处理s3收尾工作

app/core/command_spec.py

@@ -313,6 +313,33 @@ def normalize_code(code: str) -> str:
     return c
 
 
+# A 股股票代码的前缀 × 交易所对照。只认股票, 不含 ETF/基金/B 股 —— PMS 管的是个股。
+# 用途见 ledger_service.reconcile: 下游是券商(或联调期的模拟)持仓表, 会混进测试垃圾。
+# 2026-07-29 下游就只剩一只 `999999.SH` 2000 股 —— 我们自己联调时用的假代码 ——
+# 差点被对账当成真持仓补进账本。
+STOCK_PREFIXES = {
+    "SH": ("600", "601", "603", "605", "688", "689"),          # 沪主板 + 科创板
+    "SZ": ("000", "001", "002", "003", "300", "301"),          # 深主板 + 创业板
+    "BJ": ("430", "830", "831", "832", "833", "834", "835",    # 北交所
+           "836", "837", "838", "839", "870", "871", "872", "873", "874", "920"),
+}
+
+
+def is_stock_code(code: str) -> bool:
+    """点式代码是不是一只真实 A 股个股。
+
+    宁可保守: 将来新代码段出现时这里会误杀, 但调用方都会带着原始代码告警, 补一行前缀
+    即可 —— 比把垃圾放进账本便宜得多。
+    """
+    c = normalize_code(code)
+    if "." not in c:
+        return False
+    num, mkt = c.split(".", 1)
+    if not (num.isdigit() and len(num) == 6):
+        return False
+    return num[:3] in STOCK_PREFIXES.get(mkt, ())
+
+
 def validate(cmd_type: str, params: dict) -> tuple:
     """校验并归一命令参数。返回 (normalized_params, errors)。errors 非空即拒绝下达。"""
     spec = SPECS.get(cmd_type)

app/services/ledger_service.py

@@ -18,6 +18,7 @@ from __future__ import annotations
 import logging
 from datetime import datetime, timedelta
 
+from app.core import command_spec as cs
 from app.core import cushion as cu
 from app.core import recon as rc
 from app.core import tradedays as td
@@ -421,6 +422,20 @@ def reconcile(*, apply_fix: bool = True, force: bool = False) -> dict:
             "把列名补进 downstream_repo.QTY_CANDIDATES"]})
         return out
 
+    # 下游行先过一遍代码合法性。券商表里混进非个股代码的原因很多 (联调测试单、B 股、
+    # 基金、脏数据), 而对账是**会照着它改账本**的, 放进来就变成真持仓。
+    # 2026-07-29: 下游只剩一只 `999999.SH` 2000 股 —— 我们自己测拒绝路径用的假代码 ——
+    # 若照单全收, 明天日终就会给账本凭空补出 2000 股不存在的票。
+    ds_rows, junk = [], []
+    for r in ds["rows"]:
+        (ds_rows if cs.is_stock_code(r.get("ts_code")) else junk).append(r)
+    if junk:
+        out["junk_codes"] = [{"ts_code": r.get("ts_code"), "qty": r.get("qty")} for r in junk]
+        logger.error("[对账] 下游有 %s 条非 A 股个股代码, 已忽略不入账: %s。"
+                     "若其中确有真实标的, 说明 command_spec.STOCK_PREFIXES 缺了代码段",
+                     len(junk), out["junk_codes"])
+    ds = {**ds, "rows": ds_rows}
+
     book = [{"ts_code": r["ts_code"], "total_qty": int(r.get("total_qty") or 0)}
             for r in pms_repo.list_positions()]
     diffs = rc.diff_positions(book, [{"ts_code": r["ts_code"], "qty": r["qty"]}

scripts/test_wiring.py

@@ -841,9 +841,9 @@ def _():
 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)})
+    fake = install_fakes(prices={f"6000{i:02d}.SH": 10.0 for i in range(22)})
     for i in range(22):
-        code = f"C{i}.SH"
+        code = f"6000{i:02d}.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)
@@ -859,7 +859,7 @@ def _():
         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, "持仓不得被动过"
+            assert fake.positions[f"6000{i:02d}.SH"]["total_qty"] == 1000, "持仓不得被动过"
         r2 = ls.reconcile(force=True)          # 人工确认下游读数无误后放行
         assert not r2.get("blocked"), "force 应绕过限制"
     finally:
@@ -870,21 +870,21 @@ def _():
 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)})
+    fake = install_fakes(prices={f"6000{i:02d}.SH": 10.0 for i in range(22)})
     for i in range(22):
-        code = f"C{i}.SH"
+        code = f"6000{i:02d}.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 = [{"ts_code": f"6000{i:02d}.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, "以下游为准"
+        assert fake.positions["600000.SH"]["total_qty"] == 1500, "以下游为准"
     finally:
         downstream_repo.fetch_positions = orig