209 lines
9.5 KiB
Python
209 lines
9.5 KiB
Python
"""输入冻结（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
|