big-qmt/py-client/strategy/ipo/boot.py

"""新股自动申购，提供交易日校验、券商对账和本地幂等保护。"""

from __future__ import annotations

import json
import logging
from datetime import datetime, time
from pathlib import Path
from typing import Any

import config
from sdk import Client
from libs.calc import trading_time
from libs.lockfile import is_lock,write_lockfile

IPO_SESSIONS = ((time(9, 30), time(11, 30)), (time(13, 0), time(15, 0)))


def AutoBuyIpo():
    """安全执行一次新股申购，返回成功提交的证券数量。"""
    if not config.account_config.enable_auto_ipo:
        logging.info("[IPO] 自动申购未启用")
        return 0
    if not trading_time(datetime.now()):
        logging.info("[IPO] 非交易时间")
        return 0

    try:
        with Client(
            config.global_config.qmt_base_url,
            config.global_config.qmt_token,
            config.HTTP_TIMEOUT,
        ) as client:
            result = client.ipo_data("STOCK")
            for item in result:
                try:
                    if not isinstance(item, dict):
                        raise TypeError("IPO 数据项必须是字典")

                    stock = str(item.get("stock", "")).strip()
                    if not is_target_stock(stock):
                        continue

                    ipo_price = float(item["issuePrice"])
                    max_purchase_num = int(item["maxPurchaseNum"])
                    if ipo_price <= 0 or max_purchase_num <= 0:
                        raise ValueError("发行价或申购额度必须大于 0")

                    lock_path = Path(config.global_config.qmt_data_dir) / f"{stock}.lock"
                    if is_lock(lock_path):
                        continue

                    client.passorder(
                        op_type=23,
                        stock=stock,
                        volume=max_purchase_num,
                        pr_type=11,
                        price=ipo_price,
                        strategy_name="ipo",
                    )
                    write_lockfile(lock_path)
                    logging.info(
                        "[IPO] %s 申购，发行价：%s 可申购额度：%s",
                        stock,
                        ipo_price,
                        max_purchase_num,
                    )
                except Exception:
                    logging.exception("[IPO] 单条申购处理失败，数据=%r", item)
    except Exception:
        logging.exception("[IPO] 自动申购任务失败")

def is_target_stock(symbol: str) -> bool:
    """
    判断是否为上证、深证、科创板的A股。
    symbol格式示例: '600519.SH', '000001.SZ'
    """
    # 提取纯数字代码
    code = symbol.split(".")[0]
    
    # 判断是否为合规板块
    if code.startswith(('60', '688', '689')):  # 沪市主板 + 科创板
        return True
    if code.startswith(('000', '001', '002', '003', '300', '301')):  # 深市主板 + 创业板
        return True
        
    return False