ptrade 风控实战：动态组合波动率风控与仓位压制

在量化投资中，随着市场波动的加剧，固定仓位策略往往承受较大的最大回撤。通过**目标波动率模型（Target Volatility）**对持仓组合的整体波动率进行实时监控，并在波动率过高时动态压制总仓位，是实现资产平滑收益曲线的有效手段。

本文将结合 ptrade API 介绍如何在策略中实现组合波动率计算与动态仓位压制风控。

1. 核心风控逻辑

1. 计算资产收益率与协方差矩阵：定期（如每日盘前）获取持仓股票池的历史日收盘价，计算各标的日收益率及协方差矩阵。
2. 计算组合年化波动率：结合当前各标的权重，利用公式计算组合整体波动率：
   $$\sigma_{portfolio} = \sqrt{w^T \Sigma w} \times \sqrt{252}$$
3. 动态计算仓位调整系数：根据预设的目标年化波动率（Target Vol）计算仓位系数：
   $$Scale = \min\left(1.0, \frac{Target\ Vol}{\sigma_{portfolio}}\right)$$
4. 仓位压制执行：若当前组合波动率超出目标值，按系数 $Scale$ 缩减各标的的目标持仓市值。

2. ptrade 代码实现

以下为在 ptrade 平台中实现的完整动态波动率风控示例：

import numpy as np
import pandas as pd

def initialize(context):
    # 设置股票池
    g.security = ['600570.SS', '000001.SZ', '600000.SS', '000002.SZ']
    set_universe(g.security)
    
    # 风控参数设置
    g.target_vol = 0.15  # 目标年化波动率 15%
    g.lookback_days = 60 # 历史数据回溯天数
    g.scale = 1.0        # 仓位压制系数

def before_trading_start(context, data):
    """
    盘前计算持仓组合波动率并更新仓位调整系数
    """
    # 1. 获取持仓组合历史收盘价数据
    df = get_history(g.lookback_days + 1, '1d', 'close', security_list=g.security, fq='pre')
    if df is None or df.empty:
        return
    
    # 提取价格数据计算日收益率
    if isinstance(df.index, pd.MultiIndex) or 'code' in df.columns:
        price_df = df.pivot(columns='code', values='close')
    else:
        price_df = df
        
    returns = price_df.pct_change().dropna()
    
    # 2. 计算协方差矩阵（年化）
    cov_matrix = returns.cov() * 252
    
    # 3. 计算当前持仓权重（若无持仓则默认等权重）
    positions = get_positions()
    total_value = context.portfolio.portfolio_value
    weights = []
    
    for sec in g.security:
        pos = positions.get(sec, None)
        if pos and total_value > 0:
            weights.append((pos.amount * pos.last_sale_price) / total_value)
        else:
            weights.append(1.0 / len(g.security))
            
    weights = np.array(weights)
    # 权重归一化
    if weights.sum() > 0:
        weights = weights / weights.sum()
    
    # 4. 计算组合年化波动率
    port_variance = np.dot(weights.T, np.dot(cov_matrix, weights))
    port_vol = np.sqrt(port_variance)
    
    # 5. 计算仓位压制系数 Scale
    if port_vol > g.target_vol:
        g.scale = g.target_vol / port_vol
        log.info(f"[风控触发] 当前组合年化波动率: {port_vol:.2%}, 超过目标值: {g.target_vol:.2%}, 仓位压制系数调整为: {g.scale:.2f}")
    else:
        g.scale = 1.0
        log.info(f"[风控正常] 当前组合年化波动率: {port_vol:.2%}, 无需压制仓位")

def handle_data(context, data):
    # 示例逻辑：根据策略买入，并应用风控压制系数
    cash = context.portfolio.cash
    target_per_stock = (context.portfolio.portfolio_value / len(g.security)) * g.scale
    
    for sec in g.security:
        if data[sec].is_open > 0:
            # 根据动态调整后的目标市值下单
            order_target_value(sec, target_per_stock)

3. 注意事项与开发建议

1. 数据准备与清洗：在调用 get_history 时，需注意过滤停牌及新股上市初期数据，避免异常波动对协方差矩阵产生误导。
2. 调仓频率控制：波动率风控不建议过于频繁地大幅调整仓位，可通过设置调仓阈值缓冲带（如波动率偏差超过 10% 才触发调整），避免换手率过高带来的印花税与手续费磨损。
3. 与单股风控结合：组合波动率风控属于宏观仓位控制，建议配合单股止损（如个股回撤 8% 止损）联合使用，构建多层次风控体系。