ptrade 自动打新教程：如何结合 get_lucky_info 实现新股中签自动查询？

ptrade 自动打新与中签查询实现指南

在 A 股市场中，新股申购（打新）和可转债申购是低风险且收益稳健的投资方式。利用 ptrade 平台的量化接口，我们可以实现每日自动一键打新，并结合 get_lucky_info 接口自动查询中签信息。以下是完整的实现教程与策略源码。

一、 核心 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}]。

二、 自动打新与中签查询策略源码

以下是一个完整的 ptrade 策略，包含每日上午 10:00 自动一键打新以及盘前 09:15 自动查询中签信息的逻辑：

import datetime
import time

def initialize(context):
    # 设置基准和股票池（打新策略通常不需要特定股票池，但框架要求初始化）
    g.security = '600570.SS'
    set_universe(g.security)
    
    # 1. 盘前 09:15 自动查询中签信息
    run_daily(context, check_lucky_results, time='09:15')
    
    # 2. 盘中 10:00 自动执行一键打新
    run_daily(context, auto_ipo_subscription, time='10:00')
    
    # 初始化打新标记
    g.ipo_done = False

def before_trading_start(context, data):
    # 每日开盘前重置打新标记
    g.ipo_done = False

def auto_ipo_subscription(context):
    """
    自动打新函数：一键申购当日所有新股与可转债
    """
    if not g.ipo_done:
        log.info("开始执行每日自动打新...")
        
        # 申购新股（可根据需要分市场申购，不传 market_type 则默认申购全部）
        try:
            # 申购全部新股与可转债
            result = ipo_stocks_order()
            log.info(f"打新委托提交完成，返回结果: {result}")
            g.ipo_done = True
        except Exception as e:
            log.error(f"自动打新执行失败，错误信息: {e}")

def check_lucky_results(context):
    """
    中签查询函数：查询上一交易日至今日的中签结果
    """
    log.info("开始查询中签信息...")
    try:
        # 获取上一个交易日日期
        pre_date = str(get_trading_day(-1)).replace("-", "")
        # 获取当前交易日日期
        current_date = context.blotter.current_dt.strftime("%Y%m%d")
        
        # 查询中签信息
        lucky_info = get_lucky_info(pre_date, current_date)
        
        if lucky_info:
            log.info(f"★ 恭喜！发现中签信息 ★ : {lucky_info}")
            for item in lucky_info:
                msg = f"中签提醒 - 证券代码: {item['stock_code']}, 证券名称: {item['stock_name']}, 数量: {item['occur_amount']}, 价格: {item['business_price']}"
                log.info(msg)
                # 如果券商环境支持，可在此处调用发送邮件或企业微信通知接口
        else:
            log.info("未查询到新增中签信息。")
    except Exception as e:
        log.error(f"中签信息查询失败，错误信息: {e}")

def handle_data(context, data):
    pass

三、 注意事项与实盘要点

1. 接口缓存限制：为了减小柜台压力，get_lucky_info 接口在同一分钟内多次调用会返回缓存数据。因此，无需在 handle_data 或 tick_data 中高频调用，推荐使用 run_daily 每天定时调用一次。
2. 实盘环境支持：get_lucky_info 接口对接部分柜台（如 jz_ufx）时可能不受支持。在实盘运行前，请先在模拟盘或咨询券商确认该接口是否可用。
3. 资金预留：中签后请确保账户内有足够的可用资金用于缴款，避免因资金不足导致弃购。