Upload signals_db.py with huggingface_hub

signals_db.py

@@ -4,26 +4,43 @@ Stores one row per inference call and records the actual buyer outcome
 when it becomes available — used as training labels on the next retrain.
 """
 
+import os
 import sqlite3
+from contextlib import closing
 from pathlib import Path
 from typing import Optional
 import pandas as pd
 
-DB_PATH = Path(__file__).parent / "signals.db"
+# Prefer a persistent-volume mount so the feedback log survives HF Space restarts.
+# Override with SIGNALS_DB_PATH env var for local dev or alternative mount points.
+DB_PATH = Path(os.environ.get("SIGNALS_DB_PATH", "/data/signals.db"))
 
 
 def get_connection(db_path: Path = DB_PATH) -> sqlite3.Connection:
-    """Return a WAL-mode connection with row_factory = sqlite3.Row."""
-    conn = sqlite3.connect(str(db_path))
+    """Return a WAL-mode connection.
+
+    check_same_thread=False: FastAPI runs sync handlers in a thread pool and
+    a daemon thread drives retraining — connections must be usable from any
+    thread that opens them. Each call site opens *and closes* its own
+    connection, so there is no shared-connection cross-thread hazard.
+
+    busy_timeout=5000: WAL mode handles read/write concurrency well, but a
+    brief write-lock collision (e.g. retrain thread + request thread writing
+    simultaneously) would otherwise raise OperationalError immediately.
+    """
+    conn = sqlite3.connect(str(db_path), check_same_thread=False)
     conn.row_factory = sqlite3.Row
     conn.execute("PRAGMA journal_mode=WAL")
     conn.execute("PRAGMA foreign_keys=ON")
+    conn.execute("PRAGMA busy_timeout=5000")
     return conn
 
 
 def init_db(db_path: Path = DB_PATH) -> None:
     """Create the prediction_log table and indexes. Idempotent."""
-    with get_connection(db_path) as conn:
+    db_path = Path(db_path)
+    db_path.parent.mkdir(parents=True, exist_ok=True)
+    with closing(get_connection(db_path)) as conn, conn:
         conn.executescript("""
             CREATE TABLE IF NOT EXISTS prediction_log (
                 id                    INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -53,7 +70,7 @@ def log_prediction(
     db_path: Path = DB_PATH,
 ) -> None:
     """Insert a new prediction entry. One row per prediction call."""
-    with get_connection(db_path) as conn:
+    with closing(get_connection(db_path)) as conn, conn:
         conn.execute(
             """INSERT INTO prediction_log
                (record_id, predicted_at, predicted_probability, predicted_label, predicted_roi_pct)
@@ -68,27 +85,65 @@ def record_outcome(
     outcome_at: str,
     db_path: Path = DB_PATH,
 ) -> int:
-    """Update the latest unresolved prediction for record_id with the actual outcome.
-    Returns number of rows updated."""
-    with get_connection(db_path) as conn:
+    """Update the single most-recent unresolved prediction for record_id.
+
+    Returns 1 if a row was updated, 0 if no unresolved prediction existed.
+    Targets only the latest row so that a record scored multiple times before
+    the buyer responds does not get every prior prediction overwritten.
+    """
+    with closing(get_connection(db_path)) as conn, conn:
         cursor = conn.execute(
             """UPDATE prediction_log
                SET actual_outcome = ?, outcome_at = ?
-               WHERE record_id = ? AND actual_outcome IS NULL""",
+               WHERE id = (
+                   SELECT id FROM prediction_log
+                   WHERE record_id = ? AND actual_outcome IS NULL
+                   ORDER BY predicted_at DESC
+                   LIMIT 1
+               )""",
             (actual_outcome, outcome_at, record_id),
         )
         return cursor.rowcount
 
 
+def record_outcomes_many(
+    outcomes: list[dict],
+    db_path: Path = DB_PATH,
+) -> int:
+    """Update the most-recent unresolved prediction for each record_id in one transaction.
+
+    outcomes: list of dicts with keys record_id, actual_outcome, outcome_at.
+    Returns total rows updated.
+    """
+    if not outcomes:
+        return 0
+    with closing(get_connection(db_path)) as conn, conn:
+        total = 0
+        for o in outcomes:
+            cursor = conn.execute(
+                """UPDATE prediction_log
+                   SET actual_outcome = ?, outcome_at = ?
+                   WHERE id = (
+                       SELECT id FROM prediction_log
+                       WHERE record_id = ? AND actual_outcome IS NULL
+                       ORDER BY predicted_at DESC
+                       LIMIT 1
+                   )""",
+                (o["actual_outcome"], o["outcome_at"], o["record_id"]),
+            )
+            total += cursor.rowcount
+        return total
+
+
 def get_prediction_log_df(db_path: Path = DB_PATH) -> pd.DataFrame:
     """Return the full prediction log as a DataFrame."""
-    with get_connection(db_path) as conn:
+    with closing(get_connection(db_path)) as conn:
         return pd.read_sql_query("SELECT * FROM prediction_log ORDER BY predicted_at DESC", conn)
 
 
 def get_resolved_predictions(db_path: Path = DB_PATH) -> pd.DataFrame:
     """Return only predictions where the actual outcome is known."""
-    with get_connection(db_path) as conn:
+    with closing(get_connection(db_path)) as conn:
         return pd.read_sql_query(
             "SELECT * FROM prediction_log WHERE actual_outcome IS NOT NULL ORDER BY outcome_at DESC",
             conn,
@@ -97,7 +152,7 @@ def get_resolved_predictions(db_path: Path = DB_PATH) -> pd.DataFrame:
 
 def db_summary(db_path: Path = DB_PATH) -> dict:
     """Return row count for prediction_log."""
-    with get_connection(db_path) as conn:
+    with closing(get_connection(db_path)) as conn:
         count = conn.execute("SELECT COUNT(*) FROM prediction_log").fetchone()[0]
         return {"prediction_log": count}