251 lines
10 KiB
Python
251 lines
10 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
行业划分接口 (设计 §5 的可插拔适配器)
|
|
======================================
|
|
IndustryClassifier.get(ts_code) -> 行业名 | None
|
|
|
|
数据源经参数 PMS_SECTOR_SOURCE 切换:
|
|
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`, 由
|
|
`plan_feed` 每次刷新时 upsert 进 pms_industry_map (见 UPSTREAM_PLAN_API.md §2)。
|
|
所以本模块**不需要**新增 plan_api 数据源 —— 即时查当日计划有个洞: 今天没上榜的持仓票
|
|
查不到 theme, 它的行业硬拦截就悄悄失效了。落库以后覆盖面随每天刷新累积, 且页面可见可手改。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import threading
|
|
import time
|
|
|
|
from app.repo import downstream_repo, industry_repo, pms_repo
|
|
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
|
|
|
|
|
|
VALID_SOURCES = ("gp_hybk", "custom_table", "gp_stock_category")
|
|
|
|
|
|
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"):
|
|
e = str(r["error"])[:200]
|
|
if e not in errors: # 同一个列名错误会对每只样本各报一遍, 去重
|
|
errors.append(e)
|
|
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 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:
|
|
"""行业约束是否**真的**生效 —— 配置了数据源, 且那个源确实给得出行业。
|
|
|
|
只看参数值不够: 空的映射表、连不上的行业表, 都会让 get() 一律返回 None, 而 None 在
|
|
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
|
|
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:
|
|
src = source()
|
|
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["cached_today"] = sum(1 for v in _hybk["data"].values() if v)
|
|
st["count"] = st["cached_today"] # count 沿用旧字段名, 含义见 hint
|
|
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 最小的那个主行业); "
|
|
f"本日已解析 {st['cached_today']} 只 —— 这是**按需查询的日缓存计数**, "
|
|
f"0 只表示今天还没查过任何票 (账本空/没下过命令), 不是查不到")
|
|
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())
|
|
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":
|
|
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"][:2]) if pr["errors"]
|
|
else "; 无报错, 是表里没有这些票 (代码写法已试过点式/前缀式/纯数字)")
|
|
+ "。用 运维 → 导出下游表结构 看 _category_probe 的实际列名")
|
|
return st
|
|
|
|
|
|
def _load_custom() -> dict:
|
|
now = time.time()
|
|
if _cache["source"] == "custom_table" and now - _cache["at"] < CACHE_TTL:
|
|
return _cache["map"]
|
|
with _lock:
|
|
rows = pms_repo.list_industry()
|
|
m = {r["ts_code"]: r["industry"] for r in rows}
|
|
_cache.update({"at": now, "map": m, "source": "custom_table"})
|
|
return m
|
|
|
|
|
|
def get(ts_code: str):
|
|
"""返回行业名; 未配置数据源或查不到 → None (调用方据此跳过行业约束)。"""
|
|
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()
|
|
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
|