646 lines
32 KiB
Python
646 lines
32 KiB
Python
"""产业研判与因果论断的覆盖率读数（只读，不写任何库、不落任何文件）。
|
||
|
||
## 为什么要有它
|
||
|
||
《主观量化系统方案_2026-09-03》第四之五之三节把逻辑状态四态定了稿，其中乙路是产业研判。
|
||
接之前必须先量出能对上多少：产业研判只有八个主题，投影表里有一千九百多个环节，而候选卡是按
|
||
环节判的；两边的命名也不在一个体系里（产业研判按股票池主题聚簇，候选卡按环节名）。
|
||
不先量就接，结果是大面积为空。本脚本出四组读数，末尾给一句结论：产业研判这一路今天
|
||
能覆盖多少只票。
|
||
|
||
## 四组读数
|
||
|
||
甲 产业研判覆盖 —— 主题簇、环节簇、概念簇、个股评析各有多少个，各自最新一条的生成日期
|
||
与距今天数，方向取值分布，自我校验未通过的条数；主题簇只有几个，逐条列出。
|
||
乙 命名对齐 —— 产业研判的主题名里有多少个与投影表里的环节名逐字相同，对不上的逐个列出
|
||
并给出包含关系的候选环节名（这决定要不要做主题与环节的对照表）。
|
||
环节簇评析的名字也照同样口径对一遍。
|
||
丙 候选卡侧可达 —— 当日被传导指向的环节里有多少个能对上任意一条产业研判；当日档位表里的票，
|
||
按当日指向的环节算能对上多少只，按投影表全部所属环节算又能对上多少只。
|
||
丁 因果论断覆盖 —— 当日档位表的票里有多少只有因果论断，论断的披露日最新、中位、最老是哪天，
|
||
距数据日超过三十、六十、九十、一百八十、三百六十五天的各占多少。
|
||
|
||
## 纪律
|
||
|
||
全部是 SELECT，没有任何写操作，也不生成文件，只往终端打印。
|
||
每一路读不到都打印一行原因然后继续，任何一组失败都不影响其余组。
|
||
本脚本不接进常规链路，也不注册到 run.py 的子命令里：它是一次性的决策读数，跑法见下。
|
||
|
||
## 跑法【桥机 192.168.16.155 · ~/akg-factor-bridge】
|
||
|
||
docker compose exec -T akg-factor-bridge python judgement_coverage.py
|
||
docker compose exec -T akg-factor-bridge python judgement_coverage.py --date 2026-09-02
|
||
|
||
不传日期时，数据日取平台因子表 t_factor_akg_score 的最新一天（与出计划的口径同源）；
|
||
取不到时退回基座传导视图的最新扫描日。
|
||
|
||
## 已知口径与局限（读数时要一起看）
|
||
|
||
一，投影表每天 06:10 删后插，只存当前态，所以环节名与成员集合都是"今天的"；跑本脚本
|
||
避开 06:10 那个窗口。
|
||
二，桥的只读视图 v_factor_segment_members 暴露的是投影后的环节名 segment_name，而传导扫描
|
||
与因果论断视图连接用的是图上原名 COALESCE(orig_segment_name, segment_name)。改派重排过的
|
||
环节这两个名字会不同。脚本会额外试读一次投影底表拿原名，读不到就只用视图名并打印一行说明。
|
||
三，因果论断视图里客体是环节的论断会按成员展开成多行，所以论断条数一律按 claim_id 去重计。
|
||
四，研判结论表以簇键唯一，重评时整行覆盖，库里只有每个簇的最新一版，没有历史版本，
|
||
所以"距今天数"读的是最新一版的生成日期，不是这个簇第一次生成的日期。
|
||
"""
|
||
from __future__ import annotations # 注解不在定义时求值：开发机的 Python 3.9 也能导入本模块
|
||
|
||
import argparse
|
||
import datetime as dt
|
||
import unicodedata
|
||
|
||
import common
|
||
import db
|
||
|
||
# 论断披露日距数据日的分桶边界（天）。这里只是把分布摊开给人看，不是判定阈值，
|
||
# 四态合成真正要用的陈旧天数在方案第四之五之三节里一次定死，不由本脚本决定。
|
||
_AGE_BUCKETS = (30, 60, 90, 180, 365)
|
||
|
||
|
||
# ---------------------------------------------------------------- 打印与表格
|
||
def say(msg: str = "") -> None:
|
||
print(msg)
|
||
|
||
|
||
def sec(title: str) -> None:
|
||
print(f"\n{'=' * 76}\n{title}\n{'=' * 76}")
|
||
|
||
|
||
def _w(s) -> int:
|
||
"""终端显示宽度：中日韩宽字符按两列算，其余按一列算。"""
|
||
return sum(2 if unicodedata.east_asian_width(ch) in ("W", "F") else 1
|
||
for ch in str(s))
|
||
|
||
|
||
def _pad(s, width: int, right: bool = False) -> str:
|
||
gap = max(0, width - _w(s))
|
||
return (" " * gap + str(s)) if right else (str(s) + " " * gap)
|
||
|
||
|
||
def table(headers, rows, right_cols=()) -> None:
|
||
"""打印一张定宽文字表。rows 是二维序列，None 显示为空。"""
|
||
body = [["" if c is None else str(c) for c in r] for r in rows]
|
||
if not body:
|
||
say("（无行）")
|
||
return
|
||
n = len(headers)
|
||
widths = [max([_w(headers[i])] + [_w(r[i]) for r in body]) for i in range(n)]
|
||
say(" ".join(_pad(headers[i], widths[i], i in right_cols) for i in range(n)))
|
||
say(" ".join("─" * widths[i] for i in range(n)))
|
||
for r in body:
|
||
say(" ".join(_pad(r[i], widths[i], i in right_cols) for i in range(n)))
|
||
|
||
|
||
def _pct(part: int, whole: int) -> str:
|
||
return "—" if not whole else f"{part / whole:.1%}"
|
||
|
||
|
||
# ---------------------------------------------------------------- 只读取数
|
||
def _rows(df) -> list:
|
||
"""把 db.read_* 的返回统一成字典列表；空表返回空列表。"""
|
||
if df is None:
|
||
return []
|
||
if getattr(df, "empty", False):
|
||
return []
|
||
if hasattr(df, "to_dict"):
|
||
return list(df.to_dict("records"))
|
||
return [dict(r) for r in df]
|
||
|
||
|
||
def read_pg(sql: str, params=None, what: str = "") -> list:
|
||
"""只读一条 PG 查询。读不到打印一行原因并返回空列表，不抛出。"""
|
||
try:
|
||
return _rows(db.read_pg(sql, params))
|
||
except Exception as e: # noqa: BLE001 —— 单路失败不拖累其余读数
|
||
say(f" （读不到{what}，本项留空: {e!r}）")
|
||
return []
|
||
|
||
|
||
def read_factor(sql: str, params=None, what: str = "") -> list:
|
||
"""只读一条平台因子库查询。读不到打印一行原因并返回空列表，不抛出。"""
|
||
try:
|
||
return _rows(db.read_mysql("factor", sql, params))
|
||
except Exception as e: # noqa: BLE001
|
||
say(f" （读不到{what}，本项留空: {e!r}）")
|
||
return []
|
||
|
||
|
||
def _ymd(v):
|
||
"""各表的日期列形态不一，统一成 ISO 日期串；认不出返回 None。"""
|
||
if v is None or (isinstance(v, float) and v != v):
|
||
return None
|
||
if isinstance(v, dt.datetime):
|
||
return v.date().isoformat()
|
||
if isinstance(v, dt.date):
|
||
return v.isoformat()
|
||
s = str(v).strip()
|
||
if len(s) == 8 and s.isdigit():
|
||
return f"{s[:4]}-{s[4:6]}-{s[6:]}"
|
||
if len(s) >= 10 and s[4] == "-" and s[7] == "-" and s[:4].isdigit():
|
||
return s[:10]
|
||
return None
|
||
|
||
|
||
def _days_between(d_from, d_to):
|
||
"""两个 ISO 日期串相差多少自然日；任一为空返回 None。"""
|
||
if not d_from or not d_to:
|
||
return None
|
||
try:
|
||
return (dt.date.fromisoformat(d_to) - dt.date.fromisoformat(d_from)).days
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def _s(v):
|
||
if v is None or (isinstance(v, float) and v != v):
|
||
return None
|
||
s = str(v).strip()
|
||
return s or None
|
||
|
||
|
||
def _median(vals: list):
|
||
xs = sorted(v for v in vals if v is not None)
|
||
return xs[len(xs) // 2] if xs else None
|
||
|
||
|
||
def _tri(v):
|
||
"""三态布尔：真、假、说不上（空值）。不能直接拿列值判真假——pandas 读回来的布尔列是
|
||
numpy 的布尔，与 Python 的 True / False 不是同一个对象；空值是 NaN，而 NaN 本身为真。"""
|
||
if v is None or (isinstance(v, float) and v != v):
|
||
return None
|
||
return bool(v)
|
||
|
||
|
||
def pick_date(arg):
|
||
"""数据日：优先命令行，其次平台因子表最新一天，再次基座传导视图最新扫描日。"""
|
||
if arg:
|
||
return arg.strip()
|
||
rows = read_factor("SELECT MAX(trade_date) AS d FROM t_factor_akg_score",
|
||
what="平台因子表 t_factor_akg_score 的最新日期")
|
||
d = _ymd(rows[0]["d"]) if rows else None
|
||
if d:
|
||
return d
|
||
say(" （退回基座传导视图取最新扫描日）")
|
||
rows = read_pg("SELECT MAX(scan_date) AS d FROM v_factor_transmission",
|
||
what="传导视图 v_factor_transmission 的最新扫描日")
|
||
return _ymd(rows[0]["d"]) if rows else None
|
||
|
||
|
||
# ---------------------------------------------------------------- 数据装载
|
||
def load_judgements() -> list:
|
||
"""研判结论视图全表（列子集）。scope 分产业研判、个股评析、环节评析、概念评析。
|
||
|
||
这里不走取数层的 sources.judgement_rows：那个函数是给日频快照用的，只取配置里那两类
|
||
scope，并且会跳过簇键或主题名为空的行。覆盖率读数要的恰恰是"库里到底有什么"，
|
||
少数一行都会让结论偏乐观，所以本脚本自己读全量、不做任何过滤。
|
||
"""
|
||
return read_pg(
|
||
"SELECT scope, subject_name, segment_name, ts_code, leaning, verified, "
|
||
"review_date, n_materials, n_bull, n_bear, input_version, cluster_key "
|
||
"FROM v_factor_judgement",
|
||
what="研判结论视图 v_factor_judgement（视图未建时属正常）")
|
||
|
||
|
||
def load_segment_names() -> tuple:
|
||
"""投影表里的环节名与各自的上市成员数。
|
||
|
||
返回 (视图名到成员数的字典, 图上原名集合)。原名从投影底表试读，只读账号没有底表权限时
|
||
返回空集合并打印一行说明——此时命名对齐只按视图名判，改派重排过的环节可能被少算。
|
||
"""
|
||
rows = read_pg(
|
||
"SELECT segment_name, count(DISTINCT ts_code) AS n_members "
|
||
"FROM v_factor_segment_members "
|
||
"WHERE ts_code IS NOT NULL AND ts_code <> '' GROUP BY segment_name",
|
||
what="环节投影视图 v_factor_segment_members")
|
||
by_name = {}
|
||
for r in rows:
|
||
name = _s(r.get("segment_name"))
|
||
if name:
|
||
by_name[name] = int(r.get("n_members") or 0)
|
||
orig = set()
|
||
try:
|
||
raw = _rows(db.read_pg(
|
||
"SELECT DISTINCT COALESCE(orig_segment_name, segment_name) AS seg "
|
||
"FROM segment_members_projection"))
|
||
orig = {_s(r.get("seg")) for r in raw if _s(r.get("seg"))}
|
||
except Exception as e: # noqa: BLE001 —— 底表没权限是常态，视图名已经够用
|
||
say(f" （读不到投影底表的图上原名，命名对齐只按视图里的环节名判: {e!r}）")
|
||
return by_name, orig
|
||
|
||
|
||
def load_pointed(ds: str) -> tuple:
|
||
"""当日被传导指向的环节，以及每只票挂在哪些被指向环节上。
|
||
|
||
两张视图合起来才是完整的被指向成员：v_factor_transmission 摊平的是未动名单，
|
||
v_factor_transmission_moved 出的是已启动成员。返回 (环节到源数的字典, 票到环节集合的字典)。
|
||
"""
|
||
seg_sources, by_code = {}, {}
|
||
for sql, what in (
|
||
("SELECT ts_code, target, n_sources FROM v_factor_transmission "
|
||
"WHERE scan_date = %s", "传导视图 v_factor_transmission 当日行"),
|
||
("SELECT ts_code, target, n_sources FROM v_factor_transmission_moved "
|
||
"WHERE scan_date = %s", "已动成员视图 v_factor_transmission_moved 当日行")):
|
||
for r in read_pg(sql, (ds,), what=what):
|
||
seg = _s(r.get("target"))
|
||
if not seg:
|
||
continue
|
||
try:
|
||
n = int(r.get("n_sources") or 0)
|
||
except (TypeError, ValueError):
|
||
n = 0
|
||
seg_sources[seg] = max(seg_sources.get(seg, 0), n)
|
||
code = _s(r.get("ts_code"))
|
||
if code:
|
||
by_code.setdefault(common.to_prefix(code), set()).add(seg)
|
||
return seg_sources, by_code
|
||
|
||
|
||
def load_panel(ds: str) -> tuple:
|
||
"""当日档位表：主榜与观察档的票。返回 (主榜集合, 观察档集合, 档位表覆盖只数)。
|
||
|
||
分界与 plan.py 一致：主榜分不低于 150，其余是观察档。档位表覆盖只数单独从
|
||
t_factor_akg_gate 数，与计划里的 gate_covered 同源。
|
||
"""
|
||
main, obs = set(), set()
|
||
for r in read_factor(
|
||
"SELECT stock_code, factor_value FROM t_factor_akg_score WHERE trade_date = %s",
|
||
(ds,), what=f"平台因子表 t_factor_akg_score 的 {ds} 截面"):
|
||
code = _s(r.get("stock_code"))
|
||
if not code:
|
||
continue
|
||
try:
|
||
v = float(r.get("factor_value"))
|
||
except (TypeError, ValueError):
|
||
continue
|
||
(main if v >= 150.0 else obs).add(common.to_prefix(code))
|
||
gate_rows = read_factor(
|
||
"SELECT count(*) AS n FROM t_factor_akg_gate WHERE trade_date = %s",
|
||
(ds,), what=f"平台因子表 t_factor_akg_gate 的 {ds} 截面")
|
||
gate_n = int(gate_rows[0]["n"]) if gate_rows else 0
|
||
return main, obs, gate_n
|
||
|
||
|
||
def load_member_segments(codes: set) -> dict:
|
||
"""投影表里每只票所属的全部环节（不限于当日被指向）。只留给定的票。"""
|
||
out = {}
|
||
for r in read_pg(
|
||
"SELECT segment_name, ts_code FROM v_factor_segment_members "
|
||
"WHERE ts_code IS NOT NULL AND ts_code <> ''",
|
||
what="环节投影视图的成员关系"):
|
||
code = _s(r.get("ts_code"))
|
||
seg = _s(r.get("segment_name"))
|
||
if not code or not seg:
|
||
continue
|
||
k = common.to_prefix(code)
|
||
if k in codes:
|
||
out.setdefault(k, set()).add(seg)
|
||
return out
|
||
|
||
|
||
def load_logic(ds: str) -> list:
|
||
"""因果论断视图按票聚合：论断条数、最新与最老披露日、经环节展开的条数、两类标记。
|
||
|
||
论断条数一律按 claim_id 去重：客体是环节的论断在视图里会按成员展开成多行。
|
||
"""
|
||
return read_pg(
|
||
"SELECT ts_code, "
|
||
"count(DISTINCT claim_id) AS n_claims, "
|
||
"max(disclosure_date) AS d_max, min(disclosure_date) AS d_min, "
|
||
"count(DISTINCT claim_id) FILTER (WHERE link_method = 'segment') AS n_via_segment, "
|
||
"count(DISTINCT claim_id) FILTER (WHERE disputed) AS n_disputed, "
|
||
"count(DISTINCT claim_id) FILTER (WHERE review_flag IS NOT NULL) AS n_flagged "
|
||
"FROM v_factor_logic WHERE disclosure_date <= %s GROUP BY ts_code",
|
||
(ds,), what="因果论断视图 v_factor_logic（视图未建时属正常）")
|
||
|
||
|
||
# ---------------------------------------------------------------- 甲
|
||
_SCOPE_LABEL = {"industry": "产业研判（主题簇）", "segment": "环节评析（环节簇）",
|
||
"concept": "概念评析（概念簇）", "company": "个股评析（公司簇）",
|
||
"other": "其他簇键"}
|
||
|
||
|
||
def part_jia(judg: list, today: str) -> None:
|
||
sec("甲 · 产业研判覆盖：各类簇各有多少、最新一条多久没动")
|
||
if not judg:
|
||
say("研判结论视图没有出行：视图未建、无权限，或表里还没有跑完第二轮的簇。本组读数留空。")
|
||
return
|
||
groups = {}
|
||
for r in judg:
|
||
groups.setdefault(_s(r.get("scope")) or "other", []).append(r)
|
||
rows = []
|
||
for scope in ("industry", "segment", "concept", "company", "other"):
|
||
g = groups.get(scope)
|
||
if not g:
|
||
rows.append([_SCOPE_LABEL[scope], 0, "—", "—", "—", "—"])
|
||
continue
|
||
dates = [d for d in (_ymd(r.get("review_date")) for r in g) if d]
|
||
newest, oldest = (max(dates), min(dates)) if dates else (None, None)
|
||
gap = _days_between(newest, today)
|
||
n_unverified = sum(1 for r in g if _tri(r.get("verified")) is False)
|
||
rows.append([_SCOPE_LABEL[scope], len(g), newest or "—", oldest or "—",
|
||
"—" if gap is None else f"{gap} 天", n_unverified])
|
||
table(["簇的类别", "簇数", "最新生成日", "最老生成日", "最新一条距今", "自我校验未通过"],
|
||
rows, right_cols=(1, 5))
|
||
|
||
lean = {}
|
||
for r in judg:
|
||
scope = _s(r.get("scope")) or "other"
|
||
lean.setdefault(scope, {})
|
||
key = _s(r.get("leaning")) or "（方向为空）"
|
||
lean[scope][key] = lean[scope].get(key, 0) + 1
|
||
say("\n方向取值分布（四态合成的乙路判据用的就是这个受控枚举）：")
|
||
lr = []
|
||
for scope in ("industry", "segment", "concept", "company", "other"):
|
||
if scope in lean:
|
||
for k, v in sorted(lean[scope].items(), key=lambda kv: -kv[1]):
|
||
lr.append([_SCOPE_LABEL[scope], k, v])
|
||
table(["簇的类别", "方向", "簇数"], lr, right_cols=(2,))
|
||
|
||
ind = groups.get("industry") or []
|
||
if ind:
|
||
say(f"\n产业研判逐条清单（共 {len(ind)} 个主题，乙路今天全部的材料就是这几行）：")
|
||
ir = []
|
||
for r in sorted(ind, key=lambda x: _ymd(x.get("review_date")) or "", reverse=True):
|
||
d = _ymd(r.get("review_date"))
|
||
gap = _days_between(d, today)
|
||
ir.append([_s(r.get("subject_name")) or "（无名）",
|
||
_s(r.get("leaning")) or "—",
|
||
r.get("n_bull"), r.get("n_bear"), r.get("n_materials"),
|
||
{True: "是", False: "否", None: "未知"}[_tri(r.get("verified"))],
|
||
d or "—", "—" if gap is None else f"{gap} 天",
|
||
(_s(r.get("input_version")) or "—")[:12]])
|
||
table(["主题名", "方向", "多头条数", "空头条数", "材料条数", "自我校验通过",
|
||
"生成日", "距今", "材料指纹"], ir, right_cols=(2, 3, 4, 7))
|
||
|
||
|
||
# ---------------------------------------------------------------- 乙
|
||
def _substr_hints(name: str, seg_names, limit: int = 3) -> str:
|
||
"""给对不上的名字找包含关系的候选环节名（一方是另一方的子串），最多给几个。"""
|
||
hits = [s for s in seg_names if s != name and (name in s or s in name)]
|
||
hits.sort(key=len)
|
||
if not hits:
|
||
return "（无包含关系的候选）"
|
||
tail = " …" if len(hits) > limit else ""
|
||
return "、".join(hits[:limit]) + tail
|
||
|
||
|
||
def part_yi(judg: list, seg_members: dict, seg_orig: set) -> set:
|
||
"""返回能被逐字对上的环节名集合（主题名命中的加上环节簇名命中的）。"""
|
||
sec("乙 · 命名对齐：产业研判的主题名与投影表里的环节名对不对得上")
|
||
all_seg = set(seg_members) | set(seg_orig)
|
||
if not all_seg:
|
||
say("投影表里一个环节名都没读到，命名对齐无法判定。本组读数留空。")
|
||
return set()
|
||
say(f"投影表里的环节名：视图名 {len(seg_members)} 个"
|
||
+ (f"，加上图上原名后去重 {len(all_seg)} 个。" if seg_orig else "（图上原名未读到）。"))
|
||
if not judg:
|
||
say("研判结论视图没有出行，没有主题名可对。本组读数留空。")
|
||
return set()
|
||
|
||
matched = set()
|
||
for scope, label in (("industry", "主题名"), ("segment", "环节簇名")):
|
||
rows = [r for r in judg if (_s(r.get("scope")) or "") == scope]
|
||
if not rows:
|
||
say(f"\n{label}：一条都没有，跳过。")
|
||
continue
|
||
out, hit = [], 0
|
||
for r in sorted(rows, key=lambda x: _ymd(x.get("review_date")) or "", reverse=True):
|
||
name = _s(r.get("segment_name")) if scope == "segment" else None
|
||
name = name or _s(r.get("subject_name")) or "（无名）"
|
||
ok = name in all_seg
|
||
if ok:
|
||
hit += 1
|
||
matched.add(name)
|
||
out.append([name, "是" if ok else "否",
|
||
seg_members.get(name, "—") if ok else "—",
|
||
_ymd(r.get("review_date")) or "—",
|
||
"" if ok else _substr_hints(name, all_seg)])
|
||
say(f"\n{label} 共 {len(rows)} 个，与环节名逐字相同 {hit} 个"
|
||
f"（{_pct(hit, len(rows))}），对不上 {len(rows) - hit} 个：")
|
||
table([label, "逐字命中环节名", "该环节上市成员数", "生成日", "包含关系的候选环节名"],
|
||
out, right_cols=(2,))
|
||
say("\n读法：逐字命中列全是否，就意味着乙路按名字直连一只票也接不上，"
|
||
"要接必须先做主题与环节的对照表；包含关系那一列是起草对照表时的线索，不是自动匹配的依据。")
|
||
return matched
|
||
|
||
|
||
# ---------------------------------------------------------------- 丙
|
||
def part_bing(judg: list, seg_sources: dict, code_pointed: dict,
|
||
main: set, obs: set, gate_n: int, member_segs: dict) -> tuple:
|
||
"""返回 (能对上研判的票集合, 按投影环节能对上的票集合)。"""
|
||
sec("丙 · 候选卡侧的可达性：当日被指向的环节与档位表的票，有多少能对上产业研判")
|
||
theme_names = {_s(r.get("subject_name")) for r in judg
|
||
if (_s(r.get("scope")) or "") == "industry"}
|
||
seg_cluster_names = {_s(r.get("segment_name")) or _s(r.get("subject_name"))
|
||
for r in judg if (_s(r.get("scope")) or "") == "segment"}
|
||
theme_names.discard(None)
|
||
seg_cluster_names.discard(None)
|
||
reach = theme_names | seg_cluster_names
|
||
say(f"能被对上的名字一共 {len(reach)} 个：同名主题 {len(theme_names)} 个，"
|
||
f"环节簇 {len(seg_cluster_names)} 个。")
|
||
|
||
if not seg_sources:
|
||
say("当日没有读到任何被传导指向的环节，本组前半留空。")
|
||
else:
|
||
by_seg = sum(1 for s in seg_sources if s in seg_cluster_names)
|
||
by_theme = sum(1 for s in seg_sources if s in theme_names)
|
||
both = sum(1 for s in seg_sources if s in reach)
|
||
say(f"\n当日被传导指向的环节 {len(seg_sources)} 个：按环节簇对上 {by_seg} 个，"
|
||
f"按同名主题对上 {by_theme} 个，去重合计 {both} 个（{_pct(both, len(seg_sources))}）。")
|
||
top = sorted(seg_sources.items(), key=lambda kv: (-kv[1], kv[0]))[:20]
|
||
rows = [[seg, n,
|
||
"环节簇" if seg in seg_cluster_names else
|
||
("同名主题" if seg in theme_names else "对不上")]
|
||
for seg, n in top]
|
||
say("\n被指向最强的二十个环节（源数从高到低）：")
|
||
table(["被指向的环节", "源数", "对上哪一路研判"], rows, right_cols=(1,))
|
||
|
||
panel = main | obs
|
||
say(f"\n当日档位表：主榜 {len(main)} 只，观察档 {len(obs)} 只，合计 {len(panel)} 只；"
|
||
f"档位表覆盖 {gate_n} 只。")
|
||
if not panel:
|
||
say("档位表当日没有票，本组后半留空。")
|
||
return set(), set()
|
||
|
||
hit_pointed = {k for k in panel if (code_pointed.get(k) or set()) & reach}
|
||
hit_member = {k for k in panel if (member_segs.get(k) or set()) & reach}
|
||
n_pointed = sum(1 for k in panel if code_pointed.get(k))
|
||
n_member = sum(1 for k in panel if member_segs.get(k))
|
||
rows = [
|
||
["当日被传导指向（卡上的主题就是它）", n_pointed, _pct(n_pointed, len(panel)),
|
||
len(hit_pointed), _pct(len(hit_pointed), len(panel))],
|
||
["投影表里的全部所属环节（不限当日）", n_member, _pct(n_member, len(panel)),
|
||
len(hit_member), _pct(len(hit_member), len(panel))],
|
||
]
|
||
say("\n档位表的票能不能对上研判，按两种口径各算一遍：")
|
||
table(["算所在环节的口径", "有所在环节的票数", "占档位表", "能对上研判的票数", "占档位表"],
|
||
rows, right_cols=(1, 2, 3, 4))
|
||
say("两种口径的差别：上一行是候选卡当天真正用的（卡上的主题来自当日传导指向），"
|
||
"下一行是把票在投影表里挂着的环节全算上，是乙路理论上的天花板。")
|
||
return hit_pointed, hit_member
|
||
|
||
|
||
# ---------------------------------------------------------------- 丁
|
||
def part_ding(logic: list, main: set, obs: set, ds: str) -> set:
|
||
"""返回档位表里有因果论断的票集合。"""
|
||
sec("丁 · 因果论断覆盖：档位表的票有多少只有论断、论断有多旧")
|
||
panel = main | obs
|
||
if not logic:
|
||
say("因果论断视图没有出行：视图未建、无权限，或披露日不晚于数据日的论断为零。本组读数留空。")
|
||
return set()
|
||
by_code = {}
|
||
for r in logic:
|
||
code = _s(r.get("ts_code"))
|
||
if code:
|
||
by_code[common.to_prefix(code)] = r
|
||
say(f"因果论断视图里一共有 {len(by_code)} 只票带论断（披露日不晚于数据日 {ds}）。")
|
||
if not panel:
|
||
say("档位表当日没有票，只能给出全库读数。")
|
||
return set()
|
||
|
||
have = {k for k in panel if k in by_code}
|
||
have_main = {k for k in main if k in by_code}
|
||
rows = [["主榜", len(main), len(have_main), _pct(len(have_main), len(main))],
|
||
["观察档", len(obs), len(have) - len(have_main),
|
||
_pct(len(have) - len(have_main), len(obs))],
|
||
["合计", len(panel), len(have), _pct(len(have), len(panel))]]
|
||
table(["档位", "票数", "有因果论断", "覆盖率"], rows, right_cols=(1, 2, 3))
|
||
if not have:
|
||
say("档位表里一只票都没有论断，披露日分布无从谈起。")
|
||
return set()
|
||
|
||
n_claims, n_via_seg, n_disputed, n_flagged = 0, 0, 0, 0
|
||
per_stock, newest_dates = [], []
|
||
for k in have:
|
||
r = by_code[k]
|
||
try:
|
||
n_claims += int(r.get("n_claims") or 0)
|
||
per_stock.append(int(r.get("n_claims") or 0))
|
||
n_via_seg += int(r.get("n_via_segment") or 0)
|
||
n_disputed += int(r.get("n_disputed") or 0)
|
||
n_flagged += int(r.get("n_flagged") or 0)
|
||
except (TypeError, ValueError):
|
||
pass
|
||
d = _ymd(r.get("d_max"))
|
||
if d:
|
||
newest_dates.append(d)
|
||
say(f"\n这 {len(have)} 只票身上一共 {n_claims} 条论断（按论断编号去重），"
|
||
f"每票中位数 {_median(per_stock)} 条，最多 {max(per_stock) if per_stock else 0} 条；"
|
||
f"其中 {n_via_seg} 条是客体为环节、按成员展开挂上来的，不是直接讲这家公司的。")
|
||
say(f"标记：处于未结的多空分歧复核 {n_disputed} 条；逻辑评析标了疑似误抽 {n_flagged} 条。"
|
||
"这两类在接进四态之前要先决定怎么处置。")
|
||
|
||
if not newest_dates:
|
||
say("论断的披露日全部认不出来，日期分布留空。")
|
||
return have
|
||
ages = [a for a in (_days_between(d, ds) for d in newest_dates) if a is not None]
|
||
say(f"\n每只票最新一条论断的披露日：最新 {max(newest_dates)}，"
|
||
f"中位 {_median(newest_dates)}，最老 {min(newest_dates)}"
|
||
f"（距数据日 {ds} 分别是 {min(ages) if ages else '—'}、{_median(ages)}、"
|
||
f"{max(ages) if ages else '—'} 天）。")
|
||
if ages:
|
||
rows, prev = [], 0
|
||
for b in _AGE_BUCKETS:
|
||
n = sum(1 for a in ages if prev < a <= b)
|
||
rows.append([f"{prev + 1} 到 {b} 天", n, _pct(n, len(ages)),
|
||
sum(1 for a in ages if a > b),
|
||
_pct(sum(1 for a in ages if a > b), len(ages))])
|
||
prev = b
|
||
n_last = sum(1 for a in ages if a > _AGE_BUCKETS[-1])
|
||
rows.append([f"超过 {_AGE_BUCKETS[-1]} 天", n_last, _pct(n_last, len(ages)), 0, "0.0%"])
|
||
say("\n按距数据日的天数分桶（末两列是超过本桶上界的累计，看陈旧面有多大）：")
|
||
table(["距数据日", "票数", "占比", "超过本桶上界的票数", "占比"],
|
||
rows, right_cols=(1, 2, 3, 4))
|
||
return have
|
||
|
||
|
||
# ---------------------------------------------------------------- 结论
|
||
def conclusion(main: set, obs: set, hit_pointed: set, hit_member: set,
|
||
have_logic: set, judg: list, today: str) -> None:
|
||
sec("结论")
|
||
panel = main | obs
|
||
if not panel:
|
||
say("当日档位表没有票，给不出覆盖结论；先确认那一天的因子表已经构建。")
|
||
return
|
||
ind = [r for r in judg if (_s(r.get("scope")) or "") == "industry"]
|
||
dates = [d for d in (_ymd(r.get("review_date")) for r in ind) if d]
|
||
gap = _days_between(max(dates), today) if dates else None
|
||
say(f"产业研判这一路今天只有 {len(ind)} 个主题，最新一条生成于 "
|
||
f"{max(dates) if dates else '未知'}"
|
||
f"{'' if gap is None else f'，距今 {gap} 天'}。")
|
||
say(f"当日档位表 {len(panel)} 只票里，按候选卡当天用的口径（当日被传导指向的环节）"
|
||
f"能对上产业研判的有 {len(hit_pointed)} 只，占 {_pct(len(hit_pointed), len(panel))}；"
|
||
f"把投影表里挂着的环节全算上也只有 {len(hit_member)} 只，"
|
||
f"占 {_pct(len(hit_member), len(panel))}——这是乙路的天花板。")
|
||
diff = len(have_logic) - len(hit_pointed)
|
||
say(f"同一批票里有因果论断的是 {len(have_logic)} 只，占 {_pct(len(have_logic), len(panel))}；"
|
||
+ (f"甲路比乙路多覆盖 {diff} 只。" if diff > 0 else
|
||
(f"乙路比甲路多覆盖 {-diff} 只。" if diff < 0 else "两路覆盖的只数一样多。")))
|
||
if len(hit_pointed) == 0:
|
||
say("\n判断：乙路现在按名字直连一只票都接不上，接进四态合成等于全票都落到"
|
||
"无法判断加证据不足。要么先做主题与环节的对照表再接，要么这一轮先只接甲路，"
|
||
"乙路等主题数上来、对照表建好再说。")
|
||
elif len(hit_pointed) * 10 < len(panel):
|
||
say("\n判断：乙路能覆盖的不到档位表的一成，接进去的话绝大多数票仍然是证据不足。"
|
||
"可以接，但要在卡上明写缺的是哪一路，不要让读的人误以为是系统判过了。")
|
||
else:
|
||
say("\n判断：乙路的覆盖面够得上接。仍要在卡上分开标注缺失与矛盾两个子因，"
|
||
"缺失不当成负面证据。")
|
||
say("\n本脚本只读不写，跑完不留任何文件。以上读数随投影表每天 06:10 重建而变，"
|
||
"拍板前建议连着两个交易日各跑一次看是否稳定。")
|
||
|
||
|
||
# ---------------------------------------------------------------- 入口
|
||
def run(date=None) -> int:
|
||
today = dt.date.today().isoformat()
|
||
say("产业研判与因果论断覆盖率读数（只读，不写任何库、不落任何文件）")
|
||
ds = pick_date(date)
|
||
if not ds:
|
||
say("取不到数据日：平台因子表与基座传导视图都读不到。先确认连接与当日构建，再跑本脚本。")
|
||
return 1
|
||
say(f"数据日 {ds}；脚本运行日 {today}（容器走国际时间，与北京时间可能差一天，"
|
||
f"两个日期都打出来是为了让距今天数有据可查）。")
|
||
|
||
judg = load_judgements()
|
||
seg_members, seg_orig = load_segment_names()
|
||
seg_sources, code_pointed = load_pointed(ds)
|
||
main, obs, gate_n = load_panel(ds)
|
||
member_segs = load_member_segments(main | obs) if (main or obs) else {}
|
||
logic = load_logic(ds)
|
||
|
||
hit_pointed, hit_member, have_logic = set(), set(), set()
|
||
for name, fn in (
|
||
("甲", lambda: part_jia(judg, today)),
|
||
("乙", lambda: part_yi(judg, seg_members, seg_orig)),
|
||
("丙", lambda: part_bing(judg, seg_sources, code_pointed, main, obs,
|
||
gate_n, member_segs)),
|
||
("丁", lambda: part_ding(logic, main, obs, ds))):
|
||
try:
|
||
r = fn()
|
||
if name == "丙" and isinstance(r, tuple):
|
||
hit_pointed, hit_member = r
|
||
elif name == "丁" and isinstance(r, set):
|
||
have_logic = r
|
||
except Exception as e: # noqa: BLE001 —— 单组失败不拖累其余读数
|
||
say(f"\n第{name}组读数失败（其余照出）: {e!r}")
|
||
try:
|
||
conclusion(main, obs, hit_pointed, hit_member, have_logic, judg, today)
|
||
except Exception as e: # noqa: BLE001
|
||
say(f"\n结论一节失败: {e!r}")
|
||
return 0
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(
|
||
description="产业研判与因果论断的覆盖率读数（只读）")
|
||
ap.add_argument("--date", default=None,
|
||
help="数据日 YYYY-MM-DD，不传时取平台因子表最新一天")
|
||
a = ap.parse_args()
|
||
return run(a.date)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|