Fix runtime SQLite contention and inference reliability

.env.docker.example

@@ -67,7 +67,7 @@ TRACKED_COINS=BTC,ETH,SOL,BNB,XRP,ADA,DOGE,AVAX,DOT,LINK
 # Embedding Backend (sentiment scoring)
 # ============================================
 # EMBEDDING_BACKEND=local          → local sentence-transformers (default, safe)
-# EMBEDDING_BACKEND=openrouter     → OpenRouter API (needs key below)
+# EMBEDDING_BACKEND=openrouter     → OpenRouter API (needs key below; fail-closed)
 EMBEDDING_BACKEND=local
 EMBEDDING_MODEL=all-MiniLM-L6-v2
 

.env.example

@@ -31,6 +31,8 @@ OPENROUTER_API_KEY=
 # Embedding backend for sentiment scoring.
 #   EMBEDDING_BACKEND=local         → sentence-transformers (default, no API key)
 #   EMBEDDING_BACKEND=openrouter    → OpenRouter /embeddings (requires key above)
+# OpenRouter selection is fail-closed; the crawler never falls back to a local
+# model because mixing embedding spaces changes sentiment scores.
 EMBEDDING_BACKEND=local
 # Local: all-MiniLM-L6-v2 | all-mpnet-base-v2 | BAAI/bge-m3 ...
 # OpenRouter (opt-in): qwen/qwen3-embedding-8b | openai/text-embedding-3-large ...

README.md

@@ -74,7 +74,7 @@ uv run python -m crypto_sentiment_crawler.taskmanager start signals
 # Check status
 uv run python -m crypto_sentiment_crawler.taskmanager status
 
-# Run inference (price prediction)
+# Run the experimental inference diagnostic (not the production signal path)
 uv run python -m crypto_sentiment_crawler.inference
 
 # Run single signal check
@@ -364,7 +364,7 @@ crypto_sentiment_crawler/
     ├── main.py              # Entry point
     ├── taskmanager.py       # Task manager CLI
     ├── orchestrator.py      # Integration layer
-    ├── inference.py         # Price prediction
+    ├── inference.py         # Experimental live-data inference diagnostic
     │
     ├── bayesian/            # Decision layer
     │   ├── beliefs.py       # SourceBelief model

crypto_sentiment_crawler/analysis/source_weights.py

@@ -10,6 +10,7 @@
 from pathlib import Path
 
 from ..logging_config import logger
+from ..sqlite_utils import connect_sqlite
 from ..storage.db import Database
 
 
@@ -288,16 +289,16 @@ def read_belief_version() -> int | None:
             logger.warning("Could not read belief version from %s", resolved_state_path)
             return None
 
-    conn = sqlite3.connect(db_path)
+    conn = connect_sqlite(db_path)
     conn.row_factory = sqlite3.Row
-    columns = {row[1] for row in conn.execute("PRAGMA table_info(source_weights)")}
-    has_belief_version = "belief_version" in columns
-    snapshot_columns = {
-        row[1] for row in conn.execute("PRAGMA table_info(source_weight_snapshots)")
-    }
-    has_snapshots = "belief_version" in snapshot_columns
-
     try:
+        columns = {row[1] for row in conn.execute("PRAGMA table_info(source_weights)")}
+        has_belief_version = "belief_version" in columns
+        snapshot_columns = {
+            row[1] for row in conn.execute("PRAGMA table_info(source_weight_snapshots)")
+        }
+        has_snapshots = "belief_version" in snapshot_columns
+
         for _ in range(2):
             expected_version = read_belief_version()
             if expected_version is not None and has_snapshots:

crypto_sentiment_crawler/config.py

@@ -37,9 +37,6 @@ class Settings(BaseSettings):
     # OpenRouter is opt-in: set EMBEDDING_BACKEND=openrouter explicitly.
     embedding_backend: str = "local"
     embedding_model: str = "all-MiniLM-L6-v2"
-    # When True, the pipeline asserts a valid OpenRouter key before starting a
-    # crawl (rather than falling back silently).  Default False = safe.
-    embedding_require_openrouter: bool = False
 
     # Database
     database_path: str = "data/sentiment.db"

crypto_sentiment_crawler/confounders/collector.py

@@ -6,16 +6,16 @@
 
 import asyncio
 import json
-import sqlite3
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
 
-from .news import NewsCollector
+from ..logging_config import logger
+from ..sqlite_utils import sqlite_transaction
 from .macro import MacroCollector
-from .regime import RegimeCollector
 from .models import ConfounderSnapshot
-from ..logging_config import logger
+from .news import NewsCollector
+from .regime import RegimeCollector
 
 
 class ConfounderCollector:
@@ -56,9 +56,9 @@ def _init_db(self) -> None:
         """Initialize database table for confounders."""
         Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)
 
-        conn = sqlite3.connect(self.db_path)
-        conn.execute("""
-            CREATE TABLE IF NOT EXISTS confounders (
+        with sqlite_transaction(self.db_path) as conn:
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS confounders (
                 id INTEGER PRIMARY KEY AUTOINCREMENT,
                 timestamp TEXT NOT NULL,
 
@@ -101,14 +101,12 @@ def _init_db(self) -> None:
                 collection_errors TEXT,
 
                 UNIQUE(timestamp)
-            )
-        """)
-        conn.execute("""
-            CREATE INDEX IF NOT EXISTS idx_confounders_timestamp
-            ON confounders(timestamp)
-        """)
-        conn.commit()
-        conn.close()
+                )
+            """)
+            conn.execute("""
+                CREATE INDEX IF NOT EXISTS idx_confounders_timestamp
+                ON confounders(timestamp)
+            """)
 
     async def collect_all(self) -> ConfounderSnapshot:
         """
@@ -182,9 +180,7 @@ async def collect_all(self) -> ConfounderSnapshot:
 
     def store_snapshot(self, snapshot: ConfounderSnapshot) -> None:
         """Store a confounder snapshot to database."""
-        conn = sqlite3.connect(self.db_path)
-
-        try:
+        with sqlite_transaction(self.db_path) as conn:
             conn.execute("""
                 INSERT OR REPLACE INTO confounders (
                     timestamp,
@@ -226,14 +222,11 @@ def store_snapshot(self, snapshot: ConfounderSnapshot) -> None:
                 snapshot.reddit_avg_score,
                 json.dumps(snapshot.collection_errors) if snapshot.collection_errors else None,
             ))
-            conn.commit()
-        finally:
-            conn.close()
 
     async def collect_and_store(self) -> ConfounderSnapshot:
         """Collect all confounders and store to database."""
         snapshot = await self.collect_all()
-        self.store_snapshot(snapshot)
+        await asyncio.to_thread(self.store_snapshot, snapshot)
 
         vix_str = f"{snapshot.vix_level:.1f}" if snapshot.vix_level else "N/A"
         logger.info(

crypto_sentiment_crawler/dashboard/queries.py

@@ -7,10 +7,12 @@
 from pathlib import Path
 from typing import Optional
 
+from ..sqlite_utils import connect_sqlite
+
 
 def get_db_connection(db_path: str = "data/sentiment.db") -> sqlite3.Connection:
     """Get a database connection with row factory."""
-    conn = sqlite3.connect(db_path)
+    conn = connect_sqlite(db_path)
     conn.row_factory = sqlite3.Row
     return conn
 

crypto_sentiment_crawler/dashboard/routes.py

@@ -8,8 +8,8 @@
 from fastapi import APIRouter, HTTPException, Query
 
 from ..config import settings
-
 from .affiliates import get_affiliates_for_context
+from .news_schemas import TrendingResponse
 from .queries import (
     compute_beta_std,
     get_all_sources_sentiment_history,
@@ -27,12 +27,13 @@
     get_sentiment_history,
     get_source_sentiment_history,
     get_source_similarity,
-    get_source_type_label as get_source_type_label_query,
     get_source_weights,
     load_bayesian_beliefs,
     merge_history_data,
 )
-from .news_schemas import TrendingResponse
+from .queries import (
+    get_source_type_label as get_source_type_label_query,
+)
 from .schemas import (
     AffiliateLink,
     AffiliateResponse,
@@ -53,6 +54,10 @@
 
 router = APIRouter(prefix="/api", tags=["dashboard"])
 
+# These handlers intentionally use ``def`` rather than ``async def``. FastAPI
+# runs them in its thread pool so synchronous sqlite3 queries cannot block the
+# server event loop while waiting on another writer.
+
 DB_PATH = os.environ.get("DB_PATH", "data/sentiment.db")
 STATE_PATH = os.environ.get("STATE_PATH", "data/orchestrator_state.json")
 TRENDING_PATH = os.environ.get(
@@ -108,7 +113,7 @@ def get_source_type_label(accuracy: float | None, is_contrarian: bool) -> str:
 
 
 @router.get("/dashboard/summary", response_model=DashboardSummary)
-async def get_dashboard_summary():
+def get_dashboard_summary():
     """
     Get current dashboard summary metrics.
 
@@ -153,7 +158,7 @@ async def get_dashboard_summary():
 
 
 @router.get("/dashboard/history", response_model=DashboardHistory)
-async def get_dashboard_history(
+def get_dashboard_history(
     days: int = Query(30, ge=1, le=90, description="Number of days of history"),
 ):
     """
@@ -195,7 +200,7 @@ async def get_dashboard_history(
 
 
 @router.get("/dashboard/sources", response_model=SourceRankings)
-async def get_source_rankings():
+def get_source_rankings():
     """
     Get source accuracy rankings.
 
@@ -227,7 +232,7 @@ async def get_source_rankings():
 
 
 @router.get("/affiliates", response_model=AffiliateResponse)
-async def get_affiliates():
+def get_affiliates():
     """
     Get contextual affiliate recommendations.
 
@@ -251,7 +256,7 @@ async def get_affiliates():
 
 
 @router.get("/dashboard/beliefs", response_model=BeliefsResponse)
-async def get_bayesian_beliefs():
+def get_bayesian_beliefs():
     """
     Get current Bayesian beliefs from the model.
 
@@ -340,7 +345,7 @@ async def get_bayesian_beliefs():
 
 
 @router.get("/dashboard/beliefs/similarity")
-async def get_beliefs_similarity():
+def get_beliefs_similarity():
     """
     Get source similarity data computed from GP feature vectors.
 
@@ -355,7 +360,7 @@ async def get_beliefs_similarity():
 
 
 @router.get("/dashboard/sources/{source}/history", response_model=SourceSentimentHistory)
-async def get_source_history(
+def get_source_history(
     source: str,
     days: int = Query(30, ge=1, le=90, description="Number of days of history"),
 ):
@@ -395,7 +400,7 @@ async def get_source_history(
 
 
 @router.get("/dashboard/sources/history/all", response_model=AllSourcesHistory)
-async def get_all_sources_history(
+def get_all_sources_history(
     days: int = Query(30, ge=1, le=90, description="Number of days of history"),
 ):
     """
@@ -434,7 +439,7 @@ async def get_all_sources_history(
 
 
 @router.get("/dashboard/sources/list")
-async def get_sources_list():
+def get_sources_list():
     """
     Get list of all available sources with sentiment data.
 
@@ -456,7 +461,7 @@ async def get_sources_list():
 
 
 @router.get("/dashboard/coins")
-async def get_coins_list():
+def get_coins_list():
     """
     Get tracked coins sorted by latest market cap (descending).
     Stablecoins are excluded. Guarantees at least 6 coins by
@@ -501,7 +506,7 @@ async def get_coins_list():
 
 
 @router.get("/dashboard/coins/{coin}", response_model=CoinPrice)
-async def get_coin_price(coin: str):
+def get_coin_price(coin: str):
     """
     Get price data for a specific coin including 24h, 7d, and 30d changes.
     """
@@ -524,7 +529,7 @@ async def get_coin_price(coin: str):
 
 
 @router.get("/dashboard/coins/{coin}/history")
-async def get_coin_price_history(
+def get_coin_price_history(
     coin: str,
     days: int = Query(30, ge=1, le=90, description="Number of days of history"),
 ):
@@ -551,7 +556,7 @@ async def get_coin_price_history(
 
 
 @router.get("/dashboard/posts/recent", response_model=RecentPostsResponse)
-async def get_recent_posts(
+def get_recent_posts(
     limit: int = Query(20, ge=1, le=100, description="Number of posts to return"),
     source: str | None = Query(None, description="Filter by source"),
 ):
@@ -624,7 +629,7 @@ async def get_recent_posts(
 
 
 @router.get("/dashboard/sentiment/daily")
-async def get_daily_sentiment(
+def get_daily_sentiment(
     days: int = Query(30, ge=1, le=90, description="Number of days of history"),
     source: str | None = Query(None, description="Filter by source"),
 ):
@@ -647,7 +652,7 @@ async def get_daily_sentiment(
 
 
 @router.get("/dashboard/panic-score")
-async def get_panic_score():
+def get_panic_score():
     """
     Get Reddit-based panic score for the last 24 hours.
 
@@ -662,7 +667,7 @@ async def get_panic_score():
 
 
 @router.get("/ops/health")
-async def get_ops_health():
+def get_ops_health():
     """
     Pipeline health monitoring endpoint.
 
@@ -825,7 +830,7 @@ async def get_ops_health():
 
 
 @router.get("/news/trending", response_model=TrendingResponse)
-async def get_trending_signals():
+def get_trending_signals():
     """
     Get today's trending signals and market intelligence.