行业源换 gp_hybk(199,三级884,单主行业); gp_stock_category 作废

README.md

@@ -61,7 +61,7 @@ scripts/
   test_batch5_units.py 决策系统信号流解析与消化口径                  8 例
   test_batch6_units.py ws 通道: 测试向量/签名/公钥/水位/DDL/逐笔入账 65 例
   test_batch7_units.py 上游选股计划: 解析/新鲜度/候选筛选/取数守卫   32 例
-  test_wiring.py       装配自检: 服务层→核心→落表 全链路 (内存桩)   55 例
+  test_wiring.py       装配自检: 服务层→核心→落表 全链路 (内存桩)   58 例
   init_db.py           建表 (应用 ddl_pms_v1.sql, 幂等, 默认演练; 含 DDL 体检)
   check_db.py          实机连通性与表结构自检 (需真实 .env)
   gen_keys.py          ws 通道密钥: 生成 / 只取公钥(--pubkey) / PEM(--pem) / 自检(--check)
@@ -199,7 +199,7 @@ QMT ──trade/order_update──▶ pms-ws ──落 pms_qmt_inbox──▶
 
 ## 已实现 / 待开发
 
-**已实现**：建表 DDL 与建表脚本；配置与运行参数中心；仓位规划器与安全垫账；命令系统（27 类命令全目录 + 双状态机 + 冲突识别）；方案生成器（降仓凑额四档、升仓、建仓分批、清仓/减至、行业清仓与限额、暂停买入撤单）；账本回放与对账引擎（成交认领、外部成交并入 BASE 告警、以下游为准修正、除权检测、T+1 可用量、连续不一致升级）；规则闸终检；择时执行器实现 B（分日配额、分笔、VWAP/回踩/不追高、14:45 兜底、停牌一字板顺延、窗口耗尽收口、挂单有效期）；动作引擎四类自主动作 + 研判闸客户端 + 提议分流；决策系统信号消化（两条流独立消费组订阅、置信度分档转清仓指令或提议）；管理页面四块 + 运维/日报抽屉；调度器九个调度位；**上游选股计划接口接入**（`/plan` 取候选池、交易日龄硬校验、`theme` 灌行业映射表、页面预览抽屉与不可用横幅）；**ws 直连通道的连接层**（常驻进程 + 出口队列 + 签名 + seq 水位与累积确认，见下）；**单测 241 例**。
+**已实现**：建表 DDL 与建表脚本；配置与运行参数中心；仓位规划器与安全垫账；命令系统（27 类命令全目录 + 双状态机 + 冲突识别）；方案生成器（降仓凑额四档、升仓、建仓分批、清仓/减至、行业清仓与限额、暂停买入撤单）；账本回放与对账引擎（成交认领、外部成交并入 BASE 告警、以下游为准修正、除权检测、T+1 可用量、连续不一致升级）；规则闸终检；择时执行器实现 B（分日配额、分笔、VWAP/回踩/不追高、14:45 兜底、停牌一字板顺延、窗口耗尽收口、挂单有效期）；动作引擎四类自主动作 + 研判闸客户端 + 提议分流；决策系统信号消化（两条流独立消费组订阅、置信度分档转清仓指令或提议）；管理页面四块 + 运维/日报抽屉；调度器九个调度位；**上游选股计划接口接入**（`/plan` 取候选池、交易日龄硬校验、`theme` 灌行业映射表、页面预览抽屉与不可用横幅）；**ws 直连通道的连接层**（常驻进程 + 出口队列 + 签名 + seq 水位与累积确认，见下）；**单测 244 例**。
 
 ### 下一步（按可动工顺序）
 
@@ -207,7 +207,7 @@ QMT ──trade/order_update──▶ pms-ws ──落 pms_qmt_inbox──▶
 |---|---|---|
 | 1 | ~~ws 通道的账本侧改造（清单 4~6）~~ | ✅ 2026-07-29 |
 | 2 | ~~上游选股计划接入（`/plan`）~~ | ✅ 2026-07-31：候选池独占来源、交易日龄硬校验、盘前昨收兜底、ST 剔除、PMS 侧主题限额、页面抽屉与探活脚本。口径与实测记录见 `UPSTREAM_PLAN_API.md` |
-| 3 | **行业硬拦截开闸** | 参数已切 `gp_stock_category`；`ready()` 改成小样本实探（参数填了不算数，见 `UPSTREAM_PLAN_API.md` §7.4）。**待验证**：部署后再查 `GET /api/industry`，看 `status.probe.hit` 是否 > 0、命中的列名是什么 —— hit=0 就说明这张表用不了，得回头灌 `custom_table` |
+| 3 | **行业硬拦截开闸** | 数据源换成 `gp_hybk`（199 库行业板块表，三级 884*，每票只认 bk_code 最小的主行业；`gp_stock_category` 作废——无 `stock_code` 列）。代码写法与 database 均做了自适应/显式报错。**待验证**：`PMS_SECTOR_SOURCE=gp_hybk` 后查 `GET /api/industry` 看 `probe.form` 与 `ready`，再看 `/api/positions` 的 `sector` 有没有填上。详见 `UPSTREAM_PLAN_API.md` §8 |
 | 4 | **账本重建**：清账后账本为空，需对端持仓就绪后 `POST /api/ops/reconcile?apply_fix=true` 以下游为准补回 | 阻塞：等 QMT 侧按真实成本价重建模拟持仓（`QMT_SIDE_S3_CLOSEOUT.md` §5） |
 | 5 | **ws 通道联调收尾**（协议 §9 S3） | 阻塞：`trade_no` 格式不合 §5.5（`QMT_SIDE_S3_CLOSEOUT.md` §1，这条挡住切 ws） |
 | 6 | 上游计划的剩余待确认口径 | 等上游：`UPSTREAM_PLAN_API.md` §4 剩 5 条 + §7.3 新增两条（同一 `date` 计划不幂等、档位规模与文档不符） |

UPSTREAM_PLAN_API.md

@@ -438,3 +438,73 @@ gp_stock_category 一个字的证据都没有。再加上 repo 层 `fetch_sector
 让 planner 把 sector 置空、页面顶部挂黄色横幅。**该停的时候明着停**, 比绿着失效强。
 
 三条单测锁住: 查得到才 ready / 查不到 ready=False 且错误原文冒到 hint / 空映射表不算 ready。
+
+#### 实探结果 (2026-07-31)
+
+```
+ready: false
+hint:  gp_stock_category **一只都没查到** ...
+errors: Unknown column 'stock_code' in 'where clause'
+```
+
+**这张表没有 `stock_code` 列。** 老代码撞的两个候选列名 (`stock_code` / `ts_code`) 里,
+第一个不存在, 第二个查得动但三只样本一只都不在表里 —— 而两种情况在老实现里都变成 `None`,
+分不出来。行业硬拦截当时就是**开着但全程失效**的状态, 只不过现在它明着说了。
+
+据此又改了一版: **先 `SHOW COLUMNS` 认列, 再查**。
+
+- `downstream_repo.category_columns()` —— 认代码列 (8 个候选) 与行业列 (10 个候选, 加了
+  `sw_l1` / `board_name` / `plate_name` 这类), 缓存 10 分钟。认不出来时报错直接带上**实际列名**,
+  是一条可执行的信息而不是一句"查不到"。
+- `probe_category()` —— 拿认出来的那一列, 三种代码写法 (点式 / 前缀式 / 纯数字) 逐个试。
+- `gp_stock_category` 加进 `describe()` 白名单与「运维 → 导出下游表结构」, 多出一个
+  `_category_probe` 段。**下个窗口的第一条命令就是它** —— 看清列名才能决定是改列名映射
+  还是干脆放弃这张表、去灌 `pms_industry_map`。
+
+```bash
+curl -s -X GET http://127.0.0.1:38100/api/ops/downstream-schema | python3 -m json.tool
+```
+
+
+---
+
+## 8. 行业源换成 gp_hybk (2026-07-31 定案)
+
+`gp_stock_category` 这条路作废 (§7.4: 没有 `stock_code` 列, `ts_code` 也查不到样本票)。
+参考另一项目的持仓行业分析实现, 换成 **`gp_hybk`**:
+
+| | |
+|---|---|
+| 库 | `DB_MYSQL_URL` (192.168.18.199) —— PMS 原本只用它读 `zs_day_data`, `fetch_all(..., source="index")` 直接可用, **不新增连接** |
+| 表 / 列 | `gp_hybk`: `gp_code` / `bk_code` (数值型) / `bk_name` |
+| 分级 | `bk_code` 前缀 `881`=二级, `884`=三级 |
+| PMS 取哪级 | **三级 884** (`PMS_SECTOR_HYBK_LEVEL=l3`) —— 二级太粗, 4 只堆一个二级行业拦不住 |
+| 一票多行业 | **只认一个主行业**, `ctx["sector"]` 保持单值, sizer/planner 一行不改 |
+
+主行业的选法定为 **该级板块里 `bk_code` 升序第一个**。要点是**稳定**: 同一只票每次都得到
+同一个行业, 否则今天算「工程机械」明天算「专用设备」, `sector_names_map` / `sector_mv_map`
+的累计会自己跳。按 bk_code 数值排, 不依赖查询返回次序, 也不依赖表里有没有主次标记。
+
+> 与参考实现的差异是故意的: 那边把金额**均分**到多个行业, 因为它要的是统计准确;
+> 这边是**拦不拦**, 只认一个主行业——代价是一票同属多个热门三级行业时集中度会算漏,
+> 这是明确接受的取舍 (用户 2026-07-31 定)。
+
+### 两个没确认、改成自适应的点
+
+1. **`gp_hybk` 在 199 的哪个 database?** 按 `DB_MYSQL_URL` (默认 `db_gp_cj`) 实现。
+   若不在那儿, `industry.status()` 的 hint 会直接写「gp_hybk 在 DB_MYSQL_URL 指向的库,
+   若不在那儿要改 .env」并带上原始报错 —— 不会静默失效。
+2. **`gp_code` 的代码写法?** 点式 `600000.SH` / 前缀式 `SH600000` / 纯数字 `600000`
+   三种各试一遍 (样本: 浦发、平安、茅台), 谁命中用谁并缓存 10 分钟。三种都不命中时明确区分
+   「表能查但没有这些票」和「表根本查不了」。
+
+### 验证
+
+```bash
+curl -s http://127.0.0.1:38100/api/params -X POST -H 'Content-Type: application/json' \
+  -d '{"key":"PMS_SECTOR_SOURCE","value":"gp_hybk"}'
+curl -s http://127.0.0.1:38100/api/industry | python3 -m json.tool
+```
+
+看 `status.probe.form` (认出来的代码写法) 与 `status.ready`。ready=true 之后再看
+`/api/positions` 里持仓票的 `sector` 有没有真填上。

app/repo/downstream_repo.py

@@ -62,7 +62,7 @@ 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"):
+                     "strategy_daily_results", "gp_stock_category"):
         raise ValueError(f"不允许探测的表: {table}")
     return fetch_all(f"SHOW COLUMNS FROM {table}")
 
@@ -203,7 +203,46 @@ def fetch_refs(ts_code: str):
             "trade_date": r.get("trade_date")}
 
 
-INDUSTRY_COL_CANDIDATES = ("industry", "category", "sector", "industry_name", "sw_industry")
+INDUSTRY_COL_CANDIDATES = ("industry", "category", "sector", "industry_name", "sw_industry",
+                           "sw_l1", "sw_l2", "industry_l1", "board_name", "plate_name")
+CODE_COL_CANDIDATES = ("ts_code", "stock_code", "code", "symbol", "sec_code", "secu_code",
+                       "security_code", "gp_code")
+
+_cat_cols = {"at": 0.0, "data": None}
+CAT_COLS_TTL = 600.0
+
+
+def category_columns(force: bool = False) -> dict:
+    """`SHOW COLUMNS FROM gp_stock_category` → 认出代码列与行业列 (缓存 10 分钟)。
+
+    先认列再查, 而不是拿一堆候选列名硬撞: 2026-07-31 实测撞出来的是
+    `Unknown column 'stock_code'` —— 报错本身没错, 但撞法会把「列不存在」和「这只票不在
+    表里」混成一锅, 而且错误串长到没法看。SHOW COLUMNS 一次就说清了。
+    """
+    import time
+    now = time.time()
+    if not force and _cat_cols["data"] is not None and now - _cat_cols["at"] < CAT_COLS_TTL:
+        return _cat_cols["data"]
+    d = {"columns": None, "code_col": None, "industry_col": None, "error": None}
+    try:
+        rows = fetch_all("SHOW COLUMNS FROM gp_stock_category")
+        cols = [str(r.get("Field") or r.get("field") or r.get("COLUMN_NAME") or "")
+                for r in (rows or [])]
+        cols = [c for c in cols if c]
+        d["columns"] = cols
+        low = {c.lower(): c for c in cols}
+        d["code_col"] = next((low[k] for k in CODE_COL_CANDIDATES if k in low), None)
+        d["industry_col"] = next((low[k] for k in INDUSTRY_COL_CANDIDATES if k in low), None)
+        if not cols:
+            d["error"] = "SHOW COLUMNS 回了空 —— 表可能不存在或代理不给看"
+        elif not d["code_col"]:
+            d["error"] = f"没有可识别的代码列; 实际列: {cols}"
+        elif not d["industry_col"]:
+            d["error"] = f"没有可识别的行业列; 实际列: {cols}"
+    except Exception as e:
+        d["error"] = f"{type(e).__name__}: {str(e)[:200]}"
+    _cat_cols.update({"at": now, "data": d})
+    return d
 
 
 def probe_category(ts_code: str) -> dict:
@@ -214,27 +253,35 @@ def probe_category(ts_code: str) -> dict:
     于是行业硬拦截会**静默失效**。调用方要判"数据源到底通不通", 必须看得见 error。
 
     返回 {value, code_col, industry_col, columns, error}: value=None 且 error=None
-    才是真正的"查到了行, 但没有行业列"或"这只票不在表里"。
+    才是真正的"表能用, 只是这只票不在里面"。
     """
-    out = {"value": None, "code_col": None, "industry_col": None, "columns": None,
+    meta = category_columns()
-           "error": None}
+    out = {"value": None, "code_col": meta.get("code_col"),
-    errs = []
+           "industry_col": meta.get("industry_col"), "columns": meta.get("columns"),
-    for col in ("stock_code", "ts_code"):
+           "error": meta.get("error")}
+    if out["error"] or not out["code_col"]:
+        return out
+    col = out["code_col"]
+    dot = to_dot(ts_code)
+    forms = [dot, to_prefix(dot), dot.split(".")[0]]          # 点式 / 前缀式 / 纯数字
+    for form in dict.fromkeys(forms):
         try:
             r = fetch_one(f"SELECT * FROM gp_stock_category WHERE {col} = :code LIMIT 1",
-                          {"code": to_prefix(ts_code) if col == "stock_code" else ts_code})
+                          {"code": form})
         except Exception as e:
-            errs.append(f"{col}: {type(e).__name__}: {e}")
+            out["error"] = f"{col}={form}: {type(e).__name__}: {str(e)[:160]}"
-            continue
+            return out
         if r:
-            out["code_col"], out["columns"] = col, sorted(r.keys())
+            out["matched_form"] = form
-            for k in INDUSTRY_COL_CANDIDATES:
+            ind = out["industry_col"]
-                if r.get(k):
+            if ind and r.get(ind):
-                    out["value"], out["industry_col"] = str(r[k]), k
+                out["value"] = str(r[ind])
-                    return out
+            else:
-            return out                      # 有行但没有任何已知行业列 —— 也是有效信息
+                for k in INDUSTRY_COL_CANDIDATES:
-    if errs and out["code_col"] is None:
+                    if r.get(k):
-        out["error"] = "; ".join(errs[:2])
+                        out["value"], out["industry_col"] = str(r[k]), k
+                        break
+            return out
     return out
 
 

app/repo/industry_repo.py

@@ -0,0 +1,137 @@
+# -*- coding: utf-8 -*-
+"""
+行业板块表 gp_hybk (199 库, 只读)
+==================================
+数据源由另一项目的持仓行业分析给出 (2026-07-31 参考其实现确认):
+
+    库    DB_MYSQL_URL (192.168.18.199) —— PMS 里原本只用来读大盘指数 zs_day_data,
+          `session.fetch_all(..., source="index")` 直接可用, 不新增连接。
+    表    gp_hybk
+    列    gp_code (股票代码) / bk_code (板块代码, 数值型) / bk_name (板块名)
+    口径  bk_code 前缀 881 = 二级行业, 884 = 三级行业
+    特点  **一只票会对应多条同级行业**
+
+PMS 用**三级 (884)** 做行业集中度硬拦截, 且**每票只认一个主行业** ——
+`sizer.check_caps` 的 `ctx["sector"]` 是单值, 累计口径 (`sector_names_map` /
+`sector_mv_map`) 也按单值建。主行业的选法必须**稳定**: 同一只票每次都要得到同一个行业,
+否则今天算「工程机械」明天算「专用设备」, 集中度累计会自己跳。故定为
+**bk_code 升序取第一个** —— 纯粹按数值排, 不依赖查询返回次序, 也不依赖表里有没有主次标记。
+
+代码写法未确认 (点式 600000.SH / 前缀式 SH600000 / 纯数字 600000 都可能), 所以
+`probe()` 会三种各试一遍, 认出哪种能命中就缓存下来, 之后批量查一律用那种。
+"""
+from __future__ import annotations
+
+import logging
+import threading
+import time
+
+from app.db.session import fetch_all
+from app.repo.downstream_repo import to_dot, to_prefix
+
+logger = logging.getLogger("pms.hybk")
+
+TABLE = "gp_hybk"
+SOURCE = "index"                      # DB_MYSQL_URL
+LEVEL_PREFIX = {"l2": "881", "l3": "884"}
+BATCH = 400                           # 一次 IN 多少个代码 (持仓+候选也就几十, 留足余量)
+
+# 代码写法探测样本: 沪/深各一, 一定在任何行业表里
+PROBE_CODES = ("600000.SH", "000001.SZ", "600519.SH")
+
+_fmt = {"at": 0.0, "form": None, "error": None, "columns": None}
+_lock = threading.Lock()
+FMT_TTL = 600.0
+
+
+def _forms(ts_code: str) -> dict:
+    dot = to_dot(ts_code)
+    return {"dot": dot, "prefix": to_prefix(dot), "num": dot.split(".")[0]}
+
+
+def _as(ts_code: str, form: str) -> str:
+    return _forms(ts_code).get(form) or to_dot(ts_code)
+
+
+def _rows_for(codes: list) -> list:
+    """一次 IN 查询。不用 CAST/LIKE —— 前缀判断放 Python 侧, 免得 SQL 方言与单表守卫扯皮。"""
+    if not codes:
+        return []
+    keys = [f"c{i}" for i in range(len(codes))]
+    sql = (f"SELECT gp_code, bk_code, bk_name FROM {TABLE} "
+           f"WHERE gp_code IN ({', '.join(':' + k for k in keys)})")
+    return fetch_all(sql, dict(zip(keys, codes)), source=SOURCE)
+
+
+def probe(force: bool = False) -> dict:
+    """认代码写法。返回 {form, error, columns, tried}; form=None 表示这张表用不了。
+
+    三种写法各拿 PROBE_CODES 试一次, 谁先命中用谁。命中不了但也没报错, 说明表能查、
+    只是这些票不在里面 (或代码写法是第四种) —— 这跟"表根本查不了"要分开报, 不能都吞成
+    "没有行业"。行业约束是硬拦截, 拿不到就该明着停, 见 UPSTREAM_PLAN_API.md §7.4。
+    """
+    now = time.time()
+    if not force and _fmt["form"] is not None and now - _fmt["at"] < FMT_TTL:
+        return dict(_fmt, tried=list(PROBE_CODES))
+    err = None
+    for form in ("dot", "prefix", "num"):
+        codes = [_as(c, form) for c in PROBE_CODES]
+        try:
+            rows = _rows_for(codes)
+        except Exception as e:
+            err = f"{form}({codes[0]}): {type(e).__name__}: {str(e)[:180]}"
+            continue
+        if rows:
+            cols = sorted(rows[0].keys())
+            with _lock:
+                _fmt.update({"at": now, "form": form, "error": None, "columns": cols})
+            logger.info("[行业] gp_hybk 代码写法 = %s (样本 %s), 列 %s", form, codes[0], cols)
+            return dict(_fmt, tried=codes)
+    with _lock:
+        _fmt.update({"at": now, "form": None, "columns": None,
+                     "error": err or f"三种代码写法都没命中 (试了 {list(PROBE_CODES)}); "
+                                     f"表能查但里面没有这些票, 或 gp_code 是第四种写法"})
+    return dict(_fmt, tried=list(PROBE_CODES))
+
+
+def invalidate():
+    with _lock:
+        _fmt.update({"at": 0.0, "form": None, "error": None, "columns": None})
+
+
+def fetch_industries(codes, level: str = "l3") -> dict:
+    """{ts_code(点式): [(bk_code, bk_name), ...]}, 已按 bk_code 升序。
+
+    只回该 level 的板块; 查不到的票不出现在结果里 (调用方据此判 None)。
+    """
+    prefix = LEVEL_PREFIX.get(level, LEVEL_PREFIX["l3"])
+    p = probe()
+    form = p.get("form")
+    if not form:
+        raise RuntimeError(p.get("error") or "gp_hybk 不可用")
+    want = {}
+    for c in codes or []:
+        d = to_dot(c)
+        if d:
+            want.setdefault(_as(d, form), d)
+    out = {}
+    keys = list(want)
+    for i in range(0, len(keys), BATCH):
+        for r in _rows_for(keys[i:i + BATCH]):
+            raw = str(r.get("gp_code") or "").strip().upper()
+            dot = want.get(raw) or want.get(raw.upper())
+            bk = str(r.get("bk_code") or "").strip()
+            name = str(r.get("bk_name") or "").strip()
+            if not dot or not name or not bk.startswith(prefix):
+                continue
+            out.setdefault(dot, [])
+            if (bk, name) not in out[dot]:
+                out[dot].append((bk, name))
+    for v in out.values():
+        v.sort(key=lambda x: x[0])          # bk_code 升序 —— 主行业选取要可复现
+    return out
+
+
+def primary_industry_map(codes, level: str = "l3") -> dict:
+    """{ts_code: 主行业名}。主行业 = 该级板块里 bk_code 最小的那个 (见模块头部)。"""
+    return {c: v[0][1] for c, v in fetch_industries(codes, level=level).items() if v}

app/services/industry.py

@@ -5,8 +5,11 @@
     IndustryClassifier.get(ts_code) -> 行业名 | None
 
 数据源经参数 PMS_SECTOR_SOURCE 切换:
-  custom_table       自定义映射表 pms_industry_map (默认建议; 用户后续灌何种划分都只是灌数)
+  gp_hybk            **当前口径** —— 199 库的行业板块表, 取三级 (bk_code 884*), 每票只认
-  gp_stock_category  决策系统生态已有的行业表 (映射表就绪前的临时数据源)
+                     一个主行业 (bk_code 升序第一个)。详见 app/repo/industry_repo.py
+  custom_table       自定义映射表 pms_industry_map (要自己灌数时用)
+  gp_stock_category  ~~决策系统生态的行业表~~ **实测不可用**: 该表没有 stock_code 列,
+                     ts_code 也查不到样本票 (2026-07-31)。保留枚举只为兼容, 别再用
   ""(空)             未配置 → 行业约束自动停用, 页面显著提示 (不静默放行也不误拦)
 
 映射表现在有稳定的灌数来源: 上游选股计划 `/plan` 的 `evidence.theme`, 由
@@ -20,7 +23,7 @@ import logging
 import threading
 import time
 
-from app.repo import downstream_repo, pms_repo
+from app.repo import downstream_repo, industry_repo, pms_repo
 from app.services import param_store
 
 logger = logging.getLogger("pms.industry")
@@ -35,6 +38,9 @@ PROBE_FALLBACK = ("600000.SH", "000001.SZ", "600519.SH")
 PROBE_SAMPLE_MAX = 5
 
 
+VALID_SOURCES = ("gp_hybk", "custom_table", "gp_stock_category")
+
+
 def source() -> str:
     return (param_store.get("PMS_SECTOR_SOURCE", "") or "").strip()
 
@@ -70,7 +76,9 @@ def _probe_category() -> dict:
             errors.append(f"{c}: {type(e).__name__}: {e}")
             continue
         if r.get("error"):
-            errors.append(f"{c}: {r['error']}")
+            e = str(r["error"])[:200]
+            if e not in errors:            # 同一个列名错误会对每只样本各报一遍, 去重
+                errors.append(e)
             continue
         cols = cols or r.get("columns")
         if r.get("value"):
@@ -83,6 +91,12 @@ def _probe_category() -> dict:
     return data
 
 
+def level() -> str:
+    """gp_hybk 取哪一级 (l2=881 / l3=884)。默认三级 —— 二级太粗, 4 只堆一个二级行业拦不住。"""
+    v = (param_store.get("PMS_SECTOR_HYBK_LEVEL", "l3") or "l3").strip().lower()
+    return v if v in ("l2", "l3") else "l3"
+
+
 def ready() -> bool:
     """行业约束是否**真的**生效 —— 配置了数据源, 且那个源确实给得出行业。
 
@@ -90,6 +104,12 @@ def ready() -> bool:
     sizer.check_caps 里等于"跳过行业约束"。所以这里必须要有证据。
     """
     src = source()
+    if src == "gp_hybk":
+        try:
+            return bool(industry_repo.probe().get("form"))
+        except Exception as e:
+            logger.warning("gp_hybk 探测失败, 按数据源不可用处理: %s", e)
+            return False
     if src == "custom_table":
         try:
             return len(_load_custom()) > 0
@@ -109,6 +129,25 @@ def status() -> dict:
     st = {"source": src, "ready": ready(), "count": 0,
           "hint": "行业划分数据源未配置 —— 行业集中度硬拦截已自动停用, "
                   "行业类命令置灰 (设计 §5/§13)"}
+    if src == "gp_hybk":
+        try:
+            pr = industry_repo.probe()
+        except Exception as e:
+            st.update({"ready": False, "hint": f"gp_hybk 探测失败: {type(e).__name__}: {e}"})
+            return st
+        st["probe"] = {k: pr.get(k) for k in ("form", "error", "columns", "tried")}
+        st["level"] = level()
+        st["count"] = sum(1 for v in _hybk["data"].values() if v)
+        if pr.get("form"):
+            st["hint"] = (f"gp_hybk 可用 (代码写法 {pr['form']}, 取 {st['level']} 级 "
+                          f"bk_code {industry_repo.LEVEL_PREFIX[st['level']]}*, "
+                          f"每票只认 bk_code 最小的那个主行业); 本日已缓存 {st['count']} 只")
+        else:
+            st["ready"] = False
+            st["hint"] = ("gp_hybk **用不了** —— 行业约束按未配置停用。" + str(pr.get("error"))
+                          + "。gp_hybk 在 DB_MYSQL_URL 指向的库 (默认 199/db_gp_cj), "
+                            "若不在那儿要改 .env")
+        return st
     if src == "custom_table":
         try:
             st["count"] = len(_load_custom())
@@ -133,8 +172,9 @@ def status() -> dict:
         else:
             st["hint"] = ("gp_stock_category **一只都没查到** —— 行业约束等同于未配置且会"
                           "静默失效。样本 " + ", ".join(pr["sample_codes"][:3])
-                          + ("; 错误: " + " | ".join(pr["errors"]) if pr["errors"]
+                          + ("; 原因: " + " | ".join(pr["errors"][:2]) if pr["errors"]
-                             else "; 无报错, 是表里没有这些票或没有行业列"))
+                             else "; 无报错, 是表里没有这些票 (代码写法已试过点式/前缀式/纯数字)")
+                          + "。用 运维 → 导出下游表结构 看 _category_probe 的实际列名")
     return st
 
 
@@ -151,26 +191,57 @@ def _load_custom() -> dict:
 
 def get(ts_code: str):
     """返回行业名; 未配置数据源或查不到 → None (调用方据此跳过行业约束)。"""
-    src = source()
+    if not ts_code:
-    if not ts_code or src not in ("custom_table", "gp_stock_category"):
-        return None
-    try:
-        if src == "custom_table":
-            return _load_custom().get(ts_code)
-        return downstream_repo.fetch_sector_from_category(ts_code)
-    except Exception as e:
-        logger.warning("行业查询失败 [%s]: %s", ts_code, e)
         return None
+    return get_many([ts_code]).get(ts_code)
+
+
+_hybk = {"day": None, "data": {}}
+
+
+def _hybk_many(codes) -> dict:
+    """gp_hybk 批量取主行业, 按日缓存 (板块归属是日频的, 盘中重复查没意义)。"""
+    import datetime as _dt
+    today = _dt.datetime.now().strftime("%Y%m%d")
+    if _hybk["day"] != today:
+        _hybk.update({"day": today, "data": {}})
+    miss = [c for c in codes if c and c not in _hybk["data"]]
+    if miss:
+        try:
+            got = industry_repo.primary_industry_map(miss, level=level())
+        except Exception as e:
+            logger.warning("gp_hybk 批量查询失败 [%d 只]: %s", len(miss), e)
+            got = {}
+        for c in miss:
+            _hybk["data"][c] = got.get(c)      # 查不到也缓存 None, 免得每跳都重查
+    return {c: _hybk["data"].get(c) for c in (codes or [])}
 
 
 def get_many(codes) -> dict:
     src = source()
-    if src != "custom_table":
+    codes = [c for c in (codes or []) if c]
-        return {c: get(c) for c in (codes or [])}
+    if src == "gp_hybk":
-    m = _load_custom()
+        return _hybk_many(codes)
-    return {c: m.get(c) for c in (codes or [])}
+    if src == "custom_table":
+        m = _load_custom()
+        return {c: m.get(c) for c in codes}
+    if src == "gp_stock_category":
+        out = {}
+        for c in codes:
+            try:
+                out[c] = downstream_repo.fetch_sector_from_category(c)
+            except Exception as e:
+                logger.warning("行业查询失败 [%s]: %s", c, e)
+                out[c] = None
+        return out
+    return {c: None for c in codes}
 
 
 def invalidate():
     _cache["at"] = 0.0
     _probe.update({"at": 0.0, "data": None})
+    _hybk.update({"day": None, "data": {}})
+    try:
+        industry_repo.invalidate()
+    except Exception:
+        pass

app/services/param_store.py

@@ -80,7 +80,8 @@ DESC = {
     "PMS_PLAN_EXCLUDE_ST": "剔除 ST/*ST/退市整理 (上游名单明确不剔除, 这道闸在 PMS 侧)",
     "PMS_PLAN_THEME_SYNC": "把 theme 灌进 pms_industry_map。默认关 —— theme 是事件驱动的传导主题(覆盖约三成、词表不规范), 当行业标签会让集中度约束天天漂; 行业源请用 gp_stock_category",
     "PMS_PLAN_QUERY_EXTRA": "计划接口附加查询串逃生口 (上游加了新参数时不用改代码); top/obs_top/theme_cap 请用各自的显式参数, 同名键以显式参数为准",
-    "PMS_SECTOR_SOURCE": "行业划分数据源: 空=约束停用 / custom_table (推荐, 由上游计划的 theme 灌数) / gp_stock_category",
+    "PMS_SECTOR_SOURCE": "行业划分数据源: 空=约束停用 / gp_hybk (199 库行业板块表, 当前口径) / custom_table (自己灌数) / gp_stock_category (实测不可用, 勿选)",
+    "PMS_SECTOR_HYBK_LEVEL": "gp_hybk 取哪一级: l3=三级 bk_code 884* (默认) / l2=二级 881*",
     "PMS_SECTOR_MAX_NAMES": "同行业最大持仓只数 (硬拦截)",
     "PMS_SECTOR_MAX_RATIO": "同行业最大占总仓比例 (硬拦截)",
     "PMS_EXEC_WINDOW_TDAYS": "任务命令默认执行窗口 (交易日)",
@@ -278,8 +279,11 @@ _RANGES = {
 def _range_check(key, v):
     if key == "PMS_AUTONOMY" and v not in ("full", "propose_only", "off"):
         return "PMS_AUTONOMY 只能是 full / propose_only / off"
-    if key == "PMS_SECTOR_SOURCE" and v not in ("", "custom_table", "gp_stock_category"):
+    if key == "PMS_SECTOR_SOURCE" and v not in ("", "gp_hybk", "custom_table",
-        return "PMS_SECTOR_SOURCE 只能是 空 / custom_table / gp_stock_category"
+                                                "gp_stock_category"):
+        return "PMS_SECTOR_SOURCE 只能是 空 / gp_hybk / custom_table / gp_stock_category"
+    if key == "PMS_SECTOR_HYBK_LEVEL" and v not in ("l2", "l3"):
+        return "PMS_SECTOR_HYBK_LEVEL 只能是 l2 / l3"
     if key == "PMS_DISPATCH_MODE" and v not in ("shadow", "ws"):
         return "PMS_DISPATCH_MODE 只能是 shadow / ws"
     if key == "PMS_CANDIDATE_SOURCE" and v not in ("plan_api", "buy_plan", "both"):

app/web/main.py

@@ -448,7 +448,8 @@ def api_downstream_schema():
     """导出下游三表的实际列定义 —— 用于回填 QMT_INTERFACE_REQUIREMENTS D1。"""
     def _schema():
         out = {}
-        for t in ("trading_position", "trading_order", "trading_buy_plan"):
+        for t in ("trading_position", "trading_order", "trading_buy_plan",
+                  "gp_stock_category"):
             try:
                 out[t] = downstream_repo.describe(t)
             except Exception as e:
@@ -457,6 +458,10 @@ def api_downstream_schema():
             out["_position_probe"] = downstream_repo.fetch_positions()["columns"]
         except Exception as e:
             out["_position_probe"] = {"error": str(e)}
+        try:
+            out["_category_probe"] = downstream_repo.category_columns(force=True)
+        except Exception as e:
+            out["_category_probe"] = {"error": str(e)}
         return out
     return ok(_schema)
 

config/settings.py

@@ -108,7 +108,10 @@ class Settings(BaseSettings):
     #   不用改代码即可透传; top/obs_top/theme_cap 已有显式参数, 同名键以显式参数为准
 
     # --- 行业约束 (硬拦截; 数据源接口化) ---
-    PMS_SECTOR_SOURCE: str = ""            # "" = 停用并页面提示 / custom_table / gp_stock_category
+    #   行业划分数据源。gp_hybk = 199 库的行业板块表 (三级 884*, 每票只认一个主行业),
+    #   2026-07-31 起的正式口径; gp_stock_category 实测不可用 (无 stock_code 列), 勿用。
+    PMS_SECTOR_SOURCE: str = ""            # "" = 停用并页面提示 / gp_hybk / custom_table
+    PMS_SECTOR_HYBK_LEVEL: str = "l3"      # gp_hybk 取哪一级: l3=884(三级, 默认) / l2=881(二级)
     PMS_SECTOR_MAX_NAMES: int = 4
     PMS_SECTOR_MAX_RATIO: float = 0.40     # 占总仓
 

scripts/run_tests.py

@@ -12,8 +12,8 @@
   test_batch5_units.py  决策系统信号流解析与消化口径                    (8 例)
   test_batch6_units.py  ws 通道: 测试向量/签名/公钥/水位/DDL/逐笔入账    (65 例)
   test_batch7_units.py  上游选股计划: 解析/新鲜度/候选筛选/取数守卫      (32 例)
-  test_wiring.py        装配自检: 服务层→核心→落表 全链路 (内存桩)      (55 例)
+  test_wiring.py        装配自检: 服务层→核心→落表 全链路 (内存桩)      (58 例)
-                                                                  共 241 例
+                                                                  共 244 例
 任一子集失败即整体失败 (退出码 1)。
 """
 import os

scripts/test_wiring.py

@@ -470,6 +470,14 @@ def install_fakes(prices=None, positions=None, params=None, high5=None, prev_clo
     downstream_repo.fetch_positions = lambda: {"rows": [], "columns": {"qty": None},
                                                "raw_count": 0}
     downstream_repo.fetch_refs = lambda c: None
+    downstream_repo.category_columns = lambda force=False: {
+        "columns": None, "code_col": None, "industry_col": None,
+        "error": "单测桩: 未接行业表"}
+    from app.repo import industry_repo
+    industry_repo.probe = lambda force=False: {"form": None, "error": "单测桩: 未接 gp_hybk",
+                                               "columns": None, "tried": []}
+    industry_repo.primary_industry_map = lambda codes, level="l3": {}
+    industry_repo.invalidate = lambda: None
     downstream_repo.latest_filled_order_id = lambda: "ANCHOR_0"
     # 回放游标预置成非空 —— 否则 replay_fills 会走「冷启动只对齐不追认」那条路 (见
     # ledger_service._seed_cursor), 下面那几个回放用例就测不到入账。冷启动本身另有专门用例。
@@ -1688,6 +1696,14 @@ def _real_industry():
     return ind
 
 
+def _real_downstream():
+    """同上, 恢复 downstream_repo —— install_fakes 把 category_columns 也打了桩。"""
+    import importlib
+    from app.repo import downstream_repo as dr
+    importlib.reload(dr)
+    return dr
+
+
 @case("行业源·gp_stock_category 查得到才算 ready (参数填了不等于数据源能用)")
 def _():
     from app.repo import downstream_repo
@@ -1722,7 +1738,7 @@ def _():
         assert ind.ready() is False
         st = ind.status()
         assert st["ready"] is False and "一只都没查到" in st["hint"], st["hint"]
-        assert "没有行业列" in st["hint"], st["hint"]
+        assert "表里没有这些票" in st["hint"] and "导出下游表结构" in st["hint"], st["hint"]
         # ② 表根本查不了 (有报错) —— 错误必须冒到 hint 上, 不能吞
         ind.invalidate()
         downstream_repo.probe_category = lambda c: {
@@ -1734,6 +1750,119 @@ def _():
         downstream_repo.probe_category = orig
 
 
+@case("行业源·gp_hybk: 三级 884 / 每票只认 bk_code 最小的主行业 / 代码写法自适应")
+def _():
+    from app.repo import industry_repo as ir
+    install_fakes(params={"PMS_SECTOR_SOURCE": "gp_hybk"})
+    ind = _real_industry()
+    import importlib
+    importlib.reload(ir)
+    orig = ir.fetch_all
+    try:
+        # 表里用前缀式 SH600000; 一票挂 二级×1 + 三级×2
+        TBL = [
+            {"gp_code": "SH600000", "bk_code": 881155, "bk_name": "银行(二级)"},
+            {"gp_code": "SH600000", "bk_code": 884219, "bk_name": "股份制银行"},
+            {"gp_code": "SH600000", "bk_code": 884101, "bk_name": "全国性银行"},
+            {"gp_code": "SZ000001", "bk_code": 884219, "bk_name": "股份制银行"},
+        ]
+        calls = []
+
+        def _fa(sql, params=None, source=None):
+            assert source == "index", source          # 必须走 199, 不是 153 代理
+            want = set((params or {}).values())
+            calls.append(sorted(want))
+            return [r for r in TBL if r["gp_code"] in want]
+        ir.fetch_all = _fa
+        p = ir.probe(force=True)
+        assert p["form"] == "prefix", p               # 点式先试没命中, 自动落到前缀式
+        m = ir.fetch_industries(["600000.SH", "000001.SZ"], level="l3")
+        assert m["600000.SH"] == [("884101", "全国性银行"), ("884219", "股份制银行")], m
+        # 主行业 = bk_code 升序第一个 (884101 < 884219), 二级 881 不能混进来
+        pm = ir.primary_industry_map(["600000.SH", "000001.SZ"], level="l3")
+        assert pm == {"600000.SH": "全国性银行", "000001.SZ": "股份制银行"}, pm
+        assert ir.primary_industry_map(["600000.SH"], level="l2") == {"600000.SH": "银行(二级)"}
+        # 服务层: ready 有证据, get_many 批量且按日缓存
+        ind.invalidate()
+        assert ind.ready() is True
+        assert ind.get_many(["600000.SH", "000001.SZ"]) == {
+            "600000.SH": "全国性银行", "000001.SZ": "股份制银行"}
+        n = len(calls)
+        ind.get_many(["600000.SH", "000001.SZ"])      # 二次不再打库
+        assert len(calls) == n, calls[n:]
+        assert ind.get("600000.SH") == "全国性银行"
+        st = ind.status()
+        assert st["ready"] is True and st["level"] == "l3" and "884" in st["hint"], st
+    finally:
+        ir.fetch_all = orig
+        ir.invalidate()
+
+
+@case("行业源·gp_hybk 三种写法都不命中 → ready=False, 错误说清是表能查还是查不了")
+def _():
+    from app.repo import industry_repo as ir
+    install_fakes(params={"PMS_SECTOR_SOURCE": "gp_hybk"})
+    ind = _real_industry()
+    import importlib
+    importlib.reload(ir)
+    orig = ir.fetch_all
+    try:
+        ir.fetch_all = lambda sql, params=None, source=None: []      # 能查, 但没有这些票
+        ind.invalidate()
+        assert ind.ready() is False
+        st = ind.status()
+        assert st["ready"] is False and "用不了" in st["hint"], st["hint"]
+        assert "三种代码写法都没命中" in st["hint"], st["hint"]
+        assert "DB_MYSQL_URL" in st["hint"], st["hint"]     # 指到该改哪儿
+        # 查不了的情况: 错误原文要冒上来
+        ir.invalidate()
+
+        def _boom(sql, params=None, source=None):
+            raise RuntimeError("Table 'db_gp_cj.gp_hybk' doesn't exist")
+        ir.fetch_all = _boom
+        ind.invalidate()
+        assert ind.ready() is False
+        assert "doesn't exist" in ind.status()["hint"], ind.status()["hint"]
+    finally:
+        ir.fetch_all = orig
+        ir.invalidate()
+
+
+@case("行业源·先 SHOW COLUMNS 认列再查: 没有代码列时直接给可执行的报错")
+def _():
+    install_fakes(params={"PMS_SECTOR_SOURCE": "gp_stock_category"})
+    ind = _real_industry()
+    dr = _real_downstream()          # category_columns 也被 install_fakes 打过桩
+    orig_fa, orig_fo = dr.fetch_all, dr.fetch_one
+    try:
+        # ① 实机遇到的情况: 表在, 但没有 stock_code 这类代码列
+        dr._cat_cols.update({"at": 0.0, "data": None})
+        dr.fetch_all = lambda sql, p=None: [{"Field": "gp_name"}, {"Field": "sw_l1"}]
+        meta = dr.category_columns(force=True)
+        assert meta["code_col"] is None and "没有可识别的代码列" in meta["error"], meta
+        assert "gp_name" in meta["error"], meta["error"]      # 把实际列名报出来才可执行
+        assert dr.probe_category("600000.SH")["value"] is None
+        assert ind.ready() is False
+        # ② 列认出来了: 代码写法逐个试, 命中即返回
+        dr._cat_cols.update({"at": 0.0, "data": None})
+        dr.fetch_all = lambda sql, p=None: [{"Field": "ts_code"}, {"Field": "sw_l1"}]
+        seen = []
+
+        def _one(sql, p=None):
+            seen.append((p or {}).get("code"))
+            return {"ts_code": "SH600000", "sw_l1": "银行"} if (p or {}).get("code") == "SH600000" else None
+        dr.fetch_one = _one
+        dr.category_columns(force=True)
+        r = dr.probe_category("600000.SH")
+        assert r["value"] == "银行" and r["industry_col"] == "sw_l1", r
+        assert seen == ["600000.SH", "SH600000"], seen     # 点式先试, 再前缀式
+        ind.invalidate()
+        assert ind.ready() is True
+    finally:
+        dr.fetch_all, dr.fetch_one = orig_fa, orig_fo
+        dr._cat_cols.update({"at": 0.0, "data": None})
+
+
 @case("行业源·custom_table 映射表为空同样不算 ready (空表 = 全票 None = 约束失效)")
 def _():
     fake = install_fakes(params={"PMS_SECTOR_SOURCE": "custom_table"})