From f85128307dd0e3407d24350282fe7b679b591309 Mon Sep 17 00:00:00 2001 From: PureSaber Date: Mon, 12 Oct 2026 00:50:56 +0800 Subject: [PATCH] Add immutable revision impact tracing and frozen pair validation --- README.md | 3 + docs/REVISION_IMPACT.md | 57 ++++++ src/quant_data_kit/revision_impact.py | 279 ++++++++++++++++++++++++++ tests/test_revision_impact.py | 207 +++++++++++++++++++ 4 files changed, 546 insertions(+) create mode 100644 docs/REVISION_IMPACT.md create mode 100644 src/quant_data_kit/revision_impact.py create mode 100644 tests/test_revision_impact.py diff --git a/README.md b/README.md index 43749aa..6b47357 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,8 @@ # quant-data-kit +Immutable downstream revision tracing and frozen paired result comparison are +available through [Revision impact](docs/REVISION_IMPACT.md). + On Windows, deep lake partitions can exceed `MAX_PATH`. If system long paths are disabled, supply an absolute extended-length lake root (for example `Path("//?/F:/lake")`). The pytest temporary-directory fixture uses this form diff --git a/docs/REVISION_IMPACT.md b/docs/REVISION_IMPACT.md new file mode 100644 index 0000000..6a09e41 --- /dev/null +++ b/docs/REVISION_IMPACT.md @@ -0,0 +1,57 @@ +# Immutable revision impact + +`quant_data_kit.revision_impact.RevisionRegistry` extends the existing local +`ArtifactStore`; it does not replace snapshot IDs, manifests, or SHA256 identities. +Register a completed output, named `ArtifactRef` inputs, stage, exact code hashes, +and frozen controls with `register(..., owner=...)`. Registration atomically pins +the descriptor and its complete dependencies. Descriptors are immutable; a new +revision or control change creates new evidence. Do not release historical pins +while they are required for audit or replay. + +`impact(old_input, new_input)` follows registered derivations and existing store +dependency edges, including a Parquet member's enclosing `SnapshotStore` +manifest. It returns affected output references and the complete set of affected +input dependencies after closure. Affected means reachable, not economically +changed, and not causal proof. Unregistered external objects are outside scope. +Every registered descriptor and its dependency closure is integrity checked. + +```python +from pathlib import Path +from quant_data_kit.artifacts import ArtifactStore, canonical_json +from quant_data_kit.revision_impact import RevisionRegistry + +store = ArtifactStore(Path("research-store")) +registry = RevisionRegistry(store) +source = store.put_bytes(canonical_json({"price": "100"}), owner="study-v1") +result = store.put_bytes(canonical_json({"nav": "1000"}), owner="study-v1") +derivation = registry.register( + output=result, inputs={"market": source}, stage="account", + code_identity={"producer": "a" * 40}, # replace with actual verified code identity + controls={"recipe": "frozen-recipe-v1"}, owner="study-v1", +) +``` + +The generic registry records producer declarations; it cannot establish what +executable actually ran. `paired(old_derivation, new_derivation, old_input=..., +new_input=...)` requires identical stage, code identity, frozen controls and all +other input references. It returns exact JSON differences, with finite decimal +deltas where applicable. It rejects changed code/controls and multi-input +confounding. Both outputs and their prior decisions remain intact. + +The portfolio `revision-study` adapter supplies actual source/runtime hashes, +checks them before and after each execution, reruns the frozen plan through the +continuous QExec account, registers both results, and reports available economic +stage differences. Features and forecasts are explicitly unavailable there. +This is retrospective data-revision sensitivity, not historical OOS certification. + +Write old/new `ArtifactRef.to_dict()` JSON files and run the read-only impact CLI: + +```bash +python -m quant_data_kit.revision_impact --store research-store \ + --old-ref old-ref.json --new-ref new-ref.json --out impact.json +``` + +The CLI refuses to overwrite an existing report. Existing artifact deletion/GC +rules continue to apply; no automatic deletion, acquisition, network access or +live execution is added. Revert this additive module and guide to roll back; +leave published artifacts and historical pins intact. diff --git a/src/quant_data_kit/revision_impact.py b/src/quant_data_kit/revision_impact.py new file mode 100644 index 0000000..382620a --- /dev/null +++ b/src/quant_data_kit/revision_impact.py @@ -0,0 +1,279 @@ +"""Immutable derivation evidence and revision reachability over ArtifactStore. + +Reachability is limited to registered objects. It is not causal attribution and +does not infer dependencies from filenames, mutable aliases, or coincident IDs. +""" + +from __future__ import annotations + +import argparse +import hashlib +import io +import json +import re +from decimal import Decimal, InvalidOperation +from pathlib import Path + +from quant_data_kit.artifacts import ArtifactRef, ArtifactStore, canonical_json +from quant_data_kit.exceptions import ValidationError + +SCHEMA = "quant.research-derivation/v1" + + +def sha256(value): + return hashlib.sha256(canonical_json(value)).hexdigest() + + +def _identity(code): + if ( + not isinstance(code, dict) + or not code + or any( + not isinstance(k, str) + or not k + or not isinstance(v, str) + or not re.fullmatch(r"[0-9a-f]{40}|[0-9a-f]{64}", v) + for k, v in code.items() + ) + ): + raise ValidationError("exact named code hashes required") + + +class RevisionRegistry: + def __init__(self, store: ArtifactStore): + self.store = store + with store._session() as db: + db.execute( + "CREATE TABLE IF NOT EXISTS research_derivations (" + "descriptor_sha TEXT PRIMARY KEY REFERENCES objects(sha) ON DELETE CASCADE, " + "output_sha TEXT NOT NULL REFERENCES objects(sha))" + ) + + def register(self, *, output, inputs, stage, code_identity, controls, owner): + """Atomically pin a derivation and all input/output dependencies. + + Code/control values are declarations from the producer; consumers must + verify the actual executable identity separately when running a replay. + Multiple histories producing identical bytes remain distinct derivations. + """ + _identity(code_identity) + if not isinstance(stage, str) or not stage.strip() or not isinstance(controls, dict): + raise ValidationError("stage and explicit controls object required") + if ( + not isinstance(inputs, dict) + or not inputs + or any(not isinstance(k, str) or not k.strip() for k in inputs) + ): + raise ValidationError("named immutable inputs required") + if any(r.sha256 == output.sha256 for r in inputs.values()): + raise ValidationError("self-dependent output forbidden") + document = { + "schema": SCHEMA, + "stage": stage, + "output": output.to_dict(), + "inputs": {k: v.to_dict() for k, v in sorted(inputs.items())}, + "code_identity": code_identity, + "controls": controls, + "controls_sha256": sha256(controls), + } + with self.store._session() as db: + dependencies = [output, *inputs.values()] + for ref in dependencies: + self.store._verify_closure(db, ref) + # Reject cycles in the semantic output -> input graph, not just in + # ArtifactStore's descriptor -> artifact dependency edges. + known = self._all(db) + pending = {r.sha256 for r in inputs.values()} + visited = set() + while pending: + node = pending.pop() + if node == output.sha256: + raise ValidationError("cyclic derivation") + if node in visited: + continue + visited.add(node) + for _, item in known: + if item["output"]["sha256"] == node: + pending.update(r["sha256"] for r in item["inputs"].values()) + descriptor = self.store._put_stream( + db, io.BytesIO(canonical_json(document)), "application/vnd.quant.derivation+json" + ) + for ref in dependencies: + db.execute( + "INSERT OR IGNORE INTO edges VALUES (?, ?)", (descriptor.sha256, ref.sha256) + ) + db.execute( + "INSERT OR IGNORE INTO research_derivations VALUES (?, ?)", + (descriptor.sha256, output.sha256), + ) + self.store._pin(db, owner, [descriptor]) + return descriptor + + def _load(self, db, ref): + self.store._verify_closure(db, ref) + value = json.loads(self.store._verify(db, ref).read_bytes()) + if value.get("schema") != SCHEMA or value.get("controls_sha256") != sha256( + value["controls"] + ): + raise ValidationError("invalid derivation manifest") + _identity(value["code_identity"]) + for raw in [value["output"], *value["inputs"].values()]: + child = ArtifactRef.from_dict(raw) + if not db.execute( + "SELECT 1 FROM edges WHERE parent=? AND child=?", (ref.sha256, child.sha256) + ).fetchone(): + raise ValidationError("unregistered derivation dependency") + self.store._verify_closure(db, child) + return value + + def load(self, ref): + with self.store._session() as db: + return self._load(db, ref) + + def _all(self, db): + results = [] + for sha, size in db.execute( + "SELECT descriptor_sha, size FROM research_derivations " + "JOIN objects ON descriptor_sha=objects.sha ORDER BY descriptor_sha" + ).fetchall(): + ref = ArtifactRef(sha, size, f"qdk+sha256:{sha}") + results.append((ref, self._load(db, ref))) + return results + + def impact(self, old_input, new_input): + with self.store._session() as db: + self.store._verify_closure(db, old_input) + self.store._verify_closure(db, new_input) + known = self._all(db) + edges = db.execute("SELECT parent, child FROM edges").fetchall() + affected = {old_input.sha256} if old_input.sha256 != new_input.sha256 else set() + found = {} + changed = True + while changed: + changed = False + for parent, child in edges: + if child in affected and parent not in affected: + affected.add(parent) + changed = True + for ref, item in known: + causes = sorted( + r["sha256"] for r in item["inputs"].values() if r["sha256"] in affected + ) + if causes and ref.sha256 not in found: + found[ref.sha256] = { + "derivation": ref.to_dict(), + "stage": item["stage"], + "output": item["output"], + "affected_dependencies": causes, + } + affected.add(item["output"]["sha256"]) + changed = True + # A node may be discovered before all of its upstream paths. Report + # dependencies from the completed closure, not first-discovery order. + for ref, item in known: + if ref.sha256 in found: + found[ref.sha256]["affected_dependencies"] = sorted( + r["sha256"] for r in item["inputs"].values() if r["sha256"] in affected + ) + return { + "schema": "quant.revision-impact/v1", + "old_input": old_input.to_dict(), + "new_input": new_input.to_dict(), + "affected_objects": [found[k] for k in sorted(found)], + "affected_artifact_hashes": sorted(affected - {old_input.sha256}), + "scope": "registered immutable dependency graph only", + "causal_claim": False, + "historical_results_modified": False, + } + + def paired(self, old_derivation, new_derivation, *, old_input, new_input): + """Validate a frozen data-only pair; never waive the normal comparison guard.""" + old, new = self.load(old_derivation), self.load(new_derivation) + for field in ("stage", "code_identity", "controls", "controls_sha256"): + if old[field] != new[field]: + raise ValidationError(f"not a frozen data-only pair: changed {field}") + expected = { + name: new_input.to_dict() if ref["sha256"] == old_input.sha256 else ref + for name, ref in old["inputs"].items() + } + if old_input.sha256 not in {r["sha256"] for r in old["inputs"].values()}: + raise ValidationError("old revision is not an input of the old result") + if expected != new["inputs"]: + raise ValidationError("pair changed other inputs or missing revision replacement") + old_result = json.loads(self.store.read_bytes(ArtifactRef.from_dict(old["output"]))) + new_result = json.loads(self.store.read_bytes(ArtifactRef.from_dict(new["output"]))) + return { + "schema": "quant.paired-revision/v1", + "intervention": "data_revision", + "code_identity": old["code_identity"], + "controls_sha256": old["controls_sha256"], + "old_derivation": old_derivation.to_dict(), + "new_derivation": new_derivation.to_dict(), + "old_input": old_input.to_dict(), + "new_input": new_input.to_dict(), + "old_result": old["output"], + "new_result": new["output"], + "changes": json_changes(old_result, new_result), + "historical_results_modified": False, + "causal_claim": False, + } + + +def json_changes(old, new, path=""): + """Stable exact JSON differences. Numeric delta only for finite decimal values.""" + if type(old) is type(new) and old == new: + return [] + if isinstance(old, dict) and isinstance(new, dict): + changes = [] + for key in sorted(set(old) | set(new)): + child = path + "/" + key.replace("~", "~0").replace("/", "~1") + if key not in old or key not in new: + changes.append( + { + "path": child, + "old": old.get(key), + "new": new.get(key), + "change": "added" if key not in old else "removed", + } + ) + else: + changes.extend(json_changes(old[key], new[key], child)) + return changes + if isinstance(old, list) and isinstance(new, list) and len(old) == len(new): + return [ + change + for i, (a, b) in enumerate(zip(old, new)) + for change in json_changes(a, b, f"{path}/{i}") + ] + result = {"path": path, "old": old, "new": new, "change": "modified"} + if not isinstance(old, bool) and not isinstance(new, bool): + try: + a, b = Decimal(str(old)), Decimal(str(new)) + if a.is_finite() and b.is_finite(): + result["delta"] = str(b - a) + except InvalidOperation: + pass + return [result] + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--store", type=Path, required=True) + parser.add_argument("--old-ref", type=Path, required=True) + parser.add_argument("--new-ref", type=Path, required=True) + parser.add_argument("--out", type=Path, required=True) + args = parser.parse_args(argv) + registry = RevisionRegistry(ArtifactStore(args.store)) + report = registry.impact( + *[ + ArtifactRef.from_dict(json.loads(path.read_text(encoding="utf-8"))) + for path in (args.old_ref, args.new_ref) + ] + ) + with args.out.open("x", encoding="utf-8") as stream: + json.dump(report, stream, indent=2, allow_nan=False) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_revision_impact.py b/tests/test_revision_impact.py new file mode 100644 index 0000000..0a6d0a7 --- /dev/null +++ b/tests/test_revision_impact.py @@ -0,0 +1,207 @@ +import json + +import pytest + +from quant_data_kit.artifacts import ArtifactStore, canonical_json +from quant_data_kit.exceptions import ValidationError +from quant_data_kit.revision_impact import RevisionRegistry, json_changes, main + + +def test_transitive_impact_preserves_original_and_pins_dependencies(tmp_path): + store = ArtifactStore(tmp_path / "s") + registry = RevisionRegistry(store) + old = store.put_bytes(b'{"price":"100"}') + new = store.put_bytes(b'{"price":"110"}') + feature = store.put_bytes(b'{"feature":"1"}') + forecast = store.put_bytes(b'{"forecast":"2"}') + separate = store.put_bytes(b'{"unrelated":true}') + for output, source, stage in [ + (feature, old, "feature"), + (forecast, feature, "forecast"), + (separate, new, "unrelated"), + ]: + registry.register( + output=output, + inputs={"input": source}, + stage=stage, + controls={"fixed": True}, + code_identity={"producer": "a" * 40}, + owner=stage, + ) + report = registry.impact(old, new) + assert {x["stage"] for x in report["affected_objects"]} == {"feature", "forecast"} + assert not registry.impact(old, old)["affected_objects"] + assert store.read_bytes(old) == b'{"price":"100"}' + with pytest.raises(ValidationError, match="referenced|protected|pinned"): + store.delete(old) + refs = [] + for name, ref in [("old", old), ("new", new)]: + path = tmp_path / (name + ".json") + path.write_text(json.dumps(ref.to_dict()), encoding="utf-8") + refs.append(path) + out = tmp_path / "impact.json" + assert ( + main( + [ + "--store", + str(store.root), + "--old-ref", + str(refs[0]), + "--new-ref", + str(refs[1]), + "--out", + str(out), + ] + ) + == 0 + ) + assert json.loads(out.read_bytes()) == report + with pytest.raises(FileExistsError): + main( + [ + "--store", + str(store.root), + "--old-ref", + str(refs[0]), + "--new-ref", + str(refs[1]), + "--out", + str(out), + ] + ) + + +def test_frozen_pair_and_changed_controls_code_or_other_input_rejected(tmp_path): + store = ArtifactStore(tmp_path) + registry = RevisionRegistry(store) + old, new, control = [store.put_bytes(v) for v in (b"old", b"new", b"control")] + outputs = [store.put_bytes(canonical_json({"nav": v})) for v in ("100", "102")] + + def register(i, **changes): + kwargs = { + "output": outputs[i], + "inputs": {"data": (old, new)[i], "control": control}, + "stage": "result", + "controls": {"fee": "1"}, + "code_identity": {"producer": "a" * 64}, + "owner": "pair", + } + return registry.register(**{**kwargs, **changes}) + + a, b = register(0), register(1) + pair = registry.paired(a, b, old_input=old, new_input=new) + assert pair["changes"] == [ + {"path": "/nav", "old": "100", "new": "102", "change": "modified", "delta": "2"} + ] + for changes in [ + {"controls": {"fee": "2"}}, + {"code_identity": {"producer": "b" * 64}}, + {"inputs": {"data": new}}, + {"stage": "other"}, + ]: + different = register(1, **changes) + with pytest.raises(ValidationError): + registry.paired(a, different, old_input=old, new_input=new) + with pytest.raises(ValidationError): + registry.paired(a, b, old_input=control, new_input=new) + + +def test_corruption_cycles_and_unknown_identity_fail(tmp_path): + store = ArtifactStore(tmp_path) + registry = RevisionRegistry(store) + a, b = store.put_bytes(b"a"), store.put_bytes(b"b") + kwargs = { + "stage": "feature", + "controls": {}, + "code_identity": {"producer": "a" * 40}, + "owner": "cycle", + } + ref = registry.register(output=b, inputs={"data": a}, **kwargs) + with pytest.raises(ValidationError, match="cyclic"): + registry.register(output=a, inputs={"data": b}, **kwargs) + with pytest.raises(ValidationError, match="self-dependent"): + registry.register(output=a, inputs={"data": a}, **kwargs) + with pytest.raises(ValidationError, match="code hashes"): + registry.register( + output=a, inputs={"data": b}, **{**kwargs, "code_identity": {"producer": "main"}} + ) + store._object_path(a.sha256).write_bytes(b"corrupted") + with pytest.raises(ValidationError, match="integrity"): + registry.load(ref) + with pytest.raises(ValidationError, match="integrity"): + registry.impact(a, b) + + +def test_json_diff_missing_and_bool_are_not_numeric_zero(): + result = json_changes( + {"a/b": True, "gone": None, "list": [1]}, {"a/b": 1, "added": None, "list": [2]} + ) + assert result[0] == {"path": "/a~1b", "old": True, "new": 1, "change": "modified"} + assert result[1]["change"] == "added" + assert result[2]["change"] == "removed" + assert result[3]["delta"] == "1" + + +def test_diamond_reports_all_affected_dependencies_independent_of_order(tmp_path, monkeypatch): + store = ArtifactStore(tmp_path) + registry = RevisionRegistry(store) + a, x, b, z, revised = [ + store.put_bytes(value.encode()) for value in ("a", "x", "b", "z", "revised") + ] + records = [] + for output, inputs in [(x, {"a": a}), (b, {"x": x}), (z, {"a": a, "b": b})]: + records.append( + registry.register( + output=output, + inputs=inputs, + stage="diamond", + code_identity={"p": "a" * 40}, + controls={}, + owner="diamond", + ) + ) + original = registry._all + monkeypatch.setattr( + registry, + "_all", + lambda db: sorted(original(db), key=lambda pair: pair[1]["output"]["sha256"] != z.sha256), + ) + impact = registry.impact(a, revised) + result = next(row for row in impact["affected_objects"] if row["output"]["sha256"] == z.sha256) + assert result["affected_dependencies"] == sorted([a.sha256, b.sha256]) + monkeypatch.setattr(registry, "_all", lambda db: list(reversed(original(db)))) + assert registry.impact(a, revised) == impact + + +def test_snapshot_partition_dependency_is_traced(tmp_path): + import pandas as pd + + from quant_data_kit.shared_snapshots import SnapshotStore + + store = ArtifactStore(tmp_path / "store") + registry = RevisionRegistry(store) + path = tmp_path / "bars.parquet" + pd.DataFrame({"date": ["2026-01-02"], "symbol": ["A"], "price": [100]}).to_parquet(path) + snapshot = SnapshotStore(store).publish( + {"bars": path}, + dataset="prices", + version="v1", + temporal={"availability": "synthetic"}, + owner="snapshot", + ) + from quant_data_kit.artifacts import ArtifactRef + + member = ArtifactRef.from_dict(snapshot.document["members"][0]["artifact"]) + output = store.put_bytes(b'{"nav":"100"}') + registry.register( + output=output, + inputs={"snapshot": snapshot.artifact}, + stage="result", + code_identity={"p": "a" * 40}, + controls={}, + owner="result", + ) + revised = store.put_bytes(b"new partition") + report = registry.impact(member, revised) + assert snapshot.artifact.sha256 in report["affected_artifact_hashes"] + assert report["affected_objects"][0]["output"] == output.to_dict()