项目3：库存分析与销量预测，我用Python让仓库不再积压也不断货

2026-03-01

大家好，我是正在实战各种AI项目的程序员晚枫。

欢迎来到第三个实战项目！这次我们要解决一个让供应链同学头疼的问题：库存管理。

库存太多，资金积压、仓储成本高；库存太少，断货丢单、客户流失。怎么找到平衡点？答案是：用数据预测未来销量，科学备货。

项目背景

需求场景

你是某零售公司的供应链分析师，面临这些痛点：

热销品经常断货，损失销售额
滞销品堆积如山，占用资金和仓库
采购凭经验拍脑袋，没有数据支撑

目标产出

建立销量预测模型，输出补货建议和安全库存策略。

准备数据

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

# 生成模拟销售数据（带季节性和趋势）
np.random.seed(42)

def generate_sales_data(start_date='2022-01-01', periods=730):
    """生成带趋势和季节性的销售数据"""
    dates = pd.date_range(start=start_date, periods=periods, freq='D')
    
    # 基础销量
    base_sales = 100
    
    # 趋势：逐年增长15%
    trend = np.linspace(0, 50, periods)
    
    # 季节性：周末高、工作日低；夏季高、冬季低
    seasonal = []
    for i, date in enumerate(dates):
        # 星期效应（周末+30%）
        weekend_boost = 30 if date.weekday() >= 5 else 0
        # 月份效应（夏季+20%，冬季-10%）
        month_factor = {6: 20, 7: 25, 8: 20, 12: -10, 1: -15}.get(date.month, 0)
        seasonal.append(weekend_boost + month_factor)
    
    # 随机波动
    noise = np.random.normal(0, 15, periods)
    
    # 促销 spikes（每月一次）
    promotions = np.zeros(periods)
    promo_days = np.random.choice(periods, size=24, replace=False)
    promotions[promo_days] = np.random.randint(50, 150, size=24)
    
    # 组合所有因素
    sales = base_sales + trend + np.array(seasonal) + noise + promotions
    sales = np.maximum(sales, 10).astype(int)  # 确保非负
    
    df = pd.DataFrame({
        '日期': dates,
        '销量': sales,
        '是否促销': promotions > 0
    })
    
    return df

# 生成数据
df = generate_sales_data()
df.to_csv('daily_sales.csv', index=False, encoding='utf-8-sig')
print(f"生成了 {len(df)} 天的销售数据")
print(f"平均日销量: {df['销量'].mean():.0f}")
print(f"总销量: {df['销量'].sum():,}")

数据探索与特征工程

class InventoryAnalyzer:
    def __init__(self, data_path):
        self.df = pd.read_csv(data_path, encoding='utf-8-sig')
        self.df['日期'] = pd.to_datetime(self.df['日期'])
        self.df.set_index('日期', inplace=True)
        print(f"加载了 {len(self.df)} 天的数据")
    
    def create_features(self):
        """创建时间特征"""
        df = self.df.copy()
        
        # 基础时间特征
        df['年'] = df.index.year
        df['月'] = df.index.month
        df['日'] = df.index.day
        df['星期'] = df.index.weekday
        df['是否周末'] = (df['星期'] >= 5).astype(int)
        df['季度'] = df.index.quarter
        df['一年中的第几天'] = df.index.dayofyear
        
        # 滞后特征（过去N天的销量）
        for lag in [1, 7, 14, 30]:
            df[f'销量_lag_{lag}'] = df['销量'].shift(lag)
        
        # 滚动统计特征
        for window in [7, 14, 30]:
            df[f'销量_ma_{window}'] = df['销量'].shift(1).rolling(window=window).mean()
            df[f'销量_std_{window}'] = df['销量'].shift(1).rolling(window=window).std()
        
        # 同比环比
        df['销量_同比'] = df['销量'].shift(365)
        df['销量_环比'] = df['销量'].shift(30)
        
        self.df_features = df.dropna()
        return self.df_features
    
    def analyze_patterns(self):
        """分析销售模式"""
        print("\n【销售模式分析】")
        
        # 按星期分析
        weekday_avg = self.df.groupby(self.df.index.weekday)['销量'].mean()
        print("\n星期平均销量:")
        days = ['周一', '周二', '周三', '周四', '周五', '周六', '周日']
        for i, day in enumerate(days):
            print(f"  {day}: {weekday_avg[i]:.0f}")
        
        # 按月分析
        monthly_avg = self.df.groupby(self.df.index.month)['销量'].mean()
        print("\n月度平均销量:")
        for month, avg in monthly_avg.items():
            print(f"  {month}月: {avg:.0f}")
        
        # 促销效果
        promo_effect = self.df.groupby('是否促销')['销量'].mean()
        print(f"\n促销效果:")
        print(f"  非促销日: {promo_effect[False]:.0f}")
        print(f"  促销日: {promo_effect[True]:.0f}")
        print(f"  提升幅度: {(promo_effect[True]/promo_effect[False]-1)*100:.1f}%")

销量预测模型

from sklearn.ensemble import RandomForestRegressor
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_absolute_error, mean_squared_error
import warnings
warnings.filterwarnings('ignore')

class SalesForecaster:
    def __init__(self, analyzer):
        self.analyzer = analyzer
        self.models = {}
    
    def prepare_data(self, forecast_days=30):
        """准备训练和测试数据"""
        df = self.analyzer.df_features.copy()
        
        # 特征列（排除原始销量和目标变量）
        feature_cols = [col for col in df.columns if col not in ['销量', '是否促销']]
        
        # 划分训练集和测试集
        train_size = len(df) - forecast_days
        self.train_df = df.iloc[:train_size]
        self.test_df = df.iloc[train_size:]
        
        self.X_train = self.train_df[feature_cols]
        self.y_train = self.train_df['销量']
        self.X_test = self.test_df[feature_cols]
        self.y_test = self.test_df['销量']
        self.feature_cols = feature_cols
        
        print(f"训练集: {len(self.train_df)} 天")
        print(f"测试集: {len(self.test_df)} 天")
        
        return self.X_train, self.y_train, self.X_test, self.y_test
    
    def train_models(self):
        """训练多个模型"""
        # 线性回归（基线）
        lr = LinearRegression()
        lr.fit(self.X_train, self.y_train)
        self.models['LinearRegression'] = lr
        
        # 随机森林
        rf = RandomForestRegressor(n_estimators=100, random_state=42, n_jobs=-1)
        rf.fit(self.X_train, self.y_train)
        self.models['RandomForest'] = rf
        
        print("\n模型训练完成")
        return self.models
    
    def evaluate(self):
        """评估模型性能"""
        results = {}
        
        print("\n【模型评估结果】")
        print("-" * 50)
        
        for name, model in self.models.items():
            y_pred = model.predict(self.X_test)
            
            mae = mean_absolute_error(self.y_test, y_pred)
            rmse = np.sqrt(mean_squared_error(self.y_test, y_pred))
            mape = np.mean(np.abs((self.y_test - y_pred) / self.y_test)) * 100
            
            results[name] = {
                'MAE': mae,
                'RMSE': rmse,
                'MAPE': mape,
                'predictions': y_pred
            }
            
            print(f"\n{name}:")
            print(f"  MAE: {mae:.2f}")
            print(f"  RMSE: {rmse:.2f}")
            print(f"  MAPE: {mape:.2f}%")
        
        self.results = results
        return results
    
    def get_feature_importance(self):
        """获取特征重要性（仅随机森林）"""
        if 'RandomForest' in self.models:
            rf = self.models['RandomForest']
            importance = pd.DataFrame({
                'feature': self.feature_cols,
                'importance': rf.feature_importances_
            }).sort_values('importance', ascending=False)
            
            print("\n【特征重要性 Top 10】")
            print(importance.head(10).to_string(index=False))
            return importance

可视化分析

import matplotlib.pyplot as plt

class InventoryVisualizer:
    def __init__(self, analyzer, forecaster):
        self.analyzer = analyzer
        self.forecaster = forecaster
        plt.style.use('seaborn-v0_8')
    
    def plot_sales_trend(self):
        """销量趋势图"""
        fig, axes = plt.subplots(2, 1, figsize=(14, 8))
        
        # 完整趋势
        axes[0].plot(self.analyzer.df.index, self.analyzer.df['销量'], 
                    alpha=0.7, linewidth=0.8)
        axes[0].set_title('Daily Sales Trend (Full History)', fontsize=14)
        axes[0].set_ylabel('Sales Volume')
        axes[0].grid(True, alpha=0.3)
        
        # 最近90天细节
        recent = self.analyzer.df.tail(90)
        axes[1].plot(recent.index, recent['销量'], marker='o', markersize=3)
        axes[1].set_title('Daily Sales Trend (Last 90 Days)', fontsize=14)
        axes[1].set_ylabel('Sales Volume')
        axes[1].grid(True, alpha=0.3)
        
        plt.tight_layout()
        plt.savefig('sales_trend.png', dpi=300, bbox_inches='tight')
        plt.show()
    
    def plot_seasonal_patterns(self):
        """季节性模式图"""
        fig, axes = plt.subplots(1, 2, figsize=(14, 5))
        
        # 星期模式
        weekday_avg = self.analyzer.df.groupby(self.analyzer.df.index.weekday)['销量'].mean()
        days = ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun']
        bars1 = axes[0].bar(days, weekday_avg.values, color='steelblue')
        axes[0].set_title('Average Sales by Weekday', fontsize=14)
        axes[0].set_ylabel('Average Sales')
        for bar in bars1:
            height = bar.get_height()
            axes[0].text(bar.get_x() + bar.get_width()/2., height,
                        f'{height:.0f}', ha='center', va='bottom')
        
        # 月度模式
        monthly_avg = self.analyzer.df.groupby(self.analyzer.df.index.month)['销量'].mean()
        months = ['Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun',
                 'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec']
        bars2 = axes[1].bar(months, monthly_avg.values, color='coral')
        axes[1].set_title('Average Sales by Month', fontsize=14)
        axes[1].set_ylabel('Average Sales')
        axes[1].tick_params(axis='x', rotation=45)
        
        plt.tight_layout()
        plt.savefig('seasonal_patterns.png', dpi=300, bbox_inches='tight')
        plt.show()
    
    def plot_forecast_comparison(self):
        """预测对比图"""
        test_dates = self.forecaster.test_df.index
        actual = self.forecaster.y_test.values
        
        plt.figure(figsize=(14, 6))
        plt.plot(test_dates, actual, label='Actual', marker='o', linewidth=2)
        
        colors = ['orange', 'green']
        for (name, result), color in zip(self.forecaster.results.items(), colors):
            plt.plot(test_dates, result['predictions'], 
                    label=f'{name} (MAPE: {result["MAPE"]:.1f}%)',
                    marker='s', alpha=0.7, color=color)
        
        plt.title('Sales Forecast Comparison', fontsize=16)
        plt.xlabel('Date')
        plt.ylabel('Sales Volume')
        plt.legend()
        plt.grid(True, alpha=0.3)
        plt.tight_layout()
        plt.savefig('forecast_comparison.png', dpi=300, bbox_inches='tight')
        plt.show()

库存策略建议

def generate_inventory_strategy(analyzer, forecaster, current_stock=500, lead_time=7):
    """
    生成库存策略建议
    
    参数:
        current_stock: 当前库存量
        lead_time: 补货提前期（天）
    """
    # 使用最佳模型的预测结果
    best_model = min(forecaster.results.items(), key=lambda x: x[1]['MAPE'])
    predictions = best_model[1]['predictions']
    
    # 计算关键指标
    historical_sales = analyzer.df['销量']
    avg_daily_sales = historical_sales.mean()
    std_daily_sales = historical_sales.std()
    max_daily_sales = historical_sales.max()
    
    # 预测期总需求
    forecast_total = predictions.sum()
    forecast_avg = predictions.mean()
    
    print("="*60)
    print("【智能库存分析报告】")
    print("="*60)
    
    print(f"\n📊 历史销售概况:")
    print(f"  平均日销量: {avg_daily_sales:.0f}")
    print(f"  销量标准差: {std_daily_sales:.0f}")
    print(f"  最高日销量: {max_daily_sales}")
    print(f"  变异系数(CV): {std_daily_sales/avg_daily_sales:.2f}")
    
    print(f"\n📈 未来{len(predictions)}天预测:")
    print(f"  预测总需求: {forecast_total:.0f}")
    print(f"  预测日均: {forecast_avg:.0f}")
    print(f"  预测区间: {predictions.min():.0f} - {predictions.max():.0f}")
    
    # 安全库存计算（95%服务水平）
    z_score = 1.65  # 95%置信度
    safety_stock = z_score * std_daily_sales * np.sqrt(lead_time)
    reorder_point = forecast_avg * lead_time + safety_stock
    
    print(f"\n📦 库存策略建议:")
    print(f"  当前库存: {current_stock}")
    print(f"  补货提前期: {lead_time}天")
    print(f"  安全库存: {safety_stock:.0f} (95%服务水平)")
    print(f"  再订货点: {reorder_point:.0f}")
    
    # 库存状态判断
    days_of_supply = current_stock / forecast_avg
    print(f"\n⚠️ 库存健康度:")
    print(f"  可供应天数: {days_of_supply:.1f}天")
    
    if current_stock < reorder_point:
        shortage = reorder_point - current_stock
        print(f"  状态: 🔴 库存不足")
        print(f"  建议: 立即补货 {shortage:.0f} 件")
    elif days_of_supply > 30:
        excess = current_stock - reorder_point
        print(f"  状态: 🟡 库存偏高")
        print(f"  建议: 暂停补货，考虑促销清库存")
    else:
        print(f"  状态: 🟢 库存健康")
        print(f"  建议: 维持当前策略，持续监控")
    
    # 补货计划
    print(f"\n📝 未来7天补货计划:")
    cumulative_demand = 0
    for i, pred in enumerate(predictions[:7], 1):
        cumulative_demand += pred
        remaining = current_stock - cumulative_demand
        status = "✅ 充足" if remaining > safety_stock else "⚠️ 需关注" if remaining > 0 else "🔴 缺货"
        print(f"  Day{i}: 预测需求{pred:.0f}, 预计剩余{remaining:.0f} {status}")
    
    return {
        'safety_stock': safety_stock,
        'reorder_point': reorder_point,
        'days_of_supply': days_of_supply,
        'forecast': predictions
    }

主程序

def main():
    """主程序"""
    print("="*60)
    print("库存分析与销量预测系统")
    print("="*60)
    
    # 1. 加载和分析数据
    print("\n[1/5] 加载销售数据...")
    analyzer = InventoryAnalyzer('daily_sales.csv')
    
    print("\n[2/5] 特征工程...")
    analyzer.create_features()
    analyzer.analyze_patterns()
    
    # 2. 训练预测模型
    print("\n[3/5] 训练预测模型...")
    forecaster = SalesForecaster(analyzer)
    forecaster.prepare_data(forecast_days=30)
    forecaster.train_models()
    forecaster.evaluate()
    forecaster.get_feature_importance()
    
    # 3. 可视化
    print("\n[4/5] 生成可视化图表...")
    visualizer = InventoryVisualizer(analyzer, forecaster)
    visualizer.plot_sales_trend()
    visualizer.plot_seasonal_patterns()
    visualizer.plot_forecast_comparison()
    
    # 4. 库存策略
    print("\n[5/5] 生成库存策略...")
    strategy = generate_inventory_strategy(analyzer, forecaster, 
                                          current_stock=800, lead_time=7)
    
    # 导出预测结果
    forecast_df = pd.DataFrame({
        '日期': forecaster.test_df.index,
        '实际销量': forecaster.y_test.values,
        '预测销量': forecaster.results['RandomForest']['predictions'].round(0)
    })
    forecast_df.to_csv('sales_forecast.csv', index=False, encoding='utf-8-sig')
    print("\n  ✓ 预测结果已保存: sales_forecast.csv")
    
    print("\n" + "="*60)
    print("分析完成！")
    print("="*60)

if __name__ == "__main__":
    main()

项目总结

学到的技能

✅ 时间序列特征工程
✅ 销量预测建模
✅ 安全库存计算
✅ 库存健康度评估

核心公式

指标	公式	说明
安全库存	Z × σ × √L	Z是服务水平系数，σ是需求标准差，L是提前期
再订货点	日均需求×L + 安全库存	低于此值就触发补货
库存周转	销售成本 / 平均库存	越高越好