ptrade 平台如何利用 get_cb_info 过滤临期高风险可转债？（附 Python 源码）

在 ptrade 量化交易平台中，进行可转债交易时，临近到期的可转债通常伴随着较高的强赎风险或价格波动风险。我们可以通过 get_cb_info 接口获取可转债的基础信息，并利用其中的 maturity_date（到期日）字段进行风控过滤。

一、接口介绍

get_cb_info() 接口在研究和交易模块可用，返回一个包含所有可转债基础信息的 DataFrame。其中包含的关键字段有：

• bond_code: 可转债代码
• bond_name: 可转债名称
• maturity_date: 到期日（格式通常为 YYYY-MM-DD）

注：此 API 依赖券商的可转债基础数据权限，使用前请确保已开通。

二、风控过滤策略实现思路

1. 在每日盘前准备阶段（before_trading_start）调用 get_cb_info() 获取全市场可转债数据。
2. 获取当前交易日日期。
3. 遍历可转债列表，计算到期日（maturity_date）与当前日期的差值。
4. 剔除距离到期不足指定天数（例如 90 天）的临期高风险转债，构建安全的转债池。

三、Python 策略源码示例

以下是一个在 ptrade 中实现可转债到期日过滤的完整策略框架：

import datetime
import pandas as pd

def initialize(context):
    # 初始化策略，设置默认股票池
    g.security = '600570.SS'
    set_universe(g.security)
    g.safe_bond_pool = [] # 用于存放过滤后的安全可转债池

def before_trading_start(context, data):
    # 每日盘前执行风控过滤
    log.info("开始进行可转债到期日风控过滤...")
    
    # 1. 获取可转债基础信息
    cb_df = get_cb_info()
    if cb_df.empty:
        log.warning("未能获取到可转债基础数据，请检查权限！")
        return
    
    # 2. 获取当前交易日期
    today = context.blotter.current_dt.date()
    
    # 定义风控阈值：过滤掉距离到期不足 90 天的转债
    risk_threshold_days = 90
    
    safe_bonds = []
    
    # 3. 遍历并计算剩余期限
    for index, row in cb_df.iterrows():
        bond_code = row['bond_code']
        bond_name = row['bond_name']
        maturity_str = row['maturity_date']
        
        if not maturity_str:
            continue
            
        try:
            # 将到期日字符串转换为 date 对象
            maturity_date = datetime.datetime.strptime(maturity_str, '%Y-%m-%d').date()
            # 计算剩余天数
            days_left = (maturity_date - today).days
            
            # 4. 过滤临期债券
            if days_left > risk_threshold_days:
                safe_bonds.append(bond_code)
            else:
                log.info(f"【风控触发】剔除临期转债: {bond_code}({bond_name}), 距离到期仅剩 {days_left} 天")
        except Exception as e:
            log.error(f"解析转债 {bond_code} 到期日异常: {e}")
            continue
            
    g.safe_bond_pool = safe_bonds
    log.info(f"风控过滤完成。全市场转债总数: {len(cb_df)}，安全转债池数量: {len(g.safe_bond_pool)}")

def handle_data(context, data):
    # 在盘中，仅在安全的转债池 g.safe_bond_pool 中选择标的进行交易
    if len(g.safe_bond_pool) > 0:
        # 示例：获取安全池中第一个转债并下单（实际策略中需加入选债逻辑）
        target_bond = g.safe_bond_pool[0]
        # order(target_bond, 10) # 买入10张
        pass

四、注意事项

1. 异常处理：由于网络或数据源原因，get_cb_info() 可能会返回空数据，代码中必须加入 cb_df.empty 的保护性判断。
2. 日期格式兼容：不同券商柜台返回的 maturity_date 格式可能略有差异（如 YYYYMMDD 或 YYYY-MM-DD），请根据实际返回格式调整 strptime 中的解析模板。