From 580e03fc76e6547a637c9dcbe46df647a404b345 Mon Sep 17 00:00:00 2001 From: zlt Date: Wed, 29 Jul 2026 16:55:45 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A4=84=E7=90=86s3=E6=94=B6=E5=B0=BE=E5=B7=A5?= =?UTF-8?q?=E4=BD=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/services/ledger_service.py | 54 ++++++++++++++++++++++++++++++++-- app/web/main.py | 5 ++-- scripts/test_wiring.py | 52 ++++++++++++++++++++++++++++++++ 3 files changed, 107 insertions(+), 4 deletions(-) diff --git a/app/services/ledger_service.py b/app/services/ledger_service.py index 821469b..c1cd575 100644 --- a/app/services/ledger_service.py +++ b/app/services/ledger_service.py @@ -371,8 +371,43 @@ def recompute_position(ts_code: str) -> dict: # ================================================================ 对账 -def reconcile(*, apply_fix: bool = True) -> dict: - """账本 vs 下游持仓, 以下游为准修正并留痕。""" +def _recon_blast_guard(book: list, ds: dict, diffs: list, *, force: bool = False): + """对账修正的爆炸半径限制。返回 None 表示放行, 否则返回拦截原因。 + + 只拦「大面积重写」, 不拦日常漂移 —— 后者正是对账存在的意义。两条判据: + + 1. **下游读回空、而本端有持仓** —— 直接拦。真的一夜清仓与「读空」在数据上完全同形, + 但前者极罕见、后者(重启/连错库/权限/网络)很常见, 且代价不对称: 拦错了只是晚一天 + 修账, 放错了是 22 个持仓的批次结构不可逆地没了。 + 2. **受影响的股票数既超绝对值又超占比** —— 两个条件同时满足才拦, 避免持仓很少时 + (比如只剩 3 只) 一点正常漂移就被占比判据挡住。 + + 首次建账 (本端无持仓、下游有) 不受限制: 那时 held=0, 两条判据都不成立。 + """ + if force: + return None + held = [b for b in book if int(b.get("total_qty") or 0) > 0] + if not held: + return None # 空账本 → 建账/认领, 放行 + if not (ds.get("rows") or []): + return {"why": "下游持仓读回空集, 而本端有持仓 —— 视为读数异常而非真实清仓", + "held": len(held), "diffs": len(diffs)} + max_names = param_store.get_int("PMS_RECON_MAX_FIX_NAMES", 5) + max_ratio = param_store.get_float("PMS_RECON_MAX_FIX_RATIO", 0.34) + ratio = len(diffs) / len(held) + if len(diffs) > max_names and ratio > max_ratio: + return {"why": f"差异面过大: {len(diffs)} 只 / 持仓 {len(held)} 只 " + f"({ratio:.0%} > {max_ratio:.0%} 且 > {max_names} 只)", + "held": len(held), "diffs": len(diffs)} + return None + + +def reconcile(*, apply_fix: bool = True, force: bool = False) -> dict: + """账本 vs 下游持仓, 以下游为准修正并留痕。 + + force=True 绕过爆炸半径限制 (见 _recon_blast_guard) —— 只在人工确认下游读数确实 + 正确之后使用。 + """ out = {"ok": True, "diffs": [], "fixes": [], "errors": [], "columns": {}, "severity": rc.SEV_OK} try: ds = downstream_repo.fetch_positions() @@ -402,6 +437,21 @@ def reconcile(*, apply_fix: bool = True) -> dict: if not diffs or not apply_fix: return out + blast = _recon_blast_guard(book, ds, diffs, force=force) + if blast: + # 差异面太大, 不自动改账。**这一条是 2026-07-29 的血的教训**: 下游一次读空 + # (对端模拟环境重启), 对账照着「以下游为准」把 22 个持仓 126 万市值全核销了, + # 全程没有任何拦截 —— 上面那道数量列校验写的是 `if ds["rows"] and ...`, + # 空结果集连它都不走。 + # 「以下游为准」是对的, 但它的前提是「读到的确实是下游的真实状态」。读空、读漏、 + # 连错库都会长成同一个样子, 而代价是不可逆的批次核销。所以给它加个爆炸半径: + # 小幅漂移照常自动修 (那正是对账的价值), 大面积重写一律停下来等人。 + out.update({"severity": rc.SEV_ERROR, "fixes": [], "blocked": blast}) + logger.error("[对账] 拒绝自动修正: %s。差异 %s 项, 持仓 %s 只。" + "确认下游读数无误后, 用 apply_fix=true&force=true 手工放行", + blast["why"], len(diffs), blast["held"]) + return out + codes = [d["ts_code"] for d in diffs] prices = market.get_prices(codes) # 下游的成本价 —— 补仓位时的开仓价优先取它, 现价只兜底 (见 rc.build_recon_fixes 注释: diff --git a/app/web/main.py b/app/web/main.py index 82bb461..3104b92 100644 --- a/app/web/main.py +++ b/app/web/main.py @@ -288,8 +288,9 @@ def api_replay(limit: int = Query(500)): @app.post("/api/ops/reconcile") -def api_reconcile(apply_fix: bool = Query(True)): - return ok(ledger_service.reconcile, apply_fix=apply_fix) +def api_reconcile(apply_fix: bool = Query(True), force: bool = Query(False)): + """force=true 绕过「差异面过大不自动改账」的限制, 仅在人工确认下游读数无误后使用。""" + return ok(ledger_service.reconcile, apply_fix=apply_fix, force=force) @app.post("/api/ops/premarket") diff --git a/scripts/test_wiring.py b/scripts/test_wiring.py index f0145bd..8d9d4ed 100644 --- a/scripts/test_wiring.py +++ b/scripts/test_wiring.py @@ -837,6 +837,58 @@ def _(): downstream_repo.fetch_positions = orig +@case("账本服务·下游读空绝不清账 (2026-07-29 实际发生过的 22 只全核销)") +def _(): + from app.services import ledger_service as ls + from app.repo import downstream_repo + fake = install_fakes(prices={f"C{i}.SH": 10.0 for i in range(22)}) + for i in range(22): + code = f"C{i}.SH" + fake.insert_lot(ts_code=code, lot_type="BASE", qty=1000, open_price=10.0, + open_date="2026-07-01") + fake.update_position(code, total_qty=1000, avail_qty=1000) + orig = downstream_repo.fetch_positions + try: + # 对端模拟环境重启, 下游持仓表读回空集。「一夜清仓」与「读空」在数据上完全同形, + # 但代价不对称: 拦错了只是晚一天修账, 放错了是 22 个持仓的批次结构不可逆地没了。 + downstream_repo.fetch_positions = lambda: { + "rows": [], "columns": {"code": "stock_code", "qty": "current_qty"}, + "raw_count": 0} + r = ls.reconcile() + assert r.get("blocked"), "下游读空必须拦截, 不能照着清账" + assert r["severity"] == "ERROR", r["severity"] + assert not r["fixes"], "拦截时不得产生任何修正" + for i in range(22): + assert fake.positions[f"C{i}.SH"]["total_qty"] == 1000, "持仓不得被动过" + r2 = ls.reconcile(force=True) # 人工确认下游读数无误后放行 + assert not r2.get("blocked"), "force 应绕过限制" + finally: + downstream_repo.fetch_positions = orig + + +@case("账本服务·日常小幅漂移仍照常自动修正 (不被爆炸半径误伤)") +def _(): + from app.services import ledger_service as ls + from app.repo import downstream_repo + fake = install_fakes(prices={f"C{i}.SH": 10.0 for i in range(22)}) + for i in range(22): + code = f"C{i}.SH" + fake.insert_lot(ts_code=code, lot_type="BASE", qty=1000, open_price=10.0, + open_date="2026-07-01") + fake.update_position(code, total_qty=1000, avail_qty=1000) + orig = downstream_repo.fetch_positions + try: + rows = [{"ts_code": f"C{i}.SH", "qty": 1000} for i in range(22)] + rows[0]["qty"] = 1500 # 22 只里只有 1 只对不上 + downstream_repo.fetch_positions = lambda: { + "rows": rows, "columns": {"qty": "current_qty"}, "raw_count": 22} + r = ls.reconcile() + assert not r.get("blocked"), "小幅漂移不该被拦 —— 那正是对账的价值" + assert fake.positions["C0.SH"]["total_qty"] == 1500, "以下游为准" + finally: + downstream_repo.fetch_positions = orig + + @case("账本服务·日报生成 (关注区 + 次日除权检测快照)") def _(): from app.services import ledger_service as ls