ptrade 平台如何实现新股一键申购与中签自动查询？（附 Python 源码）

在 A 股量化交易中，新股与可转债申购（俗称“打新”）是获取超额收益的低风险手段。通过 ptrade 平台的自动化运维能力，我们可以实现每日一键申购与中签结果自动查询。本文将为您详细解析相关 API 并提供完整的策略源码。

一、 核心 API 介绍

1. 新股一键申购：ipo_stocks_order

• 接口说明：用于一键申购当日全部新股或可转债。
• 核心参数：
  • market_type：申购代码所属市场。不传时默认申购全部。0 代表上证普通代码，1 代表上证科创板，2 代表深证普通，3 代表深证创业板，4 代表可转债。
  • black_stocks：黑名单列表，加入此列表的股票将不进行申购。

2. 历史中签信息查询：get_lucky_info

• 接口说明：用于获取指定时间范围内的中签信息。
• 核心参数：
  • start_date：开始日期（格式："YYYYmmdd"）。
  • end_date：结束日期（格式："YYYYmmdd"）。
• 返回格式：列表套字典，如 [{'stock_code': '371002.SZ', 'occur_amount': 10.0, 'business_price': 100.0, 'stock_name': '...发债', 'init_date': 20220928}]。

二、 完整策略源码

以下策略配置为：

1. 每天 09:35 自动触发一键申购（避开开盘极度繁忙阶段）。
2. 每天 09:15（盘前阶段）自动查询前一日至今日的中签结果，并打印到日志中。

# -*- coding: utf-8 -*-
import datetime

def initialize(context):
    # 设置一个基础股票池（打新策略通常不需要频繁操作特定股票，但框架要求设置）
    g.security = '600570.SS'
    set_universe(g.security)
    
    # 1. 每天 09:15 盘前自动查询中签结果
    run_daily(context, query_lucky_draw, time='09:15')
    
    # 2. 每天 09:35 自动执行一键申购
    run_daily(context, auto_ipo_order, time='09:35')

def before_trading_start(context, data):
    # 每日初始化标识
    g.ipo_executed = False

def auto_ipo_order(context):
    """
    自动申购当日所有新股与可转债
    """
    if not g.ipo_executed:
        log.info("====== 开始执行每日一键打新 ======")
        try:
            # 一键申购全部新股与可转债（不传 market_type 默认申购全部）
            result = ipo_stocks_order()
            log.info("申购提交结果: %s" % str(result))
            g.ipo_executed = True
        except Exception as e:
            log.error("一键申购过程中出现异常: %s" % str(e))
        log.info("====== 一键打新执行完毕 ======")

def query_lucky_draw(context):
    """
    自动查询中签信息
    """
    log.info("====== 开始查询中签信息 ======")
    try:
        # 获取前一交易日日期
        pre_date_str = str(get_trading_day(-1)).replace("-", "")
        # 获取当前交易日日期
        current_date_str = context.blotter.current_dt.strftime("%Y%m%d")
        
        log.info("查询区间: %s 至 %s" % (pre_date_str, current_date_str))
        
        # 调用中签查询接口
        lucky_info = get_lucky_info(pre_date_str, current_date_str)
        
        if lucky_info:
            log.info("【恭喜中签！】中签详细信息如下:")
            for info in lucky_info:
                log.info("中签日期: %s | 代码: %s | 名称: %s | 数量: %s | 价格: %s" % 
                         (info.get('init_date'), 
                          info.get('stock_code'), 
                          info.get('stock_name'), 
                          info.get('occur_amount'), 
                          info.get('business_price')))
        else:
            log.info("此区间内未查询到中签记录。")
            
    except Exception as e:
        log.error("查询中签信息时出现异常: %s" % str(e))
    log.info("====== 中签信息查询完毕 ======")

def handle_data(context, data):
    pass

三、 自动化运维注意事项

1. 实盘资金账户限制：get_lucky_info 接口在同一分钟内多次调用会返回缓存数据，以减小柜台压力。实盘对接时，请确保券商柜台支持该函数。
2. 服务器重启处理：建议在 initialize 中使用 set_parameters(not_restart_trade="1")，避免交易时间内服务器异常重启导致策略重复初始化或重复触发申购逻辑。
3. 黑名单机制：若有不想申购的个股，可在 ipo_stocks_order 中传入 black_stocks=['代码1', '代码2'] 进行过滤。