88 lines
3.0 KiB
Python
88 lines
3.0 KiB
Python
"""新股自动申购，提供交易日校验、券商对账和本地幂等保护。"""
|
|
|
|
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
|