处理s3收尾工作

app/services/ledger_service.py

@@ -371,8 +371,43 @@ def recompute_position(ts_code: str) -> dict:
 
 
 # ================================================================ 对账
-def reconcile(*, apply_fix: bool = True) -> dict:
+def _recon_blast_guard(book: list, ds: dict, diffs: list, *, force: bool = False):
-    """账本 vs 下游持仓, 以下游为准修正并留痕。"""
+    """对账修正的爆炸半径限制。返回 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 注释:

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)):
+def api_reconcile(apply_fix: bool = Query(True), force: bool = Query(False)):
-    return ok(ledger_service.reconcile, apply_fix=apply_fix)
+    """force=true 绕过「差异面过大不自动改账」的限制, 仅在人工确认下游读数无误后使用。"""
+    return ok(ledger_service.reconcile, apply_fix=apply_fix, force=force)
 
 
 @app.post("/api/ops/premarket")

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