feat(registry): record whether a research result can still be trusted

lab/registry/__init__.py

@@ -76,12 +76,26 @@ def _init_schema(self) -> None:
                     evaluation        TEXT,                    -- JSON: ic/icir/dsr/bh_fdr/...
                     input_hash        TEXT,                    -- reproducibility fingerprint
                     report_path       TEXT,
-                    warnings          TEXT                     -- JSON list
+                    warnings          TEXT,                    -- JSON list
+                    validity          TEXT DEFAULT 'valid',    -- valid|affected|invalidated|superseded
+                    affected_by       TEXT DEFAULT '[]'        -- JSON list of reasons, e.g. ["BUG-03"]
                 );
                 CREATE INDEX IF NOT EXISTS idx_runs_status ON research_runs(status);
                 CREATE INDEX IF NOT EXISTS idx_runs_factor ON research_runs(factor);
                 """
             )
+
+            # Databases created before the validity columns exist need them
+            # added: CREATE TABLE IF NOT EXISTS leaves an existing table alone.
+            cols = {r["name"] for r in conn.execute("PRAGMA table_info(research_runs)")}
+            if "validity" not in cols:
+                conn.execute(
+                    "ALTER TABLE research_runs ADD COLUMN validity TEXT DEFAULT 'valid'"
+                )
+            if "affected_by" not in cols:
+                conn.execute(
+                    "ALTER TABLE research_runs ADD COLUMN affected_by TEXT DEFAULT '[]'"
+                )
             conn.commit()
 
     def begin_run(self, slice_name: str, inputs: dict[str, Any]) -> str:
@@ -157,3 +171,86 @@ def list_runs(self, limit: int = 50) -> list[dict]:
                 "SELECT * FROM research_runs ORDER BY timestamp DESC LIMIT ?", (limit,)
             ).fetchall()
         return [dict(r) for r in rows]
+
+    # ── Result validity ─────────────────────────────────────────────────
+    #
+    # A bug fix that changes numerical output invalidates earlier results
+    # without making them disappear. NO_LOOKAHEAD_CONTRACT's own remediation
+    # procedure asks for the affected versions to be *marked*, not deleted:
+    #
+    #     1. 这是一个 bug，提交 issue 或 PR
+    #     2. 修复后在该文件中更新"已在代码中实现的位置"
+    #     3. 如果该 bug 影响了之前的回测结果，标注 affected 版本
+    #
+    # These methods are what makes step 3 answerable by machine instead of by
+    # reading commit messages.
+
+    VALIDITY_STATES = ("valid", "affected", "invalidated", "superseded")
+
+    def mark_validity(self, run_id: str, validity: str, reason: str = "") -> bool:
+        """Record whether a run's result can still be trusted.
+
+        Args:
+            run_id: The run to mark.
+            validity: One of VALIDITY_STATES.
+            reason: What affected it, e.g. "BUG-03". Appended to the run's
+                affected_by list if not already present, so a run can be
+                affected by several things and the history is additive.
+
+        Returns:
+            True if the run existed and was updated.
+        """
+        if validity not in self.VALIDITY_STATES:
+            raise ValueError(
+                f"validity must be one of {self.VALIDITY_STATES}, got {validity!r}"
+            )
+
+        with self._conn() as conn:
+            row = conn.execute(
+                "SELECT affected_by FROM research_runs WHERE run_id=?", (run_id,)
+            ).fetchone()
+            if row is None:
+                return False
+
+            try:
+                existing = json.loads(row["affected_by"] or "[]")
+            except json.JSONDecodeError:
+                existing = []
+            if reason and reason not in existing:
+                existing.append(reason)
+
+            conn.execute(
+                "UPDATE research_runs SET validity=?, affected_by=? WHERE run_id=?",
+                (validity, json.dumps(existing), run_id),
+            )
+        return True
+
+    def list_affected(self, reason: str | None = None) -> list[dict]:
+        """Runs whose results are no longer plain 'valid'.
+
+        Args:
+            reason: Optional filter, e.g. "BUG-03" -- returns only runs marked
+                as affected by that specific cause.
+
+        Returns:
+            Matching runs, newest first.
+        """
+        with self._conn() as conn:
+            rows = conn.execute(
+                "SELECT * FROM research_runs "
+                "WHERE COALESCE(validity, 'valid') != 'valid' "
+                "ORDER BY timestamp DESC"
+            ).fetchall()
+
+        out = [dict(r) for r in rows]
+        if reason:
+            filtered = []
+            for r in out:
+                try:
+                    causes = json.loads(r.get("affected_by") or "[]")
+                except json.JSONDecodeError:
+                    causes = []
+                if reason in causes:
+                    filtered.append(r)
+            out = filtered
+        return out

main.py

@@ -626,6 +626,50 @@ def cmd_cache(args) -> int:
     return 0
 
 
+def cmd_runs(args) -> int:
+    """Research run registry: list results and their validity.
+
+    Exists so that "which results does this bug fix invalidate?" is a query
+    rather than an exercise in reading commit messages. A fix that changes
+    numerical output marks the earlier runs affected; it does not delete them.
+    """
+    from quant_platform.lab.registry import DEFAULT_DB, RunStore
+
+    store = RunStore(getattr(args, "db", None) or DEFAULT_DB)
+    sub = getattr(args, "subcommand", None) or "list"
+
+    if sub == "mark":
+        ok = store.mark_validity(args.run_id, args.state, args.reason or "")
+        if not ok:
+            print(f"Unknown run: {args.run_id}")
+            return 1
+        print(
+            f"{args.run_id}: validity={args.state}"
+            + (f"  affected_by={args.reason}" if args.reason else "")
+        )
+        return 0
+
+    if sub == "affected":
+        rows = store.list_affected(args.reason)
+        title = "Runs affected" + (f" by {args.reason}" if args.reason else "")
+    else:
+        rows = store.list_runs(getattr(args, "limit", 50))
+        title = "Recent runs"
+
+    if not rows:
+        print(f"{title}: none")
+        return 0
+
+    print(f"{title} ({len(rows)}):")
+    print(f"  {'run_id':36s} {'validity':12s} {'factor':18s} affected_by")
+    for r in rows:
+        print(
+            f"  {r['run_id']:36s} {r.get('validity') or 'valid':12s} "
+            f"{(r.get('factor') or '-'):18s} {r.get('affected_by') or '[]'}"
+        )
+    return 0
+
+
 def cmd_web(args) -> int:
     """Start the web server (FastAPI + Vue frontend)."""
     import uvicorn
@@ -1698,6 +1742,29 @@ def main() -> int:
     cache_parser.add_argument("--cache-dir", type=str, default=".quant_cache",
                               help="Cache directory path")
 
+    # runs — research run registry and result validity
+    runs_parser = subparsers.add_parser(
+        "runs", help="Research runs: list results and mark their validity"
+    )
+    runs_sub = runs_parser.add_subparsers(dest="subcommand", help="Runs action")
+    runs_list = runs_sub.add_parser("list", help="List recent runs with validity")
+    runs_list.add_argument("--limit", type=int, default=50, help="How many runs")
+    runs_list.add_argument("--db", type=str, default=None, help="Registry path")
+    runs_aff = runs_sub.add_parser(
+        "affected", help="Runs marked affected/invalidated rather than valid"
+    )
+    runs_aff.add_argument("--reason", type=str, default=None,
+                          help="Filter by cause, e.g. BUG-03")
+    runs_aff.add_argument("--db", type=str, default=None, help="Registry path")
+    runs_mark = runs_sub.add_parser("mark", help="Mark a run's result validity")
+    runs_mark.add_argument("run_id", type=str, help="Run id to mark")
+    runs_mark.add_argument("state", type=str,
+                           choices=["valid", "affected", "invalidated", "superseded"],
+                           help="New validity state")
+    runs_mark.add_argument("--reason", type=str, default="",
+                           help="Cause, e.g. BUG-03")
+    runs_mark.add_argument("--db", type=str, default=None, help="Registry path")
+
     # ml
     ml_parser = subparsers.add_parser("ml", help="ML alpha signal operations")
     ml_sub = ml_parser.add_subparsers(dest="subcommand", help="ML action")
@@ -1874,6 +1941,8 @@ def main() -> int:
         return cmd_sweep(args)
     elif args.command == "cache":
         return cmd_cache(args)
+    elif args.command == "runs":
+        return cmd_runs(args)
     elif args.command == "web":
         return cmd_web(args)
     elif args.command == "ml":

tests/test_lab/test_registry_validity.py

@@ -0,0 +1,127 @@
+"""Tests for research-run result validity metadata.
+
+A fix that changes numerical output invalidates earlier results without
+deleting them. `NO_LOOKAHEAD_CONTRACT.md`'s remediation procedure asks for the
+affected versions to be *marked*, so the Registry needs somewhere to record
+that -- otherwise "which results does this fix invalidate?" can only be
+answered by reading commit messages.
+"""
+
+import sqlite3
+
+import pytest
+from quant_platform.lab.registry import RunStore
+
+
+@pytest.fixture
+def store(tmp_path):
+    return RunStore(str(tmp_path / "registry.db"))
+
+
+@pytest.fixture
+def two_runs(store):
+    # Inputs differ per run on purpose: run_id is `run_<unix seconds>_<input
+    # hash>`, so identical inputs within the same second collide on the primary
+    # key. That collision is a separate defect; this fixture sidesteps it so
+    # these tests stay about validity.
+    for i in (1, 2):
+        store.begin_run(f"slice_{i}", {"factor": "roe", "n": i})
+    return store.list_runs()
+
+
+class TestSchema:
+    def test_new_database_has_validity_columns(self, tmp_path):
+        db = str(tmp_path / "new.db")
+        RunStore(db)
+        cols = {
+            r[1] for r in sqlite3.connect(db).execute("PRAGMA table_info(research_runs)")
+        }
+        assert "validity" in cols
+        assert "affected_by" in cols
+
+    def test_existing_database_is_migrated(self, tmp_path):
+        """CREATE TABLE IF NOT EXISTS does not add columns to an existing
+        table, so the store has to alter it."""
+        db = str(tmp_path / "old.db")
+        conn = sqlite3.connect(db)
+        conn.execute(
+            """CREATE TABLE research_runs (
+                   run_id TEXT PRIMARY KEY, timestamp TEXT NOT NULL,
+                   status TEXT NOT NULL, slice TEXT NOT NULL,
+                   factor TEXT, evaluation TEXT)"""
+        )
+        conn.execute(
+            "INSERT INTO research_runs VALUES "
+            "('legacy', '2026-01-01', 'success', 's', 'roe', '{}')"
+        )
+        conn.commit()
+        conn.close()
+
+        store = RunStore(db)  # opening it must migrate
+
+        cols = {
+            r[1] for r in sqlite3.connect(db).execute("PRAGMA table_info(research_runs)")
+        }
+        assert {"validity", "affected_by"} <= cols
+        # And the pre-existing row survives with a sensible default.
+        assert store.get_run("legacy") is not None
+        assert store.get_run("legacy")["validity"] == "valid"
+
+
+class TestMarkValidity:
+    def test_marking_a_run(self, store, two_runs):
+        run_id = two_runs[0]["run_id"]
+        assert store.mark_validity(run_id, "affected", "BUG-03") is True
+
+        run = store.get_run(run_id)
+        assert run["validity"] == "affected"
+        assert "BUG-03" in run["affected_by"]
+
+    def test_unknown_run_returns_false(self, store):
+        assert store.mark_validity("does-not-exist", "affected", "BUG-03") is False
+
+    def test_reasons_accumulate_without_duplicating(self, store, two_runs):
+        """A run can be affected by several findings, but the same cause
+        should not be recorded twice."""
+        run_id = two_runs[0]["run_id"]
+        store.mark_validity(run_id, "affected", "BUG-03")
+        store.mark_validity(run_id, "affected", "BUG-03")  # duplicate
+        store.mark_validity(run_id, "affected", "BUG-22")
+
+        import json
+
+        causes = json.loads(store.get_run(run_id)["affected_by"])
+        assert causes == ["BUG-03", "BUG-22"]
+
+    def test_other_runs_are_left_alone(self, store, two_runs):
+        store.mark_validity(two_runs[0]["run_id"], "invalidated", "BUG-03")
+        assert store.get_run(two_runs[1]["run_id"])["validity"] == "valid"
+
+    def test_rejects_unknown_state(self, store, two_runs):
+        with pytest.raises(ValueError, match="validity must be one of"):
+            store.mark_validity(two_runs[0]["run_id"], "probably-fine", "x")
+
+
+class TestListAffected:
+    def test_empty_when_nothing_is_marked(self, store, two_runs):
+        assert store.list_affected() == []
+
+    def test_returns_only_marked_runs(self, store, two_runs):
+        store.mark_validity(two_runs[0]["run_id"], "affected", "BUG-03")
+        affected = store.list_affected()
+        assert [r["run_id"] for r in affected] == [two_runs[0]["run_id"]]
+
+    def test_filters_by_reason(self, store, two_runs):
+        store.mark_validity(two_runs[0]["run_id"], "affected", "BUG-03")
+        store.mark_validity(two_runs[1]["run_id"], "affected", "BUG-22")
+
+        assert len(store.list_affected("BUG-03")) == 1
+        assert store.list_affected("BUG-03")[0]["run_id"] == two_runs[0]["run_id"]
+        assert len(store.list_affected("BUG-22")) == 1
+        assert store.list_affected("BUG-99") == []
+
+    def test_superseded_and_invalidated_are_included(self, store, two_runs):
+        """Anything that is no longer plain 'valid' needs to surface."""
+        store.mark_validity(two_runs[0]["run_id"], "superseded", "BUG-03")
+        store.mark_validity(two_runs[1]["run_id"], "invalidated", "BUG-22")
+        assert len(store.list_affected()) == 2