akg-factor-bridge/sources.py

"""候选卡的取数层：把各条证据线从三处库读成"按前缀码索引的字典"（全部只读）。

## 为什么要有它

候选卡的规则在 card.py，是纯函数；证据从哪来、怎么对齐代码格式、缺了怎么办，
全部收在这里，plan.py 只做装配。三处来源：

  基座 PG   v_factor_transmission_moved   已启动成员及其所在环节的传导证据（只认 Segment 目标）
            v_factor_stock_daily          数据日涨幅、主力净额异常值、热度变化
  153 代理  strategy_daily_results        决策系统昨夜结论：信号、支撑压力位、吸筹块
            （吸筹的评分、状态、评分日三项一次读齐；基座落库的吸筹版本没有评分日，所以不从基座取）
  平台 MySQL gp_day_data                  交易日历（算评分日龄用；取一只长期存在的票的日期序列）
  基座 PG   v_factor_logic                数据基座抽取的因果论断（方向、机制、时效、出处），每票最多几条，
                                          只展示不作门槛（2026-09-03 方案第 3.3 节）
  平台 MySQL zs_day_data / eastmoney_rzrq_data / fear_greed_index
                                          计划环境段的市场四项：两市成交额、融资余额、恐贪指数；
                                          市场广度从基座 v_factor_stock_daily 当日行自算（同一节）
  基座 PG   v_factor_judgement            产业研判与环节评析的最新一版结论（采信倾向、多空条数、
                                          自我校验、材料指纹），每个计划日抄一份存版本史（judgement.py）

代码格式：基座是点后缀式 600000.SH，决策系统与桥是前缀式 SH600000，进出都过 common.to_prefix。
读失败的语义：每一路读不到都返回空字典并打印一行原因，候选卡按"缺失"处理（进关注或仅展示），
不让计划断产——与 pool.py 的安全边界一致。

离线单测：logic_claims 与 market_context 都接受注入的读函数（read_pg / read_mysql），
test_market_context.py 用假数据函数替换真实连接，不连库；两者内部只对"记录列表"做计算，
读函数返回 DataFrame 或普通的字典列表都可以（见 _records）。
"""
from __future__ import annotations

import datetime as dt
import json
import statistics
from collections import defaultdict

import pandas as pd

import common
import config
import db

# 上证指数与深证成指在指数日线表 zs_day_data 里的代码，两市成交额取二者当日 amount 之和。
MARKET_INDEX_CODES = ("000001.SH", "399001.SZ")
# 市场广度里"涨停家数"的近似口径：涨幅达到 9.8%（不读涨跌停价表，与环节日行情视图注释一致）。
LIMIT_UP_PCT = 9.8
# 融资表与恐贪指数表的日期列名不在本仓库内核实过，按候选名逐个匹配（第一个命中的用）。
_DATE_COL_CANDIDATES = ("trade_date", "date", "stat_date", "data_date", "report_date",
                        "dt", "timestamp", "update_date", "record_date", "created_at")


def moved_members(ds: str) -> dict[str, dict]:
    """数据日 ds 被传导指向的环节里，已启动（不在未动名单）的成员。
    同一票挂在多个被指向环节上时，取源数最多、其次链符最高的那条作卡上的证据。"""
    try:
        df = db.read_pg(
            "SELECT ts_code, target, n_sources, chain_fit, members_total, moved, "
            "moved_ratio, mkt_trade_date FROM v_factor_transmission_moved "
            "WHERE scan_date = %s", (ds,))
    except Exception as e:  # noqa: BLE001
        print(f"  （已动成员视图读取失败，候选卡的传导门槛整体缺席: {e!r}）")
        return {}
    if df.empty:
        return {}
    df["k"] = df["ts_code"].map(lambda s: common.to_prefix(str(s).strip()))
    df["n_sources"] = pd.to_numeric(df["n_sources"], errors="coerce").fillna(0)
    df["chain_fit"] = pd.to_numeric(df["chain_fit"], errors="coerce").fillna(0)
    df = df.sort_values(["n_sources", "chain_fit"], ascending=False).drop_duplicates("k")
    out = {}
    for r in df.itertuples():
        out[r.k] = {"theme": str(r.target), "n_sources": int(r.n_sources),
                    "chain_fit": float(r.chain_fit),
                    "members_total": None if pd.isna(r.members_total) else int(r.members_total),
                    "moved": None if pd.isna(r.moved) else int(r.moved),
                    "mkt_trade_date": None if pd.isna(r.mkt_trade_date) else str(r.mkt_trade_date)}
    return out


def stock_daily(ds: str) -> dict[str, dict]:
    """数据日 ds 的个股行情三列：涨幅（百分数）、主力净额异常值、热度五日变化。"""
    try:
        df = db.read_pg(
            "SELECT ts_code, pct_change, net_z, heat_chg FROM v_factor_stock_daily "
            "WHERE trade_date = %s", (ds,))
    except Exception as e:  # noqa: BLE001
        print(f"  （个股日行情视图读取失败，候选卡的已启动门槛整体缺席: {e!r}）")
        return {}
    out = {}
    for r in df.itertuples():
        k = common.to_prefix(str(r.ts_code).strip())
        out[k] = {"pct0": _f(r.pct_change), "net_z": _f(r.net_z), "heat_chg": _f(r.heat_chg)}
    return out


def trading_days(upto: str, back_days: int = 90) -> list[str]:
    """交易日历：取一只长期存在的票在 gp_day_data 里的日期序列（表有一千四百万行，
    全表 DISTINCT 太慢；按 symbol 走索引）。返回升序 ISO 日期串。"""
    start = (dt.date.fromisoformat(upto) - dt.timedelta(days=back_days)).isoformat()
    try:
        df = db.read_mysql(
            "factor", "SELECT DISTINCT DATE(`timestamp`) d FROM gp_day_data "
                      "WHERE symbol = %s AND `timestamp` >= %s AND `timestamp` <= %s "
                      "ORDER BY d", ("SH600519", start, upto))
        return [pd.Timestamp(x).date().isoformat() for x in df["d"].tolist()]
    except Exception as e:  # noqa: BLE001
        print(f"  （交易日历读取失败，评分日龄按自然日×5/7 近似: {e!r}）")
        return []


def night_conclusions(codes, ds: str) -> dict[str, dict]:
    """决策系统昨夜结论（每票最新一行）：信号、支撑压力位、吸筹块与评分日龄。

    评分日龄 = 结论行的 trade_date 到数据日 ds 之间的交易日数（含头不含尾）。
    该表会被盘中补扫就地改写、没有落库时刻列，日龄以行的 trade_date 为准，
    这是已知局限（方案 2.2 第三项）。"""
    codes = sorted({common.to_prefix(str(c).strip()) for c in codes if c})
    if not codes:
        return {}
    # 只取不晚于数据日 ds 的结论行：生产上 ds 就是数据日，与"取最新一行"等价；
    # 复盘重建历史日时这一条防前视（否则历史日会读到今天的吸筹状态）。
    # trade_date 是整数 YYYYMMDD（数据源盘点 §1a），直接比大小。
    ds_int = int(ds.replace("-", "")) if ds else 99999999
    try:
        marks = ",".join(["%s"] * len(codes))
        df = db.read_mysql(
            "pms",
            f"SELECT stock_code, trade_date, signal_type, support_level, pressure_level, "
            f"raw_logic_json FROM strategy_daily_results "
            f"WHERE stock_code IN ({marks}) AND trade_date <= %s",
            tuple(codes) + (ds_int,))
    except Exception as e:  # noqa: BLE001
        print(f"  （决策系统结论表读取失败，吸筹确认线与坏信号风险整体缺席: {e!r}）")
        return {}
    if df.empty:
        return {}
    df = df.sort_values("trade_date").drop_duplicates("stock_code", keep="last")
    cal = trading_days(ds)
    cal_index = {d: i for i, d in enumerate(cal)}
    out = {}
    for r in df.itertuples():
        k = common.to_prefix(str(r.stock_code).strip())
        tdate = _ymd(r.trade_date)
        age = _age(tdate, ds, cal_index)
        ff = {}
        try:
            raw = json.loads(r.raw_logic_json) if isinstance(r.raw_logic_json, str) else (r.raw_logic_json or {})
            ff = (raw or {}).get("fund_flow") or {}
        except Exception:  # noqa: BLE001 —— 坏 JSON 当无吸筹块
            ff = {}
        out[k] = {
            "signal": str(r.signal_type or "").strip().upper() or None,
            "support": _f(r.support_level), "pressure": _f(r.pressure_level),
            "conclusion_date": tdate,
            "accum_state": str(ff.get("state") or "") or None,
            "accum_score": _f(ff.get("score")),
            "accum_structure": ff.get("structure"), "accum_pos_tag": ff.get("pos_tag"),
            "accum_age": age,
        }
    return out


def _ymd(v) -> str | None:
    """各表的日期列形态不一（整数 YYYYMMDD、date、datetime、ISO 串），统一成 ISO 日期串。
    前几种形态不经 pandas 就能认出来，这样离线单测里的假数据不依赖 pandas；认不出的最后才交给
    pandas 解析，仍失败返回 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]
    try:
        return pd.Timestamp(s).date().isoformat()
    except Exception:  # noqa: BLE001
        return None


def _age(tdate: str | None, ds: str, cal_index: dict) -> int | None:
    if not tdate:
        return None
    if tdate in cal_index and ds in cal_index:
        return cal_index[ds] - cal_index[tdate]
    try:  # 日历缺失或日期在日历之外：自然日 × 5/7 近似
        nat = (dt.date.fromisoformat(ds) - dt.date.fromisoformat(tdate)).days
        return max(0, round(nat * 5 / 7))
    except ValueError:
        return None


def _f(v):
    try:
        x = float(v)
    except (TypeError, ValueError):
        return None
    return None if x != x else x


def _records(df) -> list[dict]:
    """把读函数的返回统一成字典列表：DataFrame 走 to_dict，普通列表原样返回，None 与空表返回空列表。
    这样取数函数的计算部分只面对普通 Python 对象，离线单测的假读函数直接返回字典列表即可。"""
    if df is None:
        return []
    if isinstance(df, list):
        return [dict(r) for r in df]
    if hasattr(df, "to_dict"):
        if getattr(df, "empty", False):
            return []
        return list(df.to_dict("records"))
    return list(df)


def _to_dot(code: str) -> str:
    """前缀式 SH600000 转成数据基座的点后缀式 600000.SH；已是点后缀式或纯数字则原样返回。"""
    s = str(code or "").strip().upper()
    if "." in s or len(s) < 3:
        return s
    if s[:2] in ("SH", "SZ", "BJ") and s[2:].isdigit():
        return f"{s[2:]}.{s[:2]}"
    return s


# ============================================================================
# 因果论断（数据基座 v_factor_logic，2026-09-03 方案第 3.3 节"候选卡读因果论断"）
# ============================================================================
_LOGIC_COLS = ("ts_code", "subject_name", "object_name", "direction", "mechanism", "condition",
               "horizon", "strength", "tier", "confidence", "disclosure_date", "doc_id",
               "doc_title", "source_span", "claim_id", "via_segment",
               # 质量两列（2026-09-03 起取）：disputed 是这条论断还在多空分歧里没结论，
               # review_flag 是逻辑评析读过之后标的"疑似误抽"。两者都只作提示，不过滤——
               # 那是模型的判断不是确定性事实，过滤会让人看不到系统曾经抽到过什么。
               "disputed", "review_flag")


def logic_claims(codes, ds: str, per_stock: int | None = None, read_pg=None) -> dict[str, list[dict]]:
    """这批票在数据基座因果论断视图里、披露日不晚于数据日 ds 的论断，每票取最近披露日的最多
    per_stock 条（默认 config.LOGIC_CLAIMS_PER_STOCK），按前缀码索引。

    每条论断带：方向、机制、条件、时效、强度、层级、置信度、披露日、出处文档标题与编号、
    论断编号、经由环节、主体与客体名。只作展示与出处，不进判决；读失败返回空字典并打印一行原因。
    read_pg 可注入（离线单测），默认走 db.read_pg。"""
    n_per = config.LOGIC_CLAIMS_PER_STOCK if per_stock is None else int(per_stock)
    if n_per <= 0:
        return {}
    wanted = sorted({common.to_prefix(str(c).strip()) for c in codes if c})
    if not wanted:
        return {}
    dots = [_to_dot(c) for c in wanted]
    reader = read_pg or db.read_pg
    try:
        marks = ",".join(["%s"] * len(dots))
        rows = _records(reader(
            f"SELECT {', '.join(_LOGIC_COLS)} FROM v_factor_logic "
            f"WHERE ts_code IN ({marks}) AND disclosure_date <= %s",
            tuple(dots) + (ds,)))
    except Exception as e:  # noqa: BLE001
        print(f"  （因果论断视图 v_factor_logic 读取失败，候选卡的论断证据线整体缺席: {e!r}）")
        return {}
    if not rows:
        return {}
    want = set(wanted)
    by_code: dict[str, list[dict]] = {}
    seen: set[tuple[str, str]] = set()
    for r in rows:
        k = common.to_prefix(str(r.get("ts_code") or "").strip())
        if k not in want:                 # 只留请求的票，视图返回的多余行不带进卡
            continue
        cid = _s(r.get("claim_id"))
        if cid and (k, cid) in seen:      # 视图里客体为环节的论断按成员展开，同票同论断只留一条
            continue
        if cid:
            seen.add((k, cid))
        by_code.setdefault(k, []).append({
            "direction": _s(r.get("direction")), "mechanism": _s(r.get("mechanism")),
            "condition": _s(r.get("condition")), "horizon": _s(r.get("horizon")),
            "strength": _s(r.get("strength")), "tier": _s(r.get("tier")),
            "confidence": _f(r.get("confidence")),
            "disclosure_date": _ymd(r.get("disclosure_date")),
            "doc_id": _s(r.get("doc_id")), "doc_title": _s(r.get("doc_title")),
            "source_span": (_s(r.get("source_span")) or "")[:200] or None,
            "claim_id": _s(r.get("claim_id")), "via_segment": _s(r.get("via_segment")),
            "subject": _s(r.get("subject_name")), "object": _s(r.get("object_name")),
            "disputed": bool(r.get("disputed")), "review_flag": _s(r.get("review_flag")),
        })
    out = {}
    for k, items in by_code.items():
        items.sort(key=lambda c: (c["disclosure_date"] or "", c["confidence"] or -1.0), reverse=True)
        # 质量指标必须在截断之前算：截断后只剩最近几条，"这票一共有几份研报来源"就看不出来了。
        # 算完挂在每条论断上（每条都带同一份，卡上取第一条即可），不另开一个返回值，
        # 免得上层要同时接两个字典、少接一个就静默丢掉质量提示。
        quality = _logic_quality(items, ds)
        kept = items[:n_per]
        for c in kept:
            c["quality"] = quality
        out[k] = kept
    return out


def _logic_quality(items: list, ds: str) -> dict:
    """一票全部论断的质量画像，在截断之前算。四项：

      age_days     最新一条论断距数据日多少天。同一张卡上吸筹评分超过三十个交易日就判陈旧，
                   而论断此前永不过期，这是双标。这里给出天数，阈值判定在 card.py。
      n_docs       论断来自几份不同研报。实测近三成带多条论断的票，全部论断出自同一份研报——
                   那份研报一旦过时或本身有偏，这票的整条研究证据一起失效，而卡上看不出来。
      n_flagged    被逻辑评析标为疑似误抽的条数；n_disputed 还在多空分歧里没结论的条数。
      n_via_segment  客体是环节、按成员展开挂上来的条数。那类不是直接讲这家公司，
                   与直接讲这家公司的论断分量不同，卡上要分得开。
    """
    days = None
    dates = [c["disclosure_date"] for c in items if c.get("disclosure_date")]
    if dates and ds:
        try:
            days = (dt.date.fromisoformat(ds) - dt.date.fromisoformat(max(dates))).days
        except ValueError:
            days = None
    return {
        "n": len(items),
        "age_days": days,
        "latest_date": max(dates) if dates else None,
        "n_docs": len({c["doc_id"] for c in items if c.get("doc_id")}),
        "n_flagged": sum(1 for c in items if c.get("review_flag")),
        "n_disputed": sum(1 for c in items if c.get("disputed")),
        "n_via_segment": sum(1 for c in items if c.get("via_segment")),
    }


def _s(v) -> str | None:
    if v is None or (isinstance(v, float) and v != v):
        return None
    s = str(v).strip()
    return s or None


def _i(v) -> int | None:
    x = _f(v)
    return None if x is None else int(x)


def _b(v) -> bool | None:
    """自我校验结果这类三值列：真、假、还没有结果。空值保持为空，不当成假。"""
    if v is None or (isinstance(v, float) and v != v):
        return None
    if isinstance(v, bool):
        return v
    s = str(v).strip().lower()
    if s in ("true", "t", "1", "yes", "y"):
        return True
    if s in ("false", "f", "0", "no", "n"):
        return False
    return None


def _n_items(v) -> int | None:
    """JSON 数组列只留条数：列表直接数，字符串先按 JSON 解析，认不出返回空。"""
    if v is None or (isinstance(v, float) and v != v):
        return None
    if isinstance(v, (list, tuple)):
        return len(v)
    if isinstance(v, str):
        try:
            parsed = json.loads(v)
        except ValueError:
            return None
        return len(parsed) if isinstance(parsed, list) else None
    return None


# ============================================================================
# 产业研判与环节评析（数据基座 v_factor_judgement）
# 用处见 judgement.py：选股系统每个计划日自留一份快照，攒版本史。
# ============================================================================
_JUDGEMENT_COLS = ("scope", "subject_name", "segment_name", "cluster_key", "leaning",
                   "n_bull", "n_bear", "n_flags", "verified", "verify_problems",
                   "n_materials", "model", "rounds", "review_date", "reviewed_at",
                   "input_version", "review_id")


def judgement_rows(scopes=None, read_pg=None) -> list[dict]:
    """数据基座研判结论视图里 scope 落在指定几类的全部行，一行一个评析簇。

    scopes 不传时取 config.JUDGEMENT_SCOPES（默认产业研判 industry 与环节评析 segment）。
    每行带：主题名或环节名、采信倾向、多头论点条数、空头论点条数、旗标条数、自我校验结果与
    校验问题条数、材料条数、模型与轮次、生成日期与生成时刻、材料指纹、评析编号、簇键。

    这张视图按簇键唯一、重评时整行覆盖（视图注释里的已知局限 a），所以它永远只有最新一版；
    版本史由读的一方自己攒，见 judgement.py。读失败返回空列表并打印一行原因，不阻断上层。
    read_pg 可注入（离线单测），默认走 db.read_pg。"""
    want = tuple(scopes) if scopes else tuple(sorted(config.JUDGEMENT_SCOPES))
    if not want:
        return []
    reader = read_pg or db.read_pg
    try:
        marks = ",".join(["%s"] * len(want))
        rows = _records(reader(
            f"SELECT {', '.join(_JUDGEMENT_COLS)} FROM v_factor_judgement "
            f"WHERE scope IN ({marks})", want))
    except Exception as e:  # noqa: BLE001
        print(f"  （研判结论视图 v_factor_judgement 读取失败，本计划日没有行业观点行可写: {e!r}）")
        return []
    out = []
    for r in rows:
        key = _s(r.get("cluster_key"))
        subject = _s(r.get("subject_name")) or _s(r.get("segment_name"))
        if not key or not subject:
            continue          # 簇键或主题名缺一样，这一行没法参与逐版本比对，跳过
        out.append({
            "scope": _s(r.get("scope")), "subject_name": subject,
            "segment_name": _s(r.get("segment_name")), "cluster_key": key,
            "leaning": _s(r.get("leaning")),
            "n_bull": _i(r.get("n_bull")), "n_bear": _i(r.get("n_bear")),
            "n_flags": _i(r.get("n_flags")),
            "verified": _b(r.get("verified")),
            "n_verify_problems": _n_items(r.get("verify_problems")),
            "n_materials": _i(r.get("n_materials")),
            "model": _s(r.get("model")), "rounds": _i(r.get("rounds")),
            "review_date": _ymd(r.get("review_date")),
            "reviewed_at": _dt_str(r.get("reviewed_at")),
            "input_version": _s(r.get("input_version")),
            "review_id": _s(r.get("review_id")),
        })
    return out


def _dt_str(v) -> str | None:
    """时刻列统一成 MySQL 认的 'YYYY-MM-DD HH:MM:SS'；只有日期的补零点；认不出返回空。"""
    if v is None or (isinstance(v, float) and v != v):
        return None
    if isinstance(v, dt.datetime):
        return v.strftime("%Y-%m-%d %H:%M:%S")
    if isinstance(v, dt.date):
        return f"{v.isoformat()} 00:00:00"
    s = str(v).strip().replace("T", " ")
    if len(s) >= 19 and s[4] == "-" and s[7] == "-":
        return s[:19]
    day = _ymd(s)
    return f"{day} 00:00:00" if day else None


# ============================================================================
# 计划环境段的市场四项（2026-09-03 方案第 1.4 节清单里"有"与"可自算"的项）
# ============================================================================
def market_context(ds: str, read_pg=None, read_mysql=None) -> dict:
    """数据日 ds 的市场环境四项，全部只展示与复盘分组，不拦任何票。

      turnover    两市成交额：指数日线表 zs_day_data 里上证与深成当日 amount 之和，以及相对前五个
                  交易日均值的比值（原表单位，未换算）
      breadth     市场广度：基座个股日行情视图当日行自算——上涨、下跌、平盘家数，涨幅达 9.8% 的家数
                  （涨停近似），涨幅中位数
      margin      融资：eastmoney_rzrq_data 最新一日的 financing_balance 与 change_percent_5d
      fear_greed  恐贪指数：fear_greed_index 最新一日的 index_value 与日期

    每一项读失败为 None 并把原因记进 errors，不阻断出计划。read_pg / read_mysql 可注入（离线单测）。
    融资与恐贪两张表按"最新一日"取，不按 ds 过滤：它们是 T+1 更新的情绪读数，计划日早晨看到的
    就是最新一行；行里带日期，读者自己判断新鲜度。"""
    rpg = read_pg or db.read_pg
    rmy = read_mysql or db.read_mysql
    src = config.MARKET_MYSQL_SOURCE
    out = {"date": ds, "turnover": None, "breadth": None, "margin": None, "fear_greed": None,
           "errors": {}, "fetched_at": dt.datetime.now().isoformat(timespec="seconds")}

    # 一、两市成交额。不按日期列过滤（列的类型未在本仓库内核实：DATE 与整数 YYYYMMDD 的比较
    # 语义不同），改为取每个指数最近的几十行，在 Python 里按归一化日期筛不晚于 ds 的行。
    try:
        marks = ",".join(["%s"] * len(MARKET_INDEX_CODES))
        rows = _records(rmy(
            src, f"SELECT symbol, `timestamp` AS d, amount FROM zs_day_data "
                 f"WHERE symbol IN ({marks}) ORDER BY `timestamp` DESC LIMIT 80",
            tuple(MARKET_INDEX_CODES)))
        out["turnover"] = _turnover(rows, ds)
        if out["turnover"] is None:
            out["errors"]["turnover"] = "zs_day_data 最近 40 个交易日内没有不晚于计划日、且两市齐全的行"
    except Exception as e:  # noqa: BLE001
        out["errors"]["turnover"] = repr(e)
        print(f"  （两市成交额读取失败，环境段该项为空: {e!r}）")

    # 二、市场广度：基座个股日行情视图当日全部行自算。
    try:
        rows = _records(rpg(
            "SELECT pct_change FROM v_factor_stock_daily WHERE trade_date = %s", (ds,)))
        out["breadth"] = _breadth(rows)
        if out["breadth"] is None:
            out["errors"]["breadth"] = "v_factor_stock_daily 当日无行"
    except Exception as e:  # noqa: BLE001
        out["errors"]["breadth"] = repr(e)
        print(f"  （市场广度自算失败，环境段该项为空: {e!r}）")

    # 三、融资余额与五日变化；四、恐贪指数。两张表都取最新一日一行。
    for key, table, cols, label in (
            ("margin", "eastmoney_rzrq_data", ("financing_balance", "change_percent_5d"), "融资余额"),
            ("fear_greed", "fear_greed_index", ("index_value",), "恐贪指数")):
        try:
            row, date_col = _latest_row(rmy, src, table)
            if row is None:
                out["errors"][key] = f"{table} 为空表"
                continue
            item = {"date": _ymd(row.get(date_col)) if date_col else None,
                    "date_col": date_col, "source": table}
            for c in cols:
                item[c] = _f(row.get(c))
            if all(item[c] is None for c in cols):
                out["errors"][key] = f"{table} 最新行缺列 {cols}（实际列: {sorted(row)[:12]}）"
                continue
            out[key] = item
        except Exception as e:  # noqa: BLE001
            out["errors"][key] = repr(e)
            print(f"  （{label}读取失败，环境段该项为空: {e!r}）")
    return out


def _turnover(rows: list[dict], ds: str) -> dict | None:
    """两市成交额：按日期把两个指数的 amount 相加，只认两市齐全的日子；当日取不晚于 ds 的最近一日，
    前五日均值取它之前的五个交易日（不足五个按实际个数）。"""
    by_day: dict[str, dict] = {}
    for r in rows:
        d = _ymd(r.get("d"))
        a = _f(r.get("amount"))
        if not d or a is None or d > ds:
            continue
        by_day.setdefault(d, {})[str(r.get("symbol") or "").strip()] = a
    full = sorted((d for d, m in by_day.items() if all(c in m for c in MARKET_INDEX_CODES)),
                  reverse=True)
    if not full:
        return None
    day0 = full[0]
    amt0 = sum(by_day[day0].values())
    prev = [sum(by_day[d].values()) for d in full[1:6]]
    avg5 = (sum(prev) / len(prev)) if prev else None
    return {"data_date": day0, "amount": amt0, "prev5_avg": avg5,
            "ratio_vs_prev5": (amt0 / avg5) if avg5 else None, "prev5_days": len(prev),
            # 2026-09-03 实测核对：09-02 两市 amount 合计 1,321,831,143，而当日真实成交额约
            # 1.32 万亿元，所以这一列的单位是千元，不是元。换算成亿元要除以十万（乘一千再除一亿）。
            # 渲染与下游一律用 amount_yi，原始值保留在 amount 里备查。
            "unit": "千元",
            "amount_yi": amt0 / 1e5,
            "prev5_avg_yi": (avg5 / 1e5) if avg5 else None,
            "source": "zs_day_data 上证 000001.SH 与深成 399001.SZ 当日 amount 之和"}


def _breadth(rows: list[dict]) -> dict | None:
    pcts = [p for p in (_f(r.get("pct_change")) for r in rows) if p is not None]
    if not pcts:
        return None
    return {"n": len(pcts),
            "up": sum(1 for p in pcts if p > 0), "down": sum(1 for p in pcts if p < 0),
            "flat": sum(1 for p in pcts if p == 0),
            "limit_up_approx": sum(1 for p in pcts if p >= LIMIT_UP_PCT),
            "pct_median": round(statistics.median(pcts), 3),
            "limit_up_rule": f"涨幅达 {LIMIT_UP_PCT}% 记为涨停近似",
            "source": "v_factor_stock_daily 当日行自算"}


def _latest_row(rmy, src: str, table: str) -> tuple[dict | None, str | None]:
    """取一张表按日期列排序的最新一行。日期列名先用一行样本探出（候选名见 _DATE_COL_CANDIDATES），
    探不到就按第一列倒序（通常是自增主键）并把 date_col 记为 None。"""
    sample = _records(rmy(src, f"SELECT * FROM {table} LIMIT 1"))
    if not sample:
        return None, None
    cols = list(sample[0].keys())
    date_col = next((c for c in _DATE_COL_CANDIDATES if c in cols), None)
    order = f"`{date_col}`" if date_col else "1"
    rows = _records(rmy(src, f"SELECT * FROM {table} ORDER BY {order} DESC LIMIT 1"))
    return (rows[0] if rows else None), date_col

# 丙路（券商行动）两个窗口各自的长度，自然日。等长是硬要求：窗口不等长会让八成的票
# 假显示覆盖收缩——实测前 135 天对近 45 天时有 907 只票误报。等长本身也是抗抖动的低通。
BROKER_WINDOW_DAYS = 45


def broker_actions(codes, ds: str, *, window_days: int = BROKER_WINDOW_DAYS,
                   read_mysql=None) -> dict:
    """券商用行动说话这一路：同一财年同一预测期的每股收益预测中位数与覆盖机构数，
    比较最近两个等长窗口。返回前缀码到 logic_state.signal 的字典（算不出的票不进字典）。

    三条口径必须照做，否则读数是错的：
      一，两个窗口等长（见上面那条常量的说明）。
      二，同一财年才可比，按预测期字段精确匹配，跨财年比较没有意义。
      三，同一家机构在窗口里可能发多篇，先按机构取最近一篇再算中位数，
          否则发得勤的机构会被重复计入。

    看的是券商的行动不是言辞——券商极少明说不看好某个行业，所以等不到它开口，
    只能看预测在不在下修、覆盖在不在收缩。
    """
    import logic_state as ls

    reader = read_mysql or db.read_mysql
    end = dt.date.fromisoformat(ds)
    mid = end - dt.timedelta(days=int(window_days))
    start = end - dt.timedelta(days=int(window_days) * 2)
    dotted = sorted({_to_dot(c) for c in codes if c})
    if not dotted:
        return {}
    marks = ",".join(["%s"] * len(dotted))
    try:
        df = reader(
            "factor",
            f"SELECT ts_code, report_date, quarter, org_name, eps FROM gp_report_rc "
            f"WHERE ts_code IN ({marks}) AND report_date > %s AND report_date <= %s "
            f"AND eps IS NOT NULL AND quarter IS NOT NULL",
            tuple(dotted) + (start.isoformat(), end.isoformat()))
    except Exception as e:  # noqa: BLE001
        print(f"  （券商研报明细表读取失败，券商行动这一路整体缺席: {e!r}）")
        return {}

    rows = df.itertuples() if hasattr(df, "itertuples") else []
    box: dict = defaultdict(lambda: defaultdict(lambda: {"now": {}, "prev": {}}))
    for r in rows:
        d = _ymd(r.report_date)
        if not d:
            continue
        win = "now" if d > mid.isoformat() else "prev"
        k = common.to_prefix(str(r.ts_code).strip())
        org = str(r.org_name or "").strip() or "未署名"
        slot = box[k][str(r.quarter).strip()][win]
        if org not in slot or d > slot[org][0]:      # 同机构多篇只留最近一篇
            slot[org] = (d, float(r.eps))

    out = {}
    for k, by_q in box.items():
        usable = [(q, v) for q, v in by_q.items() if v["now"] and v["prev"]]
        if not usable:
            continue
        q, v = max(usable, key=lambda kv: len(kv[1]["now"]) + len(kv[1]["prev"]))
        now = {"eps": statistics.median([x[1] for x in v["now"].values()]),
               "firms": len(v["now"])}
        prev = {"eps": statistics.median([x[1] for x in v["prev"].values()]),
                "firms": len(v["prev"])}
        sig = ls.from_broker(now, prev, as_of=ds)
        if sig["refs"]:
            sig["refs"][0]["quarter"] = q
        out[k] = sig
    return out