akg-factor-bridge/judgement_coverage.py

"""产业研判与因果论断的覆盖率读数（只读，不写任何库、不落任何文件）。

## 为什么要有它

《主观量化系统方案_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())