修复: 建表脚本/redis RESP3/参数缓存穿透/回放认领时间守卫

QMT_INTERFACE_REQUIREMENTS.md

@@ -10,8 +10,8 @@
 
 | # | 数据项 | 需要的字段 | 期望更新时延 | 用途 | 答复 |
 |---|---|---|---|---|---|
-| A1 | 持仓快照 `trading_position` | 股票代码（点式）、持仓数量、**可用数量（T+1 可卖）**、成本价（如有）、冻结数量（如有） | 盘中 ≤ 5 分钟 | 账本对账基准。当前 PMS 只确认过 `stock_code` 列可用，**请提供该表完整字段定义（DDL）**，尤其确认是否已有"可用数量"列——若无，PMS 自行按 T+1 规则推算 | |
+| A1 | 持仓快照 `trading_position` | 股票代码（点式）、持仓数量、**可用数量（T+1 可卖）**、成本价（如有）、冻结数量（如有） | 盘中 ≤ 5 分钟 | 账本对账基准。当前 PMS 只确认过 `stock_code` 列可用，**请提供该表完整字段定义（DDL）**，尤其确认是否已有"可用数量"列——若无，PMS 自行按 T+1 规则推算 | **PMS 侧实机探明（2026-07-27，待下游确认语义）**：该表 14 列，含 `stock_code / stock_name / total_quantity / available_quantity / frozen_quantity / cost_price / market_price / market_value / profit_loss`。**可用数量列已存在**（`available_quantity`），本项数据需求实质已满足，PMS 已按此列对接。仍请确认：①`available_quantity` 是否即 T+1 可卖口径 ②更新时延 ③`stock_code` 格式（点式/前缀式） |
-| A2 | 委托与成交 `trading_order` | 委托号、股票代码、方向、委托价、委托量、状态（**完整状态枚举文档**）、成交量、成交均价、委托/成交时间、来源标识 | 状态变更后 ≤ 1 分钟 | 成交回放入账（批次/成本）、指令执行确认。**请提供完整 DDL 与状态流转说明**（现掌握的 submitted/filled/completed/pending/failed 为推断口径，需正式确认） | |
+| A2 | 委托与成交 `trading_order` | 委托号、股票代码、方向、委托价、委托量、状态（**完整状态枚举文档**）、成交量、成交均价、委托/成交时间、来源标识 | 状态变更后 ≤ 1 分钟 | 成交回放入账（批次/成本）、指令执行确认。**请提供完整 DDL 与状态流转说明**（现掌握的 submitted/filled/completed/pending/failed 为推断口径，需正式确认） | **PMS 侧实机探明**：该表 20 列，已见 `order_id / strategy_id / stock_code / stock_name / order_side / order_quantity / order_price / target_price / order_status` 等。**仍缺两项，是本清单里最卡 PMS 的**：①状态枚举的正式定义 ②**成交来源标识**（哪条委托对应 PMS 的哪条指令）。在来源标识到位前，PMS 只能按「同股同向 + 下发早于成交 + FIFO」贪心认领，认领不上即判外部成交并告警——人工与 PMS 同时下单时会误判 |
 | A3 | 账户资金快照（**新增需求**） | 总资产、可用资金、冻结资金、当日卖出可用资金（T+0 回笼） | 盘中 ≤ 5 分钟；日终必须 | 总规模校准与买入前资金校验。形式不限：新表 / 现有表 / HTTP 接口均可，请给出可行方案 | |
 | A4 | 成交回报明细（可选） | 若 A2 已含逐笔或聚合成交（成交量/均价），本项可免；否则请提供逐笔成交表 | 同 A2 | 部分成交场景的精确入账 | |
 | A5 | 上游买入计划 `trading_buy_plan` | PMS 将作为该表的承接方（替代原决策系统 ENTRY_GATE 的角色）。请确认：①下游当前是否仍轮询 `is_active=6` 自动挂单？②切换后是否可以**停止**该轮询（统一走 B1 通道），或保留作为过渡（方案 X） | — | 旧通道处置（见 C2） | |
@@ -67,6 +67,6 @@ CREATE TABLE pms_order_request (
 
 | # | 文档 | 说明 | 答复 |
 |---|---|---|---|
-| D1 | `trading_position` / `trading_order` / `trading_buy_plan` 完整 DDL | 含索引与状态枚举正式定义 | |
+| D1 | `trading_position` / `trading_order` / `trading_buy_plan` 完整 DDL | 含索引与状态枚举正式定义 | **列名已由 PMS 实机探明**（`docker compose run --rm pms-web python scripts/check_db.py` 的 [3] 段会打印三表完整列定义，可直接贴进本表）：trading_position 14 列 / trading_order 20 列 / trading_buy_plan 25 列。**仍需下游正式提供的是语义而非列名**：状态枚举取值与流转、各数量列口径、索引 |
 | D2 | 下游执行行为说明 | 委托拆单逻辑（如有）、涨跌停处理、集合竞价时段行为 | |
 | D3 | 现有信号消费点清单 | 下游当前消费决策系统信号的全部位置（用于 C2 停用核对，防遗漏） | |

README.md

@@ -42,6 +42,7 @@ scripts/
   test_core_units.py   仓位与安全垫核心逻辑                        14 例
   test_batch2_units.py 命令 / 方案 / 回放对账 纯逻辑                35 例
   test_wiring.py       装配自检: 服务层→核心→落表 全链路 (内存桩)   18 例
+  init_db.py           建表 (应用 ddl_pms_v1.sql, 幂等, 默认演练)
   check_db.py          实机连通性与表结构自检 (需真实 .env)
 ```
 
@@ -63,7 +64,9 @@ docker compose build                       # 默认走清华 PyPI 镜像; 可 --
 
 # 构建验证 (不连库, 秒级): 应输出 ALL SUITES PASS
 docker compose run --rm pms-web python scripts/run_tests.py
-# 实机自检 (连库, 需 .env): 三库连通 + pms_* 十表 + 下游表结构 + 行情 Redis
+# 建表 (幂等; 不加 --yes 只演练打印)
+docker compose run --rm pms-web python scripts/init_db.py --yes
+# 实机自检 (连库, 需 .env): 库连通 + pms_* 十表 + 下游表完整列定义 + 行情 Redis
 docker compose run --rm pms-web python scripts/check_db.py
 
 docker compose up -d                       # 管理页面
@@ -79,6 +82,8 @@ git pull && docker compose build && docker compose up -d
 
 基础镜像 `python:3.11-slim` 拉取慢时，先给服务器 Docker 配置 registry 镜像加速。日志落 `./logs`（已挂载卷）；容器时区 Asia/Shanghai。管理页面的前端资源（Vue3 / ElementPlus / axios）走 unpkg CDN，浏览器需能访问外网；若内网隔离，把页面头部三行 `<script>/<link>` 换成本地文件即可（页面本身无构建步骤）。
 
+**依赖版本坑**：`redis` 必须锁在 5.x。redis-py 6.x 默认按 RESP3 握手（先发 `HELLO 3`），而 208/150 两台 Redis 是 6.0 以前的版本，会回 `unknown command HELLO`，行情与 Celery 总线一起断。`requirements.txt` 已锁版本，代码侧另有 `protocol=2` 双保险；升级依赖时别把这条放开。
+
 ## 调度总表（`--profile sched` 生效，设计 §10）
 
 | 调度 | 时间 | 任务 | 本批状态 |

app/repo/downstream_repo.py

@@ -4,23 +4,26 @@
 ==============================================================================
 这三张表**归下游系统维护**, PMS 只读 (设计 §9)。
 
-列名不确定性 (重要):
+列名口径 (2026-07-27 实机 SHOW COLUMNS 确认, 见 check_db.py 输出):
-  QMT_INTERFACE_REQUIREMENTS A1/A2/D1 仍在协商 —— 目前只确认 trading_position.stock_code
+  trading_position: id, stock_code, stock_name, total_quantity, available_quantity,
-  与 trading_order 的 order_* 系列列。因此本模块**不硬编码持仓数量列名**, 而是先取一行
+                    frozen_quantity, cost_price, market_price, market_value, profit_loss, ...
-  探测列名 (候选列表按常见命名排优先级), 并把探测结果回传给上层, 在页面与日报显式展示
+  → **可用数量列 available_quantity 下游已提供**, QMT 需求清单 A1 的核心疑问就此落地,
-  「本次对账用的是哪一列」。协商答复到位后, 把 PMS_DS_QTY_COL 参数写死即可停用探测。
+    PMS 无需自行按 T+1 推算下游可卖量 (账本侧仍按 T+1 自行维护, 对账时以下游为准)。
+探测机制保留: 候选表把已确认列名排在首位, 万一下游改名或换表也能自动兜住, 探测结果经
+「运维 → 导出下游表结构」可见, 也是回填 D1 的现成材料。
 """
 from __future__ import annotations
 
 from app.db.session import fetch_all, fetch_one
 
-# 数量/可用量列名候选 (探测顺序即优先级)
+# 数量/可用量列名候选 (探测顺序即优先级; 首项为实机确认的真实列名)
-QTY_CANDIDATES = ["current_qty", "total_qty", "position_qty", "hold_qty", "stock_qty",
+QTY_CANDIDATES = ["total_quantity", "current_qty", "total_qty", "position_qty", "hold_qty",
-                  "volume", "quantity", "qty", "position_volume", "hold_volume"]
+                  "stock_qty", "volume", "quantity", "qty", "position_volume", "hold_volume"]
-AVAIL_CANDIDATES = ["available_qty", "avail_qty", "can_use_volume", "available_volume",
+AVAIL_CANDIDATES = ["available_quantity", "available_qty", "avail_qty", "can_use_volume",
-                    "sellable_qty", "enable_amount", "can_sell_qty"]
+                    "available_volume", "sellable_qty", "enable_amount", "can_sell_qty"]
 COST_CANDIDATES = ["cost_price", "avg_cost", "open_price", "position_cost", "cost"]
-FROZEN_CANDIDATES = ["frozen_qty", "frozen_volume", "freeze_qty"]
+FROZEN_CANDIDATES = ["frozen_quantity", "frozen_qty", "frozen_volume", "freeze_qty"]
+PRICE_CANDIDATES = ["market_price", "last_price", "current_price", "price"]
 
 FILLED_STATUSES = ("completed", "filled")
 
@@ -76,6 +79,7 @@ def fetch_positions() -> dict:
     avail_col = _pick(keys, AVAIL_CANDIDATES)
     cost_col = _pick(keys, COST_CANDIDATES)
     frozen_col = _pick(keys, FROZEN_CANDIDATES)
+    price_col = _pick(keys, PRICE_CANDIDATES)
     code_col = _pick(keys, ["stock_code", "ts_code", "code", "security_code"]) or "stock_code"
 
     out = []
@@ -89,10 +93,11 @@ def fetch_positions() -> dict:
             "avail_qty": _int(r.get(avail_col)) if avail_col else None,
             "cost": _float(r.get(cost_col)) if cost_col else None,
             "frozen": _int(r.get(frozen_col)) if frozen_col else None,
+            "price": _float(r.get(price_col)) if price_col else None,
         })
     return {"rows": out, "raw_count": len(rows),
             "columns": {"code": code_col, "qty": qty_col, "avail": avail_col,
-                        "cost": cost_col, "frozen": frozen_col}}
+                        "cost": cost_col, "frozen": frozen_col, "price": price_col}}
 
 
 def _int(v):

app/services/ledger_service.py

@@ -81,12 +81,22 @@ def replay_fills(*, limit: int = 500) -> dict:
 
 
 def _open_instructions() -> list:
+    """在途指令 (回放认领的候选池)。
+
+    dispatched_at 取「下发时点 → 建单时点」, **不用 updated_at**:
+    updated_at 每次部分成交回写都会往前跳, 拿它当下发时点会让同一条指令的后续成交
+    被时间守卫挡在门外, 误判成外部成交。建单时点是天然的下界, 宁松勿紧。
+    """
     rows = pms_repo.list_instructions(statuses=list(LIVE_INSTR), limit=500)
-    return [{"instruction_id": r["instruction_id"], "ts_code": r["ts_code"],
+    out = []
-             "side": r.get("side"), "qty": int(r.get("qty") or 0),
+    for r in rows:
-             "exec_qty": int(r.get("exec_qty") or 0), "action": r.get("action"),
+        prog = r.get("progress") or {}
-             "dispatched_at": str(r.get("updated_at") or r.get("created_at") or "")}
+        out.append({"instruction_id": r["instruction_id"], "ts_code": r["ts_code"],
-            for r in rows]
+                    "side": r.get("side"), "qty": int(r.get("qty") or 0),
+                    "exec_qty": int(r.get("exec_qty") or 0), "action": r.get("action"),
+                    "dispatched_at": str(prog.get("dispatched_at")
+                                         or r.get("created_at") or "")})
+    return out
 
 
 def _apply_action(act: dict):

app/services/market.py

@@ -29,17 +29,27 @@ QUOTE_KEY = "tushare:rt_min:1MIN:{code}"
 
 
 def _r():
+    """行情 Redis 客户端。
+
+    强制 RESP2 (protocol=2): 目标 Redis 是 6.0 以前的版本, 不认 RESP3 的 HELLO 命令,
+    而 redis-py 6.x 默认按 RESP3 握手, 会直接报 "unknown command `HELLO`" 全盘取不到价。
+    requirements.txt 已把 redis 锁在 5.x, 这里再显式声明一道 (老版本 redis-py 不认这个
+    参数, 用 TypeError 兜回去)。
+    """
     global _redis
     if _redis is not None:
         return _redis
     with _lock:
         if _redis is None:
             import redis
-            _redis = redis.Redis(
+            kw = dict(host=settings.SIGNAL_REDIS_HOST, port=settings.SIGNAL_REDIS_PORT,
-                host=settings.SIGNAL_REDIS_HOST, port=settings.SIGNAL_REDIS_PORT,
+                      password=settings.SIGNAL_REDIS_PASSWORD or None,
-                password=settings.SIGNAL_REDIS_PASSWORD or None,
+                      db=settings.SIGNAL_REDIS_DB_QUOTES, decode_responses=True,
-                db=settings.SIGNAL_REDIS_DB_QUOTES, decode_responses=True,
+                      socket_timeout=settings.SIGNAL_REDIS_SOCKET_TIMEOUT)
-                socket_timeout=settings.SIGNAL_REDIS_SOCKET_TIMEOUT)
+            try:
+                _redis = redis.Redis(protocol=2, **kw)
+            except TypeError:
+                _redis = redis.Redis(**kw)
     return _redis
 
 

app/services/param_store.py

@@ -77,7 +77,10 @@ DESC = {
     "PMS_REPLAY_INTERVAL_MIN": "成交回放间隔 (分钟)", "PMS_RECON_ALARM_DAYS": "连续不一致升级天数",
 }
 
-_cache = {"at": 0.0, "data": {}, "error": None}
+# loaded 标记必不可少: 不能用「data 是否为空」判断缓存是否有效 ——
+# 参数表为空 (还没在页面改过参数) 或读表失败时 data 都是 {}, 那样每次 get() 都会穿透去连库,
+# 健康时是白白打表, 故障时是刷屏重试。实测 check_db 一次跑出 50 条重复告警即此故。
+_cache = {"at": 0.0, "data": {}, "error": None, "loaded": False}
 _lock = threading.Lock()
 
 
@@ -117,15 +120,19 @@ def _type_of(key: str):
 def refresh(force: bool = False) -> dict:
     """拉一次表值 (带 TTL 缓存)。连库失败不抛异常 —— 退回文件初值并记 error。"""
     now = time.time()
-    if not force and now - _cache["at"] < CACHE_TTL and _cache["data"]:
+    if not force and _cache["loaded"] and now - _cache["at"] < CACHE_TTL:
         return _cache["data"]
     with _lock:
         try:
             rows = pms_repo.all_params()
-            _cache.update({"data": rows, "at": now, "error": None})
+            _cache.update({"data": rows, "at": now, "error": None, "loaded": True})
         except Exception as e:
-            _cache.update({"at": now, "error": f"{type(e).__name__}: {e}"})
+            msg = f"{type(e).__name__}: {e}"
-            logger.warning("参数表读取失败, 退回 settings 初值: %s", e)
+            changed = _cache.get("error") != msg
+            # 失败也要更新时间戳并置 loaded, 否则下一次 get() 立刻又去连库
+            _cache.update({"at": now, "error": msg, "loaded": True})
+            if changed:      # 同一个错误只吼一次, 恢复或换错才再吼
+                logger.warning("参数表读取失败, 退回 settings 初值: %s", e)
     return _cache["data"]
 
 
@@ -197,7 +204,7 @@ def set_param(key: str, value, updated_by: str = "user") -> dict:
         pms_repo.set_param(key, v, updated_by)
     except Exception as e:
         return {"ok": False, "error": f"写入失败: {type(e).__name__}: {e}"}
-    _cache["at"] = 0.0
+    _cache["loaded"] = False
     refresh(force=True)
     return {"ok": True, "key": key, "value": v}
 

config/settings.py

@@ -24,7 +24,10 @@ class Settings(BaseSettings):
     #   153 代理: pms_* 全部表 + trading_position / trading_order / trading_buy_plan /
     #   strategy_daily_results (决策系统结论) —— 严格单表访问, 代理禁多表联查
     SOURCE_DB_EXT_DSN: str = "mysql+pymysql://user:pass@192.168.16.150:3306/factordb_mysql"
-    #   因子分表 gp_stock_factor_pro_* (自算参考位/MA/ATR/量比)
+    #   因子库直连。**PMS 当前不使用**: 因子分表 gp_stock_factor_pro_YYYYMM 经
+    #   ShardingSphere 动态分配物理落点, 直连单台会在分片变动的月份漏表, 必须走
+    #   PROXY_DB_URL 逐表单查 (沿用 bionic_trader 的踩坑结论)。保留本键仅为与 bionic
+    #   的 .env 对齐, 连不通不影响任何功能。
     DB_MYSQL_URL: str = "mysql+pymysql://user:pass@192.168.18.199:3306/db_gp_cj"
     #   大盘指数 zs_day_data (页面区制提示用, 非约束)
 

requirements.txt

@@ -2,7 +2,10 @@ pydantic>=2.5
 pydantic-settings>=2.1
 SQLAlchemy>=2.0
 PyMySQL>=1.1
-redis>=5.0
+# redis-py 6.x 默认按 RESP3 握手 (先发 HELLO 3), 而 208/150 两台 Redis 是 6.0 以前的版本,
+# 会回 "unknown command HELLO" 导致行情与 Celery 总线全断 —— 锁在 5.x (默认 RESP2)。
+# 代码侧另有 protocol=2 双保险 (app/services/market.py)。
+redis>=5.0,<6.0
 celery>=5.3
 fastapi>=0.110
 uvicorn>=0.27

scripts/check_db.py

@@ -43,8 +43,8 @@ def main():
     from app.db import session as dbs
 
     print("\n[1] 数据库连通性")
-    for name, label in (("proxy", "153 代理 (pms_* / trading_*)"),
+    for name, label in (("proxy", "153 代理 (pms_* / trading_* / 因子分表)"),
-                        ("factor", "因子分表库"), ("index", "大盘指数库")):
+                        ("index", "大盘指数库")):
         r = dbs.ping(name)
         if r["ok"]:
             line("OK", f"{label}")
@@ -52,31 +52,51 @@ def main():
             fail(f"{label} 连接失败: {r['error']}")
         else:
             warn(f"{label} 连接失败 (非致命): {r['error']}")
+    rf = dbs.ping("factor")
+    line("OK" if rf["ok"] else "INFO",
+         "因子库直连 (SOURCE_DB_EXT_DSN) —— PMS 当前**不使用**: 因子分表经 "
+         "ShardingSphere 动态分片, 必须走 153 代理读 (沿用 bionic 结论), "
+         f"故此项连不通不影响任何功能。当前: {'通' if rf['ok'] else rf['error'].splitlines()[0]}")
 
     print("\n[2] PMS 自有表")
+    missing_tables = []
     for t in PMS_TABLES:
         try:
             r = dbs.fetch_one(f"SELECT COUNT(*) AS n FROM {t}")
             line("OK", f"{t:<20} {r['n']} 行")
         except Exception as e:
-            fail(f"{t} 不可用: {type(e).__name__}: {e}")
+            msg = str(e)
+            if "does not exist" in msg or "doesn't exist" in msg:
+                missing_tables.append(t)
+                fail(f"{t} 不存在")
+            else:
+                fail(f"{t} 不可用: {type(e).__name__}: {e}")
+    if missing_tables:
+        line("", "→ 建表: docker compose run --rm pms-web python scripts/init_db.py --yes")
 
-    print("\n[3] 下游只读表")
+    print("\n[3] 下游只读表 (完整列定义, 可直接回填 QMT 需求清单 D1)")
     from app.repo import downstream_repo
     for t in DOWNSTREAM:
         try:
             cols = downstream_repo.describe(t)
-            line("OK", f"{t:<20} {len(cols)} 列: "
+            line("OK", f"{t} —— {len(cols)} 列")
-                       f"{', '.join(str(c.get('Field')) for c in cols[:10])}"
+            for c in cols:
-                       f"{' ...' if len(cols) > 10 else ''}")
+                line("", f"    {str(c.get('Field')):<26} {str(c.get('Type')):<18} "
+                         f"{'NULL' if str(c.get('Null')).upper() == 'YES' else 'NOT NULL':<9}"
+                         f"{str(c.get('Comment') or '')[:40]}")
         except Exception as e:
             warn(f"{t} 读取失败: {type(e).__name__}: {e}")
     try:
         ds = downstream_repo.fetch_positions()
-        line("OK" if ds["columns"].get("qty") else "WARN",
-             f"持仓列探测: {ds['columns']} (共 {ds['raw_count']} 行)")
         if ds["raw_count"] and not ds["columns"].get("qty"):
-            warn("未识别出持仓数量列 —— 请按 QMT 需求清单 A1/D1 取得 DDL 后补 QTY_CANDIDATES")
+            warn(f"未识别出持仓数量列 {ds['columns']} —— 把真实列名补进 "
+                 f"downstream_repo.QTY_CANDIDATES")
+        else:
+            line("OK", f"持仓列映射: {ds['columns']} (共 {ds['raw_count']} 行)")
+            sample = ds["rows"][:3]
+            for s in sample:
+                line("", f"    {s['ts_code']:<12} 持仓 {s['qty']} 可用 {s['avail_qty']} "
+                         f"成本 {s['cost']} 现价 {s['price']}")
     except Exception as e:
         warn(f"持仓快照读取失败: {e}")
 
@@ -89,12 +109,19 @@ def main():
             pos = [p["ts_code"] for p in pms_repo.list_positions(only_open=True)][:3]
         except Exception:
             pass
+        if not pos:
+            try:
+                pos = [r["ts_code"] for r in downstream_repo.fetch_positions()["rows"][:3]]
+            except Exception:
+                pos = []
         probe = pos or ["600000.SH"]
         got = {c: market.get_price(c) for c in probe}
         if any(v for v in got.values()):
             line("OK", f"实时价样本: {got}")
         else:
-            warn(f"取不到实时价样本 {got} (非交易时段属正常)")
+            warn(f"取不到实时价样本 {got} (非交易时段属正常; 若报 unknown command HELLO, "
+                 f"说明 redis-py 版本过新走了 RESP3, 重新 build 镜像即可 —— "
+                 f"requirements 已锁 redis<6)")
     except Exception as e:
         warn(f"Redis 行情不可用: {type(e).__name__}: {e}")
 

scripts/init_db.py

@@ -0,0 +1,145 @@
+# -*- coding: utf-8 -*-
+"""
+建表脚本: 把 ddl_pms_v1.sql 应用到 153 代理侧
+===============================================
+运行:
+  docker compose run --rm pms-web python scripts/init_db.py          # 演练, 只打印不执行
+  docker compose run --rm pms-web python scripts/init_db.py --yes    # 实际建表
+  docker compose run --rm pms-web python scripts/init_db.py --yes --only pms_command,pms_plan
+
+说明:
+  * 全部语句为 `CREATE TABLE IF NOT EXISTS`, 重复执行安全 (已存在的表不会被改动)。
+  * 直接用引擎执行, 绕过 db.session 的单表守卫 —— DDL 本身就只涉及一张表,
+    但语句里的中文注释可能撞上守卫的关键字启发式, 没必要让它误伤。
+  * ShardingSphere-Proxy 对 DDL 的支持视配置而定。若某条语句被代理拒绝,
+    脚本会把完整语句打出来, 直接拿去物理库 (my_quant_db) 执行即可, 其余语句不受影响。
+"""
+import argparse
+import os
+import re
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+DDL_FILE = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+                        "ddl_pms_v1.sql")
+
+
+def split_sql(text: str) -> list:
+    """按分号切语句, 但**跳过字符串字面量里的分号**。
+
+    DDL 的列注释里有 `COMMENT '参数命令: EFFECTIVE/SUPERSEDED; 任务命令: PENDING/...'`,
+    直接 split(";") 会把一条 CREATE TABLE 劈成三段残句 —— 必须带引号状态机。
+    """
+    out, buf, in_str, i = [], [], False, 0
+    while i < len(text):
+        ch = text[i]
+        if in_str:
+            if ch == "\\" and i + 1 < len(text):        # 反斜杠转义
+                buf.append(text[i:i + 2])
+                i += 2
+                continue
+            if ch == "'":
+                if i + 1 < len(text) and text[i + 1] == "'":   # '' 转义
+                    buf.append("''")
+                    i += 2
+                    continue
+                in_str = False
+            buf.append(ch)
+        elif ch == "'":
+            in_str = True
+            buf.append(ch)
+        elif ch == ";":
+            out.append("".join(buf))
+            buf = []
+        else:
+            buf.append(ch)
+        i += 1
+    out.append("".join(buf))
+    return out
+
+
+def parse_statements(sql_text: str) -> list:
+    """去掉整行 `--` 注释后切语句。返回 [(表名, 语句)]; 非 CREATE TABLE 的残句会被标 '?'。"""
+    body = "\n".join(ln for ln in sql_text.splitlines() if not ln.strip().startswith("--"))
+    out = []
+    for stmt in split_sql(body):
+        s = stmt.strip()
+        if not s:
+            continue
+        m = re.search(r"CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?`?([A-Za-z0-9_]+)`?", s, re.I)
+        out.append((m.group(1) if m else "?", s))
+    return out
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--yes", action="store_true", help="确认执行 (缺省只演练)")
+    ap.add_argument("--only", default="", help="只处理指定表, 逗号分隔")
+    args = ap.parse_args()
+
+    if not os.path.exists(DDL_FILE):
+        print(f"FAIL  找不到 DDL 文件: {DDL_FILE}")
+        sys.exit(1)
+    with open(DDL_FILE, encoding="utf-8") as f:
+        stmts = parse_statements(f.read())
+    only = {x.strip() for x in args.only.split(",") if x.strip()}
+    if only:
+        stmts = [(t, s) for t, s in stmts if t in only]
+    if not stmts:
+        print("FAIL  没有可执行的语句 (检查 --only 是否写对)")
+        sys.exit(1)
+    broken = [s for t, s in stmts if t == "?"]
+    if broken:
+        print(f"FAIL  解析出 {len(broken)} 条非 CREATE TABLE 残句, 说明 DDL 切分有误, "
+              f"已中止 (不执行任何语句)。首条残句:\n{broken[0][:200]}")
+        sys.exit(1)
+
+    print(f"DDL 文件: {DDL_FILE}")
+    print(f"待建表 {len(stmts)} 张: {', '.join(t for t, _ in stmts)}")
+    if not args.yes:
+        print("\n[演练模式] 未执行任何语句。确认无误后加 --yes 重跑:")
+        print("  docker compose run --rm pms-web python scripts/init_db.py --yes")
+        return
+
+    # 连库依赖放到演练之后再导入: 演练模式只做语句切分校验, 无依赖环境也能跑
+    from sqlalchemy import text
+    from app.db.session import get_engine
+
+    eng = get_engine("proxy")
+    okc, failed = 0, []
+    print()
+    for tbl, stmt in stmts:
+        try:
+            with eng.begin() as c:
+                c.execute(text(stmt))
+            print(f"  OK    {tbl}")
+            okc += 1
+        except Exception as e:
+            print(f"  FAIL  {tbl}: {type(e).__name__}: {e}")
+            failed.append((tbl, stmt, str(e)))
+
+    print("\n[验证] 逐表 COUNT(*)")
+    missing = []
+    for tbl, _ in stmts:
+        try:
+            with eng.connect() as c:
+                n = c.execute(text(f"SELECT COUNT(*) FROM {tbl}")).scalar()
+            print(f"  OK    {tbl:<20} {n} 行")
+        except Exception as e:
+            print(f"  FAIL  {tbl:<20} {type(e).__name__}: {e}")
+            missing.append(tbl)
+
+    print("\n" + "-" * 62)
+    if failed:
+        print(f"以下 {len(failed)} 张表建失败, 完整语句如下 —— 可直接拿到物理库执行:")
+        for tbl, stmt, err in failed:
+            print(f"\n### {tbl}  ({err.splitlines()[0]})\n{stmt};")
+    if missing:
+        print(f"\nFAILED: {len(missing)} 张表仍不可用: {', '.join(missing)}")
+        sys.exit(1)
+    print(f"ALL OK: {okc} 张表就绪, 接着跑 python scripts/check_db.py 做完整自检")
+
+
+if __name__ == "__main__":
+    main()

scripts/test_wiring.py

@@ -332,7 +332,7 @@ def install_fakes(prices=None, positions=None, params=None):
     downstream_repo.fetch_positions = lambda: {"rows": [], "columns": {"qty": None},
                                                "raw_count": 0}
     downstream_repo.fetch_refs = lambda c: None
-    param_store._cache.update({"at": 0.0, "data": {}, "error": None})
+    param_store._cache.update({"at": 0.0, "data": {}, "error": None, "loaded": False})
     portfolio.save_neg_streak({})
     return fake
 
@@ -583,6 +583,12 @@ def _():
     fake.insert_instruction(instruction_id="INS_A", origin_type="plan", origin_id="P1",
                             ts_code="600000.SH", action="OPEN", side="buy", qty=6000,
                             limit_price=10.0, status="DISPATCHED")
+    # 建单时点必须显式钉死: 用 datetime.now() 会让本用例的通过与否取决于跑测试的钟点
+    # (时间守卫要求 下发 ≤ 成交), 17:12 跑就会误判成外部成交 —— 实机上已踩到。
+    fake.instructions["INS_A"]["created_at"] = "2026-07-27 09:30:00"
+    # updated_at 故意置在成交之后: 认领必须看建单时点, 不能看 updated_at
+    # (否则部分成交回写一次 updated_at, 后续成交就会被自己挡住)
+    fake.instructions["INS_A"]["updated_at"] = "2026-07-27 23:59:59"
     fills = [{"order_id": 101, "ts_code": "600000.SH", "side": "buy", "qty": 6000,
               "price": 10.0, "done_time": "2026-07-27 09:40:00"}]
     orig = downstream_repo.fetch_filled_orders