目录移动平均法计算示例方法选择Python代码30天示例数据版12个月示例数据版365天示例数据版运行环境jupyter notebook (python 3.12.7)移动平均法在时间序列分析中简单移动平均SMA、加权移动平均WMA、指数移动平均EMA和赫尔移动平均HMA是四种常用的移动平均方法它们在计算逻辑、灵敏度、滞后性以及适用场景上各有特点。适用场景方法适用场景SMA 简单移动平均(Simple Moving Average)长期趋势分析、数据平稳且噪声多的场景如宏观经济指标。WMA 加权移动平均(Weighted Moving Average)需平衡近期与远期数据的场景如季节性商品需求预测。EMA 指数移动平均(Exponential Moving Average)高频交易、快速响应趋势变化如股票/加密货币价格跟踪。HMA 赫尔移动平均(Hull Moving Average)短期交易信号、需最小化滞后的场景如日内交易或突破策略。平稳销量 → SMA明显趋势 → WMA频繁波动 → EMA既要快又要稳 → HMA计算公式方法计算公式核心思想SMA对过去 n 期数据取算术平均权重均等。WMA线性递减权重如 win−i1近期数据权重更高。EMA指数递减权重αn12强调近期数据且无限回溯历史。HMA通过双重平滑减少滞后性结合短期和长期趋势。计算示例假设某药品过去7天的出库量单位箱为[10,12,15,8,20,18,25]第7天为最新数据反映突发需求方法计算过程结果解释SMA(158201825)/586/517.2忽略早期数据10,12但滞后于突发需求。WMA(5×254×183×202×81×15)/(54321)250/1516.67近期权重高但被早期低值8,15拖累。EMA假设 α0.3上期EMA16: 0.3×250.7×167.511.218.7更贴近最新数据25响应最快。HMA1. 计算WMA(5) 16.672. 计算WMA(3) 21.673. HMA 2×16.67−21.6711.67双重平滑后结果异常需谨慎使用。SMA17.2反映过去5天平均出库量适合评估稳态需求但未捕捉到第7天的突发需求25箱。WMA16.67比SMA更重视近期数据但因第4天异常低值8箱拉低结果可能低估实际需求。EMA18.7对第7天的25箱响应明显适合动态调整补货计划但需警惕噪声如第4天的8箱干扰。HMA11.67因窗口大小和权重问题产生反常识结果不推荐直接用于库存决策需参数调优。方法选择1.数据频率影响若数据为小时级如冷链物流EMA或HMA更适用月度数据则优先SMA。2.组合策略用SMA判断长期趋势EMA触发补货信号人工排除异常值如盘点误差。3.验证方法通过MAE/MSE对比各方法在历史数据上的误差选择最优模型场景与方法匹配稳态库存监控如常规药品SMA 安全库存阈值如SMA ± 2倍标准差。季节性备货如流感疫苗WMA权重近期远期 历史同期数据修正。应急药品调度EMAα0.2−0.4 人工复核突发订单。低库存预警可尝试HMA但需验证参数如 n3 时 HMA13.33更合理。参数优化方向窗口大小n医药物流通常建议 n7周周期或 n30月周期。EMA的平滑系数α突发需求多时取 α0.3−0.5稳定时取 α0.1−0.2。在医药物流中EMA突发需求和WMA季节性通常是首选而SMA适合稳态分析HMA需谨慎验证。实际应用中应结合业务逻辑如药品有效期、供应商交期调整参数而非依赖单一数学结果。Python代码30天示例数据版import numpy as np import pandas as pd import matplotlib.pyplot as plt from matplotlib.dates import DateFormatter # 设置随机种子以确保结果可重复 np.random.seed(42) # 1. 生成示例数据30天出库量 def generate_sample_data(): base_demand np.random.randint(10, 20, size30) base_demand[15] 40 # 第15天突发需求 base_demand[25] 5 # 第25天库存短缺 dates pd.date_range(start2023-01-01, periods30) return pd.DataFrame({Date: dates, Outbound: base_demand}) # 2. 计算四种移动平均 def calculate_moving_averages(data, window7): df data.copy() # SMA df[SMA] df[Outbound].rolling(windowwindow).mean() # WMA线性递减权重 weights np.arange(1, window1) # e.g. [1,2,3,4,5] for window5 def wma(x, w): return np.dot(x, w) / w.sum() # 使用固定窗口大小的WMA df[WMA] df[Outbound].rolling(windowwindow).apply(lambda x: wma(x, weights[:len(x)])) # EMA alpha 2 / (window 1) # 平滑系数 df[EMA] df[Outbound].ewm(alphaalpha, adjustFalse).mean() # HMA half_window max(1, window // 2) # 确保至少为1 sqrt_window int(np.sqrt(window)) # 计算WMA(n/2) wma_half_weights np.arange(1, half_window1) wma_half df[Outbound].rolling(windowhalf_window).apply(lambda x: wma(x, wma_half_weights[:len(x)])) # 计算WMA(n) wma_full_weights np.arange(1, window1) wma_full df[Outbound].rolling(windowwindow).apply(lambda x: wma(x, wma_full_weights[:len(x)])) # 计算2*WMA(n/2) - WMA(n) df[HMA_temp] 2 * wma_half - wma_full # 对结果再取WMA(sqrt(n)) sqrt_weights np.arange(1, sqrt_window1) df[HMA] df[HMA_temp].rolling(windowsqrt_window).apply(lambda x: wma(x, sqrt_weights[:len(x)])) return df # 3. 可视化结果 def plot_results(df, window): plt.figure(figsize(14, 7)) # 绘制实际出库量 plt.plot(df[Date], df[Outbound], ko-, labelActual Outbound, markersize5, linewidth1) # 绘制四种移动平均 plt.plot(df[Date], df[SMA], b--, labelfSMA (n{window}), linewidth1.5) plt.plot(df[Date], df[WMA], g-., labelfWMA (n{window}), linewidth1.5) plt.plot(df[Date], df[EMA], r:, labelfEMA (n{window}), linewidth1.5) plt.plot(df[Date], df[HMA], c-, labelfHMA (n{window}), linewidth1.5) # 标记特殊事件 plt.axvline(xdf[Date][15], colororange, linestyle--, alpha0.5, labelDemand Spike (Day 15)) plt.axvline(xdf[Date][25], colorpurple, linestyle--, alpha0.5, labelStockout (Day 25)) # 图表格式设置 plt.title(Pharmaceutical Outbound: Moving Averages Comparison, fontsize14) plt.xlabel(Date, fontsize12) plt.ylabel(Outbound Quantity (Cases), fontsize12) # 日期格式 date_form DateFormatter(%m-%d) plt.gca().xaxis.set_major_formatter(date_form) plt.legend(fontsize10, bbox_to_anchor(1.05, 1), locupper left) plt.grid(True, linestyle--, alpha0.6) plt.tight_layout() plt.show() # 4. 主程序 def main(): # 生成数据 df generate_sample_data() # 计算移动平均默认窗口7天 window 7 df_ma calculate_moving_averages(df, window) # 打印最后10天数据 print(Last 10 days data with moving averages:) print(df_ma.tail(10).to_string(indexFalse)) # 可视化 plot_results(df_ma, window) if __name__ __main__: main()运行结果尝试不同窗口大小例如571015对比结果12个月示例数据版import numpy as np import pandas as pd import matplotlib.pyplot as plt from matplotlib.dates import DateFormatter # 设置随机种子以确保结果可重复 np.random.seed(42) # 1. 生成示例数据2022年1-12月出库量 def generate_sample_data(): # 基础需求模式年初低年中高年末回落 base_pattern [50, 55, 60, 70, 80, 90, 85, 75, 65, 60, 55, 50] # 添加随机波动 noise np.random.randint(-10, 10, size12) base_demand np.array(base_pattern) noise # 确保没有负值 base_demand np.clip(base_demand, 20, None) dates pd.date_range(start2022-01-01, end2022-12-31, freqMS) return pd.DataFrame({Date: dates, Outbound: base_demand}) # 2. 计算四种移动平均 def calculate_moving_averages(data, window3): df data.copy() # SMA df[SMA] df[Outbound].rolling(windowwindow).mean() # WMA线性递减权重 weights np.arange(1, window1) # e.g. [1,2,3] for window3 def wma(x, w): return np.dot(x, w) / w.sum() df[WMA] df[Outbound].rolling(windowwindow).apply(lambda x: wma(x, weights[:len(x)])) # EMA alpha 2 / (window 1) # 平滑系数 df[EMA] df[Outbound].ewm(alphaalpha, adjustFalse).mean() # HMA half_window max(1, window // 2) sqrt_window int(np.sqrt(window)) # 计算WMA(n/2) wma_half_weights np.arange(1, half_window1) wma_half df[Outbound].rolling(windowhalf_window).apply(lambda x: wma(x, wma_half_weights[:len(x)])) # 计算WMA(n) wma_full_weights np.arange(1, window1) wma_full df[Outbound].rolling(windowwindow).apply(lambda x: wma(x, wma_full_weights[:len(x)])) # 计算2*WMA(n/2) - WMA(n) df[HMA_temp] 2 * wma_half - wma_full # 对结果再取WMA(sqrt(n)) sqrt_weights np.arange(1, sqrt_window1) df[HMA] df[HMA_temp].rolling(windowsqrt_window).apply(lambda x: wma(x, sqrt_weights[:len(x)])) return df # 3. 预测2023年1月出库量 def predict_january(df, window3): # 使用最后window个月的数据进行预测 last_data df.tail(window) # SMA预测 sma_pred last_data[Outbound].mean() # WMA预测 weights np.arange(1, window1) wma_pred np.dot(last_data[Outbound], weights) / weights.sum() # EMA预测 alpha 2 / (window 1) ema_pred df[EMA].iloc[-1] * (1 - alpha) df[Outbound].iloc[-1] * alpha # HMA预测 hma_pred df[HMA].iloc[-1] # 简单使用最后一个HMA值 predictions { SMA: sma_pred, WMA: wma_pred, EMA: ema_pred, HMA: hma_pred } return predictions # 4. 可视化结果 def plot_results(df, window): plt.figure(figsize(14, 7)) # 绘制实际出库量 plt.plot(df[Date], df[Outbound], ko-, label2022 Actual Outbound, markersize5, linewidth1) # 绘制四种移动平均 plt.plot(df[Date], df[SMA], b--, labelfSMA (n{window}), linewidth1.5) plt.plot(df[Date], df[WMA], g-., labelfWMA (n{window}), linewidth1.5) plt.plot(df[Date], df[EMA], r:, labelfEMA (n{window}), linewidth1.5) plt.plot(df[Date], df[HMA], c-, labelfHMA (n{window}), linewidth1.5) # 图表格式设置 plt.title(Pharmaceutical Outbound (2022) Moving Averages, fontsize14) plt.xlabel(Month, fontsize12) plt.ylabel(Outbound Quantity (Cases), fontsize12) # 日期格式 date_form DateFormatter(%b) plt.gca().xaxis.set_major_formatter(date_form) plt.legend(fontsize10, bbox_to_anchor(1.05, 1), locupper left) plt.grid(True, linestyle--, alpha0.6) plt.tight_layout() plt.show() # 5. 主程序 def main(): # 生成2022年数据 df_2022 generate_sample_data() print(2022 Monthly Outbound Data:) print(df_2022.to_string(indexFalse)) # 计算移动平均窗口3个月 window 3 df_ma calculate_moving_averages(df_2022, window) # 预测2023年1月 predictions predict_january(df_ma, window) print(\n2023 January Predictions:) for method, value in predictions.items(): print(f{method}: {value:.1f} cases) # 可视化 plot_results(df_ma, window) if __name__ __main__: main()运行结果365天示例数据版import numpy as np import pandas as pd import matplotlib.pyplot as plt from matplotlib.dates import DateFormatter, MonthLocator # 设置随机种子以确保结果可重复 np.random.seed(42) # 1. 生成示例数据2022年每日出库量 def generate_daily_data(): # 生成日期范围2022年全年 dates pd.date_range(start2022-01-01, end2022-12-31, freqD) n_days len(dates) # 基础模式周周期工作日高周末低和月周期月中高峰 day_of_week np.array([d.dayofweek for d in dates]) # 周一0周日6 day_of_month np.array([d.day for d in dates]) day_of_year np.array([d.dayofyear for d in dates]) # 基础需求模式 base_demand 50 20 * np.sin(2 * np.pi * day_of_month / 30) # 月周期 base_demand - 15 * (day_of_week 5) # 周末减少 # 添加随机波动和季节性 noise np.random.normal(0, 10, sizen_days) seasonal 10 * np.sin(2 * np.pi * (day_of_year - 80) / 365) # 季节性波动 # 合成最终需求 daily_demand np.round(base_demand seasonal noise) daily_demand np.clip(daily_demand, 20, 120) # 确保在合理范围内 # 添加一些特殊事件节假日和促销 special_dates [ 2022-01-01, 2022-02-01, 2022-05-01, 2022-10-01, 2022-06-18, 2022-11-11, 2022-12-25 ] for date in special_dates: # 使用日期字符串直接比较 idx np.where(dates.astype(str) date)[0] if len(idx) 0: daily_demand[idx] daily_demand[idx] * 1.5 # 特殊日期需求增加50% return pd.DataFrame({Date: dates, Outbound: daily_demand}) # 2. 计算四种移动平均 def calculate_moving_averages(data, window30): df data.copy() # SMA df[SMA] df[Outbound].rolling(windowwindow).mean() # WMA线性递减权重 weights np.arange(1, window1) def wma(x, w): return np.dot(x, w) / w.sum() df[WMA] df[Outbound].rolling(windowwindow).apply( lambda x: wma(x, weights[:len(x)]), rawTrue) # EMA alpha 2 / (window 1) df[EMA] df[Outbound].ewm(alphaalpha, adjustFalse).mean() # HMA half_window max(1, window // 2) sqrt_window int(np.sqrt(window)) # 计算WMA(n/2) wma_half_weights np.arange(1, half_window1) wma_half df[Outbound].rolling(windowhalf_window).apply( lambda x: wma(x, wma_half_weights[:len(x)]), rawTrue) # 计算WMA(n) wma_full df[Outbound].rolling(windowwindow).apply( lambda x: wma(x, weights[:len(x)]), rawTrue) # 计算2*WMA(n/2) - WMA(n) df[HMA_temp] 2 * wma_half - wma_full # 对结果再取WMA(sqrt(n)) sqrt_weights np.arange(1, sqrt_window1) df[HMA] df[HMA_temp].rolling(windowsqrt_window).apply( lambda x: wma(x, sqrt_weights[:len(x)]), rawTrue) return df # 3. 预测2023年1月每日出库量 def predict_january_daily(df_2022, window30): # 使用最后window天的数据作为基础 last_window df_2022[Outbound].iloc[-window:].values # 生成2023年1月的日期 jan_dates pd.date_range(start2023-01-01, end2023-01-31, freqD) # 初始化预测数据框 predictions pd.DataFrame({ Date: jan_dates, DayOfWeek: [d.dayofweek for d in jan_dates], DayOfMonth: [d.day for d in jan_dates] }) # 计算各方法的基准预测值使用最后window天的移动平均值 last_sma df_2022[SMA].iloc[-1] last_wma df_2022[WMA].iloc[-1] last_ema df_2022[EMA].iloc[-1] last_hma df_2022[HMA].iloc[-1] # 添加周周期调整因子工作日/周末 weekday_adjustment np.where(predictions[DayOfWeek] 5, 1.1, 0.9) # 生成各方法的预测考虑周周期 predictions[SMA_Pred] last_sma * weekday_adjustment predictions[WMA_Pred] last_wma * weekday_adjustment predictions[EMA_Pred] last_ema * weekday_adjustment predictions[HMA_Pred] last_hma * weekday_adjustment # 对特殊日期如元旦进行调整 predictions.loc[predictions[Date] 2023-01-01, [SMA_Pred, WMA_Pred, EMA_Pred, HMA_Pred]] * 1.5 return predictions # 4. 可视化结果 def plot_results(df_2022, predictions, window): fig, (ax1, ax2) plt.subplots(2, 1, figsize(16, 12)) # 绘制2022年数据 ax1.plot(df_2022[Date], df_2022[Outbound], k-, label2022 Daily Outbound, linewidth0.5, alpha0.7) ax1.plot(df_2022[Date], df_2022[SMA], b-, labelfSMA ({window} days)) ax1.plot(df_2022[Date], df_2022[EMA], r-, labelfEMA ({window} days)) ax1.set_title(2022 Daily Outbound Moving Averages, fontsize14) ax1.set_ylabel(Outbound Quantity (Cases), fontsize12) ax1.legend(fontsize10) ax1.grid(True, linestyle--, alpha0.6) # 格式化x轴 ax1.xaxis.set_major_locator(MonthLocator()) ax1.xaxis.set_major_formatter(DateFormatter(%b)) # 绘制2023年1月预测 ax2.plot(predictions[Date], predictions[SMA_Pred], b--, labelSMA Prediction) ax2.plot(predictions[Date], predictions[WMA_Pred], g-., labelWMA Prediction) ax2.plot(predictions[Date], predictions[EMA_Pred], r:, labelEMA Prediction) ax2.plot(predictions[Date], predictions[HMA_Pred], c-, labelHMA Prediction) # 标记周末 weekends predictions[predictions[DayOfWeek] 5] for date in weekends[Date]: ax2.axvspan(date - pd.Timedelta(days0.5), date pd.Timedelta(days0.5), colorgray, alpha0.1) ax2.set_title(2023 January Daily Predictions, fontsize14) ax2.set_ylabel(Predicted Outbound, fontsize12) ax2.legend(fontsize10) ax2.grid(True, linestyle--, alpha0.6) # 格式化x轴 ax2.xaxis.set_major_locator(MonthLocator()) ax2.xaxis.set_major_formatter(DateFormatter(%d-%b)) plt.tight_layout() plt.show() # 5. 主程序 def main(): # 生成2022年每日数据 df_2022 generate_daily_data() print(2022 Daily Data (sample):) print(df_2022.head(10).to_string(indexFalse)) # 计算移动平均窗口30天 window 30 df_ma calculate_moving_averages(df_2022, window) # 预测2023年1月每日出库量 predictions predict_january_daily(df_ma, window) print(\n2023 January Predictions (sample):) print(predictions.head(10).to_string(indexFalse)) # 可视化 plot_results(df_ma, predictions, window) # 保存预测结果 predictions.to_csv(january_2023_predictions.csv, indexFalse) print(\nPredictions saved to january_2023_predictions.csv) if __name__ __main__: main()运行结果调整了下载文件格式的版本import numpy as np import pandas as pd import matplotlib.pyplot as plt from matplotlib.dates import DateFormatter, MonthLocator from tabulate import tabulate import warnings import os # Ignore warnings warnings.filterwarnings(ignore, categoryUserWarning) # Set random seed for reproducibility np.random.seed(42) # 1. Generate sample data (daily outbound for 2022) def generate_daily_data(): dates pd.date_range(start2022-01-01, end2022-12-31, freqD) n_days len(dates) day_of_week np.array([d.dayofweek for d in dates]) day_of_month np.array([d.day for d in dates]) day_of_year np.array([d.dayofyear for d in dates]) base_demand 50 20 * np.sin(2 * np.pi * day_of_month / 30) base_demand - 15 * (day_of_week 5) noise np.random.normal(0, 10, sizen_days) seasonal 10 * np.sin(2 * np.pi * (day_of_year - 80) / 365) daily_demand np.round(base_demand seasonal noise) daily_demand np.clip(daily_demand, 20, 120) special_dates [ 2022-01-01, 2022-02-01, 2022-05-01, 2022-10-01, 2022-06-18, 2022-11-11, 2022-12-25 ] for date in special_dates: idx np.where(dates.astype(str) date)[0] if len(idx) 0: daily_demand[idx] daily_demand[idx] * 1.5 return pd.DataFrame({Date: dates, Outbound: daily_demand}) # 2. Calculate four types of moving averages def calculate_moving_averages(data, window30): df data.copy() # SMA df[SMA] df[Outbound].rolling(windowwindow).mean() # WMA weights np.arange(1, window1) def wma(x, w): return np.dot(x, w) / w.sum() df[WMA] df[Outbound].rolling(windowwindow).apply( lambda x: wma(x, weights[:len(x)]), rawTrue) # EMA alpha 2 / (window 1) df[EMA] df[Outbound].ewm(alphaalpha, adjustFalse).mean() # HMA half_window max(1, window // 2) sqrt_window int(np.sqrt(window)) wma_half_weights np.arange(1, half_window1) wma_half df[Outbound].rolling(windowhalf_window).apply( lambda x: wma(x, wma_half_weights[:len(x)]), rawTrue) wma_full df[Outbound].rolling(windowwindow).apply( lambda x: wma(x, weights[:len(x)]), rawTrue) df[HMA_temp] 2 * wma_half - wma_full sqrt_weights np.arange(1, sqrt_window1) df[HMA] df[HMA_temp].rolling(windowsqrt_window).apply( lambda x: wma(x, sqrt_weights[:len(x)]), rawTrue) return df # 3. Predict daily outbound for January 2023 def predict_january_daily(df_2022, window30): jan_dates pd.date_range(start2023-01-01, end2023-01-31, freqD) predictions pd.DataFrame({ Date: jan_dates, DayOfWeek: [d.dayofweek for d in jan_dates], DayOfMonth: [d.day for d in jan_dates] }) last_sma df_2022[SMA].iloc[-1] last_wma df_2022[WMA].iloc[-1] last_ema df_2022[EMA].iloc[-1] last_hma df_2022[HMA].iloc[-1] weekday_adjustment np.where(predictions[DayOfWeek] 5, 1.1, 0.9) predictions[SMA_Pred] last_sma * weekday_adjustment predictions[WMA_Pred] last_wma * weekday_adjustment predictions[EMA_Pred] last_ema * weekday_adjustment predictions[HMA_Pred] last_hma * weekday_adjustment predictions.loc[predictions[Date] 2023-01-01, [SMA_Pred, WMA_Pred, EMA_Pred, HMA_Pred]] * 1.5 return predictions # 4. Display results as tables def display_results(df_2022, predictions): # Last 10 days of December 2022 dec_2022 df_2022[df_2022[Date].dt.month 12].tail(10).copy() dec_table dec_2022[[Date, Outbound, SMA, WMA, EMA, HMA]].copy() dec_table[Date] dec_table[Date].dt.strftime(%m-%d) # Format date consistently # January 2023 predictions pred_table predictions.copy() pred_table[Date] pred_table[Date].dt.strftime(%m-%d) # Format date consistently pred_table[Weekday] pred_table[DayOfWeek].map({ 0: Mon, 1: Tue, 2: Wed, 3: Thu, 4: Fri, 5: Sat, 6: Sun }) print(\nLast 10 days of December 2022 actual outbound and moving averages:) print(tabulate( dec_table.round(1), headers[Date, Outbound, SMA, WMA, EMA, HMA], tablefmtgrid, showindexFalse, floatfmt.1f )) print(\nJanuary 2023 daily outbound predictions (first 10 days):) print(tabulate( pred_table[[Date, Weekday, SMA_Pred, WMA_Pred, EMA_Pred, HMA_Pred]].head(10).round(1), headers[Date, Weekday, SMA, WMA, EMA, HMA], tablefmtgrid, showindexFalse, floatfmt.1f )) avg_pred pd.DataFrame({ Method: [SMA, WMA, EMA, HMA], Avg_Prediction: [ predictions[SMA_Pred].mean(), predictions[WMA_Pred].mean(), predictions[EMA_Pred].mean(), predictions[HMA_Pred].mean() ] }) print(\nAverage predictions for January by method:) print(tabulate( avg_pred.round(1), headerskeys, tablefmtgrid, showindexFalse, floatfmt.1f )) # 5. Visualize results def plot_results(df_2022, predictions, window30): plt.figure(figsize(14, 10)) ax1 plt.subplot(2, 1, 1) ax2 plt.subplot(2, 1, 2) # Subplot 1: 2022 actual data with moving averages ax1.plot(df_2022[Date], df_2022[Outbound], k-, labelActual Outbound, linewidth0.8, alpha0.7) ax1.plot(df_2022[Date], df_2022[SMA], b-, labelfSMA ({window} days), linewidth1.5) ax1.plot(df_2022[Date], df_2022[EMA], r-, labelfEMA ({window} days), linewidth1.5) ax1.set_title(2022 Daily Outbound with Moving Averages, fontsize14, pad20) ax1.set_ylabel(Outbound (units), fontsize12) ax1.legend(fontsize10, locupper left) ax1.grid(True, linestyle--, alpha0.6) ax1.xaxis.set_major_locator(MonthLocator()) ax1.xaxis.set_major_formatter(DateFormatter(%b)) # Subplot 2: January 2023 predictions ax2.plot(predictions[Date], predictions[SMA_Pred], b--, labelSMA Prediction, linewidth1.5) ax2.plot(predictions[Date], predictions[WMA_Pred], g-., labelWMA Prediction, linewidth1.5) ax2.plot(predictions[Date], predictions[EMA_Pred], r:, labelEMA Prediction, linewidth1.5) ax2.plot(predictions[Date], predictions[HMA_Pred], c-, labelHMA Prediction, linewidth1.5) weekends predictions[predictions[DayOfWeek] 5] for date in weekends[Date]: ax2.axvspan(date - pd.Timedelta(days0.5), date pd.Timedelta(days0.5), colorgray, alpha0.1) ax2.set_title(January 2023 Daily Outbound Predictions, fontsize14, pad20) ax2.set_ylabel(Predicted Outbound (units), fontsize12) ax2.legend(fontsize10, locupper left) ax2.grid(True, linestyle--, alpha0.6) ax2.xaxis.set_major_locator(MonthLocator()) ax2.xaxis.set_major_formatter(DateFormatter(%d-%b)) plt.tight_layout() # Save plot with error handling try: plot_path prediction_comparison.png plt.savefig(plot_path, dpi300, bbox_inchestight) except PermissionError: home_path os.path.expanduser(~) plot_path os.path.join(home_path, prediction_comparison.png) plt.savefig(plot_path, dpi300, bbox_inchestight) print(fCouldnt save plot to current directory. Saved to: {plot_path}) plt.show() # 6. Main program def main(): df_2022 generate_daily_data() window 30 df_ma calculate_moving_averages(df_2022, window) predictions predict_january_daily(df_ma, window) display_results(df_ma, predictions) plot_results(df_ma, predictions, window) try: # Prepare CSV output with visualization-consistent headers csv_output predictions.copy() csv_output[Date] csv_output[Date].dt.strftime(%m-%d) csv_output[Weekday] csv_output[DayOfWeek].map({ 0: Mon, 1: Tue, 2: Wed, 3: Thu, 4: Fri, 5: Sat, 6: Sun }) # Select and rename columns to match visualization csv_output csv_output[[ Date, Weekday, SMA_Pred, WMA_Pred, EMA_Pred, HMA_Pred ]].rename(columns{ SMA_Pred: SMA, WMA_Pred: WMA, EMA_Pred: EMA, HMA_Pred: HMA }) # Round values to 1 decimal place csv_output csv_output.round(1) # Try saving to current directory first try: csv_path january_2023_predictions.csv csv_output.to_csv(csv_path, indexFalse) print(f\nPredictions saved to {csv_path} in current directory) except PermissionError: # Fall back to users home directory if current directory fails home_path os.path.expanduser(~) csv_path os.path.join(home_path, january_2023_predictions.csv) csv_output.to_csv(csv_path, indexFalse) print(f\nCouldnt write to current directory. Predictions saved to: {csv_path}) except Exception as e: print(f\nError saving files: {str(e)}) print(Please check your directory permissions or specify a different output path.) if __name__ __main__: main()CSV/表格文件 Python | 移动平均法 销量预测
目录移动平均法计算示例方法选择Python代码30天示例数据版12个月示例数据版365天示例数据版运行环境jupyter notebook (python 3.12.7)移动平均法在时间序列分析中简单移动平均SMA、加权移动平均WMA、指数移动平均EMA和赫尔移动平均HMA是四种常用的移动平均方法它们在计算逻辑、灵敏度、滞后性以及适用场景上各有特点。适用场景方法适用场景SMA 简单移动平均(Simple Moving Average)长期趋势分析、数据平稳且噪声多的场景如宏观经济指标。WMA 加权移动平均(Weighted Moving Average)需平衡近期与远期数据的场景如季节性商品需求预测。EMA 指数移动平均(Exponential Moving Average)高频交易、快速响应趋势变化如股票/加密货币价格跟踪。HMA 赫尔移动平均(Hull Moving Average)短期交易信号、需最小化滞后的场景如日内交易或突破策略。平稳销量 → SMA明显趋势 → WMA频繁波动 → EMA既要快又要稳 → HMA计算公式方法计算公式核心思想SMA对过去 n 期数据取算术平均权重均等。WMA线性递减权重如 win−i1近期数据权重更高。EMA指数递减权重αn12强调近期数据且无限回溯历史。HMA通过双重平滑减少滞后性结合短期和长期趋势。计算示例假设某药品过去7天的出库量单位箱为[10,12,15,8,20,18,25]第7天为最新数据反映突发需求方法计算过程结果解释SMA(158201825)/586/517.2忽略早期数据10,12但滞后于突发需求。WMA(5×254×183×202×81×15)/(54321)250/1516.67近期权重高但被早期低值8,15拖累。EMA假设 α0.3上期EMA16: 0.3×250.7×167.511.218.7更贴近最新数据25响应最快。HMA1. 计算WMA(5) 16.672. 计算WMA(3) 21.673. HMA 2×16.67−21.6711.67双重平滑后结果异常需谨慎使用。SMA17.2反映过去5天平均出库量适合评估稳态需求但未捕捉到第7天的突发需求25箱。WMA16.67比SMA更重视近期数据但因第4天异常低值8箱拉低结果可能低估实际需求。EMA18.7对第7天的25箱响应明显适合动态调整补货计划但需警惕噪声如第4天的8箱干扰。HMA11.67因窗口大小和权重问题产生反常识结果不推荐直接用于库存决策需参数调优。方法选择1.数据频率影响若数据为小时级如冷链物流EMA或HMA更适用月度数据则优先SMA。2.组合策略用SMA判断长期趋势EMA触发补货信号人工排除异常值如盘点误差。3.验证方法通过MAE/MSE对比各方法在历史数据上的误差选择最优模型场景与方法匹配稳态库存监控如常规药品SMA 安全库存阈值如SMA ± 2倍标准差。季节性备货如流感疫苗WMA权重近期远期 历史同期数据修正。应急药品调度EMAα0.2−0.4 人工复核突发订单。低库存预警可尝试HMA但需验证参数如 n3 时 HMA13.33更合理。参数优化方向窗口大小n医药物流通常建议 n7周周期或 n30月周期。EMA的平滑系数α突发需求多时取 α0.3−0.5稳定时取 α0.1−0.2。在医药物流中EMA突发需求和WMA季节性通常是首选而SMA适合稳态分析HMA需谨慎验证。实际应用中应结合业务逻辑如药品有效期、供应商交期调整参数而非依赖单一数学结果。Python代码30天示例数据版import numpy as np import pandas as pd import matplotlib.pyplot as plt from matplotlib.dates import DateFormatter # 设置随机种子以确保结果可重复 np.random.seed(42) # 1. 生成示例数据30天出库量 def generate_sample_data(): base_demand np.random.randint(10, 20, size30) base_demand[15] 40 # 第15天突发需求 base_demand[25] 5 # 第25天库存短缺 dates pd.date_range(start2023-01-01, periods30) return pd.DataFrame({Date: dates, Outbound: base_demand}) # 2. 计算四种移动平均 def calculate_moving_averages(data, window7): df data.copy() # SMA df[SMA] df[Outbound].rolling(windowwindow).mean() # WMA线性递减权重 weights np.arange(1, window1) # e.g. [1,2,3,4,5] for window5 def wma(x, w): return np.dot(x, w) / w.sum() # 使用固定窗口大小的WMA df[WMA] df[Outbound].rolling(windowwindow).apply(lambda x: wma(x, weights[:len(x)])) # EMA alpha 2 / (window 1) # 平滑系数 df[EMA] df[Outbound].ewm(alphaalpha, adjustFalse).mean() # HMA half_window max(1, window // 2) # 确保至少为1 sqrt_window int(np.sqrt(window)) # 计算WMA(n/2) wma_half_weights np.arange(1, half_window1) wma_half df[Outbound].rolling(windowhalf_window).apply(lambda x: wma(x, wma_half_weights[:len(x)])) # 计算WMA(n) wma_full_weights np.arange(1, window1) wma_full df[Outbound].rolling(windowwindow).apply(lambda x: wma(x, wma_full_weights[:len(x)])) # 计算2*WMA(n/2) - WMA(n) df[HMA_temp] 2 * wma_half - wma_full # 对结果再取WMA(sqrt(n)) sqrt_weights np.arange(1, sqrt_window1) df[HMA] df[HMA_temp].rolling(windowsqrt_window).apply(lambda x: wma(x, sqrt_weights[:len(x)])) return df # 3. 可视化结果 def plot_results(df, window): plt.figure(figsize(14, 7)) # 绘制实际出库量 plt.plot(df[Date], df[Outbound], ko-, labelActual Outbound, markersize5, linewidth1) # 绘制四种移动平均 plt.plot(df[Date], df[SMA], b--, labelfSMA (n{window}), linewidth1.5) plt.plot(df[Date], df[WMA], g-., labelfWMA (n{window}), linewidth1.5) plt.plot(df[Date], df[EMA], r:, labelfEMA (n{window}), linewidth1.5) plt.plot(df[Date], df[HMA], c-, labelfHMA (n{window}), linewidth1.5) # 标记特殊事件 plt.axvline(xdf[Date][15], colororange, linestyle--, alpha0.5, labelDemand Spike (Day 15)) plt.axvline(xdf[Date][25], colorpurple, linestyle--, alpha0.5, labelStockout (Day 25)) # 图表格式设置 plt.title(Pharmaceutical Outbound: Moving Averages Comparison, fontsize14) plt.xlabel(Date, fontsize12) plt.ylabel(Outbound Quantity (Cases), fontsize12) # 日期格式 date_form DateFormatter(%m-%d) plt.gca().xaxis.set_major_formatter(date_form) plt.legend(fontsize10, bbox_to_anchor(1.05, 1), locupper left) plt.grid(True, linestyle--, alpha0.6) plt.tight_layout() plt.show() # 4. 主程序 def main(): # 生成数据 df generate_sample_data() # 计算移动平均默认窗口7天 window 7 df_ma calculate_moving_averages(df, window) # 打印最后10天数据 print(Last 10 days data with moving averages:) print(df_ma.tail(10).to_string(indexFalse)) # 可视化 plot_results(df_ma, window) if __name__ __main__: main()运行结果尝试不同窗口大小例如571015对比结果12个月示例数据版import numpy as np import pandas as pd import matplotlib.pyplot as plt from matplotlib.dates import DateFormatter # 设置随机种子以确保结果可重复 np.random.seed(42) # 1. 生成示例数据2022年1-12月出库量 def generate_sample_data(): # 基础需求模式年初低年中高年末回落 base_pattern [50, 55, 60, 70, 80, 90, 85, 75, 65, 60, 55, 50] # 添加随机波动 noise np.random.randint(-10, 10, size12) base_demand np.array(base_pattern) noise # 确保没有负值 base_demand np.clip(base_demand, 20, None) dates pd.date_range(start2022-01-01, end2022-12-31, freqMS) return pd.DataFrame({Date: dates, Outbound: base_demand}) # 2. 计算四种移动平均 def calculate_moving_averages(data, window3): df data.copy() # SMA df[SMA] df[Outbound].rolling(windowwindow).mean() # WMA线性递减权重 weights np.arange(1, window1) # e.g. [1,2,3] for window3 def wma(x, w): return np.dot(x, w) / w.sum() df[WMA] df[Outbound].rolling(windowwindow).apply(lambda x: wma(x, weights[:len(x)])) # EMA alpha 2 / (window 1) # 平滑系数 df[EMA] df[Outbound].ewm(alphaalpha, adjustFalse).mean() # HMA half_window max(1, window // 2) sqrt_window int(np.sqrt(window)) # 计算WMA(n/2) wma_half_weights np.arange(1, half_window1) wma_half df[Outbound].rolling(windowhalf_window).apply(lambda x: wma(x, wma_half_weights[:len(x)])) # 计算WMA(n) wma_full_weights np.arange(1, window1) wma_full df[Outbound].rolling(windowwindow).apply(lambda x: wma(x, wma_full_weights[:len(x)])) # 计算2*WMA(n/2) - WMA(n) df[HMA_temp] 2 * wma_half - wma_full # 对结果再取WMA(sqrt(n)) sqrt_weights np.arange(1, sqrt_window1) df[HMA] df[HMA_temp].rolling(windowsqrt_window).apply(lambda x: wma(x, sqrt_weights[:len(x)])) return df # 3. 预测2023年1月出库量 def predict_january(df, window3): # 使用最后window个月的数据进行预测 last_data df.tail(window) # SMA预测 sma_pred last_data[Outbound].mean() # WMA预测 weights np.arange(1, window1) wma_pred np.dot(last_data[Outbound], weights) / weights.sum() # EMA预测 alpha 2 / (window 1) ema_pred df[EMA].iloc[-1] * (1 - alpha) df[Outbound].iloc[-1] * alpha # HMA预测 hma_pred df[HMA].iloc[-1] # 简单使用最后一个HMA值 predictions { SMA: sma_pred, WMA: wma_pred, EMA: ema_pred, HMA: hma_pred } return predictions # 4. 可视化结果 def plot_results(df, window): plt.figure(figsize(14, 7)) # 绘制实际出库量 plt.plot(df[Date], df[Outbound], ko-, label2022 Actual Outbound, markersize5, linewidth1) # 绘制四种移动平均 plt.plot(df[Date], df[SMA], b--, labelfSMA (n{window}), linewidth1.5) plt.plot(df[Date], df[WMA], g-., labelfWMA (n{window}), linewidth1.5) plt.plot(df[Date], df[EMA], r:, labelfEMA (n{window}), linewidth1.5) plt.plot(df[Date], df[HMA], c-, labelfHMA (n{window}), linewidth1.5) # 图表格式设置 plt.title(Pharmaceutical Outbound (2022) Moving Averages, fontsize14) plt.xlabel(Month, fontsize12) plt.ylabel(Outbound Quantity (Cases), fontsize12) # 日期格式 date_form DateFormatter(%b) plt.gca().xaxis.set_major_formatter(date_form) plt.legend(fontsize10, bbox_to_anchor(1.05, 1), locupper left) plt.grid(True, linestyle--, alpha0.6) plt.tight_layout() plt.show() # 5. 主程序 def main(): # 生成2022年数据 df_2022 generate_sample_data() print(2022 Monthly Outbound Data:) print(df_2022.to_string(indexFalse)) # 计算移动平均窗口3个月 window 3 df_ma calculate_moving_averages(df_2022, window) # 预测2023年1月 predictions predict_january(df_ma, window) print(\n2023 January Predictions:) for method, value in predictions.items(): print(f{method}: {value:.1f} cases) # 可视化 plot_results(df_ma, window) if __name__ __main__: main()运行结果365天示例数据版import numpy as np import pandas as pd import matplotlib.pyplot as plt from matplotlib.dates import DateFormatter, MonthLocator # 设置随机种子以确保结果可重复 np.random.seed(42) # 1. 生成示例数据2022年每日出库量 def generate_daily_data(): # 生成日期范围2022年全年 dates pd.date_range(start2022-01-01, end2022-12-31, freqD) n_days len(dates) # 基础模式周周期工作日高周末低和月周期月中高峰 day_of_week np.array([d.dayofweek for d in dates]) # 周一0周日6 day_of_month np.array([d.day for d in dates]) day_of_year np.array([d.dayofyear for d in dates]) # 基础需求模式 base_demand 50 20 * np.sin(2 * np.pi * day_of_month / 30) # 月周期 base_demand - 15 * (day_of_week 5) # 周末减少 # 添加随机波动和季节性 noise np.random.normal(0, 10, sizen_days) seasonal 10 * np.sin(2 * np.pi * (day_of_year - 80) / 365) # 季节性波动 # 合成最终需求 daily_demand np.round(base_demand seasonal noise) daily_demand np.clip(daily_demand, 20, 120) # 确保在合理范围内 # 添加一些特殊事件节假日和促销 special_dates [ 2022-01-01, 2022-02-01, 2022-05-01, 2022-10-01, 2022-06-18, 2022-11-11, 2022-12-25 ] for date in special_dates: # 使用日期字符串直接比较 idx np.where(dates.astype(str) date)[0] if len(idx) 0: daily_demand[idx] daily_demand[idx] * 1.5 # 特殊日期需求增加50% return pd.DataFrame({Date: dates, Outbound: daily_demand}) # 2. 计算四种移动平均 def calculate_moving_averages(data, window30): df data.copy() # SMA df[SMA] df[Outbound].rolling(windowwindow).mean() # WMA线性递减权重 weights np.arange(1, window1) def wma(x, w): return np.dot(x, w) / w.sum() df[WMA] df[Outbound].rolling(windowwindow).apply( lambda x: wma(x, weights[:len(x)]), rawTrue) # EMA alpha 2 / (window 1) df[EMA] df[Outbound].ewm(alphaalpha, adjustFalse).mean() # HMA half_window max(1, window // 2) sqrt_window int(np.sqrt(window)) # 计算WMA(n/2) wma_half_weights np.arange(1, half_window1) wma_half df[Outbound].rolling(windowhalf_window).apply( lambda x: wma(x, wma_half_weights[:len(x)]), rawTrue) # 计算WMA(n) wma_full df[Outbound].rolling(windowwindow).apply( lambda x: wma(x, weights[:len(x)]), rawTrue) # 计算2*WMA(n/2) - WMA(n) df[HMA_temp] 2 * wma_half - wma_full # 对结果再取WMA(sqrt(n)) sqrt_weights np.arange(1, sqrt_window1) df[HMA] df[HMA_temp].rolling(windowsqrt_window).apply( lambda x: wma(x, sqrt_weights[:len(x)]), rawTrue) return df # 3. 预测2023年1月每日出库量 def predict_january_daily(df_2022, window30): # 使用最后window天的数据作为基础 last_window df_2022[Outbound].iloc[-window:].values # 生成2023年1月的日期 jan_dates pd.date_range(start2023-01-01, end2023-01-31, freqD) # 初始化预测数据框 predictions pd.DataFrame({ Date: jan_dates, DayOfWeek: [d.dayofweek for d in jan_dates], DayOfMonth: [d.day for d in jan_dates] }) # 计算各方法的基准预测值使用最后window天的移动平均值 last_sma df_2022[SMA].iloc[-1] last_wma df_2022[WMA].iloc[-1] last_ema df_2022[EMA].iloc[-1] last_hma df_2022[HMA].iloc[-1] # 添加周周期调整因子工作日/周末 weekday_adjustment np.where(predictions[DayOfWeek] 5, 1.1, 0.9) # 生成各方法的预测考虑周周期 predictions[SMA_Pred] last_sma * weekday_adjustment predictions[WMA_Pred] last_wma * weekday_adjustment predictions[EMA_Pred] last_ema * weekday_adjustment predictions[HMA_Pred] last_hma * weekday_adjustment # 对特殊日期如元旦进行调整 predictions.loc[predictions[Date] 2023-01-01, [SMA_Pred, WMA_Pred, EMA_Pred, HMA_Pred]] * 1.5 return predictions # 4. 可视化结果 def plot_results(df_2022, predictions, window): fig, (ax1, ax2) plt.subplots(2, 1, figsize(16, 12)) # 绘制2022年数据 ax1.plot(df_2022[Date], df_2022[Outbound], k-, label2022 Daily Outbound, linewidth0.5, alpha0.7) ax1.plot(df_2022[Date], df_2022[SMA], b-, labelfSMA ({window} days)) ax1.plot(df_2022[Date], df_2022[EMA], r-, labelfEMA ({window} days)) ax1.set_title(2022 Daily Outbound Moving Averages, fontsize14) ax1.set_ylabel(Outbound Quantity (Cases), fontsize12) ax1.legend(fontsize10) ax1.grid(True, linestyle--, alpha0.6) # 格式化x轴 ax1.xaxis.set_major_locator(MonthLocator()) ax1.xaxis.set_major_formatter(DateFormatter(%b)) # 绘制2023年1月预测 ax2.plot(predictions[Date], predictions[SMA_Pred], b--, labelSMA Prediction) ax2.plot(predictions[Date], predictions[WMA_Pred], g-., labelWMA Prediction) ax2.plot(predictions[Date], predictions[EMA_Pred], r:, labelEMA Prediction) ax2.plot(predictions[Date], predictions[HMA_Pred], c-, labelHMA Prediction) # 标记周末 weekends predictions[predictions[DayOfWeek] 5] for date in weekends[Date]: ax2.axvspan(date - pd.Timedelta(days0.5), date pd.Timedelta(days0.5), colorgray, alpha0.1) ax2.set_title(2023 January Daily Predictions, fontsize14) ax2.set_ylabel(Predicted Outbound, fontsize12) ax2.legend(fontsize10) ax2.grid(True, linestyle--, alpha0.6) # 格式化x轴 ax2.xaxis.set_major_locator(MonthLocator()) ax2.xaxis.set_major_formatter(DateFormatter(%d-%b)) plt.tight_layout() plt.show() # 5. 主程序 def main(): # 生成2022年每日数据 df_2022 generate_daily_data() print(2022 Daily Data (sample):) print(df_2022.head(10).to_string(indexFalse)) # 计算移动平均窗口30天 window 30 df_ma calculate_moving_averages(df_2022, window) # 预测2023年1月每日出库量 predictions predict_january_daily(df_ma, window) print(\n2023 January Predictions (sample):) print(predictions.head(10).to_string(indexFalse)) # 可视化 plot_results(df_ma, predictions, window) # 保存预测结果 predictions.to_csv(january_2023_predictions.csv, indexFalse) print(\nPredictions saved to january_2023_predictions.csv) if __name__ __main__: main()运行结果调整了下载文件格式的版本import numpy as np import pandas as pd import matplotlib.pyplot as plt from matplotlib.dates import DateFormatter, MonthLocator from tabulate import tabulate import warnings import os # Ignore warnings warnings.filterwarnings(ignore, categoryUserWarning) # Set random seed for reproducibility np.random.seed(42) # 1. Generate sample data (daily outbound for 2022) def generate_daily_data(): dates pd.date_range(start2022-01-01, end2022-12-31, freqD) n_days len(dates) day_of_week np.array([d.dayofweek for d in dates]) day_of_month np.array([d.day for d in dates]) day_of_year np.array([d.dayofyear for d in dates]) base_demand 50 20 * np.sin(2 * np.pi * day_of_month / 30) base_demand - 15 * (day_of_week 5) noise np.random.normal(0, 10, sizen_days) seasonal 10 * np.sin(2 * np.pi * (day_of_year - 80) / 365) daily_demand np.round(base_demand seasonal noise) daily_demand np.clip(daily_demand, 20, 120) special_dates [ 2022-01-01, 2022-02-01, 2022-05-01, 2022-10-01, 2022-06-18, 2022-11-11, 2022-12-25 ] for date in special_dates: idx np.where(dates.astype(str) date)[0] if len(idx) 0: daily_demand[idx] daily_demand[idx] * 1.5 return pd.DataFrame({Date: dates, Outbound: daily_demand}) # 2. Calculate four types of moving averages def calculate_moving_averages(data, window30): df data.copy() # SMA df[SMA] df[Outbound].rolling(windowwindow).mean() # WMA weights np.arange(1, window1) def wma(x, w): return np.dot(x, w) / w.sum() df[WMA] df[Outbound].rolling(windowwindow).apply( lambda x: wma(x, weights[:len(x)]), rawTrue) # EMA alpha 2 / (window 1) df[EMA] df[Outbound].ewm(alphaalpha, adjustFalse).mean() # HMA half_window max(1, window // 2) sqrt_window int(np.sqrt(window)) wma_half_weights np.arange(1, half_window1) wma_half df[Outbound].rolling(windowhalf_window).apply( lambda x: wma(x, wma_half_weights[:len(x)]), rawTrue) wma_full df[Outbound].rolling(windowwindow).apply( lambda x: wma(x, weights[:len(x)]), rawTrue) df[HMA_temp] 2 * wma_half - wma_full sqrt_weights np.arange(1, sqrt_window1) df[HMA] df[HMA_temp].rolling(windowsqrt_window).apply( lambda x: wma(x, sqrt_weights[:len(x)]), rawTrue) return df # 3. Predict daily outbound for January 2023 def predict_january_daily(df_2022, window30): jan_dates pd.date_range(start2023-01-01, end2023-01-31, freqD) predictions pd.DataFrame({ Date: jan_dates, DayOfWeek: [d.dayofweek for d in jan_dates], DayOfMonth: [d.day for d in jan_dates] }) last_sma df_2022[SMA].iloc[-1] last_wma df_2022[WMA].iloc[-1] last_ema df_2022[EMA].iloc[-1] last_hma df_2022[HMA].iloc[-1] weekday_adjustment np.where(predictions[DayOfWeek] 5, 1.1, 0.9) predictions[SMA_Pred] last_sma * weekday_adjustment predictions[WMA_Pred] last_wma * weekday_adjustment predictions[EMA_Pred] last_ema * weekday_adjustment predictions[HMA_Pred] last_hma * weekday_adjustment predictions.loc[predictions[Date] 2023-01-01, [SMA_Pred, WMA_Pred, EMA_Pred, HMA_Pred]] * 1.5 return predictions # 4. Display results as tables def display_results(df_2022, predictions): # Last 10 days of December 2022 dec_2022 df_2022[df_2022[Date].dt.month 12].tail(10).copy() dec_table dec_2022[[Date, Outbound, SMA, WMA, EMA, HMA]].copy() dec_table[Date] dec_table[Date].dt.strftime(%m-%d) # Format date consistently # January 2023 predictions pred_table predictions.copy() pred_table[Date] pred_table[Date].dt.strftime(%m-%d) # Format date consistently pred_table[Weekday] pred_table[DayOfWeek].map({ 0: Mon, 1: Tue, 2: Wed, 3: Thu, 4: Fri, 5: Sat, 6: Sun }) print(\nLast 10 days of December 2022 actual outbound and moving averages:) print(tabulate( dec_table.round(1), headers[Date, Outbound, SMA, WMA, EMA, HMA], tablefmtgrid, showindexFalse, floatfmt.1f )) print(\nJanuary 2023 daily outbound predictions (first 10 days):) print(tabulate( pred_table[[Date, Weekday, SMA_Pred, WMA_Pred, EMA_Pred, HMA_Pred]].head(10).round(1), headers[Date, Weekday, SMA, WMA, EMA, HMA], tablefmtgrid, showindexFalse, floatfmt.1f )) avg_pred pd.DataFrame({ Method: [SMA, WMA, EMA, HMA], Avg_Prediction: [ predictions[SMA_Pred].mean(), predictions[WMA_Pred].mean(), predictions[EMA_Pred].mean(), predictions[HMA_Pred].mean() ] }) print(\nAverage predictions for January by method:) print(tabulate( avg_pred.round(1), headerskeys, tablefmtgrid, showindexFalse, floatfmt.1f )) # 5. Visualize results def plot_results(df_2022, predictions, window30): plt.figure(figsize(14, 10)) ax1 plt.subplot(2, 1, 1) ax2 plt.subplot(2, 1, 2) # Subplot 1: 2022 actual data with moving averages ax1.plot(df_2022[Date], df_2022[Outbound], k-, labelActual Outbound, linewidth0.8, alpha0.7) ax1.plot(df_2022[Date], df_2022[SMA], b-, labelfSMA ({window} days), linewidth1.5) ax1.plot(df_2022[Date], df_2022[EMA], r-, labelfEMA ({window} days), linewidth1.5) ax1.set_title(2022 Daily Outbound with Moving Averages, fontsize14, pad20) ax1.set_ylabel(Outbound (units), fontsize12) ax1.legend(fontsize10, locupper left) ax1.grid(True, linestyle--, alpha0.6) ax1.xaxis.set_major_locator(MonthLocator()) ax1.xaxis.set_major_formatter(DateFormatter(%b)) # Subplot 2: January 2023 predictions ax2.plot(predictions[Date], predictions[SMA_Pred], b--, labelSMA Prediction, linewidth1.5) ax2.plot(predictions[Date], predictions[WMA_Pred], g-., labelWMA Prediction, linewidth1.5) ax2.plot(predictions[Date], predictions[EMA_Pred], r:, labelEMA Prediction, linewidth1.5) ax2.plot(predictions[Date], predictions[HMA_Pred], c-, labelHMA Prediction, linewidth1.5) weekends predictions[predictions[DayOfWeek] 5] for date in weekends[Date]: ax2.axvspan(date - pd.Timedelta(days0.5), date pd.Timedelta(days0.5), colorgray, alpha0.1) ax2.set_title(January 2023 Daily Outbound Predictions, fontsize14, pad20) ax2.set_ylabel(Predicted Outbound (units), fontsize12) ax2.legend(fontsize10, locupper left) ax2.grid(True, linestyle--, alpha0.6) ax2.xaxis.set_major_locator(MonthLocator()) ax2.xaxis.set_major_formatter(DateFormatter(%d-%b)) plt.tight_layout() # Save plot with error handling try: plot_path prediction_comparison.png plt.savefig(plot_path, dpi300, bbox_inchestight) except PermissionError: home_path os.path.expanduser(~) plot_path os.path.join(home_path, prediction_comparison.png) plt.savefig(plot_path, dpi300, bbox_inchestight) print(fCouldnt save plot to current directory. Saved to: {plot_path}) plt.show() # 6. Main program def main(): df_2022 generate_daily_data() window 30 df_ma calculate_moving_averages(df_2022, window) predictions predict_january_daily(df_ma, window) display_results(df_ma, predictions) plot_results(df_ma, predictions, window) try: # Prepare CSV output with visualization-consistent headers csv_output predictions.copy() csv_output[Date] csv_output[Date].dt.strftime(%m-%d) csv_output[Weekday] csv_output[DayOfWeek].map({ 0: Mon, 1: Tue, 2: Wed, 3: Thu, 4: Fri, 5: Sat, 6: Sun }) # Select and rename columns to match visualization csv_output csv_output[[ Date, Weekday, SMA_Pred, WMA_Pred, EMA_Pred, HMA_Pred ]].rename(columns{ SMA_Pred: SMA, WMA_Pred: WMA, EMA_Pred: EMA, HMA_Pred: HMA }) # Round values to 1 decimal place csv_output csv_output.round(1) # Try saving to current directory first try: csv_path january_2023_predictions.csv csv_output.to_csv(csv_path, indexFalse) print(f\nPredictions saved to {csv_path} in current directory) except PermissionError: # Fall back to users home directory if current directory fails home_path os.path.expanduser(~) csv_path os.path.join(home_path, january_2023_predictions.csv) csv_output.to_csv(csv_path, indexFalse) print(f\nCouldnt write to current directory. Predictions saved to: {csv_path}) except Exception as e: print(f\nError saving files: {str(e)}) print(Please check your directory permissions or specify a different output path.) if __name__ __main__: main()CSV/表格文件