diff --git a/DEVLOG.md b/DEVLOG.md index e6eb9af..acce09f 100644 --- a/DEVLOG.md +++ b/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 下游表 (决定 - 153 代理加逻辑库还是直连 MySQL)、正式 ws 端点与密钥交换、真实账户规模与纪律参数。 +1. 部署前还差两件对接: 向 QMT 侧要正式下游四张表的**实际表名** (填 .env 的 + PMS_DS_TABLE_*); 交换正式 ws 密钥对。库不用建、代理不用动 (同库表名前缀方案)。 2. 正式盘上线初期纪律: shadow + propose_only + 新建仓 off, 影子期至少 3 个交易日。 --- diff --git a/REAL_TRADING_DEPLOY_PLAN.md b/REAL_TRADING_DEPLOY_PLAN.md index feef896..25bb159 100644 --- a/REAL_TRADING_DEPLOY_PLAN.md +++ b/REAL_TRADING_DEPLOY_PLAN.md @@ -1,8 +1,10 @@ # 正式盘部署方案（模拟 / 正式双系统） -2026-08-28 定稿，同日按拍板结果更新：正式盘部署在 **192.168.16.188**，正式 QMT 在 -**192.168.16.98**（ws 端点 ws://192.168.16.98:9443/pms）；正式侧**会写** 153 的下游表， -所以 153 代理上加新逻辑库（见第二节）。现有这套 PMS 保留为**模拟仓**原地不动。 +2026-08-28 定稿，同日随拍板收敛到最终口径：正式盘部署在 **192.168.16.188**，正式 QMT +在 **192.168.16.98**（ws 端点 ws://192.168.16.98:9443/pms）；**两套系统共用 my_quant_db +一个库，全部靠表名区分**——正式 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_*，只能分库 | -| 下游表 trading_* / strategy_daily_results | 153 代理 | 同一代理经新逻辑库映射（共享读） | 正式侧确认会写 | +| PMS 自有表（19 张） | my_quant_db 的 pms_* | **同库、加前缀 real_pms_***（PMS_TABLE_PREFIX） | 表名映射在 SQL 唯一执行入口统一做 | +| 下游表 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_* 全部表映射到一个**新建的物理库**； - trading_position / trading_order / trading_buy_plan / strategy_daily_results 的映射 - 与 my_quant_db **完全相同**（共享读）。一个要核对的点：正式 QMT 写的下游表若与模拟 - 写的是**同一批物理表**，正式 PMS 的回放与对账会连模拟的旧行一起看见——两边下游数据 - 分不分表，要与 QMT/决策系统侧对一遍口径。 +1. **库与表的接线（同库表名区分，什么都不用建）**。PROXY_DB_URL 与模拟**完全相同** + （照抄，还是 153 代理的 my_quant_db）。正式 PMS 的自有表靠前缀区分：.env 设 + PMS_TABLE_PREFIX=real_，db.session 在 SQL 执行唯一入口把 pms_* 统一映射成 + real_pms_*；首次 `make deploy` 建表自动带前缀建出 real_pms_* 这一套（init_db 开头 + 会打印「表名前缀 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=mysql+pymysql://:@192.168.16.153:3307/my_quant_db_real + # 必须不同 (注意: PROXY_DB_URL **不在这里** —— 照抄模拟的, 同一个 my_quant_db) + PMS_TABLE_PREFIX=real_ PMS_REDIS_URL=redis://:@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= 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。 -3. 最小台架（只起 页面+ws，不起调度）：`PROFILES="--profile ws" ./scripts/deploy.sh` - —— 建镜像、起 pms-web/pms-ws、幂等建表。 -4. `make check`（库/参数/通道逐项）与 `make test`（ALL SUITES PASS + 指纹一致）。 -5. 生成密钥、与 QMT 侧交换公钥；.env 里填好两把钥匙并把 PMS_QMT_WS_ENABLED=true； +2. 表名口径：.env 设 PMS_TABLE_PREFIX=real_；向 QMT 侧要到四张正式下游表的表名， + 填进 PMS_DS_TABLE_*。不用建库、不用授权、不动代理。 +3. 网络预检：188 要能通 153:3307、150:6379、208:6379、199:3306、214:36379、98:9443。 +4. 建表前自证（最后一道闸）：`docker compose build` 后跑 + `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：全量部署与影子期（协议判收后再做） diff --git a/app/db/session.py b/app/db/session.py index 05e4490..77f76db 100644 --- a/app/db/session.py +++ 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 diff --git a/app/repo/downstream_repo.py b/app/repo/downstream_repo.py index 7d61804..d6f986f 100644 --- a/app/repo/downstream_repo.py +++ b/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。""" - if table not in ("trading_position", "trading_order", "trading_buy_plan", - "strategy_daily_results", "gp_stock_category"): + """SHOW COLUMNS —— 供页面导出、回填 QMT_INTERFACE_REQUIREMENTS D1。 + 入参是**逻辑名** (页面/文档口径); 双系统下实际表名可能不同, 在这里换算。""" + 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 diff --git a/app/services/strategy_advisor.py b/app/services/strategy_advisor.py index 3c9a1e6..0217d19 100644 --- a/app/services/strategy_advisor.py +++ b/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: diff --git a/config/settings.py b/config/settings.py index be286a3..ffefa34 100644 --- a/config/settings.py +++ b/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 = "" diff --git a/scripts/init_db.py b/scripts/init_db.py index 27db4d0..1aad4f9 100644 --- a/scripts/init_db.py +++ b/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() - print(f" OK {tbl:<20} {n} 行") + n = c.execute(text(map_tables(f"SELECT COUNT(*) FROM {tbl}"))).scalar() + 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) diff --git a/scripts/migrate_archived_at.py b/scripts/migrate_archived_at.py index 67df6b3..ca48932 100644 --- a/scripts/migrate_archived_at.py +++ b/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,8 +60,8 @@ 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 " - f"COMMENT '{COMMENT}'") for t in todo} + 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: print("\n[演练模式] 未执行任何语句。确认无误后加 --yes 重跑。将执行:") diff --git a/scripts/probe_strategy_signals.py b/scripts/probe_strategy_signals.py index 2dfb620..0863fad 100644 --- a/scripts/probe_strategy_signals.py +++ b/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): diff --git a/scripts/run_tests.py b/scripts/run_tests.py index 1784a0f..da340d1 100644 --- a/scripts/run_tests.py +++ b/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 diff --git a/scripts/test_batch19_units.py b/scripts/test_batch19_units.py index 80ac151..899ec3f 100644 --- a/scripts/test_batch19_units.py +++ b/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: