diff --git a/README.md b/README.md index 616181a..d4d4c60 100644 --- a/README.md +++ b/README.md @@ -58,7 +58,7 @@ scripts/ test_batch4_units.py 动作引擎 四类自主动作触发与数量口径 11 例 test_batch5_units.py 决策系统信号流解析与消化口径 8 例 test_batch6_units.py ws 通道: 测试向量/签名/公钥形态/水位/DDL体检 51 例 - test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) 33 例 + test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) 34 例 init_db.py 建表 (应用 ddl_pms_v1.sql, 幂等, 默认演练; 含 DDL 体检) check_db.py 实机连通性与表结构自检 (需真实 .env) gen_keys.py ws 通道密钥: 生成 / 只取公钥(--pubkey) / PEM(--pem) / 自检(--check) @@ -175,6 +175,12 @@ QMT ──trade/order_update──▶ pms-ws ──落 pms_qmt_inbox──▶ 已经被历史成交污染的账本，用 `scripts/reset_ledger.py` 清掉重来（只删 PMS 自有表，绝不碰下游 `trading_*`），然后跑一次 `POST /api/ops/reconcile?apply_fix=true` 让对账以下游为准把真实持仓补回来。 +### 接管既有持仓时的成本价口径 + +对账补仓位（`ADD_RECON_LOT`）的开仓价**优先取下游 `trading_position.cost_price`**，现价只在下游给不出成本价时兜底，并在 `price_source` 与 `note` 里注明是估的。 + +这条不是细节：拿现价当成本，摊薄成本就等于当天价、**安全垫齐刷刷是 0**——而安全垫是盈利加仓（≥3%）、保垫减仓（峰值≥6%）、补仓评估档（−8%/−15%）共同的判断依据。一只真实成本 20、现价 10 的票（实亏 50%）会被记成「不赚不亏」，该评估的补仓不评估；页面和日报上的浮动盈亏也全是 0。2026-07-29 首次接管 22 只持仓时踩到，已修并有单测守着。 + ## 自主提议的分流（设计 §6 / §7） 动作引擎每分钟扫一遍持仓，产出 FILL / ADD / DCA / TRIM 四类候选，然后依次过三道： diff --git a/app/core/recon.py b/app/core/recon.py index b663b59..78ec169 100644 --- a/app/core/recon.py +++ b/app/core/recon.py @@ -217,22 +217,35 @@ def diff_positions(book_rows: list, ds_rows: list) -> list: return out -def build_recon_fixes(diffs: list, price_map=None, lots_map=None) -> list: +def build_recon_fixes(diffs: list, price_map=None, lots_map=None, cost_map=None) -> list: """按差异生成修正动作 (以下游为准, 全部留痕)。 - delta > 0 → 新增 RECON 批次补足 (价格取现价/成本代理, 无价则标 NEED_PRICE) - delta < 0 → 按核销次序冲销 |delta| 股 + delta > 0 → 新增 RECON 批次补足 + delta < 0 → 按核销次序冲销 |delta| 股 (用现价算已实现盈亏) + + **补仓位时开仓价必须优先取下游的成本价, 不能拿现价充数。** 这条踩过: + 2026-07-29 首次把下游 22 只已有持仓接进账本时, RECON 批次全按现价开仓, 于是每只票的 + 摊薄成本都等于当天现价、安全垫齐刷刷是 0 —— 而安全垫是盈利加仓(≥3%)、保垫减仓(峰值≥6%)、 + 补仓评估档(−8%/−15%) 共同的判断依据。一只真实成本 20、现价 10 的票 (实亏 50%) 会被记成 + "不赚不亏", 该评估的补仓不评估; 页面与日报上的浮动盈亏也全是 0。 + 下游 `trading_position.cost_price` 本来就有这个数, 没有理由不用。 + 现价只在下游给不出成本价时兜底, 并在 note 里注明来源 —— 账本上要看得出这笔成本是估的。 """ price_map = price_map or {} lots_map = lots_map or {} + cost_map = cost_map or {} fixes = [] for d in diffs or []: code, delta = d["ts_code"], d["delta"] px = float(price_map.get(code) or 0) if delta > 0: + cost = float(cost_map.get(code) or 0) + open_px, src = (cost, "下游成本价") if cost > 0 else (px, "现价兜底(下游无成本价)") fixes.append({"ts_code": code, "op": "ADD_RECON_LOT", "qty": delta, - "price": px, "need_price": px <= 0, - "note": f"对账修正(+{delta}股): 以下游为准补入 RECON 批次"}) + "price": open_px, "need_price": open_px <= 0, + "price_source": src, + "note": f"对账修正(+{delta}股): 以下游为准补入 RECON 批次, " + f"开仓价取{src}"}) else: lots = lots_map.get(code) or [] res = apply_sell_to_lots(lots, -delta) diff --git a/app/services/ledger_service.py b/app/services/ledger_service.py index e8900db..6b9e535 100644 --- a/app/services/ledger_service.py +++ b/app/services/ledger_service.py @@ -258,10 +258,13 @@ def reconcile(*, apply_fix: bool = True) -> dict: codes = [d["ts_code"] for d in diffs] prices = market.get_prices(codes) + # 下游的成本价 —— 补仓位时的开仓价优先取它, 现价只兜底 (见 rc.build_recon_fixes 注释: + # 拿现价当成本会让安全垫齐刷刷归零, 整条纪律链跟着失灵) + costs = {r["ts_code"]: r.get("cost") for r in (ds.get("rows") or [])} lots_map = {c: [{"lot_id": l["id"], "lot_type": l["lot_type"], "qty": int(l["qty"]), "open_date": _date_key(l["open_date"])} for l in pms_repo.list_lots(c, status="OPEN")] for c in codes} - fixes = rc.build_recon_fixes(diffs, price_map=prices, lots_map=lots_map) + fixes = rc.build_recon_fixes(diffs, price_map=prices, lots_map=lots_map, cost_map=costs) for f in fixes: try: _apply_fix(f) diff --git a/scripts/reset_ledger.py b/scripts/reset_ledger.py index 08754be..ec5fc24 100644 --- a/scripts/reset_ledger.py +++ b/scripts/reset_ledger.py @@ -100,6 +100,14 @@ def main(): print(f" FAIL 游标重置: {type(e).__name__}: {e}") failed.append(CURSOR_KEY) + # 对账连续不一致天数归零 —— 账本刚清空, 上一轮攒下的 streak 会让下一次对账直接判 ERROR + try: + from app.services import param_store + param_store.set_param("PMS_RECON_STREAK", 0, "reset_ledger") + print(" OK PMS_RECON_STREAK 归零 (否则下次对账会带着旧的连续天数直接升 ERROR)") + except Exception as e: + print(f" WARN 对账连续天数归零失败: {type(e).__name__}: {e}") + # ws 通道状态一并归零 —— 指令与 inbox 都清了, 留着旧水位会让补发对不上 try: execute("UPDATE pms_ws_state SET last_seq = 0, acked_seq = 0, server_seq = 0, " diff --git a/scripts/test_wiring.py b/scripts/test_wiring.py index ea149b9..8a91b11 100644 --- a/scripts/test_wiring.py +++ b/scripts/test_wiring.py @@ -742,6 +742,39 @@ def _(): downstream_repo.fetch_filled_orders = orig +@case("账本服务·对账补仓位取下游成本价, 不拿现价充数") +def _(): + from app.core import recon as rc + from app.repo import downstream_repo + from app.services import ledger_service as ls, market + fake = install_fakes(prices={"600000.SH": 10.0, "000001.SZ": 30.0}) + # 下游有两只票: A 真实成本 20 (现价 10, 实亏 50%), B 下游没给成本价 + downstream_repo.fetch_positions = lambda: { + "columns": {"qty": "total_quantity", "cost": "cost_price"}, "raw_count": 2, + "rows": [{"ts_code": "600000.SH", "qty": 1000, "avail_qty": 1000, "cost": 20.0, + "price": 10.0}, + {"ts_code": "000001.SZ", "qty": 500, "avail_qty": 500, "cost": None, + "price": 30.0}]} + r = ls.reconcile(apply_fix=True) + assert r["ok"], r + by = {f["ts_code"]: f for f in r["fixes"]} + # A: 必须记 20 而不是 10 —— 记成 10 的话安全垫是 0, 这只实亏 50% 的票就永远不进补仓评估 + assert by["600000.SH"]["price"] == 20.0, by["600000.SH"] + assert by["600000.SH"]["price_source"] == "下游成本价" + # B: 下游没成本价才退到现价, 且留痕注明是估的 + assert by["000001.SZ"]["price"] == 30.0 and "兜底" in by["000001.SZ"]["price_source"] + # 落到账本上: 摊薄成本 = 真实成本, 安全垫 = (10-20)/20 = -50% + pos = fake.positions["600000.SH"] + assert abs(float(pos["avg_cost"]) - 20.0) < 1e-6, pos + assert abs(float(pos["cushion_pct"]) - (-0.5)) < 1e-4, pos + + # 纯逻辑侧: cost_map 缺项时逐只独立回退, 不会一只没成本就全退现价 + fixes = rc.build_recon_fixes( + [{"ts_code": "A", "delta": 100}, {"ts_code": "B", "delta": 200}], + price_map={"A": 1.0, "B": 2.0}, cost_map={"A": 9.0}) + assert [f["price"] for f in fixes] == [9.0, 2.0], fixes + + @case("账本服务·对账以下游为准 + 连续不一致升级") def _(): from app.services import ledger_service as ls