Upload 38 files

runtime.py

@@ -1244,25 +1244,7 @@ def load_tplus1_test_predictions() -> pd.DataFrame:
     return df.sort_values("date").reset_index(drop=True)
 
 
-def sync_mfe_outputs() -> bool:
-    """Refresh bundled MFE artifacts from the forecasting project when available."""
-    if not MFE_SOURCE_OUTPUT_DIR.exists():
-        return MFE_SUMMARY_PATH.exists()
-    MFE_OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
-    synced = False
-    for name in ("summary.json", "latest_prediction.csv", "test_predictions.csv"):
-        source = MFE_SOURCE_OUTPUT_DIR / name
-        target = MFE_OUTPUT_DIR / name
-        if not source.exists():
-            continue
-        if not target.exists() or source.stat().st_mtime > target.stat().st_mtime:
-            target.write_bytes(source.read_bytes())
-            synced = True
-    return synced or MFE_SUMMARY_PATH.exists()
-
-
 def load_mfe_summary() -> dict[str, Any]:
-    sync_mfe_outputs()
     if not MFE_SUMMARY_PATH.exists():
         return {}
     try:
@@ -1272,7 +1254,6 @@ def load_mfe_summary() -> dict[str, Any]:
 
 
 def load_mfe_latest() -> dict[str, Any]:
-    sync_mfe_outputs()
     summary = load_mfe_summary()
     if MFE_LATEST_PATH.exists():
         try:
@@ -1280,7 +1261,7 @@ def load_mfe_latest() -> dict[str, Any]:
             up_pts = float(row.get("predicted_up_points", summary.get("latest_predicted_up_points", 0)))
             down_pts = float(row.get("predicted_down_points", summary.get("latest_predicted_down_points", 0)))
             return {
-                "input_date": str(row.get("input_date") or summary.get("latest_input_date", ""))[:10],
+                "input_date": str(row.get("input_date") or row.get("date") or summary.get("latest_input_date", ""))[:10],
                 "first5_start": row.get("first5_start") or summary.get("latest_first5_start"),
                 "first5_end": row.get("first5_end") or summary.get("latest_first5_end"),
                 "first5_close": float(row.get("first5_close") or summary.get("latest_first5_close") or 0),
@@ -1306,7 +1287,6 @@ def load_mfe_latest() -> dict[str, Any]:
 
 
 def load_mfe_backtest() -> pd.DataFrame:
-    sync_mfe_outputs()
     if not MFE_TEST_PREDICTIONS_PATH.exists():
         return pd.DataFrame()
     frame = pd.read_csv(MFE_TEST_PREDICTIONS_PATH)
@@ -1697,6 +1677,58 @@ def refresh_first5_prediction(session_date: date | None = None, minutes: pd.Data
     return prediction
 
 
+def refresh_mfe_prediction(session_date: date | None = None, minutes: pd.DataFrame | None = None) -> dict[str, Any]:
+    if session_date is None:
+        today = datetime.now(IST).date()
+        if not is_trading_day(today):
+            raise RuntimeError(f"{today.isoformat()} is not an NSE trading session.")
+    else:
+        today = session_date
+
+    minutes = fetch_yahoo_minutes(period="7d") if minutes is None else minutes
+    first5 = first5_features_from_minutes(minutes, session_date=session_date)
+    row = build_model_row(first5)
+
+    payload = joblib.load(MFE_OUTPUT_DIR / "nifty_opening_mfe_regressor.joblib")
+    up_model = payload["up_model"]
+    down_model = payload["down_model"]
+    up_features = payload["up_features"]
+    down_features = payload["down_features"]
+
+    up_pred = float(up_model.predict(row[up_features])[0])
+    down_pred = float(down_model.predict(row[down_features])[0])
+
+    input_date_str = str(pd.to_datetime(row["date"].iloc[0]).date())
+    out = {
+        "input_date": input_date_str,
+        "first5_start": str(pd.to_datetime(row["first5_start"].iloc[0])),
+        "first5_end": str(pd.to_datetime(row["first5_end"].iloc[0])),
+        "first5_close": float(row["first5_close"].iloc[0]) if "first5_close" in row.columns else float(row["close"].iloc[0]),
+        "predicted_up_points": up_pred,
+        "predicted_down_points": down_pred,
+    }
+
+    pd.DataFrame([out]).to_csv(MFE_LATEST_PATH, index=False)
+
+    if MFE_TEST_PREDICTIONS_PATH.exists():
+        history = pd.read_csv(MFE_TEST_PREDICTIONS_PATH)
+        if "date" in history.columns:
+            history["date_obj"] = pd.to_datetime(history["date"], errors="coerce").dt.date
+            history = history[history["date_obj"] != pd.to_datetime(input_date_str).date()].copy()
+            history.drop(columns=["date_obj"], inplace=True)
+            
+            new_row = pd.DataFrame([{
+                "date": input_date_str,
+                "first5_close": out["first5_close"],
+                "predicted_up_points": out["predicted_up_points"],
+                "predicted_down_points": out["predicted_down_points"]
+            }])
+            history = pd.concat([history, new_row], ignore_index=True)
+            history.to_csv(MFE_TEST_PREDICTIONS_PATH, index=False)
+
+    return out
+
+
 def refresh_daily_data() -> dict[str, Any]:
     daily = fetch_yahoo_daily(period="1mo")
     combined = append_parquet_rows(NIFTY_1D_PATH, daily, ["date"])
@@ -2151,6 +2183,7 @@ def stale_data_status(now: datetime | None = None) -> dict[str, Any]:
     latest_daily = latest_parquet_date(NIFTY_1D_PATH)
     latest_minutes = latest_parquet_date(NIFTY_1M_PATH)
     latest_t5 = latest_prediction_input_date(LATEST_PATH)
+    latest_mfe = latest_prediction_input_date(MFE_LATEST_PATH)
     latest_tomorrow = latest_tomorrow_input_date()
     latest_tplus1 = latest_prediction_input_date(TPLUS1_LATEST_PATH)
     return {
@@ -2161,11 +2194,13 @@ def stale_data_status(now: datetime | None = None) -> dict[str, Any]:
         "latest_daily_date": latest_daily.isoformat() if latest_daily else None,
         "latest_minute_date": latest_minutes.isoformat() if latest_minutes else None,
         "latest_t5_date": latest_t5.isoformat() if latest_t5 else None,
+        "latest_mfe_date": latest_mfe.isoformat() if latest_mfe else None,
         "latest_tomorrow_date": latest_tomorrow.isoformat() if latest_tomorrow else None,
         "latest_tplus1_date": latest_tplus1.isoformat() if latest_tplus1 else None,
         "daily_stale": is_stale(latest_daily, expected_daily),
         "minutes_stale": is_stale(latest_minutes, expected_minutes),
         "t5_stale": is_stale(latest_t5, expected_minutes),
+        "mfe_stale": is_stale(latest_mfe, expected_minutes),
         "tomorrow_stale": is_stale(latest_tomorrow, expected_daily),
         "tplus1_stale": is_stale(latest_tplus1, expected_tplus1),
     }
@@ -2174,7 +2209,7 @@ def stale_data_status(now: datetime | None = None) -> dict[str, Any]:
 def refresh_stale_data_once(now: datetime | None = None) -> dict[str, Any]:
     now = now or datetime.now(IST)
     status = stale_data_status(now)
-    if not any(status[key] for key in ("daily_stale", "minutes_stale", "t5_stale", "tomorrow_stale", "tplus1_stale")):
+    if not any(status[key] for key in ("daily_stale", "minutes_stale", "t5_stale", "mfe_stale", "tomorrow_stale", "tplus1_stale")):
         return {"status": "fresh", **status, "actions": []}
     if not _stale_refresh_lock.acquire(blocking=False):
         return {"status": "skipped", "reason": "stale refresh already running", **status, "actions": []}
@@ -2221,6 +2256,13 @@ def refresh_stale_data_once(now: datetime | None = None) -> dict[str, Any]:
             prediction = refresh_first5_prediction(session_date=now.date())
             actions.append({"name": "t5_prediction", "input_date": prediction.input_date})
 
+        if status["mfe_stale"] and is_trading_day(now.date()) and now.time() >= FIRST5_READY:
+            try:
+                mfe_pred = refresh_mfe_prediction(session_date=now.date())
+                actions.append({"name": "mfe_prediction", "input_date": mfe_pred["input_date"]})
+            except Exception as e:
+                actions.append({"name": "mfe_prediction", "error": str(e)})
+
         if status["tplus1_stale"] and is_trading_day(now.date()) and now.time() >= TPLUS1_READY:
             prediction = refresh_tplus1_prediction(session_date=now.date())
             actions.append({"name": "tplus1_prediction", "input_date": prediction.get("input_date")})