fix(data_platform): VPS 双验证抓出 12f8e5a 两处静默 PIT 错(用户令「nas和vps双验证」的直接战果)——①NOTICE_DATE 与报告期错位:pandas 2.3.3 的 groupby.nth(-1) 索引=原始行号而非组键、组序=文件出现序(真实文件含招股书重述乱序块),12f8e5a 把 nth 终值行与键排序的 agg 按位置拼接=披露日张冠李戴到别的报告期(300750 Q1 披露日 04-16 错配 03-10 一代);本地 fixture 恰好全升序故 448 绿漏网,VPS 真数据(乱序块)一碰即中——修=按 REPORT_DATE 列 merge 对齐+validate 一对一(假设破坏即 raise),绝不按位置拼接;②列下推漏 schema 过滤:请求列不在文件 schema(银行股变体,000001 无 TOTAL_OPERATE_INCOME)时 read_parquet(columns=) 抛异常→整股被行级容错吞掉=漏斗静默缩水——修=按 schema 过滤后缺列走 np.nan 补齐;回归用例×2(乱序文件期日对齐/缺列不跳股);全套 451 绿;修版模块 VPS 真数据复验=300750/600519 各期 NOTICE_DATE 与 NAS 逐值一致,coverage 1/2 对齐 [nas] [no-doc]

sanguo_data/fund_pit.py

@@ -102,8 +102,12 @@ def _load_stock(domain_dir: str, symbol: str, domain: str, columns: list):
         return None
     want = ["REPORT_DATE", "NOTICE_DATE"] \
         + [c for c in ("UPDATE_DATE",) if c in schema] \
-        + [c for c in columns if c not in ("REPORT_DATE", "NOTICE_DATE",
-                                           "UPDATE_DATE")]
+        + [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:
@@ -129,18 +133,23 @@ def _load_stock(domain_dir: str, symbol: str, domain: str, columns: list):
         na_position="first", kind="stable")
     # 终值=排序末行**原样**取（nth(-1)）——GroupBy.last() 取组内最后非空值，
     # 重述行该列 null 时会静默回退旧行值=拿旧值冒充终值（strategy review B1，
-    # 宁缺毋假红线）；notice max/min 与终值行解耦各取各的；UPDATE_DATE 取终值
-    # 行自带值（该终值最后被动时刻=新鲜度可观测，strategy review M2）
-    term = df.groupby("REPORT_DATE", sort=True).nth(-1).reset_index()
-    stats = df.groupby("REPORT_DATE", sort=True).agg(
-        notice_max=("NOTICE_DATE", "max"),
-        notice_first=("NOTICE_DATE", "min")).reset_index()
-    out = term[["REPORT_DATE"]].copy()
-    out.insert(0, "symbol", symbol)
-    out["NOTICE_DATE"] = stats["notice_max"].to_numpy()
-    out["UPDATE_DATE"] = (term["UPDATE_DATE"].to_numpy()
-                          if "UPDATE_DATE" in term.columns else np.nan)
-    out["notice_first"] = stats["notice_first"].to_numpy()
+    # 宁缺毋假红线）。⚠️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 稳定）

tests/data_platform/test_fund_pit_view.py

@@ -194,3 +194,31 @@ def test_11_update_date_freshness_column(static_tree):
     df = pit_view("income", ["2026-10-30"], periods=1, symbols=["000001"],
                   columns=["REVENUE"])
     assert "UPDATE_DATE" in df.columns
+
+
+def test_12_period_row_order_independent_alignment(static_tree):
+    """09-20 VPS 双验证回归：文件行序乱序（真实文件含招股书重述块）时，
+    NOTICE_DATE 必须按 REPORT_DATE 索引对齐——按位置拼接会把披露日张冠李戴
+    到别的报告期（静默 PIT 错，本地 fixture 恰好升序曾漏网）。"""
+    _write_stock(static_tree, "000007.SZ", "income", [
+        ("2026-06-30", "2026-08-20", 7.0),
+        ("2025-12-31", "2026-03-20", 5.0),   # 行序打乱：中间夹旧期
+        ("2026-03-31", "2026-04-25", 6.0),
+    ])
+    df = pit_view("income", ["2026-09-19"], periods=3, symbols=["000007"],
+                  columns=["REVENUE"])
+    m = {r.REPORT_DATE: (r.NOTICE_DATE, r.REVENUE) for r in df.itertuples()}
+    assert m["2026-03-31"] == ("2026-04-25", 6.0), \
+        "每期必须配自己的披露日（乱序文件按位置拼接会错位）"
+    assert m["2026-06-30"] == ("2026-08-20", 7.0)
+    assert m["2025-12-31"] == ("2026-03-20", 5.0)
+
+
+def test_13_missing_requested_column_not_skipped(static_tree):
+    """09-20 VPS 双验证回归②：请求列不在文件 schema（金融股变体）→ 补 NaN，
+    绝不整股跳过（列下推不过滤 schema 时 read_parquet 抛异常→行级容错吞股）。"""
+    df = pit_view("income", ["2026-10-30"], periods=1, symbols=["000001"],
+                  columns=["REVENUE", "NOT_IN_SCHEMA_COL"])
+    assert len(df) == 1 and df["REVENUE"].iloc[0] == 100.0
+    assert pd.isna(df["NOT_IN_SCHEMA_COL"].iloc[0])
+    assert df.attrs.get("skipped_symbols") == [], "缺列≠缺股，不许静默跳过"