python案例如何利用历史大数据建模预测？

本文目录导读：

1. 📊 完整案例：电商日销售额预测
2. 📈 结果分析模板
3. 🎯 关键要点总结

我来通过一个完整的实战案例，展示如何用历史数据建模预测，以电商销售预测为例,我会用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)

🎯 关键要点总结

建模流程

• 数据探索：理解数据结构和特征
• 特征工程：时间特征、滞后特征、移动平均
• 模型选择：对比多个模型
• 参数调优：网格搜索最优参数
• 验证评估：多指标评估模型
• 实际预测：应用到新数据

常见模型对比

实际应用建议

• 数据质量优先：务必清理异常数据和缺失值
• 特征工程为王：好的特征比模型更重要
• 交叉验证：采用时间序列交叉验证
• 模型监控：定期评估并更新模型
• 稳健性测试：用不同时间段验证

这个案例涵盖了从数据处理到最终预测的完整流程，你可以根据实际场景调整数据集和参数,需要我解释任何部分或扩展到特定类型的预测吗？