diff --git a/Makefile b/Makefile index 1897b56..0d2bed0 100644 --- a/Makefile +++ b/Makefile @@ -91,6 +91,17 @@ rebuild-accept: ## 只看判收: 安全垫分布 / 批次账 / 行业集中度 # 一步一停, 每步都能看清楚 —— 别开 beat, 那会让几个调度位同时动, 出了岔子分不清是谁干的。 # 全程 PMS_DISPATCH_MODE=shadow: 指令照常生成、照常过规则闸、照常记账, **只是不写下游**。 # 顺序就是生产里那几个调度位的顺序, 只是改成手动触发。 +# +# **A3 之前必须先下一条「任务命令」。** 参数命令 (SET_SCALE / SET_PORTFOLIO_CAP / +# SET_STOCK_CAP / SET_MAX_NAMES …) 只是设约束, **本来就不产生方案** —— 页面上它们显示 +# 「0 条方案」是对的, 不是坏了。要走通链路得下 B/C 组的任务命令: +# INCREASE_EXPOSURE 升仓 pct% (全局: 垫厚票补到目标 + 候选池新票建仓) +# OPEN_TARGET 建仓某股至 x% (个股: 分批 50/25/25, 含一手合并) +# +# **这里的「择时」是 PMS 内置的实现 B, 不经决策系统。** 设计里择时有两个实现: +# 实现 A = 委托决策系统盘中择时 —— 待办 #9, 等 bionic 侧接口, **还没接** +# 实现 B = PMS 自己算 (分日配额 / 分笔 / VWAP / 回踩 / 不追高 / 14:45 兜底) —— 已实现 +# 所以跑 A 段不需要动决策系统的任何逻辑, 它这一段本来就不参与。 A ?= http://127.0.0.1:38100 J := python3 -m json.tool @@ -100,7 +111,7 @@ t-plan: ## A1 拉候选池 (= 08:40 调度位) t-pre: ## A2 盘前准备: T+1 可卖重置 / 参考位 / 刹车结算 (= 08:50 调度位) @curl -s -X POST '$(A)/api/ops/premarket' | $(J) -t-cmd: ## A3 命令轮询: 新命令 → 方案生成 → 状态机推进 (= 每分钟调度位) +t-cmd: ## A3 命令轮询: 新命令 → 方案生成 (先下任务命令! 参数命令不产生方案) @curl -s -X POST '$(A)/api/ops/plan-pending' | $(J) t-plans: ## A4 看方案 (方案是"买什么、多少股、分几批", 还没成指令) @@ -109,10 +120,10 @@ t-plans: ## A4 看方案 (方案是"买什么、多少股、分几批", 还没 t-mat: ## A5 方案 → 指令 (先记账后动作: 指令先落表) @curl -s -X POST '$(A)/api/ops/materialize' | $(J) -t-dry: ## A6 择时试算 (dry_run: 规则闸判什么、今天该出多少, 一股都不发) +t-dry: ## A6 择时试算 (实现B, 不经决策系统; dry_run 只算不发) @curl -s -X POST '$(A)/api/ops/exec-tick?dry_run=true' | $(J) -t-tick: ## A7 真出手 (shadow 下 = 置 DISPATCHED 但不写下游, 你照着在 QMT 手工执行) +t-tick: ## A7 真出手 (shadow 下 = 置 DISPATCHED 但不写下游, 你照着在 QMT 手工下) @curl -s -X POST '$(A)/api/ops/exec-tick' | $(J) t-ins: ## A8 看指令与子单状态 diff --git a/README.md b/README.md index bcd13c8..526ba0a 100644 --- a/README.md +++ b/README.md @@ -66,7 +66,7 @@ scripts/ test_batch6_units.py ws 通道: 测试向量/签名/公钥/水位/DDL/逐笔入账 65 例 test_batch7_units.py 上游选股计划: 解析/新鲜度/候选筛选/取数守卫 32 例 test_batch8_units.py 榜单变化: 名册指纹/三种语义/尾部闸/落库往返 60 例 - test_batch9_units.py 成本价体检 + 连续不一致按日推进 29 例 + test_batch9_units.py 成本价体检 + 连续不一致按日推进 31 例 test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) 58 例 init_db.py 建表 (应用 ddl_pms_v1.sql, 幂等, 默认演练; 含 DDL 体检) check_db.py 实机连通性与表结构自检 (需真实 .env) @@ -225,12 +225,18 @@ make rebuild-accept # 只看判收：安全垫分布 / 批次账 / 行业集 `shadow` 正好是这一段的天然测试环境，不是妥协：指令照常生成、照常过规则闸、照常记账，**只是不写下游**。所以这一段不用碰任何开关，也不需要 ws 通道就绪。 +两件容易踩的事，先说在前面。 + +**A3 之前必须先下一条「任务命令」。** 参数命令（`SET_SCALE` / `SET_PORTFOLIO_CAP` / `SET_STOCK_CAP` / `SET_MAX_NAMES` 这些）只是设约束，**本来就不产生方案**——页面上它们显示「0 条方案」是对的，不是坏了。要走通链路得下 B/C 组的任务命令：`INCREASE_EXPOSURE`（升仓 pct%，全局：垫厚票补到目标 + 候选池新票建仓）或 `OPEN_TARGET`（建仓某股至 x%，个股：分批 50/25/25 含一手合并）。 + +**这里的「择时」是 PMS 内置的实现 B，不经决策系统。** 设计里择时有两个实现：实现 A 是委托决策系统盘中择时（待办 #9，等 bionic 侧接口，**还没接**）；实现 B 是 PMS 自己算（分日配额、分笔、VWAP/回踩/不追高、14:45 兜底），已实现。所以跑 A 段不需要动决策系统的任何逻辑——它这一段本来就不参与。 + **别开 beat**——那会让几个调度位同时动，出了岔子分不清是谁干的。`Makefile` 里有一组 `t-*` 目标，就是把那几个调度位改成手动逐跳触发，顺序与生产一致： ```bash make t-plan # A1 拉候选池 (= 08:40 调度位) make t-pre # A2 盘前准备 (= 08:50 调度位) -# —— 这里在页面「命令台」下一条建仓或升仓命令 —— +# —— 页面「命令台」下一条**任务命令**: INCREASE_EXPOSURE 或 OPEN_TARGET —— make t-cmd # A3 命令轮询: 新命令 → 方案生成 make t-plans # A4 看方案: 买什么、多少股、分几批 (还没成指令) make t-mat # A5 方案 → 指令 (先记账后动作) @@ -263,7 +269,7 @@ make t-gate # 随时: 规则闸/研判闸拒了什么、为什么 ## 已实现 / 待开发 -**已实现**：建表 DDL 与建表脚本；配置与运行参数中心；仓位规划器与安全垫账；命令系统（27 类命令全目录 + 双状态机 + 冲突识别）；方案生成器（降仓凑额四档、升仓、建仓分批、清仓/减至、行业清仓与限额、暂停买入撤单）；账本回放与对账引擎（成交认领、外部成交并入 BASE 告警、以下游为准修正、除权检测、T+1 可用量、连续不一致升级）；规则闸终检；择时执行器实现 B（分日配额、分笔、VWAP/回踩/不追高、14:45 兜底、停牌一字板顺延、窗口耗尽收口、挂单有效期）；动作引擎四类自主动作 + 研判闸客户端 + 提议分流；决策系统信号消化（两条流独立消费组订阅、置信度分档转清仓指令或提议）；管理页面四块 + 运维/日报抽屉；调度器九个调度位；**上游选股计划接口接入**（`/plan` 取候选池、交易日龄硬校验、`theme` 灌行业映射表、页面预览抽屉与不可用横幅）；**榜单变化提示**（名册快照 + 新进/掉榜/档位升降/覆盖翻转/名次跳变，持仓票单列，榜尾截断噪音闸）；**ws 直连通道的连接层**（常驻进程 + 出口队列 + 签名 + seq 水位与累积确认，见下）；**单测 333 例**。 +**已实现**：建表 DDL 与建表脚本；配置与运行参数中心；仓位规划器与安全垫账；命令系统（27 类命令全目录 + 双状态机 + 冲突识别）；方案生成器（降仓凑额四档、升仓、建仓分批、清仓/减至、行业清仓与限额、暂停买入撤单）；账本回放与对账引擎（成交认领、外部成交并入 BASE 告警、以下游为准修正、除权检测、T+1 可用量、连续不一致升级）；规则闸终检；择时执行器实现 B（分日配额、分笔、VWAP/回踩/不追高、14:45 兜底、停牌一字板顺延、窗口耗尽收口、挂单有效期）；动作引擎四类自主动作 + 研判闸客户端 + 提议分流；决策系统信号消化（两条流独立消费组订阅、置信度分档转清仓指令或提议）；管理页面四块 + 运维/日报抽屉；调度器九个调度位；**上游选股计划接口接入**（`/plan` 取候选池、交易日龄硬校验、`theme` 灌行业映射表、页面预览抽屉与不可用横幅）；**榜单变化提示**（名册快照 + 新进/掉榜/档位升降/覆盖翻转/名次跳变，持仓票单列，榜尾截断噪音闸）；**ws 直连通道的连接层**（常驻进程 + 出口队列 + 签名 + seq 水位与累积确认，见下）；**单测 335 例**。 ### 下一步（按可动工顺序） diff --git a/app/core/rebuild_check.py b/app/core/rebuild_check.py index 66900a0..d551cfd 100644 --- a/app/core/rebuild_check.py +++ b/app/core/rebuild_check.py @@ -88,14 +88,23 @@ def check_row(row: dict, price=None) -> dict: def check_costs(rows, price_map=None, *, eq_ratio_block: float = 0.8) -> dict: """整批持仓的成本价体检。 - `blocking` 为真表示**不该照这份数据建账**。两条阻断判据: + `blocking` 为真表示**不该照这份数据建账**。三条阻断判据, 共同点是"**系统性**不可用": - 1. **任何一只 MISSING / ABSURD / AVAIL_BAD。** 这些是数据本身不可用, 不是判断问题。 - 一只错就够呛 —— 账本是一本账, 里面混一只错成本的票, 之后每次看这本账都得先想起 - "那只是估的"。宁可等对端填对。 - 2. **EQ_PRICE 占比过高** (默认 ≥80%)。单只票成本≈现价完全可能 (当日买入); 整个组合 - 都这样, 那就是有人拿现价当成本填进去了 —— 而这正是 07-29 踩的那个坑的模样, - 区别只在于上次是我方拿现价兜底, 这次是对端拿现价填。结果一模一样: 安全垫全 0。 + 1. **任何一只 ABSURD / AVAIL_BAD。** 这不是"缺", 是"错"—— 成本与现价差两个数量级 + 多半是单位(分/元)或字段串位, 可用量大于总量是 T+1 口径本身写错了。一只出现就说明 + 这一批数据的生产方式有问题, 不该只怀疑那一只。 + 2. **全部 MISSING。** 对端那一列整个没填 —— 这时建账等于全组拿现价当成本, 就是 + 2026-07-29 那个坑。 + 3. **EQ_PRICE 占比过高** (默认 ≥80%)。单只成本≈现价完全可能 (当日买入); 整组都这样 + 就是有人拿现价填的 —— 与上一条同样的结果 (安全垫全 0), 只是这次是对端填的。 + + **少数几只 MISSING 不阻断。** `recon.build_recon_fixes` 对这种情况本来就有逐只回退现价 + 的路径, 且在 `price_source` / `note` 里注明是估的 —— 那是设计里定好的行为, 不该被这道 + 闸推翻。闸的职责是拦下"一本从头就错的账", 不是替既有设计做二次判断。 + + 但要知道**估出来的成本是改不回来的**: 数量对得上就没有对账差异, 后续每天的对账都不会 + 再碰它。所以这几只会单独列进 `estimated`, 由调用方大声报出来 —— 它是一条已知的、 + 带标签的坏账, 不是一条被忽略的。 NO_PRICE 不阻断: 现价取不到是本端行情的事, 成本值本身没毛病, 建账照常, 只是没法核。 """ @@ -107,17 +116,21 @@ def check_costs(rows, price_map=None, *, eq_ratio_block: float = 0.8) -> dict: by.setdefault(c["verdict"], []).append(c) n = len(checked) - hard = by.get(MISSING, []) + by.get(ABSURD, []) + by.get(AVAIL_BAD, []) + corrupt = by.get(ABSURD, []) + by.get(AVAIL_BAD, []) + missing = by.get(MISSING, []) n_eq = len(by.get(EQ_PRICE, [])) # 占比只在能判的那部分里算 —— 取不到现价的票不该稀释这个比例 judgeable = n - len(by.get(NO_PRICE, [])) eq_ratio = (n_eq / judgeable) if judgeable else 0.0 reasons = [] - if hard: - reasons.append(f"{len(hard)} 只成本价不可用 (" - + ", ".join(f"{c['ts_code']}:{c['verdict']}" for c in hard[:6]) - + (" …" if len(hard) > 6 else "") + ")") + if corrupt: + reasons.append(f"{len(corrupt)} 只成本价/可用量是**错的**不是缺的 (" + + ", ".join(f"{c['ts_code']}:{c['verdict']}" for c in corrupt[:6]) + + (" …" if len(corrupt) > 6 else "") + ")") + all_missing = bool(n and len(missing) == n) + if all_missing: + reasons.append(f"{n} 只**全部**没有成本价 —— 建账等于全组拿现价当成本, 安全垫会齐刷刷是 0") eq_block = bool(judgeable >= 2 and eq_ratio >= eq_ratio_block) if eq_block: reasons.append(f"{n_eq}/{judgeable} 只的成本≈现价 ({eq_ratio:.0%}) —— " @@ -125,8 +138,9 @@ def check_costs(rows, price_map=None, *, eq_ratio_block: float = 0.8) -> dict: return {"n": n, "counts": {k: len(v) for k, v in sorted(by.items())}, "rows": checked, "eq_ratio": round(eq_ratio, 4), - "blocking": bool(hard or eq_block), "reasons": reasons, - "hint": _hint(n, by, reasons)} + "estimated": [c["ts_code"] for c in missing] if not all_missing else [], + "blocking": bool(corrupt or all_missing or eq_block), "reasons": reasons, + "hint": _hint(n, by, reasons, missing if not all_missing else [])} def coverage(rows_checked) -> dict: @@ -153,7 +167,7 @@ def coverage(rows_checked) -> dict: "缺这几种情形, 对应的纪律分支这轮验不到: " + " / ".join(missing))} -def _hint(n, by, reasons) -> str: +def _hint(n, by, reasons, estimated=()) -> str: if not n: return "下游一只持仓都没有 —— 无账可建 (等对端装持仓)" if reasons: @@ -161,8 +175,14 @@ def _hint(n, by, reasons) -> str: + "。开仓价一错, 摊薄成本与安全垫跟着错, 而错账与真账在页面上长得一模一样 —— " "请对端把 trading_position 的 cost_price / available_quantity 填成真实值再来") extra = "" + if estimated: + codes = ", ".join(str(c["ts_code"]) for c in estimated[:6]) + extra += (f"。**{len(estimated)} 只会拿现价当成本建账** ({codes}" + + (" …" if len(estimated) > 6 else "") + ") —— 它们的安全垫从 0 起算, " + "而且**改不回来**: 数量对得上就没有对账差异, 后续对账不会再碰它。" + "能等的话, 请对端把这几只的 cost_price 填上再建") if by.get(EQ_PRICE): - extra = f" (其中 {len(by[EQ_PRICE])} 只成本≈现价, 若确为当日买入则正常)" + extra += f" ({len(by[EQ_PRICE])} 只成本≈现价, 若确为当日买入则正常)" if by.get(NO_PRICE): extra += f" ({len(by[NO_PRICE])} 只取不到现价, 成本没核但可用)" return f"{n} 只持仓的成本价可用, 可以建账{extra}" diff --git a/app/services/ledger_service.py b/app/services/ledger_service.py index b1d55a0..b5209f1 100644 --- a/app/services/ledger_service.py +++ b/app/services/ledger_service.py @@ -681,6 +681,13 @@ def reconcile(*, apply_fix: bool = True, force: bool = False) -> dict: if chk["blocking"]: logger.warning("[对账] 成本价闸本应拦下 (%s), 但 force=true 放行 —— " "安全垫将从 0 起算, 补仓/加仓/保垫减仓这一轮判不准", chk["reasons"]) + if chk.get("estimated"): + # 少数几只没成本价 —— 走 build_recon_fixes 的逐只兜底 (设计定好的行为), 不拦。 + # 但必须吼一声: 估出来的成本**改不回来** (数量对得上就没有对账差异, 后续对账 + # 不会再碰它), 它是一条已知的坏账, 得让人知道是哪几只。 + logger.error("[对账] %s 只没有下游成本价, 将拿现价建账且**此后不会被对账修正**: %s。" + "能等的话请对端补上 cost_price 再建", + len(chk["estimated"]), chk["estimated"]) if not out["coverage"]["enough"]: logger.warning("[对账] 建账数据的情形覆盖不全: %s", out["coverage"]["hint"]) diff --git a/scripts/run_tests.py b/scripts/run_tests.py index 0976abf..dd89984 100644 --- a/scripts/run_tests.py +++ b/scripts/run_tests.py @@ -13,9 +13,9 @@ test_batch6_units.py ws 通道: 测试向量/签名/公钥/水位/DDL/逐笔入账 (65 例) test_batch7_units.py 上游选股计划: 解析/新鲜度/候选筛选/取数守卫 (32 例) test_batch8_units.py 榜单变化: 名册指纹/三种语义/尾部闸/落库往返 (60 例) - test_batch9_units.py 成本价体检 + 连续不一致按日推进 (29 例) + test_batch9_units.py 成本价体检 + 连续不一致按日推进 (31 例) test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) (58 例) - 共 333 例 + 共 335 例 任一子集失败即整体失败 (退出码 1)。 """ import os diff --git a/scripts/test_batch9_units.py b/scripts/test_batch9_units.py index ebc5b93..717bcff 100644 --- a/scripts/test_batch9_units.py +++ b/scripts/test_batch9_units.py @@ -102,12 +102,33 @@ def t_b1(): assert "可以建账" in out["hint"] -@case("[B2] 只要有一只成本不可用就阻断 —— 一本账里混一只错的也不行") +@case("[B2] 少数几只没成本价不阻断, 但要单独列出来") def t_b2(): + # 既有设计对"某一只没成本价"有逐只回退现价的路径 (build_recon_fixes, 带 price_source + # 留痕)。闸不该推翻它 —— 闸管的是"一本从头就错的账", 不是替既有设计做二次判断。 rows = [_p("600000.SH", cost=20.0), _p("600519.SH", cost=0)] out = rb.check_costs(rows, {"600000.SH": 10.0, "600519.SH": 1800.0}) - assert out["blocking"] is True - assert "600519.SH:MISSING" in out["reasons"][0] + assert out["blocking"] is False + assert out["estimated"] == ["600519.SH"] + assert "改不回来" in out["hint"], "估出来的成本后续对账不会再碰, 这句必须说出来" + + +@case("[B2b] 全部没成本价 → 阻断 (对端那一列整个没填)") +def t_b2b(): + rows = [_p("600000.SH", cost=0), _p("600519.SH", cost=None)] + out = rb.check_costs(rows, {"600000.SH": 10.0, "600519.SH": 1800.0}) + assert out["blocking"] is True and out["estimated"] == [] + assert "全部" in out["reasons"][0] + + +@case("[B2c] 数据是**错的**而不是缺的 → 一只就阻断") +def t_b2c(): + # ABSURD/AVAIL_BAD 说明这批数据的生产方式有问题, 不该只怀疑那一只 + for bad in ({"cost": 1000.0}, {"avail": 9999}): + rows = [_p("600000.SH", cost=20.0), _p("600519.SH", **bad)] + out = rb.check_costs(rows, {"600000.SH": 10.0, "600519.SH": 10.0}) + assert out["blocking"] is True, bad + assert "错的" in out["reasons"][0] @case("[B3] 整组成本≈现价 → 阻断 (这就是 07-29 那个坑的模样)") @@ -163,7 +184,7 @@ def t_b8(): @case("[B9] 阻断时的 hint 要给出可执行的下一步") def t_b9(): - out = rb.check_costs([_p("600000.SH", cost=0)], {"600000.SH": 10.0}) + out = rb.check_costs([_p("600000.SH", cost=0)], {"600000.SH": 10.0}) # 唯一一只且缺失 assert "cost_price" in out["hint"], "得说清要对端改哪一列, 不是只说'数据有问题'" assert "长得一模一样" in out["hint"], "得说清为什么不能将就着建"