本文目录导读：
我来通过一个完整的实战案例，展示如何用历史数据建模预测，以电商销售预测为例,我会用Python一步步实现。
📊 完整案例：电商日销售额预测
数据准备与探索
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.model_selection import train_test_split, TimeSeriesSplit
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score
import warnings
warnings.filterwarnings('ignore')
# 设置中文显示
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False
# 生成模拟历史销售数据（实际应用时读取真实数据）
np.random.seed(42)
dates = pd.date_range(start='2020-01-01', end='2023-12-31', freq='D')
n = len(dates)
# 构建有季节性和趋势的模拟数据
trend = np.linspace(500, 2000, n) # 增长趋势
seasonality = 300 * np.sin(2 * np.pi * np.arange(n) / 365) # 年度季节性
weekend_effect = np.where(pd.Series(dates).dt.dayofweek >= 5, 200, 0) # 周末效应
noise = np.random.normal(0, 100, n) # 随机噪声
sales = trend + seasonality + weekend_effect + noise + 1000
# 创建DataFrame
df = pd.DataFrame({
'日期': dates,
'销售额': sales,
'星期': pd.Series(dates).dt.day_name(),
'月份': pd.Series(dates).dt.month,
'是否周末': (pd.Series(dates).dt.dayofweek >= 5).astype(int)
})
print("\n=== 数据基本信息 ===")
print(df.head())
print(f"\n数据形状: {df.shape}")
print(f"\n数据统计描述:\n{df['销售额'].describe()}")
特征工程
def create_features(data):
"""
创建时间序列特征
"""
df = data.copy()
df['日期'] = pd.to_datetime(df['日期'])
# 时间特征
df['年'] = df['日期'].dt.year
df['月'] = df['日期'].dt.month
df['日'] = df['日期'].dt.day
df['星期'] = df['日期'].dt.weekday
df['季度'] = df['日期'].dt.quarter
# 周期性特征
df['日正弦'] = np.sin(2 * np.pi * df['日期'].dt.dayofyear / 365)
df['日余弦'] = np.cos(2 * np.pi * df['日期'].dt.dayofyear / 365)
df['月正弦'] = np.sin(2 * np.pi * df['月'] / 12)
df['月余弦'] = np.cos(2 * np.pi * df['月'] / 12)
# 是否节假日（简化处理）
df['是否周末'] = (df['星期'] >= 5).astype(int)
df['是否月初'] = (df['日'] <= 5).astype(int)
df['是否月末'] = (df['日'] >= 26).astype(int)
# 滞后特征（历史值）
for lag in [1, 2, 3, 7, 14, 30]:
df[f'滞后期_{lag}'] = df['销售额'].shift(lag)
# 移动平均特征
for window in [7, 14, 30]:
df[f'移动平均_{window}'] = df['销售额'].rolling(window=window).mean()
# 移动标准差
for window in [7, 30]:
df[f'移动标准差_{window}'] = df['销售额'].rolling(window=window).std()
# 删除缺失值
df = df.dropna()
return df
# 创建特征
df_features = create_features(df[['日期', '销售额']])
print("\n=== 特征工程后的数据 ===")
print(f"特征数量: {df_features.shape[1] - 2}")
print(f"可用样本数: {df_features.shape[0]}")
数据可视化分析
# 可视化分析
fig, axes = plt.subplots(2, 2, figsize=(15, 10))
# 1. 时间序列趋势
axes[0, 0].plot(df_features['日期'], df_features['销售额'], alpha=0.7, linewidth=0.8)
axes[0, 0].set_title('日销售额时间序列')
axes[0, 0].set_xlabel('日期')
axes[0, 0].set_ylabel('销售额')
# 2. 季节性分析（按月）
monthly_sales = df.set_index('日期').resample('M')['销售额'].mean()
axes[0, 1].plot(monthly_sales.index, monthly_sales.values, marker='o')
axes[0, 1].set_title('月度平均销售额趋势')
axes[0, 1].set_xlabel('月份')
axes[0, 1].set_ylabel('平均销售额')
# 3. 星期效应
weekday_avg = df.groupby('星期')['销售额'].mean()
axes[1, 0].bar(range(7), weekday_avg.values)
axes[1, 0].set_xticks(range(7))
axes[1, 0].set_xticklabels(['周一', '周二', '周三', '周四', '周五', '周六', '周日'])
axes[1, 0].set_title('星期平均销售额')
axes[1, 0].set_ylabel('平均销售额')
# 4. 特征相关性
corr_matrix = df_features.select_dtypes(include=[np.number]).corr()['销售额'].sort_values(ascending=False)[:15]
axes[1, 1].barh(corr_matrix.index, corr_matrix.values)
axes[1, 1].set_title('与销售额相关性最高的特征')
axes[1, 1].invert_yaxis()
plt.tight_layout()
plt.savefig('销售数据分析.png', dpi=300, bbox_inches='tight')
plt.show()
模型训练与对比
class SalesPredictor:
def __init__(self):
self.models = {}
self.scaler = StandardScaler()
self.feature_importance = None
def prepare_data(self, df_features):
"""
准备训练数据和测试数据
"""
# 特征列（排除日期和销售额）
feature_cols = [col for col in df_features.columns
if col not in ['日期', '销售额']]
X = df_features[feature_cols]
y = df_features['销售额']
# 时间序列分割（避免随机打乱）
split_idx = int(len(X) * 0.8)
X_train, X_val = X.iloc[:split_idx], X.iloc[split_idx:]
y_train, y_val = y.iloc[:split_idx], y.iloc[split_idx:]
# 标准化特征
X_train_scaled = self.scaler.fit_transform(X_train)
X_val_scaled = self.scaler.transform(X_val)
return X_train_scaled, X_val_scaled, y_train, y_val, feature_cols
def train_models(self, X_train, y_train):
"""
训练多个模型并比较
"""
models = {
'随机森林': RandomForestRegressor(
n_estimators=200,
max_depth=10,
min_samples_split=5,
random_state=42
),
'梯度提升': GradientBoostingRegressor(
n_estimators=200,
learning_rate=0.1,
max_depth=5,
random_state=42
),
'ExtraTrees': RandomForestRegressor(
n_estimators=200,
max_depth=12,
min_samples_split=3,
criterion='absolute_error',
random_state=42
)
}
trained_models = {}
for name, model in models.items():
print(f"训练 {name}...")
model.fit(X_train, y_train)
trained_models[name] = model
return trained_models
def evaluate_models(self, models, X_val, y_val):
"""
评估模型性能
"""
results = {}
for name, model in models.items():
predictions = model.predict(X_val)
mae = mean_absolute_error(y_val, predictions)
mse = mean_squared_error(y_val, predictions)
rmse = np.sqrt(mse)
r2 = r2_score(y_val, predictions)
# 计算MAPE
mape = np.mean(np.abs((y_val - predictions) / y_val)) * 100
results[name] = {
'MAE': mae,
'RMSE': rmse,
'R2': r2,
'MAPE%': mape
}
print(f"{name}:")
print(f" MAE = {mae:.2f}")
print(f" RMSE = {rmse:.2f}")
print(f" R2 = {r2:.4f}")
print(f" MAPE = {mape:.2f}%")
return results
# 创建预测器实例
predictor = SalesPredictor()
# 准备数据
X_train, X_val, y_train, y_val, feature_cols = predictor.prepare_data(df_features)
print(f"\n=== 数据划分 ===")
print(f"训练集大小: {len(X_train)}")
print(f"验证集大小: {len(X_val)}")
print(f"特征数量: {len(feature_cols)}")
# 训练模型
models = predictor.train_models(X_train, y_train)
# 评估模型
results = predictor.evaluate_models(models, X_val, y_val)
模型优化与选择
from sklearn.model_selection import GridSearchCV
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
def optimize_model(X_train, y_train):
"""
超参数调优
"""
print("\n=== 超参数调优 ===")
# 随机森林调参
print("优化随机森林参数...")
rf_param_grid = {
'n_estimators': [100, 200, 300],
'max_depth': [5, 10, 15],
'min_samples_split': [5, 10],
'min_samples_leaf': [2, 5]
}
rf_model = RandomForestRegressor(random_state=42)
rf_grid = GridSearchCV(
rf_model,
rf_param_grid,
cv=3,
scoring='neg_mean_absolute_error',
n_jobs=-1,
verbose=1
)
# 使用少量数据加速
sample_idx = np.random.choice(len(X_train), min(5000, len(X_train)), replace=False)
rf_grid.fit(X_train[sample_idx], y_train.iloc[sample_idx] if hasattr(y_train, 'iloc') else y_train[sample_idx])
print(f"随机森林最优参数: {rf_grid.best_params_}")
# 梯度提升调参
print("\n优化梯度提升参数...")
gb_param_grid = {
'n_estimators': [100, 200],
'learning_rate': [0.05, 0.1],
'max_depth': [4, 5, 6],
'subsample': [0.8, 1.0]
}
gb_model = GradientBoostingRegressor(random_state=42)
gb_grid = GridSearchCV(
gb_model,
gb_param_grid,
cv=3,
scoring='neg_mean_absolute_error',
n_jobs=-1,
verbose=1
)
gb_grid.fit(X_train[sample_idx],
y_train.iloc[sample_idx] if hasattr(y_train, 'iloc') else y_train[sample_idx])
print(f"梯度提升最优参数: {gb_grid.best_params_}")
return rf_grid.best_estimator_, gb_grid.best_estimator_
# 执行参数调优（注意：可能会运行较长时间）
# best_rf, best_gb = optimize_model(X_train, y_train)
最终模型验证与预测
def final_prediction_analysis(model, X_val, y_val, df_features, split_idx):
"""
最终预测分析和可视化
"""
# 预测
predictions = model.predict(X_val)
# 获取日期
val_dates = df_features['日期'].iloc[split_idx:]
# 创建比较DataFrame
comparison_df = pd.DataFrame({
'日期': val_dates,
'实际值': y_val.values if hasattr(y_val, 'values') else y_val,
'预测值': predictions
})
# 计算误差
comparison_df['误差'] = comparison_df['实际值'] - comparison_df['预测值']
comparison_df['绝对误差'] = np.abs(comparison_df['误差'])
comparison_df['误差百分比'] = (comparison_df['绝对误差'] / comparison_df['实际值']) * 100
return comparison_df
# 选择最佳模型（这里用梯度提升）
best_model = models['梯度提升']
# 生成预测
comparison_df = final_prediction_analysis(
best_model, X_val, y_val, df_features,
int(len(df_features) * 0.8)
)
# 可视化预测结果
fig, axes = plt.subplots(2, 2, figsize=(15, 12))
# 1. 预测vs实际
axes[0, 0].plot(comparison_df['日期'], comparison_df['实际值'],
label='实际值', linewidth=2)
axes[0, 0].plot(comparison_df['日期'], comparison_df['预测值'],
label='预测值', linewidth=1, linestyle='--')
axes[0, 0].set_title('预测值与实际值对比')
axes[0, 0].legend()
axes[0, 0].set_xlabel('日期')
axes[0, 0].set_ylabel('销售额')
# 2. 误差分布
axes[0, 1].hist(comparison_df['误差'], bins=30, alpha=0.7)
axes[0, 1].set_title('预测误差分布')
axes[0, 1].set_xlabel('误差')
axes[0, 1].set_ylabel('频次')
# 3. 最近30天对比
recent_30_days = comparison_df.tail(30)
axes[1, 0].plot(recent_30_days['日期'], recent_30_days['实际值'],
marker='o', label='实际值')
axes[1, 0].plot(recent_30_days['日期'], recent_30_days['预测值'],
marker='s', label='预测值')
axes[1, 0].set_title('最近30天预测对比')
axes[1, 0].legend()
axes[1, 0].set_xlabel('日期')
axes[1, 0].set_ylabel('销售额')
# 4. 散点图和相关性
axes[1, 1].scatter(comparison_df['实际值'], comparison_df['预测值'], alpha=0.5)
axes[1, 1].plot([comparison_df['实际值'].min(), comparison_df['实际值'].max()],
[comparison_df['实际值'].min(), comparison_df['实际值'].max()],
'r--')
axes[1, 1].set_title('预测值vs实际值散点图')
axes[1, 1].set_xlabel('实际值')
axes[1, 1].set_ylabel('预测值')
plt.tight_layout()
plt.savefig('模型预测结果.png', dpi=300, bbox_inches='tight')
plt.show()
# 输出统计结果
print("\n=== 预测精度统计 ===")
print(f"平均误差: {comparison_df['误差'].mean():.2f}")
print(f"平均绝对误差: {comparison_df['绝对误差'].mean():.2f}")
print(f"平均百分比误差: {comparison_df['误差百分比'].mean():.2f}%")
print(f"最大绝对误差: {comparison_df['绝对误差'].max():.2f}")
未来预测
def predict_future(df_features, model, steps=30):
"""
预测未来30天
"""
# 复制最新数据
future_df = df_features.copy()
# 获取最后日期
last_date = df_features['日期'].iloc[-1]
# 准备预测存储
future_predictions = []
print("\n=== 未来30天预测 ===")
for i in range(steps):
# 创建预测特征的占位
future_date = last_date + pd.Timedelta(days=1)
# 获取最新特征（滞后值更新）
last_row = future_df.iloc[-1:].copy()
# 更新日期相关特征
last_row['日期'] = future_date
last_row['年'] = future_date.year
last_row['月'] = future_date.month
last_row['日'] = future_date.day
last_row['星期'] = future_date.weekday
last_row['季度'] = future_date.quarter
last_row['日正弦'] = np.sin(2 * np.pi * future_date.timetuple().tm_yday / 365)
last_row['日余弦'] = np.cos(2 * np.pi * future_date.timetuple().tm_yday / 365)
last_row['月正弦'] = np.sin(2 * np.pi * future_date.month / 12)
last_row['月余弦'] = np.cos(2 * np.pi * future_date.month / 12)
last_row['是否周末'] = (future_date.weekday >= 5).astype(int)
last_row['是否月初'] = (future_date.day <= 5).astype(int)
last_row['是否月末'] = (future_date.day >= 26).astype(int)
# 准备特征列（与训练时一致）
feature_cols = [col for col in df_features.columns
if col not in ['日期', '销售额']]
# 预测
X_pred = predictor.scaler.transform(last_row[feature_cols])
prediction = model.predict(X_pred)[0]
# 记录预测结果
future_predictions.append({
'日期': future_date,
'预测值': prediction
})
# 更新最新的销售额（用于后续滞后特征计算）
last_row.loc[:, '销售额'] = prediction
# 重新计算滞后特征
future_df.loc[future_df.index[-1], '销售额'] = prediction
# 更新滞后特征
for lag in [1, 2, 3, 7, 14, 30]:
if i >= lag:
future_df.loc[future_df.index[-1], f'滞后期_{lag}'] = future_df['销售额'].iloc[-lag]
else:
# 使用已有数据
actual_sales = df_features['销售额'].iloc[-lag+i:]
future_df.loc[future_df.index[-1], f'滞后期_{lag}'] = actual_sales.iloc[0]
# 移动平均
for window in [7, 14, 30]:
if i >= window:
future_df.loc[future_df.index[-1], f'移动平均_{window}'] = \
future_df['销售额'].iloc[-window:].mean()
else:
# 结合历史数据
combined = pd.concat([
df_features['销售额'].iloc[-max(0, window-i):],
future_df['销售额'].iloc[:i+1]
])
future_df.loc[future_df.index[-1], f'移动平均_{window}'] = combined.mean()
# 更新日期
last_date = future_date
last_row.loc[:, '日期'] = future_date
return pd.DataFrame(future_predictions)
# 预测未来30天
future_predictions = predict_future(df_features, best_model, steps=30)
# 可视化未来预测
plt.figure(figsize=(15, 6))
# 绘制历史数据
hist_start = -90 # 展示最近90天
plt.plot(df_features['日期'].iloc[hist_start:],
df_features['销售额'].iloc[hist_start:],
label='历史销售额', linewidth=2)
# 绘制未来预测
plt.plot(future_predictions['日期'], future_predictions['预测值'],
label='未来预测', linewidth=2, linestyle='--', marker='o')
'未来30天销售预测')
plt.xlabel('日期')
plt.ylabel('销售额')
plt.legend()
plt.grid(True, alpha=0.3)
plt.savefig('未来预测结果.png', dpi=300, bbox_inches='tight')
plt.show()
print("\n=== 未来30天预测结果 ===")
print(future_predictions.head(10))
模型诊断与特征重要性
def model_diagnostics(df_features, best_model):
"""
模型诊断和特征重要性分析
"""
# 特征重要性
feature_cols = [col for col in df_features.columns
if col not in ['日期', '销售额']]
importance = pd.DataFrame({
'特征': feature_cols,
'重要性': best_model.feature_importances_
}).sort_values('重要性', ascending=False)
# 可视化特征重要性
plt.figure(figsize=(12, 8))
top_features = importance.head(20)
colors = plt.cm.viridis(np.linspace(0, 1, len(top_features)))
bars = plt.barh(range(len(top_features)), top_features['重要性'],
color=colors, alpha=0.7)
plt.yticks(range(len(top_features)), top_features['特征'])
plt.xlabel('重要性权重')
plt.title('特征重要性排名')
plt.tight_layout()
plt.savefig('特征重要性分析.png', dpi=300, bbox_inches='tight')
plt.show()
print("\n=== 特征重要性分析 ===")
print(importance.head(15).to_string(index=False))
return importance
# 执行模型诊断
importance_df = model_diagnostics(df_features, best_model)
# 残差分析
y_pred = best_model.predict(X_val)
residuals = y_val - y_pred
fig, axes = plt.subplots(1, 2, figsize=(12, 5))
# 残差随时间变化
axes[0].plot(range(len(residuals)), residuals, alpha=0.7, linewidth=0.8)
axes[0].axhline(y=0, color='red', linestyle='--')
axes[0].set_title('残差序列')
axes[0].set_xlabel('时间顺序')
axes[0].set_ylabel('残差')
# 残差直方图
axes[1].hist(residuals, bins=30, edgecolor='black')
axes[1].set_title('残差分布')
axes[1].set_xlabel('残差值')
axes[1].set_ylabel('频数')
plt.tight_layout()
plt.savefig('残差分析.png', dpi=300, bbox_inches='tight')
plt.show()
📈 结果分析模板
def generate_summary_report(df_features, comparison_df, future_predictions):
"""
生成预测分析总结报告
"""
print("\n" + "="*60)
print("销售预测分析报告")
print("="*60)
# 数据特征
print("\n【数据集统计】")
print(f"总样本数: {len(df_features)}")
print(f"开始日期: {df_features['日期'].min()}")
print(f"结束日期: {df_features['日期'].max()}")
# 预测精度
print("\n【模型性能】")
print(f"平均绝对误差: {comparison_df['绝对误差'].mean():.2f} 元")
print(f"平均相对误差: {comparison_df['误差百分比'].mean():.2f}%")
# 最优/最差预测
best_idx = comparison_df['绝对误差'].idxmin()
worst_idx = comparison_df['绝对误差'].idxmax()
print("\n【最准确预测】")
print(f"日期: {comparison_df.loc[best_idx, '日期'].strftime('%Y-%m-%d')}")
print(f"实际值: {comparison_df.loc[best_idx, '实际值']:.2f}")
print(f"预测值: {comparison_df.loc[best_idx, '预测值']:.2f}")
print("\n【最不准确预测】")
print(f"日期: {comparison_df.loc[worst_idx, '日期'].strftime('%Y-%m-%d')}")
print(f"实际值: {comparison_df.loc[worst_idx, '实际值']:.2f}")
print(f"预测值: {comparison_df.loc[worst_idx, '预测值']:.2f}")
# 未来趋势
print("\n【未来30天预测】")
future_mean = future_predictions['预测值'].mean()
recent_mean = df_features['销售额'].iloc[-30:].mean()
print(f"近30天平均销售额: {recent_mean:.2f}")
print(f"未来30天平均预测: {future_mean:.2f}")
growth_rate = (future_mean - recent_mean) / recent_mean * 100
print(f"预计增长率: {growth_rate:.2f}%")
# 输出总结报告
generate_summary_report(df_features, comparison_df, future_predictions)
🎯 关键要点总结
建模流程
- 数据探索：理解数据结构和特征
- 特征工程：时间特征、滞后特征、移动平均
- 模型选择：对比多个模型
- 参数调优：网格搜索最优参数
- 验证评估：多指标评估模型
- 实际预测：应用到新数据
常见模型对比
| 模型 | 优点 | 缺点 |
|---|---|---|
| 线性回归 | 简单、可解释 | 无法捕捉复杂非线性 |
| 随机森林 | 抗过拟合好 | 可能过拟合时间序列 |
| 梯度提升 | 精度高、稳健 | 训练时间较长 |
| LSTM等深度学习 | 能捕捉长期依赖 | 需要大量数据和时间 |
实际应用建议
- 数据质量优先：务必清理异常数据和缺失值
- 特征工程为王：好的特征比模型更重要
- 交叉验证：采用时间序列交叉验证
- 模型监控：定期评估并更新模型
- 稳健性测试：用不同时间段验证
这个案例涵盖了从数据处理到最终预测的完整流程，你可以根据实际场景调整数据集和参数,需要我解释任何部分或扩展到特定类型的预测吗？
