diff --git a/Makefile b/Makefile index dcd12d8..2578b93 100644 --- a/Makefile +++ b/Makefile @@ -22,7 +22,7 @@ DC := docker compose $(PROFILES) RUN := docker compose run --rm --no-deps pms-web .PHONY: help deploy deploy-local build up down ps logs test initdb check health \ - probe changes industry ws-status reset-ledger shell + probe changes industry ws-status rebuild rebuild-accept reset-ledger shell help: ## 列出所有目标 @grep -hE '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) \ @@ -78,6 +78,13 @@ industry: ## 行业源实探 (ready=false 就是硬拦截失效, 别当它绿着 ws-status: ## ws 通道状态 (连接态 / seq 水位 / 出口队列) $(RUN) python scripts/ws_smoke.py status +# ---------------------------------------------------------------- 账本重建 (待办 #4) +rebuild: ## 账本重建预检 (只读)。加 GO=1 才真的建账, FORCE=1 连预检不过也建 + $(RUN) python scripts/rebuild_ledger.py $(if $(GO),--yes,) $(if $(FORCE),--force,) + +rebuild-accept: ## 只看判收: 安全垫分布 / 批次账 / 行业集中度 + $(RUN) python scripts/rebuild_ledger.py --accept + # ---------------------------------------------------------------- 危险操作 (要确认) reset-ledger: ## 清空账本重来 (影子期专用; 必须 CONFIRM=1) @if [ "$(CONFIRM)" != "1" ]; then \ diff --git a/README.md b/README.md index 50750a0..a7ab638 100644 --- a/README.md +++ b/README.md @@ -33,6 +33,7 @@ app/ action_engine.py 动作引擎: FILL 回踩补足 / ADD 盈利加仓 / DCA 补仓 / TRIM 保垫减仓 signal_rules.py 决策系统两条信号流的解析与消化口径 (含置信度尺度归一) plan_diff.py 上游榜单的版本比对: 名册指纹 / 新进掉榜 / 档位升降 / 榜尾噪音闸 + rebuild_check.py 接管既有持仓前的成本价体检 (阻断判据 + §5.2 情形覆盖) tradedays.py 交易日历: 调度守卫与执行窗口计算 ws_codec.py QMT 协议编解码: 规范化串 / Ed25519 签名验签 / 信封 / seq 水位推进 db/session.py 三库连接 + **严格单表访问守卫** (JOIN/逗号连表/跨表子查询一律拒绝) @@ -65,6 +66,7 @@ scripts/ test_batch6_units.py ws 通道: 测试向量/签名/公钥/水位/DDL/逐笔入账 65 例 test_batch7_units.py 上游选股计划: 解析/新鲜度/候选筛选/取数守卫 32 例 test_batch8_units.py 榜单变化: 名册指纹/三种语义/尾部闸/落库往返 60 例 + test_batch9_units.py 接管持仓前的成本价体检: 阻断判据/情形覆盖 23 例 test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) 58 例 init_db.py 建表 (应用 ddl_pms_v1.sql, 幂等, 默认演练; 含 DDL 体检) check_db.py 实机连通性与表结构自检 (需真实 .env) @@ -72,6 +74,7 @@ scripts/ reset_ledger.py 清空账本并把回放游标对齐到当前 (影子运行期重来一次; 不碰下游表) probe_plan_api.py 上游计划实机探活: 通不通 / 字段口径 / 候选筛选结果 / 有没有价 / 榜单变化 (只读; --snapshot 才落库, 那是它唯一的写操作) + rebuild_ledger.py 账本重建: 预检 → 执行 → 判收 (默认只预检; --yes 才改账) ws_smoke.py ws 联调工具: status/watch/place/cancel/inbox (绕开 dispatch_mode) ``` @@ -196,6 +199,22 @@ QMT ──trade/order_update──▶ pms-ws ──落 pms_qmt_inbox──▶ 这条不是细节：拿现价当成本，摊薄成本就等于当天价、**安全垫齐刷刷是 0**——而安全垫是盈利加仓（≥3%）、保垫减仓（峰值≥6%）、补仓评估档（−8%/−15%）共同的判断依据。一只真实成本 20、现价 10 的票（实亏 50%）会被记成「不赚不亏」，该评估的补仓不评估；页面和日报上的浮动盈亏也全是 0。2026-07-29 首次接管 22 只持仓时踩到，已修并有单测守着。 +**但「优先取成本价」还不够——兜底是静默的。** 下游那一列填 0、留空、或干脆等于现价时，对账照样建账，只在 note 里留一句「现价兜底」。结果是账本建起来了、页面一切正常、每个数都长得像真的，只有安全垫齐刷刷是 0，而没人会盯着一个「看起来就该是 0」的字段。所以 2026-07-31 加了一道**首次建账的成本价硬闸**（`app/core/rebuild_check.py`）： + +- **只在账本为空时生效**——那一刻是不可逆的，这批 RECON 批次的开仓价会定死每只票的摊薄成本。日常漂移不走这道闸，否则每天拦一次对账才是真的坏事。 +- 拦两类：**任何一只成本价缺失/为 0/与现价差两个数量级/可用量大于总量**；以及**整组成本≈现价**（默认 ≥80%）——单只可能是当日买入，整组都这样就是有人拿现价填的。 +- 拦下来只是不建账，不改任何东西。确认这份数据就是对的，可以 `--force` 放行，那等于声明你接受安全垫从 0 起算。 + +配套一条命令，默认只预检不改账： + +```bash +make rebuild # 预检（只读）：事实源 → 成本价体检 → 情形覆盖 → 开关现状 +make rebuild GO=1 # 预检过了才执行，执行完自动判收 +make rebuild-accept # 只看判收：安全垫分布 / 批次账 / 行业集中度 +``` + +判收里最硬的一条是**安全垫分布**：如果重建后每一只票的安全垫都是 0，那就是踩了这个坑——页面上每个数都合理，只有这一处露馅。行业集中度那一项顺带回答 `PMS_SECTOR_MAX_RATIO=40%` 在三级粒度下偏不偏松（分母与 `sizer.check_caps` 一致，是**组合持仓市值**不是总规模，两边用不同分母会得出两个都自称「行业集中度」的数）。 + ## 自主提议的分流（设计 §6 / §7） 动作引擎每分钟扫一遍持仓，产出 FILL / ADD / DCA / TRIM 四类候选，然后依次过三道： @@ -216,7 +235,7 @@ QMT ──trade/order_update──▶ pms-ws ──落 pms_qmt_inbox──▶ ## 已实现 / 待开发 -**已实现**：建表 DDL 与建表脚本；配置与运行参数中心；仓位规划器与安全垫账；命令系统（27 类命令全目录 + 双状态机 + 冲突识别）；方案生成器（降仓凑额四档、升仓、建仓分批、清仓/减至、行业清仓与限额、暂停买入撤单）；账本回放与对账引擎（成交认领、外部成交并入 BASE 告警、以下游为准修正、除权检测、T+1 可用量、连续不一致升级）；规则闸终检；择时执行器实现 B（分日配额、分笔、VWAP/回踩/不追高、14:45 兜底、停牌一字板顺延、窗口耗尽收口、挂单有效期）；动作引擎四类自主动作 + 研判闸客户端 + 提议分流；决策系统信号消化（两条流独立消费组订阅、置信度分档转清仓指令或提议）；管理页面四块 + 运维/日报抽屉；调度器九个调度位；**上游选股计划接口接入**（`/plan` 取候选池、交易日龄硬校验、`theme` 灌行业映射表、页面预览抽屉与不可用横幅）；**榜单变化提示**（名册快照 + 新进/掉榜/档位升降/覆盖翻转/名次跳变，持仓票单列，榜尾截断噪音闸）；**ws 直连通道的连接层**（常驻进程 + 出口队列 + 签名 + seq 水位与累积确认，见下）；**单测 304 例**。 +**已实现**：建表 DDL 与建表脚本；配置与运行参数中心；仓位规划器与安全垫账；命令系统（27 类命令全目录 + 双状态机 + 冲突识别）；方案生成器（降仓凑额四档、升仓、建仓分批、清仓/减至、行业清仓与限额、暂停买入撤单）；账本回放与对账引擎（成交认领、外部成交并入 BASE 告警、以下游为准修正、除权检测、T+1 可用量、连续不一致升级）；规则闸终检；择时执行器实现 B（分日配额、分笔、VWAP/回踩/不追高、14:45 兜底、停牌一字板顺延、窗口耗尽收口、挂单有效期）；动作引擎四类自主动作 + 研判闸客户端 + 提议分流；决策系统信号消化（两条流独立消费组订阅、置信度分档转清仓指令或提议）；管理页面四块 + 运维/日报抽屉；调度器九个调度位；**上游选股计划接口接入**（`/plan` 取候选池、交易日龄硬校验、`theme` 灌行业映射表、页面预览抽屉与不可用横幅）；**榜单变化提示**（名册快照 + 新进/掉榜/档位升降/覆盖翻转/名次跳变，持仓票单列，榜尾截断噪音闸）；**ws 直连通道的连接层**（常驻进程 + 出口队列 + 签名 + seq 水位与累积确认，见下）；**单测 327 例**。 ### 下一步（按可动工顺序） @@ -225,7 +244,7 @@ QMT ──trade/order_update──▶ pms-ws ──落 pms_qmt_inbox──▶ | 1 | ~~ws 通道的账本侧改造（清单 4~6）~~ | ✅ 2026-07-29 | | 2 | ~~上游选股计划接入（`/plan`）~~ | ✅ 2026-07-31：候选池独占来源、交易日龄硬校验、盘前昨收兜底、ST 剔除、PMS 侧主题限额、页面抽屉与探活脚本。口径与实测记录见 `UPSTREAM_PLAN_API.md` | | 3 | ~~行业硬拦截开闸~~ | ✅ 2026-07-31：`PMS_SECTOR_SOURCE=gp_hybk`（199 库，三级 884*，每票只认 bk_code 最小的主行业），实测 `ready=true`、代码写法 `prefix`。**这是 PMS 第一次有可用的行业源。**待账本重建后回看 `PMS_SECTOR_MAX_RATIO=40%` 对三级粒度是否偏松（参考项目三级用 20%）。详见 `UPSTREAM_PLAN_API.md` §8 | -| 4 | **账本重建**：清账后账本为空，需对端持仓就绪后 `POST /api/ops/reconcile?apply_fix=true` 以下游为准补回 | 阻塞：等 QMT 侧按真实成本价重建模拟持仓（`QMT_SIDE_S3_CLOSEOUT.md` §5） | +| 4 | **账本重建**：预检/执行/判收已就绪（`make rebuild`），**首次建账加了成本价硬闸** | 阻塞：等 QMT 侧按真实成本价重建模拟持仓（`QMT_SIDE_S3_CLOSEOUT.md` §5）。对端一装好就 `make rebuild` 看预检，过了再 `make rebuild GO=1` | | 5 | **ws 通道联调收尾**（协议 §9 S3） | 阻塞：`trade_no` 格式不合 §5.5（`QMT_SIDE_S3_CLOSEOUT.md` §1，这条挡住切 ws） | | 6 | 上游计划的剩余待确认口径 | 等上游：`UPSTREAM_PLAN_API.md` §4 剩 5 条 + §7.3 新增两条（同一 `date` 计划不幂等、档位规模与文档不符） | | 7 | ~~榜单变化做成页面提示（新进传导链 / 掉榜）~~ | ✅ 2026-07-31：上游 `changes` 恒为 `null`，所以**改成 PMS 自己算**——每次拉计划落一份名册快照（第 15 张表），差异由纯逻辑比。顺带补上「同一 `date` 多版本、下游无从分辨手上是哪一版」那个洞。详见 `UPSTREAM_PLAN_API.md` §9 | diff --git a/UPSTREAM_PLAN_API.md b/UPSTREAM_PLAN_API.md index 9f447e0..5223696 100644 --- a/UPSTREAM_PLAN_API.md +++ b/UPSTREAM_PLAN_API.md @@ -648,7 +648,7 @@ PMS_PLAN_DIFF_TAIL_GUARD 0.5 榜尾进出的噪音闸, 见 9.4 ```bash # 【服务器 factorevaluation · ~/project/tradingSystem】 -make test # 应输出 ALL SUITES PASS (304 例) +make test # 应输出 ALL SUITES PASS (327 例) make initdb # 幂等, 建第 15 张表 make check # [2] 段应能看到 pms_plan_snapshot make probe SNAP=1 # 落第一份快照; [7] 段此时应报"库里还没有快照" diff --git a/app/core/rebuild_check.py b/app/core/rebuild_check.py new file mode 100644 index 0000000..66900a0 --- /dev/null +++ b/app/core/rebuild_check.py @@ -0,0 +1,184 @@ +# -*- coding: utf-8 -*- +""" +接管既有持仓前的成本价体检 (纯逻辑, 零外部依赖) +================================================ +账本清空之后要按「以下游为准」把真实持仓认领回来 (README 待办 #4)。这一步只发生一次, +但它定死了每一只票的**开仓价**, 而开仓价一错, 后面所有纪律都错在一个看不出来的地方。 + +链条是这样的: + + 下游 cost_price → RECON 批次开仓价 → 摊薄成本 → 安全垫 + ↓ + 盈利加仓(≥3%) · 保垫减仓(峰值≥6%) · 补仓评估档(−8%/−15%) + +`recon.build_recon_fixes` 已经做了对的事: 开仓价优先取下游成本价, 现价只兜底。**但兜底是 +静默的** —— 下游那一列填 0、填空、或干脆等于现价时, 它照样建账, 只在 note 里留一句"现价 +兜底"。结果是摊薄成本等于当天价、**安全垫齐刷刷是 0**: 一只真实成本 20、现价 10 的票 +(实亏 50%) 被记成"不赚不亏", 该评估的补仓不评估, 页面和日报的浮动盈亏也全是 0。 +2026-07-29 首次接管 22 只持仓时踩过一次。 + +**漏账看得见, 错账看不见。** 拒绝建账的代价是等对端把数据填对 (一句话的事); 建错账的代价 +是一本从头到尾都合理、但每个数都错的账 —— 而且它跟真账长得一模一样。所以这里的默认行为 +是**拿不准就不建**, 与本仓库「拿不到 ≠ 通过」是同一条纪律。 + +本模块只做判断、不碰库, 也不决定要不要放行 —— 判据交给调用方 (`ledger_service.reconcile` +的建账闸 / `scripts/rebuild_ledger.py` 的预检段)。 +""" +from __future__ import annotations + +# 单只票的体检结论 +OK = "OK" # 成本价可用 +MISSING = "MISSING" # 缺失 / <=0 —— 会走现价兜底, 安全垫必为 0 +EQ_PRICE = "EQ_PRICE" # 与现价几乎相等 —— 安全垫≈0, 单只可能是真的, 全组合则不可能 +ABSURD = "ABSURD" # 与现价差了两个数量级 —— 多半是单位或字段错了 +AVAIL_BAD = "AVAIL_BAD" # 可用量 > 总量, 或为负 —— T+1 口径错 +NO_PRICE = "NO_PRICE" # 本端取不到现价, 判不了 EQ_PRICE / ABSURD (成本本身仍可用) + +# 判为"与现价几乎相等"的相对阈值。0.5% 已经远小于一个交易日的正常波动 —— +# 真实成本恰好落进这个带里是可能的, 但整个组合都落进来就不是巧合了。 +EQ_EPS = 0.005 +# 与现价的比值超出这个区间即判 ABSURD。取 100 倍是因为 A 股不存在这种成本差, +# 出现只可能是单位错 (分/元)、字段串位或脏数据。 +ABSURD_LO, ABSURD_HI = 0.01, 100.0 + + +def check_row(row: dict, price=None) -> dict: + """一只票的成本价体检。row: {ts_code, qty, avail_qty, cost, ...}; price: 本端现价。""" + code = row.get("ts_code") + qty = _int(row.get("qty")) + avail = row.get("avail_qty") + cost = _float(row.get("cost")) + px = _float(price if price is not None else row.get("price")) + + out = {"ts_code": code, "qty": qty, "avail_qty": avail, "cost": cost, "price": px, + "verdict": OK, "why": "", "cushion_pct": None} + + if avail is not None: + a = _int(avail) + if a < 0 or a > qty: + out.update({"verdict": AVAIL_BAD, + "why": f"可用量 {a} 不合理 (总量 {qty}) —— T+1 可卖口径会跟着错"}) + return out + + if cost is None or cost <= 0: + out.update({"verdict": MISSING, + "why": "下游没给成本价 —— 建账会退回现价, 摊薄成本=当天价、安全垫恒为 0"}) + return out + + if px is None or px <= 0: + out.update({"verdict": NO_PRICE, + "why": "本端取不到现价, 无法核对成本是否合理 (成本值本身可用)"}) + return out + + ratio = cost / px + if ratio <= ABSURD_LO or ratio >= ABSURD_HI: # 边界含在内: 正好 100 倍也不是真成本 + out.update({"verdict": ABSURD, + "why": f"成本 {cost:.4g} 与现价 {px:.4g} 差了 {ratio:.3g} 倍 —— " + f"多半是单位(分/元)或字段串位, 不是真成本"}) + return out + + out["cushion_pct"] = (px - cost) / cost + if abs(px - cost) / px <= EQ_EPS: + out.update({"verdict": EQ_PRICE, + "why": f"成本 {cost:.4g} ≈ 现价 {px:.4g} —— 安全垫≈0。" + f"单只可能是当日买入, 整组都这样则是拿现价充的成本"}) + return out + + +def check_costs(rows, price_map=None, *, eq_ratio_block: float = 0.8) -> dict: + """整批持仓的成本价体检。 + + `blocking` 为真表示**不该照这份数据建账**。两条阻断判据: + + 1. **任何一只 MISSING / ABSURD / AVAIL_BAD。** 这些是数据本身不可用, 不是判断问题。 + 一只错就够呛 —— 账本是一本账, 里面混一只错成本的票, 之后每次看这本账都得先想起 + "那只是估的"。宁可等对端填对。 + 2. **EQ_PRICE 占比过高** (默认 ≥80%)。单只票成本≈现价完全可能 (当日买入); 整个组合 + 都这样, 那就是有人拿现价当成本填进去了 —— 而这正是 07-29 踩的那个坑的模样, + 区别只在于上次是我方拿现价兜底, 这次是对端拿现价填。结果一模一样: 安全垫全 0。 + + NO_PRICE 不阻断: 现价取不到是本端行情的事, 成本值本身没毛病, 建账照常, 只是没法核。 + """ + rows = [r for r in (rows or []) if _int(r.get("qty")) > 0] + price_map = price_map or {} + checked = [check_row(r, price_map.get(r.get("ts_code"))) for r in rows] + by = {} + for c in checked: + by.setdefault(c["verdict"], []).append(c) + + n = len(checked) + hard = by.get(MISSING, []) + by.get(ABSURD, []) + by.get(AVAIL_BAD, []) + 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 "") + ")") + eq_block = bool(judgeable >= 2 and eq_ratio >= eq_ratio_block) + if eq_block: + reasons.append(f"{n_eq}/{judgeable} 只的成本≈现价 ({eq_ratio:.0%}) —— " + f"整组安全垫会是 0, 补仓/加仓/保垫减仓全都判不出来") + + 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)} + + +def coverage(rows_checked) -> dict: + """这批持仓能不能把纪律验到 (QMT_SIDE_S3_CLOSEOUT §5.2 的四种情形)。 + + **不阻断建账** —— 数据是真的就该建。只是提醒: 全是"不赚不亏"的话, 盈利加仓、保垫减仓、 + 补仓评估档这些分支一条都跑不到, 重建完等于没验。 + """ + ok = [c for c in (rows_checked or []) if c.get("cushion_pct") is not None] + gain = [c for c in ok if c["cushion_pct"] >= 0.05] + loss = [c for c in ok if c["cushion_pct"] <= -0.10] + intraday = [c for c in (rows_checked or []) + if c.get("avail_qty") is not None and _int(c["avail_qty"]) < _int(c["qty"])] + want = {"明显浮盈(≥+5%) 验盈利加仓与保垫减仓": len(gain), + "明显浮亏(≤−10%) 验补仓评估档": len(loss), + "当日买入(可用<总量) 验 T+1 可卖口径": len(intraday), + "持仓只数(≥3 只才验得到组合上限与行业集中度)": len(rows_checked or [])} + missing = [k for k, v in want.items() if v == 0] + if len(rows_checked or []) < 3: + missing.append("持仓不足 3 只") + return {"counts": want, "missing": missing, + "enough": not missing, + "hint": ("这批持仓覆盖了四种情形, 重建后各条纪律都验得到" if not missing else + "缺这几种情形, 对应的纪律分支这轮验不到: " + " / ".join(missing))} + + +def _hint(n, by, reasons) -> str: + if not n: + return "下游一只持仓都没有 —— 无账可建 (等对端装持仓)" + if reasons: + return ("**不建议照这份数据建账**: " + "; ".join(reasons) + + "。开仓价一错, 摊薄成本与安全垫跟着错, 而错账与真账在页面上长得一模一样 —— " + "请对端把 trading_position 的 cost_price / available_quantity 填成真实值再来") + extra = "" + if by.get(EQ_PRICE): + extra = f" (其中 {len(by[EQ_PRICE])} 只成本≈现价, 若确为当日买入则正常)" + if by.get(NO_PRICE): + extra += f" ({len(by[NO_PRICE])} 只取不到现价, 成本没核但可用)" + return f"{n} 只持仓的成本价可用, 可以建账{extra}" + + +def _int(v): + try: + return int(float(v or 0)) + except (TypeError, ValueError): + return 0 + + +def _float(v): + try: + if v is None or (isinstance(v, str) and not v.strip()): + return None + return float(v) + except (TypeError, ValueError): + return None diff --git a/app/services/ledger_service.py b/app/services/ledger_service.py index e945e13..20afe56 100644 --- a/app/services/ledger_service.py +++ b/app/services/ledger_service.py @@ -20,6 +20,7 @@ from datetime import datetime, timedelta from app.core import command_spec as cs from app.core import cushion as cu +from app.core import rebuild_check as rbc from app.core import recon as rc from app.core import tradedays as td from app.core import ws_codec as wsc @@ -639,6 +640,39 @@ def reconcile(*, apply_fix: bool = True, force: bool = False) -> dict: codes = [d["ts_code"] for d in diffs] prices = market.get_prices(codes) + + # 首次建账的成本价闸 (2026-07-31 加; QMT_SIDE_S3_CLOSEOUT.md §5) + # --------------------------------------------------------------- + # 只在**账本为空**时生效 —— 那一刻是不可逆的: 这一批 RECON 批次的开仓价会定死每只票 + # 的摊薄成本, 而摊薄成本是安全垫的分母, 安全垫又是补仓/加仓/保垫减仓的共同判据。 + # 日常漂移不走这道闸: 那时已有批次各自带着自己的成本, 补几百股用什么价影响有限, + # 而每天拦一次对账才是真的坏事。 + # + # 为什么非要有这道闸: `rc.build_recon_fixes` 在下游给不出成本价时**静默退回现价**, + # 只在 note 里留一句"现价兜底"。于是账本建起来了、页面一切正常、每个数都长得像真的, + # 只有安全垫齐刷刷是 0 —— 而没人会盯着一个"看起来就该是 0"的字段。 + # 漏账看得见 (有行、有 ERROR、有数字), 错账看不见。所以拿不准就不建。 + if not [b for b in book if int(b.get("total_qty") or 0) > 0] and \ + param_store.get_bool("PMS_RECON_REQUIRE_COST", True): + chk = rbc.check_costs(ds.get("rows") or [], prices) + out["cost_check"] = {k: chk[k] for k in ("n", "counts", "eq_ratio", "blocking", + "reasons", "hint")} + out["coverage"] = rbc.coverage(chk["rows"]) + if chk["blocking"] and not force: + out.update({"severity": rc.SEV_ERROR, "fixes": [], + "blocked": {"why": "首次建账被成本价闸拦下: " + "; ".join(chk["reasons"]), + "hint": chk["hint"], + "how": "请对端把 trading_position 的 cost_price 填成真实" + "成本价再重试; 确认这份数据就是对的可用 force=true " + "放行 (那意味着你接受安全垫从 0 起算)"}}) + logger.error("[对账] 首次建账被成本价闸拦下: %s。%s", chk["reasons"], chk["hint"]) + return out + if chk["blocking"]: + logger.warning("[对账] 成本价闸本应拦下 (%s), 但 force=true 放行 —— " + "安全垫将从 0 起算, 补仓/加仓/保垫减仓这一轮判不准", chk["reasons"]) + if not out["coverage"]["enough"]: + logger.warning("[对账] 建账数据的情形覆盖不全: %s", out["coverage"]["hint"]) + # 下游的成本价 —— 补仓位时的开仓价优先取它, 现价只兜底 (见 rc.build_recon_fixes 注释: # 拿现价当成本会让安全垫齐刷刷归零, 整条纪律链跟着失灵) costs = {r["ts_code"]: r.get("cost") for r in (ds.get("rows") or [])} @@ -660,6 +694,158 @@ def reconcile(*, apply_fix: bool = True, force: bool = False) -> dict: return out +def rebuild_preflight() -> dict: + """账本重建的**只读**预检 (README 待办 #4)。一个字都不写, 随时可跑。 + + 回答三个问题, 顺序就是它们该被回答的顺序: + 1. 事实源给不给得出持仓? (给不出就没什么可谈的) + 2. 这份持仓的成本价能不能用? (`rebuild_check` 的成本价闸, 这是重建的成败所在) + 3. 建完之后验不验得到纪律? (§5.2 的四种情形; 不阻断, 但缺了就是白建一轮) + + 再加一段开关现状。清账期间关掉的东西, 重建完必须记得打开 —— 尤其 `PMS_SIGNAL_ENABLED`: + 账本空时信号消化判 IGNORE 也照样 ACK, 卖出信号会被消费组静默吃掉且跨日拿不回来。 + """ + out = {"ok": True, "ready": False, "steps": [], "switches": {}, "hint": ""} + book = [p for p in pms_repo.list_positions() if int(p.get("total_qty") or 0) > 0] + out["book_held"] = len(book) + out["first_build"] = not book + + src = positions_source() + out["source"] = {"source": src["source"], "mode": src["mode"], "age_sec": src.get("age_sec"), + "rows": len(src["rows"]), "columns": src["columns"], + "alerts": src["alerts"]} + if src["source"] == SRC_NONE: + out["steps"].append({"step": "事实源", "ok": False, + "why": "ws 快照与 trading_position 都没有应答 —— 不是空集, 是没读到。" + "先确认 pms-ws 在跑且对端实现了 query_positions"}) + out["hint"] = "拿不到事实源, 谈不上重建 (读不到 ≠ 清仓)" + return out + rows = [r for r in src["rows"] if cs.is_stock_code(r.get("ts_code"))] + out["steps"].append({"step": "事实源", "ok": bool(rows), + "why": f"{src['source']} 给出 {len(rows)} 只持仓" + + ("" if rows else " —— 对端还没装持仓, 等它")}) + if not rows: + out["hint"] = "下游一只持仓都没有 —— 等对端装好再来" + return out + + prices = market.get_prices([r["ts_code"] for r in rows]) + chk = rbc.check_costs(rows, prices) + out["cost_check"] = chk + out["steps"].append({"step": "成本价体检", "ok": not chk["blocking"], "why": chk["hint"]}) + + cov = rbc.coverage(chk["rows"]) + out["coverage"] = cov + out["steps"].append({"step": "情形覆盖", "ok": cov["enough"], "why": cov["hint"], + "blocking": False}) + + out["switches"] = { + "PMS_DISPATCH_MODE": param_store.get("PMS_DISPATCH_MODE", "shadow"), + "PMS_AUTONOMY": param_store.get("PMS_AUTONOMY", "propose_only"), + "PMS_SIGNAL_ENABLED": param_store.get_bool("PMS_SIGNAL_ENABLED", True), + "PMS_EXEC_HALT": param_store.get_bool("PMS_EXEC_HALT", False), + "PMS_SECTOR_SOURCE": param_store.get("PMS_SECTOR_SOURCE", ""), + } + out["after"] = [ + "重建完立刻把 PMS_SIGNAL_ENABLED 打开 —— 账本空时信号消化判 IGNORE 也照样 ACK, " + "卖出信号被消费组静默吃掉且跨日拿不回来", + "确认账本无误后再重开 pms-beat (15:10 日终结算会自动认领 trading_position, " + "对端装到一半时跨过 15:10 会拿半成品建账)", + "跑一次 ledger_service.rebuild_accept() 看安全垫分布与行业集中度", + ] + out["ready"] = not chk["blocking"] + out["hint"] = (chk["hint"] if chk["blocking"] else + f"可以重建: {len(rows)} 只持仓、成本价可用。" + + ("" if cov["enough"] else "注意 " + cov["hint"])) + return out + + +def rebuild_accept() -> dict: + """重建之后的**只读**判收。回答"这本账建对了没有"。 + + 最硬的一条判据是**安全垫分布**: 如果重建后每一只票的安全垫都是 0, 那就是踩了 + 「拿现价当成本」那个坑 —— 页面上每个数都合理, 只有这一处露馅。所以专门数它。 + + 顺带回答 README 待办 #3 留的那个问题: `PMS_SECTOR_MAX_RATIO=40%` 是当初按"二级或更粗" + 的粒度定的, 换到 gp_hybk 三级 (884*) 之后偏不偏松 —— 有了真实持仓分布才算得出来。 + """ + out = {"ok": True, "checks": [], "hint": ""} + view = portfolio.positions_view() + held = view.get("positions") or [] + t = view.get("totals") or {} + out["held"] = len(held) + if not held: + out.update({"ok": False, "hint": "账本还是空的 —— 重建没跑, 或跑了但被闸拦下了"}) + return out + + cush = [x.get("cushion_pct") for x in held if x.get("cushion_pct") is not None] + zero = [x for x in held if abs(float(x.get("cushion_pct") or 0)) < 1e-9] + out["cushion"] = {"n": len(cush), "zero": len(zero), + "min": round(min(cush), 4) if cush else None, + "max": round(max(cush), 4) if cush else None, + "solid": t.get("solid_names"), "neg": t.get("neg_names")} + all_zero = bool(held and len(zero) == len(held)) + out["checks"].append({ + "check": "安全垫分布", "ok": not all_zero, + "why": ("**每一只的安全垫都是 0** —— 这正是拿现价当成本的样子。摊薄成本等于当天价, " + "补仓/加仓/保垫减仓这一整条纪律链会全程判不出来。请核对下游 cost_price" + if all_zero else + f"{len(cush)} 只有安全垫, 区间 {out['cushion']['min']:+.1%} ~ " + f"{out['cushion']['max']:+.1%}, 其中 {len(zero)} 只为 0")}) + + lots = {} + for x in held: + for l in pms_repo.list_lots(x["ts_code"], status="OPEN"): + lots[l["lot_type"]] = lots.get(l["lot_type"], 0) + 1 + out["lots"] = lots + out["checks"].append({"check": "批次账", "ok": bool(lots), + "why": f"批次分布 {lots or '(空)'}" + + ("" if lots else " —— 持仓有行但没有批次, 账本结构不完整")}) + + ready = bool(t.get("sector_ready", view.get("sector_ready"))) + names, mv = t.get("sector_names") or {}, t.get("sector_mv") or {} + # **分母必须与规则闸一致**: sizer.check_caps 的行业判据是 + # (sector_mv + add) / port_after > sector_max_ratio + # 即「占组合持仓市值」, 不是占总规模 PMS_SCALE。两边用不同分母的话, 这里报"没超"而 + # 规则闸拦人 (或反过来), 而两个数字都自称是"行业集中度"——同名不同物最难查。 + # (settings.py 里那句注释写的是"占总仓", 容易被读成占 PMS_SCALE, 已在此写明。) + port_mv = float(t.get("portfolio_mv") or 0.0) or sum( + float(x.get("market_value") or 0.0) for x in held) + max_ratio = param_store.get_float("PMS_SECTOR_MAX_RATIO", 0.40) + max_names = param_store.get_int("PMS_SECTOR_MAX_NAMES", 4) + top = sorted(((s, (mv[s] / port_mv if port_mv else 0.0), names.get(s, 0)) for s in mv), + key=lambda x: -x[1])[:5] + over_ratio = [x for x in top if x[1] > max_ratio + 1e-9] + over_names = [s for s, n in names.items() if n > max_names] + out["sector"] = {"ready": ready, "max_ratio": max_ratio, "max_names": max_names, + "denominator": "组合持仓市值 (与 sizer.check_caps 一致)", + "portfolio_mv": round(port_mv, 2), + "over_ratio": [s for s, _, _ in over_ratio], "over_names": over_names, + "top": [{"sector": s, "ratio": round(r, 4), "names": n} for s, r, n in top]} + worst = top[0][1] if top else 0.0 + if not ready: + why = "行业源没就绪 —— 集中度约束此刻等同未配置且会静默失效" + elif over_ratio or over_names: + # 接管进来的持仓超限**不算重建失败** —— 账本忠实反映了下游的真实状态, 那才是它的职责。 + # 但必须说出来: 规则闸从此会挡住这些行业的加仓, 不说的话下次加不进去会以为是 bug。 + why = (f"接管进来的持仓**已经超限**: " + + (f"占比超上限 {max_ratio:.0%} 的有 {[s for s, _, _ in over_ratio]}" + f" (最大 {worst:.1%})" if over_ratio else "") + + (f" 只数超上限 {max_names} 的有 {over_names}" if over_names else "") + + "。这不是重建出错 —— 账本忠实反映了下游真实持仓; 但规则闸从此会挡住这几个" + "行业的加仓, 心里要有数 (减持方向不受影响)") + else: + why = f"最大行业占组合 {worst:.1%} (上限 {max_ratio:.0%}), 只数上限 {max_names}, 未超限" + if worst < max_ratio * 0.5: + why += ("。阈值是当初按二级或更粗的粒度定的, 现在是三级 884* —— " + "拿这份真实分布回看它偏不偏松 (参考项目三级用 20%)") + out["checks"].append({"check": "行业集中度", "ok": ready, "why": why}) + + out["ok"] = all(c["ok"] for c in out["checks"]) + bad = [c["check"] for c in out["checks"] if not c["ok"]] + out["hint"] = "账本重建判收通过" if out["ok"] else "这几项没过: " + " / ".join(bad) + return out + + def _apply_fix(f: dict): code = f["ts_code"] pms_repo.ensure_position(code) diff --git a/app/web/main.py b/app/web/main.py index 5918ca5..f7ae42e 100644 --- a/app/web/main.py +++ b/app/web/main.py @@ -287,6 +287,27 @@ def api_replay(limit: int = Query(500)): return ok(ledger_service.replay_fills, limit=limit) +@app.get("/api/ops/rebuild-preflight") +def api_rebuild_preflight(): + """账本重建的只读预检 (README 待办 #4)。不写任何东西, 随时可点。 + + 成本价能不能用是这一步的成败所在 —— 详见 app/core/rebuild_check.py 的模块说明。 + """ + def _pf(): + from app.services import ledger_service + return ledger_service.rebuild_preflight() + return ok(_pf) + + +@app.get("/api/ops/rebuild-accept") +def api_rebuild_accept(): + """重建之后的只读判收: 安全垫分布 / 批次账 / 行业集中度。""" + def _ac(): + from app.services import ledger_service + return ledger_service.rebuild_accept() + return ok(_ac) + + @app.post("/api/ops/reconcile") def api_reconcile(apply_fix: bool = Query(True), force: bool = Query(False)): """force=true 绕过「差异面过大不自动改账」的限制, 仅在人工确认下游读数无误后使用。""" diff --git a/scripts/rebuild_ledger.py b/scripts/rebuild_ledger.py new file mode 100644 index 0000000..6f21114 --- /dev/null +++ b/scripts/rebuild_ledger.py @@ -0,0 +1,156 @@ +# -*- coding: utf-8 -*- +""" +账本重建: 预检 → 执行 → 判收 (README 待办 #4) +============================================== + docker compose run --rm --no-deps pms-web python scripts/rebuild_ledger.py # 只预检 + docker compose run --rm --no-deps pms-web python scripts/rebuild_ledger.py --yes # 预检通过就执行 + docker compose run --rm --no-deps pms-web python scripts/rebuild_ledger.py --accept # 只看判收 + (或 make rebuild / make rebuild GO=1 / make rebuild-accept) + +清账之后账本是空的, 真实持仓要按「以下游为准」认领回来。这一步**只发生一次**, 但它定死了 +每一只票的开仓价, 而开仓价一错, 后面每一条纪律都错在一个看不出来的地方: + + 下游 cost_price → RECON 批次开仓价 → 摊薄成本 → 安全垫 + ↓ + 盈利加仓(≥3%) · 保垫减仓(峰值≥6%) · 补仓评估档(−8%/−15%) + +所以本脚本**默认只预检不执行**。加 `--yes` 才真的改账, 且预检不过照样不执行 (要强行放行得 +再加 `--force`, 那等于声明"我确认这份数据就是对的, 接受安全垫从 0 起算")。 + +三段各自的意思: + [1] 预检 —— 事实源给不给得出持仓 / 成本价能不能用 / 建完验不验得到纪律 (只读) + [2] 执行 —— reconcile(apply_fix=True), 走的是生产同一条路, 不是另写一套 + [3] 判收 —— 安全垫分布是最硬的判据: 全 0 就是踩了"拿现价当成本"那个坑 +""" +from __future__ import annotations + +import argparse +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +OK, BAD, WARN = " OK ", " FAIL ", " WARN " + + +def _line(tag, msg): + print(f"[{tag}] {msg}") + + +def _steps(steps): + for s in steps: + tag = OK if s.get("ok") else (WARN if s.get("blocking") is False else BAD) + print(f" [{tag}] {s['step']}: {s['why']}") + + +def show_preflight(pf) -> bool: + print("=" * 72) + print("[1] 预检 (只读, 一个字都不写)") + print("=" * 72) + _line("账本", f"当前持仓 {pf['book_held']} 只" + + (" —— 空账本, 这是首次建账, 成本价闸生效" if pf["first_build"] + else " —— 非空, 走的是日常对账口径, 成本价闸不生效")) + src = pf.get("source") or {} + _line("事实源", f"{src.get('source')} (mode={src.get('mode')}, " + f"age={src.get('age_sec')}s, 列={src.get('columns')})") + for a in (src.get("alerts") or []): + _line(a.get("level", "WARN"), a.get("message")) + _steps(pf.get("steps") or []) + + chk = pf.get("cost_check") or {} + if chk.get("rows"): + print(f"\n 成本价明细 ({chk['n']} 只, 判定分布 {chk['counts']}):") + for r in chk["rows"]: + cu = "" if r.get("cushion_pct") is None else f" 安全垫 {r['cushion_pct']:+.1%}" + mark = " " if r["verdict"] == "OK" else "! " + print(f" {mark}{r['ts_code']:<12} {str(r['qty']):>7} 股 " + f"成本 {_g(r['cost'])} 现价 {_g(r['price'])} " + f"[{r['verdict']}]{cu}") + if r.get("why"): + print(f" {r['why']}") + + cov = pf.get("coverage") or {} + if cov: + print("\n 情形覆盖 (不阻断建账, 只影响这轮验不验得到纪律):") + for k, v in (cov.get("counts") or {}).items(): + print(f" [{OK if v else WARN}] {k}: {v}") + + sw = pf.get("switches") or {} + if sw: + print("\n 开关现状: " + " · ".join(f"{k}={v}" for k, v in sw.items())) + print(f"\n 结论: {pf.get('hint')}") + return bool(pf.get("ready")) + + +def show_accept(ac): + print("=" * 72) + print("[3] 判收 (只读)") + print("=" * 72) + _line("持仓", f"{ac.get('held')} 只") + for c in ac.get("checks") or []: + print(f" [{OK if c['ok'] else BAD}] {c['check']}: {c['why']}") + if ac.get("sector", {}).get("top"): + print("\n 行业分布 (占规模):") + for s in ac["sector"]["top"]: + print(f" {s['sector'] or '(无)':<20} {s['ratio']:>7.1%} {s['names']} 只") + print(f"\n 结论: {ac.get('hint')}") + return bool(ac.get("ok")) + + +def _g(v): + return "-" if v in (None, "") else f"{float(v):.4g}" + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--yes", action="store_true", help="预检通过就执行重建 (缺省只预检)") + ap.add_argument("--force", action="store_true", + help="预检不过也执行。**这等于声明你接受安全垫从 0 起算**") + ap.add_argument("--accept", action="store_true", help="只跑判收段 (重建已经做过了)") + args = ap.parse_args() + + from app.services import ledger_service as ls + + if args.accept: + return 0 if show_accept(ls.rebuild_accept()) else 1 + + pf = ls.rebuild_preflight() + ready = show_preflight(pf) + + if not args.yes: + print("\n(只预检, 没有改任何东西。要执行加 --yes)") + return 0 if ready else 2 + if not ready and not args.force: + print("\n预检没过, 不执行。请对端把 trading_position 的 cost_price / " + "available_quantity 填成真实值后重跑;\n" + "确认这份数据就是对的可以加 --force 放行 (那意味着安全垫从 0 起算, " + "这一轮的补仓/加仓/保垫减仓都判不准)。") + return 2 + + print("\n" + "=" * 72) + print("[2] 执行 reconcile(apply_fix=True" + (", force=True" if args.force else "") + ")") + print("=" * 72) + res = ls.reconcile(apply_fix=True, force=args.force) + _line("结果", f"ok={res.get('ok')} 事实源={res.get('source')} " + f"差异 {len(res.get('diffs') or [])} 项 修正 {len(res.get('fixes') or [])} 项") + if res.get("blocked"): + _line("BLOCKED", res["blocked"].get("why")) + if res["blocked"].get("how"): + _line("怎么办", res["blocked"]["how"]) + return 1 + for f in res.get("fixes") or []: + print(f" {f['ts_code']:<12} {f['op']:<14} {f['qty']:>7} 股 " + f"@ {_g(f.get('price'))} {f.get('price_source') or ''}") + for e in res.get("errors") or []: + _line("ERROR", e) + + print() + ok = show_accept(ls.rebuild_accept()) + print("\n收尾 (脚本不替你做, 因为这几件事该由人确认):") + for s in pf.get("after") or []: + print(f" - {s}") + return 0 if ok else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/run_tests.py b/scripts/run_tests.py index a75cfd8..032b1bd 100644 --- a/scripts/run_tests.py +++ b/scripts/run_tests.py @@ -13,8 +13,9 @@ test_batch6_units.py ws 通道: 测试向量/签名/公钥/水位/DDL/逐笔入账 (65 例) test_batch7_units.py 上游选股计划: 解析/新鲜度/候选筛选/取数守卫 (32 例) test_batch8_units.py 榜单变化: 名册指纹/三种语义/尾部闸/落库往返 (60 例) + test_batch9_units.py 接管持仓前的成本价体检: 阻断判据/情形覆盖 (23 例) test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) (58 例) - 共 304 例 + 共 327 例 任一子集失败即整体失败 (退出码 1)。 """ import os @@ -25,7 +26,8 @@ HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.dirname(HERE) 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_wiring.py"] + "test_batch7_units.py", "test_batch8_units.py", "test_batch9_units.py", + "test_wiring.py"] def main(): diff --git a/scripts/test_batch9_units.py b/scripts/test_batch9_units.py new file mode 100644 index 0000000..e8b4c74 --- /dev/null +++ b/scripts/test_batch9_units.py @@ -0,0 +1,250 @@ +# -*- coding: utf-8 -*- +""" +第九批模块单测 (零外部依赖, 不连库不触网) +========================================== +运行: 在 tradingSystem 仓库根目录执行 python scripts/test_batch9_units.py +覆盖: 接管既有持仓前的成本价体检 (app/core/rebuild_check.py)。 + +这一批守的是一个**只发生一次、且不可逆**的决定: 账本清空后按「以下游为准」认领真实持仓, +那一刻定死每只票的开仓价 → 摊薄成本 → 安全垫 → 补仓/加仓/保垫减仓的共同判据。 +2026-07-29 首次接管 22 只持仓时踩过一次 (全按现价开仓, 安全垫齐刷刷是 0)。 + +所以用例的重点不是"算得对不对", 而是**该拦的拦不拦得住**, 以及**不该拦的会不会误伤** —— +误伤的代价是多等一天, 漏拦的代价是一本每个数都错、却和真账长得一模一样的账。 +""" +import os +import sys +import traceback + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from app.core import rebuild_check as rb # noqa: E402 + +RESULTS = [] + + +def case(name): + def deco(fn): + RESULTS.append((name, fn)) + return fn + return deco + + +def _p(code, qty=1000, cost=10.0, avail=None, price=None): + """一行下游持仓 (downstream_repo.fetch_positions 的行形态)。""" + return {"ts_code": code, "qty": qty, "cost": cost, + "avail_qty": qty if avail is None else avail, "price": price} + + +# ================================================================ [A] 单只票的判定 +@case("[A1] 成本价正常 → OK, 并算出安全垫") +def t_a1(): + r = rb.check_row(_p("600000.SH", cost=20.0), price=10.0) + assert r["verdict"] == rb.OK + assert abs(r["cushion_pct"] - (-0.5)) < 1e-9, "真实成本 20、现价 10 就是实亏 50%" + + +@case("[A2] 成本价缺失 / 为 0 / 为负 → MISSING") +def t_a2(): + for bad in (None, 0, 0.0, -1.0, ""): + r = rb.check_row(_p("600000.SH", cost=bad), price=10.0) + assert r["verdict"] == rb.MISSING, f"cost={bad!r} 应判 MISSING" + assert "安全垫恒为 0" in rb.check_row(_p("600000.SH", cost=0), price=10.0)["why"] + + +@case("[A3] 成本 ≈ 现价 → EQ_PRICE (安全垫≈0)") +def t_a3(): + r = rb.check_row(_p("600000.SH", cost=10.02), price=10.0) # 差 0.2% < 0.5% + assert r["verdict"] == rb.EQ_PRICE + r2 = rb.check_row(_p("600000.SH", cost=10.30), price=10.0) # 差 3% > 0.5% + assert r2["verdict"] == rb.OK, "正常的小幅浮盈不该被当成'拿现价充数'" + + +@case("[A4] 成本与现价差两个数量级 → ABSURD (多半是单位错)") +def t_a4(): + assert rb.check_row(_p("600000.SH", cost=1000.0), price=10.0)["verdict"] == rb.ABSURD + assert rb.check_row(_p("600000.SH", cost=0.05), price=10.0)["verdict"] == rb.ABSURD + # 分/元 混用是最典型的一种: 成本记成 1002 分而现价是 10.02 元 + r = rb.check_row(_p("600000.SH", cost=1002.0), price=10.02) + assert r["verdict"] == rb.ABSURD and "单位" in r["why"] + + +@case("[A5] 可用量 > 总量 或为负 → AVAIL_BAD") +def t_a5(): + assert rb.check_row(_p("600000.SH", qty=1000, avail=1500), price=10.0)["verdict"] \ + == rb.AVAIL_BAD + assert rb.check_row(_p("600000.SH", qty=1000, avail=-1), price=10.0)["verdict"] \ + == rb.AVAIL_BAD + # 当日买入是合法的: 可用 < 总量 (成本给个与现价不同的值, 免得撞上 EQ_PRICE) + assert rb.check_row(_p("600000.SH", qty=1000, avail=0, cost=20.0), + price=10.0)["verdict"] == rb.OK + + +@case("[A6] 取不到现价 → NO_PRICE, 但成本本身仍算可用") +def t_a6(): + r = rb.check_row(_p("600000.SH", cost=20.0), price=None) + assert r["verdict"] == rb.NO_PRICE and r["cushion_pct"] is None + assert "成本值本身可用" in r["why"] + + +@case("[A7] 可用量的判定排在成本之前 —— 两个都坏时先报可用量") +def t_a7(): + r = rb.check_row(_p("600000.SH", qty=100, avail=999, cost=0), price=10.0) + assert r["verdict"] == rb.AVAIL_BAD + + +# ================================================================ [B] 整批的阻断判据 +@case("[B1] 全部正常 → 不阻断") +def t_b1(): + rows = [_p("600000.SH", cost=20.0), _p("600519.SH", cost=1500.0)] + out = rb.check_costs(rows, {"600000.SH": 10.0, "600519.SH": 1800.0}) + assert out["blocking"] is False and out["reasons"] == [] + assert "可以建账" in out["hint"] + + +@case("[B2] 只要有一只成本不可用就阻断 —— 一本账里混一只错的也不行") +def t_b2(): + 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] + + +@case("[B3] 整组成本≈现价 → 阻断 (这就是 07-29 那个坑的模样)") +def t_b3(): + rows = [_p(f"60000{i}.SH", cost=10.0) for i in range(1, 6)] + out = rb.check_costs(rows, {f"60000{i}.SH": 10.0 for i in range(1, 6)}) + assert out["blocking"] is True and out["eq_ratio"] == 1.0 + assert "安全垫会是 0" in out["reasons"][0] + + +@case("[B4] 少数几只成本≈现价 → 不阻断 (当日买入是正常的)") +def t_b4(): + rows = [_p("600000.SH", cost=20.0), _p("600519.SH", cost=1800.0), + _p("600036.SH", cost=30.0), _p("601318.SH", cost=10.0)] + px = {"600000.SH": 10.0, "600519.SH": 1500.0, "600036.SH": 40.0, "601318.SH": 10.0} + out = rb.check_costs(rows, px) + assert out["counts"].get(rb.EQ_PRICE) == 1 + assert out["blocking"] is False, "四只里一只当日买入很正常, 不该拦" + + +@case("[B5] 只有一只票时不按占比阻断 —— 样本太小, 它可能真是当日买的") +def t_b5(): + out = rb.check_costs([_p("600000.SH", cost=10.0)], {"600000.SH": 10.0}) + assert out["counts"].get(rb.EQ_PRICE) == 1 + assert out["blocking"] is False + + +@case("[B6] 取不到现价的票不稀释 EQ_PRICE 占比") +def t_b6(): + # 两只成本≈现价 + 八只取不到现价。若拿 10 做分母, 占比 20% 不阻断 —— 那是错的: + # 能判的两只全是 EQ_PRICE, 该阻断。 + rows = [_p("600001.SH", cost=10.0), _p("600002.SH", cost=10.0)] + \ + [_p(f"6001{i:02d}.SH", cost=5.0) for i in range(1, 9)] + px = {"600001.SH": 10.0, "600002.SH": 10.0} # 其余取不到价 + out = rb.check_costs(rows, px) + assert out["counts"].get(rb.NO_PRICE) == 8 + assert out["eq_ratio"] == 1.0 and out["blocking"] is True + + +@case("[B7] 零持仓行不参与体检") +def t_b7(): + rows = [_p("600000.SH", qty=0, cost=0), _p("600519.SH", cost=1800.0)] + out = rb.check_costs(rows, {"600519.SH": 1500.0}) + assert out["n"] == 1 and out["blocking"] is False, "qty=0 的行是历史残留, 不该拖累建账" + + +@case("[B8] 下游一只持仓都没有 → 不阻断但说清是没账可建") +def t_b8(): + out = rb.check_costs([], {}) + assert out["n"] == 0 and out["blocking"] is False + assert "无账可建" in out["hint"] + + +@case("[B9] 阻断时的 hint 要给出可执行的下一步") +def t_b9(): + out = rb.check_costs([_p("600000.SH", cost=0)], {"600000.SH": 10.0}) + assert "cost_price" in out["hint"], "得说清要对端改哪一列, 不是只说'数据有问题'" + assert "长得一模一样" in out["hint"], "得说清为什么不能将就着建" + + +# ================================================================ [C] 情形覆盖 (不阻断) +@case("[C1] 四种情形齐全 → enough") +def t_c1(): + rows = [_p("600000.SH", cost=10.0), # 浮盈 +50% + _p("600519.SH", cost=20.0), # 浮亏 −25% + _p("600036.SH", cost=10.0, qty=1000, avail=0), # 当日买入 + _p("601318.SH", cost=10.0)] + px = {"600000.SH": 15.0, "600519.SH": 15.0, "600036.SH": 11.0, "601318.SH": 11.0} + cov = rb.coverage(rb.check_costs(rows, px)["rows"]) + assert cov["enough"] is True and cov["missing"] == [] + + +@case("[C2] 全是不赚不亏 → 报出这轮验不到哪些纪律") +def t_c2(): + rows = [_p(f"60000{i}.SH", cost=10.0) for i in range(1, 4)] + cov = rb.coverage(rb.check_costs(rows, {f"60000{i}.SH": 10.05 for i in range(1, 4)})["rows"]) + assert cov["enough"] is False + assert any("浮盈" in m for m in cov["missing"]) + assert any("浮亏" in m for m in cov["missing"]) + + +@case("[C3] 只有一只持仓 → 报持仓不足 3 只") +def t_c3(): + cov = rb.coverage(rb.check_costs([_p("600000.SH", cost=20.0)], {"600000.SH": 10.0})["rows"]) + assert cov["enough"] is False and any("不足 3 只" in m for m in cov["missing"]) + + +@case("[C4] 覆盖不全绝不阻断建账 —— 数据是真的就该建") +def t_c4(): + rows = [_p("600000.SH", cost=20.0), _p("600519.SH", cost=3000.0)] + out = rb.check_costs(rows, {"600000.SH": 10.0, "600519.SH": 1500.0}) + cov = rb.coverage(out["rows"]) + assert out["blocking"] is False and cov["enough"] is False + + +# ================================================================ [D] 健壮性 +@case("[D1] 脏数据不炸") +def t_d1(): + for bad in ({"ts_code": "600000.SH"}, {"ts_code": None, "qty": "x", "cost": "y"}, + {"ts_code": "600000.SH", "qty": "1000", "cost": "20.0", "avail_qty": "1000"}): + rb.check_row(bad, price=10.0) + out = rb.check_costs([{"ts_code": "600000.SH", "qty": "1000", "cost": "20.0"}], + {"600000.SH": "10.0"}) + assert out["n"] == 1, "字符串数字要能吃进去 —— 库里 DECIMAL 列取出来常是字符串" + + +@case("[D2] rows 为 None 不炸") +def t_d2(): + assert rb.check_costs(None, None)["n"] == 0 + assert rb.coverage(None)["enough"] is False + + +@case("[D3] 判定常量互不相同 —— 别把两种故障混成一个码") +def t_d3(): + vs = [rb.OK, rb.MISSING, rb.EQ_PRICE, rb.ABSURD, rb.AVAIL_BAD, rb.NO_PRICE] + assert len(set(vs)) == len(vs) + + +def main(): + import logging + logging.disable(logging.CRITICAL) + passed, failed = 0, 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()