我来用一个完整的股票价格预测案例,展示如何利用历史大数据进行建模预测。
完整案例：股票价格预测
数据准备与加载
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}")
数据处理流程
- 数据清洗：处理缺失值、异常值
- 特征工程：技术指标、市场指标、时间序列特征
- 数据标准化：避免不同量纲的影响
模型选择建议
| 模型类型 | 适用场景 | 优势 | 劣势 |
|---|---|---|---|
| LSTM | 序列预测 | 擅长捕捉长期依赖 | 训练时间较长 |
| GRU | 简单序列预测 | 计算效率高 | 表达能力略弱 |
| XGBoost | 特征预测 | 处理非线性 | 不擅长序列 |
| ARIMA | 简单时间序列 | 统计基础好 | 线性假设较强 |
优化建议
# 使用早停法防止过拟合
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)
注意事项
- ⚠️ 股票预测具有高度不确定性
- 历史数据不等于未来表现
- 建议结合基本面分析
- 定期重新训练模型
- 监控模型漂移
这个完整案例展示了如何从数据准备到模型部署的完整流程，实际应用中需要根据具体需求调整参数和模型结构,同时要注意风险控制。
