"""产业研判与因果论断的覆盖率读数（只读，不写任何库、不落任何文件）。 ## 为什么要有它 《主观量化系统方案_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())