From ff97024b3a34af79c76160a7af6e762a3d84c0e5 Mon Sep 17 00:00:00 2001 From: zlt Date: Fri, 31 Jul 2026 11:11:07 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A1=8C=E4=B8=9A=E6=BA=90=E6=8D=A2=20gp=5Fhyb?= =?UTF-8?q?k(199,=E4=B8=89=E7=BA=A7884,=E5=8D=95=E4=B8=BB=E8=A1=8C?= =?UTF-8?q?=E4=B8=9A);=20gp=5Fstock=5Fcategory=20=E4=BD=9C=E5=BA=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 6 +- UPSTREAM_PLAN_API.md | 70 ++++++++++++++++++ app/repo/downstream_repo.py | 83 +++++++++++++++++----- app/repo/industry_repo.py | 137 ++++++++++++++++++++++++++++++++++++ app/services/industry.py | 109 +++++++++++++++++++++++----- app/services/param_store.py | 10 ++- app/web/main.py | 7 +- config/settings.py | 5 +- scripts/run_tests.py | 4 +- scripts/test_wiring.py | 131 +++++++++++++++++++++++++++++++++- 10 files changed, 514 insertions(+), 48 deletions(-) create mode 100644 app/repo/industry_repo.py diff --git a/README.md b/README.md index 8d32850..572b925 100644 --- a/README.md +++ b/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` 计划不幂等、档位规模与文档不符） | diff --git a/UPSTREAM_PLAN_API.md b/UPSTREAM_PLAN_API.md index f54eb16..ebd31ce 100644 --- a/UPSTREAM_PLAN_API.md +++ b/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` 有没有真填上。 diff --git a/app/repo/downstream_repo.py b/app/repo/downstream_repo.py index ef00e55..58a9f21 100644 --- a/app/repo/downstream_repo.py +++ b/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, - "error": None} - errs = [] - for col in ("stock_code", "ts_code"): + meta = category_columns() + out = {"value": None, "code_col": meta.get("code_col"), + "industry_col": meta.get("industry_col"), "columns": meta.get("columns"), + "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}") - continue + out["error"] = f"{col}={form}: {type(e).__name__}: {str(e)[:160]}" + return out if r: - out["code_col"], out["columns"] = col, sorted(r.keys()) - for k in INDUSTRY_COL_CANDIDATES: - if r.get(k): - out["value"], out["industry_col"] = str(r[k]), k - return out - return out # 有行但没有任何已知行业列 —— 也是有效信息 - if errs and out["code_col"] is None: - out["error"] = "; ".join(errs[:2]) + out["matched_form"] = form + ind = out["industry_col"] + if ind and r.get(ind): + out["value"] = str(r[ind]) + else: + for k in INDUSTRY_COL_CANDIDATES: + if r.get(k): + out["value"], out["industry_col"] = str(r[k]), k + break + return out return out diff --git a/app/repo/industry_repo.py b/app/repo/industry_repo.py new file mode 100644 index 0000000..678be33 --- /dev/null +++ b/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} diff --git a/app/services/industry.py b/app/services/industry.py index 060ec5d..7a4656a 100644 --- a/app/services/industry.py +++ b/app/services/industry.py @@ -5,8 +5,11 @@ IndustryClassifier.get(ts_code) -> 行业名 | None 数据源经参数 PMS_SECTOR_SOURCE 切换: - custom_table 自定义映射表 pms_industry_map (默认建议; 用户后续灌何种划分都只是灌数) - gp_stock_category 决策系统生态已有的行业表 (映射表就绪前的临时数据源) + gp_hybk **当前口径** —— 199 库的行业板块表, 取三级 (bk_code 884*), 每票只认 + 一个主行业 (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"] - else "; 无报错, 是表里没有这些票或没有行业列")) + + ("; 原因: " + " | ".join(pr["errors"][:2]) if pr["errors"] + 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 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) + if not ts_code: 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": - return {c: get(c) for c in (codes or [])} - m = _load_custom() - return {c: m.get(c) for c in (codes or [])} + codes = [c for c in (codes or []) if c] + if src == "gp_hybk": + return _hybk_many(codes) + 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 diff --git a/app/services/param_store.py b/app/services/param_store.py index 017dadb..0be0dea 100644 --- a/app/services/param_store.py +++ b/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"): - return "PMS_SECTOR_SOURCE 只能是 空 / custom_table / gp_stock_category" + if key == "PMS_SECTOR_SOURCE" and v not in ("", "gp_hybk", "custom_table", + "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"): diff --git a/app/web/main.py b/app/web/main.py index f2cacb2..12d6bce 100644 --- a/app/web/main.py +++ b/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) diff --git a/config/settings.py b/config/settings.py index 1d17ebe..471d340 100644 --- a/config/settings.py +++ b/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 # 占总仓 diff --git a/scripts/run_tests.py b/scripts/run_tests.py index d2a1d77..9be510f 100644 --- a/scripts/run_tests.py +++ b/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 例) - 共 241 例 + test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) (58 例) + 共 244 例 任一子集失败即整体失败 (退出码 1)。 """ import os diff --git a/scripts/test_wiring.py b/scripts/test_wiring.py index 273435e..71038c4 100644 --- a/scripts/test_wiring.py +++ b/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"})