模拟库和正式库分离

DEVLOG.md

@@ -1033,8 +1033,23 @@ PMS 业务库 (pms_* 必须新库)、Celery 总线 redis db (db8→db9, 同 db
 硬编码 38100 改走 PORT 变量。**默认值与从前逐字一致, 现有部署零感知**; 将来两套要
 合并到一台机器时也不用再改代码。
 
+同日随拍板收敛到最终口径 —— **两套系统共用 my_quant_db, 全部靠表名区分**:
+① 下游表: 正式 QMT 在原库写另一套表名, 四张下游只读表的表名做成 .env 可配
+   (settings.PMS_DS_TABLE_* → downstream_repo 统一 T_* 常量, 白名单校验, 不合法直接
+   拒绝启动 —— 宁可大声失败, 不能静默退回默认表名把另一套系统的账读进来);
+② PMS 自有表: 表名前缀 (PMS_TABLE_PREFIX="real_" → real_pms_*)。全仓库 129 处表名
+   引用**不逐条改**, 在 db.session 这个 SQL 执行唯一入口统一映射 (词边界匹配小写
+   pms_* 标识符; 空前缀零改写, 模拟行为逐字节不变; 已带前缀不二次加; 前缀白名单校验,
+   非法值拒绝启动)。两个绕过入口直执 DDL 的脚本 (init_db / migrate_archived_at) 显式
+   过 map_tables —— 不接的话正式实例 make deploy 会建出**无前缀**的表, 与模拟的表撞
+   在一起; test_batch19 加 [K2] 旁路扫描钉死"不许有第三个绕行者"。
+
 **动了哪些文件**
-docker-compose.yml、Makefile、新增 REAL_TRADING_DEPLOY_PLAN.md。业务代码零改动。
+docker-compose.yml、Makefile、config/settings.py、app/db/session.py、
+app/repo/downstream_repo.py、app/services/strategy_advisor.py、scripts/init_db.py、
+scripts/migrate_archived_at.py、scripts/probe_strategy_signals.py、
+scripts/test_batch19_units.py、scripts/run_tests.py（新增 2 例, 全套 587 例）、
+新增 REAL_TRADING_DEPLOY_PLAN.md。
 
 **部署方式**
 模拟 (tlai4090): 随下次 git pull + make deploy 自然带上, 无需专门发版。
@@ -1042,12 +1057,13 @@ docker-compose.yml、Makefile、新增 REAL_TRADING_DEPLOY_PLAN.md。业务代
 
 **真机判收**
 开发机: compose YAML 解析通过; make -n health 默认仍打 38100、PORT=38200 时打 38200;
-全量单测与本轮改动无交集 (业务代码零改动)。模拟侧部署后 `docker compose ps` 容器名
+表名映射按七种 SQL 形态现场验证 (查/插/改/DDL/反引号/大写参数键名/幂等), 映射后仍过
-应与从前完全一致 (前缀默认空)。
+单表守卫; 全量 20 套 587 例单测 ALL PASS。模拟侧空前缀零改写, 部署后行为与从前逐字
+一致; `docker compose ps` 容器名也应与从前完全一致 (前缀默认空)。
 
 **还欠着什么**
-1. 部署前要拍板/对接的三件事在方案文档第二节: 正式 QMT 写不写 153 下游表 (决定
+1. 部署前还差两件对接: 向 QMT 侧要正式下游四张表的**实际表名** (填 .env 的
-   153 代理加逻辑库还是直连 MySQL)、正式 ws 端点与密钥交换、真实账户规模与纪律参数。
+   PMS_DS_TABLE_*); 交换正式 ws 密钥对。库不用建、代理不用动 (同库表名前缀方案)。
 2. 正式盘上线初期纪律: shadow + propose_only + 新建仓 off, 影子期至少 3 个交易日。
 
 ---

REAL_TRADING_DEPLOY_PLAN.md

@@ -1,8 +1,10 @@
 # 正式盘部署方案（模拟 / 正式双系统）
 
-2026-08-28 定稿，同日按拍板结果更新：正式盘部署在 **192.168.16.188**，正式 QMT 在
+2026-08-28 定稿，同日随拍板收敛到最终口径：正式盘部署在 **192.168.16.188**，正式 QMT
-**192.168.16.98**（ws 端点 ws://192.168.16.98:9443/pms）；正式侧**会写** 153 的下游表，
+在 **192.168.16.98**（ws 端点 ws://192.168.16.98:9443/pms）；**两套系统共用 my_quant_db
-所以 153 代理上加新逻辑库（见第二节）。现有这套 PMS 保留为**模拟仓**原地不动。
+一个库，全部靠表名区分**——正式 PMS 的 19 张自有表加表名前缀（PMS_TABLE_PREFIX=real_
+→ real_pms_*），下游表由正式 QMT 写另一套表名（PMS 侧 PMS_DS_TABLE_* 指过去）。
+不建新库、不动代理、不用跨库授权。现有这套 PMS 保留为**模拟仓**原地不动。
 **一份代码，两份 .env** —— 不开分支、不复制仓库：两份代码必然分叉，修一个账务 bug 要
 修两遍测两遍；本仓库的设计本来就是「settings 只是初值，实例身份由 .env 与各自库里的
 参数表决定」。
@@ -27,8 +29,8 @@ ws_smoke 发的联调单绕开这个开关、且带 SMOKE 前缀——成交回
 | 资源 | 模拟（现状） | 正式（新） | 说明 |
 |---|---|---|---|
 | 代码 | git 仓库 | 同一仓库 | 两边各自 `git pull`，永远同版本 |
-| PMS 业务库（pms_* 全部表） | 153 代理 my_quant_db | **153 代理新逻辑库 my_quant_db_real** | 表名固定 pms_*，只能分库 |
+| PMS 自有表（19 张） | my_quant_db 的 pms_* | **同库、加前缀 real_pms_***（PMS_TABLE_PREFIX） | 表名映射在 SQL 唯一执行入口统一做 |
-| 下游表 trading_* / strategy_daily_results | 153 代理 | 同一代理经新逻辑库映射（共享读） | 正式侧确认会写 |
+| 下游表 trading_* / strategy_daily_results | 原库、原表名 | **原库、新表名**（.env 的 PMS_DS_TABLE_* 指过去） | 正式 QMT 写一套新表名 |
 | Celery 总线 | 150 redis **db8** | 150 redis **db9** | 共用一个 db 两套 worker 互相抢任务 |
 | 决策系统信号流 | 208 db2/db3（只读共享） | 同左 | 流共享，**消费组必须不同名** |
 | 行情/大盘/mtf | 208 db13 / 199 / 214 | 同左 | 纯只读，直接共用 |
@@ -41,11 +43,14 @@ ws_smoke 发的联调单绕开这个开关、且带 SMOKE 前缀——成交回
 
 ## 二、部署前的准备件
 
-1. **153 代理加逻辑库 my_quant_db_real**：pms_* 全部表映射到一个**新建的物理库**；
+1. **库与表的接线（同库表名区分，什么都不用建）**。PROXY_DB_URL 与模拟**完全相同**
-   trading_position / trading_order / trading_buy_plan / strategy_daily_results 的映射
+   （照抄，还是 153 代理的 my_quant_db）。正式 PMS 的自有表靠前缀区分：.env 设
-   与 my_quant_db **完全相同**（共享读）。一个要核对的点：正式 QMT 写的下游表若与模拟
+   PMS_TABLE_PREFIX=real_，db.session 在 SQL 执行唯一入口把 pms_* 统一映射成
-   写的是**同一批物理表**，正式 PMS 的回放与对账会连模拟的旧行一起看见——两边下游数据
+   real_pms_*；首次 `make deploy` 建表自动带前缀建出 real_pms_* 这一套（init_db 开头
-   分不分表，要与 QMT/决策系统侧对一遍口径。
+   会打印「表名前缀 real_ 生效」；个别 DDL 若被代理拒绝，脚本会打印**带前缀的**完整
+   语句，直接拿去物理库执行即可，与模拟当年建表同一套兜底）。下游表由正式 QMT 在同库
+   写一套**新表名**，向 QMT 侧要到四张表名填进 PMS_DS_TABLE_*。模拟与正式的数据从
+   表名上就分开，不会混。
 2. **正式密钥对**：容器里 `python scripts/gen_keys.py` 生成；seed 填正式 .env 的
    PMS_QMT_SIGN_SEED_HEX，公钥带外交给 QMT 侧，拿回对方公钥填 PMS_QMT_PEER_PUBKEY_B64。
 3. **真实账户规模与纪律参数**：PMS_TOTAL_SCALE 按真实资金；上线初期 shadow +
@@ -55,8 +60,8 @@ ws_smoke 发的联调单绕开这个开关、且带 SMOKE 前缀——成交回
 
 照抄模拟的 .env，改动以下键（其余全部照抄；行情/信号这些共享源**不要改**）：
 
-    # 必须不同
+    # 必须不同 (注意: PROXY_DB_URL **不在这里** —— 照抄模拟的, 同一个 my_quant_db)
-    PROXY_DB_URL=mysql+pymysql://<user>:<pass>@192.168.16.153:3307/my_quant_db_real
+    PMS_TABLE_PREFIX=real_
     PMS_REDIS_URL=redis://:<pass>@192.168.16.150:6379/9
     PMS_SIGNAL_GROUP=pms_signal_consumer_real
     PMS_SIGNAL_CONSUMER=pms_real_1
@@ -65,6 +70,12 @@ ws_smoke 发的联调单绕开这个开关、且带 SMOKE 前缀——成交回
     PMS_QMT_PEER_PUBKEY_B64=<QMT 正式侧公钥>
     PMS_SESSION_SECRET=<新生成, 别抄模拟的>
 
+    # 下游表名: 按正式 QMT 实际写的表名填 (同库, 直接写表名)
+    PMS_DS_TABLE_POSITION=<正式持仓表名>
+    PMS_DS_TABLE_ORDER=<正式委托成交表名>
+    PMS_DS_TABLE_BUY_PLAN=<正式买入计划表名>
+    PMS_DS_TABLE_DAILY=<正式定性结论表名>
+
     # 阶段A 先关, 密钥交换完再开
     PMS_QMT_WS_ENABLED=false
 
@@ -77,20 +88,26 @@ ws_smoke 发的联调单绕开这个开关、且带 SMOKE 前缀——成交回
 ## 四、阶段A：最小台架 + ws 协议实测（在 188 上）
 
 1. 装 docker 与 docker compose；`git clone` 仓库；按第三节写好 .env（密钥先留空）。
-2. 网络预检：188 要能通 153:3307、150:6379、208:6379、199:3306、214:36379、98:9443。
+2. 表名口径：.env 设 PMS_TABLE_PREFIX=real_；向 QMT 侧要到四张正式下游表的表名，
-3. 最小台架（只起 页面+ws，不起调度）：`PROFILES="--profile ws" ./scripts/deploy.sh`
+   填进 PMS_DS_TABLE_*。不用建库、不用授权、不动代理。
-   —— 建镜像、起 pms-web/pms-ws、幂等建表。
+3. 网络预检：188 要能通 153:3307、150:6379、208:6379、199:3306、214:36379、98:9443。
-4. `make check`（库/参数/通道逐项）与 `make test`（ALL SUITES PASS + 指纹一致）。
+4. 建表前自证（最后一道闸）：`docker compose build` 后跑
-5. 生成密钥、与 QMT 侧交换公钥；.env 里填好两把钥匙并把 PMS_QMT_WS_ENABLED=true；
+   `docker compose run --rm --no-deps pms-web python -c "from config.settings import settings; print(repr(settings.PMS_TABLE_PREFIX))"`
+   ——必须打印 'real_'；打印空串绝不往下走（会建出与模拟同名的表）。
+5. 最小台架（只起 页面+ws，不起调度）：`PROFILES="--profile ws" ./scripts/deploy.sh --no-pull`
+   —— 建镜像、起 pms-web/pms-ws、幂等建表。输出必须出现
+   「[双实例] 表名前缀 'real_' 生效」，COUNT 验证列出的全是 real_pms_*；没看到立即中止。
+6. `make check`（库/参数/通道逐项）与 `make test`（ALL SUITES PASS + 指纹一致）。
+7. 生成密钥、与 QMT 侧交换公钥；.env 里填好两把钥匙并把 PMS_QMT_WS_ENABLED=true；
    重跑 `PROFILES="--profile ws" ./scripts/deploy.sh --no-pull` 让新 env 生效。
-6. 通道判收（不动钱）：`make ws-status` 看 连接 ONLINE、心跳在走、能读到持仓快照；
+8. 通道判收（不动钱）：`make ws-status` 看 连接 ONLINE、心跳在走、能读到持仓快照；
    `ws_smoke.py inbox` 里有 snapshot/心跳类上行。
-7. S2 式发单（挂不上才是预期）：盘中发一张限价远离市价的买单（--ttl 5 到点对端自动撤），
+9. S2 式发单（挂不上才是预期）：盘中发一张限价远离市价的买单（--ttl 5 到点对端自动撤），
    watch 里状态走 QUEUED→SENT→受理，inbox 里有 ack 与回报。
-8. **一笔真实买卖**（判收核心）：建议选一只当日可回转（T+0）的跨境或货币 ETF，一手
+10. **一笔真实买卖**（判收核心）：建议选一只当日可回转（T+0）的跨境或货币 ETF，一手
    一两百块，当天买进当天卖出；用普通股票也行，但 A 股 T+1，卖出要等次日。判收三条：
    两张单全成、inbox 有 trade 回报、**账本零变化**（SMOKE 联调单闸把成交挡在账外）。
-9. 清场：测试买的货**必须卖光**再进阶段B——否则影子期对账会把它当真实持仓收编进账本。
+11. 清场：测试买的货**必须卖光**再进阶段B——否则影子期对账会把它当真实持仓收编进账本。
 
 ## 五、阶段B：全量部署与影子期（协议判收后再做）
 

app/db/session.py

@@ -83,6 +83,42 @@ def assert_single_table(sql: str) -> str:
     return sql
 
 
+# ---------------------------------------------------------------- 双实例表名前缀
+# (2026-08-28 双系统定稿) 模拟与正式两套 PMS **共用 my_quant_db**, 靠表名区分:
+# 正式实例在 .env 里设 PMS_TABLE_PREFIX (如 "real_"), 这里在 SQL 执行的**唯一入口**
+# (fetch_all / fetch_one / execute / execute_many) 把每个 pms_* 表名统一改写成
+# {前缀}pms_* —— 全仓库 129 处表名引用一处改写, 不逐条改 SQL。默认空串 = 不改写,
+# 模拟环境行为逐字节不变。
+#
+# 改写按词边界匹配**小写 pms_ 开头的完整标识符**:
+#   · 19 张自有表全部命中; 已带前缀的 real_pms_* 不会二次加前缀 (前面无词边界);
+#   · 绑定参数的值不在 SQL 文本里, 不受影响; 大写参数键名 ('PMS_TOTAL_SCALE') 不匹配;
+#   · 字符串字面量里的表名 (如 information_schema 按 TABLE_NAME 查列) 一并改写,
+#     恰好是想要的语义 —— migrate 类脚本查的就该是本实例的表;
+#   · 下游表 (trading_* 等) 不带 pms_ 前缀, 一个字不动 (它们另走 PMS_DS_TABLE_* 配置)。
+# 两个绕过本入口直接拿引擎执行 DDL 的脚本 (init_db / migrate_archived_at) 已显式调
+# map_tables; test_batch19 的旁路扫描钉住"不许再有第三个"。
+_PREFIX_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_]*$")
+_PMS_TBL_RE = re.compile(r"\b(pms_[a-z0-9_]+)\b")
+
+
+def _load_prefix() -> str:
+    p = (settings.PMS_TABLE_PREFIX or "").strip()
+    if p and not _PREFIX_RE.match(p):
+        raise ValueError(f"PMS_TABLE_PREFIX 不合法 (只许字母数字下划线): {p!r}")
+    return p
+
+
+TABLE_PREFIX = _load_prefix()
+
+
+def map_tables(sql: str) -> str:
+    """按实例表名前缀改写 SQL 里的 pms_* 表名; 前缀为空原样返回。"""
+    if not TABLE_PREFIX:
+        return str(sql)
+    return _PMS_TBL_RE.sub(TABLE_PREFIX + r"\1", str(sql))
+
+
 # ---------------------------------------------------------------- 引擎
 def get_engine(name: str = "proxy"):
     if name in _engines:
@@ -132,6 +168,7 @@ def ping(name: str = "proxy") -> dict:
 
 # ---------------------------------------------------------------- 执行助手
 def fetch_all(sql: str, params=None, source: str = "proxy") -> list:
+    sql = map_tables(sql)
     assert_single_table(sql)
     _check_cooldown(source)
     try:
@@ -164,6 +201,7 @@ def fetch_one(sql: str, params=None, source: str = "proxy"):
 
 def execute(sql: str, params=None, source: str = "proxy") -> int:
     """写操作 (INSERT/UPDATE/DELETE)。返回受影响行数。"""
+    sql = map_tables(sql)
     assert_single_table(sql)
     _check_cooldown(source)
     try:
@@ -182,6 +220,7 @@ def execute(sql: str, params=None, source: str = "proxy") -> int:
 
 
 def execute_many(sql: str, seq_params: list, source: str = "proxy") -> int:
+    sql = map_tables(sql)
     assert_single_table(sql)
     if not seq_params:
         return 0

app/repo/downstream_repo.py

@@ -4,6 +4,9 @@
 ==============================================================================
 这三张表**归下游系统维护**, PMS 只读 (设计 §9)。
 
+表名可配 (2026-08-28 双系统): 正式 QMT 在同一个库里写另一套表名的下游表, 表名经 .env
+(settings.PMS_DS_TABLE_*) 注入, 模拟环境全用默认值。本文件所有 SQL 统一引用 T_* 常量。
+
 列名口径 (2026-07-27 实机 SHOW COLUMNS 确认, 见 check_db.py 输出):
   trading_position: id, stock_code, stock_name, total_quantity, available_quantity,
                     frozen_quantity, cost_price, market_price, market_value, profit_loss, ...
@@ -14,7 +17,10 @@
 """
 from __future__ import annotations
 
+import re
+
 from app.db.session import fetch_all, fetch_one
+from config.settings import settings
 
 # 数量/可用量列名候选 (探测顺序即优先级; 首项为实机确认的真实列名)
 QTY_CANDIDATES = ["total_quantity", "current_qty", "total_qty", "position_qty", "hold_qty",
@@ -27,6 +33,24 @@ PRICE_CANDIDATES = ["market_price", "last_price", "current_price", "price"]
 
 FILLED_STATUSES = ("completed", "filled")
 
+# ---------------------------------------------------------------- 下游表名 (双系统可配)
+# 表名要拼进 SQL, 先做白名单校验: 字母数字下划线, 至多带一个库名前缀。不合法直接抛 ——
+# 宁可启动就大声失败, 也不能静默退回默认表名把另一套系统的账读进来。
+_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*(\.[A-Za-z_][A-Za-z0-9_]*)?$")
+
+
+def _t(name):
+    n = str(name or "").strip()
+    if not _NAME_RE.match(n):
+        raise ValueError(f"下游表名不合法 (查 .env 的 PMS_DS_TABLE_*): {n!r}")
+    return n
+
+
+T_POSITION = _t(settings.PMS_DS_TABLE_POSITION)
+T_ORDER = _t(settings.PMS_DS_TABLE_ORDER)
+T_BUY_PLAN = _t(settings.PMS_DS_TABLE_BUY_PLAN)
+T_DAILY = _t(settings.PMS_DS_TABLE_DAILY)
+
 
 def to_dot(code: str) -> str:
     """SH600000 / 600000 → 600000.SH (PMS 内部统一点式)。"""
@@ -60,17 +84,20 @@ def _pick(keys, candidates):
 
 
 def describe(table: str) -> list:
-    """SHOW COLUMNS —— 供页面导出、回填 QMT_INTERFACE_REQUIREMENTS D1。"""
+    """SHOW COLUMNS —— 供页面导出、回填 QMT_INTERFACE_REQUIREMENTS D1。
-    if table not in ("trading_position", "trading_order", "trading_buy_plan",
+    入参是**逻辑名** (页面/文档口径); 双系统下实际表名可能不同, 在这里换算。"""
-                     "strategy_daily_results", "gp_stock_category"):
+    actual = {"trading_position": T_POSITION, "trading_order": T_ORDER,
+              "trading_buy_plan": T_BUY_PLAN, "strategy_daily_results": T_DAILY,
+              "gp_stock_category": "gp_stock_category"}.get(table)
+    if not actual:
         raise ValueError(f"不允许探测的表: {table}")
-    return fetch_all(f"SHOW COLUMNS FROM {table}")
+    return fetch_all(f"SHOW COLUMNS FROM {actual}")
 
 
 # ================================================================ trading_position
 def fetch_positions() -> dict:
     """下游持仓快照。返回 {"rows":[{ts_code, qty, avail_qty, cost, frozen}], "columns":{...}}"""
-    rows = fetch_all("SELECT * FROM trading_position LIMIT 1000")
+    rows = fetch_all(f"SELECT * FROM {T_POSITION} LIMIT 1000")
     if not rows:
         return {"rows": [], "columns": {"qty": None, "avail": None, "cost": None},
                 "raw_count": 0}
@@ -129,7 +156,7 @@ def fetch_filled_orders(*, since_id=None, since_time=None, limit: int = 500) ->
     elif since_time:
         where.append("(filled_time >= :st OR order_time >= :st)")
         p["st"] = since_time
-    sql = ("SELECT * FROM trading_order WHERE " + " AND ".join(where) +
+    sql = (f"SELECT * FROM {T_ORDER} WHERE " + " AND ".join(where) +
            " ORDER BY order_id ASC LIMIT :n")
     rows = fetch_all(sql, p)
     return [_norm_order(r) for r in rows]
@@ -144,14 +171,14 @@ def latest_filled_order_id():
     游标已改为「时间窗 + 已见 order_id 集合」(见 ledger_service.replay_fills)。
     本函数保留给旧数据排查用。
     """
-    r = fetch_one("SELECT MAX(order_id) AS mx FROM trading_order WHERE order_status IN ('"
+    r = fetch_one(f"SELECT MAX(order_id) AS mx FROM {T_ORDER} WHERE order_status IN ('"
                   + "', '".join(FILLED_STATUSES) + "')")
     return (r or {}).get("mx")
 
 
 def latest_filled_time():
     """已成交单里最新的成交/委托时间 —— 时间游标冷启动的锚点 (不追认历史)。"""
-    r = fetch_one("SELECT MAX(filled_time) AS ft, MAX(order_time) AS ot FROM trading_order "
+    r = fetch_one(f"SELECT MAX(filled_time) AS ft, MAX(order_time) AS ot FROM {T_ORDER} "
                   "WHERE order_status IN ('" + "', '".join(FILLED_STATUSES) + "')")
     r = r or {}
     vals = [str(v) for v in (r.get("ft"), r.get("ot")) if v]
@@ -180,7 +207,7 @@ def fetch_buy_plans(*, is_active=None, limit: int = 200) -> list:
     """
     sql = ("SELECT id, strategy_id, stock_code, stock_name, target_price, buy_amount, "
            "factor_code, trading_time, create_time, update_time, tp_ratio, sl_ratio, "
-           "prob_thresh, hold_days, is_active, approved_by FROM trading_buy_plan")
+           f"prob_thresh, hold_days, is_active, approved_by FROM {T_BUY_PLAN}")
     p = {"n": int(limit)}
     if is_active is not None:
         sql += " WHERE is_active = :ia"
@@ -204,7 +231,7 @@ def fetch_refs(ts_code: str):
     """决策系统昨夜结论: 支撑/压力参考位 (主口径; 停更超期由 services 兜底自算)。"""
     r = fetch_one(
         "SELECT stock_code, signal_type, support_level, pressure_level, trade_date "
-        "FROM strategy_daily_results WHERE stock_code = :code "
+        f"FROM {T_DAILY} WHERE stock_code = :code "
         "ORDER BY trade_date DESC LIMIT 1", {"code": to_prefix(ts_code)})
     if not r:
         return None

app/services/strategy_advisor.py

@@ -41,7 +41,7 @@ from datetime import datetime, timedelta
 from app.core import tradedays as td
 from app.db.session import fetch_all, fetch_one
 from app.repo import pms_repo
-from app.repo.downstream_repo import to_dot, to_prefix
+from app.repo.downstream_repo import T_DAILY, to_dot, to_prefix
 from app.services import param_store
 
 logger = logging.getLogger("pms.strategy_advisor")
@@ -285,7 +285,7 @@ def accum_of(dot_codes) -> dict:
     since = int((datetime.now().date() - timedelta(days=ACCUM_WINDOW_DAYS)).strftime("%Y%m%d"))
     p = {"since": since}
     rows = fetch_all(
-        "SELECT stock_code, trade_date, raw_logic_json FROM strategy_daily_results "
+        f"SELECT stock_code, trade_date, raw_logic_json FROM {T_DAILY} "
         f"WHERE trade_date >= :since AND stock_code IN ({_in(list(rev), 'c', p)})", p)
     best = {}
     for r in rows:

config/settings.py

@@ -31,6 +31,21 @@ class Settings(BaseSettings):
     DB_MYSQL_URL: str = "mysql+pymysql://user:pass@192.168.18.199:3306/db_gp_cj"
     #   大盘指数 zs_day_data (页面区制提示用, 非约束)
 
+    # 下游只读表的表名 (2026-08-28 双系统): 正式 QMT 在同一个库里写**另一套表名**的下游表,
+    # 表名经 .env 注入; 模拟环境保持默认值一个字不用动。允许带库名前缀的写法
+    # ("my_quant_db.trading_order_real"), 供 pms 库与下游表不在同一 schema 时跨库读。
+    # 表名要拼进 SQL, downstream_repo 加载时会做白名单校验, 不合法直接拒绝启动。
+    PMS_DS_TABLE_POSITION: str = "trading_position"
+    PMS_DS_TABLE_ORDER: str = "trading_order"
+    PMS_DS_TABLE_BUY_PLAN: str = "trading_buy_plan"
+    PMS_DS_TABLE_DAILY: str = "strategy_daily_results"
+
+    # PMS 自有表的表名前缀 (2026-08-28 双系统定稿): 模拟与正式两套系统**共用 my_quant_db**,
+    # 靠表名区分 —— 正式实例设 PMS_TABLE_PREFIX="real_", 它的 19 张表就是 real_pms_* 这一套。
+    # 映射在 db.session 的 SQL 执行唯一入口统一做 (全仓库 SQL 不逐条改), 默认空串 = 不改写,
+    # 模拟环境行为逐字节不变。只许字母数字下划线, 非法值拒绝启动。
+    PMS_TABLE_PREFIX: str = ""
+
     SIGNAL_REDIS_HOST: str = "192.168.18.208"
     SIGNAL_REDIS_PORT: int = 6379
     SIGNAL_REDIS_PASSWORD: str = ""

scripts/init_db.py

@@ -169,16 +169,18 @@ def main():
 
     # 连库依赖放到演练之后再导入: 演练模式只做语句切分校验, 无依赖环境也能跑
     from sqlalchemy import text
-    from app.db.session import get_engine
+    from app.db.session import TABLE_PREFIX, get_engine, map_tables
 
     eng = get_engine("proxy")
+    if TABLE_PREFIX:
+        print(f"[双实例] 表名前缀 {TABLE_PREFIX!r} 生效: 实际建的是 {TABLE_PREFIX}pms_* 这一套表")
     okc, failed = 0, []
     print()
     for kind, tbl, stmt in stmts:
         label = tbl if kind == "table" else f"{tbl} (初始行)"
         try:
             with eng.begin() as c:
-                c.execute(text(stmt))
+                c.execute(text(map_tables(stmt)))
             print(f"  OK    {label}")
             okc += 1
         except Exception as e:
@@ -190,8 +192,8 @@ def main():
     for tbl in dict.fromkeys(t for _, t, _ in stmts):      # 去重且保序
         try:
             with eng.connect() as c:
-                n = c.execute(text(f"SELECT COUNT(*) FROM {tbl}")).scalar()
+                n = c.execute(text(map_tables(f"SELECT COUNT(*) FROM {tbl}"))).scalar()
-            print(f"  OK    {tbl:<20} {n} 行")
+            print(f"  OK    {map_tables(tbl):<20} {n} 行")
         except Exception as e:
             print(f"  FAIL  {tbl:<20} {type(e).__name__}: {e}")
             missing.append(tbl)
@@ -200,7 +202,7 @@ def main():
     if failed:
         print(f"以下 {len(failed)} 张表建失败, 完整语句如下 —— 可直接拿到物理库执行:")
         for tbl, stmt, err in failed:
-            print(f"\n### {tbl}  ({err.splitlines()[0]})\n{stmt};")
+            print(f"\n### {map_tables(tbl)}  ({err.splitlines()[0]})\n{map_tables(stmt)};")
     if missing:
         print(f"\nFAILED: {len(missing)} 张表仍不可用: {', '.join(missing)}")
         sys.exit(1)

scripts/migrate_archived_at.py

@@ -39,14 +39,16 @@ def main():
     args = ap.parse_args()
 
     from sqlalchemy import text
-    from app.db.session import get_engine
+    from app.db.session import TABLE_PREFIX, get_engine, map_tables
     eng = get_engine("proxy")
+    if TABLE_PREFIX:
+        print(f"[双实例] 表名前缀 {TABLE_PREFIX!r} 生效: 实际迁移的是 {TABLE_PREFIX}pms_* 这一套表")
 
     todo, already, unknown = [], [], []
     with eng.connect() as c:
         for t in TABLES:
             try:
-                (already if _has_column(c, t, COL) else todo).append(t)
+                (already if _has_column(c, map_tables(t), COL) else todo).append(t)
             except Exception as e:
                 print(f"  ?     {t}: 查列失败 {type(e).__name__}: {e} (表不存在就先跑 init_db)")
                 unknown.append(t)
@@ -58,7 +60,7 @@ def main():
         print("\n无需迁移: 四张表都已有该列 (或表还没建, 先 init_db)。")
         sys.exit(1 if unknown else 0)
 
-    stmts = {t: (f"ALTER TABLE {t} ADD COLUMN {COL} DATETIME NULL DEFAULT NULL "
+    stmts = {t: map_tables(f"ALTER TABLE {t} ADD COLUMN {COL} DATETIME NULL DEFAULT NULL "
                            f"COMMENT '{COMMENT}'") for t in todo}
 
     if not args.yes:

scripts/probe_strategy_signals.py

@@ -26,7 +26,7 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 from app.db.session import fetch_all, fetch_one          # noqa: E402
 from app.repo import pms_repo                             # noqa: E402
-from app.repo.downstream_repo import to_dot, to_prefix    # noqa: E402
+from app.repo.downstream_repo import T_DAILY, to_dot, to_prefix    # noqa: E402
 from app.services import param_store                      # noqa: E402
 
 # 决策系统定性词表 (契约; 见方案「口径同源声明」)。不在表里的词一律当无标志。
@@ -103,13 +103,13 @@ def _in_clause(values, prefix, params):
 
 # ================================================================ 取数
 def latest_accum_day():
-    r = fetch_one("SELECT MAX(trade_date) AS td FROM strategy_daily_results")
+    r = fetch_one(f"SELECT MAX(trade_date) AS td FROM {T_DAILY}")
     return (r or {}).get("td")
 
 
 def accum_distribution(td):
     """最新一日的定性分布 (全池)。返回 (总数, 分布, 词表外原文样本)。"""
-    rows = fetch_all("SELECT stock_code, raw_logic_json FROM strategy_daily_results "
+    rows = fetch_all(f"SELECT stock_code, raw_logic_json FROM {T_DAILY} "
                      "WHERE trade_date = :td", {"td": td})
     dist, total, oov = {}, 0, {}
     for r in rows:
@@ -131,7 +131,7 @@ def accum_of(dot_codes):
     rev = _variants(dot_codes)
     since = int((datetime.now().date() - timedelta(days=ACCUM_WINDOW_DAYS)).strftime("%Y%m%d"))
     p = {"since": since}
-    sql = ("SELECT stock_code, trade_date, raw_logic_json FROM strategy_daily_results "
+    sql = (f"SELECT stock_code, trade_date, raw_logic_json FROM {T_DAILY} "
            f"WHERE trade_date >= :since AND stock_code IN ({_in_clause(list(rev), 'c', p)})")
     best = {}
     for r in fetch_all(sql, p):

scripts/run_tests.py

@@ -37,9 +37,9 @@
   test_batch19_units.py 2026-08-28 审查修复回归: 科创板最小申报统一口径/
                         取整与部分卖/同轮买卖互斥/信号百分制契约/除权核销
                         缩放/日历按年降级/网格中枢与止盈闩锁/买入暂停
-                        按来源分记/宏观失败路径保留留痕               (17 例)
+                        按来源分记/宏观失败路径保留留痕/双实例表名前缀   (19 例)
   test_wiring.py        装配自检: 服务层→核心→落表 全链路 (内存桩)      (66 例)
-                                                                  共 585 例
+                                                                  共 587 例
 任一子集失败即整体失败 (退出码 1)。
 """
 import os

scripts/test_batch19_units.py

@@ -12,7 +12,8 @@
   7. 交易日历按年探测降级 (chinesecalendar 装了但没有当年数据);
   8. 网格只买中枢下方 / 跟踪止盈部分卖一次性闩锁 (strategy_runner);
   9. 策略买入暂停按来源分记 (strategy_service, accum 与 signal 互不误伤);
- 10. 宏观失败路径保留当日动作留痕 (macro_service._upsert_unavailable)。
+ 10. 宏观失败路径保留当日动作留痕 (macro_service._upsert_unavailable);
+ 11. 双实例表名前缀 (SQL 执行唯一入口统一映射 pms_* 表名 + 旁路扫描)。
 运行: python scripts/test_batch19_units.py
 """
 import os
@@ -376,6 +377,75 @@ def _():
         macro_repo.get_signal, macro_repo.upsert_signal = orig_get, orig_up
 
 
+# ================================================================
+# [K] 双实例表名前缀: 一处映射, 不许有旁路
+# ================================================================
+@case("[K1] 表名映射: 唯一执行入口统一改写 pms_*; 前缀空原样; 已带前缀不二次加; 非法前缀拒绝")
+def _():
+    from app.db import session as se
+    orig = se.TABLE_PREFIX
+    try:
+        se.TABLE_PREFIX = "real_"
+        f = se.map_tables
+        assert f("SELECT * FROM pms_command WHERE status = :s") == \
+               "SELECT * FROM real_pms_command WHERE status = :s"
+        assert f("INSERT INTO pms_lot (c) VALUES (:c) ON DUPLICATE KEY UPDATE q = :q") == \
+               "INSERT INTO real_pms_lot (c) VALUES (:c) ON DUPLICATE KEY UPDATE q = :q"
+        assert f("UPDATE pms_plan_snapshot SET x=1") == "UPDATE real_pms_plan_snapshot SET x=1"
+        # DDL 与反引号写法 (init_db 建表走的就是这条路)
+        assert f("CREATE TABLE IF NOT EXISTS `pms_qmt_order` (id INT)") == \
+               "CREATE TABLE IF NOT EXISTS `real_pms_qmt_order` (id INT)"
+        # 长表名贪婪匹配整个标识符, 不会只改前半截
+        assert f("DELETE FROM pms_plan_snapshot") == "DELETE FROM real_pms_plan_snapshot"
+        # 下游表不带 pms_ 前缀, 一个字不动; 大写参数键名不受影响
+        assert f("SELECT * FROM trading_order") == "SELECT * FROM trading_order"
+        s = "SELECT v FROM pms_runtime_param WHERE param_key = 'PMS_TOTAL_SCALE'"
+        assert f(s) == s.replace("pms_runtime_param", "real_pms_runtime_param"), f(s)
+        # 幂等: 已带前缀的不会被二次加前缀 (real_pms_ 里的 pms_ 前面没有词边界)
+        assert f("SELECT * FROM real_pms_command") == "SELECT * FROM real_pms_command"
+        assert f(f("SELECT * FROM pms_command")) == "SELECT * FROM real_pms_command"
+        # 映射后的 SQL 仍过单表守卫
+        se.assert_single_table(f("SELECT * FROM pms_command WHERE id = :i"))
+        se.TABLE_PREFIX = ""
+        assert f("SELECT * FROM pms_command") == "SELECT * FROM pms_command"
+    finally:
+        se.TABLE_PREFIX = orig
+    # 前缀白名单: 注入形态直接拒绝 (点号/分号/引号都不许)
+    assert se._PREFIX_RE.match("real_") and se._PREFIX_RE.match("r2_")
+    for bad in ("a.b", "x;drop", "p'q", "1abc", "a b"):
+        assert not se._PREFIX_RE.match(bad), bad
+
+
+@case("[K2] 映射不许有旁路: 自建引擎的文件不得写 pms_ 表; 直执 DDL 的脚本必须显式过映射")
+def _():
+    import re as _re
+    root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+    offenders = []
+    for base in ("app", "scripts"):
+        for dp, _dn, fns in os.walk(os.path.join(root, base)):
+            for fn in fns:
+                if not fn.endswith(".py"):
+                    continue
+                p = os.path.join(dp, fn)
+                rel = p[len(root) + 1:].replace(os.sep, "/")
+                if rel == "app/db/session.py" or rel.startswith("scripts/test_"):
+                    continue
+                src = open(p, encoding="utf-8").read()
+                if "create_engine" not in src:
+                    continue
+                # 自建引擎绕开了执行入口的表名映射 —— 这样的文件绝不能碰 pms_ 表
+                if _re.search(r"\bpms_[a-z0-9_]+", src):
+                    offenders.append(rel)
+    assert not offenders, ("这些文件绕开 db.session 自建引擎、又直接写 pms_ 表 SQL, "
+                           "双实例表名前缀对它们不生效: %s" % offenders)
+    # init_db / migrate_archived_at 拿 get_engine 直执行 DDL (有意绕过单表守卫),
+    # 它们必须显式调 map_tables —— 否则正式实例 make deploy 会建出**无前缀**的表,
+    # 与模拟实例的表撞在一起
+    for f in ("scripts/init_db.py", "scripts/migrate_archived_at.py"):
+        src = open(os.path.join(root, f), encoding="utf-8").read()
+        assert "map_tables" in src, f + " 直执 DDL 必须过 map_tables"
+
+
 def main():
     passed, failed = 0, []
     for name, fn in RESULTS: