机器学习在量化交易中的应用：从LSTM到 Transformer 的完整实战指南

import numpy as np
import pandas as pd
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout

def build_lstm_model(input_shape, units=64):
    """
    构建LSTM模型用于价格预测
    
    Parameters:
    -----------
    input_shape : tuple
        (timesteps, features)
    units : int
        LSTM单元数量
    
    Returns:
    --------
    model : keras.Model
        编译好的LSTM模型
    """
    model = Sequential([
        LSTM(units, return_sequences=True, input_shape=input_shape),
        Dropout(0.2),
        LSTM(units // 2, return_sequences=False),
        Dropout(0.2),
        Dense(32, activation='relu'),
        Dense(1, activation='linear')  # 预测下一期收益率
    ])
    
    model.compile(
        optimizer='adam',
        loss='mse',
        metrics=['mae', 'mape']
    )
    
    return model

# 数据预处理
def prepare_lstm_data(price_data, lookback=20):
    """
    将时间序列转换为监督学习格式
    
    Parameters:
    -----------
    price_data : pd.DataFrame
        包含OHLCV数据
    lookback : int
        回看窗口长度
    
    Returns:
    --------
    X : np.array
        形状为 (samples, lookback, features)
    y : np.array
        形状为 (samples,)
    """
    features = ['open', 'high', 'low', 'close', 'volume']
    data = price_data[features].values
    
    X, y = [], []
    for i in range(lookback, len(data)):
        X.append(data[i-lookback:i])
        y.append(data[i, 3])  # 预测收盘价
    
    return np.array(X), np.array(y)

from tensorflow.keras.layers import MultiHeadAttention, LayerNormalization
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Dense, GlobalAveragePooling1D

def build_transformer_model(input_shape, num_heads=4, ff_dim=64):
    """
    构建Transformer模型用于量化交易
    
    Transformer优势：
    1. 并行计算效率高
    2. 捕捉长期依赖（不受梯度消失影响）
    3. 可解释性（Attention权重）
    """
    inputs = Input(shape=input_shape)
    
    # Multi-Head Attention
    attention_output = MultiHeadAttention(
        num_heads=num_heads,
        key_dim=input_shape[-1]
    )(inputs, inputs)
    
    # Add & Norm
    attention_output = LayerNormalization()(inputs + attention_output)
    
    # Feed Forward
    ff_output = Dense(ff_dim, activation='relu')(attention_output)
    ff_output = Dense(input_shape[-1])(ff_output)
    
    # Add & Norm
    outputs = LayerNormalization()(attention_output + ff_output)
    
    # Global Pooling + Output
    outputs = GlobalAveragePooling1D()(outputs)
    outputs = Dense(32, activation='relu')(outputs)
    outputs = Dense(1, activation='sigmoid')(outputs)  # 上涨概率
    
    model = Model(inputs=inputs, outputs=outputs)
    model.compile(
        optimizer='adam',
        loss='binary_crossentropy',
        metrics=['accuracy']
    )
    
    return model

from sklearn.ensemble import RandomForestClassifier
from xgboost import XGBClassifier
import lightgbm as lgb

class EnsembleFactorModel:
    """
    集成因子选股模型
    
    结合：
    1. Random Forest：捕捉非线性交互
    2. XGBoost：处理缺失值 + 正则化
    3. LightGBM：高效训练 + 类别特征
    """
    
    def __init__(self, n_estimators=500):
        self.rf = RandomForestClassifier(
            n_estimators=n_estimators,
            max_depth=10,
            min_samples_split=20,
            random_state=42
        )
        
        self.xgb = XGBClassifier(
            n_estimators=n_estimators,
            max_depth=6,
            learning_rate=0.01,
            subsample=0.8,
            colsample_bytree=0.8,
            random_state=42
        )
        
        self.lgb = lgb.LGBMClassifier(
            n_estimators=n_estimators,
            num_leaves=31,
            learning_rate=0.01,
            feature_fraction=0.8,
            random_state=42
        )
        
        self.meta_model = LogisticRegression()
    
    def fit(self, X, y):
        """三层集成：基模型 + 元模型"""
        # 第一层：训练基模型
        self.rf.fit(X, y)
        self.xgb.fit(X, y)
        self.lgb.fit(X, y)
        
        # 第二层：用基模型预测结果作为新特征
        rf_pred = self.rf.predict_proba(X)[:, 1].reshape(-1, 1)
        xgb_pred = self.xgb.predict_proba(X)[:, 1].reshape(-1, 1)
        lgb_pred = self.lgb.predict_proba(X)[:, 1].reshape(-1, 1)
        
        X_meta = np.hstack([rf_pred, xgb_pred, lgb_pred])
        self.meta_model.fit(X_meta, y)
    
    def predict(self, X):
        """预测"""
        rf_pred = self.rf.predict_proba(X)[:, 1].reshape(-1, 1)
        xgb_pred = self.xgb.predict_proba(X)[:, 1].reshape(-1, 1)
        lgb_pred = self.lgb.predict_proba(X)[:, 1].reshape(-1, 1)
        
        X_meta = np.hstack([rf_pred, xgb_pred, lgb_pred])
        return self.meta_model.predict(X_meta)

def create_advanced_features(df):
    """
    创建高级特征
    
    包含：
    1. 传统因子（价值/动量/质量）
    2. 非线性变换（平方/交互项）
    3. 时序特征（斜率/加速度）
    4. 横截面特征（排名/分位数）
    """
    features = pd.DataFrame(index=df.index)
    
    # 1. 动量因子（多时间尺度）
    for period in [5, 10, 20, 60]:
        features[f'momentum_{period}'] = df['close'].pct_change(period)
    
    # 2. 波动率因子
    for period in [20, 60]:
        features[f'volatility_{period}'] = df['close'].pct_change().rolling(period).std()
    
    # 3. 成交量因子
    features['volume_ma_ratio'] = df['volume'] / df['volume'].rolling(20).mean()
    features['volume_price_trend'] = (df['close'].pct_change() * df['volume'].pct_change())
    
    # 4. 技术指标
    features['rsi_14'] = calculate_rsi(df['close'], 14)
    features['macd'] = calculate_macd(df['close'])
    
    # 5. 非线性变换
    features['momentum_20_sq'] = features['momentum_20'] ** 2
    features['vol_x_momentum'] = features['volatility_20'] * features['momentum_20']
    
    # 6. 时序特征（斜率）
    features['price_slope_20'] = calculate_slope(df['close'], window=20)
    
    # 7. 横截面排名（需要跨股票数据）
    # features['momentum_rank'] = features.groupby(level='date')['momentum_20'].rank(pct=True)
    
    return features.dropna()

def extract_alternative_features(text_data):
    """
    从另类数据中提取特征
    
    输入：
    - 新闻文本
    - 社交媒体数据
    - 研报文本
    
    输出：
    - 情感得分
    - 主题向量
    - 关注度指标
    """
    from transformers import AutoTokenizer, AutoModel
    import torch
    
    # 使用FinBERT提取情感
    tokenizer = AutoTokenizer.from_pretrained("yiyanghkust/finbert-tone")
    model = AutoModel.from_pretrained("yiyanghkust/finbert-tone")
    
    def extract_sentiment(text):
        inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=512)
        outputs = model(**inputs)
        # 获取[CLS]向量
        cls_embedding = outputs.last_hidden_state[:, 0, :].detach().numpy()
        return cls_embedding
    
    # 批量处理
    sentiments = text_data['text'].apply(extract_sentiment)
    
    # 降维（PCA）
    from sklearn.decomposition import PCA
    pca = PCA(n_components=10)
    sentiment_pca = pca.fit_transform(np.vstack(sentiments.values))
    
    return pd.DataFrame(
        sentiment_pca,
        index=text_data.index,
        columns=[f'sentiment_pc_{i}' for i in range(10)]
    )

from sklearn.model_selection import TimeSeriesSplit

def time_series_cv(X, y, model, n_splits=5):
    """
    时间序列交叉验证
    
    训练集始终在测试集之前，避免前视偏差
    """
    tscv = TimeSeriesSplit(n_splits=n_splits)
    scores = []
    
    for train_idx, test_idx in tscv.split(X):
        X_train, X_test = X.iloc[train_idx], X.iloc[test_idx]
        y_train, y_test = y.iloc[train_idx], y.iloc[test_idx]
        
        model.fit(X_train, y_train)
        score = model.score(X_test, y_test)
        scores.append(score)
    
    return np.mean(scores), np.std(scores)

# LSTM正则化
lstm_model = Sequential([
    LSTM(64, return_sequences=True,
         kernel_regularizer=l2(0.01),  # L2正则化
         recurrent_dropout=0.2),       # 递归Dropout
    Dropout(0.3),                      # 输入Dropout
    LSTM(32, return_sequences=False),
    Dropout(0.3),
    Dense(1)
])

# 树模型正则化
xgb_model = XGBClassifier(
    reg_alpha=0.1,   # L1正则化
    reg_lambda=1.0,  # L2正则化
    gamma=0.1,       # 分裂最小损失降低
    min_child_weight=10  # 子节点最小样本权重
)

def bootstrap_ensemble(X_train, y_train, n_models=10):
    """
    Bootstrap集成
    
    训练多个模型，每个模型用有放回抽样的数据集
    降低方差，提高泛化能力
    """
    models = []
    
    for i in range(n_models):
        # Bootstrap抽样
        boot_indices = np.random.choice(
            len(X_train),
            size=len(X_train),
            replace=True
        )
        X_boot = X_train.iloc[boot_indices]
        y_boot = y_train.iloc[boot_indices]
        
        # 训练模型
        model = build_lstm_model(input_shape=(X_boot.shape[1], X_boot.shape[2]))
        model.fit(
            X_boot, y_boot,
            epochs=50,
            batch_size=32,
            validation_split=0.2,
            verbose=0
        )
        models.append(model)
    
    return models

def ensemble_predict(models, X_test):
    """集成预测（平均）"""
    predictions = np.array([model.predict(X_test) for model in models])
    return predictions.mean(axis=0)

class MLSignalGenerator:
    """
    ML信号生成器
    
    将模型输出转换为交易信号
    """
    
    def __init__(self, model, threshold=0.5, n_classes=3):
        self.model = model
        self.threshold = threshold
        self.n_classes = n_classes
    
    def generate_signals(self, X, method='probability'):
        """
        生成交易信号
        
        Parameters:
        -----------
        method : str
            'probability' - 使用预测概率
            'ranking' - 使用排名分位数
            'top_k' - 选择Top K股票
        """
        if method == 'probability':
            probs = self.model.predict_proba(X)[:, 1]
            signals = (probs > self.threshold).astype(int)
            return signals, probs
        
        elif method == 'ranking':
            probs = self.model.predict_proba(X)[:, 1]
            # 将概率转换为排名分位数
            ranks = pd.Series(probs).rank(pct=True)
            signals = ranks.apply(lambda x: 1 if x > 0.8 else (-1 if x < 0.2 else 0))
            return signals.values, probs
        
        elif method == 'top_k':
            probs = self.model.predict_proba(X)[:, 1]
            top_k = int(len(probs) * 0.1)  # Top 10%
            threshold = np.sort(probs)[-top_k]
            signals = (probs >= threshold).astype(int)
            return signals, probs

class MLPortfolioOptimizer:
    """
    ML + 风险平价组合优化
    
    结合：
    1. ML预测收益
    2. 风险模型（协方差矩阵）
    3. 约束条件（行业中性/风格中性）
    """
    
    def __init__(self, risk_budget=None):
        self.risk_budget = risk_budget  # 风险预算（风险平价）
    
    def optimize(self, expected_returns, cov_matrix, constraints=None):
        """
        优化组合权重
        
        Parameters:
        -----------
        expected_returns : np.array
            ML模型预测的预期收益
        cov_matrix : np.array
            收益率协方差矩阵
        constraints : dict
            约束条件（最大权重/行业暴露等）
        """
        n_assets = len(expected_returns)
        
        # 目标函数：最大化夏普比率
        def objective(weights):
            portfolio_return = np.dot(weights, expected_returns)
            portfolio_risk = np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights)))
            sharpe = portfolio_return / portfolio_risk
            return -sharpe  # 负号因为scipy是最小化
        
        # 约束条件
        cons = [
            {'type': 'eq', 'fun': lambda w: np.sum(w) - 1.0},  # 权重和为1
        ]
        
        if constraints:
            if 'max_weight' in constraints:
                cons.append({'type': 'ineq', 'fun': lambda w: constraints['max_weight'] - np.max(w)})
            if 'min_weight' in constraints:
                cons.append({'type': 'ineq', 'fun': lambda w: np.min(w) - constraints['min_weight']})
        
        # 初始权重（等权）
        w0 = np.ones(n_assets) / n_assets
        
        # 优化
        result = minimize(
            objective,
            w0,
            method='SLSQP',
            constraints=cons,
            bounds=[(0, 1) for _ in range(n_assets)]
        )
        
        return result.x

def evaluate_trading_model(y_true, y_pred, y_prob, returns):
    """
    评估交易模型的综合指标
    
    不仅看分类准确率，更要看：
    1. 信息系数（IC）
    2. 多头收益率
    3. 空头收益率
    4. 多空组合收益率
    """
    from scipy.stats import spearmanr
    
    metrics = {}
    
    # 1. 信息系数（IC）
    ic, p_value = spearmanr(y_prob, returns)
    metrics['IC'] = ic
    metrics['IC_p_value'] = p_value
    
    # 2. 多空组合收益
    top_decile = np.percentile(y_prob, 90)
    bottom_decile = np.percentile(y_prob, 10)
    
    long_returns = returns[y_prob >= top_decile].mean()
    short_returns = -returns[y_prob <= bottom_decile].mean()  # 做空
    long_short_returns = long_returns + short_returns
    
    metrics['long_return'] = long_returns
    metrics['short_return'] = short_returns
    metrics['long_short_return'] = long_short_returns
    
    # 3. 胜率
    correct_predictions = ((y_prob > 0.5) & (returns > 0)) | ((y_prob <= 0.5) & (returns <= 0))
    metrics['win_rate'] = correct_predictions.mean()
    
    # 4. 盈亏比
    profitable_trades = returns[correct_predictions]
    unprofitable_trades = returns[~correct_predictions]
    metrics['profit_loss_ratio'] = profitable_trades.mean() / abs(unprofitable_trades.mean())
    
    return metrics

class MLBacktester:
    """
    ML策略回测框架
    
    支持：
    1. 滚动训练（Retrain）
    2. 在线学习（Online Learning）
    3. 交易成本调整
    """
    
    def __init__(self, model, retrain_frequency=20):
        self.model = model
        self.retrain_frequency = retrain_frequency
        self.transaction_cost = 0.001  # 双边0.1%
    
    def backtest(self, X, y, returns, initial_capital=1000000):
        """
        回测ML策略
        
        Parameters:
        -----------
        X : pd.DataFrame
            特征数据
        y : pd.Series
            标签（未来收益）
        returns : pd.Series
            实际收益
        """
        portfolio_value = [initial_capital]
        positions = []
        
        for i in range(self.retrain_frequency, len(X), self.retrain_frequency):
            # 滚动训练
            X_train = X.iloc[i-self.retrain_frequency:i]
            y_train = y.iloc[i-self.retrain_frequency:i]
            self.model.fit(X_train, y_train)
            
            # 预测
            X_test = X.iloc[i:i+self.retrain_frequency]
            predictions = self.model.predict_proba(X_test)[:, 1]
            
            # 生成持仓
            position = (predictions > 0.5).astype(int)
            positions.extend(position)
            
            # 计算收益（扣除交易成本）
            period_returns = (position * returns.iloc[i:i+self.retrain_frequency]).sum()
            cost = abs(position - positions[-2] if len(positions) > 1 else position).sum() * self.transaction_cost
            net_return = period_returns - cost
            
            portfolio_value.append(portfolio_value[-1] * (1 + net_return))
        
        return pd.Series(portfolio_value)

import torch
import torch_geometric as pyg

class StockGNN(torch.nn.Module):
    """
    股票关系图神经网络
    
    节点：股票
    边：行业关联/供应链/共同基金持仓
    
    通过消息传递捕捉系统性风险传导
    """
    def __init__(self, node_features, edge_features, hidden_channels):
        super().__init__()
        
        self.conv1 = pyg.nn.GCNConv(node_features, hidden_channels)
        self.conv2 = pyg.nn.GCNConv(hidden_channels, hidden_channels)
        self.classifier = torch.nn.Linear(hidden_channels, 1)
    
    def forward(self, data):
        x, edge_index, edge_attr = data.x, data.edge_index, data.edge_attr
        
        # 图卷积层
        x = self.conv1(x, edge_index, edge_attr)
        x = torch.relu(x)
        x = self.conv2(x, edge_index, edge_attr)
        
        # 节点分类（预测每只股票的超额收益）
        out = self.classifier(x)
        return out

import gym
from stable_baselines3 import PPO

class TradingEnv(gym.Env):
    """
    交易环境（OpenAI Gym接口）
    
    状态空间：
    - 当前持仓
    - 历史价格
    - 账户余额
    
    动作空间：
    - 买入/卖出/持有
    - 仓位大小
    """
    def __init__(self, price_data, initial_cash=1000000):
        super().__init__()
        self.price_data = price_data
        self.initial_cash = initial_cash
        self.reset()
    
    def reset(self):
        self.cash = self.initial_cash
        self.position = 0
        self.current_step = 0
        return self._get_observation()
    
    def step(self, action):
        """执行交易动作，返回（状态, 奖励, 是否结束, 信息）"""
        # action: (direction, quantity)
        # direction: -1(卖出), 0(持有), 1(买入)
        # quantity: 0-1之间，表示仓位比例
        
        current_price = self.price_data.iloc[self.current_step]['close']
        
        # 执行交易
        if action[0] == 1:  # 买入
            cost = self.cash * action[1] * (1 + self.transaction_cost)
            shares_to_buy = cost / current_price
            self.position += shares_to_buy
            self.cash -= cost
        
        elif action[0] == -1:  # 卖出
            shares_to_sell = self.position * action[1]
            revenue = shares_to_sell * current_price * (1 - self.transaction_cost)
            self.position -= shares_to_sell
            self.cash += revenue
        
        # 计算奖励（组合价值变化）
        portfolio_value = self.cash + self.position * current_price
        reward = (portfolio_value - self.previous_portfolio_value) / self.previous_portfolio_value
        
        self.current_step += 1
        done = self.current_step >= len(self.price_data) - 1
        
        return self._get_observation(), reward, done, {}

# 训练RL代理
env = TradingEnv(price_data)
model = PPO('MlpPolicy', env, verbose=1)
model.learn(total_timesteps=100000)