Upload runtime.py

runtime.py

@@ -1441,10 +1441,10 @@ def load_live_accuracy() -> dict[str, Any]:
                     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["backtest_count"] = int(len(backtest_entries))
+                    current["live_count"] = int(len(live_entries))
+                    current["total"] = int(total)
+                    current["correct_count"] = int(correct)
                     current["accuracy"] = (current["correct_count"] / current["total"]) if current["total"] > 0 else None
                     default[model_id].update(current)
                 return default
@@ -1499,10 +1499,22 @@ def update_live_accuracy(session_date: date) -> dict[str, Any]:
 
     # --- T+5: today's 9:20 AM prediction vs close > open ---
     logged_t5 = {e["date"] for e in ledger["t5"]["entries"]}
-    if session_iso not in logged_t5 and LATEST_PATH.exists():
-        try:
-            t5_row = pd.read_csv(LATEST_PATH).iloc[-1].to_dict()
-            if str(t5_row.get("input_date", ""))[:10] == session_iso:
+    if session_iso not in logged_t5:
+        t5_history = _load_prediction_history(T5_PREDICTION_HISTORY_PATH)
+        if not t5_history.empty and "target_date" in t5_history.columns:
+            t5_rows = t5_history[t5_history["target_date"].dt.date == session_date]
+        else:
+            t5_rows = pd.DataFrame()
+        if t5_rows.empty and LATEST_PATH.exists():
+            try:
+                t5_row = pd.read_csv(LATEST_PATH).iloc[-1].to_dict()
+                if str(t5_row.get("input_date", ""))[:10] == session_iso:
+                    t5_rows = pd.DataFrame([t5_row])
+            except Exception:
+                t5_rows = pd.DataFrame()
+        if not t5_rows.empty:
+            try:
+                t5_row = t5_rows.iloc[-1].to_dict()
                 pred = str(t5_row.get("prediction", "")).upper()
                 if pred in ("UP", "DOWN"):
                     ledger["t5"]["entries"].append({
@@ -1510,16 +1522,29 @@ def update_live_accuracy(session_date: date) -> dict[str, Any]:
                         "prediction": pred,
                         "actual": actual_close_gt_open,
                         "correct": pred == actual_close_gt_open,
+                        "source": "live",
                     })
-        except Exception:
-            pass
+            except Exception:
+                pass
 
     # --- Tomorrow: yesterday's prediction targeting today vs close > open ---
     logged_tom = {e["date"] for e in ledger["tomorrow"]["entries"]}
-    if session_iso not in logged_tom and TOMORROW_LATEST_PATH.exists():
-        try:
-            tom_row = pd.read_csv(TOMORROW_LATEST_PATH).iloc[-1].to_dict()
-            if str(tom_row.get("target_date", ""))[:10] == session_iso:
+    if session_iso not in logged_tom:
+        tom_history = _load_prediction_history(TOMORROW_PREDICTION_HISTORY_PATH)
+        if not tom_history.empty and "target_date" in tom_history.columns:
+            tom_rows = tom_history[tom_history["target_date"].dt.date == session_date]
+        else:
+            tom_rows = pd.DataFrame()
+        if tom_rows.empty and TOMORROW_LATEST_PATH.exists():
+            try:
+                tom_row = pd.read_csv(TOMORROW_LATEST_PATH).iloc[-1].to_dict()
+                if str(tom_row.get("target_date", ""))[:10] == session_iso:
+                    tom_rows = pd.DataFrame([tom_row])
+            except Exception:
+                tom_rows = pd.DataFrame()
+        if not tom_rows.empty:
+            try:
+                tom_row = tom_rows.iloc[-1].to_dict()
                 pred = str(tom_row.get("prediction", "")).upper()
                 if pred in ("UP", "DOWN"):
                     ledger["tomorrow"]["entries"].append({
@@ -1527,17 +1552,23 @@ def update_live_accuracy(session_date: date) -> dict[str, Any]:
                         "prediction": pred,
                         "actual": actual_close_gt_open,
                         "correct": pred == actual_close_gt_open,
+                        "source": "live",
                     })
-        except Exception:
-            pass
+            except Exception:
+                pass
 
     # --- T+1: yesterday's 14:20 prediction targeting today ---
     # T+1 target: today's close > yesterday's 14:20 close
     logged_t1 = {e["date"] for e in ledger["tplus1"]["entries"]}
-    if session_iso not in logged_t1 and TPLUS1_LATEST_PATH.exists():
-        try:
-            t1_row = pd.read_csv(TPLUS1_LATEST_PATH).iloc[-1].to_dict()
-            if str(t1_row.get("target_date", ""))[:10] == session_iso:
+    if session_iso not in logged_t1:
+        t1_history = _load_prediction_history(TPLUS1_PREDICTION_HISTORY_PATH)
+        if not t1_history.empty and "target_date" in t1_history.columns:
+            t1_rows = t1_history[t1_history["target_date"].dt.date == session_date]
+        else:
+            t1_rows = pd.DataFrame()
+        if not t1_rows.empty:
+            try:
+                t1_row = t1_rows.iloc[-1].to_dict()
                 pred = str(t1_row.get("prediction", "")).upper()
                 input_date_str = str(t1_row.get("input_date", ""))[:10]
                 input_day = date.fromisoformat(input_date_str)
@@ -1560,17 +1591,21 @@ def update_live_accuracy(session_date: date) -> dict[str, Any]:
                         "prediction": pred,
                         "actual": t1_actual,
                         "correct": pred == t1_actual,
+                        "source": "live",
                     })
-        except Exception:
-            pass
+            except Exception:
+                pass
 
     # Recompute summary stats
     for model_id in ("t5", "tomorrow", "tplus1"):
         entries = ledger[model_id]["entries"]
         total = len(entries)
         correct = sum(1 for e in entries if e.get("correct"))
+        backtest_count = sum(1 for e in entries if str(e.get("source", "backtest")).lower() == "backtest")
         ledger[model_id]["total"] = total
         ledger[model_id]["correct_count"] = correct
+        ledger[model_id]["backtest_count"] = backtest_count
+        ledger[model_id]["live_count"] = total - backtest_count
         ledger[model_id]["accuracy"] = correct / total if total > 0 else None
 
     save_live_accuracy(ledger)