From 554dd0f4cbfe8f39b00443db6ccc0e968f3e30bc Mon Sep 17 00:00:00 2001 From: yanweidong Date: Sat, 12 Sep 2026 16:25:34 +0800 Subject: [PATCH] fix bug --- docs/ipo-submission-state.md | 19 +++ docs/strategy-audit-2026-09-12.md | 86 ++++++++++++ py-client/libs/lockfile.py | 32 +++++ py-client/libs/order.py | 8 +- py-client/libs/state.py | 55 +++++++- py-client/sdk/trade.py | 6 +- py-client/strategy/ipo/boot.py | 177 ++++++++++++++----------- py-client/strategy/trend/boot.py | 2 +- py-client/strategy/zt/boot.py | 13 +- py-client/strategy/zt/positions.py | 3 +- py-client/strategy/zt/profit.py | 31 +++++ py-client/tests/test_ipo.py | 175 ++++++++++++++++++++++++ py-client/tests/test_zt_audit_fixes.py | 132 ++++++++++++++++++ 13 files changed, 646 insertions(+), 93 deletions(-) create mode 100644 docs/ipo-submission-state.md create mode 100644 docs/strategy-audit-2026-09-12.md create mode 100644 py-client/strategy/zt/profit.py create mode 100644 py-client/tests/test_ipo.py create mode 100644 py-client/tests/test_zt_audit_fixes.py diff --git a/docs/ipo-submission-state.md b/docs/ipo-submission-state.md new file mode 100644 index 0000000..aa57ad7 --- /dev/null +++ b/docs/ipo-submission-state.md @@ -0,0 +1,19 @@ +# IPO 提交与结果核对 + +当前实现使用文件记录，不增加数据库或后台任务。 + +- 路径：`qmt_data_dir/ipo/<账户 SHA-256>//<证券>/<尝试序号>.json`。 +- 提交前用独占创建方式写入并落盘 `pending`；占位失败不提交。 +- HTTP 正常返回仍保留 `pending`，不代表最终申购成功。 +- 下次 IPO 任务先查询账户委托。查不到、状态未知、仍在处理或已有部分成交时，不重发。 +- 同一本地订单编号查到状态 56 后记录 `confirmed`，后续保持防重。 +- 同一编号查到状态 57 且成交量为 0，才记录 `rejected` 并原子占位下一次尝试。旧尝试的废单不能授权新尝试再次重发。 +- 不同账户、日期使用不同记录。每次尝试都有写入记录且传给 QMT 的本地订单编号。 + +状态码依据：[迅投官方委托核对示例](https://dict.thinktrader.net/innerApi/code_examples.html)。撤单、部撤等情况未自动视为可以再次申购。 + +旧版只有证券名的 `.lock` 文件不再作为新账户的申购记录，避免跨账户锁冲突；首次运行通过当天同证券买入委托防重。升级应在当前账户委托可查询的条件下进行，旧锁本身无法证明归属账户或最终结果。 + +损坏或无法读取的记录不会触发重新提交；记录写入失败也不会提交。若长期查不到回报，需要先人工核对柜台结果，不能直接删除待确认记录后重跑。 + +接口响应必须是列表，合法空列表表示无候选；非列表响应记录错误。候选要求完整代码及对应交易所、有限正价格、正整数数量。当前参与板块范围保持不变。 diff --git a/docs/strategy-audit-2026-09-12.md b/docs/strategy-audit-2026-09-12.md new file mode 100644 index 0000000..bd5a75b --- /dev/null +++ b/docs/strategy-audit-2026-09-12.md @@ -0,0 +1,86 @@ +# IPO、Trend、ZT 策略审计 + +复审日期：2026-09-12。基线：`7a7049ce44965d27c42f08ede1818d16495d8232` 加本次读取的工作区代码，包含手工修改及未提交文件。 + +范围：三个策略及直接相关的委托簿、状态存储、SDK、服务端成交映射、网格和快照模块。本次只更新报告，不修改策略、测试或生产数据，不连接交易账户。 + +仅保留复审后仍需处理的问题。已修复及用户已标记忽略的条目、历史复现和过时统计已移除；编号沿用原报告。部分修复的问题仅描述尚未解决的策略范围。 + +## 结论 + +**待处理：P0 0 项、P1 2 项、P2 2 项。** IPO 本次未确认新的待处理问题。 + +| 编号 | 级别 | 范围 | 问题 | +|---|---|---|---| +| P1-07 | P1 | ZT | 按委托编号去重，分笔成交可能漏记 | +| P1-08 | P1 | Trend | 超时撤单仍覆盖手工单及其他非 IPO 策略订单 | +| P2-01 | P2 | Trend | 止盈峰值跨仓位沿用 | +| P2-07 | P2 | 共用模块测试 | 旧 OrderBook 构造调用导致回归测试错误 | + +P0 表示已确认的紧急严重风险；P1 表示主要交易控制或记账问题；P2 表示特定条件下的可靠性和回归验证问题。本次未确认 P0。 + +## P1 + +### P1-07：ZT 按委托编号去重，分笔成交可能漏记 + +**位置：** [state.py](../py-client/libs/state.py) 第 55、146–156、362–389 行；[服务端成交映射](../api/qmt_rest_new.py) 第 371–387 行；[ZT 启动](../py-client/strategy/zt/boot.py) 第 49–51 行。 + +**证据：** 服务端将 `m_strOrderSysID` 映射为 `order_sys_id`，成交表以该字段建立唯一索引。ZT 增量路径插入第一条后将编号加入 `existing`，其余同编号记录不再插入。不同 `ref`、时间和数量不参与身份识别。启动时两次成交查询也按此编号转成字典比较，会折叠同委托的多条记录。 + +**本轮复现：** 临时数据库中，同一委托编号、不同 `ref` 的 40 股和 60 股买入成交，对应账户快照为 100 股。`sync_account()` 仅记入 40 股并隔离该证券；重新加载数据库并再次同步，仍为 40 股，未恢复。 + +**影响与边界：** 逐笔回报可能漏记后续成交；累计回报也无法通过当前只追加逻辑更新已存数量。库存、金额及成本可能不完整，证券持续暂停交易。复现证明代码无法处理上述输入，不代表已确认目标柜台采用哪一种回报契约。 + +**修改建议：** 先核验实际回报是逐笔还是累计，以及唯一身份字段。逐笔回报透传真实成交编号并据此去重；累计回报按增量差额更新。启动稳定性比较也需保留完整回报。不要未经核验拼接时间、价格或 `ref` 作为唯一键；修改存储键时同时明确已有数据的处理方式。 + +**验收：** 同委托分笔成交累计为 100 股；重复查询和重启不重复记账；累计回报按已确认契约处理；初始化比较不因同编号覆盖而遗漏变化。 + +### P1-08：Trend 超时撤单仍覆盖手工单及其他策略订单 + +**位置：** [Trend boot.py](../py-client/strategy/trend/boot.py) 第 40、119 行；[OrderBook.refresh()](../py-client/libs/order.py) 第 61、77–88 行。 + +**证据：** Trend 启动及每轮运行均传入完整账户委托，未指定 `cancel_prefix`，默认值为 `None`。所有满足超时及可撤状态的非 `IPO-*` 委托都会进入撤单路径，包括空备注手工单和 `zt-*` 订单。 + +**本轮复现：** 四笔超时、状态 50 的模拟订单分别使用 `TREN-BUY-*`、`zt-base-*`、`IPO-*` 和空备注。按 Trend 当前调用方式刷新，Trend、ZT 和手工订单触发撤单，IPO 未触发；四笔均保留缓存。 + +**影响与边界：** 运行 Trend 可能干扰人工或其他策略希望继续等待的委托。本项仅保留 Trend 范围；当前 IPO 排除有效，不再沿用旧报告的 IPO 误撤结论。Mock 只能证明撤单调用，不能证明柜台受理。 + +**修改建议：** 若 Trend 只管理自己的订单，应明确归属后限制自动撤单，同时保留全账户在途防重。Trend 开仓使用信号前缀、持仓管理使用 `TREN-*`，不能简单改为单一 `trend-` 前缀。如果全账户非 IPO 撤单是预期行为，应明确记录这一范围后再决定是否关闭本项。 + +**验收：** Trend 自有超时订单能撤，手工单和其他策略订单不被 Trend 自动撤销；全部活动委托继续参与防重。 + +## P2 + +### P2-01：Trend 止盈峰值跨仓位沿用 + +**位置：** [Trend positions.py](../py-client/strategy/trend/positions.py) 第 120–121、198–199 行；[Trend boot.py](../py-client/strategy/trend/boot.py) 第 60、113–119 行；[网格跟踪器](../py-client/libs/grid_take_profit.py) 第 44–74 行。 + +**证据：** Trend 使用普通 `GridTrailingTracker`，键仅包含账户及证券。清仓、重新建仓和补仓成本变化时，没有清理旧峰值；每轮快照更新也未同步网格生命周期。 + +**本轮复现：** 使用真实 `manage_positions()`、真实 tracker 和模拟委托簿，依次输入成本 10 元、价格 12 元的持仓，价格 11.9 元的同一持仓，空持仓，再输入成本 10 元、价格 11 元的新持仓。各步新增卖出调用次数为 `0、1、0、1`。新仓位第一次观察仍沿用旧峰值触发卖出。 + +**影响：** 重新建仓或成本基准变化后，当前收益率可能被当成旧仓位的回撤，提前触发止盈。本项仅保留 Trend 范围。 + +**修改建议：** 根据已确认的清仓、重新建仓及成本变化同步止盈基准。不要仅在提交成功时清理，避免撤单或部分成交丢失有效峰值。 + +**验收：** 新仓位首次观察只建立峰值；同一仓位正常回撤仍触发卖出；未成交提交、撤单及同一成本下部分卖出不误清理。 + +### P2-07：OrderBook 旧构造调用导致回归测试错误 + +**位置：** [test_deal_model.py](../py-client/tests/test_deal_model.py) 第 71 行；[OrderBook 构造函数](../py-client/libs/order.py) 第 35–38 行。 + +**本轮复现：** 测试调用 `ActiveOrders('trend')`，字符串被当成 `lock_timeout_sec`，在 `max(1, lock_timeout_sec)` 处抛出 `TypeError`。完整测试 86 项中 85 项通过、1 项错误、0 项断言失败。 + +**影响与边界：** 回归套件无法全绿。实际 Trend/ZT 入口已使用无参构造，本项不意味着策略必然启动失败。 + +**修改建议：** 将测试改为当前构造接口；撤单范围通过 `refresh()` 参数和调用方约定验证，不为旧测试重新引入已删除的构造参数。 + +**验收：** 该测试通过，并保持默认刷新、ZT 范围过滤、IPO 排除和在途防重断言有效。 + +## 本轮验证及边界 + +- 完整执行 `unittest` 发现的 86 项测试：85 通过、1 项错误，详情见 P2-07。其中 IPO 专项 15 项、ZT 相关 26 项全部通过。 +- 额外执行真实状态同步、委托刷新和 Trend 持仓管理路径，分别复现 P1-07、P1-08、P2-01，结果已写入对应条目。 +- 验证使用模拟客户端及临时 SQLite 数据库，未发送真实委托、撤单或采集请求，未修改生产数据库和锁文件。 +- 成交唯一身份、逐笔/累计语义及柜台实际受理结果仍需目标环境确认，未将未核验事实写成确定结论。 +- 不新增旧 IPO 编号兼容要求，不恢复已忽略事项，不将已修复内容作为待办重复列出。 diff --git a/py-client/libs/lockfile.py b/py-client/libs/lockfile.py index a272318..b20fc2e 100644 --- a/py-client/libs/lockfile.py +++ b/py-client/libs/lockfile.py @@ -2,6 +2,9 @@ from os import PathLike from pathlib import Path +import json +import os +from tempfile import NamedTemporaryFile def is_lock(file_path: str | PathLike[str]) -> bool: @@ -14,3 +17,32 @@ def write_lockfile(file_path: str | PathLike[str]) -> None: path = Path(file_path) path.parent.mkdir(parents=True, exist_ok=True) path.write_text("LOCK", encoding="utf-8") + + +def claim_json(path: Path, record: dict) -> bool: + """原子占位并落盘，成功后才允许提交；异常留下记录等待核对。""" + path.parent.mkdir(parents=True, exist_ok=True) + try: + stream = path.open('x', encoding='utf-8') + except FileExistsError: + return False + with stream: + json.dump(record, stream, ensure_ascii=False) + stream.flush() + os.fsync(stream.fileno()) + return True + + +def replace_json(path: Path, record: dict) -> None: + """原子替换核对结果，避免其他任务读取到半条记录。""" + temporary = None + try: + with NamedTemporaryFile(mode='w', encoding='utf-8', dir=path.parent, delete=False) as stream: + temporary = Path(stream.name) + json.dump(record, stream, ensure_ascii=False) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + finally: + if temporary is not None: + temporary.unlink(missing_ok=True) diff --git a/py-client/libs/order.py b/py-client/libs/order.py index 5773542..3bc6517 100644 --- a/py-client/libs/order.py +++ b/py-client/libs/order.py @@ -33,9 +33,8 @@ class OrderBook: """线程安全的活动委托缓存。""" def __init__( - self, order_prefix: str, lock_timeout_sec: int = 180, cancel_timeout_sec: float = 30 + self, lock_timeout_sec: int = 180, cancel_timeout_sec: float = 30 ) -> None: - self.order_prefix = order_prefix self.lock_timeout_sec = max(1, lock_timeout_sec) self.cancel_timeout_sec = timedelta(seconds=cancel_timeout_sec) self.data: list[OrderItem] = [] @@ -59,7 +58,7 @@ class OrderBook: def _busy_key(side: str, code: str) -> str: return f"{side}-{code}" - def refresh(self, client: Client, orders: list[OrderItem]) -> None: + def refresh(self, client: Client, orders: list[OrderItem], *, cancel_prefix: str | None = None) -> None: """用账户快照刷新委托，并撤销超时的活动委托。""" current = datetime.now() data: list[OrderItem] = [] @@ -77,7 +76,8 @@ class OrderBook: created_at = item.created_at if ( created_at is not None - and item.local_order_id.startswith(f"{self.order_prefix}-") + and not item.local_order_id.startswith('IPO-') + and (cancel_prefix is None or item.local_order_id.startswith(cancel_prefix)) and status in CANCELABLE_STATUSES and current - created_at > self.cancel_timeout_sec ): diff --git a/py-client/libs/state.py b/py-client/libs/state.py index 2cd66ec..58b890a 100644 --- a/py-client/libs/state.py +++ b/py-client/libs/state.py @@ -1,6 +1,7 @@ """SQLite 策略状态与成交存储；每个数据库仅使用一个写入者，不做数据迁移。""" import math +import json import logging as log import sqlite3 from contextlib import closing @@ -144,9 +145,11 @@ class State: self.deals, self.deals_sys_ids = deals, set(deals) @staticmethod - def _insert_deals(db: sqlite3.Connection, deals: list[DealItem]) -> None: + def _insert_deals(db: sqlite3.Connection, deals: list[DealItem], + existing_ids: set[str] | None = None) -> None: """共享事务内保存成交；空本地编号使用明确的待核对标记。""" - existing = {row[0] for row in db.execute('SELECT order_sys_id FROM deals')} + existing = existing_ids if existing_ids is not None else { + row[0] for row in db.execute('SELECT order_sys_id FROM deals')} new_deals: dict[str, DealItem] = {} for deal in deals: if deal.order_sys_id not in existing: @@ -280,13 +283,16 @@ class State: 'AND is_arch = 0', (code,), ) - def _archive_pending(self, db: sqlite3.Connection, snapshot_dedup: bool = True) -> None: + def _archive_pending(self, db: sqlite3.Connection, snapshot_dedup: bool = True, + blocked_codes: set[str] | None = None) -> None: """保留现有逐证券保存点；ZT 增量路径禁用数量相等推断。""" codes = db.execute( 'SELECT DISTINCT stock_code FROM deals WHERE is_arch = 0' ).fetchall() for row in codes: code = row['stock_code'] + if blocked_codes and code in blocked_codes: + continue db.execute('SAVEPOINT archive_stock') try: self._archive_stock(db, code, snapshot_dedup) @@ -314,6 +320,11 @@ class State: holdings = {p.stock_code: p for p in positions if p.volume > 0} with closing(self._connect()) as db, db: db.execute('BEGIN IMMEDIATE') + db.execute('''CREATE TABLE IF NOT EXISTS zt_rejected_deals ( + stock_code TEXT NOT NULL, order_sys_id TEXT NOT NULL, + payload TEXT NOT NULL, error TEXT NOT NULL, + PRIMARY KEY (stock_code, order_sys_id) + )''') if initialize: if (db.execute('SELECT 1 FROM state LIMIT 1').fetchone() or db.execute('SELECT 1 FROM deals LIMIT 1').fetchone()): @@ -330,11 +341,13 @@ class State: # 这些成交已包含在已核对的基准中，不再增减快照库存。 db.execute('UPDATE deals SET is_arch = 1') else: - self._insert_deals(db, deals) - self._archive_pending(db, snapshot_dedup=False) + self._insert_account_deals(db, deals) + rejected = {row[0] for row in db.execute('SELECT stock_code FROM zt_rejected_deals')} + self._archive_pending(db, snapshot_dedup=False, blocked_codes=rejected) state = self._read_state(db) cached_deals = self._read_deals(db) blocked = {d['stock_code'] for d in cached_deals.values() if d['is_arch'] != 1} + blocked.update(row[0] for row in db.execute('SELECT stock_code FROM zt_rejected_deals')) for code in set(state) | set(holdings): row = state.get(code, {}) recorded = row.get('base_qty', 0) + row.get('added_qty', 0) @@ -345,3 +358,35 @@ class State: self.blocked_codes = blocked if blocked: log.warning('[ZT 同步] 以下证券状态待核对，暂停交易：%s', ', '.join(sorted(blocked))) + + def _insert_account_deals(self, db: sqlite3.Connection, deals: list[DealItem]) -> None: + """仅 ZT 增量路径隔离坏成交；保留原始字段，缺席回报不会解除隔离。""" + existing = {row[0] for row in db.execute('SELECT order_sys_id FROM deals')} + for deal in deals: + if not deal.stock_code: + raise ValueError('Trade without stock code cannot be isolated') + db.execute('SAVEPOINT insert_zt_deal') + try: + if not deal.stock_code or not deal.order_sys_id: + raise ValueError('Trade code and order ID are required') + if (deal.offset_flag not in (FLAG_BUY, FLAG_SELL) + or not math.isfinite(deal.volume) or deal.volume <= 0 + or int(deal.volume) != deal.volume + or not math.isfinite(deal.price) or deal.price < 0 + or not math.isfinite(deal.trade_amount) + or not math.isfinite(deal.close_profit)): + raise ValueError('Invalid trade direction, volume or amount') + self._insert_deals(db, [deal], existing) + except (ValueError, TypeError, OverflowError, sqlite3.IntegrityError) as exc: + db.execute('ROLLBACK TO insert_zt_deal') + db.execute('''INSERT OR REPLACE INTO zt_rejected_deals + (stock_code, order_sys_id, payload, error) VALUES (?, ?, ?, ?)''', + (deal.stock_code, deal.order_sys_id, json.dumps(asdict(deal), ensure_ascii=False), str(exc))) + log.warning('[ZT 成交] 隔离证券=%s，委托=%s，原因=%s', + deal.stock_code, deal.order_sys_id, exc) + else: + existing.add(deal.order_sys_id) + db.execute('DELETE FROM zt_rejected_deals WHERE stock_code = ? AND order_sys_id = ?', + (deal.stock_code, deal.order_sys_id)) + finally: + db.execute('RELEASE insert_zt_deal') diff --git a/py-client/sdk/trade.py b/py-client/sdk/trade.py index 25ff0a4..5e6262b 100644 --- a/py-client/sdk/trade.py +++ b/py-client/sdk/trade.py @@ -47,8 +47,10 @@ class TradeMixin: response = self._post_json( "/api/trade/ipo_data", {"type": str(ipo_type).strip().upper()}, - ) or [] - return response if isinstance(response, list) else [] + ) + if not isinstance(response, list): + raise ValueError('IPO response must be a list') + return response def cancel_by_id(self, order_id: str) -> dict[str, Any]: diff --git a/py-client/strategy/ipo/boot.py b/py-client/strategy/ipo/boot.py index fbf3d1e..da00e48 100644 --- a/py-client/strategy/ipo/boot.py +++ b/py-client/strategy/ipo/boot.py @@ -1,85 +1,112 @@ -"""新股自动申购，提供交易日校验、券商对账和本地幂等保护。""" +"""IPO 申购：账户隔离、提交前占位、通过券商委托核对结果。""" +import hashlib import json import logging -from datetime import datetime, time +import math +import re +from datetime import datetime +from decimal import Decimal, InvalidOperation 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] 自动申购任务失败") +from libs.lockfile import claim_json, replace_json +import secrets 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 + """保留原板块范围，同时校验完整代码和对应交易所。""" + return isinstance(symbol, str) and re.fullmatch( + r'(?:60[0-9]{4}|68[89][0-9]{3})\.SH|(?:00[0-3][0-9]{3}|30[01][0-9]{3})\.SZ', + symbol, + ) is not None + + +def _candidate(item: dict) -> tuple[str, float, int]: + if not isinstance(item, dict): + raise ValueError('IPO candidate must be an object') + stock = item.get('stock') + if not isinstance(stock, str) or not is_target_stock(stock.strip()): + raise ValueError('Invalid or unsupported IPO stock code') + if isinstance(item.get('issuePrice'), bool) or isinstance(item.get('maxPurchaseNum'), bool): + raise ValueError('IPO price and volume cannot be boolean') + price = float(item['issuePrice']) + try: + quantity = Decimal(str(item['maxPurchaseNum'])) + except InvalidOperation as exc: + raise ValueError('Invalid IPO volume') from exc + if not math.isfinite(price) or price <= 0: + raise ValueError('IPO price must be positive and finite') + if not quantity.is_finite() or quantity <= 0 or quantity != quantity.to_integral_value(): + raise ValueError('IPO volume must be a positive integer') + return stock.strip(), price, int(quantity) + + +def _subscribe(client, orders, account: str, day: str, stock: str, price: float, volume: int) -> bool: + # 每次明确拒绝后的尝试单独占位，避免两个进程同时重试。 + account_key = hashlib.sha256(account.encode('utf-8')).hexdigest() + folder = Path(config.global_config.qmt_data_dir) / 'ipo' / account_key / day / stock + paths = sorted(folder.glob('[0-9]*.json'), key=lambda p: int(p.stem)) + path = paths[-1] if paths else None + record = json.loads(path.read_text(encoding='utf-8')) if path else None + if record and record['status'] == 'confirmed': + return False + + matching = [o for o in orders if o.stock_code == stock and o.side == 'BUY' + and (o.insert_date.replace('-', '') == day + or (record and o.local_order_id == record['order_id']))] + # 已有其他有效/未知状态买单也阻止再申购，覆盖人工提交和旧版记录。 + if any(str(o.order_status) != '57' or o.volume_traded > 0 for o in matching): + if record and any(o.local_order_id == record['order_id'] and str(o.order_status) == '56' + for o in matching): + record['status'] = 'confirmed' + replace_json(path, record) + logging.info('[IPO] %s 已有委托，保持防重', stock) + return False + if record: + rejected = any(o.local_order_id == record['order_id'] and str(o.order_status) == '57' + and o.volume_traded == 0 for o in matching) + if not rejected: + logging.info('[IPO] %s 结果待确认，暂不重发', stock) + return False + record['status'] = 'rejected' + replace_json(path, record) + + attempt = int(path.stem) + 1 if path else 1 + order_id = f'IPO-{secrets.token_hex(12)}' + record = dict(account=account, date=day, stock=stock, order_id=order_id, + status='pending', price=price, volume=volume) + if not claim_json(folder / f'{attempt:04d}.json', record): + return False + # HTTP 正常返回或异常均不代表最终结果；仅券商回报可改变 pending。 + client.passorder(op_type=23, stock=stock, volume=volume, pr_type=11, + price=price, strategy_name='ipo', order_id=order_id) + logging.info('[IPO] %s 已提交 %d 股，等待委托确认，编号=%s', stock, volume, order_id) + return True + + +def AutoBuyIpo() -> int: + """返回本次正常返回的提交数量，不代表最终申购成功。""" + now = datetime.now() + if not config.account_config.enable_auto_ipo or not trading_time(now): + return 0 + submitted = 0 + try: + account = str(config.account_config.account_id).strip() + if not account: + raise ValueError('IPO account ID is required') + with Client(config.global_config.qmt_base_url, config.global_config.qmt_token, + config.HTTP_TIMEOUT) as client: + candidates = client.ipo_data('STOCK') + orders = client.orders() # 查询失败时不提交，不能将未知当作无委托。 + for item in candidates: + try: + stock, price, volume = _candidate(item) + submitted += _subscribe(client, orders, account, now.strftime('%Y%m%d'), + stock, price, volume) + except Exception: + logging.exception('[IPO] 单条申购处理失败，数据=%r', item) + except Exception: + logging.exception('[IPO] 自动申购任务失败') + return submitted diff --git a/py-client/strategy/trend/boot.py b/py-client/strategy/trend/boot.py index 08434c7..4f963ba 100644 --- a/py-client/strategy/trend/boot.py +++ b/py-client/strategy/trend/boot.py @@ -36,7 +36,7 @@ def StartTrend() -> None: assets = portfolio.assets positions = list(portfolio.positions.values()) cache_portfolio(config.account_config.account_id, assets, positions, client.deals()) - order_book = OrderBook("trend") + order_book = OrderBook() order_book.refresh(client, portfolio.orders) # 获取本策略的信号开仓数据 diff --git a/py-client/strategy/zt/boot.py b/py-client/strategy/zt/boot.py index 78bd36f..029617d 100644 --- a/py-client/strategy/zt/boot.py +++ b/py-client/strategy/zt/boot.py @@ -7,7 +7,7 @@ from pathlib import Path from concurrent.futures import Future, ThreadPoolExecutor import config from libs.calc import trading_time -from libs.grid_take_profit import GridTrailingTracker +from .profit import ZTProfitTracker from libs.market import market_allow_open from libs.order import OrderBook from libs.overview import Overview @@ -33,8 +33,8 @@ def StartZT() -> None: executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="zt") run = Runtime( client=client, global_cfg=config.global_config, account_cfg=config.account_config, - orders=OrderBook('zt'), open_watch=DipWatch(), add_watch=DipWatch(), - profit_tracker=GridTrailingTracker(config.account_config.grid_step_pct), + orders=OrderBook(), open_watch=DipWatch(), add_watch=DipWatch(), + profit_tracker=ZTProfitTracker(config.account_config.grid_step_pct), executor=executor ) @@ -56,7 +56,8 @@ def StartZT() -> None: config.account_config.account_id, len(signals), len(positions)) cache_portfolio(config.account_config.account_id, assets, positions, deals) state.sync_account(positions, deals, initialize=initialize) - run.orders.refresh(client, portfolio.orders) + run.profit_tracker.sync_positions(positions, state) + run.orders.refresh(client, portfolio.orders, cancel_prefix='zt-') Overview(assets, positions, config.account_config) DEFAULT_TICK_INTERVAL = 30 @@ -111,9 +112,11 @@ def RunOnce(run: Runtime, state: State, signals: list[SignalItem]) -> None: assets = portfolio.assets positions = list(portfolio.positions.values()) position_codes = list(portfolio.positions) + cache_portfolio(run.account_cfg.account_id, assets, positions, deals) state.sync_account(positions, deals) - run.orders.refresh(run.client, portfolio.orders) + run.profit_tracker.sync_positions(positions, state) + run.orders.refresh(run.client, portfolio.orders, cancel_prefix='zt-') except Exception: log.exception("[Portfolio] 刷新账户快照失败") return diff --git a/py-client/strategy/zt/positions.py b/py-client/strategy/zt/positions.py index 02c0f11..d6ef9ad 100644 --- a/py-client/strategy/zt/positions.py +++ b/py-client/strategy/zt/positions.py @@ -213,7 +213,8 @@ def handle_loss( def _position_key(runtime: Runtime, code: str) -> str: - return f"{runtime.account_cfg.account_id}:{code}" + # tracker 为该账户的 ZT Runtime 独享，与 sync_positions 使用同一个键。 + return code def get_add_num(hands: int, market_value: float) -> int: diff --git a/py-client/strategy/zt/profit.py b/py-client/strategy/zt/profit.py new file mode 100644 index 0000000..33b2432 --- /dev/null +++ b/py-client/strategy/zt/profit.py @@ -0,0 +1,31 @@ +"""ZT 按已同步的仓位及实际成本管理止盈峰值。""" + +from libs.grid_take_profit import GridTrailingTracker + + +class ZTProfitTracker(GridTrailingTracker): + def __init__(self, step: float = 1.0): + super().__init__(step) + self._bases: dict[str, tuple] = {} + + def sync_positions(self, positions, state) -> None: + # 与交易线程串行执行；提交委托本身不会改变这里的基准。 + current = {} + for position in positions: + code = position.stock_code + row = state.get_by_code(code) + if position.volume <= 0 or not row: + continue + bucket = 'added' if row.get('added_qty', 0) > 0 else 'base' + cost = row.get('added_price', 0) if bucket == 'added' else position.open_price + current[code] = (bucket, cost, row.get(f'{bucket}_order_local_id', ''), + row.get(f'{bucket}_created_at', '')) + for code in self._bases.keys() | current.keys(): + if code in state.blocked_codes: + continue + if self._bases.get(code) != current.get(code): + self.clear(code) + if code in current: + self._bases[code] = current[code] + else: + self._bases.pop(code, None) diff --git a/py-client/tests/test_ipo.py b/py-client/tests/test_ipo.py new file mode 100644 index 0000000..6304a87 --- /dev/null +++ b/py-client/tests/test_ipo.py @@ -0,0 +1,175 @@ +import json +import tempfile +import unittest +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime +from pathlib import Path +from threading import Barrier +from types import SimpleNamespace as NS +from unittest.mock import Mock, patch + +from libs.lockfile import claim_json +from sdk import OrderItem +from sdk.trade import TradeMixin +from strategy.ipo import boot + + +class IPOTests(unittest.TestCase): + def setUp(self): + temp = tempfile.TemporaryDirectory() + self.addCleanup(temp.cleanup) + self.root = Path(temp.name) + self.account = NS(account_id='account-a', enable_auto_ipo=True) + self.global_cfg = NS(qmt_data_dir=temp.name, qmt_base_url='unused', qmt_token='') + self.client = Mock() + self.client.__enter__ = Mock(return_value=self.client) + self.client.__exit__ = Mock(return_value=False) + self.client.orders.return_value = [] + self.client.ipo_data.return_value = [dict(stock='600001.SH', issuePrice=10, maxPurchaseNum=100)] + for target, value in [('account_config', self.account), ('global_config', self.global_cfg)]: + ctx = patch.object(boot.config, target, value) + ctx.start() + self.addCleanup(ctx.stop) + ctx = patch.object(boot, 'Client', return_value=self.client) + ctx.start() + self.addCleanup(ctx.stop) + ctx = patch.object(boot, 'datetime') + self.clock = ctx.start() + self.clock.now.return_value = datetime(2026, 9, 11, 10) + self.addCleanup(ctx.stop) + + def records(self): + return [json.loads(p.read_text(encoding='utf-8')) for p in sorted(self.root.rglob('*.json'))] + + def order(self, status, traded=0): + return OrderItem(stock_code='600001.SH', insert_date='20260911', + remark=self.records()[-1]['order_id'] + '|ipo', offset_flag=23, + order_status=status, volume_traded=traded) + + def test_timeout_stays_pending_and_queries_before_next_attempt(self): + self.client.passorder.side_effect = TimeoutError('response lost') + self.assertEqual(boot.AutoBuyIpo(), 0) + self.assertEqual(self.records()[0]['status'], 'pending') + self.assertEqual(boot.AutoBuyIpo(), 0) + self.assertEqual(self.client.passorder.call_count, 1) + self.assertEqual(self.client.orders.call_count, 2) + + def test_normal_http_response_is_not_confirmation(self): + self.client.passorder.return_value = {'status': 'success'} + self.assertEqual(boot.AutoBuyIpo(), 1) + self.assertEqual(self.records()[0]['status'], 'pending') + self.assertEqual(boot.AutoBuyIpo(), 0) + self.client.passorder.assert_called_once() + + def test_rejection_allows_one_new_attempt_with_new_identity(self): + boot.AutoBuyIpo() + old_id = self.records()[0]['order_id'] + self.client.orders.return_value = [self.order(57)] + self.assertEqual(boot.AutoBuyIpo(), 1) + self.assertEqual([r['status'] for r in self.records()], ['rejected', 'pending']) + self.assertNotEqual(self.records()[1]['order_id'], old_id) + boot.AutoBuyIpo() # Old rejection cannot authorize retry of the new attempt. + self.assertEqual(self.client.passorder.call_count, 2) + + def test_completed_order_confirms_and_prevents_resubmission(self): + boot.AutoBuyIpo() + self.client.orders.return_value = [self.order(56)] + self.assertEqual(boot.AutoBuyIpo(), 0) + self.assertEqual(self.records()[0]['status'], 'confirmed') + self.client.orders.return_value = [] + boot.AutoBuyIpo() + self.client.passorder.assert_called_once() + + def test_active_unknown_and_partial_orders_never_retry(self): + boot.AutoBuyIpo() + for status, traded in [(50, 0), (255, 0), (55, 40), (57, 40), (54, 0)]: + with self.subTest(status=status, traded=traded): + self.client.orders.return_value = [self.order(status, traded)] + self.assertEqual(boot.AutoBuyIpo(), 0) + self.client.passorder.assert_called_once() + + def test_orders_query_failure_does_not_submit(self): + self.client.orders.side_effect = TimeoutError('unavailable') + self.assertEqual(boot.AutoBuyIpo(), 0) + self.client.passorder.assert_not_called() + self.assertEqual(self.records(), []) + + def test_accounts_do_not_share_reservations(self): + boot.AutoBuyIpo() + self.account.account_id = 'account-b' + self.assertEqual(boot.AutoBuyIpo(), 1) + self.assertEqual(self.client.passorder.call_count, 2) + self.assertEqual({r['account'] for r in self.records()}, {'account-a', 'account-b'}) + + def test_manual_same_day_order_prevents_new_submission(self): + self.client.orders.return_value = [OrderItem(stock_code='600001.SH', offset_flag=23, + order_status=50, insert_date='2026-09-11')] + boot.AutoBuyIpo() + self.client.passorder.assert_not_called() + + def test_duplicate_candidates_only_submit_once(self): + self.client.ipo_data.return_value *= 2 + self.assertEqual(boot.AutoBuyIpo(), 1) + self.client.passorder.assert_called_once() + + def test_concurrent_initial_and_rejected_attempts_are_atomic(self): + for orders in [[], None]: + if orders is None: + orders = [self.order(57)] + barrier = Barrier(2) + def claim(path, record): + barrier.wait(timeout=5) + return claim_json(path, record) + with patch.object(boot, 'claim_json', side_effect=claim), ThreadPoolExecutor(2) as pool: + futures = [pool.submit(boot._subscribe, self.client, orders, 'account-a', + '20260911', '600001.SH', 10, 100) for _ in range(2)] + self.assertEqual(sum(f.result() for f in futures), 1) + self.assertEqual(self.client.passorder.call_count, 2) + + def test_pending_record_is_durable_before_request(self): + def submitted(**kwargs): + record = self.records()[0] + self.assertEqual(record['status'], 'pending') + self.assertEqual(record['order_id'], kwargs['order_id']) + self.client.passorder.side_effect = submitted + self.assertEqual(boot.AutoBuyIpo(), 1) + + def test_claim_failure_never_submits(self): + with patch.object(boot, 'claim_json', side_effect=OSError('disk failed')): + self.assertEqual(boot.AutoBuyIpo(), 0) + self.client.passorder.assert_not_called() + + def test_invalid_candidate_is_skipped_without_blocking_good_one(self): + good = self.client.ipo_data.return_value[0] + invalid = [dict(good, stock='600../x.SH'), dict(good, issuePrice=float('nan')), + dict(good, issuePrice=float('inf')), dict(good, maxPurchaseNum=100.5), + dict(good, maxPurchaseNum=True), dict(good, maxPurchaseNum='NaN'), + dict(good, maxPurchaseNum=0), dict(good, issuePrice=False), + dict(good, stock='600001.SZ'), None] + self.client.ipo_data.return_value = invalid + [good] + self.assertEqual(boot.AutoBuyIpo(), 1) + self.client.passorder.assert_called_once() + self.assertEqual(len(self.records()), 1) + + def test_supported_codes_and_numeric_strings(self): + for stock in ['600001.SH', '688001.SH', '689001.SH', '000001.SZ', + '001001.SZ', '002001.SZ', '003001.SZ', '300001.SZ', '301001.SZ']: + self.assertEqual(boot._candidate(dict(stock=stock, issuePrice='10.5', maxPurchaseNum='100')), + (stock, 10.5, 100)) + for stock in ['600abc.SH', '600001', '600001.SH/x', '688001.SZ', '300001.SH', None]: + self.assertFalse(boot.is_target_stock(stock)) + + +class IPOResponseTests(unittest.TestCase): + def test_only_list_response_is_accepted(self): + client = TradeMixin() + for value in [None, {}, {'error': 'bad'}, '', 0, False]: + client._post_json = Mock(return_value=value) + with self.subTest(value=value), self.assertRaises(ValueError): + client.ipo_data() + client._post_json = Mock(return_value=[]) + self.assertEqual(client.ipo_data(), []) + + +if __name__ == '__main__': + unittest.main() diff --git a/py-client/tests/test_zt_audit_fixes.py b/py-client/tests/test_zt_audit_fixes.py new file mode 100644 index 0000000..a365212 --- /dev/null +++ b/py-client/tests/test_zt_audit_fixes.py @@ -0,0 +1,132 @@ +import tempfile +import unittest +from contextlib import closing +from dataclasses import replace +from datetime import datetime, timedelta +from pathlib import Path +from types import SimpleNamespace as NS +from unittest.mock import Mock, patch + +from libs.grid_take_profit import GridState +from libs.order import OrderBook +from libs.snapshot import get_collector_snapshot +from libs.state import State +from sdk import Assets, DealItem, OrderItem, PositionItem +from strategy.zt import boot +from strategy.zt.profit import ZTProfitTracker + + +class ZTAuditFixTests(unittest.TestCase): + def setUp(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.store = State(Path(tmp.name) / 'state.db') + self.code = '600000.SH' + + def deal(self, code, identity): + return DealItem(stock_code=code, order_sys_id=identity, + remark=f'zt-base-{identity}|zt', offset_flag=48, + volume=100, price=10, trade_amount=1000) + + def position(self, code): + return PositionItem(stock_code=code, volume=100, open_price=10) + + def test_zt_cancels_only_owned_orders_and_tracks_all(self): + stamp = datetime.now() - timedelta(minutes=2) + orders = [OrderItem(stock_code=f'60000{i}.SH', order_sys_id=str(i), remark=remark, + order_status=50, offset_flag=23, + insert_date=stamp.strftime('%Y%m%d'), insert_time=stamp.strftime('%H%M%S')) + for i, remark in enumerate(['zt-base-own|zt', 'zt-SELL-own|zt', + 'zt-added-own|zt', 'IPO-new|ipo', '', 'TREN-BUY-other'])] + client = Mock() + book = OrderBook() + book.refresh(client, orders, cancel_prefix='zt-') + self.assertEqual([c.args[0] for c in client.cancel_by_id.call_args_list], ['0', '1', '2']) + self.assertEqual(book.data, orders) + self.assertTrue(all(book.busy(o.stock_code, 'BUY') for o in orders)) + client.reset_mock() + book.refresh(client, orders) + self.assertEqual([c.args[0] for c in client.cancel_by_id.call_args_list], ['0', '1', '2', '4', '5']) + + def test_invalid_trade_persists_blocks_only_its_stock_and_recovers(self): + good = '600001.SH' + bad = replace(self.deal(self.code, 'bad'), price=float('nan')) + valid = self.deal(good, 'good') + positions = [self.position(c) for c in (self.code, good)] + self.store.sync_account(positions, [bad, valid]) + self.assertEqual(self.store.blocked_codes, {self.code}) + self.assertEqual(self.store.state[good]['base_qty'], 100) + with closing(self.store._connect()) as db: + payload = db.execute('SELECT payload FROM zt_rejected_deals').fetchone()[0] + self.assertIn('bad', payload) + self.assertIn('NaN', payload) + self.store = State(self.store.path) + self.store.sync_account(positions, [valid]) + self.assertEqual(self.store.blocked_codes, {self.code}) + self.store.sync_account(positions, [replace(bad, price=10), valid]) + self.assertEqual(self.store.blocked_codes, set()) + self.assertEqual(self.store.state[self.code]['base_qty'], 100) + + def test_invalid_fields_do_not_block_other_stocks(self): + for field, value in [('volume', 0), ('volume', 1.5), ('offset_flag', 99), + ('trade_amount', float('inf')), ('price', -1)]: + with self.subTest(field=field): + bad = replace(self.deal(self.code, f'bad-{field}'), **{field: value}) + good = self.deal('600001.SH', 'good') + self.store.sync_account([self.position('600001.SH')], [bad, good]) + self.assertEqual(self.store.state['600001.SH']['base_qty'], 100) + self.assertIn(self.code, self.store.blocked_codes) + + def test_bad_stock_does_not_archive_other_trades_until_corrected(self): + first = self.deal(self.code, 'first') + bad = replace(self.deal(self.code, 'bad'), price=float('nan')) + position = replace(self.position(self.code), volume=200) + self.store.sync_account([position], [first, bad]) + self.assertEqual(self.store.deals['first']['is_arch'], 0) + self.assertNotIn(self.code, self.store.state) + self.store.sync_account([position], [replace(bad, price=10)]) + self.assertEqual(self.store.state[self.code]['base_qty'], 200) + self.assertEqual(self.store.blocked_codes, set()) + + def test_profit_basis_changes_reset_peak_but_partial_sell_does_not(self): + tracker = ZTProfitTracker() + position = self.position(self.code) + row = dict(base_qty=100, base_order_local_id='one', base_created_at='now', added_qty=0) + state = NS(blocked_codes=set(), get_by_code=lambda code: row) + tracker.sync_positions([position], state) + self.assertEqual(tracker.observe(self.code, 20).state, GridState.ARMED) + row['base_qty'] = 50 + tracker.sync_positions([replace(position, volume=50)], state) + self.assertEqual(tracker.observe(self.code, 19).state, GridState.RETREAT) + for update, cost in [({'added_qty': 100, 'added_price': 11}, 10), + ({'added_qty': 0}, 10), ({}, 12), + ({'base_order_local_id': 'reopened'}, 12)]: + row.update(update) + tracker.sync_positions([replace(position, open_price=cost)], state) + self.assertEqual(tracker.observe(self.code, 10).state, GridState.ARMED) + tracker.sync_positions([], state) + tracker.sync_positions([position], state) + self.assertEqual(tracker.observe(self.code, 5).state, GridState.ARMED) + + def test_run_once_updates_collector_before_market_fetch(self): + positions = [self.position(self.code)] + self.store.sync_account(positions, [], initialize=True) + client = Mock() + run = NS(client=client, account_cfg=NS(account_id='zt-test', min_cash_ratio=0.1), + orders=Mock(), profit_tracker=ZTProfitTracker()) + client.deals.return_value = [] + client.portfolio.return_value = NS(assets=Assets(20000, 10000), + positions={self.code: positions[0]}, orders=[]) + client.full_tick.side_effect = RuntimeError('no market data') + with patch.object(boot, 'trading_time', return_value=True), \ + patch.object(boot, 'market_allow_open', return_value=True): + boot.RunOnce(run, self.store, []) + snapshot = get_collector_snapshot() + self.assertEqual(snapshot[0], 'zt-test') + self.assertEqual(snapshot[1].total, 20000) + self.assertEqual(snapshot[2], positions) + run.orders.refresh.assert_called_once_with(client, [], cancel_prefix='zt-') + + +if __name__ == '__main__': + unittest.main()