关于 retail-allocation
面向零售商的商品分配规划技能，帮助按门店需求、容量和库存目标分配新品。它涵盖门店分群、需求驱动分配、尺码曲线优化，以及预包装与逐店拣配的权衡。
主要功能
- 01按需求优化门店分配
- 02门店分群与等级划分
- 03服装尺码曲线优化
- 04比较预包装与逐店拣配
- 05兼顾销售与降价风险
适用场景
- 01为新品制定门店初始分货计划
- 02按本地需求调整门店尺码配比
- 03为小型门店规划精选商品组合
- 04分析预包装和逐店拣配的运营权衡
安装
复制下面的命令，在终端中运行。
$
npx skills add https:/安装后在 Claude Code 中输入/retail-allocation确认已加载。
详细介绍
零售分货
你是零售分货和商品组合规划方面的专家。你的目标是帮助零售商将新品最优地分配到各门店，兼顾本地需求模式、门店容量和库存效率，从而最大化销售额并尽量减少降价销售。
初步评估
在制定分货策略前，先了解以下信息：
-
商品特征
- 分配的是哪些商品类别？
- 时尚商品还是基本款商品？（时尚商品＝风险更高）
- SKU 数量和复杂程度如何？（尺码／颜色／款式矩阵）
- 单位成本和零售价格是多少？
- 是否有季节性？（返校季、节假日、春季）
-
门店网络
- 连锁体系共有多少家门店？
- 门店类型／级别有哪些？（旗舰店、标准店、奥特莱斯店）
- 门店如何分组？（人口统计、气候、销售额）
- 门店规模有哪些差异？（占地面积、库存容量）
- 门店的地理分布如何？（区域差异）
-
当前流程
- 目前如何进行分货？（手动、基于系统）
- 分货决策由哪些因素驱动？（平均分配、历史销售、占地面积）
- 分货频率如何？（每周、每季、临时）
- 当前售罄率是多少？
- 当前降价率是多少？
-
业务目标
- 是要最大化销售额还是尽量减少降价销售？
- 各门店级别的服务水平目标是什么？
- 库存周转目标是什么？
- 期望的库存覆盖时间是多少？（可供销售周数）
- 区域／本地化定制程度如何？
零售分货框架
分货原则
1. 以需求为导向的分货
- 根据预测的本地需求进行分货
- 考虑人口统计特征、气候和历史销售
- 将合适的商品以合适的数量分配到合适的门店
2. 门店分组
- 将相似门店归为一组
- 根据组别特征进行分货
- 与逐家门店分货相比，可降低复杂度
3. 门店分级与分配
- 根据销售额／重要性对门店分级
- A 级门店获得完整商品组合和更深的库存
- C 级门店获得精选商品组合和较浅的库存
4. 优化尺码曲线
- 根据本地需求特征分配尺码
- 避免采用一刀切的方法
- 考虑城市、郊区和区域性尺码差异
5. 预包装与按店拣配
- 预包装：固定商品组合，配送中心运营效率高
- 按店拣配：按门店定制，准确性更高
- 权衡：效率与优化程度
门店分组与分级
门店分组分析
import numpy as np
import pandas as pd
from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler
import matplotlib.pyplot as plt
class StoreClusteringEngine:
"""
Cluster stores for allocation planning
Group stores with similar characteristics for efficient allocation
"""
def __init__(self, store_data):
"""
Parameters:
- store_data: DataFrame with store attributes
columns: ['store_id', 'sales_volume', 'demographics', 'climate',
'square_feet', 'location_type', 'income_level', etc.]
"""
self.stores = store_data
def create_clustering_features(self):
"""
Engineer features for clustering
Combine multiple attributes into clustering dimensions
"""
features = self.stores.copy()
# Sales volume (log-transformed for better distribution)
features['log_sales'] = np.log1p(features['annual_sales'])
# Demographics encoding
features['high_income'] = (features['median_income'] > 75000).astype(int)
features['urban'] = (features['location_type'] == 'urban').astype(int)
features['suburban'] = (features['location_type'] == 'suburban').astype(int)
# Climate encoding
features['warm_climate'] = (features['avg_temp_f'] > 65).astype(int)
features['cold_climate'] = (features['avg_temp_f'] < 45).astype(int)
# Size normalized
features['size_normalized'] = features['square_feet'] / 1000
# Customer profile
features['fashion_forward'] = features.get('fashion_index', 50) / 100
# Select clustering features
clustering_cols = [
'log_sales', 'high_income', 'urban', 'suburban',
'warm_climate', 'cold_climate', 'size_normalized',
'fashion_forward'
]
return features[['store_id'] + clustering_cols]
def perform_clustering(self, n_clusters=8, method='kmeans'):
"""
Cluster stores using K-means
Returns cluster assignments and profiles
"""
# Prepare features
feature_df = self.create_clustering_features()
feature_cols = [col for col in feature_df.columns if col != 'store_id']
X = feature_df[feature_cols].values
# Standardize features
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)
# K-means clustering
kmeans = KMeans(n_clusters=n_clusters, random_state=42, n_init=10)
clusters = kmeans.fit_predict(X_scaled)
# Add cluster labels
feature_df['cluster'] = clusters
# Profile each cluster
cluster_profiles = []
for cluster_id in range(n_clusters):
cluster_stores = feature_df[feature_df['cluster'] == cluster_id]
profile = {
'cluster_id': cluster_id,
'num_stores': len(cluster_stores),
'avg_sales': np.exp(cluster_stores['log_sales'].mean()) - 1,
'high_income_pct': cluster_stores['high_income'].mean() * 100,
'urban_pct': cluster_stores['urban'].mean() * 100,
'warm_climate_pct': cluster_stores['warm_climate'].mean() * 100,
'avg_sqft': cluster_stores['size_normalized'].mean() * 1000,
'fashion_index': cluster_stores['fashion_forward'].mean() * 100
}
# Generate cluster name
profile['cluster_name'] = self._generate_cluster_name(profile)
cluster_profiles.append(profile)
return feature_df[['store_id', 'cluster']], pd.DataFrame(cluster_profiles)
def _generate_cluster_name(self, profile):
"""Generate descriptive cluster name"""
size = 'Large' if profile['avg_sqft'] > 15000 else 'Medium' if profile['avg_sqft'] > 8000 else 'Small'
location = 'Urban' if profile['urban_pct'] > 50 else 'Suburban'
income = 'High-Income' if profile['high_income_pct'] > 50 else 'Mid-Income'
climate = 'Warm' if profile['warm_climate_pct'] > 50 else 'Cold'
return f"{size} {location} {income} {climate}"
def optimize_cluster_count(self, max_clusters=12):
"""
Find optimal number of clusters using elbow method
Returns: Plot of inertia by cluster count
"""
feature_df = self.create_clustering_features()
feature_cols = [col for col in feature_df.columns if col != 'store_id']
X = feature_df[feature_cols].values
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)
inertias = []
cluster_range = range(2, max_clusters + 1)
for k in cluster_range:
kmeans = KMeans(n_clusters=k, random_state=42, n_init=10)
kmeans.fit(X_scaled)
inertias.append(kmeans.inertia_)
# Plot elbow curve
plt.figure(figsize=(10, 6))
plt.plot(cluster_range, inertias, 'bo-')
plt.xlabel('Number of Clusters')
plt.ylabel('Inertia')
plt.title('Elbow Method For Optimal Clusters')
plt.grid(True)
return cluster_range, inertias
# Example usage
store_data = pd.DataFrame({
'store_id': [f'S{i:03d}' for i in range(1, 151)],
'annual_sales': np.random.lognormal(14, 0.5, 150),
'median_income': np.random.normal(65000, 20000, 150),
'location_type': np.random.choice(['urban', 'suburban', 'rural'], 150, p=[0.3, 0.5, 0.2]),
'avg_temp_f': np.random.normal(55, 15, 150),
'square_feet': np.random.normal(12000, 4000, 150),
'fashion_index': np.random.normal(60, 20, 150)
})
clustering_engine = StoreClusteringEngine(store_data)
cluster_assignments, cluster_profiles = clustering_engine.perform_clustering(n_clusters=6)
print("Cluster Profiles:")
print(cluster_profiles[['cluster_id', 'cluster_name', 'num_stores', 'avg_sales']])
门店分级系统
class StoreGradingSystem:
"""
Grade stores for allocation planning
A stores: Highest volume, full assortment, deepest stock
B stores: Medium volume, core assortment
C stores: Lower volume, curated assortment
"""
def __init__(self, store_sales_data):
"""
Parameters:
- store_sales_data: DataFrame with store performance
columns: ['store_id', 'annual_sales', 'sales_per_sqft',
'inventory_turns', 'markdown_rate']
"""
self.stores = store_sales_data
def grade_stores(self, a_pct=20, b_pct=30):
"""
Assign grades to stores
Parameters:
- a_pct: Percentage of stores to grade as A (top performers)
- b_pct: Percentage to grade as B
- Remainder are C stores
"""
# Calculate composite score
# Normalize each metric
self.stores['sales_score'] = self._normalize_score(self.stores['annual_sales'])
self.stores['efficiency_score'] = self._normalize_score(self.stores['sales_per_sqft'])
self.stores['turns_score'] = self._normalize_score(self.stores['inventory_turns'])
self.stores['markdown_score'] = self._normalize_score(
-self.stores['markdown_rate'] # Lower markdown is better
)
# Weighted composite score
self.stores['composite_score'] = (
self.stores['sales_score'] * 0.5 +
self.stores['efficiency_score'] * 0.2 +
self.stores['turns_score'] * 0.2 +
self.stores['markdown_score'] * 0.1
)
# Assign grades based on percentiles
a_threshold = self.stores['composite_score'].quantile(1 - a_pct/100)
b_threshold = self.stores['composite_score'].quantile(1 - (a_pct + b_pct)/100)
def assign_grade(score):
if score >= a_threshold:
return 'A'
elif score >= b_threshold:
return 'B'
else:
return 'C'
self.stores['grade'] = self.stores['composite_score'].apply(assign_grade)
# Calculate grade statistics
grade_stats = self.stores.groupby('grade').agg({
'store_id': 'count',
'annual_sales': 'sum',
'sales_per_sqft': 'mean',
'markdown_rate': 'mean'
}).rename(columns={'store_id': 'num_stores'})
grade_stats['sales_pct'] = (
grade_stats['annual_sales'] / grade_stats['annual_sales'].sum() * 100
)
return self.stores[['store_id', 'grade', 'composite_score']], grade_stats
def _normalize_score(self, series):
"""Normalize to 0-100 scale"""
min_val = series.min()
max_val = series.max()
if max_val == min_val:
return pd.Series([50] * len(series))
return ((series - min_val) / (max_val - min_val)) * 100
def recommend_allocation_depth(self, grade):
"""
Recommend inventory depth by store grade
Returns weeks of supply and min/max units
"""
allocation_rules = {
'A': {
'weeks_of_supply': 8,
'min_units': 10,
'max_units': 100,
'assortment_breadth': 'Full',
'size_curve_depth': 'Deep (all sizes)'
},
'B': {
'weeks_of_supply': 6,
'min_units': 5,
'max_units': 50,
'assortment_breadth': 'Core (80% of SKUs)',
'size_curve_depth': 'Medium (popular sizes)'
},
'C': {
'weeks_of_supply': 4,
'min_units': 2,
'max_units': 20,
'assortment_breadth': 'Curated (50% of SKUs)',
'size_curve_depth': 'Shallow (best-sellers only)'
}
}
return allocation_rules.get(grade, allocation_rules['C'])
# Example
store_performance = pd.DataFrame({
'store_id': [f'S{i:03d}' for i in range(1, 101)],
'annual_sales': np.random.lognormal(14, 0.6, 100),
'sales_per_sqft': np.random.normal(350, 80, 100),
'inventory_turns': np.random.normal(4.5, 1.0, 100),
'markdown_rate': np.random.normal(0.25, 0.08, 100)
})
grading_system = StoreGradingSystem(store_performance)
store_grades, grade_stats = grading_system.grade_stores(a_pct=20, b_pct=30)
print("Store Grade Distribution:")
print(grade_stats)
print("\nAllocation Rules by Grade:")
for grade in ['A', 'B', 'C']:
rules = grading_system.recommend_allocation_depth(grade)
print(f"\nGrade {grade}:")
print(f" Weeks of supply: {rules['weeks_of_supply']}")
print(f" Assortment: {rules['assortment_breadth']}")
分货优化
基于需求的分货
class RetailAllocationEngine:
"""
Optimize retail allocation using demand forecasts
Allocate inventory to maximize sales while minimizing markdowns
"""
def __init__(self, store_data, product_data):
"""
Parameters:
- store_data: Store attributes and clusters
- product_data: Product attributes and total available inventory
"""
self.stores = store_data
self.products = product_data
def calculate_store_demand(self, sku, store_id, forecast_period_weeks=12):
"""
Calculate expected demand for SKU at store
Uses:
- Historical sales patterns
- Store cluster characteristics
- Seasonality
- New product benchmarks
"""
store = self.stores[self.stores['store_id'] == store_id].iloc[0]
product = self.products[self.products['sku'] == sku].iloc[0]
# Base demand from historical data
if product.get('is_new_product', False):
# New product: use similar product analogs
base_weekly_demand = self._estimate_new_product_demand(product, store)
else:
# Existing product: use historical sales
base_weekly_demand = store.get(f'{sku}_weekly_sales', 0)
# Adjust for store characteristics
cluster_factor = store.get('cluster_demand_index', 1.0)
seasonality_factor = product.get('season_factor', 1.0)
adjusted_weekly_demand = (
base_weekly_demand *
cluster_factor *
seasonality_factor
)
total_forecast_demand = adjusted_weekly_demand * forecast_period_weeks
return {
'store_id': store_id,
'sku': sku,
'weekly_demand': adjusted_weekly_demand,
'total_forecast_demand': total_forecast_demand,
'forecast_weeks': forecast_period_weeks
}
def allocate_sku_to_stores(self, sku, total_available_units,
allocation_method='proportional'):
"""
Allocate SKU inventory across stores
Methods:
- proportional: Allocate based on demand forecast
- equal: Equal distribution (simple but suboptimal)
- tiered: Different depths by store grade
"""
# Calculate demand for each store
store_demands = []
for store_id in self.stores['store_id']:
demand = self.calculate_store_demand(sku, store_id)
store_demands.append(demand)
demands_df = pd.DataFrame(store_demands)
if allocation_method == 'proportional':
allocations = self._allocate_proportional(
demands_df, total_available_units
)
elif allocation_method == 'tiered':
allocations = self._allocate_tiered(
demands_df, total_available_units
)
else: # equal
allocations = self._allocate_equal(
demands_df, total_available_units
)
return allocations
def _allocate_proportional(self, demands_df, total_units):
"""
Proportional allocation based on demand
Stores with higher demand get more inventory
"""
total_demand = demands_df['total_forecast_demand'].sum()
if total_demand == 0:
# No demand - distribute equally
return self._allocate_equal(demands_df, total_units)
# Calculate allocation percentage
demands_df['allocation_pct'] = (
demands_df['total_forecast_demand'] / total_demand
)
# Allocate units
demands_df['allocated_units'] = (
demands_df['allocation_pct'] * total_units
).astype(int)
# Handle rounding - allocate remainder to highest demand stores
allocated_total = demands_df['allocated_units'].sum()
remainder = total_units - allocated_total
if remainder > 0:
# Give remainder to top stores by demand
top_stores = demands_df.nlargest(remainder, 'total_forecast_demand')
for idx in top_stores.index:
demands_df.at[idx, 'allocated_units'] += 1
# Calculate expected service level (allocation / demand)
demands_df['service_level'] = np.minimum(
demands_df['allocated_units'] / demands_df['total_forecast_demand'],
1.0
)
return demands_df[['store_id', 'sku', 'allocated_units',
'total_forecast_demand', 'service_level']]
def _allocate_tiered(self, demands_df, total_units):
"""
Tiered allocation by store grade
A stores get priority, then B, then C
"""
# Merge with store grades
demands_df = demands_df.merge(
self.stores[['store_id', 'grade']],
on='store_id'
)
# Allocate by tier
allocations = []
remaining_units = total_units
for grade in ['A', 'B', 'C']:
grade_stores = demands_df[demands_df['grade'] == grade]
if len(grade_stores) == 0 or remaining_units == 0:
continue
# Allocate proportionally within grade
grade_demand = grade_stores['total_forecast_demand'].sum()
if grade_demand > 0:
for idx, row in grade_stores.iterrows():
store_allocation = min(
int(remaining_units * (row['total_forecast_demand'] / grade_demand)),
row['total_forecast_demand'],
remaining_units
)
allocations.append({
'store_id': row['store_id'],
'sku': row['sku'],
'allocated_units': store_allocation,
'total_forecast_demand': row['total_forecast_demand'],
'grade': grade
})
remaining_units -= store_allocation
return pd.DataFrame(allocations)
def _allocate_equal(self, demands_df, total_units):
"""Simple equal distribution"""
num_stores = len(demands_df)
units_per_store = total_units // num_stores
remainder = total_units % num_stores
demands_df['allocated_units'] = units_per_store
# Distribute remainder
if remainder > 0:
demands_df.iloc[:remainder, demands_df.columns.get_loc('allocated_units')] += 1
return demands_df[['store_id', 'sku', 'allocated_units']]
def _estimate_new_product_demand(self, product, store):
"""
Estimate demand for new product using analog products
Use similar products as reference
"""
# Find similar products (same category, price range)
similar_products_avg_sales = 10 # Simplified
# Adjust for store characteristics
store_factor = store.get('sales_index', 1.0)
estimated_weekly_demand = similar_products_avg_sales * store_factor
return estimated_weekly_demand
def optimize_allocation_with_constraints(self, sku, total_units,
min_units_per_store=2,
max_units_per_store=50):
"""
Optimize allocation with business constraints
Ensures min/max bounds per store
"""
# Calculate base allocation
base_allocation = self.allocate_sku_to_stores(
sku, total_units, allocation_method='proportional'
)
# Apply constraints
base_allocation['allocated_units'] = base_allocation['allocated_units'].clip(
lower=min_units_per_store,
upper=max_units_per_store
)
# Rebalance if total exceeds available
total_allocated = base_allocation['allocated_units'].sum()
if total_allocated > total_units:
# Reduce proportionally
reduction_factor = total_units / total_allocated
base_allocation['allocated_units'] = (
base_allocation['allocated_units'] * reduction_factor
).astype(int)
return base_allocation
# Example usage
store_data = pd.DataFrame({
'store_id': [f'S{i:03d}' for i in range(1, 51)],
'grade': np.random.choice(['A', 'B', 'C'], 50, p=[0.2, 0.3, 0.5]),
'sales_index': np.random.uniform(0.7, 1.3, 50),
'cluster_demand_index': np.random.uniform(0.8, 1.2, 50)
})
product_data = pd.DataFrame({
'sku': ['SKU001'],
'is_new_product': [False],
'season_factor': [1.1]
})
allocation_engine = RetailAllocationEngine(store_data, product_data)
# Allocate 5000 units of SKU001
allocation_result = allocation_engine.allocate_sku_to_stores(
sku='SKU001',
total_available_units=5000,
allocation_method='proportional'
)
print(f"Total allocated: {allocation_result['allocated_units'].sum()} units")
print(f"Average service level: {allocation_result['service_level'].mean():.1%}")
print(f"\nTop 5 allocations:")
print(allocation_result.nlargest(5, 'allocated_units'))
尺码曲线优化
class SizeCurveOptimizer:
"""
Optimize size curves for apparel allocation
Different stores need different size distributions
"""
def __init__(self, historical_size_sales):
"""
Parameters:
- historical_size_sales: Historical sales by size and store
columns: ['store_id', 'sku', 'size', 'units_sold']
"""
self.size_sales = historical_size_sales
def calculate_store_size_profile(self, store_id, category='tops'):
"""
Calculate size distribution for a store
Returns percentage distribution across sizes
"""
store_sales = self.size_sales[
(self.size_sales['store_id'] == store_id) &
(self.size_sales['category'] == category)
]
if len(store_sales) == 0:
# No history - use national average
return self._get_national_average_curve(category)
# Calculate size distribution
size_totals = store_sales.groupby('size')['units_sold'].sum()
total_units = size_totals.sum()
size_distribution = (size_totals / total_units * 100).to_dict()
return size_distribution
def _get_national_average_curve(self, category):
"""Default national average size curve"""
# Typical size distributions (example for women's tops)
default_curves = {
'tops': {
'XS': 8,
'S': 22,
'M': 32,
'L': 24,
'XL': 10,
'XXL': 4
},
'bottoms': {
'0': 5,
'2': 10,
'4': 15,
'6': 18,
'8': 20,
'10': 15,
'12': 10,
'14': 7
}
}
return default_curves.get(category, {})
def allocate_with_size_curve(self, sku, store_id, total_units, category='tops'):
"""
Allocate units across sizes based on store's size profile
Returns unit allocation by size
"""
# Get store's size curve
size_curve = self.calculate_store_size_profile(store_id, category)
# Allocate units
size_allocation = {}
remaining_units = total_units
# Sort sizes by percentage (allocate largest first)
sorted_sizes = sorted(size_curve.items(), key=lambda x: x[1], reverse=True)
for size, pct in sorted_sizes:
allocated = min(
int(total_units * pct / 100),
remaining_units
)
size_allocation[size] = allocated
remaining_units -= allocated
# Distribute any remainder to most popular sizes
if remaining_units > 0:
for size, pct in sorted_sizes[:remaining_units]:
size_allocation[size] += 1
return size_allocation
def compare_size_curves(self, store_ids, category='tops'):
"""
Compare size curves across stores
Useful for identifying regional differences
"""
curves = []
for store_id in store_ids:
curve = self.calculate_store_size_profile(store_id, category)
curve['store_id'] = store_id
curves.append(curve)
curves_df = pd.DataFrame(curves)
return curves_df
def optimize_size_mix_for_pack(self, target_stores, pack_size=12,
category='tops'):
"""
Optimize pre-pack size mix
Find size distribution that works well across target stores
"""
# Calculate average size curve across target stores
all_curves = []
for store_id in target_stores:
curve = self.calculate_store_size_profile(store_id, category)
all_curves.append(curve)
# Average the curves
curves_df = pd.DataFrame(all_curves)
avg_curve = curves_df.mean()
# Allocate pack_size units using average curve
size_allocation = {}
remaining = pack_size
sorted_sizes = sorted(avg_curve.items(), key=lambda x: x[1], reverse=True)
for size, pct in sorted_sizes:
allocated = min(
max(1, int(pack_size * pct / 100)), # At least 1 of each
remaining
)
size_allocation[size] = allocated
remaining -= allocated
return size_allocation
# Example
historical_size_sales = pd.DataFrame({
'store_id': ['S001'] * 6 + ['S002'] * 6,
'category': ['tops'] * 12,
'size': ['XS', 'S', 'M', 'L', 'XL', 'XXL'] * 2,
'units_sold': [80, 220, 350, 240, 100, 40, 50, 180, 380, 280, 90, 20]
})
size_optimizer = SizeCurveOptimizer(historical_size_sales)
# Get size profile for store
size_profile = size_optimizer.calculate_store_size_profile('S001', 'tops')
print("Store S001 size profile:")
for size, pct in size_profile.items():
print(f" {size}: {pct:.1f}%")
# Allocate 60 units with size curve
size_allocation = size_optimizer.allocate_with_size_curve('SKU123', 'S001', 60, 'tops')
print(f"\nSize allocation for 60 units:")
print(size_allocation)
预包装与按店拣配策略
权衡分析：
class PrePackAnalyzer:
"""
Analyze trade-offs between pre-pack and pick-pack allocation
"""
def compare_strategies(self, num_skus, num_stores, avg_units_per_store):
"""
Compare operational costs and accuracy
Pre-pack: Fixed assortments, efficient but less accurate
Pick-pack: Custom per store, accurate but labor-intensive
"""
# Pre-pack strategy
prepack_labor_per_pack = 2 # minutes
prepack_packs_needed = num_stores
prepack_labor_minutes = prepack_packs_needed * prepack_labor_per_pack
prepack_labor_cost = prepack_labor_minutes / 60 * 25 # $25/hour
prepack_accuracy = 0.75 # 75% match to store needs
# Pick-pack strategy
pickpack_labor_per_unit = 0.5 # minutes
pickpack_total_units = num_stores * avg_units_per_store
pickpack_labor_minutes = pickpack_total_units * pickpack_labor_per_unit
pickpack_labor_cost = pickpack_labor_minutes / 60 * 25
pickpack_accuracy = 0.95 # 95% match to store needs
# Calculate expected sales impact
# Better accuracy = higher sell-through, lower markdowns
avg_selling_price = 50
total_revenue_potential = pickpack_total_units * avg_selling_price
prepack_sellthrough = 0.70 # Lower due to mismatches
pickpack_sellthrough = 0.85 # Higher due to better matching
prepack_revenue = total_revenue_potential * prepack_sellthrough
pickpack_revenue = total_revenue_potential * pickpack_sellthrough
prepack_markdowns = total_revenue_potential * (1 - prepack_sellthrough) * 0.5
pickpack_markdowns = total_revenue_potential * (1 - pickpack_sellthrough) * 0.5
return {
'pre_pack': {
'labor_cost': prepack_labor_cost,
'accuracy': prepack_accuracy,
'expected_revenue': prepack_revenue,
'markdowns': prepack_markdowns,
'net_benefit': prepack_revenue - prepack_labor_cost - prepack_markdowns
},
'pick_pack': {
'labor_cost': pickpack_labor_cost,
'accuracy': pickpack_accuracy,
'expected_revenue': pickpack_revenue,
'markdowns': pickpack_markdowns,
'net_benefit': pickpack_revenue - pickpack_labor_cost - pickpack_markdowns
}
}
def recommend_strategy(self, num_skus, num_stores, product_type):
"""
Recommend pre-pack or pick-pack
Factors:
- Product type (fashion vs. basic)
- Store count
- Store diversity
"""
comparison = self.compare_strategies(num_skus, num_stores, avg_units_per_store=20)
prepack_net = comparison['pre_pack']['net_benefit']
pickpack_net = comparison['pick_pack']['net_benefit']
# Decision rules
if product_type == 'fashion' and pickpack_net > prepack_net * 1.1:
return {
'recommendation': 'Pick-Pack',
'reason': 'Fashion products benefit from customization',
'incremental_benefit': pickpack_net - prepack_net
}
elif num_stores > 500 and prepack_net > pickpack_net * 0.9:
return {
'recommendation': 'Pre-Pack',
'reason': 'Large store count makes pick-pack prohibitive',
'cost_savings': prepack_net - pickpack_net
}
elif product_type == 'basics':
return {
'recommendation': 'Pre-Pack',
'reason': 'Basic products have consistent demand across stores',
'efficiency_gain': 'Significant'
}
else:
return {
'recommendation': 'Pick-Pack',
'reason': 'Better accuracy justifies additional cost',
'incremental_benefit': pickpack_net - prepack_net
}
# Example
analyzer = PrePackAnalyzer()
comparison = analyzer.compare_strategies(num_skus=100, num_stores=200, avg_units_per_store=20)
print("Pre-Pack Strategy:")
print(f" Labor cost: ${comparison['pre_pack']['labor_cost']:,.0f}")
print(f" Net benefit: ${comparison['pre_pack']['net_benefit']:,.0f}")
print("\nPick-Pack Strategy:")
print(f" Labor cost: ${comparison['pick_pack']['labor_cost']:,.0f}")
print(f" Net benefit: ${comparison['pick_pack']['net_benefit']:,.0f}")
recommendation = analyzer.recommend_strategy(100, 200, 'fashion')
print(f"\nRecommendation: {recommendation['recommendation']}")
print(f"Reason: {recommendation['reason']}")
工具和库
Python 库
优化：
pulp、pyomo：用于分货优化的线性规划scipy.optimize：优化算法ortools：Google OR-Tools
聚类与 ML：
scikit-learn：K-means 聚类、分类scipy：统计分析pandas：数据处理
可视化：
matplotlib、seaborn：绘图plotly：交互式可视化geopandas：地理可视化
商业软件
分货与规划：
- Blue Yonder Luminate：零售规划套件
- o9 Solutions：数字化规划平台
- SAP Integrated Business Planning：企业规划
- Oracle Retail：商品管理与分货
- Aptos Retail：分货优化
专业工具：
- Infor Allocation：零售分货
- JustEnough：商品组合与分货
- TXT Retail：时尚商品分货
- 板Assortment Optimization by SAS：数据分析驱动的分货
常见挑战与解决方案
挑战：新品分配（无历史数据）
问题：
- 新品没有销售历史
- 难以预测各门店的需求
- 有库存过剩或缺货的风险
解决方案：
- 使用类似产品作为参照（风格、价格、品类相似）
- 在测试门店试点上新
- 以较小数量开始，快速补货
- 使用预测属性（颜色、风格、价格）
- 监测早期销售速度并进行调整
- 对门店进行聚类，每个集群测试一家门店
挑战：尺码/颜色失衡
问题：
- 某些尺码/颜色很快售罄
- 其他尺码/颜色则滞销
- 滞销颜色有降价风险
解决方案：
- 优化本地尺码比例
- 分析地区颜色偏好
- 小批量测试流行色
- 核心色备足，时尚色少量备货
- 通过门店间调拨解决失衡
- 根据早期售罄率动态分配
挑战：门店容量限制
问题：
- 小型门店无法容纳完整商品组合
- 库存超出后仓/卖场空间
- 空间拥挤会影响顾客体验
解决方案：
- 为小型门店精选商品组合
- 按规模对门店分级并相应调整
- 畅销商品采用越库配送（减少后仓库存）
- 频繁小批量配送，而非批量配送
- 缺货时提供数字化无限货架
- 按门店业态配置商品组合
挑战：地区需求差异
问题：
- 统一的分配方案行不通
- 气候差异（佛罗里达州卖大衣？）
- 人口特征差异
- 文化偏好差异
解决方案：
- 按地理区域聚类
- 制定基于气候的分配规则
- 按人口特征细分
- 部分品类由当地采购团队决定
- 制定区域分配计划
- 按区域测试并总结经验
挑战：库存过剩与降价
问题：
- 分配的库存过多
- 售罄率低
- 为清库存而大幅降价
解决方案：
- 初始分配从保守开始（根据需求追加订货）
- 每周监测售罄率
- 及早采取降价措施
- 门店间调拨
- 分配至奥特莱斯门店
- 改进需求预测
输出格式
配货计划报告
执行摘要：
- 产品：2025 年秋季服装系列
- 配货总量：125,000 件，涵盖 850 个 SKU
- 门店网络：185 家门店（40 家 A 级、55 家 B 级、90 家 C 级）
- 配货方式：基于需求，并进行门店聚类
- 预计正价售罄率：78%
门店集群概况：
| 集群 | 门店数 | 特征 | 平均每周销售额 | 配货策略 |
|---|---|---|---|---|
| 1 - 大型城市高收入地区 | 25 | 市中心旗舰店，高收入客群 | $85K | 全系列，充足库存 |
| 2 - 郊区购物中心 | 60 | 区域购物中心，家庭客群 | $45K | 核心系列，中等库存深度 |
| 3 - 温暖气候地区 | 40 | 南部各州，全年适销 | $38K | 调整季节性，选择较轻薄的款式 |
| 4 - 小型门店 | 35 | 街边商业中心、乡村地区 | $22K | 精选系列，少量库存 |
| 5 - 奥特莱斯 | 25 | 以清仓为主 | $32K | 积压库存接收门店 |
按等级划分的配货摘要：
| 等级 | 门店数 | 配货件数 | 平均每店件数 | SKU 覆盖率 | 库存供应周数 |
|---|---|---|---|---|---|
| A | 40 | 48,000 (38%) | 1,200 | 100% | 8 周 |
| B | 55 | 45,000 (36%) | 818 | 85% | 6 周 |
| C | 90 | 32,000 (26%) | 356 | 60% | 4 周 |
尺码配比（示例：女装上衣）：
| 门店集群 | XS | S | M | L | XL | XXL | 每店总件数 |
|---|---|---|---|---|---|---|---|
| 城市高收入地区 | 12% | 24% | 30% | 22% | 9% | 3% | 120 件 |
| 郊区 | 8% | 22% | 32% | 24% | 10% | 4% | 80 件 |
| 小型门店 | 6% | 20% | 34% | 26% | 11% | 3% | 35 件 |
销量前 10 家门店——配货示例：
| 门店 | 等级 | 集群 | 总件数 | 主推 SKU | 件数 | 预计销售额 |
|---|---|---|---|---|---|---|
| S001 - 纽约旗舰店 | A | 城市 | 1,450 | Jacket-001 | 85 | $124K |
| S005 - 洛杉矶 Westfield | A | 城市 | 1,320 | Dress-042 | 78 | $108K |
| S012 - 芝加哥 | A | 城市 | 1,280 | Top-015 | 72 | $98K |
| S023 - 波士顿 | B | 郊区 | 950 | Sweater-008 | 58 | $65K |
| S034 - 亚特兰大 | B | 温暖地区 | 880 | Shorts-011 | 52 | $58K |
财务影响：
| 指标 | 数值 |
|---|---|
| 配货总件数 | 125,000 |
| 正价零售价值 | $6,250,000 |
| 预计正价售罄率 | 78% |
| 预计正价销售收入 | $4,875,000 |
| 预计折价销售收入 | $687,500 |
| 预计总收入 | $5,562,500 |
| 预计折价率 | 22% |
| 目标折价率 | <25% |
风险评估：
-
高风险 SKU（预测置信度低）
- 45 个时尚前卫 SKU
- 保守配货（按需求量的 60% 配货）
- 计划：监控前 2 周的销售情况，必要时追加配货
-
区域风险
- 东北部：寒冷天气提前到来的风险
- 缓解措施：预留 10% 缓冲库存，以便快速重新配货
-
新品风险
- 120 个 SKU 为新款
- 缓解措施：在测试门店试销，并基于类似产品进行配货
行动计划：
| 周 | 行动 | 责任方 |
|---|---|---|
| 第 1 周 | 完成配货计划，并向门店传达 | 配货团队 |
| 第 2 周 | DC 接收库存，进行预包装/拣货作业 | DC 运营团队 |
| 第 3 周 | 向门店发货，开始跟踪在途货物 | 物流团队 |
| 第 4 周 | 门店收货，进行商品陈列 | 门店运营团队 |
| 第 5 周 | 监控售罄率，识别畅销和滞销商品 | 计划团队 |
| 第 6 周 | 针对库存失衡执行重新配货 | 配货团队 |
要提出的问题
如果需要更多背景信息：
- 你要配货的产品类别是什么？
- 你的门店网络有多少家门店？
- 是否已定义门店集群/等级？
- 这是新品还是补货？
- 历史售罄率是多少？
- 目标折价率是多少？
- 你们使用预包装还是按单拣货包装？
- 目前使用什么配货系统/流程？
- 最大的挑战是什么？（折价、缺货、复杂度）
相关技能
- retail-replenishment：初始分配后的持续补货
- markdown-optimization：滞销商品的 Markdown 策略
- demand-forecasting：用于分配规划的需求预测
- seasonal-planning：季节性采购规划
- inventory-optimization：安全库存和服务水平
- warehouse-slotting-optimization：优化配送中心货位配置以提升拣货效率
- supply-chain-analytics：绩效指标和报告
评论
支持 Markdown