QMT 策略实战：如何运用凯利公式进行动态仓位管理？

这是一个基于 QMT (Quant Trader) 平台的量化策略，使用了 凯利公式 (Kelly Criterion) 进行动态仓位管理。

策略设计思路

1. 基础择时策略：为了演示仓位管理，我们需要一个基础的开平仓信号。这里使用经典的 双均线策略 (Dual Moving Average)：
   • 买入信号：短期均线（MA5）上穿长期均线（MA20），即金叉。
   • 卖出信号：短期均线（MA5）下穿长期均线（MA20），即死叉。
2. 凯利公式仓位管理：
   • 公式：$f^* = \frac{p(b+1) - 1}{b}$
   • 其中：
     • $f^*$：最优仓位比例。
     • $p$：胜率（获利交易次数 / 总交易次数）。
     • $b$：赔率（平均盈利金额 / 平均亏损金额）。
   • 应用逻辑：
     • 在每次产生买入信号时，根据历史交易记录计算当前的胜率 $p$ 和赔率 $b$。
     • 代入公式计算出 $f^*$。
     • 为了防止全仓凯利带来的剧烈波动，通常使用 半凯利 (Half-Kelly) 或设置最大仓位上限（如 99%）。
     • 如果历史交易数据不足（冷启动阶段），则使用默认的固定仓位（如 30%）。

QMT 策略代码

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

'''
策略名称：凯利公式仓位管理策略 (Kelly Criterion Position Sizing)
策略逻辑：
1. 使用双均线 (MA5, MA20) 作为买卖信号。
2. 记录每一笔平仓交易的盈亏情况。
3. 开仓时，根据历史胜率和盈亏比计算凯利公式建议的仓位比例。
4. 包含防风险机制：限制最大仓位，处理数据不足的情况。
'''

def init(ContextInfo):
    # ================= 策略参数设置 =================
    ContextInfo.stock = '600000.SH'  # 标的股票：浦发银行
    ContextInfo.short_period = 5     # 短期均线周期
    ContextInfo.long_period = 20     # 长期均线周期
    ContextInfo.account_id = '600000248' # 请修改为您的资金账号
    ContextInfo.account_type = 'STOCK'   # 账号类型：STOCK, FUTURE
    
    # 凯利公式相关参数
    ContextInfo.max_leverage = 0.99      # 最大仓位限制 (防止满仓无法扣费)
    ContextInfo.kelly_fraction = 0.5     # 凯利乘数 (0.5代表半凯利，1.0代表全凯利，建议0.5降低风险)
    ContextInfo.default_pos = 0.3        # 初始/数据不足时的默认仓位
    ContextInfo.min_trades_required = 5  # 启用凯利公式所需的最小历史交易次数
    
    # ================= 内部变量初始化 =================
    ContextInfo.trade_records = []       # 记录历史盈亏比例 [0.05, -0.02, ...]
    ContextInfo.entry_price = 0.0        # 记录开仓价格
    ContextInfo.holding = False          # 持仓状态标记
    
    # 设置股票池
    ContextInfo.set_universe([ContextInfo.stock])
    
    print("策略初始化完成，标的：{}, 账号：{}".format(ContextInfo.stock, ContextInfo.account_id))

def handlebar(ContextInfo):
    # 获取当前K线位置
    index = ContextInfo.barpos
    # 获取当前时间
    realtime = ContextInfo.get_bar_timetag(index)
    
    # 获取标的
    stock = ContextInfo.stock
    
    # ================= 数据获取 =================
    # 获取足够的历史数据计算均线
    data_len = ContextInfo.long_period + 2
    # 使用 get_market_data_ex 获取数据 (推荐)
    klines = ContextInfo.get_market_data_ex(
        ['close'], 
        [stock], 
        period=ContextInfo.period, 
        count=data_len, 
        dividend_type='front'
    )
    
    if stock not in klines or len(klines[stock]) < ContextInfo.long_period:
        return # 数据不足，跳过

    close_prices = klines[stock]['close']
    current_price = close_prices.iloc[-1]
    
    # 计算均线
    ma_short = close_prices.rolling(ContextInfo.short_period).mean()
    ma_long = close_prices.rolling(ContextInfo.long_period).mean()
    
    # 获取当前和上一根K线的均线值
    curr_ma_s = ma_short.iloc[-1]
    prev_ma_s = ma_short.iloc[-2]
    curr_ma_l = ma_long.iloc[-1]
    prev_ma_l = ma_long.iloc[-2]
    
    # ================= 信号判断 =================
    # 金叉：短线上穿长线
    golden_cross = (prev_ma_s <= prev_ma_l) and (curr_ma_s > curr_ma_l)
    # 死叉：短线下穿长线
    death_cross = (prev_ma_s >= prev_ma_l) and (curr_ma_s < curr_ma_l)
    
    # ================= 交易逻辑 =================
    
    # 1. 卖出逻辑 (死叉且持仓)
    if death_cross and ContextInfo.holding:
        # 计算本次交易盈亏比例
        if ContextInfo.entry_price > 0:
            pnl_ratio = (current_price - ContextInfo.entry_price) / ContextInfo.entry_price
            ContextInfo.trade_records.append(pnl_ratio)
            print("【平仓记录】价格: {:.2f}, 成本: {:.2f}, 盈亏比率: {:.2%}".format(
                current_price, ContextInfo.entry_price, pnl_ratio))
        
        # 执行清仓
        order_target_value(stock, 0, ContextInfo, ContextInfo.account_id)
        ContextInfo.holding = False
        ContextInfo.entry_price = 0.0
        print("【卖出信号】死叉触发，全额平仓")

    # 2. 买入逻辑 (金叉且空仓)
    elif golden_cross and not ContextInfo.holding:
        # --- 凯利公式计算仓位 ---
        target_percent = calculate_kelly_position(ContextInfo)
        
        # 获取账户总资产
        # 注意：回测模式下使用 ContextInfo.capital，实盘需用 get_trade_detail_data 获取
        total_asset = get_account_total_asset(ContextInfo)
        
        # 计算目标持仓市值
        target_value = total_asset * target_percent
        
        if target_value > 0:
            # 执行下单
            order_target_value(stock, target_value, ContextInfo, ContextInfo.account_id)
            ContextInfo.holding = True
            ContextInfo.entry_price = current_price
            print("【买入信号】金叉触发，凯利建议仓位: {:.2%}，目标市值: {:.2f}".format(
                target_percent, target_value))
        else:
            print("【买入信号】金叉触发，但凯利公式建议仓位为0 (可能历史表现不佳)，跳过交易")

def calculate_kelly_position(ContextInfo):
    """
    计算凯利公式建议的仓位比例
    f = (p * (b + 1) - 1) / b
    p: 胜率
    b: 赔率 (平均盈利 / 平均亏损)
    """
    records = ContextInfo.trade_records
    
    # 如果交易记录不足，返回默认仓位
    if len(records) < ContextInfo.min_trades_required:
        print("历史交易次数不足 ({}/{})，使用默认仓位 {:.2%}".format(
            len(records), ContextInfo.min_trades_required, ContextInfo.default_pos))
        return ContextInfo.default_pos
    
    # 分离盈利和亏损交易
    wins = [r for r in records if r > 0]
    losses = [r for r in records if r <= 0]
    
    # 计算胜率 p
    p = len(wins) / len(records)
    
    # 计算赔率 b
    avg_win = np.mean(wins) if len(wins) > 0 else 0
    avg_loss = abs(np.mean(losses)) if len(losses) > 0 else 0
    
    # 避免除以零
    if avg_loss == 0:
        b = float('inf') # 如果没有亏损过，理论上可以满仓，但我们做限制
    else:
        b = avg_win / avg_loss
        
    # 凯利公式核心计算
    if b == 0:
        f = 0
    elif b == float('inf'):
        f = ContextInfo.max_leverage
    else:
        f = (p * (b + 1) - 1) / b
    
    # 应用凯利乘数 (如半凯利)
    f = f * ContextInfo.kelly_fraction
    
    # 边界限制
    if f < 0:
        f = 0 # 期望收益为负，不交易
    elif f > ContextInfo.max_leverage:
        f = ContextInfo.max_leverage # 不超过最大杠杆限制
        
    print("【凯利计算】胜率(p): {:.2%}, 赔率(b): {:.2f}, 原始凯利: {:.2%}, 调整后仓位: {:.2%}".format(
        p, b, f / ContextInfo.kelly_fraction if ContextInfo.kelly_fraction!=0 else 0, f))
        
    return f

def get_account_total_asset(ContextInfo):
    """
    获取账户总资产，兼容回测和实盘
    """
    if ContextInfo.do_back_test:
        # 回测模式下，capital 可能只是初始资金，这里简化处理，
        # 实际回测中建议通过 get_trade_detail_data 获取动态权益
        # 这里为了演示方便，尝试获取动态权益，如果失败则用 ContextInfo.capital
        try:
            acct_info = get_trade_detail_data(ContextInfo.account_id, ContextInfo.account_type, 'ACCOUNT')
            if acct_info:
                return acct_info[0].m_dBalance
        except:
            pass
        return ContextInfo.capital
    else:
        # 实盘模式
        acct_info = get_trade_detail_data(ContextInfo.account_id, ContextInfo.account_type, 'ACCOUNT')
        if len(acct_info) > 0:
            return acct_info[0].m_dBalance
        else:
            return 0

代码关键点解析

1. 数据记录 (ContextInfo.trade_records)：

   • 凯利公式的核心依赖于历史表现。我们在 init 中初始化了一个列表。
   • 在 handlebar 的卖出逻辑中，计算本次交易的收益率 pnl_ratio 并存入列表。这是计算胜率和赔率的基础。

2. 凯利公式计算函数 (calculate_kelly_position)：

   • 冷启动处理：如果交易次数少于 min_trades_required（例如5次），统计学意义不大，此时强制使用 default_pos（默认仓位），避免初期数据偏差导致不交易或重仓。
   • 胜率 ($p$)：盈利次数 / 总次数。
   • 赔率 ($b$)：平均盈利金额 / 平均亏损金额（取绝对值）。
   • 公式实现：f = (p * (b + 1) - 1) / b。
   • 风险控制：
     • 半凯利：通过 kelly_fraction = 0.5 降低风险。全凯利虽然理论增长最快，但波动极大，容易导致爆仓。
     • 负值处理：如果计算结果小于0（意味着策略期望收益为负），则仓位设为0，停止开仓。
     • 最大值限制：限制最大仓位为 99% (max_leverage)，预留少量资金防止手续费不足。

3. 下单执行：

   • 使用 order_target_value。这是最适合仓位管理策略的函数，因为它会自动计算需要买入多少金额才能达到目标持仓市值。
   • 目标市值 = 总资产 * 凯利比例。

Q&A

Q: 为什么我的凯利公式计算出的仓位是 0？
A: 这通常意味着根据历史记录，您的策略期望收益为负（即 $p \times b < (1-p)$）。凯利公式会建议您不要下注。此时您应该优化基础的择时策略（提高胜率或盈亏比），而不是强行开仓。

Q: 如何在回测中看到效果？
A: 在 QMT 的回测界面，设置较长的时间跨度。策略初期会使用固定仓位，随着交易次数增加，仓位会根据表现动态调整。如果策略表现好（高胜率高盈亏比），仓位会逐渐增加；反之则减少。

Q: 实盘中需要注意什么？
A:

1. 数据持久化：上述代码中的 trade_records 存在内存中，如果重启 QMT 客户端，历史记录会丢失，导致重新进入冷启动阶段。实盘建议将交易记录保存到本地文件（如 CSV 或 JSON）或数据库中，在 init 时读取。
2. 账号同步：请确保 ContextInfo.account_id 填写正确，且已登录交易账号。