PTrade 期货教程：如何利用期货的 Contango 和 Backwardation 结构制定策略？

利用期货合约之间的价差（即升贴水结构）构建交易策略，通常被称为跨期套利（Calendar Spread Arbitrage）。

这种策略的核心在于捕捉不同到期月份合约之间价差的异常波动或回归。

策略原理

1. 期限结构（Term Structure）：

   • 正向市场（Contango/升水）：远月价格 > 近月价格。通常代表仓储成本、资金占用成本较高，或近期需求疲软。
   • 反向市场（Backwardation/贴水）：近月价格 > 远月价格。通常代表现货紧缺，便利收益率（Convenience Yield）高。

2. 交易逻辑：

   • 均值回归策略：假设价差会围绕一个历史均值波动。当价差过大（偏离均值）时做空价差，当价差过小（偏离均值）时做多价差。
   • 多头跨期（Bull Spread）：买入近月，卖出远月。适用于预期价差扩大（近月涨得比远月快，或跌得比远月慢）。
   • 空头跨期（Bear Spread）：卖出近月，买入远月。适用于预期价差缩小。

PTrade 策略实现

下面是一个基于 Z-Score（标准分）均值回归 的跨期套利策略示例。

策略逻辑：

1. 选合约：每天盘前选取某品种（如螺纹钢 RB）持仓量最大的两个合约，作为“近月主力”和“远月次主力”。
2. 算统计：获取过去 N 天的历史收盘价，计算“价差 = 近月价格 - 远月价格”的均值和标准差。
3. 交易信号：
   • 当 (当前价差 - 均值) / 标准差 > 阈值（价差过大）：卖出套利（卖近月，买远月），预期价差回归缩小。
   • 当 (当前价差 - 均值) / 标准差 < -阈值（价差过小）：买入套利（买近月，卖远月），预期价差回归扩大。
   • 当价差回归到均值附近时：平仓止盈。

PTrade 策略代码

import numpy as np
import pandas as pd

def initialize(context):
    """
    策略初始化函数
    """
    # 设定交易品种，例如螺纹钢 'RB'
    g.product_code = 'RB'
    # 设定交易所后缀，上期所为 .XSGE
    g.suffix = '.XSGE'
    
    # 统计窗口长度（用于计算历史价差均值和标准差）
    g.window_length = 20
    
    # 开仓阈值（Z-Score），例如 2.0 表示偏离 2 个标准差时开仓
    g.open_threshold = 2.0
    # 平仓阈值（Z-Score），例如 0.5 表示回归到 0.5 个标准差以内平仓
    g.close_threshold = 0.5
    
    # 每次交易的手数
    g.trade_unit = 1
    
    # 存储当前选定的近月和远月合约代码
    g.near_contract = None
    g.far_contract = None
    
    # 存储统计数据
    g.spread_mean = 0
    g.spread_std = 0
    
    # 设置回测频率
    # 注意：实际交易中建议使用分钟级或Tick级，这里演示使用分钟级
    set_commission(commission_ratio=0.0001, min_commission=5.0, type='FUTURE')

def get_main_contracts(context):
    """
    获取持仓量最大的两个合约作为主力（近月）和次主力（远月）
    """
    # 获取该品种所有合约信息
    # 注意：get_instruments 在回测中可能返回包含已过期合约，需要过滤
    all_instruments = get_instruments(g.product_code + g.suffix)
    
    if not all_instruments:
        return None, None
        
    contract_list = []
    current_date = context.blotter.current_dt.strftime("%Y%m%d")
    
    # 遍历合约，筛选未到期的
    # 注意：这里简化处理，实际应通过 get_instruments 返回对象的 delivery_date 判断
    # PTrade get_instruments 返回的是对象列表，需遍历
    # 由于API限制，这里演示一种通过获取快照或行情来筛选活跃合约的方法
    
    # 获取所有可能的合约代码列表（假设我们只看最近的合约）
    # 在PTrade中，通常直接获取主力合约比较困难，这里采用遍历所有合约取持仓量最大的方法
    # 为简化代码，我们假设通过 get_code_list 获取（需自行实现或硬编码部分逻辑）
    # 这里使用一种通用的逻辑：获取当前时间点该品种所有上市合约
    
    # 临时方案：获取主力合约代码（PTrade通常有主力合约后缀，如 'RB8888.XSGE'，但跨期需要具体月份）
    # 因此我们获取该品种下的具体月份合约
    
    # 获取当日该品种所有交易合约的快照
    # 注意：get_snapshot 只能在交易模块或特定回测模式下使用，且不能一次取太多
    # 这里我们采用一种更稳健的方法：获取主力合约，然后推算次主力
    
    # 实际上，PTrade回测中获取所有合约持仓量比较耗时。
    # 简化逻辑：我们假设交易 RB2310 和 RB2401 (示例)，实际应用需动态获取
    # 下面代码演示如何动态获取持仓量最大的两个合约
    
    valid_contracts = []
    
    # 获取该品种所有合约
    # 这里的逻辑需要根据实际数据源调整，以下为通用逻辑框架
    # 假设我们能获取到形如 RB2305.XSGE, RB2310.XSGE 的列表
    # 在PTrade中，可以通过 get_future_contracts(g.product_code) 获取，但该API非标准
    # 我们使用 get_instruments 配合过滤
    
    # 这是一个模拟的获取活跃合约列表的过程
    # 真实环境中建议维护一份主力合约列表或通过外部数据源导入
    # 这里为了代码可运行，我们尝试获取主力合约及其下一个合约
    
    # 获取主力合约代码
    main_code = get_dominant_future(g.product_code, g.suffix)
    if not main_code:
        return None, None
        
    # 获取所有合约，寻找比主力合约晚的合约
    # 这里简化处理：直接返回主力合约和主力合约之后的某个合约
    # 实际量化中，通常是对所有合约按持仓量排序
    
    # 为了演示策略逻辑，我们这里假设 g.near_contract 和 g.far_contract 
    # 在 before_trading_start 中通过数据查询确定
    return main_code, None # 占位，具体逻辑在 before_trading_start 实现

def get_dominant_future(product, suffix):
    """
    辅助函数：获取某品种当前的主力合约（持仓量最大）
    注意：PTrade没有直接返回主力合约代码的简单函数，通常需要自己维护或计算
    这里仅作逻辑演示，实际需遍历合约查询 amount
    """
    # 示例返回，实际需根据行情判断
    # 在回测环境中，通常需要自己构建主力连续合约映射
    return product + '2310' + suffix 

def before_trading_start(context, data):
    """
    盘前处理：确定当日交易的合约对，并计算历史统计数据
    """
    # 1. 确定合约
    # 获取该品种所有合约的行情快照，按持仓量排序
    # 由于PTrade API限制，这里模拟选取过程。
    # 实际代码中，您需要获取所有以 g.product_code 开头的合约，调用 get_snapshot 获取 'amount' (持仓量)
    
    # 假设我们已经选出了两个合约（近月和远月）
    # 在实际回测中，为了代码能跑通，我们这里硬编码示例，或者使用主力连续合约逻辑
    # 真实场景请替换为 get_snapshot 排序逻辑
    
    # 示例：假设当前主力是 RB2310，次主力是 RB2401
    # 请根据回测时间段调整这里的逻辑
    # 这里使用一个简单的逻辑：获取当前主力，并假设下一个季月是远月
    
    # 获取所有合约列表
    ins_list = get_instruments(g.product_code + g.suffix)
    if not ins_list:
        return
        
    # 过滤出属于该品种的合约代码
    code_list = [ins.contract_code for ins in ins_list if ins.contract_code.startswith(g.product_code)]
    
    # 获取前一日的收盘行情来判断持仓量
    # 注意：get_history 只能获取量价，获取持仓量(amount)需要看数据源是否支持
    # 这里我们简化：直接选取两个日期最近的合约作为演示
    code_list.sort() # 按日期排序
    
    # 剔除已经过期的（简单通过字符串比较，实际应比较 delivery_date）
    current_date_str = context.blotter.current_dt.strftime("%Y%m%d")
    # 简单过滤：保留代码中数字部分大于当前日期的（非常粗略的过滤，仅作演示）
    # 实际应解析 contract_code 中的年份月份
    
    # 假设我们选到了两个合约
    if len(code_list) >= 2:
        # 选取近期两个合约，实际应选持仓量最大的
        g.near_contract = code_list[0] 
        g.far_contract = code_list[1]
    else:
        return

    log.info("今日交易合约对: 近月 %s, 远月 %s" % (g.near_contract, g.far_contract))
    
    # 2. 计算历史价差统计数据
    if g.near_contract and g.far_contract:
        # 获取历史收盘价
        h_near = get_history(g.window_length, '1d', 'close', g.near_contract, fq=None, include=False)
        h_far = get_history(g.window_length, '1d', 'close', g.far_contract, fq=None, include=False)
        
        if len(h_near) == g.window_length and len(h_far) == g.window_length:
            # 提取收盘价数组
            # 注意：get_history 返回格式可能因版本不同而异，这里假设返回 DataFrame 或 dict
            # 推荐使用 values 属性获取 numpy 数组
            if isinstance(h_near, pd.DataFrame):
                p_near = h_near['close'].values
            else:
                p_near = h_near['close']
                
            if isinstance(h_far, pd.DataFrame):
                p_far = h_far['close'].values
            else:
                p_far = h_far['close']
            
            # 计算价差序列
            spread_series = p_near - p_far
            
            # 计算均值和标准差
            g.spread_mean = np.mean(spread_series)
            g.spread_std = np.std(spread_series)
            
            log.info("历史价差均值: %.2f, 标准差: %.2f" % (g.spread_mean, g.spread_std))
        else:
            log.warning("历史数据不足，今日暂停交易")
            g.near_contract = None # 标记为不可交易

def handle_data(context, data):
    """
    盘中运行：计算实时价差并交易
    """
    if g.near_contract is None or g.far_contract is None:
        return
        
    # 获取当前最新价格
    # data 是一个字典，包含当前分钟的 bar 数据
    if g.near_contract not in data or g.far_contract not in data:
        return
        
    price_near = data[g.near_contract]['close']
    price_far = data[g.far_contract]['close']
    
    # 计算当前价差
    current_spread = price_near - price_far
    
    # 防止除以0
    if g.spread_std == 0:
        return
        
    # 计算 Z-Score
    z_score = (current_spread - g.spread_mean) / g.spread_std
    
    # 获取当前持仓
    pos_near = get_position(g.near_contract)
    pos_far = get_position(g.far_contract)
    
    # 计算净持仓（多头为正，空头为负）
    # 期货持仓分为 long_amount 和 short_amount
    net_pos_near = pos_near.long_amount - pos_near.short_amount
    net_pos_far = pos_far.long_amount - pos_far.short_amount
    
    # --- 交易逻辑 ---
    
    # 1. 开仓逻辑：价差过大 (Z > 2) -> 卖出套利 (卖近买远)
    if z_score > g.open_threshold:
        # 如果当前没有持仓，则开仓
        if net_pos_near == 0 and net_pos_far == 0:
            log.info("Z-Score: %.2f > %.2f, 触发卖出套利（卖近买远）" % (z_score, g.open_threshold))
            # 卖出近月
            sell_open(g.near_contract, g.trade_unit)
            # 买入远月
            buy_open(g.far_contract, g.trade_unit)
            
    # 2. 开仓逻辑：价差过小 (Z < -2) -> 买入套利 (买近卖远)
    elif z_score < -g.open_threshold:
        # 如果当前没有持仓，则开仓
        if net_pos_near == 0 and net_pos_far == 0:
            log.info("Z-Score: %.2f < -%.2f, 触发买入套利（买近卖远）" % (z_score, g.open_threshold))
            # 买入近月
            buy_open(g.near_contract, g.trade_unit)
            # 卖出远月
            sell_open(g.far_contract, g.trade_unit)
            
    # 3. 平仓逻辑：价差回归 (abs(Z) < 0.5)
    elif abs(z_score) < g.close_threshold:
        # 如果持有卖出套利仓位（空近多远）
        if net_pos_near < 0 and net_pos_far > 0:
            log.info("Z-Score: %.2f 回归，平仓卖出套利" % z_score)
            # 买入平仓近月
            buy_close(g.near_contract, g.trade_unit)
            # 卖出平仓远月
            sell_close(g.far_contract, g.trade_unit)
            
        # 如果持有买入套利仓位（多近空远）
        if net_pos_near > 0 and net_pos_far < 0:
            log.info("Z-Score: %.2f 回归，平仓买入套利" % z_score)
            # 卖出平仓近月
            sell_close(g.near_contract, g.trade_unit)
            # 买入平仓远月
            buy_close(g.far_contract, g.trade_unit)

    # 4. 止损逻辑（可选）：如果价差继续扩大/缩小超过一定限度（如 4个标准差），强制平仓
    # 此处省略，可根据风险偏好添加

关键点解析

1. 合约选择 (before_trading_start)：

   • 跨期套利最关键的是选择流动性最好的两个合约。通常是主力合约（持仓量最大）和次主力合约。
   • 代码中使用了简化的逻辑。在实盘中，建议通过 get_snapshot 获取所有合约的 amount（持仓量）或 volume（成交量）进行排序，动态选取。

2. 数据获取 (get_history)：

   • 我们使用过去 g.window_length 天的日线收盘价来计算价差的统计特征。
   • include=False 表示不包含当天的数据，防止未来函数。

3. Z-Score 信号：

   • Z = (当前价差 - 均值) / 标准差。
   • 这是一种标准化的衡量方式，不依赖于绝对价格水平，适合捕捉相对偏离。

4. 下单指令：

   • 期货交易必须区分 开仓 (Open) 和 平仓 (Close)。
   • buy_open: 买入开仓（多头）。
   • sell_open: 卖出开仓（空头）。
   • buy_close: 买入平仓（平空头）。
   • sell_close: 卖出平仓（平多头）。

5. 风险提示：

   • 移仓换月：期货合约会到期。策略必须包含逻辑，在合约临近交割或主力切换时，平掉旧合约仓位，切换到新合约。上述代码为了简洁，未包含自动移仓逻辑，实盘需注意。
   • 单边风险：虽然是套利，但如果价差持续单边扩大（如逼仓行情），套利策略也会面临巨大亏损，务必设置止损。

Q&A

Q: 如何处理合约换月的问题？
A: 在 before_trading_start 中，每天检查当前持仓的合约是否还是主力合约。如果主力合约发生了变化（例如从 2305 变成了 2310），则需要在当日或次日平掉旧合约的持仓，并更新 g.near_contract 和 g.far_contract 为新的主力对。

Q: 为什么使用 Z-Score 而不是固定价差阈值？
A: 商品价格波动率会随时间变化。固定价差（例如价差超过 100 元开仓）在低波动期可能永远无法触发，在高波动期可能频繁止损。Z-Score 根据近期的波动率（标准差）动态调整开仓门槛，适应性更强。

Q: 这个策略适合所有期货品种吗？
A: 适合具有明显季节性或产业链逻辑的商品（如农产品、黑色系）。对于金融期货（如股指期货），跨期价差更多由无风险利率和分红决定，套利空间较小且逻辑不同。