Stamp as_of on dividend quality snapshot outputs for runtime guards.

Ensures built snapshot CSVs satisfy QuantPlatformKit feature snapshot date requirements and AkShare staging emits the same column.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: Pigbibi

src/cn_equity_snapshot_pipelines/akshare_staging.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import argparse
+from datetime import datetime, timezone
 from pathlib import Path
 
 import pandas as pd
@@ -127,6 +128,9 @@ def build_factor_snapshot_from_akshare(
             }
         )
     frame = pd.DataFrame(rows, columns=list(FACTOR_SNAPSHOT_COLUMNS))
+    if "as_of" not in frame.columns and "snapshot_date" not in frame.columns:
+        stamp = datetime.now(timezone.utc).date().isoformat()
+        frame.insert(0, "as_of", stamp)
     diagnostics["row_count"] = len(frame)
     return frame, diagnostics
 

src/cn_equity_snapshot_pipelines/dividend_quality.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import argparse
+from datetime import date, datetime, timezone
 from pathlib import Path
 
 import pandas as pd
@@ -15,16 +16,26 @@ def _read_csv(path: str | Path) -> pd.DataFrame:
     return pd.read_csv(Path(path))
 
 
+def _ensure_snapshot_as_of(snapshot: pd.DataFrame, *, as_of: str | date | None = None) -> pd.DataFrame:
+    frame = snapshot.copy()
+    if "as_of" in frame.columns or "snapshot_date" in frame.columns:
+        return frame
+    stamp = as_of or datetime.now(timezone.utc).date().isoformat()
+    frame.insert(0, "as_of", str(stamp))
+    return frame
+
+
 def build_and_write_snapshot(
     *,
     factor_snapshot_path: str | Path,
     output_dir: str | Path,
     min_adv20_cny: float = 0.0,
     min_market_cap_cny: float = 0.0,
+    as_of: str | date | None = None,
 ) -> SnapshotBuildResult:
     contract = get_profile_contract(CN_DIVIDEND_QUALITY_SNAPSHOT_PROFILE)
     artifact_paths = contract.artifact_paths(output_dir)
-    snapshot = _read_csv(factor_snapshot_path)
+    snapshot = _ensure_snapshot_as_of(_read_csv(factor_snapshot_path), as_of=as_of)
     ranking = score_candidates(
         snapshot,
         min_adv20_cny=float(min_adv20_cny),
@@ -74,13 +85,15 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument("--output-dir", default="data/output/dividend_quality")
     parser.add_argument("--min-adv20-cny", type=float, default=0.0)
     parser.add_argument("--min-market-cap-cny", type=float, default=0.0)
+    parser.add_argument("--as-of", default=None, help="Snapshot as_of date (YYYY-MM-DD). Defaults to UTC today.")
     args = parser.parse_args(argv)
 
     result = build_and_write_snapshot(
         factor_snapshot_path=args.factor_snapshot,
         output_dir=args.output_dir,
         min_adv20_cny=args.min_adv20_cny,
         min_market_cap_cny=args.min_market_cap_cny,
+        as_of=args.as_of,
     )
     print(f"snapshot={result.artifact_paths['snapshot']}")
     print(f"manifest={result.artifact_paths['manifest']}")

tests/test_dividend_quality.py

@@ -10,9 +10,11 @@ def test_build_dividend_quality_snapshot_writes_artifact_pack(tmp_path: Path):
     result = build_and_write_snapshot(
         factor_snapshot_path=sample,
         output_dir=tmp_path,
+        as_of="2026-02-25",
     )
 
     assert len(result.snapshot) == 4
+    assert "as_of" in result.snapshot.columns
     assert not result.ranking.empty
     for key in ("snapshot", "manifest", "ranking", "release_summary"):
         assert result.artifact_paths[key].exists()