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

我来用一个完整的股票价格预测案例,展示如何利用历史大数据进行建模预测。

完整案例：股票价格预测

数据准备与加载

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime, timedelta
import yfinance as yf
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import MinMaxScaler
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout, GRU
from tensorflow.keras.optimizers import Adam
# 设置中文显示
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False
# 下载股票数据（以苹果公司为例）
print("正在下载股票数据...")
stock_data = yf.download('AAPL', 
                        start='2018-01-01', 
                        end='2023-12-31',
                        progress=False)
print(f"数据形状: {stock_data.shape}")
print(stock_data.head())

数据探索与特征工程

# 数据探索
def explore_data(df):
    """数据探索分析"""
    print("="*50)
    print("数据基本信息:")
    print(df.info())
    print("\n统计数据:")
    print(df.describe())
    # 检查缺失值
    print("\n缺失值统计:")
    print(df.isnull().sum())
    # 检查是否有重复
    print(f"\n重复行数: {df.duplicated().sum()}")
explore_data(stock_data)
# 特征工程
def create_features(df):
    """创建技术指标特征"""
    df = df.copy()
    # 价格变化
    df['Price_Change'] = df['Close'].pct_change()
    # 移动平均线
    df['MA5'] = df['Close'].rolling(window=5).mean()
    df['MA20'] = df['Close'].rolling(window=20).mean()
    df['MA60'] = df['Close'].rolling(window=60).mean()
    # 波动率
    df['Volatility'] = df['Price_Change'].rolling(window=20).std()
    # RSI指标
    delta = df['Close'].diff()
    gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
    rs = gain / loss
    df['RSI'] = 100 - (100 / (1 + rs))
    # MACD指标
    exp1 = df['Close'].ewm(span=12, adjust=False).mean()
    exp2 = df['Close'].ewm(span=26, adjust=False).mean()
    df['MACD'] = exp1 - exp2
    df['Signal'] = df['MACD'].ewm(span=9, adjust=False).mean()
    df['MACD_Hist'] = df['MACD'] - df['Signal']
    # 交易量特征
    df['Volume_MA'] = df['Volume'].rolling(window=20).mean()
    df['Volume_Ratio'] = df['Volume'] / df['Volume_MA']
    # 价格范围
    df['High_Low_Ratio'] = (df['High'] - df['Low']) / df['Close']
    df['Close_Open_Ratio'] = (df['Close'] - df['Open']) / df['Open']
    # 清除NaN值
    df = df.dropna()
    return df
# 创建特征
featured_data = create_features(stock_data)
print(f"\n特征工程后的数据形状: {featured_data.shape}")
print(featured_data[['Close', 'MA5', 'MA20', 'RSI', 'MACD']].tail())

数据可视化

def visualize_data(df):
    """数据可视化分析"""
    fig, axes = plt.subplots(4, 2, figsize=(15, 15))
    fig.suptitle('股票数据分析', fontsize=16, fontweight='bold')
    # 1. 收盘价走势
    axes[0, 0].plot(df.index, df['Close'], label='Close Price', linewidth=2)
    axes[0, 0].plot(df.index, df['MA20'], label='MA20', alpha=0.7)
    axes[0, 0].plot(df.index, df['MA60'], label='MA60', alpha=0.7)
    axes[0, 0].set_title('股票价格走势')
    axes[0, 0].legend()
    axes[0, 0].grid(True, alpha=0.3)
    # 2. 交易量
    axes[0, 1].bar(df.index, df['Volume'], alpha=0.6, color='orange')
    axes[0, 1].set_title('交易量')
    axes[0, 1].grid(True, alpha=0.3)
    # 3. RSI指标
    axes[1, 0].plot(df.index, df['RSI'], color='purple')
    axes[1, 0].axhline(y=70, color='r', linestyle='--', alpha=0.5)
    axes[1, 0].axhline(y=30, color='g', linestyle='--', alpha=0.5)
    axes[1, 0].set_title('RSI指标')
    axes[1, 0].grid(True, alpha=0.3)
    # 4. MACD指标
    axes[1, 1].plot(df.index, df['MACD'], label='MACD', color='blue')
    axes[1, 1].plot(df.index, df['Signal'], label='Signal', color='red')
    axes[1, 1].fill_between(df.index, df['MACD_Hist'], alpha=0.3)
    axes[1, 1].set_title('MACD指标')
    axes[1, 1].legend()
    axes[1, 1].grid(True, alpha=0.3)
    # 5. 收益率分布
    axes[2, 0].hist(df['Price_Change'].dropna(), bins=50, edgecolor='black', alpha=0.7)
    axes[2, 0].set_title('收益率分布')
    axes[2, 0].set_xlabel('收益率')
    axes[2, 0].set_ylabel('频数')
    # 6. 价格相关性热图
    correlation = df[['Open', 'High', 'Low', 'Close', 'Volume', 'RSI', 'MACD']].corr()
    sns.heatmap(correlation, annot=True, cmap='coolwarm', center=0, ax=axes[2, 1])
    axes[2, 1].set_title('相关性热图')
    # 7. 波动率
    axes[3, 0].plot(df.index, df['Volatility'], color='green')
    axes[3, 0].set_title('价格波动率')
    axes[3, 0].grid(True, alpha=0.3)
    # 8. 收盘价分布
    axes[3, 1].hist(df['Close'], bins=50, edgecolor='black', alpha=0.7)
    axes[3, 1].set_title('收盘价分布')
    axes[3, 1].set_xlabel('价格')
    axes[3, 1].set_ylabel('频数')
    plt.tight_layout()
    plt.show()
visualize_data(featured_data)

数据预处理

def prepare_data(df, feature_cols, target_col, lookback=60):
    """
    准备LSTM训练数据
    参数:
    - df: 数据框
    - feature_cols: 特征列
    - target_col: 目标列
    - lookback: 历史窗口大小
    返回:
    - X_train, y_train, X_test, y_test, scaler
    """
    # 选择特征和标签
    data = df[feature_cols].values
    # 标准化
    scaler = MinMaxScaler()
    scaled_data = scaler.fit_transform(data)
    # 准备序列数据
    X, y = [], []
    for i in range(lookback, len(scaled_data)):
        X.append(scaled_data[i-lookback:i])
        y.append(scaled_data[i, feature_cols.index(target_col)])
    X, y = np.array(X), np.array(y)
    # 划分训练集和测试集
    train_size = int(len(X) * 0.8)
    X_train, X_test = X[:train_size], X[train_size:]
    y_train, y_test = y[:train_size], y[train_size:]
    print(f"训练集形状: {X_train.shape}")
    print(f"测试集形状: {X_test.shape}")
    return X_train, y_train, X_test, y_test, scaler
# 定义特征和目标
feature_cols = ['Open', 'High', 'Low', 'Close', 'Volume', 
               'MA5', 'MA20', 'RSI', 'MACD', 'Volatility']
target_col = 'Close'
# 准备数据
lookback = 60  # 使用过去60天的数据
X_train, y_train, X_test, y_test, scaler = prepare_data(
    featured_data, feature_cols, target_col, lookback
)

构建LSTM预测模型

def build_lstm_model(input_shape):
    """构建LSTM模型"""
    model = Sequential()
    # 第一层LSTM
    model.add(LSTM(units=128, return_sequences=True, input_shape=input_shape))
    model.add(Dropout(0.2))
    # 第二层LSTM
    model.add(LSTM(units=128, return_sequences=True))
    model.add(Dropout(0.2))
    # 第三层LSTM
    model.add(LSTM(units=64, return_sequences=False))
    model.add(Dropout(0.2))
    # 全连接层
    model.add(Dense(units=32, activation='relu'))
    model.add(Dropout(0.1))
    # 输出层
    model.add(Dense(units=1))
    # 编译模型
    model.compile(
        optimizer=Adam(learning_rate=0.001),
        loss='mse',
        metrics=['mae']
    )
    return model
# 创建模型
print("构建LSTM模型...")
model = build_lstm_model((X_train.shape[1], X_train.shape[2]))
model.summary()
# 训练模型
print("开始训练模型...")
history = model.fit(
    X_train, y_train,
    validation_data=(X_test, y_test),
    epochs=50,
    batch_size=32,
    verbose=1
)
# 保存模型
model.save('lstm_stock_model.h5')
print("模型已保存!")

模型评估与预测

def evaluate_and_predict(model, X_test, y_test, scaler, feature_cols, target_col):
    """模型评估和预测"""
    # 预测
    y_pred = model.predict(X_test)
    # 反标准化预测值
    y_pred_original = scaler.inverse_transform(
        np.concatenate([np.zeros((len(y_pred), len(feature_cols) - 1)), 
                       y_pred.reshape(-1, 1)], axis=1)
    )[:, -1]
    # 反标准化实际值
    y_test_original = scaler.inverse_transform(
        np.concatenate([np.zeros((len(y_test), len(feature_cols) - 1)), 
                       y_test.reshape(-1, 1)], axis=1)
    )[:, -1]
    # 计算评估指标
    mse = mean_squared_error(y_test_original, y_pred_original)
    mae = mean_absolute_error(y_test_original, y_pred_original)
    r2 = r2_score(y_test_original, y_pred_original)
    print("="*50)
    print("模型评估指标:")
    print(f"均方误差 (MSE): {mse:.4f}")
    print(f"平均绝对误差 (MAE): {mae:.4f}")
    print(f"R² 决定系数: {r2:.4f}")
    print(f"RMSE: {np.sqrt(mse):.4f}")
    return y_pred_original, y_test_original
# 预测和评估
y_pred_original, y_test_original = evaluate_and_predict(
    model, X_test, y_test, scaler, feature_cols, target_col
)
# 绘制结果
fig, axes = plt.subplots(3, 1, figsize=(15, 12))
# 1. 训练损失
axes[0].plot(history.history['loss'], label='Training Loss')
axes[0].plot(history.history['val_loss'], label='Validation Loss')
axes[0].set_title('模型损失曲线')
axes[0].set_xlabel('Epoch')
axes[0].set_ylabel('Loss')
axes[0].legend()
axes[0].grid(True, alpha=0.3)
# 2. 预测结果对比
axes[1].plot(y_test_original, label='Actual Prices', alpha=0.7)
axes[1].plot(y_pred_original, label='Predicted Prices', alpha=0.7)
axes[1].set_title('股票价格预测 vs 实际价格')
axes[1].set_xlabel('Days')
axes[1].set_ylabel('Price')
axes[1].legend()
axes[1].grid(True, alpha=0.3)
# 3. 预测误差
axes[2].plot(y_test_original - y_pred_original, color='red')
axes[2].axhline(y=0, color='black', linestyle='-', linewidth=0.5)
axes[2].set_title('预测误差')
axes[2].set_xlabel('Days')
axes[2].set_ylabel('Error')
axes[2].grid(True, alpha=0.3)
plt.tight_layout()
plt.show()

未来价格预测

def predict_future(model, last_sequence, scaler, feature_cols, target_col, days=30):
    """
    预测未来价格
    参数:
    - model: 训练好的模型
    - last_sequence: 最近的历史数据序列
    - days: 预测天数
    """
    future_predictions = []
    current_sequence = last_sequence.copy()
    for _ in range(days):
        # 预测下一个值
        next_pred = model.predict(current_sequence.reshape(1, current_sequence.shape[0], current_sequence.shape[1]))
        # 创建新序列
        new_sequence = current_sequence[1:]
        # 将预测值添加到序列中（使用Close列的位置）
        # 这里需要根据feature_cols中Close的位置调整
        close_idx = feature_cols.index('Close')
        new_point = np.zeros((1, current_sequence.shape[2]))
        new_point[0, close_idx] = next_pred[0, 0]
        # 更新其他特征（简化处理）
        new_sequence = np.append(new_sequence, new_point, axis=0)
        current_sequence = new_sequence
        # 存储预测值
        future_predictions.append(next_pred[0, 0])
    # 反标准化
    future_predictions = np.array(future_predictions)
    future_prices = scaler.inverse_transform(
        np.concatenate([np.zeros((len(future_predictions), len(feature_cols) - 1)), 
                       future_predictions.reshape(-1, 1)], axis=1)
    )[:, -1]
    return future_prices
# 预测未来30天价格
future_days = 30
last_sequence = X_test[-1]  # 使用最后一段测试数据
future_prices = predict_future(
    model, last_sequence, scaler, feature_cols, target_col, days=future_days
)
# 绘制未来预测
fig, ax = plt.subplots(figsize=(12, 6))
# 画出历史价格（最后100天）
historical_days = 100
historical_prices = featured_data['Close'][-historical_days:].values
# 预测日期
last_date = featured_data.index[-1]
future_dates = pd.date_range(start=last_date + timedelta(days=1), 
                            periods=2*future_days, 
                            freq='D')
# 实际价格和预测价格的连接
combined_prices = np.concatenate([historical_prices, future_prices])
# 绘制
ax.plot(range(historical_days), historical_prices, label='历史价格', color='blue')
ax.plot(range(historical_days-1, historical_days-1+len(future_prices)), 
        future_prices, label='预测价格', color='red', linestyle='--')
ax.set_title('股票价格预测')
ax.set_xlabel('天数')
ax.set_ylabel('价格')
ax.legend()
ax.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
print(f"future_days}天预测价格:")
for i, price in enumerate(future_prices, 1):
    print(f"第{i}天: ${price:.2f}")

模型部署示例

class StockPredictor:
    """股票价格预测器"""
    def __init__(self, model_path):
        """初始化预测器"""
        self.model = tf.keras.models.load_model(model_path)
        self.scaler = None
        self.feature_cols = ['Open', 'High', 'Low', 'Close', 'Volume', 
                           'MA5', 'MA20', 'RSI', 'MACD', 'Volatility']
        self.lookback = 60
    def load_data(self, symbol, start_date, end_date):
        """加载并准备数据"""
        data = yf.download(symbol, start=start_date, end=end_date, progress=False)
        featured = create_features(data)
        return featured
    def predict_next(self, data):
        """预测下一天价格"""
        try:
            # 准备数据
            scaled_data = self.scaler.transform(data[self.feature_cols].values)
            last_sequence = scaled_data[-self.lookback:]
            # 预测
            prediction = self.model.predict(last_sequence.reshape(1, self.lookback, len(self.feature_cols)))
            # 反标准化
            predicted_price = self.scaler.inverse_transform(
                np.concatenate([np.zeros((1, len(self.feature_cols) - 1)), 
                               prediction.reshape(-1, 1)], axis=1)
            )[0, -1]
            return predicted_price
        except Exception as e:
            print(f"预测失败: {e}")
            return None
# 使用示例
try:
    predictor = StockPredictor('lstm_stock_model.h5')
    print("预测器初始化成功!")
    # 预测示例
    # predictor.predict_next(featured_data)
except Exception as e:
    print(f"初始化失败: {e}")

数据处理流程

• 数据清洗：处理缺失值、异常值
• 特征工程：技术指标、市场指标、时间序列特征
• 数据标准化：避免不同量纲的影响

模型选择建议

优化建议

# 使用早停法防止过拟合
from tensorflow.keras.callbacks import EarlyStopping
early_stop = EarlyStopping(monitor='val_loss', 
                          patience=10, 
                          restore_best_weights=True)
# 使用学习率调度
from tensorflow.keras.callbacks import ReduceLROnPlateau
reduce_lr = ReduceLROnPlateau(monitor='val_loss', 
                             factor=0.2, 
                             patience=5, 
                             min_lr=0.0001)

注意事项

• ⚠️ 股票预测具有高度不确定性
• 历史数据不等于未来表现
• 建议结合基本面分析
• 定期重新训练模型
• 监控模型漂移

这个完整案例展示了如何从数据准备到模型部署的完整流程，实际应用中需要根据具体需求调整参数和模型结构,同时要注意风险控制。