AI量化交易实战：从多因子模型到强化学习，构建Alpha猎手的系统框架

┌─────────────────────────────────────────────────────────────────────┐
│                        策略层（Alpha生成）                          │
│   深度学习预测 → 强化学习决策 → 组合优化 → 风险控制                   │
└───────────────────────────────┬─────────────────────────────────────┘
                                │
                                ▼
┌─────────────────────────────────────────────────────────────────────┐
│                      模型层（特征工程与建模）                         │
│   ┌──────────────┐ ┌──────────────┐ ┌──────────────┐              │
│   │  传统ML模型   │ │  深度学习模型 │ │ 强化学习模型 │              │
│   │ XGBoost/LightGBM│ │   Transformer │ │   DQN/PPO   │              │
│   │  随机森林     │ │   GNN/图网络  │ │   SAC/A3C   │              │
│   └──────────────┘ └──────────────┘ └──────────────┘              │
└───────────────────────────────┬─────────────────────────────────────┘
                                │
                                ▼
┌─────────────────────────────────────────────────────────────────────┐
│                      数据层（核心基础设施）                          │
│   ┌──────────────┐ ┌──────────────┐ ┌──────────────┐              │
│   │  行情数据    │ │  基本面数据  │ │  另类数据    │              │
│   │  Tick/分钟/K线│ │  财报/研报   │ │  舆情/卫星   │              │
│   └──────────────┘ └──────────────┘ └──────────────┘              │
│   ┌──────────────────────────────────────────────────┐             │
│   │     数据仓库：ClickHouse + Parquet + Delta Lake  │             │
│   └──────────────────────────────────────────────────┘             │
└─────────────────────────────────────────────────────────────────────┘

# data_pipeline.py - 数据采集与清洗

import polars as pl
from clickhouse_driver import Client
from datetime import datetime, timedelta
import requests
import json

class MarketDataPipeline:
    def __init__(self):
        self.ch_client = Client(host='clickhouse.db', port=9000)
        
    def fetch_minute_bars(self, symbols, start_date, end_date):
        """从行情源获取分钟级数据（模拟）"""
        # 实际使用：万得/聚宽/通联等数据API
        bars = []
        for symbol in symbols:
            raw = requests.get(
                f'https://api.marketdata.com/minutes',
                params={'symbol': symbol, 'start': start_date, 'end': end_date}
            ).json()
            bars.extend(raw['data'])
        
        df = pl.DataFrame(bars)
        return self._clean_data(df)
    
    def _clean_data(self, df):
        """数据清洗：去极值、填充缺失、复权处理"""
        # 1. 去极值（MAD方法）
        median = df['close'].median()
        mad = (df['close'] - median).abs().median()
        df = df.filter(pl.col('close').is_between(
            median - 5 * mad, median + 5 * mad
        ))
        
        # 2. 填充缺失（前向填充）
        df = df.fill_null(strategy='forward')
        
        # 3. 复权处理（后复权）
        df = df.with_columns([
            (pl.col('close') * pl.col('adjust_factor')).alias('close_adj'),
            (pl.col('open') * pl.col('adjust_factor')).alias('open_adj'),
        ])
        
        return df
    
    def write_to_clickhouse(self, df, table_name='minute_bars'):
        """写入ClickHouse"""
        # 将Polars DataFrame转换为ClickHouse插入格式
        records = df.to_dicts()
        self.ch_client.execute(
            f'INSERT INTO {table_name} (symbol, timestamp, open, high, low, close, volume) VALUES',
            records
        )

# 每日调度：增量更新
if __name__ == '__main__':
    pipeline = MarketDataPipeline()
    symbols = ['000001.SZ', '600000.SH', '600036.SH']  # 沪深股票示例
    yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
    
    df = pipeline.fetch_minute_bars(symbols, yesterday, yesterday)
    pipeline.write_to_clickhouse(df)

# 时间戳验证函数
def validate_no_lookahead(df, feature_cols, timestamp_col='timestamp'):
    """验证没有未来信息"""
    for col in feature_cols:
        # 检查特征的最大时间是否超过了标签的时间
        assert df[col].timestamp.max() < df['label'].timestamp.min()

# autoencoder_factor.py - 用变分自编码器（VAE）自动挖掘因子

import torch
import torch.nn as nn
import torch.nn.functional as F

class FactorAutoEncoder(nn.Module):
    """用VAE从原始量价数据中自动学习因子表示"""
    def __init__(self, input_dim=50, latent_dim=16, hidden_dim=128):
        super().__init__()
        # Encoder
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, hidden_dim),
            nn.BatchNorm1d(hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
        )
        self.mu = nn.Linear(hidden_dim, latent_dim)
        self.log_var = nn.Linear(hidden_dim, latent_dim)
        
        # Decoder
        self.decoder = nn.Sequential(
            nn.Linear(latent_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, input_dim),
        )
    
    def encode(self, x):
        h = self.encoder(x)
        return self.mu(h), self.log_var(h)
    
    def reparameterize(self, mu, log_var):
        std = torch.exp(0.5 * log_var)
        eps = torch.randn_like(std)
        return mu + eps * std
    
    def forward(self, x):
        mu, log_var = self.encode(x)
        z = self.reparameterize(mu, log_var)  # z就是学习到的因子表示
        recon = self.decoder(z)
        return recon, mu, log_var
    
    def factor_loss(self, x):
        """提取因子表示并计算重构损失"""
        recon, mu, log_var = self.forward(x)
        # 重构损失 + KL散度
        recon_loss = F.mse_loss(recon, x)
        kl_loss = -0.5 * torch.sum(1 + log_var - mu.pow(2) - log_var.exp())
        return recon_loss + 0.001 * kl_loss, mu  # mu可以作为因子输出

# factor_evaluation.py - 因子评估工具

import pandas as pd
import numpy as np
from scipy import stats

def evaluate_factor(df, factor_col, return_col='future_return', period=20):
    """
    评估因子的IC、IR和分组收益
    df: 包含因子值和未来收益的DataFrame
    """
    # 1. 计算IC（Spearman相关性更稳健）
    ic_series = df.groupby('date').apply(
        lambda g: stats.spearmanr(g[factor_col], g[return_col])[0]
    )
    avg_ic = ic_series.mean()
    ir = avg_ic / ic_series.std()
    
    # 2. 分组收益（将因子从大到小分为10组）
    df['group'] = df.groupby('date')[factor_col].transform(
        lambda x: pd.qcut(x, 10, labels=False)
    )
    group_returns = df.groupby(['date', 'group'])[return_col].mean().unstack()
    
    # Top组 - Bottom组的收益差（多头-空头收益）
    long_short_returns = group_returns[9] - group_returns[0]
    
    print(f"IC均值: {avg_ic:.4f}")
    print(f"IR: {ir:.4f}")
    print(f"Top-Bottom年化收益: {long_short_returns.mean()*252:.2%}")
    
    return {
        'avg_ic': avg_ic,
        'ir': ir,
        'long_short_returns': long_short_returns
    }

# transformer_model.py - 基于Transformer的收益率预测

import torch
import torch.nn as nn
import torch.nn.functional as F
import math

class PositionalEncoding(nn.Module):
    """位置编码 - 保留序列顺序信息"""
    def __init__(self, d_model, max_len=252):
        super().__init__()
        pe = torch.zeros(max_len, d_model)
        position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
        div_term = torch.exp(torch.arange(0, d_model, 2).float() * 
                            (-math.log(10000.0) / d_model))
        pe[:, 0::2] = torch.sin(position * div_term)
        pe[:, 1::2] = torch.cos(position * div_term)
        self.register_buffer('pe', pe)
    
    def forward(self, x):
        return x + self.pe[:x.size(1), :]

class StockTransformer(nn.Module):
    """
    输入：过去N天的量价特征 (batch, seq_len, features)
    输出：未来收益率预测 (batch, 1)
    """
    def __init__(self, input_dim=20, d_model=64, nhead=8, num_layers=4):
        super().__init__()
        self.input_proj = nn.Linear(input_dim, d_model)
        self.pos_encoder = PositionalEncoding(d_model)
        
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=d_model,
            nhead=nhead,
            dim_feedforward=256,
            dropout=0.1,
            activation='gelu',
            batch_first=True
        )
        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=num_layers)
        
        # 输出层：用最后一个时间步的表示做预测
        self.output_proj = nn.Sequential(
            nn.Linear(d_model, 64),
            nn.GELU(),
            nn.Dropout(0.2),
            nn.Linear(64, 32),
            nn.GELU(),
            nn.Linear(32, 1)
        )
    
    def forward(self, x):
        # x: (batch, seq_len, input_dim)
        x = self.input_proj(x)  # (batch, seq_len, d_model)
        x = self.pos_encoder(x)  # 加位置编码
        x = self.transformer(x)  # (batch, seq_len, d_model)
        # 使用最后一个时间步的表示
        last_step = x[:, -1, :]  # (batch, d_model)
        return self.output_proj(last_step)  # (batch, 1)

# 稳健损失函数
def huber_loss(pred, target, delta=0.5):
    diff = pred - target
    abs_diff = torch.abs(diff)
    loss = torch.where(
        abs_diff < delta,
        0.5 * diff.pow(2),
        delta * (abs_diff - 0.5 * delta)
    )
    return loss.mean()

┌─────────────────────────────────────────────────────────────┐
│                   强化学习交易框架                           │
│                                                             │
│   状态State        ──▶    Agent（策略网络）    ──▶  动作Action│
│   (当前持仓/价格/因子)       (深度神经网络)        (买卖数量) │
│        ▲                                              │    │
│        │                                              ▼    │
│        │         ┌─────────────────────────┐             │    │
│        └─────────│    环境Environment       │◀───────────┘    │
│                  │  (市场 + 交易成本 + 滑点)  │                │
│                  └─────────────────────────┘                │
│                         │                                    │
│                         ▼                                    │
│                     奖励Reward                               │
│                 (收益率 - 交易成本)                          │
└─────────────────────────────────────────────────────────────┘

# ppo_trader.py - PPO算法实现交易策略

import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
from collections import deque
import gym
from gym import spaces

# ============ 交易环境定义 ============
class TradingEnv(gym.Env):
    """自定义交易环境"""
    def __init__(self, data, initial_capital=1000000, max_position=0.9):
        super().__init__()
        self.data = data  # 股票数据
        self.current_step = 0
        self.max_step = len(data) - 1
        self.initial_capital = initial_capital
        self.max_position = max_position
        
        # 动作空间：仓位比例 [-1, 1]（负数为做空）
        self.action_space = spaces.Box(low=-1, high=1, shape=(1,))
        
        # 状态空间：特征维度
        self.observation_space = spaces.Box(
            low=-np.inf, high=np.inf, 
            shape=(data.shape[1] + 1,),  # 行情特征 + 当前持仓
            dtype=np.float32
        )
    
    def reset(self):
        self.current_step = 0
        self.capital = self.initial_capital
        self.position = 0  # 当前持仓（股数）
        self.portfolio_value = self.initial_capital
        self.trades = []
        return self._get_state()
    
    def _get_state(self):
        # 获取当前时间步的特征 + 持仓状态
        features = self.data[self.current_step].values
        state = np.concatenate([features, [self.position / self.initial_capital]])
        return state.astype(np.float32)
    
    def step(self, action):
        # 1. 执行交易
        price = self.data.loc[self.current_step, 'close']
        target_position = np.clip(action[0], -self.max_position, self.max_position)
        target_shares = target_position * self.portfolio_value / price
        
        # 计算交易成本（冲击成本 + 佣金）
        trade_amount = abs(target_shares - self.position) * price
        cost = trade_amount * 0.0003 + trade_amount * 0.00005  # 佣金 + 滑点
        
        # 更新持仓
        self.position = target_shares
        
        # 2. 转移到下一时间步
        self.current_step += 1
        done = self.current_step >= self.max_step
        
        # 3. 计算新市值
        new_price = self.data.loc[self.current_step, 'close'] if not done else price
        self.portfolio_value = self.position * new_price + self.capital - cost
        
        # 4. 计算收益
        reward = (self.portfolio_value - self.initial_capital) / self.initial_capital
        
        return self._get_state(), reward, done, {'portfolio': self.portfolio_value}

# ============ PPO Agent实现 ============
class PPOAgent:
    def __init__(self, state_dim, action_dim, lr=3e-4, gamma=0.99, clip_ratio=0.2):
        self.gamma = gamma
        self.clip_ratio = clip_ratio
        
        # Actor网络：输出动作分布
        self.actor = nn.Sequential(
            nn.Linear(state_dim, 256),
            nn.ReLU(),
            nn.Linear(256, 128),
            nn.ReLU(),
            nn.Linear(128, action_dim),
            nn.Tanh()  # 动作范围 [-1, 1]
        )
        
        # Critic网络：输出状态价值
        self.critic = nn.Sequential(
            nn.Linear(state_dim, 256),
            nn.ReLU(),
            nn.Linear(256, 128),
            nn.ReLU(),
            nn.Linear(128, 1)
        )
        
        self.actor_optimizer = torch.optim.Adam(self.actor.parameters(), lr=lr)
        self.critic_optimizer = torch.optim.Adam(self.critic.parameters(), lr=lr)
        
    def get_action(self, state):
        """根据当前状态选择动作（带探索）"""
        state = torch.FloatTensor(state).unsqueeze(0)
        action = self.actor(state)
        # 添加探索噪声
        noise = torch.randn_like(action) * 0.1
        action = torch.clamp(action + noise, -1, 1)
        return action.detach().numpy()[0]
    
    def update(self, states, actions, rewards, next_states, dones):
        """PPO更新（简化的单步更新）"""
        states = torch.FloatTensor(np.array(states))
        actions = torch.FloatTensor(np.array(actions))
        rewards = torch.FloatTensor(np.array(rewards))
        
        # 计算优势函数
        values = self.critic(states).squeeze()
        next_values = self.critic(torch.FloatTensor(next_states)).squeeze()
        td_target = rewards + self.gamma * next_values * (1 - dones)
        advantages = td_target - values.detach()
        
        # Actor更新（PPO的clipped objective）
        new_actions = self.actor(states)
        ratio = torch.exp(torch.distributions.Normal(new_actions, 0.1).log_prob(actions) -
                         torch.distributions.Normal(actions, 0.1).log_prob(actions))
        clipped_ratio = torch.clamp(ratio, 1 - self.clip_ratio, 1 + self.clip_ratio)
        actor_loss = -torch.min(ratio * advantages, clipped_ratio * advantages).mean()
        
        # Critic更新
        critic_loss = F.mse_loss(self.critic(states).squeeze(), td_target.detach())
        
        # 反向传播
        self.actor_optimizer.zero_grad()
        actor_loss.backward()
        self.actor_optimizer.step()
        
        self.critic_optimizer.zero_grad()
        critic_loss.backward()
        self.critic_optimizer.step()
        
        return {'actor_loss': actor_loss.item(), 'critic_loss': critic_loss.item()}

# backtest_engine.py - 回测引擎核心

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class BacktestEngine:
    def __init__(self, data, initial_capital=1_000_000, commission=0.0003, slippage=0.00005):
        """
        data: 包含价格数据的DataFrame，必须包含日期、股票代码、价格
        """
        self.data = data
        self.initial_capital = initial_capital
        self.commission = commission
        self.slippage = slippage
        
    def run(self, model, start_date, end_date, rebalance_freq='daily'):
        """
        运行回测
        """
        # 1. 日期切分
        dates = pd.date_range(start_date, end_date, freq=rebalance_freq)
        
        # 2. 初始化组合
        portfolio = {
            'capital': self.initial_capital,
            'positions': {},  # {symbol: shares}
            'cash': self.initial_capital,
            'value_history': [],
            'trade_history': []
        }
        
        for date in dates:
            # 获取当日数据
            daily_data = self.data[self.data['date'] == date]
            if daily_data.empty:
                continue
                
            # 生成交易信号
            signals = model.predict(daily_data)
            
            # 执行交易
            self._execute_trades(daily_data, signals, portfolio, date)
            
            # 记录组合价值
            portfolio['value_history'].append({
                'date': date,
                'value': self._calc_portfolio_value(portfolio, daily_data)
            })
        
        # 3. 计算绩效指标
        return self._calc_metrics(portfolio['value_history'])
    
    def _calc_metrics(self, value_history):
        """计算绩效指标：年化收益、夏普、最大回撤等"""
        df = pd.DataFrame(value_history)
        df['return'] = df['value'].pct_change()
        
        # 年化收益
        total_return = df['value'].iloc[-1] / df['value'].iloc[0] - 1
        n_years = len(df) / 252
        annual_return = (1 + total_return) ** (1 / n_years) - 1
        
        # 夏普比率
        risk_free = 0.03
        excess_returns = df['return'] - risk_free / 252
        sharpe = np.sqrt(252) * excess_returns.mean() / excess_returns.std()
        
        # 最大回撤
        cummax = df['value'].cummax()
        drawdown = (df['value'] - cummax) / cummax
        max_drawdown = drawdown.min()
        
        # 胜率
        win_rate = (df['return'] > 0).mean()
        
        return {
            'total_return': total_return,
            'annual_return': annual_return,
            'sharpe_ratio': sharpe,
            'max_drawdown': max_drawdown,
            'win_rate': win_rate,
            'n_trades': len(self.trade_history)
        }

# 实盘风控规则
class RiskControl:
    def __init__(self):
        self.max_drawdown = 0.15       # 最大回撤15%
        self.max_position = 0.8        # 最大仓位80%
        self.max_concentration = 0.2   # 单只股票最大20%
        self.daily_loss_limit = 0.05   # 单日最大损失5%
    
    def check(self, portfolio, current_value):
        """风控检查，返回是否通过"""
        # 回撤检查
        if current_value < portfolio.peak * (1 - self.max_drawdown):
            return False, "触发最大回撤止损"
        
        # 单日损失检查
        if current_value < portfolio.yesterday_value * (1 - self.daily_loss_limit):
            return False, "触发单日损失止损"
        
        return True, "通过"

┌─────────────────────────────────────────────────────────────────┐
│                      实盘系统架构                                │
│                                                                 │
│  ┌─────────┐     ┌─────────┐     ┌─────────┐                  │
│  │ 行情订阅 │────▶│ 信号生成 │────▶│ 风控检查 │                  │
│  │ (WebSocket)│    │ (模型推理)│    │ (规则引擎)│                  │
│  └─────────┘     └─────────┘     └─────────┘                  │
│       │               │               │                         │
│       ▼               ▼               ▼                         │
│  ┌─────────┐     ┌─────────┐     ┌─────────┐                  │
│  │ 数据缓存 │     │ 日志记录 │     │ 订单执行 │                  │
│  │ (Redis) │     │ (数据库) │     │ (券商API)│                  │
│  └─────────┘     └─────────┘     └─────────┘                  │
│                                         │                       │
│                                         ▼                       │
│                                    ┌─────────┐                  │
│                                    │ 交易记录 │                  │
│                                    │ (ClickHouse)│              │
│                                    └─────────┘                  │
└─────────────────────────────────────────────────────────────────┘