AI量化投资：从数据到策略的完整实战指南

import akshare as ak
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from sklearn.preprocessing import StandardScaler
import warnings
warnings.filterwarnings('ignore')

# 获取股票日线数据
stock_code = "600519"
df = ak.stock_zh_a_hist(symbol=stock_code, period="daily", start_date="2015-01-01", end_date="2025-01-01", adjust="qfq")
df.columns = ['date', 'open', 'close', 'high', 'low', 'volume', 'amount', 'amplitude', 'pct_change', 'change', 'turnover']
df['date'] = pd.to_datetime(df['date'])
df.sort_values('date', inplace=True)
df.reset_index(drop=True, inplace=True)

# 检查缺失
print(df.isnull().sum())
# 使用前复权价格，已包含分红除权调整

# 可视化
plt.figure(figsize=(12,5))
plt.plot(df['date'], df['close'], label='Close Price')
plt.title('Moutai (600519) Close Price')
plt.legend()
plt.show()

def add_technical_features(df):
    # 收益率
    df['return_1d'] = df['close'].pct_change()
    df['return_5d'] = df['close'].pct_change(5)
    df['return_20d'] = df['close'].pct_change(20)
    
    # 波动率 (滚动标准差)
    df['volatility_5d'] = df['return_1d'].rolling(5).std()
    df['volatility_20d'] = df['return_1d'].rolling(20).std()
    
    # 成交量比率
    df['volume_ratio'] = df['volume'] / df['volume'].rolling(20).mean()
    
    # RSI (14天)
    delta = df['close'].diff()
    gain = (delta.where(delta > 0, 0)).rolling(14).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
    rs = gain / loss
    df['rsi_14'] = 100 - (100 / (1 + rs))
    
    # MACD (快速12, 慢速26, 信号9)
    exp1 = df['close'].ewm(span=12, adjust=False).mean()
    exp2 = df['close'].ewm(span=26, adjust=False).mean()
    df['macd'] = exp1 - exp2
    df['macd_signal'] = df['macd'].ewm(span=9, adjust=False).mean()
    df['macd_hist'] = df['macd'] - df['macd_signal']
    
    # 价格位置 (当前价在20日区间的位置)
    high_20 = df['high'].rolling(20).max()
    low_20 = df['low'].rolling(20).min()
    df['price_position'] = (df['close'] - low_20) / (high_20 - low_20 + 1e-6)
    
    return df

df = add_technical_features(df)
# 删除前面因滚动产生的NaN
df.dropna(inplace=True)
print(df.shape)

from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.callbacks import EarlyStopping

# 选取特征列
feature_cols = ['return_1d', 'return_5d', 'return_20d', 'volatility_5d', 'volatility_20d',
                'volume_ratio', 'rsi_14', 'macd', 'macd_signal', 'macd_hist', 'price_position']

# 标签：次日收益率正负
df['target'] = (df['return_1d'].shift(-1) > 0).astype(int)

# 去掉标签为NaN的行（最后一天）
df_model = df.dropna(subset=['target']).copy()

# 归一化特征
scaler = StandardScaler()
X_scaled = scaler.fit_transform(df_model[feature_cols])

# 构建时序样本：用过去WINDOW天的数据预测下一天
WINDOW = 20  # 回溯天数
X_seq, y_seq = [], []
for i in range(WINDOW, len(df_model) - 1):  # 保留第i+1天作为标签 (已有target)
    X_seq.append(X_scaled[i-WINDOW:i])
    y_seq.append(df_model['target'].iloc[i])  # 注意：target是当天对第二天的预测，所以i对应的target已经是对i+1的预测

X_seq = np.array(X_seq)
y_seq = np.array(y_seq)

print(f"X shape: {X_seq.shape}, y shape: {y_seq.shape}")

# 划分训练集（80%）、验证集（10%）、测试集（10%）按时间顺序
split1 = int(len(X_seq) * 0.8)
split2 = int(len(X_seq) * 0.9)
X_train, X_val, X_test = X_seq[:split1], X_seq[split1:split2], X_seq[split2:]
y_train, y_val, y_test = y_seq[:split1], y_seq[split1:split2], y_seq[split2:]

print(f"Train: {X_train.shape}, Val: {X_val.shape}, Test: {X_test.shape}")

model = Sequential()
model.add(LSTM(64, input_shape=(WINDOW, len(feature_cols)), return_sequences=True))
model.add(Dropout(0.2))
model.add(LSTM(32, return_sequences=False))
model.add(Dropout(0.2))
model.add(Dense(16, activation='relu'))
model.add(Dense(1, activation='sigmoid'))

model.compile(optimizer=Adam(learning_rate=0.001), loss='binary_crossentropy', metrics=['accuracy'])
model.summary()

# 早停
early_stop = EarlyStopping(monitor='val_loss', patience=10, restore_best_weights=True)

history = model.fit(X_train, y_train, epochs=100, batch_size=32, 
                    validation_data=(X_val, y_val), callbacks=[early_stop], verbose=1)

# 在测试集上预测概率
y_pred_prob = model.predict(X_test).flatten()
y_pred_class = (y_pred_prob > 0.5).astype(int)

# 计算准确率（作为参考，但不是核心）
from sklearn.metrics import accuracy_score, classification_report
print("Test Accuracy:", accuracy_score(y_test, y_pred_class))
print(classification_report(y_test, y_pred_class))

# 生成交易信号：将概率映射为仓位（阈值可调）
# 简化：预测涨则买入（全仓），预测跌则空仓（或做空，这里仅做多）
signal = np.where(y_pred_prob > 0.55, 1, 0)  # 提高阈值减少交易次数

# 获取测试集对应的日期和真实收益率（用于回测）
test_start = split2 + WINDOW  # 因为X_seq索引偏移
test_dates = df_model.index[test_start:test_start + len(y_pred_prob)]  # 注意对齐

# 实际收益率序列（测试期间每日收益）
test_returns = df_model['return_1d'].iloc[test_start:test_start + len(y_pred_prob)].values

# 策略信号与日期对齐
# signal长度等于预测样本数，但每日收益率数量少一天？仔细检查：
# 我们预测的是第i天的标签（对应第i+1天收益率），所以信号应在第i天收盘后使用，用于第i+1天的交易。
# 简化：信号序列从第i天开始，对应第i+1天的收益
# 构造一个DataFrame
backtest_df = pd.DataFrame({
    'date': df_model['date'].iloc[test_start:test_start + len(y_pred_prob)],
    'return': test_returns,
    'signal': signal
})
# 策略收益 = signal * 当日收益率（因为信号在当天生效，用于当天收盘价买入？实际上信号是预测次日，所以应该用次日收益）
# 更准确：信号在第t天产生，用于第t+1天的收益。
# 所以将信号shift(1)
backtest_df['strategy_return'] = backtest_df['signal'].shift(1) * backtest_df['return']
# 去掉首日NaN
backtest_df.dropna(inplace=True)

# 计算累计收益
backtest_df['cum_return'] = (1 + backtest_df['return']).cumprod()
backtest_df['cum_strategy'] = (1 + backtest_df['strategy_return']).cumprod()

# 计算风险指标
total_return = backtest_df['cum_strategy'].iloc[-1] - 1
annual_return = (1 + total_return) ** (252 / len(backtest_df)) - 1
daily_std = backtest_df['strategy_return'].std()
sharpe_ratio = np.sqrt(252) * daily_std  # 无风险利率为0
# 最大回撤
cum_wealth = backtest_df['cum_strategy']
rolling_max = cum_wealth.expanding().max()
drawdown = (cum_wealth - rolling_max) / rolling_max
max_drawdown = drawdown.min()

print(f"总收益率: {total_return:.2%}")
print(f"年化收益率: {annual_return:.2%}")
print(f"夏普比率: {sharpe_ratio:.3f}")
print(f"最大回撤: {max_drawdown:.2%}")

# 绘图
plt.figure(figsize=(14,6))
plt.plot(backtest_df['date'], backtest_df['cum_strategy'], label='Strategy')
plt.plot(backtest_df['date'], backtest_df['cum_return'], label='Buy & Hold')
plt.title('Backtest Performance')
plt.legend()
plt.grid()
plt.show()

import gym
from gym import spaces
from tensorflow.keras import layers, models

class TradingEnv(gym.Env):
    def __init__(self, data, initial_capital=100000):
        super().__init__()
        self.data = data
        self.initial_capital = initial_capital
        self.current_step = 0
        self.action_space = spaces.Discrete(3)  # 0:持有, 1:买入, 2:卖出
        self.observation_space = spaces.Box(low=-np.inf, high=np.inf, shape=(len(feature_cols)+1,), dtype=np.float32)  # 特征+持仓
        
    def reset(self):
        self.current_step = 0
        self.capital = self.initial_capital
        self.position = 0  # 持股数量
        return self._get_obs()
    
    def _get_obs(self):
        features = self.data.iloc[self.current_step][feature_cols].values
        return np.append(features, [self.position])
    
    def step(self, action):
        # 简化：价格变化，执行动作
        # ... 实现买卖逻辑，计算奖励
        pass