如何在 QMT 策略中计算并应用投资组合的在险价值 (VaR) 进行风控？

在量化交易中，在险价值（Value at Risk, VaR） 是一个非常核心的风险管理指标。它回答了一个问题：“在给定的置信水平（例如95%）下，我的投资组合在未来特定时间段（例如1天）内，最大的潜在损失是多少？”

核心逻辑

我们将采用 历史模拟法（Historical Simulation） 来计算 VaR，这种方法不需要假设收益率服从正态分布，能够较好地捕捉市场的“肥尾”风险。

策略逻辑步骤：

1. 获取持仓信息：获取当前账户中所有持仓股票的市值。
2. 获取历史数据：拉取这些股票过去 N 天（如252天）的历史收盘价。
3. 模拟组合收益：假设当前持仓权重在过去一直保持不变，计算该组合在历史每一天的模拟盈亏（PnL）。
4. 计算 VaR：对历史模拟盈亏进行排序，找到对应置信度（如 5% 分位数）的损失值，即为 VaR。
5. 风控操作：设定一个 VaR 限额（例如总资产的 2%）。如果计算出的当前 VaR 超过了限额，说明风险过高，需要按比例减仓，直到 VaR 回到安全线以内。

QMT 策略代码实现

以下是一个完整的 Python 策略代码。该策略会在每日盘中计算 VaR，如果风险超标，则自动执行减仓操作。

# -*- coding: gbk -*-
import pandas as pd
import numpy as np
import datetime

def init(ContextInfo):
    """
    策略初始化函数
    """
    # 1. 设置资金账号 (请修改为您的实际账号)
    ContextInfo.accid = '6000000000' 
    ContextInfo.accountType = 'STOCK'
    ContextInfo.set_account(ContextInfo.accid)
    
    # 2. VaR计算参数设置
    ContextInfo.lookback_days = 252      # 回溯历史天数 (通常取一年)
    ContextInfo.confidence_level = 0.95  # 置信水平 95%
    ContextInfo.holding_period = 1       # 持有期 1天
    
    # 3. 风控阈值设置
    # 允许的最大VaR占总资产的比例。例如 0.02 代表如果不希望单日最大潜在亏损超过总资产的 2%
    ContextInfo.max_var_ratio = 0.02     
    
    # 4. 运行控制
    ContextInfo.run_risk_control = True  # 是否开启风控交易
    print("策略初始化完成，VaR风控模块已加载。")

def get_portfolio_positions(ContextInfo):
    """
    获取当前账户持仓信息
    返回: DataFrame (index=stock_code, columns=['volume', 'market_value'])
    """
    positions = get_trade_detail_data(ContextInfo.accid, ContextInfo.accountType, 'POSITION')
    
    data_list = []
    for pos in positions:
        # 过滤掉持仓量为0的记录
        if pos.m_nVolume > 0:
            data_list.append({
                'code': pos.m_strInstrumentID + '.' + pos.m_strExchangeID,
                'volume': pos.m_nVolume,
                'market_value': pos.m_dMarketValue
            })
    
    if not data_list:
        return pd.DataFrame()
        
    df = pd.DataFrame(data_list)
    df.set_index('code', inplace=True)
    return df

def calculate_portfolio_var(ContextInfo, position_df, total_assets):
    """
    使用历史模拟法计算投资组合的 VaR
    """
    if position_df.empty:
        return 0.0

    stock_list = position_df.index.tolist()
    
    # 获取历史行情数据 (多取一些以防停牌)
    count = ContextInfo.lookback_days + 20
    # 使用 get_market_data_ex 获取数据
    # 注意：这里获取的是前复权数据，以准确计算收益率
    market_data = ContextInfo.get_market_data_ex(
        ['close'], 
        stock_list, 
        period='1d', 
        count=count, 
        dividend_type='front'
    )
    
    # 构建历史价格矩阵
    price_dict = {}
    for stock in stock_list:
        if stock in market_data:
            df = market_data[stock]
            # 确保按时间排序
            df = df.sort_index()
            # 取最近 lookback_days + 1 天的数据用于计算收益率
            price_dict[stock] = df['close'].tail(ContextInfo.lookback_days + 1)
    
    if not price_dict:
        return 0.0
        
    prices_df = pd.DataFrame(price_dict)
    
    # 计算日收益率
    returns_df = prices_df.pct_change().dropna()
    
    if returns_df.empty:
        print("警告：历史数据不足，无法计算VaR")
        return 0.0

    # 计算当前各股票的权重 (市值 / 总持仓市值)
    # 注意：这里我们模拟的是“当前持仓组合”在历史上的表现
    # 因此直接用当前的市值作为权重系数应用到历史收益率上
    
    # 模拟历史每日的盈亏金额 (Simulated Daily PnL)
    # 公式：Sum(个股当前市值 * 个股历史当日收益率)
    simulated_pnl = pd.Series(0.0, index=returns_df.index)
    
    for stock in returns_df.columns:
        if stock in position_df.index:
            current_mkt_val = position_df.loc[stock, 'market_value']
            simulated_pnl += returns_df[stock] * current_mkt_val
            
    # 找到分位数
    # 如果置信度是95%，我们需要找左尾 5% 的分位数
    quantile = 1 - ContextInfo.confidence_level
    
    # 计算 VaR (取绝对值，表示损失金额)
    # quantile 方法会返回该分位数的数值（通常是负数，代表亏损）
    var_value = abs(np.percentile(simulated_pnl, quantile * 100))
    
    return var_value

def handlebar(ContextInfo):
    """
    K线周期运行函数
    """
    # 仅在最后一根K线（实时行情）运行风控逻辑，避免回测时每根K线都重复计算导致速度过慢
    if not ContextInfo.is_last_bar():
        return

    # 1. 获取账户总资产
    account_info_list = get_trade_detail_data(ContextInfo.accid, ContextInfo.accountType, 'ACCOUNT')
    if not account_info_list:
        print("未获取到账户信息，跳过风控检查")
        return
    
    account_obj = account_info_list[0]
    total_assets = account_obj.m_dBalance # 总资产
    
    if total_assets <= 0:
        return

    # 2. 获取持仓
    pos_df = get_portfolio_positions(ContextInfo)
    if pos_df.empty:
        print("当前无持仓，无需计算VaR")
        return

    # 3. 计算当前组合 VaR
    current_var = calculate_portfolio_var(ContextInfo, pos_df, total_assets)
    
    # 计算 VaR 占总资产的比例
    current_var_ratio = current_var / total_assets
    
    # 4. 风控判断与执行
    limit_var = total_assets * ContextInfo.max_var_ratio
    
    print("=" * 30)
    print(f"【VaR风控检查】 {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"当前总资产: {total_assets:.2f}")
    print(f"当前VaR(95%,1天): {current_var:.2f} (占比: {current_var_ratio*100:.2f}%)")
    print(f"VaR限额: {limit_var:.2f} (占比: {ContextInfo.max_var_ratio*100:.2f}%)")
    
    if current_var > limit_var:
        print("警告：当前风险超出限额！准备执行减仓操作...")
        
        if ContextInfo.run_risk_control:
            # 计算需要降低的风险比例
            # 目标是将 VaR 降到 limit_var 以下
            # 假设 VaR 与持仓市值呈线性关系（简化处理），我们需要减少的持仓比例
            reduce_ratio = 1 - (limit_var / current_var)
            
            # 为了防止频繁微调，可以设置一个缓冲，比如多卖出一点点
            reduce_ratio = reduce_ratio * 1.1 
            
            # 限制最大减仓比例，防止清仓过猛
            reduce_ratio = min(reduce_ratio, 1.0)
            
            print(f"计划减仓比例: {reduce_ratio*100:.2f}%")
            
            for stock in pos_df.index:
                current_vol = pos_df.loc[stock, 'volume']
                current_val = pos_df.loc[stock, 'market_value']
                
                # 计算目标市值
                target_val = current_val * (1 - reduce_ratio)
                
                # 使用 order_target_value 进行减仓
                # 注意：order_target_value 在回测和实盘中均有效
                # 这里的价格设为 -1 表示使用市价/最新价逻辑（取决于交易设置）
                # 也可以使用 passorder 进行更精细的控制
                order_target_value(stock, target_val, ContextInfo, ContextInfo.accid)
                
                print(f"执行减仓: {stock}, 原市值: {current_val:.2f}, 目标市值: {target_val:.2f}")
        else:
            print("风控交易开关未开启，仅提示风险。")
    else:
        print("当前风险在安全范围内。")
    print("=" * 30)

代码详解

1. init 初始化：

   • 设置了 lookback_days = 252，意味着我们使用过去一年的数据来模拟当前组合的表现。
   • 设置了 max_var_ratio = 0.02，这是风控的核心。如果计算出明天有 5% 的概率亏损超过总资产的 2%，则触发风控。

2. get_portfolio_positions 获取持仓：

   • 使用 get_trade_detail_data 获取真实的持仓数据，并整理成 DataFrame 方便计算。

3. calculate_portfolio_var 计算核心：

   • 数据获取：使用 get_market_data_ex 获取持仓股票的历史收盘价。
   • 历史模拟：这是最关键的一步。我们不计算单只股票的 VaR 然后相加（那样会忽略相关性），而是计算当前持仓组合在历史每一天的假设盈亏。
   • 公式：模拟当日盈亏 = Σ(个股当前市值 * 个股历史当日涨跌幅)。
   • 分位数：使用 np.percentile 找到历史盈亏分布中 5% 位置的数值，这就是 VaR。

4. handlebar 执行逻辑：

   • 获取账户总资产。
   • 比较 Current VaR 和 Limit VaR。
   • 减仓逻辑：如果风险超标，计算 reduce_ratio（减仓比例）。假设 VaR 与仓位线性相关，如果 VaR 是限额的 2 倍，我们需要减仓 50% 才能达标。代码中使用了 order_target_value 来快速调整仓位。

如何使用

1. 修改账号：将代码中的 ContextInfo.accid = '6000000000' 修改为您实际的 QMT 资金账号。
2. 运行模式：
   • 此策略依赖 get_trade_detail_data，在实盘或模拟交易模式下效果最佳。
   • 如果在回测模式下，需要确保回测引擎正确模拟了资金和持仓变化。
3. 数据补充：确保在 QMT 客户端的“数据管理”中补充了持仓股票过去一年的日线数据，否则 get_market_data_ex 可能取不到足够的历史数据导致计算失败。

Q&A

Q1: 为什么使用历史模拟法而不是方差-协方差法？
A1: 方差-协方差法假设收益率服从正态分布，但实际股市中经常出现极端行情（肥尾效应）。历史模拟法直接使用过去发生的真实涨跌幅，能更真实地反映组合在极端情况下的潜在风险，且无需计算复杂的协方差矩阵，计算效率在 QMT 中也更高。

Q2: order_target_value 是如何下单的？
A2: order_target_value(stock, value, ...) 会自动计算当前持仓市值与目标市值 value 之间的差额，并根据差额自动发出买入或卖出指令。在风控减仓场景下，目标市值小于当前市值，系统会自动卖出相应数量的股票。

Q3: 这个策略能保证我一定不会亏损超过 2% 吗？
A3: 不能。VaR 只是一个统计估算值，基于历史数据预测未来。如果市场发生前所未有的“黑天鹅”事件（历史数据中未包含的极端波动），实际亏损可能会超过 VaR 估算值。此外，VaR 是基于 95% 置信度的，意味着仍有 5% 的概率亏损会超过这个数值。