Refresh July 30 horizon dashboard

site/data/manifest.json

@@ -1,14 +1,14 @@
 {
   "schema_version": 3,
-  "generated_at_utc": "2026-07-30T00:29:41.454374+00:00",
+  "generated_at_utc": "2026-07-30T21:12:34.167009+00:00",
   "latest_decision_date": "2026-07-28",
   "latest_monitor_date": "2026-07-29",
   "snapshot_count": 17,
   "evaluation_count": 14,
   "dashboard_sha256": "786388e9b2b040a8e0e39ad8dc46ecc44d3067b3095281a334da56467729ca9b",
   "dashboard_bytes": 586173,
-  "horizon_dashboard_sha256": "20d6bdaa4d619ecca77441801a4f2e63b93c665822979b278cb490999bd4695a",
-  "horizon_dashboard_bytes": 66861,
-  "horizon_forecast_arms": 6,
+  "horizon_dashboard_sha256": "6388d818bd0e7a103662231c17259e119126864b8fd318e70c96148ec56495cb",
+  "horizon_dashboard_bytes": 128840,
+  "horizon_forecast_arms": 12,
   "source_policy": "sanitized_live_forward_outputs_only"
 }

src/experiments/live_forward_20d.py

@@ -4,6 +4,7 @@
 
 import hashlib
 import json
+from copy import deepcopy
 from datetime import UTC
 from pathlib import Path
 from typing import Any, Iterable
@@ -90,6 +91,95 @@ def _payload_sha256(payload: dict[str, Any]) -> str:
     return hashlib.sha256(encoded).hexdigest()
 
 
+def _canonical_payload_sha256(payload: dict[str, Any]) -> str:
+    def string_keys(value: Any) -> Any:
+        if isinstance(value, dict):
+            return {
+                str(key): string_keys(local)
+                for key, local in value.items()
+            }
+        if isinstance(value, (list, tuple)):
+            return [string_keys(local) for local in value]
+        return value
+
+    return _payload_sha256(string_keys(payload))
+
+
+def _training_data_revision(config: dict[str, Any]) -> int:
+    return int(config.get("live_forward_20d", {}).get("training_data_revision", 1))
+
+
+def _experiment_design_payload(config: dict[str, Any]) -> dict[str, Any]:
+    """Remove the operational data revision from the frozen experiment design."""
+    design = deepcopy(config)
+    design.get("live_forward_20d", {}).pop("training_data_revision", None)
+    return design
+
+
+def _ensure_experiment_manifest(path: Path, config: dict[str, Any]) -> None:
+    """Freeze design while allowing an audited monotonic training-data repair."""
+    config_sha256 = _payload_sha256(config)
+    if not path.exists():
+        _write_json(
+            path,
+            {
+                "status": "frozen_paper_live_experiment",
+                "experiment_version": config.get("experiment", {}).get(
+                    "id",
+                    "live_forward_20d_v1",
+                ),
+                "config_sha256": config_sha256,
+                "design_sha256": _canonical_payload_sha256(
+                    _experiment_design_payload(config)
+                ),
+                "training_data_revision": _training_data_revision(config),
+                "config": config,
+                "config_history": [],
+                "created_utc": pd.Timestamp.now(tz=UTC).isoformat(),
+            },
+        )
+        return
+
+    frozen = json.loads(path.read_text(encoding="utf-8"))
+    if frozen.get("config_sha256") == config_sha256:
+        return
+    old_config = frozen.get("config", {})
+    old_design_sha = _canonical_payload_sha256(
+        _experiment_design_payload(old_config)
+    )
+    new_design_sha = _canonical_payload_sha256(
+        _experiment_design_payload(config)
+    )
+    old_revision = _training_data_revision(old_config)
+    new_revision = _training_data_revision(config)
+    if old_design_sha != new_design_sha or new_revision <= old_revision:
+        raise RuntimeError(
+            "Output root already contains a different frozen experiment config"
+        )
+
+    history = list(frozen.get("config_history", []))
+    history.append(
+        {
+            "config_sha256": frozen.get("config_sha256"),
+            "training_data_revision": old_revision,
+            "config": old_config,
+            "superseded_utc": pd.Timestamp.now(tz=UTC).isoformat(),
+            "reason": "training_data_revision_advanced",
+        }
+    )
+    frozen.update(
+        {
+            "config_sha256": config_sha256,
+            "design_sha256": new_design_sha,
+            "training_data_revision": new_revision,
+            "config": config,
+            "config_history": history,
+            "updated_utc": pd.Timestamp.now(tz=UTC).isoformat(),
+        }
+    )
+    _write_json(path, frozen)
+
+
 def build_timestamp_safe_nlp_features(
     posts: list[Any],
     price_dates: pd.DatetimeIndex,
@@ -1751,6 +1841,99 @@ def evaluate_horizon_matched_strategies(
     return result
 
 
+def build_strategy_interim_marks(
+    output_root: Path,
+    prices: pd.DataFrame,
+    as_of: pd.Timestamp,
+) -> pd.DataFrame:
+    """Mark each currently entered strategy at the latest official close."""
+    price_dates = pd.DatetimeIndex(
+        prices.index.get_level_values("date").unique()
+    ).sort_values()
+    as_of_pos = price_dates.get_indexer([as_of])[0]
+    if as_of_pos < 0:
+        return pd.DataFrame()
+    opened = prices["open"].unstack("ticker")
+    close_column = "adj_close" if "adj_close" in prices else "close"
+    closed = prices[close_column].unstack("ticker")
+    candidates: dict[
+        tuple[str, int],
+        list[tuple[pd.Timestamp, Path, dict[str, Any]]],
+    ] = {}
+    for manifest_path in output_root.glob("forecasts/*/*/manifest.json"):
+        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+        if not manifest.get("strategy_rebalance", False):
+            continue
+        target_path = manifest_path.parent / "strategy_rebalance_target.csv"
+        if not target_path.exists():
+            continue
+        decision = pd.Timestamp(manifest["decision_date"])
+        decision_pos = price_dates.get_indexer([decision])[0]
+        if decision_pos < 0 or decision_pos + 1 > as_of_pos:
+            continue
+        key = (str(manifest["family"]), int(manifest["horizon_sessions"]))
+        candidates.setdefault(key, []).append((decision, target_path, manifest))
+
+    rows: list[dict[str, Any]] = []
+    for (family, horizon), items in candidates.items():
+        decision, target_path, manifest = max(items, key=lambda item: item[0])
+        decision_pos = price_dates.get_indexer([decision])[0]
+        entry_pos = decision_pos + 1
+        entry_date = price_dates[entry_pos]
+        target = pd.read_csv(target_path)
+        if not {"ticker", "weight"}.issubset(target.columns):
+            raise ValueError(f"Invalid strategy target schema: {target_path}")
+        weights = target.set_index(target["ticker"].astype(str))["weight"].astype(
+            float
+        )
+        entry = opened.loc[entry_date].reindex(weights.index)
+        mark = closed.loc[as_of].reindex(weights.index)
+        usable = (
+            weights.notna()
+            & weights.gt(0)
+            & entry.notna()
+            & entry.gt(0)
+            & mark.notna()
+            & mark.gt(0)
+        )
+        if not usable.all():
+            missing = weights.index[~usable].astype(str).tolist()
+            raise RuntimeError(
+                f"Invalid interim mark prices or weights for {family} h{horizon} "
+                f"vintage {decision.date()}: {', '.join(missing[:10])}"
+            )
+        weights = weights / weights.sum()
+        gross_mark = float(((mark / entry - 1.0) * weights).sum())
+        spy_mark = float("nan")
+        if "SPY" in opened.columns and "SPY" in closed.columns:
+            spy_mark = float(
+                closed.loc[as_of, "SPY"] / opened.loc[entry_date, "SPY"] - 1.0
+            )
+        rows.append(
+            {
+                "experiment_version": manifest["experiment_version"],
+                "family": family,
+                "horizon_sessions": horizon,
+                "rebalance_decision_date": str(decision.date()),
+                "entry_date": str(entry_date.date()),
+                "mark_date": str(as_of.date()),
+                "completed_open_to_open_sessions": int(as_of_pos - entry_pos),
+                "gross_mark_to_close": gross_mark,
+                "spy_mark_to_close": spy_mark,
+                "active_mark_vs_spy": float(gross_mark - spy_mark),
+                "initial_holdings": int(len(weights)),
+                "status": "interim_official_close_mark",
+            }
+        )
+    result = pd.DataFrame(rows)
+    if not result.empty:
+        result = result.sort_values(
+            ["family", "horizon_sessions"]
+        ).reset_index(drop=True)
+        result.to_csv(output_root / "strategy_interim_marks.csv", index=False)
+    return result
+
+
 def build_live_20d_dashboard(
     output_root: Path,
     run_summary: dict[str, Any],
@@ -1812,29 +1995,7 @@ def run_live_forward_20d(
             "holdings": int(settings["holdings"]),
         }
     experiment_manifest_path = output / "experiment_manifest.json"
-    config_sha256 = _payload_sha256(config)
-    if experiment_manifest_path.exists():
-        frozen_experiment = json.loads(
-            experiment_manifest_path.read_text(encoding="utf-8")
-        )
-        if frozen_experiment.get("config_sha256") != config_sha256:
-            raise RuntimeError(
-                "Output root already contains a different frozen experiment config"
-            )
-    else:
-        _write_json(
-            experiment_manifest_path,
-            {
-                "status": "frozen_paper_live_experiment",
-                "experiment_version": config.get("experiment", {}).get(
-                    "id",
-                    "live_forward_20d_v1",
-                ),
-                "config_sha256": config_sha256,
-                "config": config,
-                "created_utc": pd.Timestamp.now(tz=UTC).isoformat(),
-            },
-        )
+    _ensure_experiment_manifest(experiment_manifest_path, config)
     universe_cfg = settings["universe"]
     eligible_features, eligibility = prepare_eligible_panel(
         data.loc[data.index.get_level_values("date") <= decision_date],
@@ -2056,6 +2217,11 @@ def run_live_forward_20d(
         price_data,
         decision_date,
     )
+    strategy_interim_marks = build_strategy_interim_marks(
+        output,
+        price_data,
+        decision_date,
+    )
     strategy_performance: list[dict[str, Any]] = []
     if not strategy_returns.empty:
         for (family, horizon), group in strategy_returns.groupby(
@@ -2131,12 +2297,18 @@ def run_live_forward_20d(
         ) if not tactical.empty else 0,
         "matured_forecasts": int(len(matured)),
         "strategy_daily_return_rows": int(len(strategy_returns)),
+        "strategy_interim_mark_rows": int(len(strategy_interim_marks)),
         "strategy_performance": strategy_performance,
         "strategy_returns_output": (
             str(output / "strategy_daily_returns.csv")
             if not strategy_returns.empty
             else None
         ),
+        "strategy_interim_marks_output": (
+            str(output / "strategy_interim_marks.csv")
+            if not strategy_interim_marks.empty
+            else None
+        ),
         "output": str(daily_dir),
     }
     _write_json(daily_dir / "run_summary.json", summary)