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

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

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

1. 波动率分层函数

import numpy as np
import pandas as pd

def volatility_tier(close, short=20, long=60):
    '''波动率分层：按短期/长期波动率比值识别市场状态
    用于波动率自适应仓位与行情风格切换判断
    '''
    ret = close.pct_change()
    vol_s = ret.rolling(short).std() * np.sqrt(252)
    vol_l = ret.rolling(long).std() * np.sqrt(252)
    ratio = vol_s / (vol_l + 1e-8)
    tier = pd.Series('normal', index=close.index)
    tier[ratio > 1.3] = 'high_vol'    # 波动放大
    tier[ratio < 0.7] = 'low_vol'     # 波动收敛
    return tier

2. 四步适配法校验框架

def four_step_adapt(params, market):
    '''四步适配法校验：通用/海外策略落地到具体市场前的强制校验
    步骤：制度约束 -> 成本测算 -> 参数本地化 -> 样本外验证
    任一步不通过则禁止上线，杜绝直接照搬参数
    '''
    checks = {
        '制度约束': market.get('t_plus') is not None,
        '成本测算': market.get('cost', 0) < params.get('edge', 0),
        '参数本地化': params.get('localized', False),
        '样本外验证': params.get('oos_passed', False),
    }
    checks['全部通过'] = all(checks.values())
    return checks

3. 三角套利核心逻辑

def triangular_arbitrage(eurusd, eurgbp, gbpusd, cost=0.0005):
    '''外汇三角套利：USD -> EUR -> GBP -> USD 闭环价差
    闭环乘积 = EURGBP × GBPUSD / EURUSD，偏离 1 即存在理论套利空间
    须扣除点差与滑点 cost，实盘空间往往被成本吞噬
    '''
    product = eurgbp * gbpusd / eurusd
    edge = abs(product - 1.0)
    return {'闭环乘积': product, '净价差': edge - cost, '是否可行': edge > cost}

4. 窄口突破策略模板

import numpy as np
import pandas as pd

def narrow_breakout(high, low, close, n=20, k=0.75):
    '''低位窄口突破：波动率收敛(窄口)后放量突破
    窄口 = n日振幅收窄至历史 k 分位以下；突破 = 收盘上破窄口箱体上沿
    '''
    width = (high - low) / close
    narrow = width.rolling(n).mean() < width.rolling(n * 4).quantile(k)
    box_top = high.rolling(n).max()
    signal = narrow.shift(1) & (close > box_top.shift(1))
    return signal.astype(int)

5. 风险平价组合计算

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()

6. VWAP 拆单算法

import numpy as np

def vwap_schedule(total_volume, volume_profile):
    '''VWAP 拆单：按历史成交量分布把大单拆到各时段，降低冲击成本
    volume_profile: 各时段成交量占比；返回各时段应下单量
    '''
    profile = np.asarray(volume_profile, dtype=float)
    profile = profile / (profile.sum() + 1e-8)   # 归一化
    return total_volume * profile

def vwap_price(prices, volumes):
    '''VWAP 成交价 = Σ(价格×成交量) / Σ成交量，用于评估执行质量'''
    return float((prices * volumes).sum() / (volumes.sum() + 1e-8))

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 order_route(venues):
    '''订单路由：多市场/多通道选择最优执行通道
    venues: list of dict, 含 name, fee(费率), liquidity(流动性), latency(延迟)
    综合流动性、费率、延迟打分，选最优通道；教学骨架
    '''
    scored = []
    for v in venues:
        score = v['liquidity'] / (v['fee'] + 1e-6) - v['latency']
        scored.append((score, v['name']))
    scored.sort(reverse=True)
    return scored[0][1]