dynamic-lot-sizing

动态批量订货

你是非平稳库存系统中动态批量订货模型方面的专家，这类系统的参数会随时间变化。你的目标是运用有限期规划方法，在需求、成本或价格随时间变化时，帮助优化库存补货决策。

初步评估

在解决动态批量订货问题之前，先了解以下信息：

1. 时间范围

   • 规划周期有多长？（月、季度、年）
   • 是有限期还是滚动规划？
   • 重新规划的频率是多少？

2. 时变参数

   • 哪些因素会随时间变化？（需求、成本、价格、产能）
   • 需求：是否存在季节性模式、趋势或已知变化？
   • 成本：是否会涨价，持有成本是否有季节性变化？
   • 产能：生产或仓储限制是否随时间变化？

3. 问题类型

   • 时间变化是确定性的还是随机的？
   • 价格变化是已知的（已公布）还是预测的？
   • 是否必须考虑通货膨胀？

4. 决策灵活性

   • 能否在每个周期调整决策？
   • 订单是已承诺的，还是补货安排可以灵活调整？
   • 是否存在提前期？会产生什么影响？

5. 特殊考量

   • 是否存在期末影响（残值、期末库存）？
   • 是否有价格投机机会？
   • 是否需要考虑产品过时或生命周期？

---

动态批量订货基础

问题特征

与静态 EOQ 的主要区别：

• 需求随周期变化：D₁、D₂、...、D_T
• 成本可能变化：订货成本 S_t、持有成本 h_t、采购成本 c_t
• 规划周期 T 有限（而非无限）
• 必须考虑期末影响

决策： 确定每个周期 t 的订货量 Q_t，以使总成本最小

应用场景：

• 季节性需求规划
• 价格投机（在涨价前采购）
• 产品生命周期管理（新品或衰退期产品）
• 时尚品或易腐品
• 促销规划

---

Python 实现：动态批量订货

时变需求与成本

import numpy as np
import pandas as pd
from typing import List, Dict, Optional
import matplotlib.pyplot as plt
from scipy.optimize import minimize

class DynamicLotSizing:
    """
    Dynamic lot-sizing with time-varying parameters

    Handles changes in demand, costs, and prices over time
    """

    def __init__(self, demands: List[float], setup_costs: List[float],
                 holding_costs: List[float], unit_costs: List[float],
                 initial_inventory: float = 0, salvage_value: float = 0):
        """
        Parameters:
        -----------
        demands : list
            Demand in each period [D₁, D₂, ..., D_T]
        setup_costs : list
            Setup cost in each period [S₁, S₂, ..., S_T]
        holding_costs : list
            Holding cost per unit per period [h₁, h₂, ..., h_T]
        unit_costs : list
            Purchase cost per unit [c₁, c₂, ..., c_T]
        initial_inventory : float
            Starting inventory
        salvage_value : float
            Value per unit of leftover inventory at end
        """
        self.T = len(demands)
        self.demands = np.array(demands)
        self.setup_costs = np.array(setup_costs)
        self.holding_costs = np.array(holding_costs)
        self.unit_costs = np.array(unit_costs)
        self.initial_inventory = initial_inventory
        self.salvage_value = salvage_value

    def dynamic_programming(self) -> Dict:
        """
        Solve using dynamic programming

        F(t, I_t) = minimum cost from period t onwards with inventory I_t

        For computational efficiency, discretize inventory levels
        """

        # Cumulative future demands
        cum_demand = np.zeros(self.T + 1)
        for t in range(self.T - 1, -1, -1):
            cum_demand[t] = cum_demand[t + 1] + self.demands[t]

        # Maximum useful inventory to consider
        max_inventory = int(cum_demand[0])

        # DP table: F[t][i] = min cost from period t with inventory i
        INF = 1e9
        F = [[INF] * (max_inventory + 1) for _ in range(self.T + 1)]
        decision = [[None] * (max_inventory + 1) for _ in range(self.T)]

        # Terminal condition: salvage value
        for i in range(max_inventory + 1):
            F[self.T][i] = -self.salvage_value * i

        # Backward recursion
        for t in range(self.T - 1, -1, -1):
            d_t = self.demands[t]
            S_t = self.setup_costs[t]
            h_t = self.holding_costs[t]
            c_t = self.unit_costs[t]

            for I_t in range(max_inventory + 1):
                # Option 1: Don't order
                if I_t >= d_t:
                    I_next = I_t - d_t
                    cost_no_order = h_t * I_next + F[t + 1][int(I_next)]
                else:
                    cost_no_order = INF  # Cannot satisfy demand

                # Option 2: Order
                best_order_cost = INF
                best_order_qty = 0

                # Try different order quantities
                max_order = int(cum_demand[t] - I_t)
                for Q in range(1, max_order + 1):
                    I_after_order = I_t + Q
                    if I_after_order >= d_t:
                        I_next = I_after_order - d_t
                        cost = S_t + c_t * Q + h_t * I_next + F[t + 1][int(I_next)]
                        if cost < best_order_cost:
                            best_order_cost = cost
                            best_order_qty = Q

                # Choose best option
                if cost_no_order < best_order_cost:
                    F[t][I_t] = cost_no_order
                    decision[t][I_t] = 0  # No order
                else:
                    F[t][I_t] = best_order_cost
                    decision[t][I_t] = best_order_qty

        # Forward pass to construct solution
        orders = np.zeros(self.T)
        inventory = np.zeros(self.T + 1)
        inventory[0] = self.initial_inventory

        for t in range(self.T):
            I_t = int(inventory[t])
            Q_t = decision[t][I_t] if I_t <= max_inventory else 0
            orders[t] = Q_t

            inventory[t + 1] = inventory[t] + Q_t - self.demands[t]

        # Calculate costs
        setup_cost = sum(self.setup_costs[t] for t in range(self.T) if orders[t] > 0)
        purchase_cost = sum(self.unit_costs[t] * orders[t] for t in range(self.T))
        holding_cost = sum(self.holding_costs[t] * inventory[t + 1]
                          for t in range(self.T))
        salvage = self.salvage_value * inventory[self.T]

        total_cost = setup_cost + purchase_cost + holding_cost - salvage

        return {
            'method': 'Dynamic Programming',
            'orders': orders,
            'inventory': inventory[:-1],
            'total_cost': total_cost,
            'setup_cost': setup_cost,
            'purchase_cost': purchase_cost,
            'holding_cost': holding_cost,
            'salvage_revenue': salvage
        }

    def price_speculation_analysis(self) -> Dict:
        """
        Analyze price speculation opportunities

        Identify periods where buying ahead is beneficial
        """

        speculation_opportunities = []

        for t in range(self.T - 1):
            # Compare buying now vs. buying later
            current_price = self.unit_costs[t]
            future_price = self.unit_costs[t + 1]

            price_increase = future_price - current_price
            holding_cost = self.holding_costs[t]

            # Net savings per unit if buy now for future demand
            net_savings = price_increase - holding_cost

            if net_savings > 0:
                speculation_opportunities.append({
                    'period': t + 1,
                    'current_price': current_price,
                    'future_price': future_price,
                    'price_increase': price_increase,
                    'holding_cost': holding_cost,
                    'net_savings_per_unit': net_savings
                })

        return speculation_opportunities

    def plot_time_varying_parameters(self):
        """Visualize how parameters change over time"""

        periods = np.arange(1, self.T + 1)

        fig, ((ax1, ax2), (ax3, ax4)) = plt.subplots(2, 2, figsize=(14, 10))

        # Demand
        ax1.plot(periods, self.demands, marker='o', linewidth=2, color='blue')
        ax1.set_xlabel('Period')
        ax1.set_ylabel('Demand (units)')
        ax1.set_title('Demand Over Time', fontweight='bold')
        ax1.grid(True, alpha=0.3)

        # Unit costs
        ax2.plot(periods, self.unit_costs, marker='s', linewidth=2, color='green')
        ax2.set_xlabel('Period')
        ax2.set_ylabel('Unit Cost ($)')
        ax2.set_title('Purchase Price Over Time', fontweight='bold')
        ax2.grid(True, alpha=0.3)

        # Setup costs
        ax3.plot(periods, self.setup_costs, marker='^', linewidth=2, color='red')
        ax3.set_xlabel('Period')
        ax3.set_ylabel('Setup Cost ($)')
        ax3.set_title('Setup Cost Over Time', fontweight='bold')
        ax3.grid(True, alpha=0.3)

        # Holding costs
        ax4.plot(periods, self.holding_costs, marker='d', linewidth=2, color='orange')
        ax4.set_xlabel('Period')
        ax4.set_ylabel('Holding Cost ($/unit/period)')
        ax4.set_title('Holding Cost Over Time', fontweight='bold')
        ax4.grid(True, alpha=0.3)

        plt.tight_layout()
        return plt

    def rolling_horizon_simulation(self, horizon_length: int = 6,
                                   actual_demands: Optional[List[float]] = None) -> Dict:
        """
        Simulate rolling horizon planning

        Replan every period with updated forecast

        Parameters:
        -----------
        horizon_length : int
            Length of planning horizon for each replan
        actual_demands : list, optional
            Actual realized demands (if different from forecast)
        """

        if actual_demands is None:
            actual_demands = self.demands

        actual_demands = np.array(actual_demands)

        orders = np.zeros(self.T)
        inventory = np.zeros(self.T + 1)
        inventory[0] = self.initial_inventory

        total_cost = 0

        for t in range(self.T):
            # Define planning horizon
            horizon_end = min(t + horizon_length, self.T)

            # Create subproblem for rolling horizon
            subproblem = DynamicLotSizing(
                demands=list(self.demands[t:horizon_end]),
                setup_costs=list(self.setup_costs[t:horizon_end]),
                holding_costs=list(self.holding_costs[t:horizon_end]),
                unit_costs=list(self.unit_costs[t:horizon_end]),
                initial_inventory=inventory[t],
                salvage_value=self.salvage_value
            )

            # Solve subproblem
            solution = subproblem.dynamic_programming()

            # Implement first-period decision only
            orders[t] = solution['orders'][0]

            # Update inventory based on actual demand
            inventory[t + 1] = inventory[t] + orders[t] - actual_demands[t]

            # Accumulate costs
            if orders[t] > 0:
                total_cost += self.setup_costs[t]
            total_cost += self.unit_costs[t] * orders[t]
            total_cost += self.holding_costs[t] * inventory[t + 1]

        return {
            'method': f'Rolling Horizon (H={horizon_length})',
            'orders': orders,
            'inventory': inventory[:-1],
            'total_cost': total_cost
        }


# Example: Seasonal Demand with Price Changes
def example_seasonal_with_price_change():
    """Example: Seasonal demand pattern with anticipated price increase"""

    print("\n" + "=" * 70)
    print("DYNAMIC LOT-SIZING: SEASONAL DEMAND WITH PRICE CHANGE")
    print("=" * 70)

    # 12-month planning horizon
    # Seasonal demand pattern (low winter, high summer)
    demands = [60, 50, 70, 80, 100, 120, 130, 120, 100, 80, 60, 50]

    # Price increase announced for month 6
    unit_costs = [10, 10, 10, 10, 10, 12, 12, 12, 12, 12, 12, 12]

    # Seasonal holding cost (higher in summer due to cooling requirements)
    holding_costs = [0.5, 0.5, 0.5, 0.6, 0.7, 0.8, 0.9, 0.8, 0.7, 0.6, 0.5, 0.5]

    # Constant setup cost
    setup_costs = [100] * 12

    problem = DynamicLotSizing(
        demands=demands,
        setup_costs=setup_costs,
        holding_costs=holding_costs,
        unit_costs=unit_costs,
        initial_inventory=0,
        salvage_value=5  # $5/unit salvage at end
    )

    print("\nProblem Overview:")
    print(f"  Planning Horizon: {problem.T} months")
    print(f"  Total Demand: {problem.demands.sum():.0f} units")
    print(f"  Price Change: ${unit_costs[0]} → ${unit_costs[5]} in month 6")

    # Analyze price speculation opportunities
    spec_opps = problem.price_speculation_analysis()

    if spec_opps:
        print("\n  Price Speculation Opportunities:")
        for opp in spec_opps:
            print(f"    Month {opp['period']}: Buy ahead to save "
                  f"${opp['net_savings_per_unit']:.2f}/unit")

    # Solve with DP
    print("\nSolving with Dynamic Programming...")
    solution = problem.dynamic_programming()

    print(f"\n{'=' * 70}")
    print("OPTIMAL SOLUTION")
    print("=" * 70)

    print(f"\n{'Total Cost:':<30} ${solution['total_cost']:,.2f}")
    print(f"{'Setup Cost:':<30} ${solution['setup_cost']:,.2f}")
    print(f"{'Purchase Cost:':<30} ${solution['purchase_cost']:,.2f}")
    print(f"{'Holding Cost:':<30} ${solution['holding_cost']:,.2f}")
    print(f"{'Salvage Revenue:':<30} ${solution['salvage_revenue']:,.2f}")

    print("\n  Optimal Order Plan:")
    print(f"\n  {'Month':<8} {'Demand':<10} {'Order':<10} {'End Inv':<12} "
          f"{'Unit Price':<12} {'Setup?'}")
    print("  " + "-" * 65)

    for t in range(problem.T):
        setup_indicator = "Yes" if solution['orders'][t] > 0 else "No"
        print(f"  {t+1:<8} {demands[t]:<10.0f} {solution['orders'][t]:<10.0f} "
              f"{solution['inventory'][t]:<12.0f} ${unit_costs[t]:<11.2f} {setup_indicator}")

    # Highlight speculation behavior
    print("\n  Key Insights:")
    if solution['orders'][4] > demands[4]:
        print(f"    • Month 5: Large order ({solution['orders'][4]:.0f} units) "
              f"to avoid price increase")
        print(f"      → Buying ahead at ${unit_costs[4]} vs ${unit_costs[5]} later")

    # Plot parameters and solution
    problem.plot_time_varying_parameters()
    plt.savefig('/tmp/dynamic_lot_sizing_parameters.png', dpi=300, bbox_inches='tight')
    print(f"\n  Parameter plots saved to /tmp/dynamic_lot_sizing_parameters.png")

    return problem, solution


# Example: Rolling Horizon
def example_rolling_horizon():
    """Example: Rolling horizon planning with forecast updates"""

    print("\n" + "=" * 70)
    print("ROLLING HORIZON PLANNING")
    print("=" * 70)

    # Forecasted demands
    forecast_demands = [100, 110, 95, 105, 100, 110, 105, 95, 100, 105, 110, 100]

    # Actual demands (slightly different from forecast)
    np.random.seed(42)
    actual_demands = forecast_demands + np.random.normal(0, 10, 12)
    actual_demands = np.maximum(actual_demands, 0)  # Non-negative

    problem = DynamicLotSizing(
        demands=forecast_demands,
        setup_costs=[150] * 12,
        holding_costs=[1.5] * 12,
        unit_costs=[20] * 12,
        initial_inventory=50
    )

    print("\nRolling Horizon Setup:")
    print(f"  Planning Horizon: {problem.T} periods")
    print(f"  Replanning Frequency: Every period")
    print(f"  Look-ahead Horizon: 6 periods")

    # Compare: Full horizon vs. Rolling horizon
    full_horizon = problem.dynamic_programming()
    rolling_6 = problem.rolling_horizon_simulation(horizon_length=6,
                                                   actual_demands=actual_demands)
    rolling_3 = problem.rolling_horizon_simulation(horizon_length=3,
                                                   actual_demands=actual_demands)

    print(f"\n{'=' * 70}")
    print("COMPARISON OF APPROACHES")
    print("=" * 70)

    comparison = pd.DataFrame([
        {'Method': 'Full Horizon (12 periods)', 'Total Cost': full_horizon['total_cost']},
        {'Method': 'Rolling Horizon (H=6)', 'Total Cost': rolling_6['total_cost']},
        {'Method': 'Rolling Horizon (H=3)', 'Total Cost': rolling_3['total_cost']}
    ])

    print("\n" + comparison.to_string(index=False))

    print(f"\n  Insight: Shorter horizons are more myopic but adapt to actual demand")

    return problem, rolling_6


if __name__ == "__main__":
    problem1, solution1 = example_seasonal_with_price_change()
    problem2, solution2 = example_rolling_horizon()

---

工具与库

Python 库

• numpy、scipy：数值计算
• pulp、pyomo：优化建模
• 动态规划实现

商业软件

• SAP APO：支持时变参数的高级规划
• Blue Yonder：动态规划与优化
• Kinaxis：支持情景规划的 RapidResponse
• o9 Solutions：分时段规划

---

常见挑战与解决方案

挑战：计算复杂度

问题： 动态规划的状态空间会变得很大
解决方案：

• 离散化库存水平
• 使用近似动态规划
• 采用滚动规划方法
• 对大型问题使用启发式方法

挑战：预测不确定性

问题： 未来需求或价格存在不确定性
解决方案：

• 采用滚动规划并频繁重新规划
• 使用随机动态规划处理不确定性
• 基于情景进行规划
• 使用鲁棒优化

挑战：期末影响

问题： 产生不自然的期末行为
解决方案：

• 使用适当的残值
• 将规划周期延长至决策周期之后
• 滚动规划可缓解这一问题
• 设定期末库存目标

---

相关技能

• economic-order-quantity：静态 EOQ 模型
• lot-sizing-problems：多周期确定性批量订货
• stochastic-inventory-models：需求不确定性
• demand-forecasting：预测时变需求
• seasonal-planning：季节性需求模式
• price-optimization：动态定价策略