diff --git a/DEVLOG.md b/DEVLOG.md index d43b14f..87df9a9 100644 --- a/DEVLOG.md +++ b/DEVLOG.md @@ -30,6 +30,37 @@ --- +## 2026-08-20 · 紧急清仓修复：配额不重复扣今日成交 / 清仓命令即时撤策略 + +**背景** +2026-08-19 盘中下一键清仓，部分票卖完，剩三支拖到次日早盘（08-20 开盘）才清。用户观察：这三支既没跌停也在盘中，按紧急清仓逻辑本应当天直接卖掉。据 make watch 与 make t-ins/t-gate 的实机数据排查，定位两条根因，本次一并修，都在执行与命令这一层，其他系统不动。 + +**根因甲，当日配额把今日已成交量重复扣了一次。** +当日配额（今天最多投放多少股）是拿“剩余未成交量”算的，剩余量等于委托量减累计成交量，已经扣过成交一次。可是判断“今天还能不能再投放”时，用的是“配额减今日已投放量”，而“今日已投放量”里既算今天还挂着没成交的委托、又算今天已经成交的委托。于是今天成交的那部分被扣了两次。当天成交越多，“可再投放”越快变成负数，指令发一两笔、成交回来之后就误判“当日配额已出完”，当天不再补单。更糟的是这道判断排在紧急直通和 14:45 强制兜底前面，连兜底都被挡住，要等次日配额清零才继续。实机印证：300953 昨天两笔卖了 200，剩 100 停在“配额已出完”，今早才卖；002128 今早还挂着 200 同样卡在这里。 + +**根因乙，清仓命令没有立刻停掉相关票的买入侧。** +002128 挂着网格策略，而“清仓完成清场”撤策略要等持仓归零才触发。于是那天网格整个上午都在逐档买入、清仓命令同时在卖，买进来的又是 T+1 当天卖不掉，等于清仓命令在追一个自己还在被买进的持仓。账本里那条撤策略的 CLEANUP 记录时间是今早 09:34，证明策略今天才被撤下。这违背设计铁律“命令至上”——命令一下达，冲突的自主动作就该让位。 + +**注**：三支里也有纯 T+1 的（如 300738 昨天上午 09:37 刚建仓 900 股，当天不可卖），这部分是市场规则，紧急清仓也绕不过，只能次日，非缺陷。甲、乙修的是本可当天卖却被卡住的那部分。 + +**做了什么（两处修复，都带单测）** +甲，app/services/executor.py：run_tick 里当日配额的“今日已投放量”口径分两种。紧急单或窗口末日单改用新加的 _inflight_today（只算今日在途、已终态一律不算），这样“可再投放 = 剩余减在途”，永不重复扣，紧急清仓当天会持续补单直到卖完或撞上 T+1 可卖为零；普通多日单仍用 _consumed_today（今日成交加在途）做节流，行为不变。超卖与重复下单由既有的“可卖量”闸与券商侧另兜一道，本改动不放大风险。exec_timing.py 未改，配额闸的行为靠上面这个口径修正自然纠正。 +乙，app/services/command_service.py：新增 _stop_buyside_for_exit，在 LIQUIDATE_ALL、EXIT_STOCK、SECTOR_EXIT 三类全额清仓命令下达时立刻撤掉相关票的活跃策略、驳回待确认的买入提议；在途买单本就由 planner 的撤单项处理，三层合起来做到“命令一下、这只票不再有任何新买入”。撤策略只改策略状态、不动持仓，安全；幂等、失败不抛，单只失败记 errors 不拖垮命令本体。 + +**动了哪些文件** +app/services/executor.py（run_tick 配额口径分流 + 新增 _inflight_today）；app/services/command_service.py（plan_command 加停买入侧调用 + 新增 _stop_buyside_for_exit + 进度回执 stopped_buyside）；scripts/test_batch15_units.py（新增 7 例：在途口径、老新口径下紧急清仓 FIRE 与配额已出完的分野用真函数钉死根因、全在途正确等待、run_tick 分流源码级防回归、撤策略与驳回买入提议不误伤卖出与无关票、plan_command 走一遍 LIQUIDATE_ALL 确实撤策略）；scripts/run_tests.py（登记 batch15，总数 505）。 + +**部署方式** +桥机 factorevaluation 上 make deploy（源码打进镜像，必须重建容器），make test 见 ALL SUITES PASS（已含 batch15）。收盘后部署，让下一次紧急清仓发生在你在场时。不新增参数，不动 .env。 + +**真机判收** +未判收。开发容器全量单测 ALL SUITES PASS（505 例，含新增 7 例）。桥机判收：部署后下一次一键清仓，预期同一只票在同一天内持续补单直到卖完或卖到 T+1 的墙，不再出现“卖两笔就停在配额已出完、要等次日”；且清仓命令一下达，挂着策略的票在命令进度回执 stopped_buyside 里能看到策略已撤、账本有 CLEANUP 行，网格不再边卖边买。 + +**还欠着什么** +一，本次只对全额清仓类命令即时撤策略；减仓类（REDUCE_EXPOSURE/REDUCE_STOCK）是否也要停对应票的加仓侧，未做，按需再议。二，配额“已全部在途、等成交”这一情形目前仍复用“当日配额已出完”这句话，行为正确但措辞会让人误以为是老毛病，下次顺手把这条 WAIT 的文案分开。三，PMS 侧 avail_qty 与券商侧可用持仓偶有不一致（出口委托里有 INSUFFICIENT_POSITION 拒单），本次未动，属对账口径，单独排查。 + +--- + ## 2026-08-18 · 宏观择时层上线：股汇对冲指数管整体仓位，外加个股宏观闸 **背景与依据** diff --git a/app/services/command_service.py b/app/services/command_service.py index 3134689..44ee9d2 100644 --- a/app/services/command_service.py +++ b/app/services/command_service.py @@ -236,6 +236,18 @@ def plan_command(cmd: dict) -> dict: cancelled = cancel_r["cancelled"] cancel_failed = cancel_r["failed"] + # 命令覆盖冲突 (设计铁律「命令至上」): 全额清仓类命令一下达, 立刻停掉这些票的**买入侧** + # —— 撤活跃策略 (网格/做T/跟踪) 与待你确认的买入提议, 别再边卖边买。在途买单已由上面 + # _cancel_marked_instructions 撤掉, 这里补策略与提议这两层。 + # 由来 (2026-08-19 实机): 一键清仓时 002128 挂着网格, 而"清仓完成清场"撤策略要等持仓 + # 归零才触发; 于是那天网格整个上午都在逐档买入、清仓命令同时在卖, 买进来的又是 T+1 + # 锁着当天卖不掉 —— 清仓命令在追一个自己还在被买进的持仓。改成命令下达即撤, 不等归零。 + stop_r = {} + if cmd_type in ("LIQUIDATE_ALL", "EXIT_STOCK", "SECTOR_EXIT"): + exit_codes = sorted({it["ts_code"] for it in items + if it.get("action") == "EXIT" and it.get("ts_code")}) + stop_r = _stop_buyside_for_exit(exit_codes, command_id) + # 开关类命令写运行参数。 # **这行返回值绝对不能丢**: HALT_BUY / HALT_ALL 的"刹车"就是这一次写入 —— 规则闸 # 每一跳都去 param_store 读 PMS_GLOBAL_BUY_HALT / PMS_GLOBAL_EXEC_HALT, 参数没写进去 @@ -259,6 +271,12 @@ def plan_command(cmd: dict) -> dict: if cancel_failed: notes.insert(0, f"⚠ {len(cancel_failed)} 条在途指令**没撤掉, 仍在下游挂着**: " + "; ".join(f"{x['instruction_id']}({x['why']})" for x in cancel_failed[:5])) + if stop_r.get("strategies") or stop_r.get("proposals"): + notes.insert(0, f"清仓覆盖冲突: 已撤策略 {len(stop_r['strategies'])} 条、" + f"驳回买入提议 {len(stop_r['proposals'])} 条 (停止边卖边买)") + if stop_r.get("errors"): + notes.insert(0, f"⚠ 停买入侧有 {len(stop_r['errors'])} 处未成: " + + "; ".join(str(x) for x in stop_r["errors"][:5])) progress = {"target_amount": result.get("target_amount", 0.0), "planned_amount": result.get("planned_amount", 0.0), "done_amount": 0.0, "gap": result.get("gap", 0.0), @@ -269,7 +287,8 @@ def plan_command(cmd: dict) -> dict: # 有一堆候选、全被同一道闸拒了。不聚合出来就得去翻 rejects 原始清单。 "reject_summary": pl.summarize_rejects(rejects), "cancelled_instructions": cancelled, - "cancel_failed": cancel_failed} + "cancel_failed": cancel_failed, + "stopped_buyside": stop_r or {}} # 零方案的两种情形必须分开 (2026-07-31 修) # ------------------------------------------------------------------ @@ -652,6 +671,60 @@ def _ledger_rejects(rejects: list, command_id: str): logger.warning("拒绝留痕失败: %s", e) +# ================================================================ 命令覆盖冲突: 清仓即停买入侧 +def _stop_buyside_for_exit(codes: list, command_id: str) -> dict: + """全额清仓类命令 (一键清仓 / 清仓某股 / 清仓某行业) 下达时, 立刻停掉这些票的买入侧。 + + 撤活跃策略 (网格/做T/跟踪) + 驳回待确认的买入提议。在途买单由 planner 的 HALT 撤单项 + 经 _cancel_marked_instructions 另行撤销, 这里补策略与提议两层 —— 三层合起来才真正做到 + 「命令一下, 这只票不再有任何新的买入」。幂等、不抛异常: 单只失败记进 errors, 不拖垮 + 命令本体 (故障即守成)。撤策略只改策略状态、不动持仓, 安全。""" + out = {"strategies": [], "proposals": [], "errors": []} + codeset = {c for c in (codes or []) if c} + if not codeset: + return out + from app.services import strategy_service + try: + strategies = [s for s in pms_repo.list_strategies(statuses=["ACTIVE", "PAUSED"], limit=500) + if s.get("ts_code") in codeset] + except Exception as e: + strategies = [] + out["errors"].append(f"读策略失败: {type(e).__name__}: {e}") + for s in strategies: + try: + r = strategy_service.set_status(s["strategy_id"], "CANCELLED", by="command") or {} + if r.get("ok"): + out["strategies"].append(s["strategy_id"]) + try: + pms_repo.insert_ledger( + ts_code=s["ts_code"], action="CLEANUP", arbiter="rule", verdict="PASS", + price_at=0, ref_id=command_id, + hard_numbers={"strategy_id": s["strategy_id"], "command_id": command_id}, + reason="清仓命令下达: 撤该票策略, 停止边卖边买 (命令覆盖冲突)") + except Exception as e: + logger.warning("[命令] 撤策略留痕失败 %s: %s", s.get("ts_code"), e) + else: + out["errors"].append(f"{s.get('ts_code')} 撤策略未成: {r.get('error')}") + except Exception as e: + out["errors"].append(f"{s.get('ts_code')} 撤策略失败: {type(e).__name__}: {e}") + try: + props = [p for p in pms_repo.list_proposals(statuses=("WAIT_USER",), limit=500) + if p.get("ts_code") in codeset and p.get("action") in BUILD_ACTIONS] + except Exception as e: + props = [] + out["errors"].append(f"读提议失败: {type(e).__name__}: {e}") + for p in props: + try: + if pms_repo.decide_proposal(p["proposal_id"], "DECLINED"): + out["proposals"].append(p["proposal_id"]) + except Exception as e: + out["errors"].append(f"{p.get('ts_code')} 驳回买入提议失败: {type(e).__name__}: {e}") + if out["strategies"] or out["proposals"]: + logger.warning("[命令] %s 清仓覆盖冲突: 撤策略 %s, 驳回买入提议 %s", + command_id, out["strategies"], out["proposals"]) + return out + + # ================================================================ 清仓完成后清场 CLEANUP_MARK_KEY = "PMS_EXIT_CLEANUP_DONE" # 已闭仓且已清场的代码集 (JSON) BUILD_ACTIONS = ("OPEN", "FILL", "ADD", "DCA") # 建仓类方案/指令 (买入) diff --git a/app/services/executor.py b/app/services/executor.py index 23191ba..ecd581a 100644 --- a/app/services/executor.py +++ b/app/services/executor.py @@ -163,7 +163,19 @@ def run_tick(*, now=None, dry_run: bool = False) -> dict: pos = _pos_of(view, code) quota = et.daily_quota(remaining, tdays_left, allow_odd_tail=(ins.get("action") == "EXIT")) - fired_today = _consumed_today(children, ymd_today) + # 当日配额的「今日已投放量」口径分两种: + # 普通多日单 —— _consumed_today (今日已成交 + 今日在途), 给多日出手节流用。 + # 紧急单 / 窗口末日单 —— 只按「今日在途量」算 (_inflight_today)。这两类单的 + # 配额本来就等于全部剩余量, 而剩余量 = 委托量 − 累计成交, 已经扣过成交一次; + # 若再从「已投放量」里把今日成交扣第二遍, 同一笔成交被扣两次, 当天成交越多, + # 「可再投放 = 配额 − 已投放」越快变成负数 —— 于是发一两笔、成交回来之后就误判 + # 「当日配额已出完」, 当天不再补单, 连 14:45 强制兜底也被挡在这道判断前面, + # 要等次日配额清零才继续。这正是 2026-08-19 紧急清仓剩三支拖到次日早盘的根因。 + # 只按在途量扣: 可再投放 = 剩余 − 在途, 永不重复扣, 紧急清仓当天会持续补单, + # 直到卖完或撞上 T+1 可卖为零 (可卖量那道闸在下面单独兜, 不会超卖/重复下单)。 + is_urgent = bool(prog.get("urgent")) + fired_today = (_inflight_today(children, ymd_today) if (is_urgent or is_last) + else _consumed_today(children, ymd_today)) day_ctx = {**day, "support": pos.get("support_ref"), "limit_up": _limit_up(day), "limit_down": _limit_down(day), "halted": not day or not day.get("price")} @@ -438,6 +450,41 @@ def _consumed_today(children: list, ymd_today: int) -> int: return total +def _inflight_today(children: list, ymd_today: int) -> int: + """今日已下发、但**还没有最终结局**的量 (仍挂在券商, 随时可能成交)。 + + 与 _consumed_today 的差别只有一处: **已终态的委托一律算 0**, 不再把已成交的 cum_qty + 计进来。用途见 run_tick 里那段说明 —— 紧急单 / 窗口末日单的「可再投放」= 剩余 − 在途, + 而剩余量已经扣过累计成交, 所以这里绝不能再把今日成交算进「已投放」, 否则同一笔成交 + 被扣两次。三种结局的处理: + * 仍在途 (未终态) → 按下发量算 (它随时可能成交, 不该重复下单占掉这份量) + * 已终态 (成交/被拒/过期) → 算 0 (成交的已进 exec_qty→剩余量; 被拒/过期的一股没成) + * 查不到出口行 / 影子模式 → 按下发量算 (保守当在途, 宁可少投也不重复投) + """ + total = 0 + for c in children or []: + if int(c.get("ymd") or 0) != ymd_today: + continue + qty = int(c.get("qty") or 0) + ref = c.get("ref") + if not ref or str(ref).startswith("manual:"): + total += qty + continue + try: + o = qmt_repo.get_order(ref) + except Exception: + logger.warning("[配额] 反查出口行失败 %s, 本轮按在途计入", ref) + total += qty + continue + if not o: + total += qty + continue + if str(o.get("status") or "").upper() in qmt_repo.FINAL: + continue # 已终态: 成交的已在剩余量里, 被拒/过期的不占在途 + total += qty # 仍挂在券商, 算在途 + return total + + def _limit_up(day: dict) -> bool: """一字板粗判: 当日最高=最低=现价 且 涨幅为正 (无涨跌停价字段时的兜底口径)。""" if not day: diff --git a/scripts/run_tests.py b/scripts/run_tests.py index bf24fb2..9dedcbd 100644 --- a/scripts/run_tests.py +++ b/scripts/run_tests.py @@ -22,8 +22,10 @@ test_batch13_units.py 决策系统卖出采纳: 门槛分档/清场标记/确认加速/闭仓清场 (4 例) test_batch14_units.py 宏观择时: 指数计算与对齐/区域迟滞/周期与对数映射/ 分方向触发/让路与冷却/宏观闸/建议采纳 (25 例) + test_batch15_units.py 紧急清仓修复: 配额在途口径不重复扣/紧急清仓不被 + 误判配额已出完/清仓命令即时撤策略与买入提议 (7 例) test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) (58 例) - 共 498 例 + 共 505 例 任一子集失败即整体失败 (退出码 1)。 """ import os @@ -36,7 +38,8 @@ SUITES = ["test_core_units.py", "test_batch2_units.py", "test_batch3_units.py", "test_batch4_units.py", "test_batch5_units.py", "test_batch6_units.py", "test_batch7_units.py", "test_batch8_units.py", "test_batch9_units.py", "test_batch10_units.py", "test_batch11_units.py", "test_batch12_units.py", - "test_batch13_units.py", "test_batch14_units.py", "test_wiring.py"] + "test_batch13_units.py", "test_batch14_units.py", "test_batch15_units.py", + "test_wiring.py"] def main(): diff --git a/scripts/test_batch15_units.py b/scripts/test_batch15_units.py new file mode 100644 index 0000000..07fbe2f --- /dev/null +++ b/scripts/test_batch15_units.py @@ -0,0 +1,285 @@ +# -*- coding: utf-8 -*- +""" +第十五批模块单测 (紧急清仓两处修复, 零外部依赖) +================================================ +运行: 在 tradingSystem 仓库根目录执行 python scripts/test_batch15_units.py + +背景 (2026-08-19 实机): 盘中下一键清仓, 部分票卖完、剩三支拖到次日早盘才清。排查结论两条: + 甲, 当日配额把"今日已成交量"重复扣了一次 —— 配额是拿"剩余未成交量"算的(已扣过成交), + 又从"今日已投放量"里再扣一遍今日成交, 同一笔成交扣两次, 当天成交越多、可再投放越快 + 变负, 于是紧急清仓发一两笔就误判"当日配额已出完", 连 14:45 兜底都被挡, 要等次日。 + 乙, 清仓命令没有立刻停掉相关票的买入侧 —— 挂网格的票(002128)清场撤策略要等持仓归零才触发, + 于是网格整天在买、清仓在卖, 买进来的又是 T+1 当天卖不掉, 清仓在追一个自己还在被买的持仓。 + +本批覆盖: + * _inflight_today 只算在途、已终态一律不算 (甲的口径修正); + * 同一批 children 下, _consumed_today 与 _inflight_today 的差, 以及内置择时在两种口径下 + FIRE / 配额已出完 的分野 (直接钉死根因, 用真函数不打桩); + * _stop_buyside_for_exit 撤策略 + 驳回买入提议, 不动卖出提议与无关票 (乙); + * plan_command 下 LIQUIDATE_ALL 时确实撤掉了持仓票的活跃策略 (乙的接线, 真实走一遍). +""" +import os +import sys +import traceback +from datetime import datetime + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from app.core import exec_timing as et # noqa: E402 +from app.services import executor # noqa: E402 +from app.services import command_service as csvc # noqa: E402 +from app.repo import qmt_repo, pms_repo # noqa: E402 + +RESULTS = [] + + +def case(name): + def deco(fn): + RESULTS.append((name, fn)) + return fn + return deco + + +YMD = 20260820 + +# 择时内置B 的参数快照 (卖出用得到的几项) +EXEC_PRM = {"sell_avoid_open_min": 30, "eod_force_time": "14:45", + "eod_force_discount": 0.998, "urgent_sell_discount": 0.995, + "sell_bucket_times": ""} +DAY = {"price": 27.6, "vwap": 27.5, "high": 27.8, "low": 27.3, + "day_chg_from_open": 0.01, "bars": 30} +NOW = datetime(2026, 8, 20, 13, 10) # 下午盘中, 未到 14:45 + + +# ---------------------------------------------------------------- 出口行打桩 +class _FakeQmt: + """按 ref 给出订单终态/在途, 供 _consumed_today / _inflight_today 反查。""" + def __init__(self, orders): + self.orders = orders # ref -> {"status","cum_qty"} + + def get_order(self, ref): + return self.orders.get(ref) + + +def _patch_qmt(orders): + fake = _FakeQmt(orders) + executor.qmt_repo.get_order = fake.get_order + + +def _unpatch_qmt(): + executor.qmt_repo.get_order = qmt_repo.get_order + + +# ================================================================ 甲: 配额口径 +@case("配额口径·_inflight_today 只算在途, 已成交/被拒/过期一律不算") +def _(): + children = [ + {"ymd": YMD, "qty": 900, "ref": "r_filled"}, # 已全成 + {"ymd": YMD, "qty": 500, "ref": "r_rej"}, # 被拒 + {"ymd": YMD, "qty": 400, "ref": "r_live"}, # 仍在途 + {"ymd": YMD, "qty": 100, "ref": "manual:x"}, # 影子/手动: 保守当在途 + {"ymd": 20260819, "qty": 999, "ref": "r_old"}, # 不是今天, 忽略 + ] + _patch_qmt({"r_filled": {"status": qmt_repo.OS_FILLED, "cum_qty": 900}, + "r_rej": {"status": qmt_repo.OS_REJECTED, "cum_qty": 0}, + "r_live": {"status": qmt_repo.OS_ACCEPTED, "cum_qty": 0}}) + try: + # 在途口径: 只有 r_live(400) + manual(100) = 500 + assert executor._inflight_today(children, YMD) == 500, \ + executor._inflight_today(children, YMD) + # 老口径 _consumed_today: 成交 900 + 在途 400 + manual 100 = 1400 (被拒 0) + assert executor._consumed_today(children, YMD) == 1400, \ + executor._consumed_today(children, YMD) + finally: + _unpatch_qmt() + + +@case("配额口径·同一批成交下, 老口径把紧急清仓benched, 新口径继续FIRE (根因钉死)") +def _(): + # 场景复刻 002128: 委托 2300, 今日已发两笔 900+900, 第一笔全成、第二笔在途。 + # 剩余 = 2300 - 900(已成) = 1400。窗口末日, 配额 = 全部剩余 = 1400。 + children = [{"ymd": YMD, "qty": 900, "ref": "d1"}, + {"ymd": YMD, "qty": 900, "ref": "d2"}] + _patch_qmt({"d1": {"status": qmt_repo.OS_FILLED, "cum_qty": 900}, + "d2": {"status": qmt_repo.OS_ACCEPTED, "cum_qty": 0}}) + try: + remaining = 1400 + quota = et.daily_quota(remaining, 1, allow_odd_tail=True) # 末日 → 1400 + assert quota == 1400, quota + old = executor._consumed_today(children, YMD) # 900 + 900 = 1800 + new = executor._inflight_today(children, YMD) # 0 + 900 = 900 + assert old == 1800 and new == 900, (old, new) + + # 老口径: left = 1400 - 1800 < 0 → "当日配额已出完", 紧急也被挡 (bug) + d_old = et.decide(side="sell", now=NOW, day=DAY, params=EXEC_PRM, + is_last_day=True, is_command=True, urgent=True, + fired_today=old, quota=quota) + assert d_old["action"] == et.ACT_WAIT and "配额已出完" in d_old["reason"], d_old + # 新口径: left = 1400 - 900 = 500 > 0 → 紧急直通 FIRE, 还能再投 500 + d_new = et.decide(side="sell", now=NOW, day=DAY, params=EXEC_PRM, + is_last_day=True, is_command=True, urgent=True, + fired_today=new, quota=quota) + assert d_new["action"] == et.ACT_FIRE and d_new["qty_hint"] == 500, d_new + finally: + _unpatch_qmt() + + +@case("配额口径·全部在途(没有可再投放)时正确等待, 不重复下单") +def _(): + children = [{"ymd": YMD, "qty": 1400, "ref": "d1"}] + _patch_qmt({"d1": {"status": qmt_repo.OS_ACCEPTED, "cum_qty": 0}}) + try: + remaining = 1400 + quota = et.daily_quota(remaining, 1, allow_odd_tail=True) + new = executor._inflight_today(children, YMD) # 1400 在途 + assert new == 1400 + d = et.decide(side="sell", now=NOW, day=DAY, params=EXEC_PRM, + is_last_day=True, is_command=True, urgent=True, + fired_today=new, quota=quota) + # left = 0 → 等在途成交, 不再投 (避免超卖/重复下单) + assert d["action"] == et.ACT_WAIT, d + finally: + _unpatch_qmt() + + +@case("配额口径·run_tick 选口径: 紧急或末日走在途口径, 普通多日走老口径") +def _(): + # 用源码级断言确认 run_tick 里的分流条件没被改回去 (轻量, 防回归) + import inspect + src = inspect.getsource(executor.run_tick) + assert "_inflight_today" in src and "is_urgent or is_last" in src, "run_tick 分流丢了" + assert "_consumed_today" in src, "多日节流口径丢了" + + +# ================================================================ 乙: 命令覆盖冲突 +class _Rec: + def __init__(self): + self.strategies = [] + self.proposals = [] + self.cancelled = [] + self.declined = [] + self.ledger = [] + + +def _patch_stop(rec, strategies, proposals): + from app.services import strategy_service + rec._orig = (pms_repo.list_strategies, pms_repo.list_proposals, pms_repo.decide_proposal, + pms_repo.insert_ledger, strategy_service.set_status) + pms_repo.list_strategies = lambda **kw: list(strategies) + pms_repo.list_proposals = lambda **kw: list(proposals) + + def _decide(pid, decision): + rec.declined.append((pid, decision)) + return True + pms_repo.decide_proposal = _decide + pms_repo.insert_ledger = lambda **kw: rec.ledger.append(kw) or 1 + + def _setst(sid, status, by="user"): + rec.cancelled.append((sid, status, by)) + return {"ok": True, "status": status} + strategy_service.set_status = _setst + + +def _unpatch_stop(rec): + from app.services import strategy_service + (pms_repo.list_strategies, pms_repo.list_proposals, pms_repo.decide_proposal, + pms_repo.insert_ledger, strategy_service.set_status) = rec._orig + + +@case("命令覆盖冲突·撤活跃策略 + 驳回买入提议, 不动卖出提议与无关票") +def _(): + rec = _Rec() + strategies = [{"strategy_id": "STR_A", "ts_code": "002128.SZ", "type": "GRID"}, + {"strategy_id": "STR_B", "ts_code": "600000.SH", "type": "T0"}] # 无关票 + proposals = [{"proposal_id": "P_buy", "ts_code": "002128.SZ", "action": "DCA"}, # 买, 驳 + {"proposal_id": "P_sell", "ts_code": "002128.SZ", "action": "TRIM"}, # 卖, 留 + {"proposal_id": "P_other", "ts_code": "600000.SH", "action": "OPEN"}] # 无关票 + _patch_stop(rec, strategies, proposals) + try: + out = csvc._stop_buyside_for_exit(["002128.SZ"], "CMD_T_0001") + assert out["strategies"] == ["STR_A"], out # 只撤本票策略 + assert out["proposals"] == ["P_buy"], out # 只驳本票买入提议 + assert rec.cancelled == [("STR_A", "CANCELLED", "command")], rec.cancelled + assert rec.declined == [("P_buy", "DECLINED")], rec.declined + assert any(x.get("action") == "CLEANUP" for x in rec.ledger) # 留痕 + assert not out["errors"], out + finally: + _unpatch_stop(rec) + + +@case("命令覆盖冲突·空票集不动任何东西") +def _(): + rec = _Rec() + _patch_stop(rec, [{"strategy_id": "STR_A", "ts_code": "002128.SZ"}], []) + try: + out = csvc._stop_buyside_for_exit([], "CMD_T_0002") + assert out == {"strategies": [], "proposals": [], "errors": []}, out + assert not rec.cancelled + finally: + _unpatch_stop(rec) + + +@case("命令覆盖冲突·plan_command 下 LIQUIDATE_ALL 真的撤掉持仓票策略 (接线)") +def _(): + from app.core import command_spec as cs + from app.services import portfolio, strategy_service + orig = (pms_repo.update_command, pms_repo.insert_plans, pms_repo.list_instructions, + pms_repo.list_strategies, pms_repo.list_proposals, pms_repo.insert_ledger, + pms_repo.list_plans, + portfolio.positions_view, strategy_service.set_status, csvc.param_store.get_int) + rec = _Rec() + held = [{"ts_code": "002128.SZ", "total_qty": 300, "avail_qty": 300, "price": 27.6, + "price_ok": True, "base_qty": 0}] + try: + pms_repo.update_command = lambda *a, **k: 1 + pms_repo.insert_plans = lambda rows: len(rows) + pms_repo.list_plans = lambda **kw: [] # 没有在途方案 + pms_repo.list_instructions = lambda **kw: [] # 没有在途买单 + pms_repo.list_strategies = lambda **kw: [ + {"strategy_id": "STR_A", "ts_code": "002128.SZ", "type": "GRID"}] + pms_repo.list_proposals = lambda **kw: [] + pms_repo.insert_ledger = lambda **kw: rec.ledger.append(kw) or 1 + portfolio.positions_view = lambda **kw: { + "held": held, "positions": held, + "params": {"scale": 2_000_000, "weak_neg_days": 5}, + "totals": {"scale": 2_000_000}, "sector_ready": True} + strategy_service.set_status = lambda sid, status, by="user": ( + rec.cancelled.append((sid, status, by)) or {"ok": True, "status": status}) + csvc.param_store.get_int = lambda k, d=0: d + + cmd = {"cmd_type": "LIQUIDATE_ALL", "command_id": "CMD_20260820_0009", + "params": {"window_tdays": 1, "confirm": "YES"}} + r = csvc.plan_command(cmd) + assert r["status"] == cs.ST_EXECUTING, r + assert ("STR_A", "CANCELLED", "command") in rec.cancelled, rec.cancelled + # 命令进度里带停买入侧的回执 + assert r["plan"]["stopped_buyside"]["strategies"] == ["STR_A"], r["plan"] + finally: + (pms_repo.update_command, pms_repo.insert_plans, pms_repo.list_instructions, + pms_repo.list_strategies, pms_repo.list_proposals, pms_repo.insert_ledger, + pms_repo.list_plans, + portfolio.positions_view, strategy_service.set_status, + csvc.param_store.get_int) = orig + + +# ================================================================ runner +def main(): + passed = failed = 0 + for name, fn in RESULTS: + try: + fn() + print(f" PASS {name}") + passed += 1 + except Exception: + print(f" FAIL {name}") + traceback.print_exc() + failed += 1 + print("-" * 60) + if failed: + print(f"FAILED: {failed} / {passed + failed}") + sys.exit(1) + print(f"ALL PASS ({passed} cases)") + + +if __name__ == "__main__": + main()