CTA 策略入门：趋势跟踪、期货套利与 Python 实战

import numpy as np

def simulate_trend_following(n_days=1000, short_window=20, long_window=60):
    np.random.seed(42)

    # 模拟期货价格：5 个阶段交替出现趋势和震荡
    regimes = [
        (0, 150, 0.0008, 0.012),     # 上涨趋势
        (150, 300, -0.0001, 0.018),   # 震荡
        (300, 500, -0.0006, 0.013),   # 下跌趋势
        (500, 700, 0.0001, 0.020),    # 震荡
        (700, 1000, 0.0007, 0.011),   # 上涨趋势
    ]

    returns = np.zeros(n_days)
    for start, end, mu, sigma in regimes:
        returns[start:end] = np.random.normal(mu, sigma, end - start)

    price = 100 * np.exp(np.cumsum(returns))

    # 计算均线
    ma_short = np.array([np.mean(price[max(0, i-short_window+1):i+1])
                         if i >= short_window - 1 else np.nan
                         for i in range(n_days)])
    ma_long = np.array([np.mean(price[max(0, i-long_window+1):i+1])
                        if i >= long_window - 1 else np.nan
                        for i in range(n_days)])

    # 信号：短均线 > 长均线做多，反之做空
    position = np.zeros(n_days)
    for i in range(long_window, n_days):
        position[i] = 1 if ma_short[i] > ma_long[i] else -1

    # 策略收益
    strategy_ret = position[:-1] * returns[1:]
    strategy_ret = np.insert(strategy_ret, 0, 0)
    strategy_equity = np.exp(np.cumsum(strategy_ret))
    buyhold_equity = price / price[0]

    # 统计指标
    active_ret = strategy_ret[long_window:]
    sharpe = np.mean(active_ret) / np.std(active_ret) * np.sqrt(252)
    peak = np.maximum.accumulate(strategy_equity)
    max_dd = np.min((strategy_equity - peak) / peak)

    print(f"策略终值: {strategy_equity[-1]:.2f}x")
    print(f"买入持有终值: {buyhold_equity[-1]:.2f}x")
    print(f"夏普比率: {sharpe:.2f}")
    print(f"最大回撤: {max_dd:.1%}")

simulate_trend_following()

策略终值: 1.35x
买入持有终值: 1.51x
夏普比率: 0.34
最大回撤: -42.0%

import numpy as np

def simulate_calendar_spread():
    np.random.seed(123)
    n = 500

    # 模拟两个相关期货合约（近月 vs 远月）
    common = np.cumsum(np.random.normal(0.0002, 0.012, n))
    near = 3500 + common * 100 + np.cumsum(np.random.normal(0, 0.8, n))

    # 远月 = 近月 + 升水 + 均值回复噪声
    spread_noise = np.zeros(n)
    for i in range(1, n):
        spread_noise[i] = 0.93 * spread_noise[i-1] + np.random.normal(0, 2.5)
    far = near + 30 + spread_noise

    spread = far - near
    lookback = 40

    # 滚动 Z-score
    z_score = np.full(n, np.nan)
    for i in range(lookback, n):
        window = spread[i-lookback:i]
        std_w = np.std(window)
        if std_w > 0:
            z_score[i] = (spread[i] - np.mean(window)) / std_w

    # 信号
    position = np.zeros(n)
    for i in range(lookback + 1, n):
        if np.isnan(z_score[i]):
            position[i] = position[i-1]
            continue
        if position[i-1] == 0:
            if z_score[i] > 2:    position[i] = -1   # 价差过大，做空价差
            elif z_score[i] < -2: position[i] = +1   # 价差过小，做多价差
        elif position[i-1] == 1:
            if abs(z_score[i]) < 0.5:  position[i] = 0    # 回归，平仓
            elif z_score[i] < -4:      position[i] = 0    # 止损
            else:                      position[i] = 1
        elif position[i-1] == -1:
            if abs(z_score[i]) < 0.5:  position[i] = 0
            elif z_score[i] > 4:       position[i] = 0
            else:                      position[i] = -1

    # 盈亏
    spread_ret = np.diff(spread)
    start = lookback + 1
    daily_pnl = position[start:-1] * spread_ret[start:]
    daily_pnl = daily_pnl[~np.isnan(daily_pnl)]
    n_trades = int(np.sum(np.abs(np.diff(position[start:])) > 0))

    print(f"累计盈亏: {np.sum(daily_pnl):.2f}")
    print(f"交易次数: {n_trades} ({n_trades//2} 个来回)")
    print(f"年化夏普: {np.mean(daily_pnl)/np.std(daily_pnl)*np.sqrt(252):.2f}")

simulate_calendar_spread()

累计盈亏: 96.14
交易次数: 34 (17 个来回)
年化夏普: 2.49