260 lines
12 KiB
Python
260 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""fund PIT 视图层：static 三表多期 asof 视图（spec §20.11.7，2026-09-20）。
|
||
|
||
读法（终值快照 PIT，spec §20.11.7「PIT 语义边界」段）：
|
||
- 源=static per-stock 三表 ``<root>/<domain>/<code>.<EXC>_<domain>.parquet``；
|
||
每股每期可含重述多行（含补洗窗口内临时双行）：值取 (UPDATE_DATE,NOTICE_DATE)
|
||
排序末行=终值，有效披露日 NOTICE_DATE=组内最大（保守不提前——重述公告日前
|
||
整期不可见=宁缺毋假，asof 由更早期间兜底）。
|
||
- NOTICE_DATE 为空/NaT 的期整期弃用（宁缺毋假红线，与 fund_dmsk_reader
|
||
noticed_by 同语义）。
|
||
- 第 k 期=截至 asof 已披露的第 k 个报告期（NOTICE_DATE≤asof 含端点）；
|
||
asof 列表入参不预设调仓节律——回测 asof=历史日与实盘 asof=昨交易日边界
|
||
一致（策略①契约，fixture ⑤）。
|
||
- 拼盘态=常态（spec）：static=各股最近一次洗库时刻的拼盘，本视图如实透出。
|
||
|
||
失败分级（读取核契约）：
|
||
- 缺数据=空 df 不 raise + ``df.attrs['reason']`` 原因标签
|
||
（domain_not_found / no_data / filtered_empty / ok，fixture ④）；
|
||
- 单股缺文件/坏文件/零行=行级容错跳过（洗库容错纪律同 akshare_static_download，
|
||
不拖垮整域）；键列（REPORT_DATE/NOTICE_DATE）整体缺失的文件同样跳过。
|
||
|
||
coverage 伴生（``with_coverage=True`` 返 ``(view_df, coverage_df)``）：
|
||
- 每 (asof, REPORT_DATE) 的已披露家数/覆盖率——分子=首披露到达口径
|
||
（组内 NOTICE_DATE 最小值≤asof，即「该股历史上有过有效披露」；重述把终值
|
||
口径推迟但首披露仍算到达），分母=symbols 入参家数（null/缺文件/未披露均计
|
||
入分母，fixture ⑥）；与 periods 截断无关（全披露矩阵口径）。
|
||
- 到达口径（coverage）与终值口径（视图可见性）之差=重述窗口内的快照损失，
|
||
消费方可见可量化（spec「审计关卡量化其发生频率」的数据面落点）。
|
||
"""
|
||
import glob
|
||
import os
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
DOMAINS = ("income", "balance", "cashflow")
|
||
|
||
_STATIC_ROOT_ENV = "SANGUO_STATIC_ROOT"
|
||
_DEFAULT_STATIC_ROOT_WIN = r"C:\sanguo_vnpy_v2\data\static"
|
||
_DEFAULT_STATIC_ROOT_POSIX = "/volume1/stock/sanguo_vnpy_v2/data/static"
|
||
|
||
_BAD_DATE_STRINGS = ("", "NaT", "nan", "None", "none", "NoneType")
|
||
_COVERAGE_COLS = ["asof", "REPORT_DATE", "disclosed", "total", "ratio"]
|
||
|
||
|
||
def _static_root() -> str:
|
||
env = os.environ.get(_STATIC_ROOT_ENV)
|
||
if env:
|
||
return env
|
||
return _DEFAULT_STATIC_ROOT_WIN if os.name == "nt" else _DEFAULT_STATIC_ROOT_POSIX
|
||
|
||
|
||
def _norm_date(series: pd.Series) -> pd.Series:
|
||
"""任意日期形态（str/datetime/NaT）→ 'YYYY-MM-DD' 字符串前缀。"""
|
||
return series.astype(str).str[:10]
|
||
|
||
|
||
def _is_valid_date(series: pd.Series) -> pd.Series:
|
||
return series.notna() & ~series.isin(_BAD_DATE_STRINGS) & (series.str.len() == 10)
|
||
|
||
|
||
def _empty_view(columns) -> pd.DataFrame:
|
||
# 值列 dtype=object（空帧）；非空路径随源（数值 float64）——空/非空批次
|
||
# concat 会有 dtype 漂移，消费方分批处理时自行 astype（strategy review m2）
|
||
df = pd.DataFrame({c: pd.Series(dtype=object)
|
||
for c in ("asof", "symbol", "REPORT_DATE", "NOTICE_DATE",
|
||
"UPDATE_DATE") + tuple(columns)})
|
||
df["k"] = pd.Series(dtype="int64")
|
||
return df[["asof", "symbol", "k", "REPORT_DATE", "NOTICE_DATE", "UPDATE_DATE"]
|
||
+ list(columns)]
|
||
|
||
|
||
def _empty_coverage() -> pd.DataFrame:
|
||
df = pd.DataFrame({c: pd.Series(dtype=object) for c in ("asof", "REPORT_DATE")})
|
||
df["disclosed"] = pd.Series(dtype="int64")
|
||
df["total"] = pd.Series(dtype="int64")
|
||
df["ratio"] = pd.Series(dtype="float64")
|
||
return df[_COVERAGE_COLS]
|
||
|
||
|
||
def _load_stock(domain_dir: str, symbol: str, domain: str, columns: list):
|
||
"""单股单域 → [symbol, REPORT_DATE, NOTICE_DATE, notice_first, *columns]。
|
||
|
||
值取 (UPDATE_DATE,NOTICE_DATE) 排序末行（终值，UPDATE_DATE 空=最旧，
|
||
na_position='first' 与因子侧 polars nulls_first 对齐——09-20 factor review
|
||
MAJOR-1：pandas 默认 nulls-last 会把脏行（UPDATE=null）无条件推成终值）；
|
||
NOTICE_DATE=组内 max（可见性口径）、notice_first=组内 min（coverage 到达
|
||
口径）。缺文件/坏文件/零行/键列缺失 → None（行级容错）。
|
||
列下推（09-20 review MAJOR-3）：只读键列+UPDATE_DATE+所需列（schema 探头
|
||
后按需投影，缺列补 null），全市场调用免 319 列全量 IO。
|
||
"""
|
||
hits = sorted(glob.glob(os.path.join(
|
||
domain_dir, "%s.*_%s.parquet" % (symbol, domain))))
|
||
if not hits:
|
||
return None
|
||
try:
|
||
import pyarrow.parquet as pq
|
||
schema = set(pq.ParquetFile(hits[0]).schema_arrow.names)
|
||
except Exception:
|
||
return None
|
||
if "REPORT_DATE" not in schema or "NOTICE_DATE" not in schema:
|
||
return None
|
||
want = ["REPORT_DATE", "NOTICE_DATE"] \
|
||
+ [c for c in ("UPDATE_DATE",) if c in schema] \
|
||
+ [c for c in columns
|
||
if c in schema and c not in ("REPORT_DATE", "NOTICE_DATE",
|
||
"UPDATE_DATE")]
|
||
# ⚠️请求列必须按 schema 过滤（09-20 VPS 双验证回归②）：不过滤时缺列股
|
||
# （金融股 schema 变体）read_parquet(columns=...) 抛异常→整股被行级容错
|
||
# 跳过=漏斗静默缩水；过滤后缺列由下方 np.nan 补齐路径兜住。
|
||
try:
|
||
df = pd.read_parquet(hits[0], columns=want)
|
||
except Exception:
|
||
return None
|
||
if df.empty:
|
||
return None
|
||
df = df.copy()
|
||
df["REPORT_DATE"] = _norm_date(df["REPORT_DATE"])
|
||
df["NOTICE_DATE"] = _norm_date(df["NOTICE_DATE"])
|
||
# ① 宁缺毋假：NOTICE_DATE 无效的行剔除；某报告期全组无效则该期自然消失
|
||
# （重述行有效、原始行无效时该组退化为仅重述行——可见性随之推迟，自洽）
|
||
keep = _is_valid_date(df["REPORT_DATE"]) & _is_valid_date(df["NOTICE_DATE"])
|
||
df = df[keep]
|
||
if df.empty:
|
||
return None
|
||
if "UPDATE_DATE" in df.columns:
|
||
df["UPDATE_DATE"] = _norm_date(df["UPDATE_DATE"])
|
||
df.loc[~_is_valid_date(df["UPDATE_DATE"]), "UPDATE_DATE"] = np.nan
|
||
# kind="stable"+na_position="first"：确定性行序（跨重写复算一致，strategy
|
||
# review m1）+null UPDATE_DATE=最旧（与因子侧 polars nulls_first 对齐）
|
||
df = df.sort_values(
|
||
[c for c in ("UPDATE_DATE", "NOTICE_DATE") if c in df.columns],
|
||
na_position="first", kind="stable")
|
||
# 终值=排序末行**原样**取（nth(-1)）——GroupBy.last() 取组内最后非空值，
|
||
# 重述行该列 null 时会静默回退旧行值=拿旧值冒充终值（strategy review B1，
|
||
# 宁缺毋假红线）。⚠️pandas 2.3.3 的 nth 索引=**原始行号**而非组键、组序=
|
||
# 文件出现序（真实文件含招股书重述乱序块）——notice max/min 与终值行必须
|
||
# 按 REPORT_DATE **列 merge 对齐**（validate 一对一，假设破坏即 raise），
|
||
# 绝不按位置拼接（09-20 VPS 双验证实锤：位置拼接=NOTICE_DATE 张冠李戴到
|
||
# 别的报告期，静默 PIT 错；本地 fixture 恰好全升序故首轮测试漏网）。
|
||
# UPDATE_DATE 取终值行自带值（新鲜度可观测，strategy review M2）。
|
||
g = df.groupby("REPORT_DATE")
|
||
term = g.nth(-1).drop(columns=["NOTICE_DATE"]).reset_index(drop=True)
|
||
stats = g["NOTICE_DATE"].agg(["max", "min"]).reset_index()
|
||
term = term.merge(stats, on="REPORT_DATE", how="left",
|
||
validate="one_to_one")
|
||
term.insert(0, "symbol", symbol)
|
||
term = term.rename(columns={"max": "NOTICE_DATE", "min": "notice_first"})
|
||
if "UPDATE_DATE" not in term.columns:
|
||
term["UPDATE_DATE"] = np.nan
|
||
out = term[["symbol", "REPORT_DATE", "NOTICE_DATE", "UPDATE_DATE",
|
||
"notice_first"]].copy()
|
||
for c in columns:
|
||
# 缺列显式 float64 NaN（勿用 None——object 全 NA 列进 concat 触发
|
||
# FutureWarning 且未来升 object；数值列缺列语义=null 稳定）
|
||
out[c] = term[c].to_numpy() if c in term.columns else np.nan
|
||
return out
|
||
|
||
|
||
def _master_frame(domain: str, symbols: list, columns: list):
|
||
"""→ (master_df|None, missing)。missing=缺文件/坏文件/零行被行级容错跳过的
|
||
symbols（strategy review M1：静默丢弃=09-01 事故形态复燃口，必须向消费方
|
||
暴露——pit_view 落 attrs['skipped_symbols']）。"""
|
||
domain_dir = os.path.join(_static_root(), domain)
|
||
frames, missing = [], []
|
||
for sym in symbols:
|
||
per = _load_stock(domain_dir, sym, domain, columns)
|
||
if per is not None:
|
||
frames.append(per)
|
||
else:
|
||
missing.append(sym)
|
||
if not frames:
|
||
return None, missing
|
||
return pd.concat(frames, ignore_index=True), missing
|
||
|
||
|
||
def pit_view(domain, asof_dates, periods=1, symbols=None, columns=None,
|
||
with_coverage=False):
|
||
"""static 三表多期 asof PIT 视图（语义详见模块 docstring/spec §20.11.7）。
|
||
|
||
Args:
|
||
domain: ``income``/``balance``/``cashflow`` 之一（express/forecast 不在本
|
||
视图域——两层异源零交集，spec §20.11.7 探针证据⑤）
|
||
asof_dates: asof 日列表（'YYYY-MM-DD' 或全时间戳，前缀比较）
|
||
periods: 每 asof 取最近 k 个已披露期（k=1=最新）
|
||
symbols: 裸 6 位代码列表；None=目录内全市场推导
|
||
columns: 值列名列表（static 源列名直存零转换）；缺列补 null（schema 稳定）
|
||
with_coverage: True 时返 ``(view_df, coverage_df)``
|
||
|
||
Returns:
|
||
view_df 列 = ``[asof, symbol, k, REPORT_DATE, NOTICE_DATE, UPDATE_DATE,
|
||
*columns]``（UPDATE_DATE=组内 max=新鲜度可观测，strategy review M2）；
|
||
原因标签恒在 ``view_df.attrs['reason']``——**必须在首次返回时读取**
|
||
（filter/join/merge 会丢 attrs，strategy review m3）；行级容错跳过的
|
||
symbols 在 ``view_df.attrs['skipped_symbols']``（M1）。缺数据=空 df 不
|
||
raise。columns 必须显式传值列（None=仅骨架键列）。
|
||
"""
|
||
columns = list(columns) if columns else []
|
||
if domain not in DOMAINS:
|
||
view = _empty_view(columns)
|
||
view.attrs["reason"] = "domain_not_found"
|
||
view.attrs["skipped_symbols"] = []
|
||
return (view, _empty_coverage()) if with_coverage else view
|
||
if symbols is None:
|
||
domain_dir = os.path.join(_static_root(), domain)
|
||
names = glob.glob(os.path.join(domain_dir, "*_%s.parquet" % domain))
|
||
symbols = sorted(os.path.basename(p).split(".")[0] for p in names)
|
||
symbols = list(symbols)
|
||
asofs = [str(a)[:10] for a in asof_dates]
|
||
|
||
master, missing = _master_frame(domain, symbols, columns)
|
||
if master is None:
|
||
view = _empty_view(columns)
|
||
view.attrs["reason"] = "no_data"
|
||
view.attrs["skipped_symbols"] = missing
|
||
return (view, _empty_coverage()) if with_coverage else view
|
||
|
||
parts = []
|
||
for asof in asofs:
|
||
visible = master[master["NOTICE_DATE"] <= asof]
|
||
if visible.empty:
|
||
continue
|
||
visible = visible.sort_values(["symbol", "REPORT_DATE"],
|
||
ascending=[True, False])
|
||
visible = visible.assign(
|
||
k=visible.groupby("symbol").cumcount().to_numpy() + 1)
|
||
visible = visible[visible["k"] <= periods].copy()
|
||
visible.insert(0, "asof", asof)
|
||
parts.append(visible)
|
||
|
||
if parts:
|
||
view = pd.concat(parts, ignore_index=True)
|
||
view = view[["asof", "symbol", "k", "REPORT_DATE", "NOTICE_DATE",
|
||
"UPDATE_DATE"] + columns]
|
||
reason = "ok"
|
||
else:
|
||
view = _empty_view(columns)
|
||
reason = "filtered_empty" if not missing or len(missing) < len(symbols) \
|
||
else "no_data"
|
||
view.attrs["reason"] = reason
|
||
view.attrs["skipped_symbols"] = missing
|
||
|
||
if not with_coverage:
|
||
return view
|
||
total = len(symbols) if symbols else 0
|
||
cov_parts = []
|
||
for asof in asofs:
|
||
arrived = master[master["notice_first"] <= asof]
|
||
if arrived.empty:
|
||
continue
|
||
per = (arrived.groupby("REPORT_DATE")["symbol"].nunique()
|
||
.rename("disclosed").reset_index())
|
||
per.insert(0, "asof", asof)
|
||
per["total"] = total
|
||
per["ratio"] = per["disclosed"] / total if total else 0.0
|
||
cov_parts.append(per)
|
||
coverage = (pd.concat(cov_parts, ignore_index=True)[_COVERAGE_COLS]
|
||
if cov_parts else _empty_coverage())
|
||
return view, coverage
|