"""输入冻结（G0.5 · 评审 §5）——把每日「不可复原的上游」落成不可变快照。

**为什么必须有这个模块。** 桥内写入是幂等的，但上游不是：

1. `transmission_candidates` 的 quiet 集合与 moved_ratio 依赖 Neo4j 返回序
   （基座 `topic_context` 的 Cypher 是 `LIMIT $cap` 且无 ORDER BY），同日重跑可能不同；
2. `industry_pools` 只存最新态、每周一 refresh_pools 自动长大，没有时点版本；
3. `gp_day_data` 是**前复权**、锚在最新日——每次有股票除权，历史 close 会被整体
   重写，于是 `akg_upside` 的历史值今天回填和三个月后回填不是同一组数；
4. `consensus_daily` 的 asof 行会随 sync_consensus 重跑而覆盖。

于是「可复现」唯一诚实可达的定义是：**能从冻结的输入重算出同一个值**。
这个模块就是那份冻结，一次性满足三件事——判收标准 §10 的「重跑不产生漂移」、
「赛道成员快照可审计」、以及未来任何回溯需求（含 §10 里 GRU 复活所需的
「≥1 年 live 传导史」——那个计时器只在开始存快照的那天启动）。

落点：桥工程自身 `data/frozen/<YYYY-MM-DD>/`（config.FROZEN_ROOT），
不落基座、不落平台，守 §4 的三层边界。
格式：装了 pyarrow 就 parquet，否则 csv.gz（零新增依赖）。
每天一个 manifest.json 记录各源行数、告警与代码版本——**manifest 才是可 diff 的
审计线**，建议 git 跟踪 manifest.json、忽略数据文件。

用法（全程 Docker）：
    docker compose exec akg-factor-bridge python run.py freeze --date 2026-07-24
    docker compose exec akg-factor-bridge python run.py freeze          # 默认今天
    # 日更链里 `build ... --mode daily` 跑完会自动冻结（--no-freeze 可关）
"""
from __future__ import annotations

import datetime as dt
import json
import subprocess
from pathlib import Path

import pandas as pd

import config
import db

try:                      # parquet 更省更快，但不强制装 pyarrow
    import pyarrow        # noqa: F401
    _FMT, _EXT = "parquet", ".parquet"
except Exception:         # noqa: BLE001
    _FMT, _EXT = "csv.gz", ".csv.gz"


# ---------------------------------------------------------------- 落盘
def _write(df: pd.DataFrame, path: Path) -> int:
    path.parent.mkdir(parents=True, exist_ok=True)
    df = pd.DataFrame() if df is None else df
    # 空也要留痕：0 行本身是信息（"那天上游真的没有数据" ≠ "那天没跑"）
    if _FMT == "parquet":
        df.to_parquet(path, index=False)
    else:
        df.to_csv(path, index=False, compression="gzip")
    return len(df)


def _git_rev() -> str:
    """记录冻结时的桥代码版本——快照可复算的前提是知道当时的口径。"""
    try:
        return subprocess.run(["git", "rev-parse", "--short", "HEAD"],
                              cwd="/app", capture_output=True, text=True,
                              timeout=5).stdout.strip() or "unknown"
    except Exception:     # noqa: BLE001 —— 无 git / 无 .git 都不该让冻结失败
        return "unknown"


# ---------------------------------------------------------------- 各源抓取
def _cap_universe(_day: str) -> pd.DataFrame:
    """覆盖池当日成员（industry_pools 无历史版本 → 只能靠这里留时点）。"""
    return db.read_pg("SELECT ts_code FROM v_factor_universe ORDER BY ts_code")


def _cap_transmission(day: str) -> pd.DataFrame:
    """传导原始行。四路里唯一「过期即不可复原」的推断类信号，最该冻结。"""
    try:
        return db.read_pg(
            "SELECT * FROM v_factor_transmission WHERE scan_date = %s "
            "ORDER BY ts_code", (day,))
    except Exception:     # noqa: BLE001 —— 旧视图无 rank/target 列时的兜底
        return db.read_pg(
            "SELECT * FROM v_factor_transmission WHERE scan_date = %s", (day,))


def _cap_consensus(day: str) -> pd.DataFrame:
    """当日可见的最新一致预期（as-of 口径与 build_upside 一致，每股留最新一行）。"""
    return db.read_pg(
        "SELECT DISTINCT ON (ts_code) ts_code, asof_date, target_mid_avg, "
        "       eps_med, np_med, n_orgs, n_reports_90d "
        "FROM v_factor_consensus WHERE asof_date <= %s "
        "ORDER BY ts_code, asof_date DESC", (day,))


def _cap_price(day: str) -> pd.DataFrame:
    """当日收盘（**前复权**，会被后续除权整体重写 → 必须冻结当时看到的值）。"""
    import factors                      # 复用同一套代码列探测，口径不分叉
    col = factors._price_code_col()     # noqa: SLF001
    return db.read_mysql(
        "price",
        f"SELECT `timestamp` AS trade_date, `{col}` AS code, close "
        f"FROM gp_day_data WHERE `timestamp` = %s", (day,))


def _cap_heat(day: str) -> pd.DataFrame:
    """当日最新批次热度（T+1 到达，故 day 当天常为空——空也留痕）。"""
    return db.read_mysql(
        "heat",
        """SELECT s.trade_date, s.stock_code, s.batch_no, s.score
           FROM stock_fund_heat_scores s
           JOIN (SELECT trade_date, MAX(batch_no) bn FROM stock_fund_heat_scores
                 WHERE trade_date = %s GROUP BY trade_date) m
             ON m.trade_date = s.trade_date AND m.bn = s.batch_no""", (day,))


def _cap_optional(sql: str):
    """可选视图（第五/六个插槽视图还没建时不该让冻结失败）。"""
    def _f(_day: str) -> pd.DataFrame:
        try:
            return db.read_pg(sql)
        except Exception:   # noqa: BLE001
            return pd.DataFrame()
    return _f


_SOURCES = {
    "universe":     _cap_universe,
    "transmission": _cap_transmission,
    "consensus":    _cap_consensus,
    "price_close":  _cap_price,
    "heat":         _cap_heat,
    # 第五/六个插槽视图建好后自动开始冻结（没建 → 空 DataFrame，不报错）
    "segment_members": _cap_optional("SELECT * FROM v_factor_segment_members"),
    "segment_edges":   _cap_optional("SELECT * FROM v_factor_segment_edges"),
}


# ---------------------------------------------------------------- 体检断言
def _audit(day: str, tr: pd.DataFrame) -> list[str]:
    """把评审里那几个「静默失真」写成显式告警，冻结当时就看见。"""
    warns: list[str] = []
    if tr is None or tr.empty:
        return warns
    if "mkt_trade_date" in tr.columns and tr["mkt_trade_date"].notna().any():
        bad = sorted({str(x) for x in tr["mkt_trade_date"].dropna()
                      if str(x) != day})
        if bad:
            warns.append(f"传导用的 mkt 快照日 {bad} ≠ scan_date {day}"
                         f"——movers 陈旧，该日传导项不可信（评审 §6.2）")
    if "members_total" in tr.columns:
        hit = tr[pd.to_numeric(tr["members_total"], errors="coerce")
                 >= config.UPSTREAM_MEMBER_CAP]
        if not hit.empty:
            n = hit["target"].nunique() if "target" in hit.columns else len(hit)
            warns.append(f"{n} 个环节 members_total >= {config.UPSTREAM_MEMBER_CAP}，"
                         f"撞上游 topic_context cap——moved_ratio 建立在任意顺序的抽样上"
                         f"（评审 硬伤1）")
    if "n_quiet_stored" in tr.columns:
        hit = tr[pd.to_numeric(tr["n_quiet_stored"], errors="coerce")
                 >= config.UPSTREAM_QUIET_CAP]
        if not hit.empty:
            n = hit["target"].nunique() if "target" in hit.columns else len(hit)
            warns.append(f"{n} 个环节 quiet 存满 {config.UPSTREAM_QUIET_CAP} 条"
                         f"=撞 quiet[:12] 截断——真实未动成员更多（评审 硬伤1）")
    return warns


# ---------------------------------------------------------------- 入口
def snapshot(day: str | None = None, extra_frames: dict | None = None) -> dict:
    """冻结 day 当日的全部上游输入 +（可选）当日算出的因子值。

    extra_frames: {"factor_akg_heat": df, ...} —— run.py build 完顺手传进来，
    这样「输入 + 输出」在同一目录里，任何一行因子值都能被逐步复算。
    """
    day = day or dt.date.today().isoformat()
    out = Path(config.FROZEN_ROOT) / day
    manifest = {"date": day, "format": _FMT, "git_rev": _git_rev(),
                "subfactor_universe": config.SUBFACTOR_UNIVERSE,
                "frozen_at": dt.datetime.now().astimezone().isoformat(),
                "rows": {}, "errors": {}, "warnings": []}
    frames: dict[str, pd.DataFrame] = {}

    for name, fn in _SOURCES.items():
        try:
            df = fn(day)
            frames[name] = df
            manifest["rows"][name] = _write(df, out / f"{name}{_EXT}")
        except Exception as e:      # noqa: BLE001 —— 单源失败不拖累其余源
            manifest["errors"][name] = repr(e)
            print(f"  ❌ 冻结 {name} 失败: {e!r}")

    for name, df in (extra_frames or {}).items():
        try:
            manifest["rows"][name] = _write(df, out / f"{name}{_EXT}")
        except Exception as e:      # noqa: BLE001
            manifest["errors"][name] = repr(e)

    manifest["warnings"] = _audit(day, frames.get("transmission"))

    out.mkdir(parents=True, exist_ok=True)
    (out / "manifest.json").write_text(
        json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"  ❄️ 冻结 {day} → {out}")
    for k, v in manifest["rows"].items():
        print(f"     {k}: {v} 行")
    for w in manifest["warnings"]:
        print(f"     ⚠️ {w}")
    return manifest