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

wen python案例 2026-08-06 1

本文目录导读：

数据准备与探索
特征工程
数据可视化分析
建立预测模型
模型优化与选择
模型评估与可视化
预测未来数据
模型部署准备
性能监控与模型更新

我来给你一个完整的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
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import 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)
n_samples = 1000
# 创建时间序列数据
dates = pd.date_range(start='2020-01-01', periods=n_samples, freq='D')
df = pd.DataFrame({
    'date': dates,
    'sales': np.random.normal(100, 20, n_samples) + 
             np.sin(np.arange(n_samples)/30)*10 +  # 周期性
             np.arange(n_samples)/50                # 趋势
})
# 添加特征
df['month'] = df['date'].dt.month
df['day_of_week'] = df['date'].dt.dayofweek
df['days_from_start'] = np.arange(n_samples)
print("数据预览:")
print(df.head())
print("\n数据统计:")
print(df.describe())

特征工程

# 创建滞后特征（利用历史数据）
def create_lag_features(df, target_col, lags=[1, 7, 30]):
    """
    创建历史滞后特征
    """
    df = df.copy()
    for lag in lags:
        df[f'{target_col}_lag_{lag}'] = df[target_col].shift(lag)
    # 移动平均特征
    df['sales_ma_7'] = df['sales'].rolling(window=7).mean()
    df['sales_ma_30'] = df['sales'].rolling(window=30).mean()
    # 差分特征（去除趋势）
    df['sales_diff'] = df['sales'].diff()
    df['sales_diff_7'] = df['sales'].diff(7)
    return df
# 应用特征工程
df_enhanced = create_lag_features(df, 'sales')
df_enhanced = df_enhanced.dropna()
print("特征工程后的数据:")
print(df_enhanced[['sales', 'sales_lag_1', 'sales_lag_7', 'sales_ma_7']].head())

数据可视化分析

# 绘制时间序列图
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
# 原始序列
axes[0, 0].plot(df['date'], df['sales'])
axes[0, 0].set_title('销售数据时间序列')
axes[0, 0].set_xlabel('日期')
axes[0, 0].set_ylabel('销售额')
# 分布图
axes[0, 1].hist(df['sales'], bins=50, alpha=0.7)
axes[0, 1].set_title('销售数据分布')
# 自相关图
from pandas.plotting import autocorrelation_plot
autocorrelation_plot(df['sales'], ax=axes[1, 0])
axes[1, 0].set_title('自相关图')
# 滞后散点图
axes[1, 1].scatter(df_enhanced['sales_lag_1'][:500], 
                  df_enhanced['sales'][:500], alpha=0.5)
axes[1, 1].set_title('滞后1期与当前值关系')
axes[1, 1].set_xlabel('前一时期销售额')
axes[1, 1].set_ylabel('当前销售额')
plt.tight_layout()
plt.show()

建立预测模型

from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.linear_model import LinearRegression, Ridge
from xgboost import XGBRegressor
from sklearn.model_selection import cross_val_score
# 准备数据
X = df_enhanced.drop(['date', 'sales'], axis=1)
y = df_enhanced['sales']
# 划分训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, shuffle=False
)
# 数据标准化
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)
# 定义模型
models = {
    '线性回归': LinearRegression(),
    '岭回归': Ridge(alpha=1.0),
    '随机森林': RandomForestRegressor(
        n_estimators=100, 
        max_depth=10, 
        random_state=42
    ),
    '梯度提升': GradientBoostingRegressor(
        n_estimators=100, 
        learning_rate=0.1,
        random_state=42
    ),
    'XGBoost': XGBRegressor(
        n_estimators=200,
        learning_rate=0.1,
        max_depth=5,
        random_state=42
    )
}
# 训练和评估模型
results = {}
for name, model in models.items():
    # 训练模型
    model.fit(X_train_scaled, y_train)
    # 预测
    y_pred_train = model.predict(X_train_scaled)
    y_pred_test = model.predict(X_test_scaled)
    # 评估
    r2_train = r2_score(y_train, y_pred_train)
    r2_test = r2_score(y_test, y_pred_test)
    rmse_test = np.sqrt(mean_squared_error(y_test, y_pred_test))
    results[name] = {
        'model': model,
        'r2_train': r2_train,
        'r2_test': r2_test,
        'rmse_test': rmse_test
    }
    print(f"\n{name}模型:")
    print(f"  训练集R²: {r2_train:.4f}")
    print(f"  测试集R²: {r2_test:.4f}")
    print(f"  测试集RMSE: {rmse_test:.4f}")

模型优化与选择

from sklearn.model_selection import GridSearchCV
# 随机森林参数优化
param_grid_rf = {
    'n_estimators': [50, 100, 200],
    'max_depth': [5, 10, 15, None],
    'min_samples_split': [2, 5, 10]
}
grid_rf = GridSearchCV(
    RandomForestRegressor(random_state=42),
    param_grid_rf,
    cv=5,
    scoring='r2',
    n_jobs=-1
)
grid_rf.fit(X_train_scaled, y_train)
print("随机森林最佳参数:")
print(grid_rf.best_params_)
print(f"最佳交叉验证得分: {grid_rf.best_score_:.4f}")
# XGBoost参数优化
param_grid_xgb = {
    'n_estimators': [100, 200, 300],
    'learning_rate': [0.01, 0.1, 0.3],
    'max_depth': [3, 5, 7],
    'subsample': [0.8, 1.0]
}
grid_xgb = GridSearchCV(
    XGBRegressor(random_state=42),
    param_grid_xgb,
    cv=5,
    scoring='r2',
    n_jobs=-1
)
grid_xgb.fit(X_train_scaled, y_train)
print("\nXGBoost最佳参数:")
print(grid_xgb.best_params_)
print(f"最佳交叉验证得分: {grid_xgb.best_score_:.4f}")

模型评估与可视化

# 选择最佳模型进行预测
best_model = models['XGBoost']  # 假设XGBoost表现最好
# 预测
y_pred_test = best_model.predict(X_test_scaled)
# 特征重要性
if hasattr(best_model, 'feature_importances_'):
    feature_importance = pd.DataFrame({
        'feature': X.columns,
        'importance': best_model.feature_importances_
    }).sort_values('importance', ascending=False)
    print("特征重要性:")
    print(feature_importance.head(10))
    # 绘制特征重要性图
    plt.figure(figsize=(10, 6))
    plt.bar(feature_importance['feature'][:10], 
            feature_importance['importance'][:10])
    plt.title('特征重要性排名')
    plt.xlabel('特征')
    plt.ylabel('重要性')
    plt.xticks(rotation=45)
    plt.show()
# 预测结果可视化
plt.figure(figsize=(14, 6))
plt.plot(y_test.index, y_test.values, label='实际值', alpha=0.7)
plt.plot(y_test.index, y_pred_test, label='预测值', alpha=0.7)'预测结果对比')
plt.xlabel('样本索引')
plt.ylabel('销售额')
plt.legend()
plt.grid(True, alpha=0.3)
plt.show()

预测未来数据

def predict_future(model, last_data, steps, scaler, feature_names):
    """
    预测未来数据
    Parameters:
    -----------
    model: 训练好的模型
    last_data: 最近的数据点
    steps: 预测步数
    scaler: 标准化器
    feature_names: 特征名称列表
    """
    future_predictions = []
    current_data = last_data.copy()
    # 生成未来日期
    last_date = pd.Timestamp('2023-01-01')
    future_dates = pd.date_range(
        start=last_date, 
        periods=steps, 
        freq='D'
    )
    for i in range(steps):
        # 准备特征向量（这里需要根据你的特征工程逻辑）
        future_features = []
        for feature in feature_names:
            if 'lag' in feature:
                # 滞后特征处理（简化版）
                if current_data[feature] is not None:
                    future_features.append(current_data[feature])
                else:
                    future_features.append(0)
            else:
                # 日期相关特征
                if feature == 'month':
                    future_features.append(future_dates[i].month)
                elif feature == 'day_of_week':
                    future_features.append(future_dates[i].dayofweek)
                elif feature == 'days_from_start':
                    future_features.append(len(df) + i)
                else:
                    future_features.append(0)
        # 标准化特征
        future_features_scaled = scaler.transform(
            np.array(future_features).reshape(1, -1)
        )
        # 预测
        pred = model.predict(future_features_scaled)[0]
        future_predictions.append(pred)
        # 更新滞后特征
        current_data = pred
    return future_dates, future_predictions
# 使用最近的数据进行预测
last_data_point = df_enhanced.iloc[-1].to_dict()
future_dates, future_values = predict_future(
    best_model, 
    last_data_point, 
    steps=30,
    scaler=scaler,
    feature_names=X.columns.tolist()
)
# 可视化未来预测
plt.figure(figsize=(14, 6))
plt.plot(df['date'][-100:], df['sales'][-100:], label='历史数据', alpha=0.7)
plt.plot(future_dates, future_values, 'r--', label='未来预测', alpha=0.8)'未来30天销售预测')
plt.xlabel('日期')
plt.ylabel('销售额')
plt.legend()
plt.grid(True, alpha=0.3)
plt.show()

模型部署准备

import joblib
# 保存模型和预处理对象
def save_model(model, scaler, filepath='models/'):
    """
    保存训练好的模型和标准化器
    """
    import os
    os.makedirs(filepath, exist_ok=True)
    # 保存模型
    joblib.dump(model, f'{filepath}best_model.pkl')
    joblib.dump(scaler, f'{filepath}scaler.pkl')
    print(f"模型已保存到 {filepath}")
# 保存模型
save_model(best_model, scaler)
# 模型加载示例
def load_model(filepath='models/'):
    """
    加载模型
    """
    model = joblib.load(f'{filepath}best_model.pkl')
    scaler = joblib.load(f'{filepath}scaler.pkl')
    return model, scaler
# 重新加载模型进行验证
loaded_model, loaded_scaler = load_model()
print("模型加载成功!")
# 使用加载的模型进行预测
sample_data = X_test_scaled[0].reshape(1, -1)
prediction = loaded_model.predict(sample_data)
print(f"样本预测值: {prediction[0]:.2f}")
print(f"实际值: {y_test.iloc[0]:.2f}")

性能监控与模型更新

def monitor_performance(model, X_new, y_new):
    """
    监控模型性能，检测是否需要更新
    """
    # 预测新数据
    y_pred_new = model.predict(X_new)
    # 计算新数据的性能指标
    mae = np.mean(np.abs(y_pred_new - y_new))
    rmse = np.sqrt(mean_squared_error(y_new, y_pred_new))
    r2 = r2_score(y_new, y_pred_new)
    print("新数据性能指标:")
    print(f"  MAE: {mae:.4f}")
    print(f"  RMSE: {rmse:.4f}")
    print(f"  R²: {r2:.4f}")
    # 设置阈值判断是否需要重新训练
    threshold_r2 = 0.7
    if r2 < threshold_r2:
        print(f"警告: R²低于阈值{threshold_r2}，建议重新训练模型")
        return False
    else:
        print("模型性能良好，继续使用")
        return True
# 模拟新数据到来
new_data = df_enhanced.iloc[-50:]  # 使用最后50条数据
X_new = new_data.drop(['date', 'sales'], axis=1)
y_new = new_data['sales']
# 标准化新数据
X_new_scaled = scaler.transform(X_new)
# 监控模型性能
monitor_performance(loaded_model, X_new_scaled, y_new)

这个完整的预测建模案例涵盖了：