Upload runtime.py

runtime.py

@@ -40,10 +40,11 @@ NIFTY_1D_PATH = DATA_DIR / "nifty50_1d.parquet"
 MODEL_PATH = MODEL_DIR / "nifty_opening_direction_model.joblib"
 LATEST_PATH = MODEL_DIR / "latest_prediction.csv"
 TEST_PREDICTIONS_PATH = DATA_DIR / "test_predictions.parquet"
-TOMORROW_MODEL_PATH = MODEL_DIR / "nifty_tomorrow_direction_model.joblib"
-TOMORROW_LATEST_PATH = MODEL_DIR / "tomorrow_latest_prediction.csv"
-TOMORROW_SUMMARY_PATH = MODEL_DIR / "tomorrow_summary.json"
+TOMORROW_MODEL_PATH = MODEL_DIR / "nifty_tomorrow_direction_model.joblib"
+TOMORROW_LATEST_PATH = MODEL_DIR / "tomorrow_latest_prediction.csv"
+TOMORROW_SUMMARY_PATH = MODEL_DIR / "tomorrow_summary.json"
 TOMORROW_TEST_PREDICTIONS_PATH = DATA_DIR / "tomorrow_test_predictions.parquet"
+TOMORROW_PREDICTION_HISTORY_PATH = MODEL_DIR / "tomorrow_prediction_history.parquet"
 FORECASTING_PROJECT_ROOT = Path(
     os.environ.get(
         "FORECASTING_PROJECT_ROOT",
@@ -55,14 +56,16 @@ DAILY_FORECASTER_SUMMARY_PATH = DAILY_FORECASTER_OUTPUT_DIR / "forecaster_summar
 DAILY_FORECASTER_LATEST_PATH = DAILY_FORECASTER_OUTPUT_DIR / "forecaster_latest.csv"
 DAILY_FORECASTER_PREDICTIONS_PATH = DAILY_FORECASTER_OUTPUT_DIR / "forecaster_test_predictions.csv"
 TPLUS1_MODEL_PATH = MODEL_DIR / "nifty_1420_tplus1_logistic_model.joblib"
-TPLUS1_LATEST_PATH = MODEL_DIR / "tplus1_latest_prediction.csv"
-TPLUS1_SUMMARY_PATH = MODEL_DIR / "tplus1_summary.json"
-TPLUS1_TEST_PREDICTIONS_PATH = DATA_DIR / "tplus1_test_predictions.parquet"
-REFRESH_STATE_PATH = MODEL_DIR / "refresh_state.json"
-REFRESH_WAITING = "waiting_second_payload"
-REFRESH_REFRESHING = "refreshing"
-REFRESH_READY = "ready"
-REFRESH_FAILED = "failed"
+TPLUS1_LATEST_PATH = MODEL_DIR / "tplus1_latest_prediction.csv"
+TPLUS1_SUMMARY_PATH = MODEL_DIR / "tplus1_summary.json"
+TPLUS1_TEST_PREDICTIONS_PATH = DATA_DIR / "tplus1_test_predictions.parquet"
+TPLUS1_PREDICTION_HISTORY_PATH = MODEL_DIR / "tplus1_prediction_history.parquet"
+T5_PREDICTION_HISTORY_PATH = MODEL_DIR / "t5_prediction_history.parquet"
+REFRESH_STATE_PATH = MODEL_DIR / "refresh_state.json"
+REFRESH_WAITING = "waiting_second_payload"
+REFRESH_REFRESHING = "refreshing"
+REFRESH_READY = "ready"
+REFRESH_FAILED = "failed"
 REFRESH_NORMAL = "normal"
 LIVE_ACCURACY_PATH = MODEL_DIR / "live_accuracy.json"
 
@@ -381,19 +384,24 @@ def fetch_yahoo_daily(period: str = "1mo") -> pd.DataFrame:
     return yahoo_history_to_ohlcv(raw, daily=True)
 
 
-def append_parquet_rows(path: Path, new_rows: pd.DataFrame, subset: list[str]) -> pd.DataFrame:
-    if new_rows.empty:
-        if path.exists():
-            return pd.read_parquet(path)
-        raise RuntimeError(f"No rows returned for {path.name}; leaving parquet unchanged.")
+def append_parquet_rows(path: Path, new_rows: pd.DataFrame, subset: list[str]) -> pd.DataFrame:
+    if new_rows.empty:
+        if path.exists():
+            return pd.read_parquet(path)
+        raise RuntimeError(f"No rows returned for {path.name}; leaving parquet unchanged.")
     if path.exists():
         existing = pd.read_parquet(path)
         combined = pd.concat([existing, new_rows], ignore_index=True)
     else:
         combined = new_rows.copy()
-    combined = combined.drop_duplicates(subset=subset, keep="last").sort_values(subset).reset_index(drop=True)
-    combined.to_parquet(path, index=False, compression="zstd")
-    return combined
+    combined = combined.drop_duplicates(subset=subset, keep="last").sort_values(subset).reset_index(drop=True)
+    combined.to_parquet(path, index=False, compression="zstd")
+    return combined
+
+
+def append_prediction_history(path: Path, row: dict[str, Any], subset: list[str]) -> pd.DataFrame:
+    frame = pd.DataFrame([row])
+    return append_parquet_rows(path, frame, subset)
 
 
 def latest_parquet_date(path: Path) -> date | None:
@@ -502,7 +510,7 @@ def build_model_row(first5_row: pd.DataFrame) -> pd.DataFrame:
     return output
 
 
-def predict_row(row: pd.DataFrame) -> Prediction:
+def predict_row(row: pd.DataFrame) -> Prediction:
     payload = load_model()
     model = payload["model"]
     features = payload["features"]
@@ -515,19 +523,28 @@ def predict_row(row: pd.DataFrame) -> Prediction:
     pred = apply_decision_overlays(raw_pred, row, payload.get("decision_overlays", DECISION_OVERLAYS))
     is_overridden = bool(raw_pred[0] != pred[0])
     confidence = directional_confidence(prob_up, pred, threshold)
-    prediction = Prediction(
-        input_date=pd.to_datetime(row["date"].iloc[0]).date().isoformat(),
-        first5_start=str(pd.to_datetime(row["first5_start"].iloc[0])),
-        first5_end=str(pd.to_datetime(row["first5_end"].iloc[0])),
-        prediction="UP" if int(pred[0]) == 1 else "DOWN",
+    prediction = Prediction(
+        input_date=pd.to_datetime(row["date"].iloc[0]).date().isoformat(),
+        first5_start=str(pd.to_datetime(row["first5_start"].iloc[0])),
+        first5_end=str(pd.to_datetime(row["first5_end"].iloc[0])),
+        prediction="UP" if int(pred[0]) == 1 else "DOWN",
         prob_up=float(prob_up[0]),
         confidence=float(confidence[0]),
         threshold=threshold,
-        model_name=str(payload.get("model_name", "nifty_opening_direction_model")),
-        is_overridden=is_overridden,
-    )
-    pd.DataFrame([prediction.to_dict()]).to_csv(LATEST_PATH, index=False)
-    return prediction
+        model_name=str(payload.get("model_name", "nifty_opening_direction_model")),
+        is_overridden=is_overridden,
+    )
+    pd.DataFrame([prediction.to_dict()]).to_csv(LATEST_PATH, index=False)
+    _record_prediction_history(
+        T5_PREDICTION_HISTORY_PATH,
+        {
+            **prediction.to_dict(),
+            "target_date": prediction.input_date,
+            "source": "live",
+        },
+        ["target_date"],
+    )
+    return prediction
 
 
 def _file_cache_key(path: Path) -> tuple[str, int | None, int | None]:
@@ -925,23 +942,25 @@ def refresh_tplus1_prediction(session_date: date | None = None) -> dict[str, Any
     input_day = pd.to_datetime(row["date"].iloc[0]).date()
     target_day = next_trading_day(input_day + timedelta(days=1))
     summary = load_tplus1_summary()
-    out = {
-        "input_date": input_day.isoformat(),
-        "target_date": target_day.isoformat(),
-        "forecast_for": f"next trading session after {input_day.isoformat()}",
+    out = {
+        "input_date": input_day.isoformat(),
+        "target_date": target_day.isoformat(),
+        "forecast_for": f"next trading session after {input_day.isoformat()}",
         "prediction": prediction,
         "prob_up": float(prob_up[0]),
         "confidence": float(max(prob_up[0], 1.0 - prob_up[0])),
         "threshold": threshold,
         "model_name": str(payload.get("model_name", summary.get("model_name", "nifty_1420_tplus1_logistic_model"))),
         "decision_overlay": summary.get("decision_overlay"),
-        "validation_accuracy": summary.get("validation_accuracy"),
-        "test_accuracy": summary.get("test_accuracy"),
-        "accuracy_goal": summary.get("accuracy_goal"),
-    }
-    pd.DataFrame([out]).to_csv(TPLUS1_LATEST_PATH, index=False)
-    clear_dashboard_payload_cache()
-    return out
+        "validation_accuracy": summary.get("validation_accuracy"),
+        "test_accuracy": summary.get("test_accuracy"),
+        "accuracy_goal": summary.get("accuracy_goal"),
+        "source": "live",
+    }
+    pd.DataFrame([out]).to_csv(TPLUS1_LATEST_PATH, index=False)
+    _record_prediction_history(TPLUS1_PREDICTION_HISTORY_PATH, out, ["target_date"])
+    clear_dashboard_payload_cache()
+    return out
 
 
 def _tomorrow_probability_from_daily(daily: pd.DataFrame, fallback_prob: float) -> float:
@@ -971,6 +990,8 @@ def refresh_tomorrow_prediction(session_date: date | None = None) -> dict[str, A
     if synced is not None and TOMORROW_LATEST_PATH.exists():
         latest = pd.read_csv(TOMORROW_LATEST_PATH).iloc[-1].to_dict()
         cleaned = {k: (None if pd.isna(v) else v) for k, v in latest.items()}
+        cleaned["source"] = "live"
+        _record_prediction_history(TOMORROW_PREDICTION_HISTORY_PATH, cleaned, ["target_date"])
         if session_date is None:
             clear_dashboard_payload_cache()
             return cleaned
@@ -995,20 +1016,22 @@ def refresh_tomorrow_prediction(session_date: date | None = None) -> dict[str, A
     prob_up = _tomorrow_probability_from_daily(daily[daily["date"].dt.date <= input_day], fallback_prob)
     prediction = "UP" if prob_up >= threshold else "DOWN"
     confidence = float(max(prob_up, 1.0 - prob_up))
-    row = {
-        "input_date": input_day.isoformat(),
-        "target_date": target_day.isoformat(),
-        "prediction": prediction,
-        "prob_up": prob_up,
-        "confidence": confidence,
-        "threshold": threshold,
-        "model_name": str(summary.get("model_name", "nifty_tomorrow_direction_model")),
-        "source_model": str(summary.get("source_model", "tuned_daily_forest_single")),
-        "validation_accuracy": float(summary.get("validation_accuracy", 0.5780141843971631)),
-        "test_accuracy": float(summary.get("test_accuracy", 0.6182795698924731)),
-    }
-    pd.DataFrame([row]).to_csv(TOMORROW_LATEST_PATH, index=False)
-    summary = dict(summary)
+    row = {
+        "input_date": input_day.isoformat(),
+        "target_date": target_day.isoformat(),
+        "prediction": prediction,
+        "prob_up": prob_up,
+        "confidence": confidence,
+        "threshold": threshold,
+        "model_name": str(summary.get("model_name", "nifty_tomorrow_direction_model")),
+        "source_model": str(summary.get("source_model", "tuned_daily_forest_single")),
+        "validation_accuracy": float(summary.get("validation_accuracy", 0.5780141843971631)),
+        "test_accuracy": float(summary.get("test_accuracy", 0.6182795698924731)),
+        "source": "live",
+    }
+    pd.DataFrame([row]).to_csv(TOMORROW_LATEST_PATH, index=False)
+    _record_prediction_history(TOMORROW_PREDICTION_HISTORY_PATH, row, ["target_date"])
+    summary = dict(summary)
     summary.update(
         {
             "latest_forecast_date": row["input_date"],
@@ -1393,16 +1416,41 @@ def update_opening_outcomes_from_daily() -> dict[str, Any]:
 
 def load_live_accuracy() -> dict[str, Any]:
     """Load the live accuracy ledger from disk."""
+    default = {
+        "tomorrow": {"entries": [], "accuracy": None, "total": 0, "correct_count": 0, "backtest_count": 0, "live_count": 0},
+        "t5": {"entries": [], "accuracy": None, "total": 0, "correct_count": 0, "backtest_count": 0, "live_count": 0},
+        "tplus1": {"entries": [], "accuracy": None, "total": 0, "correct_count": 0, "backtest_count": 0, "live_count": 0},
+    }
+
     if LIVE_ACCURACY_PATH.exists():
         try:
-            return json.loads(LIVE_ACCURACY_PATH.read_text(encoding="utf-8"))
+            raw = json.loads(LIVE_ACCURACY_PATH.read_text(encoding="utf-8"))
         except Exception:
-            pass
-    return {
-        "tomorrow": {"entries": [], "accuracy": None, "total": 0, "correct_count": 0},
-        "t5": {"entries": [], "accuracy": None, "total": 0, "correct_count": 0},
-        "tplus1": {"entries": [], "accuracy": None, "total": 0, "correct_count": 0},
-    }
+            raw = None
+        if isinstance(raw, dict):
+            try:
+                for model_id in default:
+                    current = raw.get(model_id, {})
+                    if not isinstance(current, dict):
+                        current = {}
+                    entries = current.get("entries", [])
+                    if not isinstance(entries, list):
+                        entries = []
+                    backtest_entries = [entry for entry in entries if str(entry.get("source", "backtest")).lower() == "backtest"]
+                    live_entries = [entry for entry in entries if str(entry.get("source", "backtest")).lower() != "backtest"]
+                    total = len(entries)
+                    correct = sum(1 for e in entries if e.get("correct"))
+                    current["entries"] = entries
+                    current["backtest_count"] = int(current.get("backtest_count") or len(backtest_entries))
+                    current["live_count"] = int(current.get("live_count") or len(live_entries))
+                    current["total"] = int(current.get("total") or total)
+                    current["correct_count"] = int(current.get("correct_count") or correct)
+                    current["accuracy"] = (current["correct_count"] / current["total"]) if current["total"] > 0 else None
+                    default[model_id].update(current)
+                return default
+            except Exception:
+                pass
+    return default
 
 
 def save_live_accuracy(data: dict[str, Any]) -> None:
@@ -1410,6 +1458,23 @@ def save_live_accuracy(data: dict[str, Any]) -> None:
     LIVE_ACCURACY_PATH.write_text(json.dumps(data, indent=2), encoding="utf-8")
 
 
+def _load_prediction_history(path: Path) -> pd.DataFrame:
+    if not path.exists():
+        return pd.DataFrame()
+    frame = pd.read_parquet(path)
+    for col in ("date", "input_date", "target_date", "forecast_date"):
+        if col in frame.columns:
+            frame[col] = pd.to_datetime(frame[col], errors="coerce")
+    sort_cols = [col for col in ("target_date", "input_date", "date", "forecast_date") if col in frame.columns]
+    if sort_cols:
+        return frame.sort_values(sort_cols).reset_index(drop=True)
+    return frame.reset_index(drop=True)
+
+
+def _record_prediction_history(path: Path, row: dict[str, Any], subset: list[str]) -> None:
+    append_prediction_history(path, row, subset)
+
+
 def update_live_accuracy(session_date: date) -> dict[str, Any]:
     """Score today's predictions against actual outcomes and update the ledger.