行业源ready改实探(参数填了不算数); probe_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       装配自检: 服务层→核心→落表 全链路 (内存桩)   52 例
+  test_wiring.py       装配自检: 服务层→核心→落表 全链路 (内存桩)   55 例
   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 水位与累积确认，见下）；**单测 235 例**。
+**已实现**：建表 DDL 与建表脚本；配置与运行参数中心；仓位规划器与安全垫账；命令系统（27 类命令全目录 + 双状态机 + 冲突识别）；方案生成器（降仓凑额四档、升仓、建仓分批、清仓/减至、行业清仓与限额、暂停买入撤单）；账本回放与对账引擎（成交认领、外部成交并入 BASE 告警、以下游为准修正、除权检测、T+1 可用量、连续不一致升级）；规则闸终检；择时执行器实现 B（分日配额、分笔、VWAP/回踩/不追高、14:45 兜底、停牌一字板顺延、窗口耗尽收口、挂单有效期）；动作引擎四类自主动作 + 研判闸客户端 + 提议分流；决策系统信号消化（两条流独立消费组订阅、置信度分档转清仓指令或提议）；管理页面四块 + 运维/日报抽屉；调度器九个调度位；**上游选股计划接口接入**（`/plan` 取候选池、交易日龄硬校验、`theme` 灌行业映射表、页面预览抽屉与不可用横幅）；**ws 直连通道的连接层**（常驻进程 + 出口队列 + 签名 + seq 水位与累积确认，见下）；**单测 241 例**。
 
 ### 下一步（按可动工顺序）
 
@@ -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 | ~~行业硬拦截开闸~~ | ✅ 2026-07-31：`PMS_SECTOR_SOURCE=gp_stock_category`（`theme` 是传导主题不是行业，已放弃灌 `pms_industry_map`，理由见 `UPSTREAM_PLAN_API.md` §6.1）。**待验证**：`GET /api/industry` 的 `status.ready` 是否为 true、覆盖多少条 |
+| 3 | **行业硬拦截开闸** | 参数已切 `gp_stock_category`；`ready()` 改成小样本实探（参数填了不算数，见 `UPSTREAM_PLAN_API.md` §7.4）。**待验证**：部署后再查 `GET /api/industry`，看 `status.probe.hit` 是否 > 0、命中的列名是什么 —— hit=0 就说明这张表用不了，得回头灌 `custom_table` |
 | 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

@@ -385,7 +385,7 @@ PMS_SECTOR_SOURCE=gp_stock_category      行业硬拦截的数据源 (07-31 开
 `白名单档位取全: 是`, 主题限额把 37 只挤出去, 候选池 10 只 —— 按 `STOCK_TARGET_DEFAULT=6%`
 算正好 60%, 与 `PORTFOLIO_CAP=60%` 对得上, 不缺票。
 
-### 7.3 要报给上游的观察: 同一 `date` 的计划会变, 且档位规模与文档不符
+### 7.3 同一 `date` 的计划会变 (已知行为, 非故障)
 
 `/plan` 不是读静态文件, 是 `plan.collect()` **实时从库里算**的。同一个 `date=2026-07-30`,
 相隔约一小时的两次请求给出了不同的结果:
@@ -396,10 +396,45 @@ PMS_SECTOR_SOURCE=gp_stock_category      行业硬拦截的数据源 (07-31 开
   数据在中途重载, 请求正好落在半截快照上。
 - **榜首换人**, 艾罗能源整个掉出主榜。
 
-**请上游确认**: ① 同一 `date` 的计划是否应当幂等? 若否, 建议给个 `generated_at` 之类的
+**已与上游对齐 (2026-07-31)**: 上游当天刚做过调整, 这个前后不一致是**预期内的**, 不是
-版本戳, 下游才能判断"手上这份是哪一版"。② 券商预期是否有重载窗口, 那期间的请求要不要
+半截快照。
-挡掉 (返回 409/503 比返回半截数据安全)。
+
+仍值得提一句的改进 (非阻塞): 给应答加个 `generated_at` / 版本戳。`/plan` 是实时算的,
+同一 `date` 可以对应多个版本, 下游现在没有任何办法判断"手上这份是哪一版"—— 出问题复盘时
+只能靠猜。PMS 侧已经把 `date` / `funnel` / `requested` 全记进日志与页面, 但那记的是
+"我拿到了什么", 不是"上游发的是哪一版"。
 
 对 PMS 的影响目前可控: 候选池只在**命令规划期**取一次, 规划完即落 `pms_plan`, 之后不再
 重取 —— 不会出现"方案在途中候选池换了"。但盘前拉与盘中下命令若跨了重载窗口, 两次看到的
 榜可能不一样, 心里要有数。
+
+
+### 7.4 行业源实探 (07-31 收尾时补的一个洞)
+
+切完 `PMS_SECTOR_SOURCE=gp_stock_category` 之后查 `/api/industry`, 回的是:
+
+```json
+{"status": {"source": "gp_stock_category", "ready": true, "count": 0}, "rows": []}
+```
+
+**这个输出什么都没证明。** 老的 `ready()` 只做了一件事: 看参数值在不在枚举里 —— 参数一填
+就报 `true`。而 `count` 与 `rows` 统计的是 `pms_industry_map` (custom_table 的表), 对
+gp_stock_category 一个字的证据都没有。再加上 repo 层 `fetch_sector_from_category` 把异常
+`except: continue` 吞成 None, 而 `sizer.check_caps` 里 `sector=None` 就是**跳过行业约束**——
+三件事叠起来的效果是: 表可能根本查不了, 页面上一片绿, 行业硬拦截静默失效。
+
+这跟本仓库到处在讲的「拿不到 ≠ 通过」是同一件事, 已修:
+
+- `downstream_repo.probe_category()` —— 把查询过程带出来 (`value` / `code_col` /
+  `industry_col` / `columns` / `error`), 区分"这只票不在表里"和"这张表查不了"。
+  `fetch_sector_from_category()` 退化成取它的 `value`。
+- `industry.ready()` —— **要有证据**。`gp_stock_category` 拿持仓票 (不足补三只大盘股)
+  做小样本实探, 命中 0 就是 `False`; `custom_table` 空映射表同样是 `False`。
+  缓存 300 秒, 不进热路径。
+- `industry.status()` —— 命中数、命中的列名、错误原文全带上, 一只没查到时 hint 直接写
+  「**一只都没查到** —— 行业约束等同于未配置且会静默失效」。
+
+`ready=False` 的连锁反应是设计里本来就有的: 行业类命令在页面置灰、`sector_source_ready`
+让 planner 把 sector 置空、页面顶部挂黄色横幅。**该停的时候明着停**, 比绿着失效强。
+
+三条单测锁住: 查得到才 ready / 查不到 ready=False 且错误原文冒到 hint / 空映射表不算 ready。

app/repo/downstream_repo.py

@@ -203,16 +203,41 @@ def fetch_refs(ts_code: str):
             "trade_date": r.get("trade_date")}
 
 
-def fetch_sector_from_category(ts_code: str):
+INDUSTRY_COL_CANDIDATES = ("industry", "category", "sector", "industry_name", "sw_industry")
-    """行业适配器 gp_stock_category (映射表就绪前的临时数据源 —— 设计 §5)。"""
+
+
+def probe_category(ts_code: str) -> dict:
+    """查 gp_stock_category 并把**过程**带出来。
+
+    存在的理由: 原来的 fetch_sector_from_category 把异常 `except: continue` 吞掉,
+    表不存在 / 列名对不上 / 代理拒绝 一律返回 None, 跟"这只票没有行业"完全无法区分 ——
+    于是行业硬拦截会**静默失效**。调用方要判"数据源到底通不通", 必须看得见 error。
+
+    返回 {value, code_col, industry_col, columns, error}: value=None 且 error=None
+    才是真正的"查到了行, 但没有行业列"或"这只票不在表里"。
+    """
+    out = {"value": None, "code_col": None, "industry_col": None, "columns": None,
+           "error": None}
+    errs = []
     for col in ("stock_code", "ts_code"):
         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})
-        except Exception:
+        except Exception as e:
+            errs.append(f"{col}: {type(e).__name__}: {e}")
             continue
         if r:
-            for k in ("industry", "category", "sector", "industry_name", "sw_industry"):
+            out["code_col"], out["columns"] = col, sorted(r.keys())
+            for k in INDUSTRY_COL_CANDIDATES:
                 if r.get(k):
-                    return str(r[k])
+                    out["value"], out["industry_col"] = str(r[k]), k
-    return None
+                    return out
+            return out                      # 有行但没有任何已知行业列 —— 也是有效信息
+    if errs and out["code_col"] is None:
+        out["error"] = "; ".join(errs[:2])
+    return out
+
+
+def fetch_sector_from_category(ts_code: str):
+    """行业适配器 gp_stock_category (映射表就绪前的临时数据源 —— 设计 §5)。"""
+    return probe_category(ts_code)["value"]

app/services/industry.py

@@ -26,17 +26,82 @@ from app.services import param_store
 logger = logging.getLogger("pms.industry")
 
 _cache = {"at": 0.0, "map": {}, "source": None}
+_probe = {"at": 0.0, "data": None}
 _lock = threading.Lock()
 CACHE_TTL = 300.0
+PROBE_TTL = 300.0
+# 实探样本: 优先拿持仓票 (它们才是真要判行业的), 不够就补几只一定在任何行业表里的大盘股。
+PROBE_FALLBACK = ("600000.SH", "000001.SZ", "600519.SH")
+PROBE_SAMPLE_MAX = 5
 
 
 def source() -> str:
     return (param_store.get("PMS_SECTOR_SOURCE", "") or "").strip()
 
 
+def _probe_category() -> dict:
+    """对 gp_stock_category 做一次小样本实探 (缓存 PROBE_TTL 秒)。
+
+    **为什么非探不可**: 老的 ready() 只看参数值 —— 参数一填就报 True, 页面显示"行业约束
+    已生效", 而那张表完全可能列名对不上/代理拒绝/根本没有这些票, 加上 repo 层把异常吞成
+    None, 结果就是行业硬拦截**静默失效**, 页面上还一片绿。2026-07-31 切到这个源时
+    `/api/industry` 回的正是 `ready=true, count=0` —— 那个 count 只统计 custom_table,
+    对本数据源一个字的证据都没有。这跟本仓库到处在讲的"拿不到 ≠ 通过"是同一件事。
+    """
+    now = time.time()
+    if _probe["data"] is not None and now - _probe["at"] < PROBE_TTL:
+        return _probe["data"]
+    codes = []
+    try:
+        from app.repo import pms_repo
+        codes = [p["ts_code"] for p in pms_repo.list_positions(only_open=True)][:PROBE_SAMPLE_MAX]
+    except Exception as e:
+        logger.warning("行业实探: 取持仓样本失败, 改用兜底样本: %s", e)
+    for c in PROBE_FALLBACK:
+        if len(codes) >= PROBE_SAMPLE_MAX:
+            break
+        if c not in codes:
+            codes.append(c)
+    hit, errors, samples, cols = 0, [], {}, None
+    for c in codes:
+        try:
+            r = downstream_repo.probe_category(c)
+        except Exception as e:
+            errors.append(f"{c}: {type(e).__name__}: {e}")
+            continue
+        if r.get("error"):
+            errors.append(f"{c}: {r['error']}")
+            continue
+        cols = cols or r.get("columns")
+        if r.get("value"):
+            hit += 1
+            samples[c] = r["value"]
+    data = {"tried": len(codes), "hit": hit, "samples": samples, "columns": cols,
+            "errors": errors[:3], "sample_codes": codes}
+    with _lock:
+        _probe.update({"at": now, "data": data})
+    return data
+
+
 def ready() -> bool:
-    """行业约束是否生效 (数据源已配置)。"""
+    """行业约束是否**真的**生效 —— 配置了数据源, 且那个源确实给得出行业。
-    return source() in ("custom_table", "gp_stock_category")
+
+    只看参数值不够: 空的映射表、连不上的行业表, 都会让 get() 一律返回 None, 而 None 在
+    sizer.check_caps 里等于"跳过行业约束"。所以这里必须要有证据。
+    """
+    src = source()
+    if src == "custom_table":
+        try:
+            return len(_load_custom()) > 0
+        except Exception:
+            return False
+    if src == "gp_stock_category":
+        try:
+            return _probe_category()["hit"] > 0
+        except Exception as e:
+            logger.warning("行业实探失败, 按数据源不可用处理: %s", e)
+            return False
+    return False
 
 
 def status() -> dict:
@@ -47,12 +112,29 @@ def status() -> dict:
     if src == "custom_table":
         try:
             st["count"] = len(_load_custom())
-            st["hint"] = f"自定义映射表 pms_industry_map 已加载 {st['count']} 条"
+            st["hint"] = (f"自定义映射表 pms_industry_map 已加载 {st['count']} 条"
+                          if st["count"] else
+                          "映射表 pms_industry_map 是空的 —— 行业约束按未配置处理, 请先灌数")
         except Exception as e:
             st["ready"] = False
             st["hint"] = f"映射表读取失败: {type(e).__name__}: {e}"
     elif src == "gp_stock_category":
-        st["hint"] = "临时数据源 gp_stock_category (逐票查询, 建议尽快切 custom_table)"
+        try:
+            pr = _probe_category()
+        except Exception as e:
+            st.update({"ready": False, "hint": f"行业实探失败: {type(e).__name__}: {e}"})
+            return st
+        st["probe"] = pr
+        st["count"] = pr["hit"]
+        if pr["hit"]:
+            st["hint"] = (f"临时数据源 gp_stock_category 实探 {pr['tried']} 只命中 "
+                          f"{pr['hit']} 只 (列: {pr.get('columns')}), 逐票查询, "
+                          f"建议尽快切 custom_table")
+        else:
+            st["hint"] = ("gp_stock_category **一只都没查到** —— 行业约束等同于未配置且会"
+                          "静默失效。样本 " + ", ".join(pr["sample_codes"][:3])
+                          + ("; 错误: " + " | ".join(pr["errors"]) if pr["errors"]
+                             else "; 无报错, 是表里没有这些票或没有行业列"))
     return st
 
 
@@ -91,3 +173,4 @@ def get_many(codes) -> dict:
 
 def invalidate():
     _cache["at"] = 0.0
+    _probe.update({"at": 0.0, "data": None})

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        装配自检: 服务层→核心→落表 全链路 (内存桩)      (52 例)
+  test_wiring.py        装配自检: 服务层→核心→落表 全链路 (内存桩)      (55 例)
-                                                                  共 238 例
+                                                                  共 241 例
 任一子集失败即整体失败 (退出码 1)。
 """
 import os

scripts/test_wiring.py

@@ -1678,6 +1678,81 @@ def _():
 
 
 # ---------------------------------------------------------------- runner
+# ================================================================ 行业数据源实探
+def _real_industry():
+    """把被 install_fakes 打过桩的 industry 恢复成真实实现 (reload 就地替换属性)。"""
+    import importlib
+    from app.services import industry as ind
+    importlib.reload(ind)
+    ind.invalidate()
+    return ind
+
+
+@case("行业源·gp_stock_category 查得到才算 ready (参数填了不等于数据源能用)")
+def _():
+    from app.repo import downstream_repo
+    fake = install_fakes(params={"PMS_SECTOR_SOURCE": "gp_stock_category"})
+    ind = _real_industry()
+    orig = downstream_repo.probe_category
+    try:
+        downstream_repo.probe_category = lambda c: {
+            "value": "电力设备", "code_col": "stock_code", "industry_col": "industry",
+            "columns": ["stock_code", "industry"], "error": None}
+        assert ind.ready() is True
+        st = ind.status()
+        assert st["ready"] is True and st["count"] == st["probe"]["tried"] > 0, st
+        assert "命中" in st["hint"], st["hint"]
+        assert ind.get("600000.SH") == "电力设备"
+    finally:
+        downstream_repo.probe_category = orig
+    _ = fake
+
+
+@case("行业源·一只都查不到 → ready=False, 约束按未配置停用 (绝不静默失效)")
+def _():
+    from app.repo import downstream_repo
+    install_fakes(params={"PMS_SECTOR_SOURCE": "gp_stock_category"})
+    ind = _real_industry()
+    orig = downstream_repo.probe_category
+    try:
+        # ① 表里没有这些票 (无报错)
+        downstream_repo.probe_category = lambda c: {
+            "value": None, "code_col": None, "industry_col": None, "columns": None,
+            "error": None}
+        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"]
+        # ② 表根本查不了 (有报错) —— 错误必须冒到 hint 上, 不能吞
+        ind.invalidate()
+        downstream_repo.probe_category = lambda c: {
+            "value": None, "code_col": None, "industry_col": None, "columns": None,
+            "error": "stock_code: ProgrammingError: Table doesn't exist"}
+        assert ind.ready() is False
+        assert "Table doesn't exist" in ind.status()["hint"]
+    finally:
+        downstream_repo.probe_category = orig
+
+
+@case("行业源·custom_table 映射表为空同样不算 ready (空表 = 全票 None = 约束失效)")
+def _():
+    fake = install_fakes(params={"PMS_SECTOR_SOURCE": "custom_table"})
+    ind = _real_industry()
+    assert ind.ready() is False
+    assert "是空的" in ind.status()["hint"], ind.status()["hint"]
+    fake.industry = {}
+    ind.invalidate()
+    from app.repo import pms_repo
+    orig = pms_repo.list_industry
+    try:
+        pms_repo.list_industry = lambda limit=5000: [
+            {"ts_code": "600000.SH", "industry": "银行", "updated_at": ""}]
+        assert ind.ready() is True
+        assert ind.get("600000.SH") == "银行"
+    finally:
+        pms_repo.list_industry = orig
+
+
 # ================================================================ 候选池 (上游 /plan)
 def _stub_sp(csvc, mapping):
     """替换 effective_stock_params, 返回 restore 闭包。"""