fix dashboard bugs and add resolution progress

api/dashboard.py

@@ -14,6 +14,71 @@ from evaluation.metrics import calculate_metrics
 
 router = APIRouter(tags=["dashboard"])
 
+
+@router.get("/resolution-status")
+def resolution_status():
+    """Return the latest durable resolution progress for dashboard polling."""
+    with connect(settings.DATABASE_URL) as conn, conn.cursor() as cur:
+        cur.execute(
+            """
+            SELECT COUNT(*) AS total,
+                   SUM(CASE WHEN resolved_at IS NOT NULL THEN 1 ELSE 0 END) AS resolved,
+                   SUM(CASE WHEN resolved_at IS NULL THEN 1 ELSE 0 END) AS pending
+            FROM predictions
+            """
+        )
+        counts = cur.fetchone()
+        cur.execute(
+            """
+            SELECT status, started_at, finished_at, last_updated_at,
+                   total_rows, rows_processed, resolved_rows, failed_rows
+            FROM job_runs
+            WHERE job_name = 'resolution'
+            ORDER BY id DESC
+            LIMIT 1
+            """
+        )
+        run = cur.fetchone()
+        cur.execute(
+            """
+            SELECT MAX(value) AS revision FROM (
+                SELECT MAX(prediction_timestamp) AS value FROM predictions
+                UNION ALL SELECT MAX(resolved_at) FROM predictions
+                UNION ALL SELECT MAX(calculated_at) FROM prediction_metrics
+                UNION ALL SELECT MAX(COALESCE(last_updated_at, finished_at, started_at)) FROM job_runs
+            )
+            """
+        )
+        revision = cur.fetchone()["revision"]
+
+    total = int(counts["total"] or 0)
+    resolved = int(counts["resolved"] or 0)
+    pending = int(counts["pending"] or 0)
+    if run is None:
+        return {
+            "status": "IDLE",
+            "processed": 0,
+            "total": total,
+            "resolved": resolved,
+            "pending": pending,
+            "failed": 0,
+            "started_at": None,
+            "last_updated_at": None,
+            "revision": revision,
+        }
+
+    return {
+        "status": run["status"],
+        "processed": int(run["rows_processed"] or 0),
+        "total": int(run["total_rows"] or total),
+        "resolved": resolved,
+        "pending": pending,
+        "failed": int(run["failed_rows"] or 0),
+        "started_at": run["started_at"],
+        "last_updated_at": run["last_updated_at"] or run["finished_at"] or run["started_at"],
+        "revision": revision,
+    }
+
 PREDICTION_COLUMNS = """
     id,
     prediction_date,
@@ -1122,6 +1187,45 @@ a:hover {{
     font-weight: 650;
 }}
 
+.resolution-card {{
+    background: var(--surface-1);
+    border: 1px solid var(--border);
+    border-radius: 8px;
+    padding: 1rem;
+    margin-bottom: 1.5rem;
+    max-width: 520px;
+}}
+
+.resolution-status {{
+    color: var(--series-1);
+    font-size: 1.15rem;
+    font-weight: 700;
+    letter-spacing: .04em;
+}}
+
+.resolution-progress {{
+    height: 10px;
+    margin: .75rem 0;
+    border-radius: 99px;
+    overflow: hidden;
+    background: rgba(128,128,128,.18);
+}}
+
+.resolution-progress > div {{
+    height: 100%;
+    width: 0;
+    background: var(--series-1);
+    transition: width .25s ease;
+}}
+
+.resolution-details {{
+    display: grid;
+    grid-template-columns: repeat(2, minmax(0, 1fr));
+    gap: .35rem 1rem;
+    color: var(--text-secondary);
+    font-variant-numeric: tabular-nums;
+}}
+
 .table-card {{
     background: var(--surface-1);
     border: 1px solid var(--border);
@@ -1236,6 +1340,21 @@ Predictions, five-session outcomes, calibration,
 and currently unresolved signals.
 </p>
 
+<section class="resolution-card" aria-live="polite">
+    <h2>Prediction resolution</h2>
+    <div class="resolution-status" id="resolution-status">Loading</div>
+    <div id="resolution-count">0 / 0 processed</div>
+    <div class="resolution-progress"><div id="resolution-progress-bar"></div></div>
+    <div id="resolution-percent">0.00%</div>
+    <div class="resolution-details">
+        <div>Resolved <strong id="resolution-resolved">0</strong></div>
+        <div>Pending <strong id="resolution-pending">0</strong></div>
+        <div>Failed <strong id="resolution-failed">0</strong></div>
+        <div>Started <strong id="resolution-started">—</strong></div>
+        <div>Last update <strong id="resolution-updated">—</strong></div>
+    </div>
+</section>
+
 {metrics_html}
 
 <h2>
@@ -1263,7 +1382,58 @@ to a higher probability of hitting the target?
 
 {pending_table}
 
+<script>
+let dashboardRevision = null;
+
+function formatResolutionTime(value) {{
+    if (!value) return "—";
+    return new Intl.DateTimeFormat("en-IN", {{
+        timeZone: "Asia/Kolkata",
+        hour: "2-digit",
+        minute: "2-digit",
+        second: "2-digit",
+        hour12: false,
+    }}).format(new Date(value));
+}}
+
+function setResolutionText(id, value) {{
+    document.getElementById(id).textContent = value;
+}}
+
+async function refreshDashboardState() {{
+    try {{
+        const response = await fetch("/resolution-status", {{ cache: "no-store" }});
+        if (!response.ok) return;
+        const state = await response.json();
+        const total = Math.max(Number(state.total) || 0, 0);
+        const processed = Math.min(Math.max(Number(state.processed) || 0, 0), total);
+        const percent = total ? (processed / total) * 100 : 0;
+
+        setResolutionText("resolution-status", state.status);
+        setResolutionText("resolution-count", `${{processed.toLocaleString()}} / ${{total.toLocaleString()}} processed`);
+        setResolutionText("resolution-percent", `${{percent.toFixed(2)}}%`);
+        setResolutionText("resolution-resolved", Number(state.resolved || 0).toLocaleString());
+        setResolutionText("resolution-pending", Number(state.pending || 0).toLocaleString());
+        setResolutionText("resolution-failed", Number(state.failed || 0).toLocaleString());
+        setResolutionText("resolution-started", formatResolutionTime(state.started_at));
+        setResolutionText("resolution-updated", formatResolutionTime(state.last_updated_at));
+        document.getElementById("resolution-progress-bar").style.width = `${{percent}}%`;
+
+        if (dashboardRevision !== null && state.revision && state.revision !== dashboardRevision) {{
+            window.location.reload();
+            return;
+        }}
+        dashboardRevision = state.revision;
+    }} catch (_) {{
+        // Keep the last rendered data visible while a transient poll fails.
+    }}
+}}
+
+refreshDashboardState();
+setInterval(refreshDashboardState, 3000);
+</script>
+
 </body>
 </html>
 """
-    )
+    )

database/init_db.py

@@ -13,6 +13,21 @@ from database.connection import connect
 
 SCHEMA_PATH = Path(__file__).parent / "schema.sql"
 
+JOB_RUN_PROGRESS_COLUMNS = {
+    "total_rows": "INTEGER NOT NULL DEFAULT 0",
+    "resolved_rows": "INTEGER NOT NULL DEFAULT 0",
+    "failed_rows": "INTEGER NOT NULL DEFAULT 0",
+    "last_updated_at": "TIMESTAMP",
+}
+
+
+def _migrate_job_runs(conn) -> None:
+    """Add progress columns for databases created before resolution tracking."""
+    existing = {row["name"] for row in conn.execute("PRAGMA table_info(job_runs)")}
+    for name, definition in JOB_RUN_PROGRESS_COLUMNS.items():
+        if name not in existing:
+            conn.execute(f"ALTER TABLE job_runs ADD COLUMN {name} {definition}")
+
 
 def init_db(database_url: str | None = None) -> None:
     from app.config import settings
@@ -20,6 +35,7 @@ def init_db(database_url: str | None = None) -> None:
     conn = connect(database_url or settings.DATABASE_URL)
     try:
         conn.executescript(SCHEMA_PATH.read_text())
+        _migrate_job_runs(conn)
         conn.commit()
     finally:
         conn.close()

database/job_runs_repository.py

@@ -22,3 +22,24 @@ def finish_job_run(database_url: str, job_run_id: int, *, success: bool, error_m
             "UPDATE job_runs SET status = ?, finished_at = STRFTIME('%Y-%m-%dT%H:%M:%fZ', 'now'), error_message = ? WHERE id = ?",
             ("SUCCESS" if success else "FAILED", error_message, job_run_id),
         )
+
+
+def update_job_progress(
+    database_url: str,
+    job_run_id: int,
+    *,
+    total_rows: int,
+    rows_processed: int,
+    resolved_rows: int,
+    failed_rows: int,
+) -> None:
+    with connect(database_url) as conn, conn.cursor() as cur:
+        cur.execute(
+            """
+            UPDATE job_runs
+            SET total_rows = ?, rows_processed = ?, resolved_rows = ?,
+                failed_rows = ?, last_updated_at = STRFTIME('%Y-%m-%dT%H:%M:%fZ', 'now')
+            WHERE id = ?
+            """,
+            (total_rows, rows_processed, resolved_rows, failed_rows, job_run_id),
+        )

database/metrics_repository.py

@@ -11,8 +11,9 @@ def get_database_url() -> str:
     return value
 
 
-def fetch_resolved_rows(conn, days: int = 20):
-    query = """
+def fetch_resolved_rows(conn, days: int | None = 20):
+    limit_clause = "" if days is None else "LIMIT :days"
+    query = f"""
         WITH latest_model AS (
             SELECT model_version FROM predictions
             WHERE resolved_at IS NOT NULL
@@ -20,7 +21,7 @@ def fetch_resolved_rows(conn, days: int = 20):
         ), latest_dates AS (
             SELECT prediction_date FROM predictions p JOIN latest_model m USING (model_version)
             WHERE p.actual_label IS NOT NULL AND p.actual_return IS NOT NULL
-            GROUP BY prediction_date ORDER BY prediction_date DESC LIMIT :days
+            GROUP BY prediction_date ORDER BY prediction_date DESC {limit_clause}
         )
         SELECT p.prediction_date, p.symbol, p.model_version, p.predicted_probability,
                p.rank, p.actual_return, p.actual_label
@@ -30,7 +31,7 @@ def fetch_resolved_rows(conn, days: int = 20):
         ORDER BY p.prediction_date, p.predicted_probability DESC, p.symbol
     """
     with conn.cursor() as cur:
-        cur.execute(query, {"days": days})
+        cur.execute(query, {} if days is None else {"days": days})
         return cur.fetchall()
 
 

database/resolution_repository.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from datetime import date, datetime, timezone
-from typing import Any
+from typing import Any, Callable
 
 from jobs.resolver import resolve_prediction_rows
 
@@ -28,25 +28,46 @@ def fetch_future_ohlcv(conn, symbol: str, prediction_date: date, forward_days: i
         return cur.fetchall()
 
 
-def resolve_pending_predictions(conn) -> int:
+def resolve_pending_predictions(
+    conn,
+    *,
+    on_progress: Callable[[int, int, int], None] | None = None,
+    commit_interval: int = 25,
+) -> int:
     resolved_count = 0
+    failed_count = 0
+    processed_count = 0
+
+    def checkpoint() -> None:
+        if processed_count % commit_interval == 0:
+            conn.commit()
+            if on_progress is not None:
+                on_progress(processed_count, resolved_count, failed_count)
+
     for prediction in fetch_unresolved_predictions(conn):
         rows = fetch_future_ohlcv(conn, prediction["symbol"], prediction["prediction_date"], prediction["target_horizon_days"])
-        resolution = resolve_prediction_rows(
-            prediction_date=prediction["prediction_date"], symbol=prediction["symbol"],
-            future_rows=[{"date": row["timestamp"], "open": row["open"], "close": row["close"]} for row in rows],
-            target_return=prediction["target_threshold"], forward_days=prediction["target_horizon_days"],
-        )
-        if resolution is None:
-            continue
-        with conn.cursor() as cur:
-            cur.execute("""
-                UPDATE predictions SET entry_date = ?, entry_open = ?, max_close_5d = ?,
-                    evaluation_end_date = ?, actual_return = ?, actual_label = ?, resolved_at = ?
-                WHERE id = ? AND resolved_at IS NULL
-            """, (resolution.entry_date, resolution.entry_open, resolution.max_close,
-                  resolution.evaluation_dates[-1], resolution.actual_return, resolution.actual_label,
-                  datetime.now(timezone.utc), prediction["id"]))
-            resolved_count += cur.rowcount
+        try:
+            resolution = resolve_prediction_rows(
+                prediction_date=prediction["prediction_date"], symbol=prediction["symbol"],
+                future_rows=[{"date": row["timestamp"], "open": row["open"], "close": row["close"]} for row in rows],
+                target_return=prediction["target_threshold"], forward_days=prediction["target_horizon_days"],
+            )
+        except ValueError:
+            failed_count += 1
+        else:
+            if resolution is not None:
+                with conn.cursor() as cur:
+                    cur.execute("""
+                        UPDATE predictions SET entry_date = ?, entry_open = ?, max_close_5d = ?,
+                            evaluation_end_date = ?, actual_return = ?, actual_label = ?, resolved_at = ?
+                        WHERE id = ? AND resolved_at IS NULL
+                    """, (resolution.entry_date, resolution.entry_open, resolution.max_close,
+                          resolution.evaluation_dates[-1], resolution.actual_return, resolution.actual_label,
+                          datetime.now(timezone.utc), prediction["id"]))
+                    resolved_count += cur.rowcount
+        processed_count += 1
+        checkpoint()
     conn.commit()
+    if on_progress is not None:
+        on_progress(processed_count, resolved_count, failed_count)
     return resolved_count

database/schema.sql

@@ -121,6 +121,10 @@ CREATE TABLE IF NOT EXISTS job_runs (
     ),
 
     rows_processed INTEGER NOT NULL DEFAULT 0,
+    total_rows INTEGER NOT NULL DEFAULT 0,
+    resolved_rows INTEGER NOT NULL DEFAULT 0,
+    failed_rows INTEGER NOT NULL DEFAULT 0,
+    last_updated_at TIMESTAMP,
 
     error_message TEXT
 );

jobs/resolve_predictions.py

@@ -11,8 +11,8 @@ from dotenv import load_dotenv
 from app.config import settings
 from database.connection import connect
 from database.job_lock_repository import acquire_job_lock, release_job_lock
-from database.job_runs_repository import finish_job_run, start_job_run
-from database.resolution_repository import resolve_pending_predictions
+from database.job_runs_repository import finish_job_run, start_job_run, update_job_progress
+from database.resolution_repository import fetch_unresolved_predictions, resolve_pending_predictions
 
 load_dotenv()
 
@@ -35,7 +35,20 @@ def run_resolution_job() -> int | None:
                 settings.DATABASE_URL, "resolution", datetime.now(IST).date()
             )
             with connect(settings.DATABASE_URL) as conn:
-                resolved = resolve_pending_predictions(conn)
+                total_rows = len(fetch_unresolved_predictions(conn))
+
+                def report_progress(processed: int, resolved: int, failed: int) -> None:
+                    update_job_progress(
+                        settings.DATABASE_URL,
+                        job_run_id,
+                        total_rows=total_rows,
+                        rows_processed=processed,
+                        resolved_rows=resolved,
+                        failed_rows=failed,
+                    )
+
+                report_progress(0, 0, 0)
+                resolved = resolve_pending_predictions(conn, on_progress=report_progress)
             finish_job_run(settings.DATABASE_URL, job_run_id, success=True)
             log.info("Resolution complete: %s predictions resolved", resolved)
             return resolved