🚀 AI 一键生成 ptrade 策略代码

PTrade 风险管理：计算投资组合 Beta 并动态对冲

ptrade | 发布时间: 2025/12/22
以下内容由 EasyQuant 生成。

问题描述

如何计算组合Beta并进行对冲？

解决方案

在量化交易中，利用Beta（$\beta$）进行对冲是一种常见的市场中性策略。其核心思想是通过做空股指期货，抵消股票组合的市场风险（系统性风险），从而获取Alpha（超额收益）。

以下是关于如何在PTrade中计算组合Beta并进行对冲的详细说明及策略代码。

1. 核心逻辑与公式

第一步：计算单只股票的Beta

Beta系数反映了个股相对于大盘（基准）的波动敏感度。计算公式为：
$$ \beta_i = \frac{Cov(r_i, r_m)}{Var(r_m)} $$
其中：

  • $r_i$：股票 $i$ 的收益率序列
  • $r_m$：基准指数（如沪深300）的收益率序列
  • $Cov$：协方差
  • $Var$：方差

第二步：计算投资组合的Beta

组合的Beta是持仓股票Beta的加权平均：
$$ \beta_p = \sum_{i=1}^{n} (w_i \times \beta_i) $$
其中 $w_i$ 是股票 $i$ 市值占组合总市值的权重。

或者更直接地，我们可以计算 组合的总风险敞口（Beta Value）：
$$ \text{Total Beta Value} = \sum (\text{市值}_i \times \beta_i) $$

第三步：计算所需期货合约数量

为了完全对冲市场风险，我们需要做空的期货合约价值应等于组合的总风险敞口。
$$ N = \frac{\text{Total Beta Value}}{\text{期货价格} \times \text{合约乘数}} $$

2. PTrade 策略实现

以下是一个完整的PTrade策略代码。该策略买入一篮子股票，并根据历史数据计算组合Beta，使用沪深300股指期货（IF）进行动态对冲。

注意：

  1. 代码中使用了 numpy 和 pandas 进行数学计算。
  2. 为了演示，我们选取了几只股票作为多头组合。
  3. 期货合约代码（如 IF2309.CCFX）在实盘中需要根据实际日期更换为主力合约，这里作为示例写死或需动态获取。
  4. 回测模式不支持 get_snapshot，因此代码中使用 get_price 获取最新价格。
import numpy as np
import pandas as pd
import math

def initialize(context):
    """
    初始化函数
    """
    # 1. 设定基准为沪深300
    set_benchmark('000300.SS')
    
    # 2. 设定要操作的股票池 (示例：几只大盘股)
    g.stocks = ['600000.SS', '600036.SS', '600519.SS', '000858.SZ']
    set_universe(g.stocks)
    
    # 3. 设定对冲使用的股指期货合约
    # 注意：实际交易中应动态获取主力合约，这里以 IF2309.CCFX 为例
    # IF 合约乘数为 300
    g.future_code = 'IF2309.CCFX' 
    g.contract_multiplier = 300
    
    # 4. 设定计算Beta的历史窗口长度 (例如过去60个交易日)
    g.beta_window = 60
    
    # 5. 设定调仓和对冲周期 (每天开盘后运行)
    run_daily(context, trade_and_hedge, time='09:35')

def before_trading_start(context, data):
    """
    盘前处理
    """
    pass

def calculate_beta(stock_code, benchmark_code, window):
    """
    计算单只股票相对于基准的Beta值
    """
    # 获取历史收盘价数据 (多取1天以计算收益率)
    # 注意：get_history 返回的数据包含股票和基准
    # 这里分别获取
    
    # 获取股票历史数据
    stock_hist = get_history(window + 1, '1d', 'close', stock_code, fq='pre', include=False)
    if stock_hist is None or len(stock_hist) < window:
        return 1.0 # 数据不足默认为1
        
    # 获取基准历史数据
    bench_hist = get_history(window + 1, '1d', 'close', benchmark_code, fq='pre', include=False)
    if bench_hist is None or len(bench_hist) < window:
        return 1.0

    # 提取收盘价序列 (处理不同Python版本返回格式差异，这里假设返回DataFrame或Series)
    # PTrade get_history 单只股票返回 DataFrame，列名为 'close'
    try:
        s_close = stock_hist['close']
        b_close = bench_hist['close']
    except:
        # 兼容性处理，如果返回格式不同
        return 1.0

    # 计算日收益率
    s_ret = s_close.pct_change().dropna()
    b_ret = b_close.pct_change().dropna()
    
    # 确保数据长度一致
    min_len = min(len(s_ret), len(b_ret))
    s_ret = s_ret[-min_len:]
    b_ret = b_ret[-min_len:]
    
    # 计算协方差矩阵
    # np.cov 返回 [[var(a), cov(a,b)], [cov(b,a), var(b)]]
    cov_mat = np.cov(s_ret, b_ret)
    
    if len(cov_mat) < 2:
        return 1.0
        
    beta = cov_mat[0, 1] / cov_mat[1, 1]
    return beta

def trade_and_hedge(context):
    """
    交易股票并调整对冲仓位
    """
    # ---------------------------------------------------
    # 1. 股票端操作 (简单的等权重买入持有)
    # ---------------------------------------------------
    target_value_per_stock = context.portfolio.portfolio_value * 0.8 / len(g.stocks)
    
    for stock in g.stocks:
        order_target_value(stock, target_value_per_stock)
    
    # ---------------------------------------------------
    # 2. 计算组合的总Beta风险敞口
    # ---------------------------------------------------
    total_beta_value = 0.0
    positions = context.portfolio.positions
    
    # 遍历当前持仓的所有股票
    for sid, pos in positions.items():
        # 排除期货持仓，只计算股票
        if pos.business_type == 'stock' and pos.amount > 0:
            # 计算该股票的Beta
            beta = calculate_beta(sid, '000300.SS', g.beta_window)
            
            # 获取股票当前市值
            market_value = pos.last_sale_price * pos.amount
            
            # 累加风险敞口 (市值 * Beta)
            total_beta_value += market_value * beta
            
            # log.info("股票: {}, 市值: {:.2f}, Beta: {:.4f}".format(sid, market_value, beta))

    log.info("当前股票组合总 Beta Value (风险敞口): {:.2f}".format(total_beta_value))

    # ---------------------------------------------------
    # 3. 计算所需期货合约数量
    # ---------------------------------------------------
    # 获取期货当前价格
    # 回测模式下使用 get_price 获取上一分钟收盘价作为参考
    future_data = get_price(g.future_code, count=1, frequency='1m', fields=['close'])
    
    if future_data is None or len(future_data) == 0:
        log.warning("无法获取期货价格，跳过对冲")
        return

    # 处理 get_price 返回格式，获取具体价格数值
    if isinstance(future_data, dict):
         # 字典格式
         future_price = future_data[g.future_code]['close'][0]
    else:
         # DataFrame格式
         future_price = future_data['close'].iloc[-1]

    # 计算每张合约的价值
    contract_value = future_price * g.contract_multiplier
    
    # 计算理论所需做空的合约数量 (四舍五入取整)
    if contract_value > 0:
        target_short_count = round(total_beta_value / contract_value)
    else:
        target_short_count = 0
        
    log.info("期货价格: {:.2f}, 理论需做空手数: {}".format(future_price, target_short_count))

    # ---------------------------------------------------
    # 4. 执行期货交易进行对冲
    # ---------------------------------------------------
    # 获取当前期货持仓
    future_pos = get_position(g.future_code)
    current_short_amount = future_pos.today_short_amount + (future_pos.short_amount - future_pos.today_short_amount) 
    # 注意：PTrade中 short_amount 通常包含今仓和昨仓，具体视接口返回而定，这里简化处理取总空头持仓
    # 更严谨的写法是直接读取 pos.short_amount
    current_short_amount = future_pos.short_amount

    diff = target_short_count - current_short_amount
    
    if diff > 0:
        # 需要增加空头 (卖出开仓)
        log.info("执行对冲：卖出开仓 {} 手 {}".format(diff, g.future_code))
        sell_open(g.future_code, int(diff))
    elif diff < 0:
        # 需要减少空头 (买入平仓)
        # 平仓数量为正数
        close_amount = abs(diff)
        log.info("调整对冲：买入平仓 {} 手 {}".format(close_amount, g.future_code))
        # 优先平今或平昨取决于交易所规则，这里简化直接平仓
        buy_close(g.future_code, int(close_amount))
    else:
        log.info("对冲仓位无需调整")

def handle_data(context, data):
    """
    盘中Tick/分钟处理 (本策略主要逻辑在 run_daily 中)
    """
    pass

3. 关键点解析

  1. 数据获取 (get_history):

    • 我们使用 get_history 获取过去 g.beta_window (60天) 的收盘价。
    • 参数 fq='pre' 表示使用前复权数据，这对于计算真实的收益率波动非常重要。

  2. Beta 计算 (np.cov):

    • 使用 pct_change() 计算日收益率。
    • np.cov(s_ret, b_ret) 计算协方差矩阵。Beta 等于 股票与基准的协方差 除以 基准的方差。

  3. 风险敞口汇总:

    • 我们没有计算单一的“组合Beta系数”，而是直接计算了“Beta价值”（$\sum Value_i \times \beta_i$）。这样做的好处是直接对应了需要对冲的金额，计算期货手数时更加直观。

  4. 期货交易 (sell_open / buy_close):

    • 对冲是做空市场，所以增加对冲仓位使用 sell_open (卖出开仓)。
    • 减少对冲仓位（例如股票市值下跌或Beta降低）使用 buy_close (买入平仓)。

  5. 合约乘数:

    • 沪深300 (IF) 和 上证50 (IH) 的乘数是 300。
    • 中证500 (IC) 和 中证1000 (IM) 的乘数是 200。
    • 代码中需根据实际品种设置 g.contract_multiplier。

Q&A

Q: 为什么回测时有时候Beta计算会报错？
A: 通常是因为新股上市时间不足，或者停牌导致数据长度不够。代码中加入了 if len(stock_hist) < window 的判断，如果数据不足默认 Beta 为 1.0，这是一种保护机制。

Q: 如何处理期货合约到期换月？
A: 在实盘中，期货合约会按月交割。需要在策略中加入逻辑：当临近交割日（如每月第三个周五）时，平掉旧合约（buy_close），开仓新月份的合约（sell_open）。可以使用 get_instruments 获取交割日信息。

Q: 这种对冲能完全消除风险吗？
A: 不能。这只能消除系统性风险（大盘涨跌带来的风险）。如果选股能力（Alpha）不强，或者个股发生特异性暴雷（非系统性风险），策略仍可能亏损。此外，Beta是基于历史数据计算的，未来的Beta可能会发生漂移。

相关推荐

PTrade 两融交易：如何进行担保品买卖 (margin_trade)

本教程专为 PTrade 融资融券新手设计，详细讲解 margin_trade 函数的用法。学习如何通过 API 实现担保品的限价和市价买卖，有效管理您的两融账户资产，并附有完整的 Python 代码...

PTrade 期货教程：主力合约换月

期货量化交易必读！本文讲解如何在 PTrade 中监控主力合约切换，编写逻辑实现到期前的自动平旧仓、开新仓（移仓换月）操作。

PTrade 基本面选股：EPS 增长

通过 get_fundamentals 接口获取历史财务数据，筛选出每股收益（EPS）连续 3 年保持增长的优质成长股。

PTrade Level-2 深度：如何解析买一卖一的委托队列？

深入解析 PTrade get_snapshot 接口返回的 Level-2 逐笔委托队列数据，分析主力挂单结构，辅助短线交易决策。

PTrade 行情快照 turnover 字段解析

解析 get_snapshot 接口在主推数据模式（tick）与在线请求模式下，total_bid_turnover（委买金额）等字段的返回值差异。

PTrade 盘前函数执行时间机制

详细对比 before_trading_start 函数在回测模式（8:30）与实盘交易（9:10）中的触发机制及数据获取注意事项。