fix: preserve evidenced action availability across refreshes

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.

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"],
     )
 
 

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,148 @@ 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"]
+
+
+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