附录三 核心策略模板与代码

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

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

1. 协整检验与配对筛选

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 pandas as pd

def pair_screen(df, corr_min=0.8):
    '''配对筛选：先相关性初筛，再逐个做协整检验（相关性 ≠ 协整）
    df: 多只股票收盘价 DataFrame；返回通过协整的配对列表
    '''
    pairs = []
    cols = df.columns
    for i in range(len(cols)):
        for j in range(i + 1, len(cols)):
            a, b = df[cols[i]], df[cols[j]]
            if a.corr(b) < corr_min:
                continue
            pv, ok = cointegration_check(a, b)
            if ok:
                pairs.append((cols[i], cols[j], round(pv, 4)))
    return pairs

2. Z 值计算与交易信号函数

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)


def trade_by_z(z, open_th=1.5, close_th=0.5, stop_th=2.5):
    '''按 Z 值生成开平仓动作：|Z|>=开仓阈值反向建仓，回归平仓，超止损阈值离场'''
    if z >= stop_th or z <= -stop_th:
        return '止损平仓'
    if z >= open_th:
        return '做空价差'      # 价差偏高，卖高买低
    if z <= -open_th:
        return '做多余价差'    # 价差偏低
    if abs(z) <= close_th:
        return '止盈平仓'
    return '持仓不动'

3. 期现基差计算模板

def futures_fair_price(spot, r, d, days):
    '''股指期货理论价格：F = S * (1 + r - d) ** (T/365)
    spot 现货指数, r 无风险利率, d 股息率, days 距交割天数
    '''
    return spot * (1 + r - d) ** (days / 365)

def basis(futures, spot):
    '''基差 = 期货价格 - 现货指数
    基差过高 -> 正向套利(买现货卖期货)；过低 -> 反向套利(受融券限制难落地)
    '''
    return futures - spot

4. AH 溢价率计算模板

def ah_premium(a_price, h_price, hkd_to_cny):
    '''AH 溢价率 = A股价格 / (H股价格 × 港币兑人民币汇率) - 1
    同股同权价格偏离；须按个股历史溢价分位数判断，忌用统一阈值
    hkd_to_cny: 1 港币兑换人民币的汇率
    '''
    h_in_cny = h_price * hkd_to_cny      # H股换算为人民币
    return a_price / (h_in_cny + 1e-8) - 1

5. 可转债双低筛选代码

import pandas as pd

def double_low_filter(cb_df, price_max=130, premium_max=0.30):
    '''可转债双低策略：低转债价格 + 低转股溢价率
    cb_df 需含 price(转债价格), premium(转股溢价率)
    债底提供下跌保护，低溢价保留正股上涨弹性；非保本，警惕强赎与退市
    '''
    pool = cb_df[(cb_df['price'] <= price_max) &
                 (cb_df['premium'] <= premium_max)].copy()
    pool['double_low'] = pool['price'] + pool['premium'] * 100
    return pool.sort_values('double_low').head(20)

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. 五级风控触发逻辑

def five_level_risk(account):
    '''五级风控防御网 + 账户三级熔断（优先保证账户存活）
    五级：策略级 -> 品种/市场级 -> 组合级 -> 账户级 -> 合规级
    account: dict, 含 drawdown(高点回撤比例)
    '''
    dd = account.get('drawdown', 0.0)
    if dd >= 0.18:
        return '全部清仓'          # 三级熔断：清仓
    if dd >= 0.12:
        return '仓位降至20%'       # 二级熔断
    if dd >= 0.08:
        return '仓位减半'          # 一级熔断
    return '正常持有'

def strategy_stop(strategy):
    '''策略级风控：单策略回撤或失效超阈值即降级/暂停，不死扛'''
    if strategy.get('drawdown', 0.0) >= strategy.get('stop_line', 0.10):
        return '暂停策略'
    if strategy.get('ic_decay', False):
        return '降权观察'
    return '继续运行'

8. 订单执行接口模板

def execute_order(symbol, volume, side, limit_price=None):
    '''订单执行接口模板（对接券商/交易 API）
    side: 'buy' / 'sell'；limit_price 为空则市价单
    实盘须处理滑点、部分成交、废单重试；此处为教学骨架
    '''
    order = {'symbol': symbol, 'volume': volume,
             'side': side, 'limit_price': limit_price}
    # TODO: 替换为真实券商 API 调用
    print('提交订单:', order)
    return {'status': 'submitted', 'order': order}