[verified] add frozen research runner and durable paper ledger

backend/app/__init__.py

@@ -17,10 +17,11 @@ from .bot_tourism import BotTourismSource, TourismSourceError
 from .event_study import EventStudyError, assess_backtest_readiness
 from .paper import PaperLedger
 from .prices import PriceSnapshotStore, PriceSourceError
+from .research import ResearchRunError, ResearchRunStore, run_tourism_research
 from .tourism import compute_tourism_signal
 from .vintages import VintageStore, VintageStoreError
 
-APP_VERSION = "0.4.0"
+APP_VERSION = "0.5.0"
 
 
 def _load_default_snapshot() -> dict[str, Any]:
@@ -47,6 +48,7 @@ def create_app(config: dict[str, Any] | None = None) -> Flask:
         PAPER_WRITE_TOKEN=os.getenv("PAPER_WRITE_TOKEN", ""),
         PAPER_COOKIE_SECURE=os.getenv("PAPER_COOKIE_SECURE", "0") == "1",
         PAPER_SESSION_SECONDS=int(os.getenv("PAPER_SESSION_SECONDS", "3600")),
+        PAPER_LEDGER_PATH=Path(os.getenv("PAPER_LEDGER_PATH", str(data_root / "paper" / "ledger.json"))),
         TOURISM_SOURCE=os.getenv("TOURISM_SOURCE", "fixture"),
         TOURISM_ADAPTER=None,
         TOURISM_DATA_ROOT=Path(os.getenv("TOURISM_DATA_ROOT", str(data_root))),
@@ -55,6 +57,8 @@ def create_app(config: dict[str, Any] | None = None) -> Flask:
         VINTAGE_STORE=None,
         PRICE_DATA_ROOT=Path(os.getenv("PRICE_DATA_ROOT", str(data_root / "prices"))),
         PRICE_STORE=None,
+        RESEARCH_DATA_ROOT=Path(os.getenv("RESEARCH_DATA_ROOT", str(data_root / "research"))),
+        RESEARCH_RUN_STORE=None,
         SNAPSHOT=_load_default_snapshot(),
     )
     if config:
@@ -64,6 +68,8 @@ def create_app(config: dict[str, Any] | None = None) -> Flask:
     app.extensions["vintage_store"] = vintage_store
     price_store = app.config.get("PRICE_STORE") or PriceSnapshotStore(app.config["PRICE_DATA_ROOT"])
     app.extensions["price_store"] = price_store
+    research_run_store = app.config.get("RESEARCH_RUN_STORE") or ResearchRunStore(app.config["RESEARCH_DATA_ROOT"])
+    app.extensions["research_run_store"] = research_run_store
     source_snapshot = app.config["SNAPSHOT"]
     source_mode = str(app.config["TOURISM_SOURCE"]).lower()
     if source_mode == "bot":
@@ -76,7 +82,10 @@ def create_app(config: dict[str, Any] | None = None) -> Flask:
         raise ValueError(f"unsupported TOURISM_SOURCE: {source_mode}")
 
     result = compute_tourism_signal(source_snapshot)
-    ledger = PaperLedger()
+    ledger_path = app.config["PAPER_LEDGER_PATH"]
+    if app.config["TESTING"] and (not config or "PAPER_LEDGER_PATH" not in config):
+        ledger_path = None
+    ledger = PaperLedger(ledger_path)
     paper_sessions: dict[str, float] = {}
     allowed_symbols = {item["symbol"] for item in result["signals"]}
     app.extensions["tourism_result"] = result
@@ -176,6 +185,35 @@ def create_app(config: dict[str, Any] | None = None) -> Flask:
     def prices_health():
         return jsonify(_price_health_payload())
 
+    @app.post("/api/v1/research/tourism/run")
+    def run_tourism_research_endpoint():
+        payload = request.get_json(silent=True) or {}
+        if not isinstance(payload, dict):
+            return jsonify({"error": "JSON object required"}), 400
+        try:
+            report = run_tourism_research(
+                app.extensions["vintage_store"],
+                app.extensions["price_store"],
+                app.extensions["research_run_store"],
+                min_events=payload.get("min_events", 12),
+                windows=payload.get("windows", (1, 3, 5, 20)),
+                cost_bps=payload.get("cost_bps", 20.0),
+                execution_lag_sessions=payload.get("execution_lag_sessions", 1),
+            )
+        except (ResearchRunError, TypeError, ValueError) as exc:
+            return jsonify({"error": str(exc)}), 400
+        return jsonify(report)
+
+    @app.get("/api/v1/research/tourism/latest")
+    def latest_tourism_research():
+        try:
+            report = app.extensions["research_run_store"].latest()
+        except FileNotFoundError:
+            return jsonify({"error": "no research runs"}), 404
+        except ResearchRunError as exc:
+            return jsonify({"error": str(exc)}), 422
+        return jsonify(report)
+
     @app.get("/api/v1/data-health")
     def data_health():
         current = app.extensions["tourism_result"]

backend/app/event_study.py

@@ -24,12 +24,30 @@ def _parse_date(value: str) -> date:
         raise EventStudyError("date must be ISO-8601") from exc
 
 
+def _canonical_timestamp(value: str) -> str:
+    try:
+        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except (TypeError, ValueError):
+        return value
+    if parsed.tzinfo is None:
+        parsed = parsed.replace(tzinfo=timezone.utc)
+    return parsed.astimezone(timezone.utc).isoformat()
+
+
 def assess_backtest_readiness(vintages: Sequence[Mapping[str, Any]], min_events: int = 12) -> dict[str, Any]:
-    if min_events < 1:
+    if isinstance(min_events, bool) or min_events < 1:
         raise EventStudyError("min_events must be positive")
-    ids = {str(item.get("vintage_id") or item.get("event_id") or "") for item in vintages}
-    ids.discard("")
-    available = len(ids)
+    independent_keys: set[tuple[str, str, str]] = set()
+    for item in vintages:
+        source_id = str(item.get("source_id") or "")
+        published_at = _canonical_timestamp(str(item.get("published_at") or ""))
+        if source_id and published_at:
+            independent_keys.add(("release", source_id, published_at))
+            continue
+        identifier = str(item.get("vintage_id") or item.get("event_id") or "")
+        if identifier:
+            independent_keys.add(("id", identifier, ""))
+    available = len(independent_keys)
     if available < min_events:
         return {
             "status": "blocked",
@@ -63,12 +81,18 @@ def _price_map(symbol: str, rows: Sequence[Mapping[str, Any]]) -> dict[date, flo
     return dict(sorted(values.items()))
 
 
-def _window_return(series: dict[date, float], event_date: date, window: int, symbol: str) -> float:
+def _window_return(series: dict[date, float], event_date: date, window: int, symbol: str, execution_lag_sessions: int) -> float:
     dates = list(series)
     anchor_candidates = [index for index, trading_day in enumerate(dates) if trading_day >= event_date]
     if not anchor_candidates:
         raise EventStudyError(f"missing post-event prices for {symbol}")
-    anchor = anchor_candidates[0]
+    first_session = anchor_candidates[0]
+    if execution_lag_sessions == 0:
+        anchor = first_session
+    elif dates[first_session] > event_date:
+        anchor = first_session + execution_lag_sessions - 1
+    else:
+        anchor = first_session + execution_lag_sessions
     end = anchor + window
     if end >= len(dates):
         raise EventStudyError(f"insufficient price history for {symbol} window {window}")
@@ -83,6 +107,7 @@ def run_event_study(
     windows: Sequence[int] = (1, 3, 5, 20),
     cost_bps: float = 0.0,
     min_events: int = 12,
+    execution_lag_sessions: int = 1,
 ) -> dict[str, Any]:
     """Calculate weighted post-publication returns from point-in-time events."""
 
@@ -93,6 +118,9 @@ def run_event_study(
         )
     if not windows or any(isinstance(window, bool) or int(window) != window or window <= 0 for window in windows):
         raise EventStudyError("windows must contain positive integers")
+    if isinstance(execution_lag_sessions, bool) or int(execution_lag_sessions) != execution_lag_sessions or execution_lag_sessions < 0:
+        raise EventStudyError("execution_lag_sessions must be a non-negative integer")
+    execution_lag_sessions = int(execution_lag_sessions)
     try:
         cost_bps = float(cost_bps)
     except (TypeError, ValueError) as exc:
@@ -133,13 +161,13 @@ def run_event_study(
                     raise EventStudyError(f"invalid target weight for {symbol}")
                 if symbol not in normalized_prices:
                     raise EventStudyError(f"missing prices for {symbol}")
-                gross += weight * _window_return(normalized_prices[symbol], event["event_date"], int(window), symbol)
+                gross += weight * _window_return(normalized_prices[symbol], event["event_date"], int(window), symbol, execution_lag_sessions)
                 turnover += abs(weight)
             cost = turnover * cost_bps / 10000.0
             gross_returns.append(gross)
             net_returns.append(gross - cost)
             if normalized_benchmark is not None:
-                benchmark_returns.append(_window_return(normalized_benchmark, event["event_date"], int(window), "benchmark"))
+                benchmark_returns.append(_window_return(normalized_benchmark, event["event_date"], int(window), "benchmark", execution_lag_sessions))
         average_gross = fmean(gross_returns)
         average_net = fmean(net_returns)
         average_benchmark = fmean(benchmark_returns) if benchmark_returns else None
@@ -152,6 +180,7 @@ def run_event_study(
             "active_return": round(average_net - average_benchmark, 8) if average_benchmark is not None else None,
             "hit_rate": round(sum(value > 0 for value in net_returns) / len(net_returns), 8),
             "cost_bps": cost_bps,
+            "execution_lag_sessions": execution_lag_sessions,
             "event_returns": [round(value, 8) for value in net_returns],
         }
 

backend/app/paper.py

@@ -2,14 +2,48 @@
 
 from __future__ import annotations
 
+import json
+from copy import deepcopy
 from datetime import datetime, timezone
 from math import isfinite
+from pathlib import Path
 from uuid import uuid4
 
+PAPER_LEDGER_SCHEMA_VERSION = 1
+
+
+class PaperLedgerError(ValueError):
+    """Raised when a persistent paper ledger is invalid or unreadable."""
+
 
 class PaperLedger:
-    def __init__(self) -> None:
+    def __init__(self, path: Path | str | None = None) -> None:
+        self.path = Path(path).resolve() if path else None
         self._entries: list[dict] = []
+        if self.path and self.path.is_file():
+            self._entries = self._load()
+
+    def _load(self) -> list[dict]:
+        if self.path is None:
+            raise PaperLedgerError("paper ledger path is not configured")
+        try:
+            payload = json.loads(self.path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError) as exc:
+            raise PaperLedgerError("paper ledger is unreadable") from exc
+        if not isinstance(payload, dict) or payload.get("schema_version") != PAPER_LEDGER_SCHEMA_VERSION or not isinstance(payload.get("entries"), list):
+            raise PaperLedgerError("unsupported paper ledger schema")
+        if not all(isinstance(entry, dict) for entry in payload["entries"]):
+            raise PaperLedgerError("paper ledger entries are invalid")
+        return deepcopy(payload["entries"])
+
+    def _persist(self, entries: list[dict]) -> None:
+        if not self.path:
+            return
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        payload = {"schema_version": PAPER_LEDGER_SCHEMA_VERSION, "entries": entries}
+        temporary = self.path.with_name(f".{self.path.name}.tmp")
+        temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+        temporary.replace(self.path)
 
     def record(self, payload: dict, allowed_symbols: set[str]) -> dict:
         symbol = str(payload.get("symbol", "")).strip().upper()
@@ -34,8 +68,10 @@ class PaperLedger:
             "assumed_price": assumed_price,
             "status": "PAPER_RECORDED",
         }
-        self._entries.append(entry)
-        return entry
+        candidate_entries = [*self._entries, entry]
+        self._persist(candidate_entries)
+        self._entries = candidate_entries
+        return deepcopy(entry)
 
     def entries(self) -> list[dict]:
-        return list(self._entries)
+        return deepcopy(self._entries)

backend/app/prices.py

@@ -230,6 +230,55 @@ class PriceSnapshotStore:
             raise PriceSourceError("unsupported price manifest schema")
         return manifest
 
+    def _snapshot_path(self, snapshot_id: str) -> Path:
+        if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}", snapshot_id):
+            raise PriceSourceError("invalid price snapshot id")
+        candidate = (self.snapshot_dir / f"{snapshot_id}.json").resolve()
+        if candidate.parent != self.snapshot_dir.resolve():
+            raise PriceSourceError("price snapshot path escaped store root")
+        return candidate
+
+    def load_snapshot(self, snapshot_id: str) -> dict[str, Any]:
+        snapshot_path = self._snapshot_path(snapshot_id)
+        if not snapshot_path.is_file():
+            raise FileNotFoundError(snapshot_id)
+        try:
+            snapshot = json.loads(snapshot_path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError) as exc:
+            raise PriceSourceError("price snapshot is unreadable") from exc
+        source = snapshot.get("source")
+        if not isinstance(source, dict) or source.get("snapshot_id") != snapshot_id:
+            raise PriceSourceError("price snapshot identity mismatch")
+        raw_files = source.get("raw_payload_files")
+        if not isinstance(raw_files, dict) or not raw_files:
+            raise PriceSourceError("price snapshot raw payload manifest is missing")
+        raw_payloads: dict[str, bytes] = {}
+        raw_root = (self.raw_dir / snapshot_id).resolve()
+        if raw_root.parent != self.raw_dir.resolve():
+            raise PriceSourceError("price raw path escaped store root")
+        for provider_symbol, filename in raw_files.items():
+            if _raw_filename(str(provider_symbol)) != filename:
+                raise PriceSourceError("price raw filename mismatch")
+            raw_path = (raw_root / filename).resolve()
+            if raw_path.parent != raw_root or not raw_path.is_file():
+                raise PriceSourceError("price raw payload is missing")
+            raw_payloads[str(provider_symbol)] = raw_path.read_bytes()
+        actual_hash = _combined_hash(raw_payloads)
+        if actual_hash != source.get("raw_payload_hash"):
+            raise PriceSourceError("price raw payload hash mismatch")
+        manifest_entry = self.load_manifest().get("snapshots", {}).get(snapshot_id)
+        if isinstance(manifest_entry, dict) and manifest_entry.get("raw_payload_hash") != actual_hash:
+            raise PriceSourceError("price manifest hash mismatch")
+        return snapshot
+
+    def list_snapshots(self) -> list[dict[str, Any]]:
+        entries = list(self.load_manifest().get("snapshots", {}).values())
+        return sorted(entries, key=lambda item: (str(item.get("retrieved_at", "")), str(item.get("snapshot_id", ""))))
+
+    def latest_snapshot_entry(self) -> dict[str, Any] | None:
+        entries = self.list_snapshots()
+        return entries[-1] if entries else None
+
     def persist(self, snapshot: dict[str, Any], raw_payloads: Mapping[str, bytes]) -> dict[str, Any]:
         source = snapshot.get("source")
         if not isinstance(source, dict) or not source.get("snapshot_id"):
@@ -258,8 +307,11 @@ class PriceSnapshotStore:
             "period_start": stored_source.get("period_start"),
             "period_end": stored_source.get("period_end"),
             "raw_payload_hash": raw_hash,
+            "parser_version": stored_source.get("parser_version"),
             "quality": stored_source.get("quality"),
             "point_in_time": stored_source.get("point_in_time"),
+            "adjusted_prices": stored_source.get("adjusted_prices"),
+            "return_price_field": stored_source.get("return_price_field"),
             "symbols": sorted(stored.get("series", {}).keys()),
             "snapshot_file": stored_source["snapshot_file"],
         }
@@ -299,6 +351,7 @@ def collect_price_snapshot(
         "quality": "revised_vendor_history",
         "point_in_time": False,
         "adjusted_prices": True,
+        "return_price_field": "close",
         "snapshot_id": snapshot_id,
         "bar_counts": {symbol: len(series["bars"]) for symbol, series in normalized.items()},
     }

backend/app/research.py

@@ -0,0 +1,296 @@
+"""Frozen, replayable Tourism research runs.
+
+A run either produces a deterministic event-study result from point-in-time
+inputs or persists a blocked report explaining the exact missing gate. It never
+falls back to revised vendor history.
+"""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+import math
+import re
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Mapping, Sequence
+
+from .event_study import EventStudyError, _canonical_timestamp, assess_backtest_readiness, run_event_study
+from .prices import PriceSnapshotStore, PriceSourceError
+from .tourism import compute_tourism_signal
+from .vintages import VintageStore, VintageStoreError
+
+RUN_SCHEMA_VERSION = 1
+_RUN_ID_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}")
+
+
+class ResearchRunError(ValueError):
+    """Raised when a research run cannot be safely created or replayed."""
+
+
+def _canonical_hash(payload: Mapping[str, Any]) -> str:
+    encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def _atomic_write(path: Path, payload: Mapping[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_name(f".{path.name}.tmp")
+    temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    temporary.replace(path)
+
+
+class ResearchRunStore:
+    """Immutable JSON reports with a compact run manifest."""
+
+    def __init__(self, root: Path | str) -> None:
+        self.root = Path(root).resolve()
+        self.report_dir = self.root / "reports"
+        self.manifest_path = self.root / "manifest.json"
+
+    def load_manifest(self) -> dict[str, Any]:
+        if not self.manifest_path.is_file():
+            return {"schema_version": RUN_SCHEMA_VERSION, "runs": {}}
+        try:
+            manifest = json.loads(self.manifest_path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError) as exc:
+            raise ResearchRunError("research run manifest is unreadable") from exc
+        if manifest.get("schema_version") != RUN_SCHEMA_VERSION or not isinstance(manifest.get("runs"), dict):
+            raise ResearchRunError("unsupported research run manifest schema")
+        return manifest
+
+    def _report_path(self, run_id: str) -> Path:
+        if not _RUN_ID_RE.fullmatch(run_id):
+            raise ResearchRunError("invalid research run id")
+        candidate = (self.report_dir / f"{run_id}.json").resolve()
+        if candidate.parent != self.report_dir.resolve():
+            raise ResearchRunError("research report path escaped store root")
+        return candidate
+
+    def load(self, run_id: str) -> dict[str, Any]:
+        path = self._report_path(run_id)
+        if not path.is_file():
+            raise FileNotFoundError(run_id)
+        try:
+            report = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError) as exc:
+            raise ResearchRunError("research report is unreadable") from exc
+        if report.get("run_id") != run_id or report.get("schema_version") != RUN_SCHEMA_VERSION:
+            raise ResearchRunError("research report identity mismatch")
+        return report
+
+    def persist(self, report: dict[str, Any]) -> dict[str, Any]:
+        run_id = str(report.get("run_id", ""))
+        path = self._report_path(run_id)
+        manifest = self.load_manifest()
+        existing = manifest["runs"].get(run_id)
+        if existing:
+            stored = self.load(run_id)
+            if stored != report:
+                raise ResearchRunError("research run id already contains a different report")
+            return stored
+        stored = copy.deepcopy(report)
+        _atomic_write(path, stored)
+        manifest["runs"][run_id] = {
+            "run_id": run_id,
+            "status": stored.get("status"),
+            "reason": stored.get("reason"),
+            "generated_at": stored.get("generated_at"),
+            "report_file": path.name,
+        }
+        _atomic_write(self.manifest_path, manifest)
+        return stored
+
+    def latest(self) -> dict[str, Any]:
+        entries = list(self.load_manifest().get("runs", {}).values())
+        if not entries:
+            raise FileNotFoundError("no research runs")
+        latest = max(entries, key=lambda item: (str(item.get("generated_at", "")), str(item.get("run_id", ""))))
+        return self.load(str(latest["run_id"]))
+
+    def list_runs(self) -> list[dict[str, Any]]:
+        entries = list(self.load_manifest().get("runs", {}).values())
+        return sorted(entries, key=lambda item: (str(item.get("generated_at", "")), str(item.get("run_id", ""))), reverse=True)
+
+
+def _validate_config(windows: Sequence[int], cost_bps: float, min_events: int, execution_lag_sessions: int) -> tuple[list[int], float, int, int]:
+    if not windows or any(isinstance(window, bool) or int(window) != window or int(window) <= 0 for window in windows):
+        raise ResearchRunError("windows must contain positive integers")
+    try:
+        parsed_cost = float(cost_bps)
+    except (TypeError, ValueError) as exc:
+        raise ResearchRunError("cost_bps must be finite and non-negative") from exc
+    if not math.isfinite(parsed_cost) or parsed_cost < 0:
+        raise ResearchRunError("cost_bps must be finite and non-negative")
+    if isinstance(min_events, bool) or int(min_events) != min_events or int(min_events) < 1:
+        raise ResearchRunError("min_events must be positive")
+    if isinstance(execution_lag_sessions, bool) or int(execution_lag_sessions) != execution_lag_sessions or int(execution_lag_sessions) < 0:
+        raise ResearchRunError("execution_lag_sessions must be a non-negative integer")
+    return [int(window) for window in windows], parsed_cost, int(min_events), int(execution_lag_sessions)
+
+
+def _price_gate(price_entry: Mapping[str, Any] | None) -> dict[str, Any]:
+    if not price_entry:
+        return {"status": "blocked", "reason": "price_snapshot_missing", "snapshot_count": 0}
+    if not bool(price_entry.get("point_in_time")):
+        return {
+            "status": "blocked",
+            "reason": "price_series_not_point_in_time",
+            "snapshot_id": price_entry.get("snapshot_id"),
+            "quality": price_entry.get("quality"),
+            "point_in_time": False,
+        }
+    return {
+        "status": "ready",
+        "reason": None,
+        "snapshot_id": price_entry.get("snapshot_id"),
+        "quality": price_entry.get("quality"),
+        "point_in_time": True,
+    }
+
+
+def _revision_key(entry: Mapping[str, Any]) -> tuple[str, str, str]:
+    source_id = str(entry.get("source_id") or "")
+    published_at = _canonical_timestamp(str(entry.get("published_at") or ""))
+    if source_id and published_at:
+        return ("release", source_id, published_at)
+    return ("id", str(entry.get("vintage_id") or ""), "")
+
+
+def _select_independent_vintages(entries: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]:
+    selected: dict[tuple[str, str, str], dict[str, Any]] = {}
+    for raw_entry in entries:
+        entry = dict(raw_entry)
+        key = _revision_key(entry)
+        current = selected.get(key)
+        candidate_rank = (str(entry.get("first_seen_at", "")), str(entry.get("last_seen_at", "")), str(entry.get("vintage_id", "")))
+        current_rank = (str(current.get("first_seen_at", "")), str(current.get("last_seen_at", "")), str(current.get("vintage_id", ""))) if current else ("", "", "")
+        if current is None or candidate_rank > current_rank:
+            selected[key] = entry
+    return sorted(selected.values(), key=lambda item: (str(item.get("published_at", "")), str(item.get("vintage_id", ""))))
+
+
+def _series_rows(series: Mapping[str, Any], price_field: str, symbol: str) -> list[dict[str, Any]]:
+    bars = series.get("bars")
+    if not isinstance(bars, list) or not bars:
+        raise ResearchRunError(f"missing price bars for {symbol}")
+    rows: list[dict[str, Any]] = []
+    for bar in bars:
+        if not isinstance(bar, dict) or price_field not in bar:
+            raise ResearchRunError(f"price field {price_field} missing for {symbol}")
+        rows.append({"date": bar.get("date"), "close": bar.get(price_field)})
+    return rows
+
+
+def run_tourism_research(
+    vintage_store: VintageStore,
+    price_store: PriceSnapshotStore,
+    run_store: ResearchRunStore,
+    *,
+    min_events: int = 12,
+    windows: Sequence[int] = (1, 3, 5, 20),
+    cost_bps: float = 20.0,
+    execution_lag_sessions: int = 1,
+) -> dict[str, Any]:
+    windows, cost_bps, min_events, execution_lag_sessions = _validate_config(windows, cost_bps, min_events, execution_lag_sessions)
+    try:
+        vintages = _select_independent_vintages(vintage_store.list_vintages())
+        price_entry = price_store.latest_snapshot_entry()
+    except (VintageStoreError, PriceSourceError) as exc:
+        raise ResearchRunError(str(exc)) from exc
+    vintage_gate = assess_backtest_readiness(vintages, min_events=min_events)
+    price_gate = _price_gate(price_entry)
+    input_vintages = [
+        {
+            "vintage_id": entry.get("vintage_id"),
+            "source_id": entry.get("source_id"),
+            "as_of": entry.get("as_of"),
+            "published_at": entry.get("published_at"),
+            "raw_payload_hash": entry.get("raw_payload_hash"),
+            "parser_version": entry.get("parser_version"),
+            "revision_status": entry.get("revision_status"),
+        }
+        for entry in vintages
+    ]
+    input_price = copy.deepcopy(price_entry) if price_entry else None
+    config = {
+        "min_events": min_events,
+        "windows": windows,
+        "cost_bps": cost_bps,
+        "execution_lag_sessions": execution_lag_sessions,
+    }
+    input_fingerprint = _canonical_hash({"theme": "tourism", "config": config, "vintages": input_vintages, "price_snapshot": input_price})
+    run_id = f"tourism-run-{input_fingerprint[:16]}"
+    try:
+        return run_store.load(run_id)
+    except FileNotFoundError:
+        pass
+    generated_at = datetime.now(timezone.utc).isoformat()
+    report: dict[str, Any] = {
+        "schema_version": RUN_SCHEMA_VERSION,
+        "run_id": run_id,
+        "theme": "tourism",
+        "status": "blocked",
+        "reason": None,
+        "generated_at": generated_at,
+        "config": config,
+        "gates": {"vintages": vintage_gate, "prices": price_gate},
+        "inputs": {"vintages": input_vintages, "price_snapshot": input_price},
+    }
+    if vintage_gate["status"] != "ready":
+        report["reason"] = vintage_gate["reason"]
+        return run_store.persist(report)
+    if price_gate["status"] != "ready":
+        report["reason"] = price_gate["reason"]
+        return run_store.persist(report)
+    if not price_entry:
+        raise ResearchRunError("price snapshot disappeared after readiness check")
+    try:
+        price_snapshot = price_store.load_snapshot(str(price_entry["snapshot_id"]))
+    except (FileNotFoundError, PriceSourceError) as exc:
+        report["reason"] = "price_snapshot_unreadable"
+        report["gates"]["prices"] = {**price_gate, "status": "blocked", "reason": report["reason"], "error": str(exc)}
+        return run_store.persist(report)
+    price_source = price_snapshot.get("source", {})
+    price_field = str(price_source.get("return_price_field", "close"))
+    series = price_snapshot.get("series")
+    if not isinstance(series, dict):
+        report["reason"] = "price_snapshot_invalid"
+        return run_store.persist(report)
+    events: list[dict[str, Any]] = []
+    try:
+        for entry in vintages:
+            snapshot = vintage_store.load_snapshot(str(entry["vintage_id"]))
+            result = compute_tourism_signal(snapshot)
+            events.append({
+                "event_id": entry["vintage_id"],
+                "published_at": snapshot["source"]["published_at"],
+                "signals": result["signals"],
+            })
+        price_rows = {symbol: _series_rows(value, price_field, symbol) for symbol, value in series.items() if symbol != price_snapshot.get("benchmark_symbol")}
+        benchmark_symbol = price_snapshot.get("benchmark_symbol")
+        benchmark_rows = _series_rows(series[benchmark_symbol], price_field, str(benchmark_symbol)) if benchmark_symbol else None
+        result = run_event_study(
+            events,
+            price_rows,
+            benchmark_prices=benchmark_rows,
+            windows=windows,
+            cost_bps=cost_bps,
+            min_events=min_events,
+            execution_lag_sessions=execution_lag_sessions,
+        )
+    except (FileNotFoundError, VintageStoreError, EventStudyError, ResearchRunError, KeyError, TypeError, ValueError) as exc:
+        report["reason"] = "research_inputs_invalid"
+        report["error"] = str(exc)
+        return run_store.persist(report)
+    report["status"] = "ready"
+    report["reason"] = None
+    report["result"] = result
+    report["input_summary"] = {
+        "event_count": len(events),
+        "price_symbols": sorted(price_rows),
+        "benchmark_symbol": benchmark_symbol,
+        "return_price_field": price_field,
+    }
+    return run_store.persist(report)

backend/app/vintages.py

@@ -116,7 +116,7 @@ class VintageStore:
         existing = manifest["vintages"].get(vintage_id)
         same_release = any(
             item.get("source_id") == source.get("source_id")
-            and item.get("published_at") == published_at
+            and _parse_timestamp(str(item.get("published_at"))) == _parse_timestamp(published_at)
             and item.get("vintage_id") != vintage_id
             for item in manifest["vintages"].values()
         )