From 2ba1f6e471eb7e0b0c24cbd0f97a8cfeb9cf2a32 Mon Sep 17 00:00:00 2001 From: PureSaber <88314620+PureSaber@users.noreply.github.com> Date: Thu, 1 Oct 2026 11:05:28 +0800 Subject: [PATCH 1/2] fix: preserve first-observed action versions with refresh receipts --- docs/RESEARCH_DATASETS.md | 31 ++++++- src/quant_data_kit/research_dataset.py | 118 ++++++++++++++++++++++-- tests/test_research_dataset.py | 120 +++++++++++++++++++++++++ 3 files changed, 257 insertions(+), 12 deletions(-) diff --git a/docs/RESEARCH_DATASETS.md b/docs/RESEARCH_DATASETS.md index e3c3d8e..4ad036a 100644 --- a/docs/RESEARCH_DATASETS.md +++ b/docs/RESEARCH_DATASETS.md @@ -146,7 +146,30 @@ adjustment change had no matching ETF cash record. The dataset does not rewrite a delayed payment to the ex-date. A downstream simulator that cannot represent dividend receivables should continue to fail until it implements that ledger. -`history.parquet` uses the snapshot capture time as `available_at`. The source -announcement only provides a calendar date, so the dataset does not invent an -intraday historical availability timestamp. It is a captured-current history, -not a historical-vintage PIT feed. +Corporate-action `captured_at` is the first observed time of that complete row +version within a verified snapshot lineage. Refresh compares every other field, +including the event ID, entitlement dates, amounts, provider and source record. +Only an identical version inherits its parent's receipt time. A changed version +keeps its new receipt; an action removed from a complete live response is removed +from the current view. Local bounded updates retain records outside their fragment. +`update_evidence.action_changes` separates additions, removals, revisions and +unchanged records; checking event IDs alone does not establish unchanged content. + +Each build or refresh seals the incoming normalized rows, their actual receipt +times and the source declaration in `action-observations.json`. Its SHA-256 is +bound by `evidence_files` and verified on load, separately from the first-observed +action view. Snapshot captures cannot move backwards; missing, naive, future or +regressed record receipts are rejected. Old snapshots are never rewritten. + +`history.parquet` uses each action version's first receipt as `available_at`. +Unchanged classification declarations retain their first observation as well. +The source announcement only provides a calendar date, so the dataset does not +invent an intraday historical availability timestamp. This remains a captured +history, not evidence of vintages before the first actual observation. The default +instrument catalog still requires a verified historical master for forward use. + +This contract changes the executing code identity. Frozen forward accounts cannot +be silently migrated or have their hashes replaced. Preserve their original code, +inputs and failure receipts; validate and preregister any replacement account with +a new code identity and a future start. Downstream full-row prefix checks remain +unchanged and still reject real historical revisions. diff --git a/src/quant_data_kit/research_dataset.py b/src/quant_data_kit/research_dataset.py index 1261e8a..e79a8d6 100644 --- a/src/quant_data_kit/research_dataset.py +++ b/src/quant_data_kit/research_dataset.py @@ -109,6 +109,65 @@ def _empty_actions() -> pd.DataFrame: return pd.DataFrame(columns=ACTION_COLUMNS) +def _record_versions(frame: pd.DataFrame, timestamp_column: str) -> list[bytes]: + """Identify the whole evidenced row version, excluding only its receipt time.""" + records = json.loads( + frame.drop(columns=timestamp_column).to_json( + orient="records", date_format="iso", date_unit="ns", force_ascii=False + ) + ) + return [_canonical_bytes(record) for record in records] + + +def _first_observed_view( + previous: pd.DataFrame, + observed: pd.DataFrame, + timestamp_column: str, + observed_at: pd.Timestamp, +) -> pd.DataFrame: + """Retain a verified parent's first receipt only for an identical row version.""" + known = dict(zip(_record_versions(previous, timestamp_column), previous[timestamp_column])) + result = observed.copy() + for index, identity in zip(result.index, _record_versions(result, timestamp_column)): + if pd.isna(result.at[index, timestamp_column]): + raise ValueError("Record receipt timestamp is missing") + receipt = _captured_at(result.at[index, timestamp_column]) + if receipt > observed_at: + raise ValueError("Record receipt cannot be later than the snapshot capture") + if identity in known: + original = _captured_at(known[identity]) + if receipt < original: + raise ValueError("An identical record version cannot move its receipt backwards") + result.at[index, timestamp_column] = known[identity] + return result + + +def _action_changes(previous: pd.DataFrame, observed: pd.DataFrame, *, complete: bool) -> dict: + keys = ["symbol", "ex_date"] + for frame in (previous, observed): + if frame.duplicated(keys).any() or frame.duplicated("event_id").any(): + raise ValueError("Corporate actions contain duplicate event identities or ex-dates") + old = dict( + zip( + previous[keys].itertuples(index=False, name=None), + _record_versions(previous, "captured_at"), + ) + ) + new = dict( + zip( + observed[keys].itertuples(index=False, name=None), + _record_versions(observed, "captured_at"), + ) + ) + common = old.keys() & new.keys() + return { + "added": len(new.keys() - old.keys()), + "removed": len(old.keys() - new.keys()) if complete else 0, + "revised": sum(old[key] != new[key] for key in common), + "unchanged": sum(old[key] == new[key] for key in common), + } + + def _read_frame(path: Path) -> pd.DataFrame: suffix = path.suffix.lower() if suffix == ".parquet": @@ -605,7 +664,7 @@ def _history( "effective_at": effective, # This is a captured-current view. We do not invent an intraday # historical publication timestamp from a date-only announcement. - "available_at": captured_at, + "available_at": _captured_at(row.captured_at), "value": str(getattr(row, field)), } ) @@ -720,6 +779,8 @@ def _publish( update_evidence: dict[str, Any], instrument_master_root: Path | None = None, preserved_history_root: Path | None = None, + previous_history: pd.DataFrame | None = None, + action_observations: pd.DataFrame | None = None, ) -> dict[str, Any]: root = root.resolve() root.mkdir(parents=True, exist_ok=True) @@ -728,6 +789,13 @@ def _publish( try: if preserved_history_root is None: history = _history(frames["actions"], symbols, start, captured_at) + if previous_history is not None: + history = _first_observed_view( + previous_history[previous_history.domain == "classification"], + history, + "available_at", + captured_at, + ) else: preserved_history_root = preserved_history_root.resolve() history = validate_history( @@ -841,6 +909,26 @@ def _publish( "sha256": _sha256(history_source), }, } + if action_observations is not None: + receipt_path = staging / "action-observations.json" + receipt_path.write_bytes( + _canonical_bytes( + { + "schema_version": "qdk.action-observations/v1", + "observed_at": captured_at.isoformat(), + "source": source, + "records": json.loads( + action_observations.to_json( + orient="records", date_format="iso", force_ascii=False + ) + ), + } + ) + ) + evidence_files["action_observations"] = { + "file": receipt_path.name, + "sha256": _sha256(receipt_path), + } if instrument_master_root is not None: for path in sorted(master_destination.rglob("*")): if path.is_file(): @@ -874,6 +962,10 @@ def _publish( "consumer_contract": { "asm_decision_workflow_load_inputs": True, "history_availability": "captured-current; no inferred intraday announcement time", + "action_availability": ( + "first receipt of identical evidenced row version; " + "fresh observations are retained separately" + ), "instrument_master_availability": ( "captured-current; strict historical replay remains blocked" if master_manifest is None @@ -959,6 +1051,9 @@ def build_dataset( license_note=license_note, ) frames = _normalize_frames(frames, symbols=normalized_symbols, start=start_day, end=end_day) + frames["actions"] = _first_observed_view( + _empty_actions(), frames["actions"], "captured_at", captured + ) validation = validate_dataset_frames( frames, symbols=normalized_symbols, start=start_day, end=end_day ) @@ -973,6 +1068,7 @@ def build_dataset( validation=validation, parent_snapshot_id=None, update_evidence={"mode": "initial_build", "revised_rows": {}}, + action_observations=frames["actions"], ) @@ -998,6 +1094,8 @@ def update_dataset( if end_day < old_end: raise ValueError("Incremental update cannot shorten the dataset") captured = _captured_at(captured_at) + if captured < _captured_at(previous["captured_at"]): + raise ValueError("Incremental snapshot capture cannot move backwards") if end_day > captured.tz_convert("Asia/Shanghai").tz_localize(None).normalize(): raise ValueError("Research dataset cannot include a future session") old_dates = pd.DatetimeIndex(old["raw"]["date"].drop_duplicates().sort_values()) @@ -1031,21 +1129,22 @@ def update_dataset( ): merged[name], revised[name] = _merge_by_key(old[name], frames[name], keys, end=end_day) merged["calendar"] = frames["calendar"] + action_changes = _action_changes( + old["actions"], frames["actions"], complete=source["mode"] == "live_public_api" + ) + action_view = _first_observed_view(old["actions"], frames["actions"], "captured_at", captured) # The ETF action endpoints return the complete fund history. Treat the new # response as the authoritative current vintage and retain old rows only # when a declared local update contains a bounded fragment. if source["mode"] == "live_public_api": - merged["actions"] = frames["actions"] - old_action_ids = set(old["actions"].get("event_id", [])) - new_action_ids = set(frames["actions"].get("event_id", [])) - revised["actions"] = len(old_action_ids.symmetric_difference(new_action_ids)) + merged["actions"] = action_view elif frames["actions"].empty: merged["actions"] = old["actions"].copy() - revised["actions"] = 0 else: - merged["actions"], revised["actions"] = _merge_by_key( - old["actions"], frames["actions"], ["symbol", "ex_date"], end=end_day + merged["actions"], _ = _merge_by_key( + old["actions"], action_view, ["symbol", "ex_date"], end=end_day ) + revised["actions"] = sum(action_changes[key] for key in ("added", "removed", "revised")) merged = _normalize_frames(merged, symbols=symbols, start=start_day, end=end_day) validation = validate_dataset_frames(merged, symbols=symbols, start=start_day, end=end_day) master_root = None @@ -1067,8 +1166,11 @@ def update_dataset( "fetched_from": pd.Timestamp(price_start).date().isoformat(), "previous_end": old_end.date().isoformat(), "revised_rows": revised, + "action_changes": action_changes, }, instrument_master_root=master_root, + previous_history=old["history"], + action_observations=frames["actions"], ) diff --git a/tests/test_research_dataset.py b/tests/test_research_dataset.py index f08e1ea..1288a12 100644 --- a/tests/test_research_dataset.py +++ b/tests/test_research_dataset.py @@ -4,6 +4,7 @@ import pandas as pd import pytest +from quant_data_kit import research_dataset as dataset from quant_data_kit.instrument_master import import_instrument_master from quant_data_kit.providers.corporate_actions import normalize_etf_actions from quant_data_kit.research_coverage import load_history @@ -297,3 +298,122 @@ def test_coverage_distinguishes_leading_unavailability_from_later_gaps(tmp_path) _build(tmp_path / "dataset", source, end="2026-01-19") assert "missing_after_first_available_bar" in str(exc.value) assert "listing_date_basis" in str(exc.value) + + +def _live_action_refresh(tmp_path, monkeypatch): + source = tmp_path / "source" + dates = pd.DatetimeIndex(["2026-01-15", "2026-01-16", "2026-01-19", "2026-01-20"]) + _source(source, dates, [10, 10, 9, 9.1], [9, 9, 9, 9.1]) + response = { + name: pd.read_parquet(source / f"{name}.parquet") + for name in ("raw", "adjusted", "benchmark", "calendar", "actions") + } + + def fetch(*args, **kwargs): + return {key: value.copy() for key, value in response.items()}, { + "mode": "live_public_api", + "provider": "akshare_sina_etf", + } + + monkeypatch.setattr(dataset, "_fetch_live_source", fetch) + root = tmp_path / "dataset" + parent = build_dataset( + root, + symbols=["510300"], + start="2026-01-15", + end="2026-01-20", + captured_at="2026-01-22T00:00:00Z", + ) + response["actions"]["captured_at"] = "2026-01-23T00:00:00Z" + return root, parent, response + + +def test_live_recapture_preserves_first_receipt_and_seals_fresh_evidence(tmp_path, monkeypatch): + root, parent, response = _live_action_refresh(tmp_path, monkeypatch) + _, previous = load_research_snapshot(root) + child = update_dataset(root, end="2026-01-20", captured_at="2026-01-23T00:00:00Z") + _, current = load_research_snapshot(root) + pd.testing.assert_frame_equal(previous["actions"], current["actions"]) + pd.testing.assert_frame_equal(previous["history"], current["history"]) + assert child["parent_snapshot_id"] == parent["snapshot_id"] + assert child["update_evidence"]["revised_rows"]["actions"] == 0 + assert child["update_evidence"]["action_changes"] == { + "added": 0, + "removed": 0, + "revised": 0, + "unchanged": 1, + } + receipt_path = root / "snapshots" / child["snapshot_id"] / "action-observations.json" + receipt = json.loads(receipt_path.read_text(encoding="utf-8")) + assert receipt["records"][0]["captured_at"] == "2026-01-23T00:00:00Z" + assert receipt["observed_at"] == child["captured_at"] + response["actions"]["captured_at"] = "2026-01-24T00:00:00Z" + update_dataset(root, end="2026-01-20", captured_at="2026-01-24T00:00:00Z") + _, repeated = load_research_snapshot(root) + pd.testing.assert_frame_equal(previous["actions"], repeated["actions"]) + receipt_path.write_text("{}", encoding="utf-8") + with pytest.raises(ValueError, match="evidence integrity"): + load_research_snapshot(root, child["snapshot_id"]) + + +@pytest.mark.parametrize( + "field,value", + [ + ("event_id", "revised-source-document"), + ("source_record", '{"document":"corrected"}'), + ("source", "another-provider"), + ("announced_date", pd.Timestamp("2026-01-13")), + ("pay_date", pd.Timestamp("2026-01-28")), + ], +) +def test_live_action_revision_cannot_inherit_old_availability(tmp_path, monkeypatch, field, value): + root, parent, response = _live_action_refresh(tmp_path, monkeypatch) + response["actions"].loc[0, field] = value + child = update_dataset(root, end="2026-01-20", captured_at="2026-01-23T00:00:00Z") + _, current = load_research_snapshot(root) + assert current["actions"].iloc[0].captured_at == "2026-01-23T00:00:00Z" + assert child["update_evidence"]["action_changes"]["revised"] == 1 + assert child["update_evidence"]["revised_rows"]["actions"] == 1 + assert child["files"]["actions"]["sha256"] != parent["files"]["actions"]["sha256"] + assert ( + current["history"] + .loc[current["history"].domain == "corporate_actions", "available_at"] + .eq(pd.Timestamp("2026-01-23T00:00:00Z")) + .all() + ) + + +def test_live_changed_cash_and_removed_action_are_not_hidden(tmp_path, monkeypatch): + root, _, response = _live_action_refresh(tmp_path, monkeypatch) + response["actions"].loc[0, "cash_per_share"] = "2" + before_ex = response["adjusted"].date < pd.Timestamp("2026-01-19") + response["adjusted"].loc[before_ex, ["open", "high", "low", "close"]] = 8 + child = update_dataset(root, end="2026-01-20", captured_at="2026-01-23T00:00:00Z") + _, current = load_research_snapshot(root) + assert current["actions"].iloc[0].captured_at == "2026-01-23T00:00:00Z" + assert child["update_evidence"]["action_changes"]["revised"] == 1 + response["actions"] = response["actions"].iloc[:0] + response["adjusted"] = response["raw"].assign(adjustment="qfq") + removed = update_dataset(root, end="2026-01-20", captured_at="2026-01-24T00:00:00Z") + _, current = load_research_snapshot(root) + assert current["actions"].empty + assert removed["update_evidence"]["action_changes"]["removed"] == 1 + assert removed["update_evidence"]["revised_rows"]["actions"] == 1 + + +@pytest.mark.parametrize( + "receipt", [None, "2026-01-23", "2026-01-24T00:00:00Z", "2026-01-21T00:00:00Z"] +) +def test_bad_or_regressed_action_receipt_cannot_publish(tmp_path, monkeypatch, receipt): + root, parent, response = _live_action_refresh(tmp_path, monkeypatch) + response["actions"]["captured_at"] = receipt + with pytest.raises(ValueError, match="receipt|time zone"): + update_dataset(root, end="2026-01-20", captured_at="2026-01-23T00:00:00Z") + assert load_research_snapshot(root)[0]["snapshot_id"] == parent["snapshot_id"] + + +def test_snapshot_capture_cannot_move_backwards(tmp_path, monkeypatch): + root, parent, _ = _live_action_refresh(tmp_path, monkeypatch) + with pytest.raises(ValueError, match="capture cannot move backwards"): + update_dataset(root, end="2026-01-20", captured_at="2026-01-21T00:00:00Z") + assert load_research_snapshot(root)[0]["snapshot_id"] == parent["snapshot_id"] From 935a9d5ea8963afa55c843c3a0b64af450321588 Mon Sep 17 00:00:00 2001 From: PureSaber <88314620+PureSaber@users.noreply.github.com> Date: Thu, 1 Oct 2026 11:10:04 +0800 Subject: [PATCH 2/2] test: validate action refresh against the real forward prefix --- tests/test_research_dataset.py | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/tests/test_research_dataset.py b/tests/test_research_dataset.py index 1288a12..2bf14e5 100644 --- a/tests/test_research_dataset.py +++ b/tests/test_research_dataset.py @@ -417,3 +417,29 @@ def test_snapshot_capture_cannot_move_backwards(tmp_path, monkeypatch): with pytest.raises(ValueError, match="capture cannot move backwards"): update_dataset(root, end="2026-01-20", captured_at="2026-01-21T00:00:00Z") assert load_research_snapshot(root)[0]["snapshot_id"] == parent["snapshot_id"] + + +def test_action_refresh_preserves_pipeline_prefix_but_revisions_do_not(tmp_path, monkeypatch): + paper = pytest.importorskip("quant_pipeline.research_paper") + root, _, response = _live_action_refresh(tmp_path, monkeypatch) + master = tmp_path / "master" + import_instrument_master(_master_source(tmp_path / "master-source"), master) + initial = bind_instrument_master(root, master, captured_at="2026-09-26T00:00:00Z") + + def inputs(manifest): + snapshot = root / "snapshots" / manifest["snapshot_id"] + return { + "bundle": str(snapshot), + "catalog": str(snapshot / "catalog.csv"), + "history": str(snapshot / "history"), + } + + frozen = paper.input_prefix(inputs(initial), "2026-09-26") + response["actions"]["captured_at"] = "2026-09-27T00:00:00Z" + refreshed = update_dataset(root, end="2026-01-20", captured_at="2026-09-27T00:00:00Z") + assert paper.input_lineage(inputs(refreshed)) == paper.input_lineage(inputs(initial)) + assert paper.input_prefix(inputs(refreshed), "2026-09-26") == frozen + response["actions"]["captured_at"] = "2026-09-28T00:00:00Z" + response["actions"]["source_record"] = '{"document":"corrected"}' + revised = update_dataset(root, end="2026-01-20", captured_at="2026-09-28T00:00:00Z") + assert paper.input_prefix(inputs(revised), "2026-09-26") != frozen