深度学习--lstm---股票预测
# 导入PyTorch相关库
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, Dataset# 导入数值计算和数据处理库
import numpy as np
import pandas as pd# 导入可视化库
import matplotlib.pyplot as plt
import seaborn as sns# 导入数据预处理库
from sklearn.preprocessing import MinMaxScaler, StandardScaler
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score# 导入系统库
import warnings
import os
from datetime import datetime, timedelta# 导入金融数据获取库
import yfinance as yf# 导入进度条库
from tqdm import tqdm# 导入JSON处理库
import json# 设置matplotlib中文显示
plt.rcParams['font.sans-serif'] = ['SimHei'] # 指定默认字体
plt.rcParams['axes.unicode_minus'] = False # 解决负号显示问题# 忽略警告信息
warnings.filterwarnings('ignore')class StockDataProcessor:"""股票数据处理类：负责数据下载、特征工程和数据预处理"""def __init__(self, sequence_length=60, # 时间序列长度（历史数据窗口大小）prediction_days=5, # 预测未来天数feature_columns=['Open', 'High', 'Low', 'Close', 'Volume'], # 特征列target_column='Close', # 目标列（预测目标）scaler_type='minmax', # 标准化类型（'minmax'或'standard'）train_ratio=0.8, # 训练集比例val_ratio=0.1): # 验证集比例"""初始化数据处理器Args:sequence_length: 序列长度，用于LSTM输入prediction_days: 预测天数feature_columns: 特征列名target_column: 目标列名scaler_type: 标准化类型 ('minmax' 或 'standard')train_ratio: 训练集比例val_ratio: 验证集比例"""# 保存初始化参数self.sequence_length = sequence_length #sequence_length: 序列长度，用于LSTM输入self.prediction_days = prediction_days #预测天数self.feature_columns = feature_columns #特征列名self.target_column = target_column #目标列名self.scaler_type = scaler_type #标准化类型self.train_ratio = train_ratio #train_ratioself.val_ratio = val_ratio #val_ratio# 初始化标准化器# 使用 minmax（适合价格预测）if scaler_type == 'minmax':"""特点：适用场景：数据分布不遵循正态分布时（如股票价格波动剧烈时）。对离群点敏感：如果数据中存在极端值（如股价暴涨暴跌），会导致缩放后的数据集中在某一端。保留原始分布形状：仅调整范围，不改变分布形态。"""self.feature_scaler = MinMaxScaler() # 特征缩放到 [0, 1]self.target_scaler = MinMaxScaler() # 目标变量缩放到 [0, 1]else:"""特点：适用场景：数据近似服从正态分布时（如股票收益率）。对离群点鲁棒性更强：均值和标准差受极端值影响较小。可能改变分布形状：即使原始数据非正态，标准化后可能更接近正态"""# 使用 standard（适合收益率或波动率预测）self.feature_scaler = StandardScaler() # 特征标准化器 # 特征标准化为均值0、方差1self.target_scaler = StandardScaler() # 目标变量标准化器 # 目标变量标准化为均值0、方差1def download_stock_data(self, symbol, start_date, end_date):"""下载股票数据"""try:# 使用yfinance下载股票数据stock_data = yf.download(symbol, start=start_date, end=end_date)return stock_dataexcept Exception as e:print(f"下载数据失败: {e}")return Nonedef add_technical_indicators(self, data):"""添加技术指标"""df = data.copy()# 移动平均线指标（MA）df['MA5'] = df['Close'].rolling(window=5).mean() # 5日均线df['MA10'] = df['Close'].rolling(window=10).mean() # 10日均线df['MA20'] = df['Close'].rolling(window=20).mean() # 20日均线df['MA50'] = df['Close'].rolling(window=50).mean() # 50日均线# 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 # 计算RS值df['RSI'] = 100 - (100 / (1 + rs)) # 计算RSI# MACD指标（指数移动平均线）exp1 = df['Close'].ewm(span=12, adjust=False).mean() # 快速EMAexp2 = df['Close'].ewm(span=26, adjust=False).mean() # 慢速EMAdf['MACD'] = exp1 - exp2 # DIF线df['MACD_Signal'] = df['MACD'].ewm(span=9, adjust=False).mean() # DEA线df['MACD_Histogram'] = df['MACD'] - df['MACD_Signal'] # MACD柱# 布林带指标（Bollinger Bands）df['BB_Middle'] = df['Close'].rolling(window=20).mean() # 中轨df['BB_Upper'] = df['BB_Middle'] + 2 * df['Close'].rolling(window=20).std() # 上轨df['BB_Lower'] = df['BB_Middle'] - 2 * df['Close'].rolling(window=20).std() # 下轨# 成交量移动平均df['Volume_MA'] = df['Volume'].rolling(window=10).mean() # 10日成交量均线# 价格波动率（标准差）df['Volatility'] = df['Close'].rolling(window=10).std() # 10日价格波动率# 收益率（收益率计算）df['Returns'] = df['Close'].pct_change() # 百分比变化# K线形态特征（上下影线和实体大小）df['Upper_Shadow'] = df['High'] - df[['Open', 'Close']].max(axis=1) # 上影线df['Lower_Shadow'] = df[['Open', 'Close']].min(axis=1) - df['Low'] # 下影线df['Body_Size'] = abs(df['Close'] - df['Open']) # 实体大小# 删除含有NaN的行df = df.dropna()return dfdef prepare_data(self, data):"""准备训练数据"""# 添加技术指标data = self.add_technical_indicators(data)# 更新特征列（包含新添加的技术指标）self.feature_columns = [col for col in data.columns if col != self.target_column]# 特征和目标变量提取features = data[self.feature_columns].values # 特征矩阵target = data[self.target_column].values.reshape(-1, 1) # 目标变量# 特征标准化features_scaled = self.feature_scaler.fit_transform(features) # 对特征进行标准化target_scaled = self.target_scaler.fit_transform(target) # 对目标变量进行标准化# 创建序列数据（将数据转换为LSTM需要的格式）X, y = [], []for i in range(self.sequence_length, len(features_scaled) - self.prediction_days + 1):# 提取历史序列X.append(features_scaled[i-self.sequence_length:i])# 提取预测目标（未来prediction_days天的数据）y.append(target_scaled[i:i+self.prediction_days].flatten())X = np.array(X) # 转换为numpy数组y = np.array(y)# 分割数据集（训练集、验证集、测试集）train_size = int(len(X) * self.train_ratio) # 计算训练集大小val_size = int(len(X) * self.val_ratio) # 计算验证集大小# 切分数据集X_train = X[:train_size] # 训练特征y_train = y[:train_size] # 训练标签X_val = X[train_size:train_size + val_size] # 验证特征y_val = y[train_size:train_size + val_size] # 验证标签X_test = X[train_size + val_size:] # 测试特征y_test = y[train_size + val_size:] # 测试标签return (X_train, y_train), (X_val, y_val), (X_test, y_test) # 返回数据集def inverse_transform_target(self, scaled_data):"""反标准化目标变量"""return self.target_scaler.inverse_transform(scaled_data.reshape(-1, 1)).flatten()class StockDataset(Dataset):"""PyTorch数据集封装，用于加载股票数据"""def __init__(self, X, y):"""初始化数据集Args:X: 特征数据 (numpy array)y: 标签数据 (numpy array)"""self.X = torch.FloatTensor(X) # 转换为PyTorch张量self.y = torch.FloatTensor(y)def __len__(self):"""返回数据集大小"""return len(self.X)def __getitem__(self, idx):"""获取单个样本Args:idx: 样本索引Returns:tuple: (特征张量, 标签张量)"""return self.X[idx], self.y[idx]class LSTMStockPredictor(nn.Module):"""LSTM股价预测模型：包含双向LSTM、注意力机制和残差连接的复杂网络结构"""def __init__(self, input_size, # 输入特征维度hidden_size=128, # LSTM隐藏层大小num_layers=3, # LSTM层数output_size=5, # 输出维度（预测天数）dropout=0.2, # Dropout概率bidirectional=True, # 是否使用双向LSTMuse_attention=True, # 是否使用注意力机制use_residual=True): # 是否使用残差连接"""初始化LSTM模型Args:input_size: 输入特征数hidden_size: 隐藏层大小num_layers: LSTM层数output_size: 输出大小（预测天数）dropout: Dropout率bidirectional: 是否使用双向LSTMuse_attention: 是否使用注意力机制use_residual: 是否使用残差连接"""super(LSTMStockPredictor, self).__init__()# 保存模型参数self.hidden_size = hidden_size #隐藏层大小self.num_layers = num_layers #LSTM层数self.bidirectional = bidirectional #是否使用双向LSTMself.use_attention = use_attention #是否使用注意力机制self.use_residual = use_residual #是否使用残差连接# 输入投影层：将输入特征映射到隐藏层维度self.input_projection = nn.Linear(input_size, hidden_size)# LSTM层：核心序列处理模块self.lstm = nn.LSTM(input_size=hidden_size, # 输入维度hidden_size=hidden_size, # 隐藏层维度num_layers=num_layers, # LSTM层数batch_first=True, # 输入数据格式(batch, seq, feature)dropout=dropout if num_layers > 1 else 0, # 除最后一层外的Dropoutbidirectional=bidirectional # 双向LSTM)# 确定LSTM输出维度（考虑双向性）lstm_output_size = hidden_size * 2 if bidirectional else hidden_size# 多头注意力机制（可选）if use_attention:self.attention = nn.MultiheadAttention(embed_dim=lstm_output_size, # 嵌入维度num_heads=8, # 头数dropout=dropout, # Dropout率batch_first=True # 输入数据格式)# 批归一化层self.batch_norm1 = nn.BatchNorm1d(lstm_output_size) # 第一层BNself.batch_norm2 = nn.BatchNorm1d(hidden_size) # 第二层BN# 全连接层self.fc1 = nn.Linear(lstm_output_size, hidden_size) # 第一层全连接self.fc2 = nn.Linear(hidden_size, hidden_size // 2) # 第二层全连接self.fc3 = nn.Linear(hidden_size // 2, output_size) # 输出层# 激活函数和正则化self.relu = nn.ReLU() # ReLU激活self.leaky_relu = nn.LeakyReLU(0.1) # LeakyReLU激活self.dropout = nn.Dropout(dropout) # Dropout层self.layer_norm = nn.LayerNorm(hidden_size) # 层归一化# 残差连接的投影层（如果启用）if use_residual:self.residual_projection = nn.Linear(lstm_output_size, hidden_size)def forward(self, x):"""前向传播Args:x: 输入张量，形状为(batch_size, sequence_length, input_size)Returns:Tensor: 输出张量，形状为(batch_size, output_size)"""batch_size = x.size(0) # 获取批次大小# 输入投影：将输入特征映射到隐藏层维度x = self.input_projection(x)# LSTM前向传播lstm_out, _ = self.lstm(x)# 注意力机制（如果启用）if self.use_attention:attn_out, _ = self.attention(lstm_out, lstm_out, lstm_out)lstm_out = lstm_out + attn_out # 残差连接# 取最后一个时间步的输出lstm_out = lstm_out[:, -1, :]# 批归一化lstm_out = self.batch_norm1(lstm_out)# 全连接层out = self.relu(self.fc1(lstm_out)) # 第一层全连接+ReLUout = self.dropout(out) # Dropout# 残差连接（如果启用）if self.use_residual:residual = self.residual_projection(lstm_out)out = out + residual# 层归一化out = self.layer_norm(out)out = self.batch_norm2(out)# 继续前向传播out = self.leaky_relu(self.fc2(out)) # 第二层全连接+LeakyReLUout = self.dropout(out) # Dropoutout = self.fc3(out) # 输出层return out# 后续类的注释类似，需要详细解释每个方法的作用和实现细节
# 由于篇幅限制，这里展示部分关键代码的注释
# 完整注释应覆盖所有类和方法，包括：
# - StockPredictionTrainer（训练器类）
# - StockPredictionSystem（系统主类）
# - 数据可视化方法
# - 模型评估指标计算
# - 模型保存/加载方法
# - 完整的训练流程
-------------------------------------------------以下是调用流程图------------------------------------------------------
