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

本文目录导读：

1. 数据准备与探索
2. 特征工程
3. 数据可视化分析
4. 建立预测模型
5. 模型优化与选择
6. 模型评估与可视化
7. 预测未来数据
8. 模型部署准备
9. 性能监控与模型更新

我来给你一个完整的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)

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

1. 数据准备：生成和加载历史数据
2. 特征工程：创建滞后特征、移动平均等
3. 数据分析：可视化和统计探索
4. 模型建立：多种机器学习模型对比
5. 模型优化：网格搜索超参数调优
6. 模型评估：使用R²、RMSE等指标
7. 未来预测：预测未来数据
8. 模型部署：保存和加载模型
9. 性能监控：跟踪模型性能

这个流程可以应用于各种预测场景,如销量预测、股票预测、需求预测等，关键是要根据具体业务场景调整特征工程和数据预处理策略。