【2026年版】AIサプライチェーン最適化：需要予測・在庫管理・物流効率化の実践ガイド

# demand_forecasting.py - LightGBMベースの需要予測
import pandas as pd
import numpy as np
import lightgbm as lgb
from sklearn.model_selection import TimeSeriesSplit
from sklearn.metrics import mean_absolute_percentage_error

class DemandForecaster:
    """AI需要予測エンジン"""

    def __init__(self):
        self.model = None
        self.feature_columns = []

    def create_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """特徴量エンジニアリング"""
        df = df.copy()

        # 時間特徴量
        df['day_of_week'] = df['date'].dt.dayofweek
        df['month'] = df['date'].dt.month
        df['week_of_year'] = df['date'].dt.isocalendar().week.astype(int)
        df['is_weekend'] = df['day_of_week'].isin([5, 6]).astype(int)
        df['is_month_start'] = df['date'].dt.is_month_start.astype(int)
        df['is_month_end'] = df['date'].dt.is_month_end.astype(int)

        # ラグ特徴量
        for lag in [1, 7, 14, 28, 90]:
            df[f'sales_lag_{lag}'] = df.groupby('product_id')['sales'].shift(lag)

        # 移動平均特徴量
        for window in [7, 14, 28]:
            df[f'sales_rolling_mean_{window}'] = (
                df.groupby('product_id')['sales']
                .transform(lambda x: x.rolling(window, min_periods=1).mean())
            )
            df[f'sales_rolling_std_{window}'] = (
                df.groupby('product_id')['sales']
                .transform(lambda x: x.rolling(window, min_periods=1).std())
            )

        # トレンド特徴量
        df['sales_trend'] = (
            df.groupby('product_id')['sales']
            .transform(lambda x: x.rolling(28).mean() - x.rolling(90).mean())
        )

        return df

    def train(self, df: pd.DataFrame):
        """モデル学習"""
        df = self.create_features(df)
        df = df.dropna()

        self.feature_columns = [
            col for col in df.columns
            if col not in ['date', 'product_id', 'sales']
        ]

        X = df[self.feature_columns]
        y = df['sales']

        # 時系列分割での交差検証
        tscv = TimeSeriesSplit(n_splits=5)
        scores = []

        for train_idx, val_idx in tscv.split(X):
            X_train, X_val = X.iloc[train_idx], X.iloc[val_idx]
            y_train, y_val = y.iloc[train_idx], y.iloc[val_idx]

            train_data = lgb.Dataset(X_train, label=y_train)
            val_data = lgb.Dataset(X_val, label=y_val)

            params = {
                'objective': 'regression',
                'metric': 'mape',
                'learning_rate': 0.05,
                'num_leaves': 63,
                'feature_fraction': 0.8,
                'bagging_fraction': 0.8,
                'bagging_freq': 5,
                'verbose': -1
            }

            model = lgb.train(
                params, train_data,
                num_boost_round=1000,
                valid_sets=[val_data],
                callbacks=[lgb.early_stopping(50)]
            )

            predictions = model.predict(X_val)
            mape = mean_absolute_percentage_error(y_val, predictions)
            scores.append(mape)

        print(f"平均MAPE: {np.mean(scores):.4f}")

        # 全データで最終モデル学習
        train_data = lgb.Dataset(X, label=y)
        self.model = lgb.train(params, train_data, num_boost_round=500)

    def predict(self, df: pd.DataFrame) -> np.ndarray:
        """需要予測"""
        df = self.create_features(df)
        X = df[self.feature_columns]
        return self.model.predict(X)

    def get_feature_importance(self) -> pd.DataFrame:
        """特徴量重要度の取得"""
        importance = pd.DataFrame({
            'feature': self.feature_columns,
            'importance': self.model.feature_importance(importance_type='gain')
        }).sort_values('importance', ascending=False)
        return importance

# inventory_optimizer.py - 動的安全在庫最適化
from dataclasses import dataclass
from typing import List
import numpy as np
from scipy import stats

@dataclass
class SKUData:
    """SKU（在庫管理単位）データ"""
    sku_id: str
    unit_cost: float
    holding_cost_rate: float  # 年間保管コスト率
    stockout_cost: float      # 欠品コスト/個
    lead_time_days: int       # リードタイム（日）
    lead_time_std: float      # リードタイム標準偏差
    daily_demand_mean: float  # 日次需要平均
    daily_demand_std: float   # 日次需要標準偏差

class InventoryOptimizer:
    """AI在庫最適化エンジン"""

    def __init__(self, service_level: float = 0.95):
        self.service_level = service_level
        self.z_score = stats.norm.ppf(service_level)

    def calculate_safety_stock(self, sku: SKUData) -> float:
        """安全在庫の計算"""
        # リードタイム中の需要の標準偏差
        demand_std_during_lt = np.sqrt(
            sku.lead_time_days * (sku.daily_demand_std ** 2) +
            (sku.daily_demand_mean ** 2) * (sku.lead_time_std ** 2)
        )
        safety_stock = self.z_score * demand_std_during_lt
        return np.ceil(safety_stock)

    def calculate_reorder_point(self, sku: SKUData) -> float:
        """発注点の計算"""
        avg_demand_during_lt = sku.daily_demand_mean * sku.lead_time_days
        safety_stock = self.calculate_safety_stock(sku)
        return np.ceil(avg_demand_during_lt + safety_stock)

    def calculate_eoq(self, sku: SKUData, ordering_cost: float) -> float:
        """経済的発注量（EOQ）の計算"""
        annual_demand = sku.daily_demand_mean * 365
        holding_cost = sku.unit_cost * sku.holding_cost_rate
        eoq = np.sqrt(
            (2 * annual_demand * ordering_cost) / holding_cost
        )
        return np.ceil(eoq)

    def optimize_all(self, skus: List[SKUData], ordering_cost: float) -> dict:
        """全SKUの在庫パラメータ最適化"""
        results = {}
        for sku in skus:
            safety_stock = self.calculate_safety_stock(sku)
            reorder_point = self.calculate_reorder_point(sku)
            eoq = self.calculate_eoq(sku, ordering_cost)

            # 年間コスト試算
            annual_demand = sku.daily_demand_mean * 365
            order_count = annual_demand / eoq
            annual_ordering_cost = order_count * ordering_cost
            avg_inventory = eoq / 2 + safety_stock
            annual_holding_cost = avg_inventory * sku.unit_cost * sku.holding_cost_rate
            total_annual_cost = annual_ordering_cost + annual_holding_cost

            results[sku.sku_id] = {
                'safety_stock': int(safety_stock),
                'reorder_point': int(reorder_point),
                'order_quantity': int(eoq),
                'avg_inventory': int(avg_inventory),
                'annual_cost': round(total_annual_cost, 2),
                'service_level': self.service_level
            }

        return results

# route_optimizer.py - Google OR-Tools による配送ルート最適化
from ortools.constraint_solver import routing_enums_pb2, pywrapcp
from typing import List, Tuple
import numpy as np

class DeliveryRouteOptimizer:
    """配送ルート最適化エンジン"""

    def __init__(self, distance_matrix: np.ndarray, num_vehicles: int,
                 depot: int = 0):
        self.distance_matrix = distance_matrix
        self.num_vehicles = num_vehicles
        self.depot = depot

    def solve(self, demands: List[int] = None,
              vehicle_capacities: List[int] = None,
              time_windows: List[Tuple[int, int]] = None) -> dict:
        """ルート最適化の実行"""

        manager = pywrapcp.RoutingIndexManager(
            len(self.distance_matrix), self.num_vehicles, self.depot
        )
        routing = pywrapcp.RoutingModel(manager)

        # 距離コールバック
        def distance_callback(from_index, to_index):
            from_node = manager.IndexToNode(from_index)
            to_node = manager.IndexToNode(to_index)
            return int(self.distance_matrix[from_node][to_node])

        transit_callback_index = routing.RegisterTransitCallback(distance_callback)
        routing.SetArcCostEvaluatorOfAllVehicles(transit_callback_index)

        # 容量制約
        if demands and vehicle_capacities:
            def demand_callback(from_index):
                from_node = manager.IndexToNode(from_index)
                return demands[from_node]

            demand_callback_index = routing.RegisterUnaryTransitCallback(
                demand_callback
            )
            routing.AddDimensionWithVehicleCapacity(
                demand_callback_index, 0, vehicle_capacities, True, 'Capacity'
            )

        # ソルバー設定
        search_parameters = pywrapcp.DefaultRoutingSearchParameters()
        search_parameters.first_solution_strategy = (
            routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC
        )
        search_parameters.local_search_metaheuristic = (
            routing_enums_pb2.LocalSearchMetaheuristic.GUIDED_LOCAL_SEARCH
        )
        search_parameters.time_limit.seconds = 30

        # 解の取得
        solution = routing.SolveWithParameters(search_parameters)

        if solution:
            return self._extract_solution(manager, routing, solution)
        return None

    def _extract_solution(self, manager, routing, solution) -> dict:
        """解の抽出"""
        routes = {}
        total_distance = 0

        for vehicle_id in range(self.num_vehicles):
            route = []
            index = routing.Start(vehicle_id)
            route_distance = 0

            while not routing.IsEnd(index):
                node = manager.IndexToNode(index)
                route.append(node)
                previous_index = index
                index = solution.Value(routing.NextVar(index))
                route_distance += routing.GetArcCostForVehicle(
                    previous_index, index, vehicle_id
                )

            route.append(manager.IndexToNode(index))
            routes[vehicle_id] = {
                'route': route,
                'distance': route_distance
            }
            total_distance += route_distance

        return {
            'routes': routes,
            'total_distance': total_distance,
            'num_vehicles_used': sum(
                1 for r in routes.values() if len(r['route']) > 2
            )
        }