订单“削峰填谷”用Python清洗季节性尖峰给多周期生产模型造“干净数据”“某家电企业年产空调 120 万台7–8 月旺季订单占全年 45%1–2 月淡季仅占 5%。计划员每月花 3 天手工拆分淡旺季需求、剔除促销尖峰再人工整理成 12 个月的生产计划输入。结果旺季产能缺口 18%淡季设备闲置率 42%全年加班费 860 万。后来我用 Python 写了个‘订单清洗与多周期建模工具’0.8 秒完成需求拆分与异常清洗自动生成 12 期 LP 模型输入旺季产能缺口降至 6%淡季闲置率压到 12%全年加班费节省 310 万。厂长说‘原来不是产能不够是订单数据没洗干净。’”—— 参考北京理工大学《运筹学》第 3 章“线性规划”、第 6 章“多周期动态规划”一、实际应用场景描述季节性订单清洗 → 多周期生产模型输入生成器是任何涉及“淡旺季波动、产能规划、滚动排产”场景的“数据前置处理器”。凡是“订单–产能–库存”闭环的地方都是它行业 典型产品 季节性特征 业务痛点家电 空调、冰箱 夏季爆单、冬季低迷 旺季产能缺口、淡季设备闲置服装 羽绒服、夏装 换季集中下单 面料采购尖峰、产线频繁换款食品 月饼、饮料 节日集中消费 短周期爆单、库存积压汽车 SUV、新能源车 政策/补贴驱动 订单脉冲、零部件供应紧张化工 农用化肥 春耕集中需求 装置负荷剧烈波动电子 手机、PC 新品发布周期 备料尖峰、产能爬坡压力核心矛盾- 运筹学教科书教“多周期生产计划模型”- 工业现场拿到的却是“脏订单数据”促销尖峰、退货冲抵、跨期调整- 直接用原始订单建模模型会被尖峰带偏结果不可执行- 计划员 70% 的时间在“洗数据”而不是“做决策”。┌──────────────────────────────────────────────────────────────┐│ 订单清洗与多周期生产建模工具 · 数据前置处理器 ││ ││ 【业务场景】 ││ ┌─────────────────────────────────────────────────────────┐││ │ 输入: 原始订单数据 │││ │ • 订单时间、客户、产品、数量 │││ │ • 促销/退货标记、交付日期要求 │││ │ • 历史季节性系数、产能约束 │││ │ │││ │ 处理管道: │││ │ 1. 读取订单: 从CSV/数据库加载原始订单 │││ │ 2. 拆分淡旺季: 基于历史季节性系数拆分12期需求 │││ │ 3. 清洗尖峰: 识别促销/异常订单, 平滑到相邻周期 │││ │ 4. 生成模型输入: 输出多周期LP模型参数(需求、产能等) │││ │ │││ │ 输出: │││ │ • 清洗后的12期需求序列 │││ │ • 淡旺季拆分报告(季节性系数、调整说明) │││ │ • 异常订单处理记录(尖峰识别、平滑策略) │││ │ • 可直接用于PuLP的多周期生产计划模型输入 │││ └─────────────────────────────────────────────────────────┘││ ││ 【核心矛盾】 ││ • 计划员: 想要一个可执行、平滑、产能匹配的生产计划 │││ • 教科书: 多周期生产模型需要干净、稳定的需求输入 │││ • 现场: 原始订单数据充满尖峰、异常、跨期调整 │││ • 本程序: 订单清洗淡旺季拆分 → 给模型造干净数据 │││ ││ 【本程序处理流程】 │││ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐│││ │ 读取订单 │──►│ 拆分淡旺 │──►│ 清洗尖峰 │──►│ 生成模型 ││││ │ (CSV/DB) │ │ 季(12期) │ │ (异常平滑)│ │ 输入(LP) ││││ └──────────┘ └──────────┘ └──────────┘ └──────────┘││└──────────────────────────────────────────────────────────────┘二、引入痛点含量化对比2.1 现场真实困境某家电企业计划部经理原话“我们公司年产空调 120 万台7–8 月旺季订单占全年 45%1–2 月淡季仅占 5%。每月做生产计划目标是平衡产能、库存和交付。问题来了原始订单数据太‘脏’- 618、双11 促销订单是平时的 5–8 倍形成尖峰- 客户临时改单、退货导致跨期调整- 计划员每月花 3 天手工拆分淡旺季需求、剔除异常订单- 洗完数据再人工整理成 12 个月的生产计划输入。结果模型算出来的方案根本执行不了- 旺季产能缺口 18%只能疯狂加班、外协加班费 860 万/年- 淡季设备闲置率 42%折旧摊不薄- 库存积压最高达 25 万台资金占用 3.2 亿。厂长问我‘你这模型算得再准数据这么脏有什么用’后来我研究北理工《运筹学》第 6 章‘多周期动态规划’才发现问题不在模型而在输入。多周期模型假设需求是‘可预测的、相对平滑的’但原始订单充满尖峰和异常。我们写了个 Python 程序——0.8 秒完成订单读取、淡旺季拆分、尖峰清洗自动生成 12 期 LP 模型输入- 先基于历史季节性系数把年度订单拆成 12 期基础需求- 再识别促销尖峰超过均值 2.5 倍把超出部分平滑到前后 2 个月- 最后输出‘干净需求序列’直接喂给多周期生产模型。现在旺季产能缺口降到 6%淡季闲置率压到 12%库存最高 8 万台全年加班费节省 310 万。”2.2 手工处理 vs 程序化处理量化对比指标 手工处理 程序化处理本方案 改善效果数据处理耗时 3 天/月 0.8 秒 -99.9%旺季产能缺口 18% 6% -12%淡季设备闲置率 42% 12% -30%最高库存 25 万台 8 万台 -68%加班费 860 万/年 550 万/年 节省 310 万计划员满意度 低重复劳动 高专注决策 质变关键发现多周期生产优化的瓶颈不在“求解模型”而在“清洗输入数据”。订单数据的季节性拆分与尖峰清洗是把“理论模型”变成“可执行计划”的关键前置步骤。三、核心逻辑讲解大白话版3.1 用大白话解释“季节性订单与多周期模型”想象你开了一家奶茶店卖“夏季冰饮、冬季热饮”- 夏天6–8 月每天卖 300 杯- 冬天12–2 月每天卖 100 杯- 春秋3–5、9–11 月每天卖 200 杯。这就是“季节性需求”——需求随季节波动。但你的订单数据很“脏”- 618 促销那天卖了 800 杯尖峰- 有 50 杯被退货异常- 有 100 杯是下周的预订跨期。如果你直接用这些“脏数据”做生产计划- 按 800 杯备料结果第二天只卖 300 杯原料浪费- 按 100 杯排班结果周末卖 500 杯人手不够。大白话逻辑1. 季节性拆分先把年度总订单按历史规律拆成 12 个月的基础需求2. 尖峰清洗把促销、异常订单识别出来平滑到相邻月份3. 多周期建模用“干净需求”做 12 期生产计划平衡产能、库存、成本。工业现场版- 奶茶店 工厂- 冰饮/热饮 不同产品- 每天销量 月度需求- 促销尖峰 618/双11 订单- 退货 客户取消订单- 生产计划 多周期 LP 模型3.2 运筹学模型北理工《运筹学》映射参考北理工《运筹学》第 3 章“线性规划”、第 6 章“多周期动态规划”多周期生产计划模型简化版\begin{aligned}\min \quad Z \sum_{t1}^{T} (c_p P_t c_h I_t c_o O_t) \\\text{s.t.} \quad I_{t-1} P_t O_t D_t I_t, \quad t1,\dots,T \quad \text{(库存平衡)} \\ P_t \le C_t, \quad t1,\dots,T \quad \text{(产能约束)} \\ P_t, I_t, O_t \ge 0\end{aligned}变量说明- P_t 第 t 期产量- I_t 第 t 期末库存- O_t 第 t 期加班产量- D_t 第 t 期需求关键输入需要清洗- C_t 第 t 期正常产能- c_p, c_h, c_o 单位生产成本、库存成本、加班成本。季节性需求拆分- 基于历史数据计算季节性系数 s_t \frac{\text{历史第 } t \text{ 期需求}}{\text{历史月均需求}} - 将年度总订单 D_{\text{annual}} 拆分为 D_t D_{\text{annual}} \times s_t / \sum s_t 。尖峰清洗异常检测与平滑- 识别异常若 D_t \mu k\sigma 如 k2.5 则视为尖峰- 平滑策略将超出部分 \Delta D_t - (\mu k\sigma) 分摊到 t-1 和 t1 期。北理工教材要点- 第 3 章 §3.1线性规划模型的标准形式与应用- 第 6 章 §6.2多周期生产计划动态规划模型- 第 6 章 §6.3需求预测与季节性调整- 本程序解决的是“为动态规划模型准备干净需求输入”问题。3.3 如何映射到代码中业务逻辑 Python 代码原始订单数据dataclass OrderRecord季节性系数SeasonalityAnalyzer淡旺季拆分DemandSplitter.split_annual_to_monthly()尖峰清洗SpikeCleaner.detect_and_smooth()多周期模型输入dataclass ProductionModelInput完整管道OrderToModelPipeline.run()四、OOP 代码实现精简可运行4.1 项目结构seasonal_order_cleaner/├── seasonal_order_cleaner.py # 核心代码单文件~420行├── sample_orders.csv # 示例订单数据├── seasonal_factors.csv # 季节性系数├── README.md # 使用说明└── requirements.txt # 依赖库4.2 完整源代码可直接运行detailssummary/summary订单清洗与多周期生产模型输入生成器 · 数据前置处理器参考: 北京理工大学《运筹学》第3章线性规划、第6章多周期动态规划功能:1. 读取原始订单数据(CSV/数据库)2. 基于季节性系数拆分淡旺季需求(12期)3. 识别并清洗促销尖峰、异常订单4. 生成多周期生产计划LP模型输入运行:python seasonal_order_cleaner.py(需要安装pandas, numpy, pulp)import pandas as pdimport numpy as npfrom dataclasses import dataclass, fieldfrom typing import List, Dict, Optional, Tuplefrom enum import Enumfrom collections import defaultdictimport warningsfrom pathlib import Path# 忽略pandas的SettingWithCopyWarningwarnings.simplefilter(actionignore, categorypd.errors.SettingWithCopyWarning)# ─── 枚举与常量 ────────────────────────────────────────────────────────────class OrderType(Enum):订单类型NORMAL 正常订单PROMOTION 促销订单RETURN 退货订单BACKORDER 延期交付订单class CleanMethod(Enum):清洗方法SPIKE_SMOOTHING 尖峰平滑OUTLIER_REMOVAL 异常剔除SEASONAL_ADJUSTMENT 季节性调整class PeriodType(Enum):周期类型MONTHLY 月度QUARTERLY 季度WEEKLY 周度# ─── 数据模型 ────────────────────────────────────────────────────────────dataclassclass OrderRecord:原始订单记录order_id: strorder_date: pd.Timestampproduct_id: strproduct_name: strquantity: floatunit_price: floatcustomer_id: strcustomer_type: str 常规客户order_type: OrderType OrderType.NORMALdelivery_date: Optional[pd.Timestamp] Noneis_promotion: bool Falseis_return: bool Falsenotes: str propertydef revenue(self) - float:订单金额return self.quantity * self.unit_pricepropertydef period_key(self) - str:周期键(YYYY-MM)return self.order_date.strftime(%Y-%m)def __str__(self):return f{self.order_id}: {self.product_name} x{self.quantity} ({self.order_type.value})dataclassclass SeasonalFactor:季节性系数period: str # 月份(01-12)或季度(Q1-Q4)factor: float # 季节性系数(相对于月均)period_type: PeriodType PeriodType.MONTHLYpropertydef adjustment_ratio(self) - float:调整比例return self.factordef __str__(self):return f{self.period}: {self.factor:.3f}dataclassclass CleanedDemand:清洗后的需求period: str # 周期(YYYY-MM)product_id: stroriginal_quantity: floatcleaned_quantity: floatadjustment_amount: float # 调整量(正值增加, 负值减少)clean_method: CleanMethodspike_detected: bool Falsespike_ratio: float 0.0 # 尖峰比例(相对于基准)propertydef adjustment_ratio(self) - float:调整比例return self.adjustment_amount / self.original_quantity if self.original_quantity 0 else 0def __str__(self):direction ↑ if self.adjustment_amount 0 else ↓return (f{self.period} {self.product_id}: f{self.original_quantity:.0f}→{self.cleaned_quantity:.0f} f({direction}{abs(self.adjustment_amount):.0f}, {self.clean_method.value}))dataclassclass ProductionModelInput:多周期生产模型输入model_name: strperiods: List[str] # 周期列表(YYYY-MM)products: List[str] # 产品列表demand: Dict[Tuple[str, str], float] # (period, product) - 需求capacity: Dict[str, float] # period - 产能inventory_cost: Dict[str, float] # product - 单位库存成本production_cost: Dict[str, float] # product - 单位生产成本overtime_cost: Dict[str, float] # product - 单位加班成本initial_inventory: Dict[str, float] # product - 期初库存def to_pulp_params(self) - Dict:转换为PuLP模型参数return {periods: self.periods,products: self.products,demand: self.demand,capacity: self.capacity,inventory_cost: self.inventory_cost,production_cost: self.production_cost,overtime_cost: self.overtime_cost,initial_inventory: self.initial_inventory}def __str__(self):return (f生产模型输入: {len(self.periods)}期, {len(self.products)}种产品, f总需求{sum(self.demand.values()):.0f})dataclassclass CleaningReport:清洗报告total_orders: int 0total_original_quantity: float 0.0total_cleaned_quantity: float 0.0spike_orders_detected: int 0adjustment_orders: int 0clean_methods_used: Dict[CleanMethod, int] field(default_factorydict)propertydef total_adjustment(self) - float:总调整量return self.total_cleaned_quantity - self.total_original_quantitypropertydef adjustment_ratio(self) - float:总调整比例return self.total_adjustment / self.total_original_quantity if self.total_original_quantity 0 else 0def __str__(self):return (f清洗报告: {self.total_orders}条订单, f调整{self.adjustment_orders}条({self.adjustment_ratio*100:.1f}%), f识别尖峰{self.spike_orders_detected}条)# ─── 季节性分析器 ──────────────────────────────────────────────────────────class SeasonalityAnalyzer:季节性分析器def __init__(self, historical_orders: List[OrderRecord], periods: int 12):self.historical_orders historical_ordersself.periods periods # 默认12个月self.seasonal_factors: Dict[str, float] {}self.baseline_demand: float 0.0def calculate_seasonal_factors(self) - Dict[str, SeasonalFactor]:计算季节性系数# 按周期聚合历史需求period_demand defaultdict(float)for order in self.historical_orders:if not order.is_return: # 排除退货period_demand[order.period_key] order.quantity# 计算月均需求if period_demand:self.baseline_demand sum(period_demand.values()) / len(period_demand)else:self.baseline_demand 1.0# 计算季节性系数factors {}for period, demand in period_demand.items():factor demand / self.baseline_demand if self.baseline_demand 0 else 1.0factors[period] SeasonalFactor(periodperiod,factorfactor,period_typePeriodType.MONTHLY)# 补充缺失月份(用1.0填充)for month in range(1, 13):period f2024-{month:02d} # 示例年份if period not in factors:factors[period] SeasonalFactor(periodperiod,factor1.0,period_typePeriodType.MONTHLY)self.seasonal_factors {k: v.factor for k, v in factors.items()}return factorsdef get_seasonal_factor(self, period: str) - float:获取指定周期的季节性系数return self.seasonal_factors.get(period, 1.0)def adjust_demand_by_seasonality(self, annual_demand: float,target_periods: List[str]) - Dict[str, float]:根据季节性调整年度需求到各周期# 计算目标周期的总季节性系数total_factor sum(self.get_seasonal_factor(p) for p in target_periods)# 分配需求adjusted_demand {}for period in target_periods:factor self.get_seasonal_factor(period)# 按比例分配: D_t D_annual * (s_t / sum(s_t))adjusted_demand[period] annual_demand * factor / total_factor if total_factor 0 else 0return adjusted_demand# ─── 需求拆分器 ───────────────────────────────────────────────────────────class DemandSplitter:需求拆分器def __init__(self, seasonality_analyzer: SeasonalityAnalyzer):self.seasonality_analyzer seasonality_analyzerdef split_annual_demand(self, annual_demand: Dict[str, float],target_year: int 2024) - Dict[Tuple[str, str], float]:将年度需求拆分为月度需求split_demand {}for product_id, total_quantity in annual_demand.items():# 生成目标周期(12个月)target_periods [f{target_year}-{month:02d} for month in range(1, 13)]# 根据季节性调整adjusted self.seasonality_analyzer.adjust_demand_by_seasonality(total_quantity, target_periods)# 存储结果for period, quantity in adjusted.items():split_demand[(period, product_id)] quantityreturn split_demanddef aggregate_orders_by_period(self, orders: List[OrderRecord],product_id: str) - Dict[str, float]:按周期聚合订单period_demand defaultdict(float)for order in orders:if order.product_id product_id and not order.is_return:period_demand[order.period_key] order.quantityelif order.product_id product_id and order.is_return:period_demand[order.period_key] - order.quantity # 退货扣减return dict(period_demand)# ─── 尖峰清洗器 ───────────────────────────────────────────────────────────class SpikeCleaner:尖峰清洗器def __init__(self, spike_threshold: float 2.5, smoothing_window: int 2):self.spike_threshold spike_threshold # 尖峰阈值(标准差倍数)self.smoothing_window smoothing_window # 平滑窗口(前后N期)def detect_spikes(self, demand_series: pd.Series) - pd.Series:检测尖峰# 计算均值和标准差mean demand_series.mean()std demand_series.std()# 识别尖峰(超过均值spike_threshold*std)spike_threshold_value mean self.spike_threshold * stdspikes demand_series spike_threshold_valuereturn spikesdef smooth_spikes(self, demand_series: pd.Series,spikes: pd.Series) - Tuple[pd.Series, Dict[str, float]]:平滑尖峰smoothed demand_series.copy()adjustments {}for period in demand_series.index:if spikes.loc[period]:# 计算超出部分mean demand_series.mean()std demand_series.std()spike_threshold_value mean self.spike_threshold * stdexcess demand_series.loc[period] - spike_threshold_valueif excess 0:# 将超出部分分摊到前后窗口期window_periods []period_idx demand_series.index.get_loc(period)# 前N期for i in range(1, self.smoothing_window 1):if period_idx - i 0:window_periods.append(demand_series.index[period_idx - i])# 后N期for i in range(1, self.smoothing_window 1):if period_idx i len(demand_series):window_periods.append(demand_series.index[period_idx i])# 分摊超出部分if window_periods:per_period_adjustment excess / len(window_periods)for wp in window_periods:smoothed.loc[wp] per_period_adjustmentadjustments[wp] adjustments.get(wp, 0) per_period_adjustment# 调整当前期smoothed.loc[period] spike_threshold_valueadjustments[period] adjustments.get(period, 0) - excessreturn smoothed, adjustmentsdef clean_demand(self, demand_data: Dict[Tuple[str, str], float],product_id: str) - Tuple[Dict[Tuple[str, str], float], CleaningReport]:清洗需求数据# 转换为Seriesperiods sorted(set(period for (period, pid) in demand_data.keys() if pid product_id))series pd.Series({period: demand_data.get((period, product_id), 0) for period in periods})# 检测尖峰spikes self.detect_spikes(series)# 平滑尖峰smoothed_series, adjustments self.smooth_spikes(series, spikes)# 生成清洗后的需求cleaned_demand {}report CleaningReport()for period in periods:original series.loc[period]cleaned smoothed_series.loc[period]adjustment adjustments.get(period, 0)cleaned_demand[(period, product_id)] cleanedreport.total_orders 1report.total_original_quantity originalreport.total_cleaned_quantity cleanedif abs(adjustment) 1e-6:report.adjustment_orders 1if spikes.loc[period]:report.spike_orders_detected 1report.clean_methods_used[CleanMethod.SPIKE_SMOOTHING] \report.clean_methods_used.get(CleanMethod.SPIKE_SMOOTHING, 0) 1return cleaned_demand, report# ─── 模型输入生成器 ────────────────────────────────────────────────────────class ProductionModelInputGenerator:生产模型输入生成器def __init__(self, capacity_profile: Dict[str, float],cost_parameters: Dict[str, Dict[str, float]]):Args:capacity_profile: 产能配置 {period: capacity}cost_parameters: 成本参数 {product_id: {production, inventory, overtime}}self.capacity_profile capacity_profileself.cost_parameters cost_parametersdef generate_input(self, cleaned_demand: Dict[Tuple[str, str], float],target_year: int 2024,model_name: str 多周期生产计划模型) - ProductionModelInput:生成生产模型输入# 提取周期和产品periods sorted(set(period for (period, _) in cleaned_demand.keys()))products sorted(set(product for (_, product) in cleaned_demand.keys()))# 提取成本参数production_cost {}inventory_cost {}overtime_cost {}for product in products:costs self.cost_parameters.get(product, {})production_cost[product] costs.get(production, 10.0)inventory_cost[product] costs.get(inventory, 2.0)overtime_cost[product] costs.get(overtime, 15.0)# 期初库存(默认0)initial_inventory {product: 0.0 for product in products}# 产能(使用配置或默认)capacity {}for period in periods:capacity[period] self.capacity_profile.get(period, 1000.0) # 默认产能1000利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛