ptrade 可转债风控：利用到期期限过滤临期高风险债

在可转债量化交易中，临近到期的可转债（通常剩余期限不足 1 年）往往面临着强赎、到期兑付压力或正股波动带来的极高不确定性。为了降低策略的信用风险与波动风险，在盘前对可转债的剩余期限进行筛选是一项非常实用的风控手段。

本文将介绍如何在 ptrade 中使用 get_cb_info 接口获取可转债的到期日（maturity_date），并编写 Python 代码过滤掉剩余期限不足 1 年的转债。

一、核心接口介绍：get_cb_info

在 ptrade 中，我们可以通过 get_cb_info() 接口获取全市场可转债的基础信息。该接口返回一个 pandas.DataFrame，其中包含了关键的到期日字段：

• 接口名称：get_cb_info()
• 支持环境：研究、交易模块（需向券商确认开通可转债基础数据权限）
• 关键返回字段：
  • bond_code: 可转债代码
  • bond_name: 可转债名称
  • maturity_date: 到期日（格式通常为 YYYY-MM-DD 的字符串）

二、风控过滤算法实现步骤

1. 调用 get_cb_info() 获取所有可转债的基础信息 DataFrame。
2. 获取当前交易日时间（回测中使用 context.blotter.current_dt，实盘中可使用 datetime.datetime.now()）。
3. 将 maturity_date 转换为 datetime 格式。
4. 计算 maturity_date 与当前日期的差值，过滤掉差值小于 365 天（1 年）的标的。

三、ptrade 策略源码示例

以下是一个完整的 ptrade 策略框架，展示了如何在盘前（before_trading_start）过滤临期转债，并动态更新股票池：

import datetime
import pandas as pd

def initialize(context):
    # 初始化策略，设置初始订阅标的（可先设为恒生电子作为占位）
    g.security = '600570.SS'
    set_universe(g.security)

def before_trading_start(context, data):
    log.info("------ 盘前风控：开始过滤临期可转债 ------")
    
    # 1. 获取全市场可转债基础信息
    cb_df = get_cb_info()
    if cb_df.empty:
        log.warning("未能获取到可转债基础信息，请检查权限！")
        g.target_bonds = []
        return
    
    # 2. 获取当前交易日期
    current_date = context.blotter.current_dt
    log.info(f"当前交易日: {current_date.strftime('%Y-%m-%d')}")
    
    # 3. 转换到期日为 datetime 格式，并计算剩余天数
    # 过滤掉 maturity_date 为空或格式不正确的行
    cb_df = cb_df.dropna(subset=['maturity_date'])
    cb_df['maturity_datetime'] = pd.to_datetime(cb_df['maturity_date'])
    
    # 计算剩余期限（天数）
    cb_df['days_left'] = (cb_df['maturity_datetime'] - current_date).dt.days
    
    # 4. 过滤掉剩余期限不足 365 天（1年）的转债
    # 同时可以过滤掉已经到期的转债（days_left <= 0）
    safe_bonds_df = cb_df[(cb_df['days_left'] > 365)]
    
    # 5. 提取过滤后的可转债代码列表
    g.target_bonds = safe_bonds_df['bond_code'].tolist()
    
    log.info(f"全市场转债总数: {len(cb_df)} 只")
    log.info(f"过滤后（剩余期限 > 1年）转债数量: {len(g.target_bonds)} 只")
    
    # 打印部分过滤掉的临期转债作为日志参考
    expired_bonds = cb_df[(cb_df['days_left'] <= 365) & (cb_df['days_left'] > 0)]
    if not expired_bonds.empty:
        log.info("今日过滤掉的临期转债示例（剩余不足1年）：")
        log.info(expired_bonds[['bond_code', 'bond_name', 'maturity_date', 'days_left']].head(5))

def handle_data(context, data):
    # 在盘中，仅对过滤后安全的转债池 g.target_bonds 进行交易逻辑编写
    if len(g.target_bonds) > 0:
        # 示例：仅打印前3只安全转债的最新价格
        for bond in g.target_bonds[:3]:
            if bond in data:
                log.info(f"安全转债 {bond} 当前收盘价: {data[bond]['close']}")

四、注意事项

1. 数据权限：get_cb_info() 依赖券商端的可转债基础数据源。如果在回测或实盘中返回空 DataFrame，请联系券商客服确认是否开通了该 API 的数据权限。
2. 日期格式兼容性：从 get_cb_info() 获取的 maturity_date 格式一般为 YYYY-MM-DD。代码中使用了 pd.to_datetime() 进行安全转换，能够自动识别标准日期格式。
3. 闰年处理：使用 365 天作为 1 年的近似过滤。如果需要极度精确的 1 年跨度，可以使用 dateutil.relativedelta 或判断年份差值。