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

wen python案例 2026-08-14 5

本文目录导读：

完整预测案例（使用Prophet）
使用SARIMA模型的备选方案
机器学习方法（XGBoost/LSTM）
模型选择建议
最佳实践建议

我来给你一个完整的Python时间序列预测案例,从数据准备到模型构建与评估。

完整预测案例（使用Prophet）

环境准备

pip install pandas numpy matplotlib prophet scikit-learn

完整代码示例

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from prophet import Prophet
from sklearn.metrics import mean_absolute_error, mean_squared_error
import warnings
warnings.filterwarnings('ignore')
# ==================== 1. 数据准备 ====================
# 生成模拟的销售历史数据
np.random.seed(42)
dates = pd.date_range(start='2020-01-01', end='2023-12-31', freq='D')
# 创建具有趋势和季节性的数据
trend = np.linspace(100, 200, len(dates))
seasonality = 20 * np.sin(2 * np.pi * dates.dayofyear / 365)  # 年度季节性
weekly = 10 * np.sin(2 * np.pi * dates.dayofweek / 7)  # 周季节性
noise = np.random.normal(0, 5, len(dates))
sales = trend + seasonality + weekly + noise
# 创建DataFrame
data = pd.DataFrame({
    'ds': dates,  # Prophet要求的日期列名
    'y': sales     # Prophet要求的目标值列名
})
# ==================== 2. 数据探索 ====================
print("数据基本信息：")
print(data.head())
print(f"\n数据范围：{data['ds'].min()} 到 {data['ds'].max()}")
print(f"总记录数：{len(data)}")
# 绘制原始数据
plt.figure(figsize=(15, 6))
plt.plot(data['ds'], data['y'])'历史销售数据')
plt.xlabel('日期')
plt.ylabel('销售额')
plt.grid(True, alpha=0.3)
plt.show()
# ==================== 3. 数据划分 ====================
# 使用前3年数据训练，最后1年数据测试
train_size = int(len(data) * 0.75)
train_data = data.iloc[:train_size]
test_data = data.iloc[train_size:]
print(f"训练数据：{len(train_data)} 条 ({(train_data['ds'].min()} 到 {train_data['ds'].max()})")
print(f"测试数据：{len(test_data)} 条 ({(test_data['ds'].min()} 到 {test_data['ds'].max()})")
# ==================== 4. 模型训练 ====================
# 创建Prophet模型
model = Prophet(
    yearly_seasonality=True,      # 年度季节性
    weekly_seasonality=True,      # 周季节性
    daily_seasonality=False,      # 日季节性（对于日数据通常不需要）
    seasonality_mode='additive',  # 加法模式
    changepoint_prior_scale=0.05, # 变化点先验比例
    seasonality_prior_scale=10.0, # 季节性先验比例
    holidays_prior_scale=10.0,    # 节假日先验比例
)
# 训练模型
model.fit(train_data)
# ==================== 5. 预测 ====================
# 创建未来时间框架（预测测试期长度）
future_periods = len(test_data)
future = model.make_future_dataframe(periods=future_periods, freq='D')
forecast = model.predict(future)
# 提取测试期间的预测结果
forecast_test = forecast.iloc[train_size:]
# ==================== 6. 模型评估 ====================
# 计算误差指标
mae = mean_absolute_error(test_data['y'], forecast_test['yhat'])
rmse = np.sqrt(mean_squared_error(test_data['y'], forecast_test['yhat']))
mape = np.mean(np.abs((test_data['y'] - forecast_test['yhat']) / test_data['y'])) * 100
print(f"\n模型评估结果：")
print(f"MAE (平均绝对误差): {mae:.2f}")
print(f"RMSE (均方根误差): {rmse:.2f}")
print(f"MAPE (平均绝对百分比误差): {mape:.2f}%")
# ==================== 7. 结果可视化 ====================
# 7.1 预测结果对比图
plt.figure(figsize=(15, 8))
# 绘制训练数据的拟合
plt.plot(train_data['ds'], train_data['y'], 'b-', label='训练数据', alpha=0.6)
# 绘制测试数据的真实值
plt.plot(test_data['ds'], test_data['y'], 'g-', label='测试数据真实值', alpha=0.8)
# 绘制预测值
plt.plot(test_data['ds'], forecast_test['yhat'], 'r--', label='预测值', linewidth=2)
# 绘制置信区间
plt.fill_between(
    test_data['ds'],
    forecast_test['yhat_lower'],
    forecast_test['yhat_upper'],
    color='r', alpha=0.2, label='95%置信区间'
)
'销售预测结果对比')
plt.xlabel('日期')
plt.ylabel('销售额')
plt.legend()
plt.grid(True, alpha=0.3)
plt.show()
# 7.2 预测误差分析
plt.figure(figsize=(15, 6))
errors = test_data['y'].values - forecast_test['yhat'].values
plt.plot(test_data['ds'], errors, 'r-', alpha=0.7)
plt.axhline(y=0, color='b', linestyle='--', alpha=0.5)'预测误差分布')
plt.xlabel('日期')
plt.ylabel('误差值')
plt.grid(True, alpha=0.3)
plt.show()
# 7.3 预测成分分析
fig2 = model.plot_components(forecast)
plt.show()
# ==================== 8. 未来预测 ====================
# 预测未来30天
future_30 = model.make_future_dataframe(periods=30, freq='D')
forecast_30 = model.predict(future_30)
# 提取未来30天的预测
future_forecast = forecast_30.tail(30)
print("\n未来30天预测结果：")
print(future_forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']].tail(10))
# 可视化未来预测
plt.figure(figsize=(15, 6))
plt.plot(data['ds'], data['y'], 'b-', label='历史数据', alpha=0.7)
plt.plot(future_forecast['ds'], future_forecast['yhat'], 'r--', label='预测值', linewidth=2)
plt.fill_between(
    future_forecast['ds'],
    future_forecast['yhat_lower'],
    future_forecast['yhat_upper'],
    color='r', alpha=0.2, label='95%置信区间'
)'销售预测（未来30天）')
plt.xlabel('日期')
plt.ylabel('销售额')
plt.legend()
plt.grid(True, alpha=0.3)
plt.show()

使用SARIMA模型的备选方案

from statsmodels.tsa.statespace.sarimax import SARIMAX
from statsmodels.graphics.tsaplots import plot_acf, plot_pacf
# 使用上面的数据
# 转换数据格式为时间序列
ts_data = train_data.set_index('ds')['y']
# 绘制ACF和PACF图确定参数
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8))
plot_acf(ts_data, lags=40, ax=ax1)
plot_pacf(ts_data, lags=40, ax=ax2)
plt.show()
# 训练SARIMA模型
sarima_model = SARIMAX(
    ts_data,
    order=(1, 1, 1),          # (AR, I, MA)
    seasonal_order=(1, 1, 1, 7),  # 周季节性
    trend='c'
)
sarima_result = sarima_model.fit(disp=False)
# 预测
sarima_forecast = sarima_result.forecast(steps=len(test_data))
# 评估
mae_sarima = mean_absolute_error(test_data['y'], sarima_forecast)
print(f"SARIMA MAE: {mae_sarima:.2f}")

机器学习方法（XGBoost/LSTM）

from xgboost import XGBRegressor
# 特征工程
def create_features(df):
    df = df.copy()
    df['year'] = df['ds'].dt.year
    df['month'] = df['ds'].dt.month
    df['day'] = df['ds'].dt.day
    df['day_of_week'] = df['ds'].dt.dayofweek
    df['week_of_year'] = df['ds'].dt.isocalendar().week
    df['day_of_year'] = df['ds'].dt.dayofyear
    return df
# 创建特征
train_features = create_features(train_data)
test_features = create_features(test_data)
# 定义特征和目标
feature_cols = ['year', 'month', 'day', 'day_of_week', 'week_of_year', 'day_of_year']
X_train = train_features[feature_cols]
y_train = train_features['y']
X_test = test_features[feature_cols]
y_test = test_features['y']
# 训练XGBoost模型
xgb_model = XGBRegressor(
    n_estimators=100,
    learning_rate=0.1,
    max_depth=5,
    random_state=42
)
xgb_model.fit(X_train, y_train)
# 预测
xgb_predictions = xgb_model.predict(X_test)
# 评估
mae_xgb = mean_absolute_error(y_test, xgb_predictions)
rmse_xgb = np.sqrt(mean_squared_error(y_test, xgb_predictions))
print(f"XGBoost MAE: {mae_xgb:.2f}")
print(f"XGBoost RMSE: {rmse_xgb:.2f}")
# 特征重要性
importance_df = pd.DataFrame({
    'feature': feature_cols,
    'importance': xgb_model.feature_importances_
}).sort_values('importance', ascending=False)
print("\n特征重要性：")
print(importance_df)

模型选择建议

def model_selection_comparison(models, X_train, y_train, X_test, y_test):
    """比较多个模型的性能"""
    results = {}
    for name, model in models.items():
        model.fit(X_train, y_train)
        predictions = model.predict(X_test)
        mae = mean_absolute_error(y_test, predictions)
        rmse = np.sqrt(mean_squared_error(y_test, predictions))
        mape = np.mean(np.abs((y_test - predictions) / y_test)) * 100
        results[name] = {
            'MAE': mae,
            'RMSE': rmse,
            'MAPE': mape
        }
    return pd.DataFrame(results).T
# 示例：比较不同模型
models = {
    'Prophet': model,  # 已训练的Prophet
    'SARIMA': sarima_result,  # 已训练的SARIMA
    'XGBoost': xgb_model  # 已训练的XGBoost
}

最佳实践建议

数据质量检查

处理缺失值（填充或删除）
检测异常值（IQR、Z-score方法）
检查数据平稳性

特征工程

时间特征（年、月、日、星期）
滞后特征（前几期值）
滚动统计量（移动平均值、标准差）

模型调优

# 使用GridSearchCV进行超参数调优
from sklearn.model_selection import GridSearchCV
param_grid = {
    'n_estimators': [50, 100, 200],
    'max_depth': [3, 5, 7],
    'learning_rate': [0.01, 0.1, 0.3]
}
grid_search = GridSearchCV(
    XGBRegressor(),
    param_grid,
    cv=3,
    scoring='neg_mean_absolute_error'
)
grid_search.fit(X_train, y_train)
print(f"最佳参数：{grid_search.best_params_}")