From 3531937db01ac37e6d3900a8c3209851e2146e09 Mon Sep 17 00:00:00 2001 From: zlt Date: Fri, 31 Jul 2026 10:13:56 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A1=8C=E4=B8=9A=E6=BA=90ready=E6=94=B9?= =?UTF-8?q?=E5=AE=9E=E6=8E=A2(=E5=8F=82=E6=95=B0=E5=A1=AB=E4=BA=86?= =?UTF-8?q?=E4=B8=8D=E7=AE=97=E6=95=B0);=20probe=5Fcategory=E8=AE=A9?= =?UTF-8?q?=E5=BC=82=E5=B8=B8=E5=8F=AF=E8=A7=81;=20=E6=94=B6=E5=B0=BE?= =?UTF-8?q?=E6=96=87=E6=A1=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 6 +-- UPSTREAM_PLAN_API.md | 43 ++++++++++++++++-- app/repo/downstream_repo.py | 37 ++++++++++++--- app/services/industry.py | 91 +++++++++++++++++++++++++++++++++++-- scripts/run_tests.py | 4 +- scripts/test_wiring.py | 75 ++++++++++++++++++++++++++++++ 6 files changed, 237 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index 91016e2..8d32850 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 装配自检: 服务层→核心→落表 全链路 (内存桩) 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` 计划不幂等、档位规模与文档不符） | diff --git a/UPSTREAM_PLAN_API.md b/UPSTREAM_PLAN_API.md index af72383..f54eb16 100644 --- a/UPSTREAM_PLAN_API.md +++ b/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` 之类的 -版本戳, 下游才能判断"手上这份是哪一版"。② 券商预期是否有重载窗口, 那期间的请求要不要 -挡掉 (返回 409/503 比返回半截数据安全)。 +**已与上游对齐 (2026-07-31)**: 上游当天刚做过调整, 这个前后不一致是**预期内的**, 不是 +半截快照。 + +仍值得提一句的改进 (非阻塞): 给应答加个 `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。 diff --git a/app/repo/downstream_repo.py b/app/repo/downstream_repo.py index d7136d0..ef00e55 100644 --- a/app/repo/downstream_repo.py +++ b/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): - """行业适配器 gp_stock_category (映射表就绪前的临时数据源 —— 设计 §5)。""" +INDUSTRY_COL_CANDIDATES = ("industry", "category", "sector", "industry_name", "sw_industry") + + +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]) - return None + 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]) + return out + + +def fetch_sector_from_category(ts_code: str): + """行业适配器 gp_stock_category (映射表就绪前的临时数据源 —— 设计 §5)。""" + return probe_category(ts_code)["value"] diff --git a/app/services/industry.py b/app/services/industry.py index 0c16f67..060ec5d 100644 --- a/app/services/industry.py +++ b/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}) diff --git a/scripts/run_tests.py b/scripts/run_tests.py index 31c82a9..d2a1d77 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 装配自检: 服务层→核心→落表 全链路 (内存桩) (52 例) - 共 238 例 + test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) (55 例) + 共 241 例 任一子集失败即整体失败 (退出码 1)。 """ import os diff --git a/scripts/test_wiring.py b/scripts/test_wiring.py index 4a193ea..273435e 100644 --- a/scripts/test_wiring.py +++ b/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 闭包。"""