#!/usr/bin/env python # -*- coding: utf-8 -*- """ 整合股票分析工具 v1.0 包含：趋势预测(CNN-LSTM-Attention)、金叉预测、死叉预测、T+1预测(LSTM) """ import os import sys import pandas as pd import numpy as np import warnings from datetime import datetime, timedelta # 添加这行 warnings.filterwarnings('ignore') # --- 导入各模块所需库 --- try: import yfinance as yf except ImportError: yf = None print("警告: 未安装 yfinance 库，金叉、死叉、T+1 预测功能将不可用。") try: import torch import torch.nn as nn import torch.optim as optim from torch.utils.data import DataLoader, TensorDataset from sklearn.preprocessing import MinMaxScaler from sklearn.metrics import mean_squared_error, mean_absolute_error import matplotlib matplotlib.use('Agg') import matplotlib.pyplot as plt import seaborn as sns except ImportError: torch = None print("警告: 未安装 PyTorch 或相关库，趋势预测功能将不可用。") try: from tensorflow.keras.models import Sequential from tensorflow.keras.layers import LSTM, Dense from tensorflow.keras.optimizers import Adam import tensorflow as tf except ImportError: tf = None print("警告: 未安装 TensorFlow/Keras，T+1预测功能将不可用。") # 设置中文字体 plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans', 'Arial Unicode MS', 'Microsoft YaHei', 'WenQuanYi Micro Hei'] plt.rcParams['axes.unicode_minus'] = False # 解决负号 '-' 显示为方块的问题 def load_stock_data(file_path): """加载Excel股票数据""" try: # 根据文件扩展名选择引擎 _, ext = os.path.splitext(file_path) if ext.lower() == '.xls': df = pd.read_excel(file_path, engine='xlrd') elif ext.lower() == '.xlsx': df = pd.read_excel(file_path, engine='openpyxl') else: print(f"不支持的文件格式: {ext}") return None, None if df.shape[1] < 2: print("Excel文件必须至少包含两列：股票名称和股票代码") return None, None # 假设第一列为股票名称，第二列为股票代码 stock_names = df.iloc[:, 0].astype(str).tolist() stock_codes = df.iloc[:, 1].astype(str).tolist() # 确保代码是6位数字字符串 stock_codes = [str(code).strip().lstrip('0').zfill(6) for code in stock_codes] if df.empty: raise ValueError("Excel文件中没有数据。") return stock_names, stock_codes except Exception as e: print(f"加载股票数据时出错: {e}") return None, None def validate_codes(names, codes): """验证股票代码格式""" valid_names = [] valid_codes = [] invalid_items = [] for name, code in zip(names, codes): if code.isdigit() and len(code) == 6 and code[0] in ['6', '0', '3']: valid_names.append(name) valid_codes.append(code) else: invalid_items.append((name, code)) if invalid_items: print(f"发现 {len(invalid_items)} 个无效代码/名称:") for name, code in invalid_items[:10]: # 只打印前10个 print(f" - 名称: {name}, 代码: {code}") print("这些无效项将被跳过。") return valid_names, valid_codes def run_trend_prediction(stock_names, stock_codes, output_base_path): """执行趋势预测分析 (CNN-LSTM-Attention)""" print("\n--- 正在执行 趋势预测 (CNN-LSTM-Attention) 分析 ---") if not torch: print(" 跳过趋势预测，因为 PyTorch 或相关库未安装。") return trend_dir = os.path.join(output_base_path, "趋势预测") os.makedirs(trend_dir, exist_ok=True) class AttentionLayer(nn.Module): """注意力机制层""" def __init__(self, hidden_dim): super(AttentionLayer, self).__init__() self.hidden_dim = hidden_dim self.attention_weights = nn.Linear(hidden_dim, 1) def forward(self, lstm_output): attention_scores = torch.tanh(self.attention_weights(lstm_output)) attention_weights = torch.softmax(attention_scores, dim=1) weighted_output = lstm_output * attention_weights output = torch.sum(weighted_output, dim=1) return output, attention_weights.squeeze(-1) class CNNLSTMAttention(nn.Module): """CNN-LSTM-Attention模型""" def __init__(self, input_dim, cnn_filters, cnn_kernel_size, lstm_hidden_dim, dense_hidden_dim, dropout_rate=0.2): super(CNNLSTMAttention, self).__init__() self.conv1 = nn.Conv1d(in_channels=input_dim, out_channels=cnn_filters, kernel_size=cnn_kernel_size, padding=1) self.relu = nn.ReLU() self.dropout_cnn = nn.Dropout(dropout_rate) self.lstm = nn.LSTM(cnn_filters, lstm_hidden_dim, batch_first=True, bidirectional=False) self.dropout_lstm = nn.Dropout(dropout_rate) self.attention = AttentionLayer(lstm_hidden_dim) self.dense1 = nn.Linear(lstm_hidden_dim, dense_hidden_dim) self.relu_dense = nn.ReLU() self.dropout_dense = nn.Dropout(dropout_rate) self.dense2 = nn.Linear(dense_hidden_dim, 1) def forward(self, x): x = x.permute(0, 2, 1) conv_out = self.relu(self.conv1(x)) conv_out = self.dropout_cnn(conv_out) conv_out = conv_out.permute(0, 2, 1) lstm_out, _ = self.lstm(conv_out) lstm_out = self.dropout_lstm(lstm_out) attn_out, attention_weights = self.attention(lstm_out) dense_out = self.relu_dense(self.dense1(attn_out)) dense_out = self.dropout_dense(dense_out) output = self.dense2(dense_out) return output, attention_weights def get_chinese_stock_data(ticker, period="2y"): ticker = str(ticker).zfill(6) if ticker.startswith(('00', '08', '20', '30', '15', '16', '18')): ticker_yahoo = f"{ticker}.SZ" elif ticker.startswith(('5', '6', '9', '11', '13')): ticker_yahoo = f"{ticker}.SS" else: ticker_yahoo = f"{ticker}.SZ" print(f" 正在获取 {ticker_yahoo} 的数据...") try: stock = yf.Ticker(ticker_yahoo) hist = stock.history(period=period) if hist.empty: print(f" 错误：无法获取 {ticker_yahoo} 的数据，请检查股票代码是否正确。") return None print(f" 成功获取 {len(hist)} 条数据记录") return hist except Exception as e: print(f" 获取 {ticker_yahoo} 数据时发生错误: {e}") return None def add_technical_indicators(df): df['MA_5'] = df['Close'].rolling(window=5).mean() df['MA_10'] = df['Close'].rolling(window=10).mean() df['MA_20'] = df['Close'].rolling(window=20).mean() 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)) exp1 = df['Close'].ewm(span=12).mean() exp2 = df['Close'].ewm(span=26).mean() df['MACD'] = exp1 - exp2 df['MACD_signal'] = df['MACD'].ewm(span=9).mean() df['BB_middle'] = df['Close'].rolling(window=20).mean() bb_std = df['Close'].rolling(window=20).std() df['BB_upper'] = df['BB_middle'] + (bb_std * 2) df['BB_lower'] = df['BB_middle'] - (bb_std * 2) df['PCT_change'] = df['Close'].pct_change() df['Volume_MA'] = df['Volume'].rolling(window=5).mean() return df def prepare_data(df, lookback_window=60): df = add_technical_indicators(df) feature_columns = [ 'Open', 'High', 'Low', 'Close', 'Volume', 'MA_5', 'MA_10', 'MA_20', 'RSI', 'MACD', 'MACD_signal', 'BB_upper', 'BB_middle', 'BB_lower', 'PCT_change', 'Volume_MA' ] df = df.dropna() if df.shape[0] < lookback_window: print(f" 数据不足，需要至少 {lookback_window} 条记录，但只有 {df.shape[0]} 条记录") return None, None, None, None, None data = df[feature_columns].values target = df['Close'].values scaler_features = MinMaxScaler() scaler_target = MinMaxScaler() scaled_data = scaler_features.fit_transform(data) scaled_target = scaler_target.fit_transform(target.reshape(-1, 1)).flatten() X, y = [], [] for i in range(lookback_window, len(scaled_data)): X.append(scaled_data[i-lookback_window:i]) y.append(scaled_target[i]) X, y = np.array(X), np.array(y) return X, y, scaler_features, scaler_target, df.index[lookback_window:] def split_data(X, y, train_ratio=0.8, val_ratio=0.1): total_samples = len(X) train_end = int(total_samples * train_ratio) val_end = int(total_samples * (train_ratio + val_ratio)) X_train, y_train = X[:train_end], y[:train_end] X_val, y_val = X[train_end:val_end], y[train_end:val_end] X_test, y_test = X[val_end:], y[val_end:] return (X_train, y_train), (X_val, y_val), (X_test, y_test) def train_model(model, train_loader, val_loader, num_epochs=100, learning_rate=0.001): device = torch.device("cuda" if torch.cuda.is_available() else "cpu") model.to(device) criterion = nn.MSELoss() optimizer = optim.Adam(model.parameters(), lr=learning_rate, weight_decay=1e-5) scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, mode='min', patience=10, factor=0.5) train_losses = [] val_losses = [] best_val_loss = float('inf') patience_counter = 0 patience = 20 print(" 开始训练模型...") for epoch in range(num_epochs): model.train() train_loss = 0.0 for batch_x, batch_y in train_loader: batch_x, batch_y = batch_x.to(device), batch_y.to(device) optimizer.zero_grad() outputs, _ = model(batch_x) loss = criterion(outputs.squeeze(), batch_y) loss.backward() optimizer.step() train_loss += loss.item() model.eval() val_loss = 0.0 with torch.no_grad(): for batch_x, batch_y in val_loader: batch_x, batch_y = batch_x.to(device), batch_y.to(device) outputs, _ = model(batch_x) loss = criterion(outputs.squeeze(), batch_y) val_loss += loss.item() train_loss /= len(train_loader) val_loss /= len(val_loader) train_losses.append(train_loss) val_losses.append(val_loss) scheduler.step(val_loss) if (epoch + 1) % 10 == 0: print(f' Epoch [{epoch+1}/{num_epochs}], Train Loss: {train_loss:.6f}, Val Loss: {val_loss:.6f}') if val_loss < best_val_loss: best_val_loss = val_loss patience_counter = 0 torch.save(model.state_dict(), 'best_stock_model.pth') else: patience_counter += 1 if patience_counter >= patience: print(f" 验证损失连续 {patience} 轮没有改善，提前停止训练") break model.load_state_dict(torch.load('best_stock_model.pth')) return train_losses, val_losses def evaluate_model(model, test_loader, scaler_target): device = torch.device("cuda" if torch.cuda.is_available() else "cpu") model.to(device) model.eval() all_predictions = [] all_targets = [] with torch.no_grad(): for batch_x, batch_y in test_loader: batch_x = batch_x.to(device) outputs, _ = model(batch_x) predictions = outputs.cpu().numpy() all_predictions.extend(predictions) all_targets.extend(batch_y.numpy()) all_predictions = np.array(all_predictions).flatten() all_targets = np.array(all_targets) pred_rescaled = scaler_target.inverse_transform(all_predictions.reshape(-1, 1)).flatten() target_rescaled = scaler_target.inverse_transform(all_targets.reshape(-1, 1)).flatten() mse = mean_squared_error(target_rescaled, pred_rescaled) rmse = np.sqrt(mse) mae = mean_absolute_error(target_rescaled, pred_rescaled) mape = np.mean(np.abs((target_rescaled - pred_rescaled) / target_rescaled)) * 100 return pred_rescaled, target_rescaled, {'MSE': mse, 'RMSE': rmse, 'MAE': mae, 'MAPE': mape} def plot_results(y_true, y_pred, train_losses, val_losses, title, save_path=None): fig, axes = plt.subplots(2, 2, figsize=(16, 12)) axes[0, 0].plot(y_true, label='真实值', alpha=0.7) axes[0, 0].plot(y_pred, label='预测值', alpha=0.7) axes[0, 0].set_title(f'{title} - 预测 vs 真实值') axes[0, 0].legend() axes[0, 0].grid(True, alpha=0.3) residuals = y_true - y_pred axes[0, 1].scatter(range(len(residuals)), residuals, alpha=0.6) axes[0, 1].axhline(y=0, color='r', linestyle='--') axes[0, 1].set_title('残差图') axes[0, 1].grid(True, alpha=0.3) axes[1, 0].plot(train_losses, label='训练损失', alpha=0.7) axes[1, 0].plot(val_losses, label='验证损失', alpha=0.7) axes[1, 0].set_title('训练和验证损失') axes[1, 0].set_xlabel('Epoch') axes[1, 0].set_ylabel('Loss') axes[1, 0].legend() axes[1, 0].grid(True, alpha=0.3) axes[1, 1].hist(residuals, bins=50, edgecolor='black', alpha=0.7) axes[1, 1].set_title('预测误差分布') axes[1, 1].set_xlabel('误差') axes[1, 1].set_ylabel('频率') axes[1, 1].grid(True, alpha=0.3) plt.tight_layout() if save_path: os.makedirs(os.path.dirname(save_path), exist_ok=True) plt.savefig(save_path, dpi=300, bbox_inches='tight') print(f" 图表已保存到: {save_path}") plt.close(fig) def is_trading_day(date): return date.weekday() < 5 def get_next_trading_day(date): next_day = date + timedelta(days=1) while not is_trading_day(next_day): next_day += timedelta(days=1) return next_day def predict_future_prices_with_constraints(model, last_sequence, scaler_features, scaler_target, days=30, initial_price=None): device = torch.device("cuda" if torch.cuda.is_available() else "cpu") model.to(device) model.eval() predictions = [] current_seq = last_sequence.copy() if initial_price is None: last_scaled_close = current_seq[0, -1, 3] initial_price = scaler_target.inverse_transform([[last_scaled_close]])[0, 0] current_price = initial_price trading_days = 0 day_count = 0 while trading_days < days: next_date = get_next_trading_day(datetime.now() + timedelta(days=day_count)) day_count += 1 if is_trading_day(next_date): with torch.no_grad(): input_tensor = torch.FloatTensor(current_seq).to(device) pred, _ = model(input_tensor) pred_value = pred.cpu().numpy()[0, 0] pred_price_unscaled = scaler_target.inverse_transform([[pred_value]])[0, 0] max_price = current_price * 1.1 min_price = current_price * 0.9 constrained_pred_price = max(min(pred_price_unscaled, max_price), min_price) predictions.append(constrained_pred_price) current_price = constrained_pred_price last_row = current_seq[0, -1, :].copy() normalized_constrained_price = scaler_target.transform([[constrained_pred_price]])[0, 0] last_row[3] = normalized_constrained_price new_seq = np.zeros_like(current_seq) new_seq[0, :-1, :] = current_seq[0, 1:, :] new_seq[0, -1, :] = last_row current_seq = new_seq trading_days += 1 return predictions def plot_future_predictions(stock_code, stock_name, y_test, date_index, future_predictions, scaler_target, save_path=None): plt.figure(figsize=(14, 7)) recent_true_prices = scaler_target.inverse_transform(y_test[-50:].reshape(-1, 1)).flatten() recent_dates = date_index[-len(recent_true_prices):] plt.plot(recent_dates, recent_true_prices, label='近期真实价格', color='blue', linewidth=1.5) future_dates = [] current_date = date_index[-1] day_offset = 1 trading_day_count = 0 while trading_day_count < 30: future_date = current_date + timedelta(days=day_offset) if is_trading_day(future_date): future_dates.append(future_date) trading_day_count += 1 day_offset += 1 plt.plot(future_dates, future_predictions, label='未来预测价格', color='red', linestyle='--', marker='o', markersize=4) plt.title(f'{stock_code} ({stock_name}) 股票价格预测 - 近期与未来30个交易日\n(考虑涨跌幅限制±10%)') plt.xlabel('日期') plt.ylabel('价格 (元)') plt.legend() plt.grid(True, alpha=0.3) plt.xticks(rotation=45) plt.tight_layout() if save_path: os.makedirs(os.path.dirname(save_path), exist_ok=True) plt.savefig(save_path, dpi=300, bbox_inches='tight') print(f" 未来预测图表已保存到: {save_path}") plt.close() # --- 执行趋势预测 --- for i, (name, code) in enumerate(zip(stock_names, stock_codes)): print(f" 处理第 {i+1}/{len(stock_codes)} 支股票: {name} ({code})") df_stock = get_chinese_stock_data(code, period="2y") if df_stock is None or df_stock.shape[0] < 60: print(f" 跳过 {name} ({code}) - 数据不足") continue X, y, scaler_features, scaler_target, date_index = prepare_data(df_stock, lookback_window=60) if X is None or y is None: print(f" 跳过 {name} ({code}) - 数据预处理失败") continue (X_train, y_train), (X_val, y_val), (X_test, y_test) = split_data(X, y) if X_train.shape[0] == 0: print(f" 跳过 {name} ({code}) - 训练数据不足") continue train_dataset = TensorDataset(torch.FloatTensor(X_train), torch.FloatTensor(y_train)) val_dataset = TensorDataset(torch.FloatTensor(X_val), torch.FloatTensor(y_val)) test_dataset = TensorDataset(torch.FloatTensor(X_test), torch.FloatTensor(y_test)) train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True) val_loader = DataLoader(val_dataset, batch_size=32, shuffle=False) test_loader = DataLoader(test_dataset, batch_size=32, shuffle=False) input_dim = X.shape[2] model = CNNLSTMAttention( input_dim=input_dim, cnn_filters=64, cnn_kernel_size=3, lstm_hidden_dim=50, dense_hidden_dim=25, dropout_rate=0.2 ) print(f" 模型结构: {model.__class__.__name__}") train_losses, val_losses = train_model(model, train_loader, val_loader, num_epochs=100) test_predictions, test_targets, metrics = evaluate_model(model, test_loader, scaler_target) safe_name = str(name).replace('/', '_').replace('\\', '_').replace(':', '_').replace('*', '_').replace('?', '_').replace('"', '_').replace('<', '_').replace('>', '_').replace('|', '_').replace("'", "_") base_filename = f"{str(code).zfill(6)}_{safe_name}" plot_results_filename = os.path.join(trend_dir, f"{base_filename}_analysis.png") plot_results(test_targets, test_predictions, train_losses, val_losses, f"{str(code).zfill(6)} ({name})", save_path=plot_results_filename) last_real_price = scaler_target.inverse_transform(y_test[-1:].reshape(-1, 1))[0, 0] last_sequence = X_test[-1:] future_predictions = predict_future_prices_with_constraints( model, last_sequence, scaler_features, scaler_target, days=30, initial_price=last_real_price ) future_plot_filename = os.path.join(trend_dir, f"{base_filename}_future_prediction.png") plot_future_predictions(str(code).zfill(6), name, y_test, date_index, future_predictions, scaler_target, save_path=future_plot_filename) print(f" {name} ({code}) 趋势预测完成。") print(f"趋势预测分析完成，结果保存在: {trend_dir}") def run_golden_cross_prediction(stock_names, stock_codes, output_base_path): """执行金叉预测分析""" print("\n--- 正在执行 金叉预测 分析 ---") if not yf: print(" 跳过金叉预测，因为 yfinance 库未安装。") return gc_dir = os.path.join(output_base_path, "金叉预测") os.makedirs(gc_dir, exist_ok=True) def calculate_macd(data, short_period=12, long_period=26, signal_period=9): ema_short = data.ewm(span=short_period).mean() ema_long = data.ewm(span=long_period).mean() dif = ema_short - ema_long dea = dif.ewm(span=signal_period).mean() histogram = (dif - dea) * 2 return dif, dea, histogram def detect_golden_cross(dif_series, dea_series): golden_cross_signal = (dif_series > dea_series) & (dif_series.shift(1) <= dea_series.shift(1)) return golden_cross_signal def get_last_n_trading_days(df, n=20): date_index = df.index date_parts = pd.Index([d.date() if hasattr(d, 'date') else d for d in date_index]) all_dates_unique = date_parts.unique() trading_dates = [date_obj for date_obj in all_dates_unique if pd.Timestamp(date_obj).weekday() < 5] last_n_trading_dates = trading_dates[-n:] mask = date_parts.isin(last_n_trading_dates) last_n_trading_data = df[mask] last_n_trading_data.sort_index(inplace=True) return last_n_trading_data # --- 执行金叉预测 --- for name, code in zip(stock_names, stock_codes): print(f" 处理股票: {name} ({code})") yf_stock_code = f"{code}.SS" if code.startswith(('5', '6', '9')) else f"{code}.SZ" hist = None try: print(f" 正在获取 {yf_stock_code} 的数据...") ticker = yf.Ticker(yf_stock_code) hist = ticker.history(period="5y") if hist.empty: print(f" 无法获取 {yf_stock_code} 的数据，跳过。") continue except Exception as e: print(f" 获取 {yf_stock_code} 数据时出错: {e}") continue close_prices = hist['Close'] dif, dea, histogram = calculate_macd(close_prices) golden_cross_points = detect_golden_cross(dif, dea) analysis_df = pd.DataFrame({ 'Close': close_prices, 'DIF': dif, 'DEA': dea, 'Histogram': histogram, 'Golden_Cross_Signal': golden_cross_points }, index=hist.index) all_golden_cross_signals = analysis_df[analysis_df['Golden_Cross_Signal']] last_20_trading_days_df = get_last_n_trading_days(analysis_df, n=20) safe_name = str(name).replace('/', '_').replace('\\', '_').replace(':', '_').replace('*', '_').replace('?', '_').replace('"', '_').replace('<', '_').replace('>', '_').replace('|', '_').replace("'", "_") base_filename = f"{str(code).zfill(6)}_{safe_name}" # 保存历史金叉信号 golden_cross_excel_filename = os.path.join(gc_dir, f"{base_filename}_golden_cross_signals.xlsx") if not all_golden_cross_signals.empty: gc_to_save = all_golden_cross_signals.reset_index()[['Date', 'Close', 'DIF', 'DEA', 'Histogram']] gc_to_save_fixed = gc_to_save.copy() gc_to_save_fixed['Date'] = gc_to_save_fixed['Date'].dt.tz_localize(None) gc_to_save_fixed.to_excel(golden_cross_excel_filename, index=False, engine='openpyxl') print(f" ✅ 历史金叉信号已保存至: {golden_cross_excel_filename}") # 保存最近20天数据 last_20_data_excel_filename = os.path.join(gc_dir, f"{base_filename}_last_20_trading_days_data.xlsx") if not last_20_trading_days_df.empty: l20_to_save = last_20_trading_days_df.reset_index()[['Date', 'Close', 'DIF', 'DEA', 'Histogram']] l20_to_save_fixed = l20_to_save.copy() l20_to_save_fixed['Date'] = l20_to_save_fixed['Date'].dt.tz_localize(None) l20_to_save_fixed.to_excel(last_20_data_excel_filename, index=False, engine='openpyxl') print(f" ✅ 最近20天数据已保存至: {last_20_data_excel_filename}") # 绘制图表 plot_filename = os.path.join(gc_dir, f"{base_filename}_last_20_trading_days_plot.png") if not last_20_trading_days_df.empty: fig, ax = plt.subplots(figsize=(14, 8)) ax.plot(last_20_trading_days_df.index, last_20_trading_days_df['Close'], label='Close Price', marker='o', linewidth=1.2) ax.plot(last_20_trading_days_df.index, last_20_trading_days_df['DIF'], label='DIF', marker='x', linewidth=1.2) ax.plot(last_20_trading_days_df.index, last_20_trading_days_df['DEA'], label='DEA', marker='^', linewidth=1.2) ax.bar(last_20_trading_days_df.index, last_20_trading_days_df['Histogram'], label='Histogram', alpha=0.3, width=0.8) ax.set_title(f'{code} - {name}\nLast 20 Trading Days Data (Golden Cross Analysis)') ax.set_xlabel('Date') ax.set_ylabel('Value') ax.legend() ax.grid(True, linestyle='--', alpha=0.6) import matplotlib.dates as mdates ax.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) ax.xaxis.set_major_locator(mdates.AutoDateLocator()) plt.xticks(rotation=45, ha='right') plt.tight_layout() plt.savefig(plot_filename, dpi=300) plt.close() print(f" ✅ 折线图已保存至: {plot_filename}") print(f"金叉预测分析完成，结果保存在: {gc_dir}") def run_death_cross_prediction(stock_names, stock_codes, output_base_path): """执行死叉预测分析""" print("\n--- 正在执行 死叉预测 分析 ---") if not yf: print(" 跳过死叉预测，因为 yfinance 库未安装。") return dc_dir = os.path.join(output_base_path, "死叉预测") os.makedirs(dc_dir, exist_ok=True) def calculate_macd(data, short_period=12, long_period=26, signal_period=9): ema_short = data.ewm(span=short_period).mean() ema_long = data.ewm(span=long_period).mean() dif = ema_short - ema_long dea = dif.ewm(span=signal_period).mean() histogram = (dif - dea) * 2 return dif, dea, histogram def detect_death_cross(dif_series, dea_series): death_cross_signal = (dif_series < dea_series) & (dif_series.shift(1) >= dea_series.shift(1)) return death_cross_signal def get_last_n_trading_days(df, n=20): date_index = df.index date_parts = pd.Index([d.date() if hasattr(d, 'date') else d for d in date_index]) all_dates_unique = date_parts.unique() trading_dates = [date_obj for date_obj in all_dates_unique if pd.Timestamp(date_obj).weekday() < 5] last_n_trading_dates = trading_dates[-n:] mask = date_parts.isin(last_n_trading_dates) last_n_trading_data = df[mask] last_n_trading_data.sort_index(inplace=True) return last_n_trading_data # --- 执行死叉预测 --- for name, code in zip(stock_names, stock_codes): print(f" 处理股票: {name} ({code})") yf_stock_code = f"{code}.SS" if code.startswith(('5', '6', '9')) else f"{code}.SZ" hist = None try: print(f" 正在获取 {yf_stock_code} 的数据...") ticker = yf.Ticker(yf_stock_code) hist = ticker.history(period="5y") if hist.empty: print(f" 无法获取 {yf_stock_code} 的数据，跳过。") continue except Exception as e: print(f" 获取 {yf_stock_code} 数据时出错: {e}") continue close_prices = hist['Close'] dif, dea, histogram = calculate_macd(close_prices) death_cross_points = detect_death_cross(dif, dea) analysis_df = pd.DataFrame({ 'Close': close_prices, 'DIF': dif, 'DEA': dea, 'Histogram': histogram, 'Death_Cross_Signal': death_cross_points }, index=hist.index) all_death_cross_signals = analysis_df[analysis_df['Death_Cross_Signal']] last_20_trading_days_df = get_last_n_trading_days(analysis_df, n=20) safe_name = str(name).replace('/', '_').replace('\\', '_').replace(':', '_').replace('*', '_').replace('?', '_').replace('"', '_').replace('<', '_').replace('>', '_').replace('|', '_').replace("'", "_") base_filename = f"{str(code).zfill(6)}_{safe_name}" # 保存历史死叉信号 death_cross_excel_filename = os.path.join(dc_dir, f"{base_filename}_death_cross_signals.xlsx") if not all_death_cross_signals.empty: dc_to_save = all_death_cross_signals.reset_index()[['Date', 'Close', 'DIF', 'DEA', 'Histogram']] dc_to_save_fixed = dc_to_save.copy() dc_to_save_fixed['Date'] = dc_to_save_fixed['Date'].dt.tz_localize(None) dc_to_save_fixed.to_excel(death_cross_excel_filename, index=False, engine='openpyxl') print(f" ✅ 历史死叉信号已保存至: {death_cross_excel_filename}") # 保存最近20天数据 last_20_data_excel_filename = os.path.join(dc_dir, f"{base_filename}_last_20_trading_days_data.xlsx") if not last_20_trading_days_df.empty: l20_to_save = last_20_trading_days_df.reset_index()[['Date', 'Close', 'DIF', 'DEA', 'Histogram']] l20_to_save_fixed = l20_to_save.copy() l20_to_save_fixed['Date'] = l20_to_save_fixed['Date'].dt.tz_localize(None) l20_to_save_fixed.to_excel(last_20_data_excel_filename, index=False, engine='openpyxl') print(f" ✅ 最近20天数据已保存至: {last_20_data_excel_filename}") # 绘制图表 plot_filename = os.path.join(dc_dir, f"{base_filename}_last_20_trading_days_plot.png") if not last_20_trading_days_df.empty: fig, ax = plt.subplots(figsize=(14, 8)) ax.plot(last_20_trading_days_df.index, last_20_trading_days_df['Close'], label='Close Price', marker='o', linewidth=1.2) ax.plot(last_20_trading_days_df.index, last_20_trading_days_df['DIF'], label='DIF', marker='x', linewidth=1.2) ax.plot(last_20_trading_days_df.index, last_20_trading_days_df['DEA'], label='DEA', marker='^', linewidth=1.2) ax.bar(last_20_trading_days_df.index, last_20_trading_days_df['Histogram'], label='Histogram', alpha=0.3, width=0.8) ax.set_title(f'{code} - {name}\nLast 20 Trading Days Data (Death Cross Analysis)') ax.set_xlabel('Date') ax.set_ylabel('Value') ax.legend() ax.grid(True, linestyle='--', alpha=0.6) import matplotlib.dates as mdates ax.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) ax.xaxis.set_major_locator(mdates.AutoDateLocator()) plt.xticks(rotation=45, ha='right') plt.tight_layout() plt.savefig(plot_filename, dpi=300) plt.close() print(f" ✅ 折线图已保存至: {plot_filename}") print(f"死叉预测分析完成，结果保存在: {dc_dir}") def run_t_plus_one_prediction(stock_names, stock_codes, output_base_path): """执行T+1预测分析 (LSTM)""" print("\n--- 正在执行 T+1预测 (LSTM) 分析 ---") if not tf: print(" 跳过T+1预测，因为 TensorFlow/Keras 库未安装。") return t1_dir = os.path.join(output_base_path, "T+1预测") os.makedirs(t1_dir, exist_ok=True) from datetime import date, timedelta, datetime from concurrent.futures import ThreadPoolExecutor from sklearn.preprocessing import MinMaxScaler from tensorflow.keras.models import Sequential from tensorflow.keras.layers import LSTM, Dense from tensorflow.keras.optimizers import Adam import matplotlib matplotlib.use('Agg') import matplotlib.pyplot as plt # 导入 tqdm 用于进度条 try: from tqdm import tqdm except ImportError: print("警告: 未安装 tqdm，将不显示进度条。") from tqdm import trange as tqdm_range # 定义一个简单的替代函数 def tqdm(iterable, desc="", **kwargs): for item in iterable: yield item def tqdm_range(n, desc="", **kwargs): for i in range(n): yield i else: from tqdm import trange as tqdm_range def download_train_predict(args): code, start_date, end_date, data_path, predict_path, stock_name = args if code.startswith("6"): yahoo_ticker = f"{code}.SS" elif code.startswith(("0", "3")): yahoo_ticker = f"{code}.SZ" else: print(f"Invalid Chinese stock code format: {code}") return False # Download data start_dt = datetime.strptime(start_date, "%Y%m%d").strftime("%Y-%m-%d") end_dt = datetime.strptime(end_date, "%Y%m%d").strftime("%Y-%m-%d") try: ticker = yf.Ticker(yahoo_ticker) hist = ticker.history(start=start_dt, end=end_dt) if hist.empty: print(f"No data found for ticker {yahoo_ticker}") return False hist_reset = hist.reset_index() hist_mapped = pd.DataFrame({ 'DT': hist_reset['Date'].dt.strftime('%Y-%m-%d'), 'CODE': code, 'NAME': stock_name, 'TCLOSE': hist_reset['Close'], 'HIGH': hist_reset['High'], 'LOW': hist_reset['Low'], 'TOPEN': hist_reset['Open'], 'LCLOSE': hist_reset['Close'].shift(1), 'CHG': hist_reset['Close'] - hist_reset['Open'], 'PCHG': ((hist_reset['Close'] - hist_reset['Open']) / hist_reset['Open']) * 100, 'VOTURNOVER': hist_reset['Volume'], }) hist_mapped.dropna(subset=['LCLOSE'], inplace=True) hist_mapped.reset_index(drop=True, inplace=True) csv_path = os.path.join(data_path, f"{code}.csv") hist_mapped.to_csv(csv_path, index=False, encoding='utf-8') print(f"Downloaded data for {yahoo_ticker}, saved as {code}.csv") except Exception as e: print(f"Error downloading data for {yahoo_ticker}: {e}") return False # Train and predict csv_file = os.path.join(data_path, f'{code}.csv') if not os.path.exists(csv_file): return False df = pd.read_csv(csv_file) if len(df) < 360: print(f"Insufficient data for {code}, length: {len(df)}") return False df = df[['DT', 'TCLOSE']].copy() df = df.set_index(['DT']) df.index = pd.to_datetime(df.index) df.sort_index(inplace=True) full_date_range = pd.date_range(start=df.index.min(), end=datetime.strptime(end_date, "%Y%m%d"), freq='D') df_full = df.reindex(full_date_range) df_full['TCLOSE'].fillna(method='ffill', inplace=True) df_full.fillna(method='bfill', inplace=True) prices_clean = df_full['TCLOSE'].dropna().values if len(prices_clean) < 60 + 1: print(f"Not enough clean data points for {code}") return False prices = prices_clean.reshape(-1, 1) scaler = MinMaxScaler(feature_range=(0, 1)) scaled_prices = scaler.fit_transform(prices) X, y = [], [] sequence_length = 60 for i in range(sequence_length, len(scaled_prices)): X.append(scaled_prices[i-sequence_length:i, 0]) y.append(scaled_prices[i, 0]) X, y = np.array(X), np.array(y) if len(X) == 0: print(f"No sequences could be formed for {code}") return False X = X.reshape((X.shape[0], X.shape[1], 1)) split_idx = int(len(X) * 0.8) X_train, X_test = X[:split_idx], X[split_idx:] y_train, y_test = y[:split_idx], y[split_idx:] model = Sequential([ LSTM(units=50, return_sequences=True, input_shape=(X_train.shape[1], 1)), LSTM(units=50, return_sequences=False), Dense(units=25), Dense(units=1) ]) model.compile(optimizer=Adam(learning_rate=0.001), loss='mean_squared_error') # 修改为 verbose=1 以显示训练进度 model.fit(X_train, y_train, batch_size=1, epochs=10, verbose=1) last_sequence = scaled_prices[-sequence_length:].reshape(1, sequence_length, 1) predicted_scaled = model.predict(last_sequence, verbose=0) predicted_price_scaled = predicted_scaled[0, 0] predicted_price = scaler.inverse_transform([[predicted_price_scaled]])[0, 0] predict_days = 2 predicted_prices = [predicted_price] current_sequence = last_sequence.copy() for _ in range(1, predict_days): next_pred_scaled = model.predict(current_sequence, verbose=0)[0, 0] next_pred = scaler.inverse_transform([[next_pred_scaled]])[0, 0] predicted_prices.append(next_pred) new_val_scaled = next_pred_scaled current_sequence = np.roll(current_sequence, -1, axis=1) current_sequence[0, -1, 0] = new_val_scaled # --- 新增：获取最近10个交易日的实际价格 --- last_10_actual_dates = df_full['TCLOSE'].tail(10).index.strftime('%Y-%m-%d').tolist() last_10_actual_prices = df_full['TCLOSE'].tail(10).tolist() last_10_df = pd.DataFrame({'DT': last_10_actual_dates, 'Actual_Price': last_10_actual_prices}) # 获取最近一个交易日的实际收盘价 last_actual_date = pd.to_datetime(df_full.index[-1]) # 预测下一个交易日的价格 # 找到下一个交易日 next_trading_day = last_actual_date + timedelta(days=1) while next_trading_day.weekday() >= 5: # 5 = Saturday, 6 = Sunday next_trading_day += timedelta(days=1) future_dates = [next_trading_day.strftime('%Y-%m-%d')] # 只取第一个预测值作为T+1预测 t_plus_1_price = predicted_prices[0] prediction_df = pd.DataFrame({'DT': future_dates, 'Predicted_Price': [t_plus_1_price]}) print(f"\n--- {code} ({stock_name}) 近10个交易日实际股价与T+1预测 ---") print("实际价格 (近10日):") for date, price in zip(last_10_actual_dates, last_10_actual_prices): print(f" {date}: {price:.2f}") print(f"\nT+1预测价格 ({future_dates[0]}): {t_plus_1_price:.2f}") print(f"前一日收盘价 ({last_10_actual_dates[-1]}): {last_10_actual_prices[-1]:.2f}") change = t_plus_1_price - last_10_actual_prices[-1] change_pct = (change / last_10_actual_prices[-1]) * 100 print(f"预测涨跌: {change:+.2f} ({change_pct:+.2f}%)\n") # --- 绘制图表 --- # 包含近10日实际价格和1日预测价格 combined_df = pd.concat([last_10_df, prediction_df[['DT', 'Predicted_Price']]], axis=1, sort=False) all_dates_for_plot = last_10_actual_dates + future_dates all_actual_for_plot = last_10_actual_prices all_predicted_for_plot = [np.nan] * 10 + [t_plus_1_price] plt.figure(figsize=(12, 6)) plt.plot(pd.to_datetime(all_dates_for_plot[:-1]), all_actual_for_plot, label='实际价格 (近10日)', marker='o', linestyle='-', color='blue') plt.plot(pd.to_datetime(all_dates_for_plot[-1:]), all_predicted_for_plot[-1:], label='预测价格 (T+1)', marker='x', linestyle='--', color='red', markersize=8) plt.title(f'{code} ({stock_name}) 股价预测 - 近10日实际与T+1预测') plt.xlabel('日期') plt.ylabel('价格') plt.legend() plt.grid(True, linestyle='--', alpha=0.6) plt.xticks(rotation=45) plt.tight_layout() output_image_path = os.path.join(predict_path, f"{code}_{stock_name}_prediction_chart.png") plt.savefig(output_image_path) plt.close() print(f" Prediction chart saved for {code} ({stock_name}) to {output_image_path}") return True # --- 执行T+1预测 --- start_date = (date.today() - timedelta(days=720)).strftime("%Y%m%d") end_date_str = date.today().strftime("%Y%m%d") temp_data_path = os.path.join(end_date_str, "data") os.makedirs(temp_data_path, exist_ok=True) args_list = [(code, start_date, end_date_str, temp_data_path, t1_dir, name) for code, name in zip(stock_codes, stock_names)] # 使用 tqdm 显示进度条 successful_predictions = 0 print(f"开始处理 {len(args_list)} 支股票的T+1预测...") for arg in tqdm(args_list, desc="Processing Stocks"): if download_train_predict(arg): successful_predictions += 1 print(f"T+1预测分析完成，结果保存在: {t1_dir}") print(f" 成功处理: {successful_predictions}/{len(stock_codes)} 支股票") def main(): print("欢迎使用整合版股票分析工具！") print("="*50) # 1. 获取股票数据文件路径 while True: excel_file_path = input("\n请输入包含股票名称和代码的Excel文件路径 (例如: C:\\stock_list.xlsx)\n注意: Excel文件应有两列，第一列为股票名称，第二列为股票代码 (6位数字)。\n路径: ").strip() if os.path.isfile(excel_file_path): break else: print(f"错误: 找不到文件 '{excel_file_path}'，请重新输入。") # 2. 获取结果保存路径 while True: output_path = input("\n请输入股票分析结果的保存文件夹路径 (例如: D:\\analysis_results):\n路径: ").strip() if not os.path.exists(output_path): try: os.makedirs(output_path, exist_ok=True) print(f"已创建文件夹: {output_path}") except OSError as e: print(f"创建文件夹失败: {e}，请重新输入路径。") continue break # 3. 加载并验证数据 print("\n正在加载和验证股票数据...") stock_names, stock_codes = load_stock_data(excel_file_path) if stock_names is None or stock_codes is None: print("加载股票数据失败，程序退出。") sys.exit(1) print(f"成功从文件 {excel_file_path} 中读取到 {len(stock_codes)} 个代码/名称对。") # 验证代码格式 valid_names, valid_codes = validate_codes(stock_names, stock_codes) if not valid_codes: print("没有提供有效的代码。程序退出。") sys.exit(1) print(f"经过验证，共有 {len(valid_codes)} 个有效代码/名称对。") # 4. 依次执行所有分析 run_trend_prediction(valid_names, valid_codes, output_path) run_golden_cross_prediction(valid_names, valid_codes, output_path) run_death_cross_prediction(valid_names, valid_codes, output_path) run_t_plus_one_prediction(valid_names, valid_codes, output_path) print("\n" + "="*50) print("所有分析任务已完成！") print(f"分析结果已保存在: {output_path}") print("- 趋势预测结果: {}/趋势预测".format(output_path)) print("- 金叉预测结果: {}/金叉预测".format(output_path)) print("- 死叉预测结果: {}/死叉预测".format(output_path)) print("- T+1预测结果: {}/T+1预测".format(output_path)) print("程序结束。") if __name__ == "__main__": main()