fix bug
This commit is contained in:
@@ -2,9 +2,11 @@
|
||||
|
||||
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
|
||||
@@ -14,14 +16,14 @@ from libs.lockfile import is_lock,write_lockfile
|
||||
IPO_SESSIONS = ((time(9, 30), time(11, 30)), (time(13, 0), time(15, 0)))
|
||||
|
||||
|
||||
def AutoBuyIpo() -> None:
|
||||
"""安全执行一次新股申购。"""
|
||||
def AutoBuyIpo():
|
||||
"""安全执行一次新股申购,返回成功提交的证券数量。"""
|
||||
if not config.account_config.enable_auto_ipo:
|
||||
logging.info("[IPO] 自动申购未启用")
|
||||
return
|
||||
return 0
|
||||
if not trading_time(datetime.now()):
|
||||
logging.info("[IPO] 非交易时间")
|
||||
return
|
||||
return 0
|
||||
|
||||
try:
|
||||
with Client(
|
||||
@@ -36,7 +38,7 @@ def AutoBuyIpo() -> None:
|
||||
raise TypeError("IPO 数据项必须是字典")
|
||||
|
||||
stock = str(item.get("stock", "")).strip()
|
||||
if not stock or stock.endswith(".BJ"):
|
||||
if not is_target_stock(stock):
|
||||
continue
|
||||
|
||||
ipo_price = float(item["issuePrice"])
|
||||
@@ -68,3 +70,18 @@ def AutoBuyIpo() -> None:
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user