附录三 Python策略开发核心代码模板

注：以下为各模块核心教学代码模板（Python 3 + numpy/pandas）；完整工程化框架、实盘系统与数据接口可访问 intoquant.com 获取。

⚠ 教学参考：以下代码为方法论教学模板，未经本站实盘验证，不构成投资建议；实盘请自行回测、样本外验证、小资金试运行。

1. 因子预处理（去极值 / 截面标准化 / 正交化）

import numpy as np
import statsmodels.api as sm

def winsorize(factor, n=3):
    '''去极值：中位数绝对偏差法(MAD)，超过 n 倍压回边界'''
    median = factor.median()
    mad = (factor - median).abs().median()
    band = n * 1.4826 * mad
    return factor.clip(median - band, median + band)

def standardize(factor):
    '''截面标准化：Z-score，均值0标准差1'''
    return (factor - factor.mean()) / (factor.std() + 1e-8)

def orthogonalize(factor, neutral):
    '''正交化：对行业/市值等中性因子回归取残差，去除多重共线性'''
    return sm.OLS(factor, sm.add_constant(neutral)).fit().resid

2. 单因子回测 / 分层测试 / IC 计算

import pandas as pd

def calc_ic(factor, fwd_return, method='spearman'):
    '''IC 信息系数：因子值与下期收益相关性；spearman 为 RankIC 更稳健'''
    return factor.corr(fwd_return, method=method)

def layered_backtest(factor, fwd_return, groups=5):
    '''分层测试：按因子分 N 组检验单调性，理想呈单调递增/递减'''
    labels = pd.qcut(factor.rank(method='first'), groups, labels=False)
    return fwd_return.groupby(labels).mean()

3. 时间序列切分 / 滚动回测框架

def time_split(df, train_end, valid_end):
    '''时间序列切分：严禁随机打乱，按时间先后划分 train/valid/test'''
    train = df[df.index <= train_end]
    valid = df[(df.index > train_end) & (df.index <= valid_end)]
    test = df[df.index > valid_end]   # 纯样本外，绝不参与训练调参
    return train, valid, test

def rolling_backtest(df, signal_fn, window=252, step=63):
    '''滚动回测：逐窗口只用历史数据生成信号，模拟真实实盘迭代过程'''
    out = []
    for s in range(0, len(df) - window, step):
        hist = df.iloc[s:s + window]
        future = df.iloc[s + window:s + window + step]
        out.append(signal_fn(hist, future))
    return out

4. 配对交易：协整检验与 Z 值信号

import statsmodels.api as sm

def cointegration_check(price_a, price_b):
    '''Engle-Granger 协整检验：p 值 < 0.05 才具备配对基础
    高相关 ≠ 协整，协整才代表价差长期存在回归动力
    '''
    pvalue = sm.tsa.coint(price_a, price_b)[1]
    return pvalue, pvalue < 0.05


import numpy as np

def zscore_signal(spread, window=120):
    '''滚动 Z 值交易信号
    开仓 |Z|>=1.5, 平仓 |Z|<=0.5, 止损 |Z|>=2.5（阈值需按品种本地化）
    '''
    mean = spread.rolling(window).mean()
    std = spread.rolling(window).std()
    return (spread - mean) / std.replace(0, np.nan)

5. CTA 策略信号模板

import numpy as np
import pandas as pd

def donchian_signal(high, low, close, n=40):
    '''唐奇安通道突破：多品种多周期、多空双向；上破开多、下破开空'''
    upper = high.rolling(n).max().shift(1)
    lower = low.rolling(n).min().shift(1)
    sig = pd.Series(0, index=close.index)
    sig[close > upper] = 1
    sig[close < lower] = -1
    return sig

def vol_target_position(returns, target_vol=0.15, lookback=20):
    '''目标波动率控仓：波动率反比仓位，控制总杠杆'''
    realized = returns.rolling(lookback).std() * np.sqrt(252)
    return (target_vol / realized.replace(0, np.nan)).clip(upper=1.0)

6. 风险平价组合计算

import numpy as np

def risk_parity_weights(returns):
    '''风险平价权重（简化版：波动率倒数加权）
    returns: (T, N) 收益率矩阵；返回 (N,) 权重，和为 1
    各策略风险贡献近似相等，鲁棒性好，适合个人投资者
    '''
    vol = returns.std(axis=0, ddof=1) * np.sqrt(252)   # 年化波动率
    inv_vol = 1.0 / np.where(vol > 1e-8, vol, 1e-8)    # 防除零
    return inv_vol / inv_vol.sum()                     # 归一化

def risk_contribution(weights, cov):
    '''各策略风险贡献占比，用于验证是否达到风险平价；理想接近等权 1/N'''
    port_vol = np.sqrt(weights @ cov @ weights)
    rc = weights * (cov @ weights) / port_vol
    return rc / rc.sum()

7. 参数扫描热力图生成模板

import itertools
import pandas as pd

def param_scan(backtest_fn, param_grid):
    '''参数扫描：遍历参数组合输出夏普矩阵用于画热力图
    选平坦高原区，不选孤立尖峰（尖峰 = 过拟合）
    param_grid 如 {'n': [20,40,60], 'k': [1.5,2.0,2.5]}
    '''
    keys = list(param_grid.keys())
    rows = []
    for combo in itertools.product(*[param_grid[k] for k in keys]):
        params = dict(zip(keys, combo))
        rows.append({**params, 'sharpe': backtest_fn(**params)})
    df = pd.DataFrame(rows)
    return df.pivot_table(index=keys[0], columns=keys[1], values='sharpe')

8. 绩效指标计算模板

import numpy as np

def performance_metrics(returns, rf=0.0):
    '''绩效指标：年化收益 / 夏普 / 最大回撤 / 卡玛；returns 为日收益率序列'''
    ann_ret = (1 + returns).prod() ** (252 / len(returns)) - 1
    ann_vol = returns.std() * np.sqrt(252)
    sharpe = (ann_ret - rf) / (ann_vol + 1e-8)
    nav = (1 + returns).cumprod()
    max_dd = ((nav - nav.cummax()) / nav.cummax()).min()
    calmar = ann_ret / (abs(max_dd) + 1e-8)
    return {'年化收益': ann_ret, '夏普': sharpe, '最大回撤': max_dd, '卡玛': calmar}