主流策略类型

系统梳理量化交易的经典策略框架

学习目标

理解量化交易的主要策略类型及其原理
掌握动量、均值回归、多因子等核心策略
了解各策略的适用场景和风险特征
学会用 Python 实现基础策略逻辑

策略分类概览

量化策略可以从多个维度分类：

分类维度	类型
信号来源	趋势跟踪、均值回归、统计套利、事件驱动
持仓周期	高频、日内、短线、中线、长线
资产类型	股票、期货、期权、债券、外汇、加密货币
风险暴露	方向性、市场中性、套利、多策略

一、统计套利 (Statistical Arbitrage)

1.1 核心思想

统计套利（StatArb）是基于统计关系的均值回归策略。核心逻辑：

找到历史上价格走势相关的资产
当它们的价格关系偏离正常范围时
做空”相对高估”的，做多”相对低估”的
等待价格关系回归正常时平仓获利

1.2 与配对交易的关系

┌─────────────────────────────────────────────────────────────┐
│                    统计套利 vs 配对交易                      │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  配对交易 (Pairs Trading)                                   │
│  ├── 最简单的统计套利形式                                   │
│  ├── 通常只涉及两只资产                                     │
│  └── 专注于价差的均值回归                                   │
│                                                             │
│  统计套利 (Statistical Arbitrage)                           │
│  ├── 更广泛的策略类别                                       │
│  ├── 可涉及多只资产的复杂关系                               │
│  ├── 包括：配对交易、多因子统计模型、ML预测                 │
│  └── 常用于市场中性组合                                     │
│                                                             │
└─────────────────────────────────────────────────────────────┘

1.3 简化示例

假设两只股票 A 和 B 历史上价格比为 2:1，现在比价为 2.5:1：

# 策略逻辑
ratio = price_A / price_B  # 2.5，高于历史均值
 
# 交易决策
if ratio > threshold:
    # A 相对 B 太贵了
    # 做空 A，做多 B
    position_A = -1
    position_B = 2  # 保持对冲比率
 
# 等待 ratio 回归到 2:1 左右时平仓

详细实现请参考 02-统计套利详解.md

二、动量策略 (Momentum Strategies)

2.1 核心思想

动量效应是市场中最稳健的异象之一：过去表现好的资产，未来倾向于继续表现好。

“The trend is your friend until the end.”

行为金融学解释：

反应不足：投资者对新信息反应缓慢，价格逐步调整
羊群效应：投资者跟随他人行为，强化趋势
确认偏误：选择性关注支持自己观点的信息

2.2 时间序列动量 (Time Series Momentum)

也称为趋势跟踪 (Trend Following)，是 CTA 策略的核心。

2.2.1 策略逻辑

对单一资产：

过去一段时间上涨 → 做多
过去一段时间下跌 → 做空

2.2.2 信号生成方法

方法一：均线交叉 (Moving Average Crossover)

import numpy as np
import pandas as pd
 
def ma_crossover_signal(prices, fast_period=20, slow_period=60):
    """
    均线交叉信号生成
 
    参数:
        prices: 价格序列
        fast_period: 快线周期
        slow_period: 慢线周期
 
    返回:
        signal: 1=做多, -1=做空, 0=中性
    """
    # 计算均线
    fast_ma = prices.rolling(window=fast_period).mean()
    slow_ma = prices.rolling(window=slow_period).mean()
 
    # 生成信号
    signal = np.where(fast_ma > slow_ma, 1, -1)
 
    return signal
 
 
# 模拟数据演示
np.random.seed(42)
dates = pd.date_range('2023-01-01', periods=500, freq='D')
 
# 生成带趋势的价格
trend = np.linspace(100, 150, 500)
noise = np.random.normal(0, 2, 500)
prices = pd.Series(trend + noise, index=dates)
 
# 生成信号
signals = ma_crossover_signal(prices, fast_period=20, slow_period=60)
 
# 可视化信号
import matplotlib.pyplot as plt
 
plt.figure(figsize=(14, 6))
plt.plot(prices.index, prices.values, label='Price', alpha=0.7)
plt.plot(prices.index, prices.rolling(20).mean(), label='MA20', alpha=0.7)
plt.plot(prices.index, prices.rolling(60).mean(), label='MA60', alpha=0.7)
 
# 标记做多区域
long_signals = signals == 1
plt.fill_between(prices.index, 0, prices.max(),
                 where=long_signals, alpha=0.1, color='green', label='Long')
 
plt.title('均线交叉策略')
plt.legend()
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()

方法二：通道突破 (Channel Breakout)

def channel_breakout_signal(prices, period=60, multiplier=2.0):
    """
    通道突破信号
 
    参数:
        prices: 价格序列
        period: 通道周期
        multiplier: 通道倍数 (用于布林带)
 
    返回:
        signal: 信号序列
    """
    # 计算通道
    mean = prices.rolling(window=period).mean()
    std = prices.rolling(window=period).std()
 
    upper_band = mean + multiplier * std
    lower_band = mean - multiplier * std
 
    # 生成信号
    signal = pd.Series(0, index=prices.index)
 
    # 突破上轨 → 做多
    signal[prices > upper_band] = 1
 
    # 突破下轨 → 做空
    signal[prices < lower_band] = -1
 
    return signal, upper_band, lower_band
 
 
# 演示
signals, upper, lower = channel_breakout_signal(prices, period=60, multiplier=2)
 
plt.figure(figsize=(14, 6))
plt.plot(prices.index, prices.values, label='Price', alpha=0.6)
plt.plot(upper.index, upper.values, label='Upper Band', linestyle='--', color='red')
plt.plot(lower.index, lower.values, label='Lower Band', linestyle='--', color='green')
plt.fill_between(prices.index, lower.values, upper.values, alpha=0.1)
 
# 标记信号
long_points = signals == 1
short_points = signals == -1
plt.scatter(prices.index[long_points], prices.values[long_points],
            color='green', marker='^', s=50, label='Long Signal', zorder=5)
plt.scatter(prices.index[short_points], prices.values[short_points],
            color='red', marker='v', s=50, label='Short Signal', zorder=5)
 
plt.title('通道突破策略')
plt.legend()
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()

2.3 截面动量 (Cross-Sectional Momentum)

截面动量是相对强弱策略，在多资产之间比较。

2.3.1 策略逻辑

计算所有资产的过去收益率
做多收益率最高的 N 个
做空收益率最低的 N 个（可选）
定期调仓

2.3.2 完整实现

def cross_sectional_momentum(returns, lookback=60, top_n=10, bottom_n=10, long_only=False):
    """
    截面动量策略
 
    参数:
        returns: 收益率矩阵 (T x N)，每列为一个资产
        lookback: 回看期
        top_n: 做多资产数量
        bottom_n: 做空资产数量
        long_only: 是否只做多
 
    返回:
        positions: 持仓矩阵 (T x N)
    """
    T, N = returns.shape
    positions = pd.DataFrame(0, index=returns.index, columns=returns.columns)
 
    for i in range(lookback, T):
        # 计算回看期累计收益率
        historical_returns = returns.iloc[i-lookback:i]
 
        # 计算累计收益
        cumulative_returns = (1 + historical_returns).prod() - 1
 
        # 排名
        ranked = cumulative_returns.rank(ascending=False)
 
        # 做多头寸
        top_assets = ranked[ranked <= top_n].index
        positions.iloc[i][top_assets] = 1 / top_n
 
        # 做空头寸
        if not long_only:
            bottom_assets = ranked[ranked >= (N - bottom_n + 1)].index
            positions.iloc[i][bottom_assets] = -1 / bottom_n
 
    return positions
 
 
# 模拟数据演示
np.random.seed(42)
n_assets = 50
n_periods = 500
 
# 生成不同特征的资产收益率
returns = pd.DataFrame(
    np.random.normal(0.0005, 0.02, (n_periods, n_assets)),
    index=pd.date_range('2023-01-01', periods=n_periods, freq='D'),
    columns=[f'Stock_{i:02d}' for i in range(n_assets)]
)
 
# 添加一些动量效应
for i in range(n_assets):
    momentum_factor = np.random.choice([-0.3, 0, 0.3])
    returns[f'Stock_{i:02d}'] += momentum_factor * returns[f'Stock_{i:02d}'].shift(1).fillna(0)
 
# 运行策略
positions = cross_sectional_momentum(returns, lookback=60, top_n=10, bottom_n=10)
 
# 计算策略收益
strategy_returns = (positions.shift(1) * returns).sum(axis=1)
cumulative_returns = (1 + strategy_returns).cumprod()
 
# 可视化
fig, axes = plt.subplots(2, 1, figsize=(14, 8))
 
# 累计收益
axes[0].plot(cumulative_returns.index, cumulative_returns.values, label='Strategy', linewidth=2)
axes[0].plot(cumulative_returns.index, (1 + returns.mean(axis=1)).cumprod().values,
             label='Equal Weight', linewidth=2, alpha=0.7)
axes[0].set_title('截面动量策略累计收益')
axes[0].legend()
axes[0].grid(True, alpha=0.3)
 
# 持仓分布
position_counts = positions[position_counts != 0].count(axis=1)
axes[1].bar(position_counts.index, position_counts.values, color='steelblue', alpha=0.7)
axes[1].set_title('持仓数量')
axes[1].set_ylabel('持仓资产数')
axes[1].grid(True, alpha=0.3)
 
plt.tight_layout()
plt.show()

2.4 动量的崩溃与反转

动量策略存在显著风险：

风险类型	描述	缓解方法
动量崩溃	趋势突然反转，快速亏损	止损、波动率调整
高换手率	频繁调仓导致交易成本高	信号平滑、降低调仓频率
市场状态依赖	在震荡市中表现差	识别市场状态，动态调整

三、均值回归策略 (Mean Reversion)

3.1 核心思想

均值回归假设价格偏离”正常水平”后会回归：

$P_{t + 1} - P_{t} \propto (μ - P_{t})$

其中 $μ$ 是均值水平。

3.2 布林带均值回归

def bollinger_bands_mean_reversion(prices, period=20, std_dev=2, entry_z=2, exit_z=0.5):
    """
    布林带均值回归策略
 
    参数:
        prices: 价格序列
        period: 均线周期
        std_dev: 标准差倍数
        entry_z: 入场 Z-score 阈值
        exit_z: 出场 Z-score 阈值
 
    返回:
        positions: 持仓序列
    """
    # 计算布林带
    mean = prices.rolling(window=period).mean()
    std = prices.rolling(window=period).std()
 
    upper_band = mean + std_dev * std
    lower_band = mean - std_dev * std
 
    # 计算 Z-score
    z_score = (prices - mean) / std
 
    # 生成信号
    positions = pd.Series(0, index=prices.index)
 
    # 价格触及上轨 → 做空（期待回归）
    positions[z_score > entry_z] = -1
 
    # 价格触及下轨 → 做多（期待回归）
    positions[z_score < -entry_z] = 1
 
    # 回归到均值附近 → 平仓
    positions[np.abs(z_score) < exit_z] = 0
 
    return positions, upper_band, lower_band, z_score
 
 
# 演示
np.random.seed(42)
# 生成均值回归价格
ornstein_uhlenbeck = []
price = 100
theta = 0.1  # 回归速度
mu = 100     # 均值
sigma = 2    # 波动率
dt = 1
 
for _ in range(500):
    d_price = theta * (mu - price) * dt + sigma * np.random.normal(0, np.sqrt(dt))
    price += d_price
    ornstein_uhlenbeck.append(price)
 
prices_mr = pd.Series(ornstein_uhlenbeck, index=pd.date_range('2023-01-01', periods=500, freq='D'))
 
# 运行策略
positions, upper, lower, z_score = bollinger_bands_mean_reversion(prices_mr)
 
# 可视化
fig, axes = plt.subplots(2, 1, figsize=(14, 8))
 
# 价格和信号
axes[0].plot(prices_mr.index, prices_mr.values, label='Price', alpha=0.7)
axes[0].plot(upper.index, upper.values, label='Upper Band', linestyle='--', color='red')
axes[0].plot(lower.index, lower.values, label='Lower Band', linestyle='--', color='green')
axes[0].fill_between(prices_mr.index, lower.values, upper.values, alpha=0.1)
 
# 标记持仓
long_positions = positions == 1
short_positions = positions == -1
axes[0].scatter(prices_mr.index[long_positions], prices_mr.values[long_positions],
                color='green', marker='^', s=30, label='Long', zorder=5)
axes[0].scatter(prices_mr.index[short_positions], prices_mr.values[short_positions],
                color='red', marker='v', s=30, label='Short', zorder=5)
axes[0].set_title('布林带均值回归策略')
axes[0].legend()
axes[0].grid(True, alpha=0.3)
 
# Z-score
axes[1].plot(z_score.index, z_score.values, label='Z-Score', color='steelblue')
axes[1].axhline(y=2, color='red', linestyle='--', label='Entry (+2)')
axes[1].axhline(y=-2, color='green', linestyle='--', label='Entry (-2)')
axes[1].axhline(y=0.5, color='orange', linestyle=':', label='Exit (±0.5)')
axes[1].axhline(y=-0.5, color='orange', linestyle=':')
axes[1].fill_between(z_score.index, -0.5, 0.5, alpha=0.1, color='gray')
axes[1].set_title('Z-Score')
axes[1].set_ylabel('Z-Score')
axes[1].legend()
axes[1].grid(True, alpha=0.3)
 
plt.tight_layout()
plt.show()

3.3 配对交易简化版

配对交易是均值回归的经典应用：

def simplified_pairs_trading(price_a, price_b, window=30, entry_threshold=2, exit_threshold=0.5):
    """
    简化版配对交易
 
    参数:
        price_a, price_b: 两只资产的价格
        window: 滚动窗口
        entry_threshold: 入场 Z-score 阈值
        exit_threshold: 出场 Z-score 阈值
 
    返回:
        positions_a, positions_b: 两只资产的持仓
        spread: 价差
        z_score: 标准化价差
    """
    # 计算价差（简化：假设对冲比率为1）
    spread = price_a - price_b
 
    # 计算滚动统计量
    spread_mean = spread.rolling(window=window).mean()
    spread_std = spread.rolling(window=window).std()
 
    # Z-score
    z_score = (spread - spread_mean) / spread_std
 
    # 生成信号
    positions_a = pd.Series(0, index=price_a.index)
    positions_b = pd.Series(0, index=price_b.index)
 
    # 价差过高 → A相对高估，做空A做多B
    positions_a[z_score > entry_threshold] = -1
    positions_b[z_score > entry_threshold] = 1
 
    # 价差过低 → A相对低估，做多A做空B
    positions_a[z_score < -entry_threshold] = 1
    positions_b[z_score < -entry_threshold] = -1
 
    # 回归 → 平仓
    close_signal = np.abs(z_score) < exit_threshold
    positions_a[close_signal] = 0
    positions_b[close_signal] = 0
 
    return positions_a, positions_b, spread, z_score
 
 
# 模拟配对数据
np.random.seed(42)
n_periods = 500
 
# 共同因子
common_factor = np.cumsum(np.random.normal(0, 1, n_periods))
 
# 两只股票受共同因子驱动，但有各自特质
price_a = 100 + common_factor + np.random.normal(0, 2, n_periods)
price_b = 100 + 0.8 * common_factor + np.random.normal(0, 1.5, n_periods)
 
dates = pd.date_range('2023-01-01', periods=n_periods, freq='D')
price_a = pd.Series(price_a, index=dates, name='Stock_A')
price_b = pd.Series(price_b, index=dates, name='Stock_B')
 
# 运行策略
pos_a, pos_b, spread, z_score = simplified_pairs_trading(price_a, price_b)
 
# 可视化
fig, axes = plt.subplots(3, 1, figsize=(14, 10))
 
# 价格
axes[0].plot(price_a.index, price_a.values, label='Stock A', alpha=0.7)
axes[0].plot(price_b.index, price_b.values, label='Stock B', alpha=0.7)
axes[0].set_title('配对交易：两只股票价格')
axes[0].legend()
axes[0].grid(True, alpha=0.3)
 
# 价差和 Z-score
ax2 = axes[1]
ax2.plot(spread.index, spread.values, label='Spread', color='steelblue')
ax2.set_ylabel('Spread', color='steelblue')
ax2.tick_params(axis='y', labelcolor='steelblue')
 
ax3 = ax2.twinx()
ax3.plot(z_score.index, z_score.values, label='Z-Score', color='orange', alpha=0.7)
ax3.axhline(y=2, color='red', linestyle='--', alpha=0.5)
ax3.axhline(y=-2, color='green', linestyle='--', alpha=0.5)
ax3.set_ylabel('Z-Score', color='orange')
ax3.tick_params(axis='y', labelcolor='orange')
 
axes[1].set_title('价差和 Z-Score')
 
# 持仓
axes[2].plot(pos_a.index, pos_a.values, label='Position A', linestyle='-', marker='.', markersize=3)
axes[2].plot(pos_b.index, pos_b.values, label='Position B', linestyle='-', marker='.', markersize=3)
axes[2].axhline(y=0, color='black', linestyle='-', alpha=0.3)
axes[2].set_title('持仓')
axes[2].set_ylabel('Position')
axes[2].legend()
axes[2].grid(True, alpha=0.3)
 
plt.tight_layout()
plt.show()

四、多因子选股 (Multi-Factor Selection)

4.1 从 ML 预测到多空组合

多因子策略是量化权益投资的主流框架：

┌─────────────────────────────────────────────────────────────┐
│                    多因子策略流程                            │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  因子数据        →  ML模型预测      →  组合构建              │
│  (估值/成长/质量)   (XGBoost/NN)      (多空组合)              │
│                                                             │
│     │                           │                          │
│     ├─── 价值因子: PE, PB, PS     │                          │
│     ├─── 成长因子: 营收增长       ├─── 信号转换:             │
│     ├─── 质量因子: ROE, ROA       │     预测 → 排名 → 分组    │
│     ├─── 动量因子: 过去收益       │                          │
│     ├─── 情绪因子: 换手率         │                          │
│     └─── 技术因子: 波动率         │                          │
│                                 │                          │
│                                 └─── 权重分配:              │
│                                      等权/市值/信号加权      │
│                                                             │
└─────────────────────────────────────────────────────────────┘

4.2 因子合成与组合构建

def multi_factor_portfolio(factor_data, lookback=20, top_quantile=0.2, bottom_quantile=0.2):
    """
    多因子组合构建
 
    参数:
        factor_data: 因子数据 DataFrame (T x N x K) 或字典形式
        lookback: 因子平滑窗口
        top_quantile: 做多头寸分位数
        bottom_quantile: 做空头寸分位数
 
    返回:
        positions: 持仓矩阵
    """
    # 简化：假设 factor_data 是多个因子加权后的综合得分
    # 实际应用中需要因子正交化、去极值、标准化等预处理
 
    # 因子合成：简单加权
    # 实际中可以用 ML 模型学习最优权重
    composite_score = factor_data.mean(axis=1)  # 如果是多因子，取平均
 
    # 横截面标准化
    standardized_score = composite_score.sub(composite_score.mean(axis=1), axis=0).div(
        composite_score.std(axis=1), axis=0
    )
 
    # 计算分位数
    positions = pd.DataFrame(0, index=composite_score.index, columns=composite_score.columns)
 
    for i in range(lookback, len(composite_score)):
        current_scores = standardized_score.iloc[i]
 
        # 做多头寸：得分最高的
        long_threshold = current_scores.quantile(1 - top_quantile)
        long_mask = current_scores >= long_threshold
        positions.iloc[i][long_mask] = 1 / long_mask.sum()
 
        # 做空头寸：得分最低的
        short_threshold = current_scores.quantile(bottom_quantile)
        short_mask = current_scores <= short_threshold
        positions.iloc[i][short_mask] = -1 / short_mask.sum()
 
    return positions
 
 
# 模拟因子数据
np.random.seed(42)
n_stocks = 100
n_periods = 500
 
# 生成模拟因子
factors = {}
factor_names = ['value', 'growth', 'quality', 'momentum']
 
for name in factor_names:
    # 每个因子有一些预测能力
    ic = np.random.uniform(0.02, 0.08)  # IC (Information Coefficient)
    factor = np.random.normal(0, 1, (n_periods, n_stocks))
 
    # 添加一些持续性（因子不是纯随机的）
    for i in range(1, n_periods):
        factor[i] = 0.9 * factor[i-1] + 0.1 * np.random.normal(0, 1, n_stocks)
 
    factors[name] = pd.DataFrame(
        factor,
        index=pd.date_range('2023-01-01', periods=n_periods, freq='D'),
        columns=[f'Stock_{i:03d}' for i in range(n_stocks)]
    )
 
# 合成因子（简单平均）
composite_factor = sum(factors.values()) / len(factors)
 
# 生成收益率（与因子有相关性）
returns = pd.DataFrame(
    np.random.normal(0.0005, 0.02, (n_periods, n_stocks)),
    index=composite_factor.index,
    columns=composite_factor.columns
)
 
# 添加因子收益贡献
for i in range(n_periods):
    factor_return = 0.03 * composite_factor.iloc[i].values  # IC 转化为收益
    returns.iloc[i] += factor_return
 
# 构建组合
positions = multi_factor_portfolio(composite_factor, lookback=20)
 
# 计算策略收益
strategy_returns = (positions.shift(1) * returns).sum(axis=1)
cumulative_returns = (1 + strategy_returns).cumprod()
 
# 对比：等权基准
equal_weight_returns = returns.mean(axis=1)
equal_weight_cumulative = (1 + equal_weight_returns).cumprod()
 
# 可视化
fig, axes = plt.subplots(2, 1, figsize=(14, 8))
 
# 累计收益对比
axes[0].plot(cumulative_returns.index, cumulative_returns.values,
             label='Multi-Factor Long-Short', linewidth=2)
axes[0].plot(equal_weight_cumulative.index, equal_weight_cumulative.values,
             label='Equal Weight Benchmark', linewidth=2, alpha=0.7)
axes[0].set_title('多因子策略累计收益')
axes[0].legend()
axes[0].grid(True, alpha=0.3)
 
# 分组收益分析
# 计算因子分位数
quantile_returns = pd.DataFrame()
for i in range(20, len(returns)):
    factor_rank = composite_factor.iloc[i].rank(pct=True)
 
    # 5分位
    q5_mask = factor_rank <= 0.2
    q1_mask = factor_rank >= 0.8
 
    q5_return = returns.iloc[i][q5_mask].mean()
    q1_return = returns.iloc[i][q1_mask].mean()
 
    quantile_returns.loc[returns.index[i], 'Q5 (High)'] = q5_return
    quantile_returns.loc[returns.index[i], 'Q1 (Low)'] = q1_return
    quantile_returns.loc[returns.index[i], 'Long-Short'] = q5_return - q1_return
 
# 分位数累计收益
axes[1].plot((1 + quantile_returns['Q5 (High)']).cumprod().index,
             (1 + quantile_returns['Q5 (High)']).cumprod().values,
             label='High Factor Score (Q5)', linewidth=1.5)
axes[1].plot((1 + quantile_returns['Q1 (Low)']).cumprod().index,
             (1 + quantile_returns['Q1 (Low)']).cumprod().values,
             label='Low Factor Score (Q1)', linewidth=1.5)
axes[1].set_title('因子分位数收益')
axes[1].legend()
axes[1].grid(True, alpha=0.3)
 
plt.tight_layout()
plt.show()

五、CTA 趋势跟踪 (CTA Trend Following)

CTA (Commodity Trading Advisor) 策略主要应用于期货市场，以趋势跟踪为核心。

5.1 CTA 策略特点

特点	描述
资产类别	商品期货、金融期货、股指期货
多空双向	可做多做空，不依赖牛市
杠杆使用	保证金交易，天然带杠杆
低相关性	与传统股债相关性低，有分散化价值
危机 Alpha	在市场恐慌时往往表现较好

5.2 典型 CTA 系统

def cta_trend_system(prices, fast_period=20, slow_period=60, vol_window=20, vol_target=0.15):
    """
    CTA 趋势跟踪系统（简化版）
 
    特点：
    1. 趋势识别（均线交叉）
    2. 波动率调整仓位
    3. 多空双向交易
 
    参数:
        prices: 价格序列
        fast_period, slow_period: 均线周期
        vol_window: 波动率计算窗口
        vol_target: 目标年化波动率
 
    返回:
        positions: 仓位序列（考虑杠杆）
    """
    # 计算趋势信号
    fast_ma = prices.rolling(window=fast_period).mean()
    slow_ma = prices.rolling(window=slow_period).mean()
 
    # 原始信号 (-1, 0, 1)
    raw_signal = pd.Series(0, index=prices.index)
    raw_signal[fast_ma > slow_ma] = 1
    raw_signal[fast_ma < slow_ma] = -1
 
    # 计算波动率
    returns = prices.pct_change()
    realized_vol = returns.rolling(window=vol_window).std() * np.sqrt(252)  # 年化
 
    # 波动率调整仓位
    # 目标：保持投资组合波动率在 vol_target
    vol_scaled_positions = raw_signal * vol_target / realized_vol
 
    # 限制最大杠杆（实际风控中很重要）
    max_leverage = 3
    vol_scaled_positions = vol_scaled_positions.clip(-max_leverage, max_leverage)
 
    return vol_scaled_positions, raw_signal, realized_vol
 
 
# 模拟期货价格（带趋势和周期）
np.random.seed(42)
n_periods = 500
 
# 趋势成分
trend = np.cumsum(np.random.normal(0.1, 1, n_periods))
 
# 周期成分
cycle = 10 * np.sin(np.linspace(0, 20*np.pi, n_periods))
 
# 噪声
noise = np.random.normal(0, 3, n_periods)
 
futures_price = pd.Series(
    100 + trend + cycle + noise,
    index=pd.date_range('2023-01-01', periods=n_periods, freq='D'),
    name='Futures'
)
 
# 运行 CTA 系统
positions, raw_signal, realized_vol = cta_trend_system(futures_price)
 
# 计算收益
returns = futures_price.pct_change()
strategy_returns = positions.shift(1) * returns
cumulative_returns = (1 + strategy_returns).cumprod()
 
# 可视化
fig, axes = plt.subplots(3, 1, figsize=(14, 10))
 
# 价格和信号
axes[0].plot(futures_price.index, futures_price.values, label='Futures Price', alpha=0.7)
axes[0].plot(futures_price.rolling(20).mean().index, futures_price.rolling(20).mean().values,
             label='MA20', alpha=0.7)
axes[0].plot(futures_price.rolling(60).mean().index, futures_price.rolling(60).mean().values,
             label='MA60', alpha=0.7)
 
# 标记多头区域
long_region = raw_signal == 1
axes[0].fill_between(futures_price.index, 0, futures_price.max(),
                     where=long_region, alpha=0.1, color='green', label='Long Zone')
 
axes[0].set_title('CTA 趋势系统：价格与信号')
axes[0].legend()
axes[0].grid(True, alpha=0.3)
 
# 仓位
axes[1].plot(positions.index, positions.values, label='Position (Vol-Adjusted)', color='steelblue')
axes[1].axhline(y=1, color='green', linestyle='--', alpha=0.5, label='Unit Long')
axes[1].axhline(y=-1, color='red', linestyle='--', alpha=0.5, label='Unit Short')
axes[1].axhline(y=3, color='black', linestyle=':', alpha=0.3, label='Max Leverage')
axes[1].axhline(y=-3, color='black', linestyle=':', alpha=0.3)
axes[1].fill_between(positions.index, 0, positions.values, where=positions.values > 0,
                     alpha=0.3, color='green')
axes[1].fill_between(positions.index, 0, positions.values, where=positions.values < 0,
                     alpha=0.3, color='red')
axes[1].set_title('仓位（波动率调整）')
axes[1].set_ylabel('Leverage')
axes[1].legend()
axes[1].grid(True, alpha=0.3)
 
# 累计收益
axes[2].plot(cumulative_returns.index, cumulative_returns.values, label='CTA Strategy', linewidth=2)
axes[2].plot((1 + returns).cumprod().index, (1 + returns).cumprod().values,
             label='Buy & Hold', linewidth=2, alpha=0.7)
axes[2].set_title('累计收益')
axes[2].legend()
axes[2].grid(True, alpha=0.3)
 
plt.tight_layout()
plt.show()

六、事件驱动策略 (Event-Driven)

6.1 策略类型

事件驱动策略基于特定事件对股价的影响：

事件类型	逻辑	典型持有期
财报超预期	盈余惊喜 → 股价上涨	数日到数周
并购重组	收购溢价 → 目标公司股价上涨	事件周期内
评级调整	分析师推荐变化 → 股价反应	短期
分红派息	除权除息 → 价格调整	短期
回购增持	公司回购 → 信心信号	中期

6.2 事件研究方法

def event_study(prices, event_dates, window_before=20, window_after=20):
    """
    事件研究：分析事件对价格的影响
 
    参数:
        prices: 价格数据 (DataFrame，每列是一只股票)
        event_dates: 事件日期字典 {stock_id: event_date}
        window_before: 事前窗口
        window_after: 事后窗口
 
    返回:
        abnormal_returns: 异常收益
        cumulative_abnormal_returns: 累积异常收益 (CAR)
    """
    abnormal_returns = {}
    cumulative_abnormal_returns = {}
 
    # 计算正常收益（简化：使用市场模型或历史均值）
    market_returns = prices.mean(axis=1).pct_change()
    estimated_returns = prices.pct_change()
 
    for stock, event_date in event_dates.items():
        if stock not in prices.columns:
            continue
 
        event_idx = prices.index.get_loc(event_date)
 
        # 定义事件窗口
        start_idx = max(0, event_idx - window_before)
        end_idx = min(len(prices), event_idx + window_after + 1)
 
        # 计算异常收益 = 实际收益 - 正常收益
        stock_returns = prices[stock].pct_change()
        ar = stock_returns.iloc[start_idx:end_idx] - market_returns.iloc[start_idx:end_idx]
 
        # 累积异常收益
        car = ar.cumsum()
 
        abnormal_returns[stock] = ar
        cumulative_abnormal_returns[stock] = car
 
    return abnormal_returns, cumulative_abnormal_returns
 
 
# 模拟事件数据
np.random.seed(42)
n_stocks = 50
n_periods = 500
 
# 生成价格数据
prices = pd.DataFrame(
    np.random.lognormal(0.0005, 0.02, (n_periods, n_stocks)) * 100,
    index=pd.date_range('2023-01-01', periods=n_periods, freq='D'),
    columns=[f'Stock_{i:02d}' for i in range(n_stocks)]
)
 
# 模拟事件（财报发布）
event_dates = {}
for i in range(20):  # 20个事件
    stock = f'Stock_{np.random.randint(0, n_stocks):02d}'
    # 随机事件日期（避开头尾）
    event_date = prices.index[np.random.randint(50, n_periods - 50)]
    event_dates[stock] = event_date
 
    # 在事件后添加价格反应（模拟超预期效应）
    event_idx = prices.index.get_loc(event_date)
    reaction = np.random.uniform(0.02, 0.08)  # 2-8% 的正向反应
    for j in range(event_idx, min(event_idx + 5, n_periods)):
        prices.iloc[j][stock] *= (1 + reaction * (1 - (j - event_idx) / 5))
 
# 运行事件研究
ar, car = event_study(prices, event_dates, window_before=20, window_after=20)
 
# 可视化平均 CAR
plt.figure(figsize=(12, 6))
 
# 对齐 CAR
aligned_car = pd.DataFrame()
for stock, car_series in car.items():
    aligned_car[stock] = car_series.values
 
# 计算平均
avg_car = aligned_car.mean(axis=1)
std_car = aligned_car.std(axis=1)
 
x_axis = np.arange(-20, 20)
plt.plot(x_axis, avg_car.values, 'o-', label='Average CAR', linewidth=2)
plt.fill_between(x_axis, avg_car.values - std_car.values, avg_car.values + std_car.values,
                 alpha=0.2)
plt.axvline(x=0, color='red', linestyle='--', label='Event Day')
plt.axhline(y=0, color='black', linestyle='-', alpha=0.3)
plt.xlabel('Days relative to event')
plt.ylabel('Cumulative Abnormal Return')
plt.title('事件研究：平均累积异常收益')
plt.legend()
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()

七、波动率策略 (Volatility Strategies)

7.1 波动率卖出策略

def volatility_strategy(prices, vol_window=20, vol_percentile_high=80, vol_percentile_low=20):
    """
    波动率卖出策略
 
    逻辑：
    - 波动率低时，卖出期权（收取权利金）
    - 波动率高时，减少暴露
 
    注意：这是简化概念展示，实际需要完整的期权定价和对冲
    """
    # 计算已实现波动率
    returns = prices.pct_change()
    realized_vol = returns.rolling(window=vol_window).std() * np.sqrt(252)
 
    # 计算波动率分位数
    vol_rank = realized_vol.rolling(window=252).apply(
        lambda x: pd.Series(x).rank(pct=True).iloc[-1]
    )
 
    # 信号：低波动率时增加暴露，高波动率时减少
    signal = pd.Series(0, index=prices.index)
    signal[vol_rank < vol_percentile_low / 100] = 1  # 低波动 → 做空波动率
    signal[vol_rank > vol_percentile_high / 100] = -1  # 高波动 → 减少
 
    return signal, realized_vol, vol_rank

7.2 波动率期限结构套利

短期波动率 vs 长期波动率

如果短期波动率 > 长期波动率（异常高）
→ 做空短期波动率，做多长期波动率

逻辑：波动率均值回归

八、跨资产宏观策略

基于宏观因子的资产配置策略：

宏观因子	相关资产	策略逻辑
经济增长	股票、商品	增长超预期 → 做多周期股
通胀	TIPS、商品	通胀上升 → 做多商品
利率	债券	利率下行 → 做多长久期债
信用利差	高收益债	利差收窄 → 做多高收益债

九、策略对比总结

策略类型	核心逻辑	频率	容量	回撤特征	适用市场
时间序列动量	趋势跟踪	低-中	高	大回撤但集中	趋势市
截面动量	相对强弱	中	中	中等	多资产市场
均值回归	价格回归	中-高	中	小而频繁	震荡市
统计套利	价差回归	高-低	低	小而平稳	相关资产
多因子选股	综合评分	低	很高	市场同步	股票市场
CTA 趋势	期货趋势	低-中	中	大回撤但分散	期货市场
事件驱动	事件反应	中	中	事件相关	全市场

十、ML 如何增强传统策略

10.1 信号生成增强

from sklearn.ensemble import RandomForestRegressor
from sklearn.model_selection import train_test_split
 
def ml_enhanced_momentum(prices, lookback=60, prediction_horizon=5):
    """
    ML 增强的动量策略
 
    传统：简单使用过去收益率
    ML增强：预测未来收益率，使用多维特征
    """
    # 构建特征
    features = pd.DataFrame(index=prices.index)
 
    # 基础动量特征
    for lag in [5, 10, 20, 60]:
        features[f'momentum_{lag}'] = prices.pct_change(lag)
 
    # 波动率特征
    for window in [10, 20, 60]:
        features[f'volatility_{window}'] = prices.pct_change().rolling(window).std()
 
    # 技术指标
    features['rsi'] = calculate_rsi(prices, 14)
    features['ma_ratio'] = prices / prices.rolling(20).mean()
 
    # 目标变量：未来收益
    target = prices.shift(-prediction_horizon) / prices - 1
 
    # 准备训练数据
    valid_idx = ~(features.isna().any(axis=1) | target.isna())
    X = features[valid_idx]
    y = target[valid_idx]
 
    # 简单的滚动训练（实际中需要更复杂的在线学习）
    predictions = pd.Series(np.nan, index=prices.index)
 
    for i in range(252, len(prices)):  # 留一年数据用于初始训练
        train_X = X.iloc[:i]
        train_y = y.iloc[:i]
 
        model = RandomForestRegressor(n_estimators=50, max_depth=5, random_state=42)
        model.fit(train_X, train_y)
 
        pred = model.predict([X.iloc[i].values])[0]
        predictions.iloc[i] = pred
 
    # 转换为信号
    signal = (predictions > 0).astype(int).replace(0, -1)
 
    return signal, predictions
 
 
def calculate_rsi(prices, period=14):
    """计算 RSI 指标"""
    delta = prices.diff()
    gain = delta.where(delta > 0, 0)
    loss = -delta.where(delta < 0, 0)
 
    avg_gain = gain.rolling(window=period).mean()
    avg_loss = loss.rolling(window=period).mean()
 
    rs = avg_gain / avg_loss
    rsi = 100 - (100 / (1 + rs))
    return rsi

10.2 ML 增强维度

传统方法	ML 增强方法
固定参数规则	自适应参数优化
单一信号源	多特征非线性组合
简单阈值	概率预测
静态权重	动态权重分配
手工特征	自动特征发现

核心知识点总结

动量策略

时间序列动量：单个资产的趋势跟踪
截面动量：跨资产的相对强弱
行为金融解释：反应不足、羊群效应

均值回归

核心假设：价格偏离后回归
布林带：经典的均值回归工具
配对交易：价差的均值回归

统计套利

价差平稳性：协整关系是关键
市场中性：消除市场方向暴露
风险控制：协整断裂风险

多因子选股

因子分类：价值、成长、质量、动量、情绪
因子合成：加权、正交化、ML 模型
组合构建：排名、分组、多空

CTA 策略

趋势跟踪：均线、突破
波动率调整：风险平价、目标波动率
危机 Alpha：与传统资产低相关

策略选择原则

市场状态        →  策略选择
─────────────────────────────────
趋势明显        →  趋势跟踪、CTA
震荡整理        →  均值回归、统计套利
个股分化        →  截面动量、多因子
高波动         →  波动率策略
特定事件       →  事件驱动

下一步

继续学习 02-统计套利详解.md，深入了解统计套利的数学原理和完整实现。

MindCarver Blog

MindCarver

探索

01-主流策略类型