mcpField
R
retail-replenishment

retail-replenishment

github.com/kishorkukreja其他8499 次安装3 周前更新
代码仓库

关于 retail-replenishment

面向零售补货与库存管理场景的技能,帮助零售团队制定补货策略、调整参数并处理门店和配送中心之间的库存分配。内容涵盖多种补货模式及优化流程。

主要功能

  1. 01梳理门店网络与商品特征
  2. 02比较五类补货模式
  3. 03优化最低与最高库存参数
  4. 04根据 POS 消耗补货
  5. 05分配配送中心库存
  6. 06生成补货优化报告

适用场景

  1. 01评估门店补货现状与系统能力
  2. 02为 SKU 设置最低和最高库存
  3. 03依据 POS 数据响应需求变化
  4. 04配送中心库存不足时安排分配

安装

复制下面的命令,在终端中运行。

$npx skills add https://github.com/kishorkukreja/awesome-supply-chain --skill retail-replenishment

安装后在 Claude Code 中输入/retail-replenishment确认已加载。

详细介绍

零售补货

你是零售门店补货和库存管理领域的专家。你的目标是帮助零售商优化产品从配送中心到门店的持续流动,维持最佳库存水平,在最大化销售额的同时,尽量减少库存投入、缺货和库存过剩。

初步评估

在制定补货策略之前,请先了解:

  1. 门店网络特征

    • 有多少家门店?有哪些业态?(大型卖场、小型门店、城市、郊区)
    • 配送中心的位置和产能如何?
    • 门店的配送频率是多少?(每天、每周 2 次、每周)
    • 从配送中心到门店的交货周期有多长?
    • 采用哪种补货模式?(门店发起、自动补货、VMI)
  2. 产品特征

    • 哪些品类/SKU 需要补货?
    • 需求模式如何?(稳定、多变、季节性、促销性)
    • 是否有保质期限制?(易腐品、有日期标记的产品)
    • 产品周转速度如何?(快、中、慢)
    • 单位成本和利润率是多少?
  3. 当前绩效

    • 当前有货率是多少?(目标值与实际值)
    • 平均库存周转天数是多少?
    • 各 SKU/门店的缺货频率如何?
    • 库存过剩造成的核销情况如何?
    • 补货订单准确率如何?
  4. 系统能力

    • 是否已部署补货系统?(手动、自动化)
    • 是否能实时查看库存?
    • 是否集成 POS 数据?
    • 是否具备预测能力?
    • 是否支持自动下单?

零售补货框架

补货模式

1. 最小-最大补货

  • 设置库存最小值和最大值
  • 库存达到最小值时,订货补至最大值
  • 简单,广泛用于基本商品
  • 挑战:正确设置最小值和最大值

2. 定期补货(定期检查)

  • 按固定间隔检查库存(例如,每周)
  • 订货补至目标水平
  • 可预测,更便于规划
  • 挑战:可能无法及时满足两次检查之间出现的紧急需求

3. 需求驱动补货

  • 根据实际消耗量(POS)进行补货
  • 对需求变化响应更迅速
  • 需要实时数据
  • 挑战:系统复杂

4. 供应商管理库存(VMI)

  • 由供应商管理库存水平
  • 由供应商发起补货
  • 减少零售商的工作量
  • 挑战:失去控制权,需要共享数据

5. 越库补货

  • 产品无需进入配送中心仓储
  • 补货周期更短
  • 减少搬运成本
  • 挑战:需要协调,且要求货量大

补货参数优化

最小-最大库存水平计算
import numpy as np
import pandas as pd
from scipy import stats

class ReplenishmentParameterOptimizer:
    """
    Calculate optimal replenishment parameters for stores

    Min-Max levels, reorder points, order quantities
    """

    def __init__(self, sku_data, store_data):
        """
        Parameters:
        - sku_data: SKU information (lead times, demand, etc.)
        - store_data: Store information (size, sales, format)
        """
        self.skus = sku_data
        self.stores = store_data

    def calculate_min_max_levels(self, sku, store_id, service_level=0.95,
                                 review_period_days=7):
        """
        Calculate Min-Max replenishment levels

        Min = Safety stock + Lead time demand
        Max = Safety stock + Lead time demand + Review period demand
        """

        # Get SKU and store data
        sku_info = self.skus[self.skus['sku'] == sku].iloc[0]
        store_info = self.stores[self.stores['store_id'] == store_id].iloc[0]

        # Daily demand at this store
        avg_daily_demand = store_info.get(f'{sku}_daily_demand', 0)
        demand_std = store_info.get(f'{sku}_demand_std', 0)

        # Lead time from DC to store
        lead_time_days = sku_info.get('lead_time_days', 2)
        lead_time_std = sku_info.get('lead_time_std_days', 0.5)

        # Calculate safety stock
        # Account for demand variability during lead time + review period
        total_period = lead_time_days + review_period_days

        # Combined standard deviation (demand and lead time variability)
        if demand_std > 0 and lead_time_std > 0:
            variance = (
                total_period * (demand_std ** 2) +
                (avg_daily_demand ** 2) * (lead_time_std ** 2)
            )
            combined_std = np.sqrt(variance)
        else:
            combined_std = demand_std * np.sqrt(total_period)

        # Z-score for service level
        z = stats.norm.ppf(service_level)

        # Safety stock
        safety_stock = z * combined_std

        # Lead time demand
        lead_time_demand = avg_daily_demand * lead_time_days

        # Review period demand
        review_period_demand = avg_daily_demand * review_period_days

        # Min level (reorder point)
        min_level = safety_stock + lead_time_demand

        # Max level (order up to level)
        max_level = safety_stock + lead_time_demand + review_period_demand

        # Capacity constraints
        shelf_capacity = store_info.get(f'{sku}_shelf_capacity', 999)
        backroom_capacity = store_info.get(f'{sku}_backroom_capacity', 999)
        total_capacity = shelf_capacity + backroom_capacity

        # Cap max level at capacity
        max_level = min(max_level, total_capacity)

        # Min-max spread check (min should be at least 40% of max)
        if min_level < max_level * 0.4:
            min_level = max_level * 0.4

        return {
            'sku': sku,
            'store_id': store_id,
            'min_level': round(min_level, 0),
            'max_level': round(max_level, 0),
            'safety_stock': round(safety_stock, 0),
            'avg_daily_demand': avg_daily_demand,
            'lead_time_days': lead_time_days,
            'service_level': service_level,
            'review_period_days': review_period_days
        }

    def calculate_economic_order_quantity(self, sku, annual_demand,
                                         order_cost=50, holding_cost_rate=0.25):
        """
        Calculate EOQ for replenishment order sizing

        Balance ordering costs vs. holding costs
        """

        sku_info = self.skus[self.skus['sku'] == sku].iloc[0]
        unit_cost = sku_info.get('unit_cost', 10)

        # Holding cost per unit per year
        holding_cost = unit_cost * holding_cost_rate

        # EOQ formula
        eoq = np.sqrt((2 * annual_demand * order_cost) / holding_cost)

        # Calculate order frequency
        orders_per_year = annual_demand / eoq
        days_between_orders = 365 / orders_per_year

        return {
            'sku': sku,
            'eoq': round(eoq, 0),
            'orders_per_year': round(orders_per_year, 1),
            'days_between_orders': round(days_between_orders, 1)
        }

    def optimize_replenishment_frequency(self, sku, store_id):
        """
        Determine optimal replenishment frequency

        Balance:
        - More frequent = lower inventory, better freshness, higher cost
        - Less frequent = higher inventory, lower cost, risk of stockouts
        """

        sku_info = self.skus[self.skus['sku'] == sku].iloc[0]
        store_info = self.stores[self.stores['store_id'] == store_id].iloc[0]

        avg_daily_demand = store_info.get(f'{sku}_daily_demand', 10)
        unit_cost = sku_info.get('unit_cost', 10)
        shelf_life_days = sku_info.get('shelf_life_days', 999)

        # Factors to consider
        factors = []
        recommended_frequency_days = 7  # Default weekly

        # Factor 1: Shelf life (for perishables)
        if shelf_life_days < 14:
            recommended_frequency_days = min(recommended_frequency_days, shelf_life_days // 3)
            factors.append(f"Short shelf life ({shelf_life_days} days)")

        # Factor 2: Demand velocity
        weekly_demand = avg_daily_demand * 7
        if weekly_demand > 100:
            recommended_frequency_days = min(recommended_frequency_days, 3)
            factors.append("High velocity item")
        elif weekly_demand < 10:
            recommended_frequency_days = max(recommended_frequency_days, 14)
            factors.append("Low velocity item")

        # Factor 3: Storage capacity
        shelf_capacity = store_info.get(f'{sku}_shelf_capacity', 50)
        days_of_capacity = shelf_capacity / avg_daily_demand if avg_daily_demand > 0 else 999

        if days_of_capacity < 7:
            recommended_frequency_days = min(recommended_frequency_days, 3)
            factors.append("Limited shelf capacity")

        # Factor 4: Product value
        if unit_cost > 50:
            recommended_frequency_days = min(recommended_frequency_days, 7)
            factors.append("High value item (minimize inventory investment)")

        return {
            'sku': sku,
            'store_id': store_id,
            'recommended_frequency_days': recommended_frequency_days,
            'frequency_description': self._frequency_to_description(recommended_frequency_days),
            'factors': factors
        }

    def _frequency_to_description(self, days):
        """Convert days to frequency description"""
        if days <= 1:
            return "Daily"
        elif days <= 3:
            return "2-3 times per week"
        elif days <= 7:
            return "Weekly"
        elif days <= 10:
            return "1-2 times per week"
        else:
            return "Bi-weekly or less"

# Example usage
sku_data = pd.DataFrame({
    'sku': ['SKU001', 'SKU002', 'SKU003'],
    'unit_cost': [5.50, 12.00, 3.25],
    'lead_time_days': [2, 3, 2],
    'lead_time_std_days': [0.5, 0.8, 0.5],
    'shelf_life_days': [999, 14, 999]
})

store_data = pd.DataFrame({
    'store_id': ['S001'],
    'SKU001_daily_demand': [25],
    'SKU001_demand_std': [8],
    'SKU001_shelf_capacity': [100],
    'SKU001_backroom_capacity': [200],
    'SKU002_daily_demand': [15],
    'SKU002_demand_std': [5],
    'SKU002_shelf_capacity': [50],
    'SKU002_backroom_capacity': [50],
    'SKU003_daily_demand': [50],
    'SKU003_demand_std': [15],
    'SKU003_shelf_capacity': [200],
    'SKU003_backroom_capacity': [300]
})

optimizer = ReplenishmentParameterOptimizer(sku_data, store_data)

# Calculate min-max for SKU001 at store S001
min_max = optimizer.calculate_min_max_levels('SKU001', 'S001', service_level=0.95)
print(f"SKU001 at Store S001:")
print(f"  Min level: {min_max['min_level']} units")
print(f"  Max level: {min_max['max_level']} units")
print(f"  Safety stock: {min_max['safety_stock']} units")

# Optimal replenishment frequency
frequency = optimizer.optimize_replenishment_frequency('SKU002', 'S001')
print(f"\nSKU002 replenishment frequency: {frequency['frequency_description']}")
print(f"  Factors: {frequency['factors']}")

自动补货引擎

生成门店订单
class AutomatedReplenishmentEngine:
    """
    Automated replenishment order generation

    Generate store orders based on current inventory, demand, and rules
    """

    def __init__(self, inventory_data, replenishment_rules, dc_inventory):
        """
        Parameters:
        - inventory_data: Current store inventory levels
        - replenishment_rules: Min-max levels and order parameters
        - dc_inventory: Available inventory at DC
        """
        self.inventory = inventory_data
        self.rules = replenishment_rules
        self.dc_inventory = dc_inventory

    def generate_store_orders(self, store_id, order_date):
        """
        Generate replenishment order for a store

        For each SKU:
        1. Check if below min level
        2. Calculate order quantity (up to max)
        3. Check DC availability
        4. Create order
        """

        store_inventory = self.inventory[
            self.inventory['store_id'] == store_id
        ]

        orders = []

        for idx, item in store_inventory.iterrows():
            sku = item['sku']
            current_level = item['on_hand'] + item['on_order']

            # Get replenishment rules
            rule = self.rules[
                (self.rules['sku'] == sku) &
                (self.rules['store_id'] == store_id)
            ]

            if len(rule) == 0:
                continue  # No rule defined

            rule = rule.iloc[0]

            # Check if replenishment needed
            if current_level <= rule['min_level']:
                # Order up to max level
                order_qty = rule['max_level'] - current_level

                # Apply order constraints
                order_qty = self._apply_order_constraints(
                    sku, order_qty, rule
                )

                # Check DC availability
                dc_available = self.dc_inventory[
                    self.dc_inventory['sku'] == sku
                ]['available_qty'].values[0]

                if dc_available >= order_qty:
                    orders.append({
                        'store_id': store_id,
                        'sku': sku,
                        'order_date': order_date,
                        'order_qty': order_qty,
                        'current_level': current_level,
                        'min_level': rule['min_level'],
                        'max_level': rule['max_level'],
                        'reason': 'Below minimum',
                        'status': 'Approved'
                    })
                elif dc_available > 0:
                    # Partial fulfillment
                    orders.append({
                        'store_id': store_id,
                        'sku': sku,
                        'order_date': order_date,
                        'order_qty': dc_available,
                        'current_level': current_level,
                        'min_level': rule['min_level'],
                        'max_level': rule['max_level'],
                        'reason': 'Partial - DC constrained',
                        'status': 'Partial'
                    })
                else:
                    # Out of stock at DC
                    orders.append({
                        'store_id': store_id,
                        'sku': sku,
                        'order_date': order_date,
                        'order_qty': 0,
                        'current_level': current_level,
                        'min_level': rule['min_level'],
                        'max_level': rule['max_level'],
                        'reason': 'DC out of stock',
                        'status': 'Backorder'
                    })

        return pd.DataFrame(orders)

    def _apply_order_constraints(self, sku, order_qty, rule):
        """
        Apply business constraints to order quantity

        - Minimum order quantity
        - Case pack multiples
        - Maximum order quantity
        """

        # Round to case pack
        case_pack = rule.get('case_pack', 1)
        if case_pack > 1:
            order_qty = int(np.ceil(order_qty / case_pack) * case_pack)

        # Minimum order
        min_order_qty = rule.get('min_order_qty', 0)
        if order_qty < min_order_qty:
            return 0  # Don't order if below minimum

        # Maximum order
        max_order_qty = rule.get('max_order_qty', 9999)
        order_qty = min(order_qty, max_order_qty)

        return order_qty

    def batch_generate_all_stores(self, order_date, store_ids=None):
        """
        Generate orders for all stores

        Batch processing for efficiency
        """

        if store_ids is None:
            store_ids = self.inventory['store_id'].unique()

        all_orders = []

        for store_id in store_ids:
            store_orders = self.generate_store_orders(store_id, order_date)
            all_orders.append(store_orders)

        return pd.concat(all_orders, ignore_index=True)

    def prioritize_orders_by_criticality(self, orders_df):
        """
        Prioritize orders when DC inventory is constrained

        Criteria:
        - Store importance (sales volume)
        - Stockout risk (how far below min)
        - Product importance (A vs. C items)
        """

        # Calculate criticality score
        def calculate_criticality(row):
            # How far below min (as %)
            below_min_pct = (row['min_level'] - row['current_level']) / row['min_level']
            below_min_pct = max(0, below_min_pct)

            # Store importance (would need actual store data)
            store_importance = 1.0  # Placeholder

            # Product importance (would need ABC classification)
            product_importance = 1.0  # Placeholder

            # Combined score
            criticality = (
                below_min_pct * 0.5 +
                store_importance * 0.3 +
                product_importance * 0.2
            )

            return criticality

        orders_df['criticality'] = orders_df.apply(calculate_criticality, axis=1)

        # Sort by criticality
        orders_df = orders_df.sort_values('criticality', ascending=False)

        return orders_df

# Example
inventory_data = pd.DataFrame({
    'store_id': ['S001', 'S001', 'S001'],
    'sku': ['SKU001', 'SKU002', 'SKU003'],
    'on_hand': [45, 15, 180],
    'on_order': [0, 0, 50]
})

replenishment_rules = pd.DataFrame({
    'store_id': ['S001', 'S001', 'S001'],
    'sku': ['SKU001', 'SKU002', 'SKU003'],
    'min_level': [50, 20, 200],
    'max_level': [150, 60, 400],
    'case_pack': [6, 12, 1],
    'min_order_qty': [12, 12, 50]
})

dc_inventory = pd.DataFrame({
    'sku': ['SKU001', 'SKU002', 'SKU003'],
    'available_qty': [5000, 800, 10000]
})

replen_engine = AutomatedReplenishmentEngine(
    inventory_data,
    replenishment_rules,
    dc_inventory
)

# Generate orders
orders = replen_engine.generate_store_orders('S001', '2024-03-15')
print("Generated Orders:")
print(orders[['sku', 'order_qty', 'current_level', 'min_level', 'max_level', 'reason']])

需求驱动补货

基于 POS 的补货
class DemandDrivenReplenishment:
    """
    Demand-driven replenishment using POS data

    Replenish based on actual consumption rather than forecasts
    """

    def __init__(self, pos_data, lead_time_days=2):
        """
        Parameters:
        - pos_data: Point-of-sale transaction data
          columns: ['store_id', 'sku', 'date', 'units_sold']
        - lead_time_days: Lead time from DC to store
        """
        self.pos = pos_data
        self.lead_time = lead_time_days

    def calculate_replenishment_signal(self, sku, store_id, current_inventory,
                                      lookback_days=7):
        """
        Calculate replenishment need based on recent consumption

        Uses exponentially weighted moving average for smoothing
        """

        # Get recent POS data
        recent_pos = self.pos[
            (self.pos['sku'] == sku) &
            (self.pos['store_id'] == store_id)
        ].tail(lookback_days)

        if len(recent_pos) == 0:
            return {'order_qty': 0, 'reason': 'No sales history'}

        # Calculate EWMA of daily sales
        alpha = 0.3  # Smoothing factor
        daily_sales = recent_pos['units_sold'].values

        ewma = daily_sales[0]
        for sale in daily_sales[1:]:
            ewma = alpha * sale + (1 - alpha) * ewma

        # Forecast demand over lead time
        forecast_demand = ewma * self.lead_time

        # Add buffer (safety stock)
        std_dev = np.std(daily_sales)
        safety_stock = 1.65 * std_dev * np.sqrt(self.lead_time)  # 95% service level

        # Target inventory level
        target_level = forecast_demand + safety_stock

        # Order quantity
        order_qty = max(0, target_level - current_inventory)

        return {
            'sku': sku,
            'store_id': store_id,
            'order_qty': round(order_qty, 0),
            'avg_daily_sales': round(ewma, 1),
            'forecast_demand': round(forecast_demand, 0),
            'safety_stock': round(safety_stock, 0),
            'target_level': round(target_level, 0),
            'current_inventory': current_inventory
        }

    def detect_demand_changes(self, sku, store_id, alert_threshold=0.3):
        """
        Detect significant demand changes

        Alert when demand shifts significantly (promotions, trends, stockouts)
        """

        # Get POS data
        sku_pos = self.pos[
            (self.pos['sku'] == sku) &
            (self.pos['store_id'] == store_id)
        ].sort_values('date')

        if len(sku_pos) < 14:
            return {'alert': False, 'reason': 'Insufficient history'}

        # Compare recent vs. baseline
        baseline_sales = sku_pos.iloc[:7]['units_sold'].mean()
        recent_sales = sku_pos.iloc[-7:]['units_sold'].mean()

        if baseline_sales == 0:
            return {'alert': False, 'reason': 'No baseline sales'}

        # Calculate change
        pct_change = (recent_sales - baseline_sales) / baseline_sales

        alert = abs(pct_change) > alert_threshold

        if pct_change > alert_threshold:
            alert_type = 'Demand spike'
            action = 'Increase replenishment'
        elif pct_change < -alert_threshold:
            alert_type = 'Demand drop'
            action = 'Reduce replenishment'
        else:
            alert_type = None
            action = None

        return {
            'sku': sku,
            'store_id': store_id,
            'alert': alert,
            'alert_type': alert_type,
            'pct_change': round(pct_change * 100, 1),
            'baseline_daily_sales': round(baseline_sales, 1),
            'recent_daily_sales': round(recent_sales, 1),
            'recommended_action': action
        }

# Example
dates = pd.date_range('2024-03-01', periods=21, freq='D')
pos_data = pd.DataFrame({
    'store_id': 'S001',
    'sku': 'SKU001',
    'date': dates,
    'units_sold': [25, 22, 28, 24, 26, 23, 27,  # Week 1
                   24, 26, 25, 27, 23, 24, 28,  # Week 2
                   45, 48, 42, 46, 44, 47, 43]  # Week 3 - spike
})

ddr = DemandDrivenReplenishment(pos_data, lead_time_days=2)

# Calculate replenishment
replen_signal = ddr.calculate_replenishment_signal(
    'SKU001', 'S001', current_inventory=80, lookback_days=7
)
print("Replenishment Signal:")
print(f"  Order quantity: {replen_signal['order_qty']}")
print(f"  Avg daily sales: {replen_signal['avg_daily_sales']}")
print(f"  Target level: {replen_signal['target_level']}")

# Detect demand changes
change_alert = ddr.detect_demand_changes('SKU001', 'S001', alert_threshold=0.3)
print(f"\nDemand Change Alert: {change_alert['alert']}")
if change_alert['alert']:
    print(f"  Type: {change_alert['alert_type']}")
    print(f"  Change: {change_alert['pct_change']}%")
    print(f"  Action: {change_alert['recommended_action']}")

配送中心到门店的分配

配送中心库存分配逻辑
class DCAllocationEngine:
    """
    Allocate limited DC inventory across stores

    When DC cannot fulfill all store orders
    """

    def __init__(self, store_orders, dc_inventory, store_attributes):
        self.orders = store_orders
        self.dc_inventory = dc_inventory
        self.stores = store_attributes

    def allocate_constrained_inventory(self, sku):
        """
        Allocate limited inventory fairly across stores

        Priority factors:
        - Store sales volume
        - Stockout risk (how critical)
        - Days of supply remaining
        - Store grade (A, B, C)
        """

        # Get orders for this SKU
        sku_orders = self.orders[self.orders['sku'] == sku].copy()

        if len(sku_orders) == 0:
            return pd.DataFrame()

        # Get DC available inventory
        dc_available = self.dc_inventory[
            self.dc_inventory['sku'] == sku
        ]['available_qty'].values[0]

        total_requested = sku_orders['order_qty'].sum()

        if dc_available >= total_requested:
            # Can fulfill all orders
            sku_orders['allocated_qty'] = sku_orders['order_qty']
            sku_orders['fill_rate'] = 1.0
            return sku_orders

        # Need to allocate limited inventory
        # Calculate priority score for each store
        def calculate_priority(row):
            store_info = self.stores[
                self.stores['store_id'] == row['store_id']
            ].iloc[0]

            # Factors
            sales_volume_score = store_info.get('annual_sales', 1000000) / 1000000
            grade_score = {'A': 3, 'B': 2, 'C': 1}.get(store_info.get('grade', 'B'), 2)

            # Days of supply remaining
            current_level = row.get('current_level', 0)
            avg_daily_sales = row.get('avg_daily_sales', 1)
            days_remaining = current_level / avg_daily_sales if avg_daily_sales > 0 else 999

            # Priority decreases with more days remaining
            urgency_score = 10 / (days_remaining + 1)

            # Combined priority
            priority = (
                sales_volume_score * 0.3 +
                grade_score * 0.3 +
                urgency_score * 0.4
            )

            return priority

        sku_orders['priority'] = sku_orders.apply(calculate_priority, axis=1)

        # Sort by priority
        sku_orders = sku_orders.sort_values('priority', ascending=False)

        # Allocate inventory
        remaining_inventory = dc_available

        allocated_quantities = []

        for idx, order in sku_orders.iterrows():
            if remaining_inventory <= 0:
                allocated_quantities.append(0)
            else:
                # Allocate proportional to priority, but capped at order qty
                allocation = min(order['order_qty'], remaining_inventory)
                allocated_quantities.append(allocation)
                remaining_inventory -= allocation

        sku_orders['allocated_qty'] = allocated_quantities
        sku_orders['fill_rate'] = sku_orders['allocated_qty'] / sku_orders['order_qty']

        return sku_orders[['store_id', 'sku', 'order_qty', 'allocated_qty',
                           'fill_rate', 'priority']]

# Example
store_orders = pd.DataFrame({
    'store_id': ['S001', 'S002', 'S003', 'S004'],
    'sku': 'SKU001',
    'order_qty': [100, 80, 120, 60],
    'current_level': [10, 25, 5, 30],
    'avg_daily_sales': [15, 10, 20, 8]
})

dc_inventory = pd.DataFrame({
    'sku': ['SKU001'],
    'available_qty': [250]  # Can't fulfill all 360 units
})

store_attributes = pd.DataFrame({
    'store_id': ['S001', 'S002', 'S003', 'S004'],
    'annual_sales': [5_000_000, 3_000_000, 6_000_000, 2_000_000],
    'grade': ['A', 'B', 'A', 'C']
})

allocator = DCAllocationEngine(store_orders, dc_inventory, store_attributes)

# Allocate constrained inventory
allocation = allocator.allocate_constrained_inventory('SKU001')
print("DC Inventory Allocation (250 units available, 360 requested):")
print(allocation)
print(f"\nTotal allocated: {allocation['allocated_qty'].sum()}")

工具与库

Python 库

优化:

  • scipy.optimize:用于优化补货参数
  • pulp、pyomo:用于分配的线性规划
  • numpy:数值计算

预测:

  • statsmodels:时间序列预测
  • prophet:Facebook Prophet 需求预测
  • scikit-learn:机器学习模型

数据处理:

  • pandas:数据处理与分析
  • matplotlib、seaborn:可视化
商业软件

补货系统:

  • Blue Yonder (JDA) Replenishment:AI 驱动的自动补货
  • o9 Solutions:数字化供应链规划
  • SAP IBP:集成业务规划
  • Oracle Retail Demand Forecasting:面向零售业的补货
  • RELEX Solutions:自动化补货

库存管理:

  • Manhattan Active Inventory:实时库存管理
  • Infor SCM:供应链管理套件
  • Aptos Retail:零售运营平台

常见挑战与解决方案

挑战:需求波动

问题:

  • 需求激增或骤降难以预测
  • 促销会扭曲需求模式
  • 季节性变化
  • 难以维持服务水平

解决方案:

  • 动态调整安全库存
  • 需求感知(实时监控 POS)
  • 分别制定基准预测和促销预测
  • 为需求波动较大的商品提高补货频率
  • 在 DC 设置缓冲库存
  • 提供加急运输选项
挑战:交货周期长

问题:

  • 从 DC 到门店的交货周期为 3-7 天
  • 难以应对需求变化
  • 需要更高的库存水平

解决方案:

  • 为畅销商品采用越库作业
  • 提高补货频率
  • 根据交货周期调整安全库存
  • 在门店附近提前部署库存
  • 部分商品由供应商直接配送至门店
  • 对紧急需求采用空运
挑战:DC 缺货

问题:

  • DC 缺货
  • 门店无法获得补货
  • 销售损失,顾客满意度下降

解决方案:

  • 优化 DC 安全库存
  • 提高上游供应可视性
  • 门店间调拨
  • 从其他 DC 紧急订货
  • 由供应商直接发货
  • 与门店清晰沟通
挑战:门店库存过剩

问题:

  • 库存过多
  • 资金被占用
  • 需要降价处理
  • 空间有限

解决方案:

  • 采用更保守的库存上限
  • 建立退回 DC 的计划
  • 门店间库存再平衡
  • 分配新到货时考虑门店库存
  • 识别滞销商品并采取行动
  • 改进需求预测
挑战:补货系统复杂

问题:

  • 数千家门店 × 数千种 SKU
  • 不同品类和门店采用不同规则
  • 系统性能与维护问题

解决方案:

  • 聚类(相似门店采用相似规则)
  • ABC 分析(根据重要性采用不同管理力度)
  • 基于例外情况进行管理
  • 自动维护规则
  • 定期调整参数
  • 基于云的可扩展系统

输出格式

补货优化报告

执行摘要:

  • 门店网络:185 家门店,涵盖 3 种业态
  • SKU 数量:12,500 个活跃 SKU
  • 当前有货率:92.3%
  • 目标有货率:96%
  • 当前平均库存天数:28 天
  • 目标平均库存天数:22 天

当前参数与优化后参数:

指标当前优化后改善幅度
有货率92.3%96.2%+3.9 个百分点
平均库存天数28 天22 天-21%
每周缺货事件340115-66%
超额库存价值$4.2M$2.8M-$1.4M
每周补货订单2,8503,120+9%

补货频率调整:

品类当前频率优化后频率理由
生鲜农产品每周 2 次每日保质期短,周转快
乳制品每周 2 次每周 3 次易腐,需求稳定
干货食品每周一次每周一次稳定,保质期长
健康与美容每周一次每两周一次周转慢,空间受限

最小值-最大值参数调整(SKU 示例):

SKU门店等级当前最小值当前最大值优化后最小值优化后最大值调整理由
SKU001A5020075175提高最小值以保障服务水平,降低最大值以减少库存
SKU002B301004090根据需求缩小范围
SKU003C20801550下调周转缓慢商品的库存

配送中心分配优先级矩阵:

当配送中心库存受限时,按以下优先级分配:

优先级门店等级可供货天数占申请量的分配比例
1(最高)A<3 天100%
2A3-5 天90%
3B<3 天90%
4A>5 天75%
5B3-5 天75%
6C<3 天70%
7(最低)C>3 天50%

预期财务影响:

收益年度影响
减少缺货(增量销售额)+$2.8M
降低超额库存持有成本+$420K
减少降价促销(改善库存管理)+$650K
减少加急货运+$180K
年度收益总计+$4.05M
成本年度影响
提高补货频率(运输)-$320K
系统实施与维护-$250K
年度成本总计-$570K

年度净收益:+$3.48M

实施路线图:

阶段时间安排活动预期影响
阶段 1第 1-2 个月优化排名靠前的 500 个 SKU(A 类商品)的最小值-最大值参数60% 的收益
阶段 2第 3-4 个月推广至所有 SKU 和所有门店85% 的收益
阶段 3第 5-6 个月实施配送中心分配优先级95% 的收益
阶段 4第 7 个月起微调并持续改进100% 的收益

需要提出的问题

如果需要更多背景信息:

  1. 您的网络中有多少家门店和配送中心?
  2. 您当前的有货率是多少?目标是多少?
  3. 补货频率如何?(每日、每周等)
  4. 从配送中心到门店的交货周期是多久?
  5. 您使用自动补货还是人工补货?
  6. 您使用什么系统进行补货?
  7. 您面临的主要痛点是什么?(缺货、库存过剩、复杂性)
  8. 您是否有实时 POS 数据?
  9. 哪些品类/SKU 需要优化?

相关技能

  • retail-allocation:向门店进行初始分配
  • demand-forecasting:补货需求预测
  • inventory-optimization:安全库存和再订货点优化
  • warehouse-design:配送中心运营与效率
  • route-optimization:配送路线优化
  • supply-chain-analytics:补货绩效指标
  • omnichannel-fulfillment:跨渠道库存管理

评论

支持 Markdown