From 44799808154ce3c6b42e8a73d70b98e9e9cfb7c7 Mon Sep 17 00:00:00 2001 From: "protostatis.dev" Date: Fri, 14 Aug 2026 13:21:24 -0500 Subject: [PATCH] fix: harden sentiment runtime reliability --- .env.docker.example | 2 +- .env.example | 2 + README.md | 4 +- .../analysis/source_weights.py | 17 +- crypto_sentiment_crawler/config.py | 3 - .../confounders/collector.py | 37 +-- crypto_sentiment_crawler/dashboard/queries.py | 4 +- crypto_sentiment_crawler/dashboard/routes.py | 45 +-- crypto_sentiment_crawler/inference.py | 129 ++++++-- crypto_sentiment_crawler/orchestrator.py | 20 +- .../processing/semantic_sentiment.py | 53 +-- .../processing/user_sentiment.py | 303 +++++++++--------- crypto_sentiment_crawler/scheduler.py | 42 ++- crypto_sentiment_crawler/signals/alerts.py | 54 ++-- crypto_sentiment_crawler/signals/api.py | 12 +- crypto_sentiment_crawler/signals/service.py | 8 +- crypto_sentiment_crawler/sqlite_utils.py | 42 +++ crypto_sentiment_crawler/storage/db.py | 295 +++++++++-------- pipeline.md | 4 +- tests/test_inference_current_data.py | 89 +++++ tests/test_scheduler_belief_job.py | 45 ++- tests/test_semantic_runtime.py | 144 +++++++++ tests/test_sqlite_runtime.py | 184 +++++++++++ 23 files changed, 1073 insertions(+), 465 deletions(-) create mode 100644 crypto_sentiment_crawler/sqlite_utils.py create mode 100644 tests/test_inference_current_data.py create mode 100644 tests/test_semantic_runtime.py create mode 100644 tests/test_sqlite_runtime.py diff --git a/.env.docker.example b/.env.docker.example index eab7e6a..4dc47c3 100644 --- a/.env.docker.example +++ b/.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 diff --git a/.env.example b/.env.example index 0f83cdd..146e3ad 100644 --- a/.env.example +++ b/.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 ... diff --git a/README.md b/README.md index 8cf4dcf..44f8c4a 100644 --- a/README.md +++ b/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 diff --git a/crypto_sentiment_crawler/analysis/source_weights.py b/crypto_sentiment_crawler/analysis/source_weights.py index 9ab2608..29eed35 100644 --- a/crypto_sentiment_crawler/analysis/source_weights.py +++ b/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: diff --git a/crypto_sentiment_crawler/config.py b/crypto_sentiment_crawler/config.py index 54c1c21..91d1b43 100644 --- a/crypto_sentiment_crawler/config.py +++ b/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" diff --git a/crypto_sentiment_crawler/confounders/collector.py b/crypto_sentiment_crawler/confounders/collector.py index 718fefc..024442e 100644 --- a/crypto_sentiment_crawler/confounders/collector.py +++ b/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( diff --git a/crypto_sentiment_crawler/dashboard/queries.py b/crypto_sentiment_crawler/dashboard/queries.py index 38eda20..0d32d6e 100644 --- a/crypto_sentiment_crawler/dashboard/queries.py +++ b/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 diff --git a/crypto_sentiment_crawler/dashboard/routes.py b/crypto_sentiment_crawler/dashboard/routes.py index ad70d1a..46d43d1 100644 --- a/crypto_sentiment_crawler/dashboard/routes.py +++ b/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. diff --git a/crypto_sentiment_crawler/inference.py b/crypto_sentiment_crawler/inference.py index 7cf5284..1cca923 100644 --- a/crypto_sentiment_crawler/inference.py +++ b/crypto_sentiment_crawler/inference.py @@ -1,7 +1,8 @@ -""" -Inference module: Analyze sentiment data and predict price movements. +"""Experimental price-direction diagnostic using current sentiment data. -Uses collected sentiment signals to infer price direction for the next 4 hours. +This module is not the production signal path. It reads the live +``user_sentiment_scores`` and ``confounders`` tables and refuses to emit a +directional result when fresh user sentiment is unavailable. """ import json @@ -14,8 +15,9 @@ import pandas as pd from scipy import stats -from .logging_config import logger from .analysis.source_weights import load_weights_from_db_sync +from .logging_config import logger +from .sqlite_utils import connect_sqlite @dataclass @@ -122,7 +124,7 @@ def get_weights_summary(self) -> str: def get_connection(self) -> sqlite3.Connection: """Get database connection.""" - conn = sqlite3.connect(self.db_path) + conn = connect_sqlite(self.db_path) conn.row_factory = sqlite3.Row return conn @@ -134,24 +136,62 @@ def get_recent_sentiment( """Get recent sentiment scores.""" conn = self.get_connection() - cutoff = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat() + now = datetime.now(timezone.utc) + cutoff = (now - timedelta(hours=hours)).isoformat() + fear_greed_cutoff = (now - timedelta(hours=max(hours, 24))).isoformat() - # Get recent sentiment, but always include fear_greed (may be older) - query = """ - SELECT source, coin, score, confidence, sample_size, timestamp - FROM sentiment_scores - WHERE (timestamp >= ? OR source = 'fear_greed') - """ + coin_filter = "" params = [cutoff] - if coin: - query += " AND (coin = ? OR coin = 'MARKET')" + coin_filter = "AND (uss.coin = ? OR uss.coin IS NULL OR uss.coin = 'MARKET')" params.append(coin) + params.append(fear_greed_cutoff) + + query = f""" + WITH recent_user_sentiment AS ( + SELECT + up.source AS source, + COALESCE(uss.coin, 'MARKET') AS coin, + uss.final_score AS score, + 1.0 AS confidence, + MAX( + 1, + COALESCE(uss.segments_scored, 0) + + CASE WHEN uss.title_score IS NULL THEN 0 ELSE 1 END + ) AS sample_size, + uss.timestamp AS timestamp + FROM user_sentiment_scores uss + JOIN user_profiles up ON up.user_id = uss.user_id + WHERE uss.timestamp >= ? + AND uss.final_score IS NOT NULL + {coin_filter} + ), + latest_fear_greed AS ( + SELECT + 'fear_greed' AS source, + 'MARKET' AS coin, + (fear_greed_index - 50.0) / 50.0 AS score, + 1.0 AS confidence, + 1 AS sample_size, + timestamp + FROM confounders + WHERE fear_greed_index IS NOT NULL + AND timestamp >= ? + ORDER BY timestamp DESC + LIMIT 1 + ) + SELECT source, coin, score, confidence, sample_size, timestamp + FROM recent_user_sentiment + UNION ALL + SELECT source, coin, score, confidence, sample_size, timestamp + FROM latest_fear_greed + ORDER BY timestamp DESC + """ - query += " ORDER BY timestamp DESC" - - df = pd.read_sql_query(query, conn, params=params) - conn.close() + try: + df = pd.read_sql_query(query, conn, params=params) + finally: + conn.close() if not df.empty: df["timestamp"] = pd.to_datetime(df["timestamp"], format="ISO8601") @@ -161,12 +201,19 @@ def get_recent_sentiment( def get_latest_fear_greed(self) -> float: """Get the most recent Fear & Greed score.""" conn = self.get_connection() - row = conn.execute(""" - SELECT score FROM sentiment_scores - WHERE source = 'fear_greed' - ORDER BY timestamp DESC LIMIT 1 - """).fetchone() - conn.close() + cutoff = (datetime.now(timezone.utc) - timedelta(hours=24)).isoformat() + try: + row = conn.execute( + """ + SELECT (fear_greed_index - 50.0) / 50.0 AS score + FROM confounders + WHERE fear_greed_index IS NOT NULL AND timestamp >= ? + ORDER BY timestamp DESC LIMIT 1 + """, + (cutoff,), + ).fetchone() + finally: + conn.close() return row[0] if row else 0.0 def get_recent_prices(self, hours: int = 4, coin: str = "BTC") -> pd.DataFrame: @@ -182,8 +229,10 @@ def get_recent_prices(self, hours: int = 4, coin: str = "BTC") -> pd.DataFrame: ORDER BY timestamp ASC """ - df = pd.read_sql_query(query, conn, params=[coin, cutoff]) - conn.close() + try: + df = pd.read_sql_query(query, conn, params=[coin, cutoff]) + finally: + conn.close() if not df.empty: df["timestamp"] = pd.to_datetime(df["timestamp"], format="ISO8601") @@ -213,6 +262,7 @@ def compute_aggregate_sentiment( "confidence": 0.0, "signal_strength": 0.0, "n_sources": 0, + "n_social_sources": 0, } # Group by source, take most recent or average @@ -262,6 +312,7 @@ def compute_aggregate_sentiment( confidence = min(1.0, (n_sources / 5) * 0.5 + agreement * 0.5) n_contrarian = sum(1 for d in by_source.values() if d.get("is_contrarian")) + n_social_sources = sum(1 for source in by_source if source != "fear_greed") return { "composite_score": composite, @@ -269,6 +320,7 @@ def compute_aggregate_sentiment( "confidence": confidence, "signal_strength": abs(composite), "n_sources": n_sources, + "n_social_sources": n_social_sources, "n_contrarian": n_contrarian, } @@ -343,6 +395,27 @@ def predict( coin=coin, ) + if sentiment["n_social_sources"] == 0: + return PredictionResult( + coin=coin, + current_price=momentum["current_price"], + predicted_direction="neutral", + confidence=0.0, + sentiment_score=sentiment["composite_score"], + signals={ + "sentiment": sentiment, + "momentum": momentum, + "adjusted_score": 0.0, + "contrarian_boost": 0.0, + "data_source": "user_sentiment_scores", + }, + reasoning=( + "No fresh user sentiment is available for this lookback; " + "directional inference was suppressed." + ), + timestamp=datetime.now(timezone.utc), + ) + # Build prediction composite = sentiment["composite_score"] confidence = sentiment["confidence"] @@ -416,6 +489,7 @@ def predict( "momentum": momentum, "adjusted_score": adjusted_score, "contrarian_boost": contrarian_boost, + "data_source": "user_sentiment_scores", }, reasoning=reasoning, timestamp=datetime.now(timezone.utc), @@ -439,7 +513,8 @@ def run_inference(lookback_hours: int = 4, horizon_hours: int = 4) -> dict: predictor = PricePredictor() print("=" * 70) - print("CRYPTO SENTIMENT INFERENCE") + print("EXPERIMENTAL CRYPTO SENTIMENT INFERENCE") + print("Diagnostic only; production alerts use the signal service") print(f"Lookback: {lookback_hours}h | Prediction Horizon: {horizon_hours}h") print("=" * 70) diff --git a/crypto_sentiment_crawler/orchestrator.py b/crypto_sentiment_crawler/orchestrator.py index a81e4ce..4d8a353 100644 --- a/crypto_sentiment_crawler/orchestrator.py +++ b/crypto_sentiment_crawler/orchestrator.py @@ -792,7 +792,8 @@ async def _store_content(self, content: CrawledContent) -> int: coin = content.coins_mentioned[0] if content.coins_mentioned else None post_timestamp = content.published_at or content.crawled_at try: - post_score = self.user_scorer.score_post( + post_score = await asyncio.to_thread( + self.user_scorer.score_post, raw_data=raw_data, raw_id=raw_id, timestamp=post_timestamp.isoformat(), @@ -800,15 +801,24 @@ async def _store_content(self, content: CrawledContent) -> int: coin=coin, ) if post_score: - score_id = self.user_scorer.save_post_score(post_score) + score_id = await asyncio.to_thread( + self.user_scorer.save_post_score, + post_score, + ) # Write score back to content for later belief evaluation content.sentiment_score = post_score.final_score if score_id > 0: # Update user profile - user_id = self.user_scorer.get_or_create_user( - post_score.username, post_score.source, post_score.timestamp + user_id = await asyncio.to_thread( + self.user_scorer.get_or_create_user, + post_score.username, + post_score.source, + post_score.timestamp, + ) + await asyncio.to_thread( + self.user_scorer.update_user_profile, + user_id, ) - self.user_scorer.update_user_profile(user_id) logger.debug( f"User scored: {content.source} final={post_score.final_score:.3f} " f"fear={post_score.fear_index:.2f} euphoria={post_score.euphoria_index:.2f}" diff --git a/crypto_sentiment_crawler/processing/semantic_sentiment.py b/crypto_sentiment_crawler/processing/semantic_sentiment.py index 7779e43..4f870ac 100644 --- a/crypto_sentiment_crawler/processing/semantic_sentiment.py +++ b/crypto_sentiment_crawler/processing/semantic_sentiment.py @@ -608,26 +608,18 @@ def __init__( # Neither provider, backend, nor model_name given — read settings. from ..config import settings - cfg_backend = settings.embedding_backend or "local" + cfg_backend = (settings.embedding_backend or "local").lower() cfg_model = settings.embedding_model or "all-MiniLM-L6-v2" if cfg_backend == "openrouter": - try: - self.provider = OpenRouterEmbeddingProvider( - model=cfg_model, - api_key=settings.openrouter_api_key, - ) - except RuntimeError: - if settings.embedding_require_openrouter: - raise - logger.warning( - "OpenRouter unavailable; falling back to local MiniLM" - ) - self.provider = LocalSentenceTransformerProvider( - model_name="all-MiniLM-L6-v2" - ) - else: + self.provider = OpenRouterEmbeddingProvider( + model=cfg_model, + api_key=settings.openrouter_api_key, + ) + elif cfg_backend == "local": self.provider = LocalSentenceTransformerProvider(model_name=cfg_model) + else: + raise ValueError(f"Unknown embedding backend: {cfg_backend!r}") logger.info( "SemanticSentimentAnalyzer using provider=%s dim=%d", type(self.provider).__name__, @@ -681,7 +673,10 @@ def analyze(self, text: str, method: str = "asymmetric") -> dict: """ # Encode input text embedding = self.provider.encode_single(text, normalize=True) + return self._analyze_embedding(embedding, method) + def _analyze_embedding(self, embedding: np.ndarray, method: str) -> dict: + """Analyze one normalized embedding with the selected scoring method.""" if method == "centroid": bullish_sim = self._cosine_similarity(embedding, self.bullish_centroid) bearish_sim = self._cosine_similarity(embedding, self.bearish_centroid) @@ -733,30 +728,10 @@ def get_score(self, text: str) -> float: """Get sentiment score from -1 (bearish) to 1 (bullish).""" return self.analyze(text)["score"] - def analyze_batch(self, texts: list[str]) -> list[dict]: - """Analyze multiple texts efficiently.""" + def analyze_batch(self, texts: list[str], method: str = "centroid") -> list[dict]: + """Analyze multiple texts in one provider request.""" embeddings = self.provider.encode(texts, normalize=True) - - results = [] - for embedding in embeddings: - bullish_sim = self._cosine_similarity(embedding, self.bullish_centroid) - bearish_sim = self._cosine_similarity(embedding, self.bearish_centroid) - neutral_sim = self._cosine_similarity(embedding, self.neutral_centroid) - - raw_score = bullish_sim - bearish_sim - score = np.tanh(raw_score * 3) - sentiment_strength = max(bullish_sim, bearish_sim) - neutral_sim - confidence = max(0.0, min(1.0, sentiment_strength * 2 + 0.5)) - - results.append({ - "score": float(score), - "bullish_sim": float(bullish_sim), - "bearish_sim": float(bearish_sim), - "neutral_sim": float(neutral_sim), - "confidence": float(confidence), - }) - - return results + return [self._analyze_embedding(embedding, method) for embedding in embeddings] def get_scores_batch(self, texts: list[str]) -> list[float]: """Get sentiment scores for multiple texts.""" diff --git a/crypto_sentiment_crawler/processing/user_sentiment.py b/crypto_sentiment_crawler/processing/user_sentiment.py index 5334ad0..730f154 100644 --- a/crypto_sentiment_crawler/processing/user_sentiment.py +++ b/crypto_sentiment_crawler/processing/user_sentiment.py @@ -11,6 +11,8 @@ import numpy as np +from ..sqlite_utils import connect_sqlite, sqlite_transaction + logger = logging.getLogger("crypto_sentiment") # Minimum human comments required for a post to be saved/scored @@ -242,7 +244,7 @@ def __init__( def _get_connection(self) -> sqlite3.Connection: """Get database connection.""" - return sqlite3.connect(self.db_path) + return connect_sqlite(self.db_path) def _split_into_segments(self, text: str) -> list[str]: """Split text into meaningful segments.""" @@ -407,14 +409,18 @@ def score_post(self, raw_data: dict, raw_id: int, timestamp: str, source: str, c logger.debug(f"Skipping post {raw_id}: only {human_comment_count} human comments (min: {min_human_comments})") return None - # Score title (always included) + # Split and categorize segments + segments = self._split_into_segments(content)[:20] + score_title = bool(title and len(title.strip()) >= 10) + texts_to_score = ([title] if score_title else []) + segments + results = self.analyzer.analyze_batch(texts_to_score, method="asymmetric") + result_index = 0 + title_score = None - if title and len(title.strip()) >= 10: - result = self.analyzer.analyze(title, method="asymmetric") - title_score = result['score'] + if score_title: + title_score = results[result_index]["score"] + result_index += 1 - # Split and categorize segments - segments = self._split_into_segments(content) segment_objs = [] # Counters for multi-dimensional signals @@ -426,10 +432,9 @@ def score_post(self, raw_data: dict, raw_id: int, timestamp: str, source: str, c # Scores for filtered sentiment (STANDARD + TRUE_BEARISH only) scored_segment_values = [] - for seg in segments[:20]: # Limit segments - # Get semantic score - result = self.analyzer.analyze(seg, method="asymmetric") - score = result['score'] + for seg in segments: + score = results[result_index]["score"] + result_index += 1 # Categorize segment category = categorize_segment(seg) @@ -508,71 +513,72 @@ def score_post(self, raw_data: dict, raw_id: int, timestamp: str, source: str, c def get_or_create_user(self, username: str, source: str, timestamp: str) -> int: """Get existing user_id or create new user profile.""" source = source.lower() - conn = self._get_connection() - cursor = conn.cursor() - - # Try to get existing user - cursor.execute( - "SELECT user_id FROM user_profiles WHERE username = ? AND source = ?", - (username, source) - ) - row = cursor.fetchone() + with sqlite_transaction(self.db_path) as conn: + cursor = conn.cursor() - if row: - user_id = row[0] - # Update last_seen + # Try to get existing user cursor.execute( - "UPDATE user_profiles SET last_seen = ? WHERE user_id = ?", - (timestamp, user_id) + "SELECT user_id FROM user_profiles WHERE username = ? AND source = ?", + (username, source), ) - else: - # Create new user - cursor.execute( - """INSERT INTO user_profiles - (username, source, first_seen, last_seen, total_posts, credibility_weight) - VALUES (?, ?, ?, ?, 0, 1.0)""", - (username, source, timestamp, timestamp) - ) - user_id = cursor.lastrowid + row = cursor.fetchone() - conn.commit() - conn.close() - return user_id + if row: + user_id = row[0] + # Update last_seen + cursor.execute( + "UPDATE user_profiles SET last_seen = ? WHERE user_id = ?", + (timestamp, user_id), + ) + else: + # Create new user + cursor.execute( + """INSERT INTO user_profiles + (username, source, first_seen, last_seen, total_posts, + credibility_weight) + VALUES (?, ?, ?, ?, 0, 1.0)""", + (username, source, timestamp, timestamp), + ) + user_id = cursor.lastrowid + + if user_id is None: + raise RuntimeError("Failed to resolve user profile ID") + return int(user_id) def save_post_score(self, post_score: PostScore, update_existing: bool = False) -> int: """Save post score to database and link to user.""" - conn = self._get_connection() - cursor = conn.cursor() - # Get or create user user_id = self.get_or_create_user( post_score.username, post_score.source, - post_score.timestamp + post_score.timestamp, ) - # Check if already scored - cursor.execute( - "SELECT id FROM user_sentiment_scores WHERE raw_id = ?", - (post_score.raw_id,) - ) - existing = cursor.fetchone() + # Serialize segment scores with categories + segment_json = json.dumps([ + { + "text": s.text, + "score": s.score, + "len": s.char_length, + "category": s.category.value, + "included": s.included, + } + for s in post_score.segment_scores + ]) - if existing: - if update_existing: - score_id = existing[0] - # Update existing record - segment_json = json.dumps([ - { - "text": s.text, - "score": s.score, - "len": s.char_length, - "category": s.category.value, - "included": s.included, - } - for s in post_score.segment_scores - ]) + with sqlite_transaction(self.db_path) as conn: + cursor = conn.cursor() + cursor.execute( + "SELECT id FROM user_sentiment_scores WHERE raw_id = ?", + (post_score.raw_id,), + ) + existing = cursor.fetchone() + + if existing: + if not update_existing: + return -1 + score_id = existing[0] cursor.execute( """UPDATE user_sentiment_scores SET title_score = ?, body_score = ?, segment_scores = ?, @@ -594,119 +600,98 @@ def save_post_score(self, post_score: PostScore, update_existing: bool = False) post_score.segments_filtered, post_score.segments_scored, score_id, - ) + ), ) - conn.commit() - conn.close() return score_id - else: - conn.close() - return -1 # Already exists - # Serialize segment scores with categories - segment_json = json.dumps([ - { - "text": s.text, - "score": s.score, - "len": s.char_length, - "category": s.category.value, - "included": s.included, - } - for s in post_score.segment_scores - ]) - - # Insert score with new fields - cursor.execute( - """INSERT INTO user_sentiment_scores - (user_id, raw_id, timestamp, coin, title_score, body_score, - segment_scores, final_score, aggregation_method, - pos_count, neg_count, neu_count, - activity_level, fear_index, euphoria_index, - segments_filtered, segments_scored) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", - ( - user_id, - post_score.raw_id, - post_score.timestamp, - post_score.coin, - post_score.title_score, - post_score.body_score, - segment_json, - post_score.final_score, - post_score.aggregation_method, - post_score.pos_count, - post_score.neg_count, - post_score.neu_count, - post_score.activity_level, - post_score.fear_index, - post_score.euphoria_index, - post_score.segments_filtered, - post_score.segments_scored, + cursor.execute( + """INSERT INTO user_sentiment_scores + (user_id, raw_id, timestamp, coin, title_score, body_score, + segment_scores, final_score, aggregation_method, + pos_count, neg_count, neu_count, + activity_level, fear_index, euphoria_index, + segments_filtered, segments_scored) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + ( + user_id, + post_score.raw_id, + post_score.timestamp, + post_score.coin, + post_score.title_score, + post_score.body_score, + segment_json, + post_score.final_score, + post_score.aggregation_method, + post_score.pos_count, + post_score.neg_count, + post_score.neu_count, + post_score.activity_level, + post_score.fear_index, + post_score.euphoria_index, + post_score.segments_filtered, + post_score.segments_scored, + ), ) - ) - score_id = cursor.lastrowid + score_id = cursor.lastrowid - conn.commit() - conn.close() - - return score_id + if score_id is None: + raise RuntimeError("Failed to create user sentiment score") + return int(score_id) def update_user_profile(self, user_id: int): """Recalculate and update user profile aggregates.""" - conn = self._get_connection() - cursor = conn.cursor() - - # Get all scores for user - cursor.execute( - """SELECT final_score, timestamp FROM user_sentiment_scores - WHERE user_id = ? ORDER BY timestamp""", - (user_id,) - ) - rows = cursor.fetchall() + with sqlite_transaction(self.db_path) as conn: + cursor = conn.cursor() - if not rows: - conn.close() - return + cursor.execute( + """SELECT final_score, timestamp FROM user_sentiment_scores + WHERE user_id = ? ORDER BY timestamp""", + (user_id,), + ) + rows = cursor.fetchall() - scores = [r[0] for r in rows] - timestamps = [r[1] for r in rows] + if not rows: + return - total_posts = len(scores) - avg_sentiment = float(np.mean(scores)) - sentiment_stddev = float(np.std(scores)) if len(scores) > 1 else 0.0 - bullish_pct = sum(1 for s in scores if s > 0.1) / total_posts - bearish_pct = sum(1 for s in scores if s < -0.1) / total_posts - tendency = self._classify_tendency(bullish_pct, bearish_pct, sentiment_stddev) + scores = [row[0] for row in rows] + timestamps = [row[1] for row in rows] - # Update profile - cursor.execute( - """UPDATE user_profiles SET - total_posts = ?, - avg_sentiment = ?, - sentiment_stddev = ?, - bullish_pct = ?, - bearish_pct = ?, - tendency = ?, - first_seen = ?, - last_seen = ?, - updated_at = ? - WHERE user_id = ?""", - ( - total_posts, - avg_sentiment, - sentiment_stddev, + total_posts = len(scores) + avg_sentiment = float(np.mean(scores)) + sentiment_stddev = float(np.std(scores)) if len(scores) > 1 else 0.0 + bullish_pct = sum(1 for score in scores if score > 0.1) / total_posts + bearish_pct = sum(1 for score in scores if score < -0.1) / total_posts + tendency = self._classify_tendency( bullish_pct, bearish_pct, - tendency, - timestamps[0], - timestamps[-1], - datetime.now(timezone.utc).isoformat(), - user_id, + sentiment_stddev, ) - ) - conn.commit() - conn.close() + cursor.execute( + """UPDATE user_profiles SET + total_posts = ?, + avg_sentiment = ?, + sentiment_stddev = ?, + bullish_pct = ?, + bearish_pct = ?, + tendency = ?, + first_seen = ?, + last_seen = ?, + updated_at = ? + WHERE user_id = ?""", + ( + total_posts, + avg_sentiment, + sentiment_stddev, + bullish_pct, + bearish_pct, + tendency, + timestamps[0], + timestamps[-1], + datetime.now(timezone.utc).isoformat(), + user_id, + ), + ) def get_user_profile(self, username: str, source: str) -> Optional[UserProfile]: """Get user profile by username and source.""" @@ -846,7 +831,7 @@ def backfill_user_scores(db_path: str = "data/sentiment.db", limit: int = None, """ scorer = UserSentimentScorer(db_path=db_path) - conn = sqlite3.connect(db_path) + conn = connect_sqlite(db_path) cursor = conn.cursor() # Default to all social media sources diff --git a/crypto_sentiment_crawler/scheduler.py b/crypto_sentiment_crawler/scheduler.py index c34acd5..b7c4597 100644 --- a/crypto_sentiment_crawler/scheduler.py +++ b/crypto_sentiment_crawler/scheduler.py @@ -14,7 +14,7 @@ import asyncio import signal import sys -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from functools import wraps from pathlib import Path @@ -30,6 +30,17 @@ from .orchestrator import CrawlerOrchestrator from .storage.db import Database +JOB_STAGGER_SECONDS = { + "crawl": 0, + "price": 5, + "evaluate": 15, + "stats": 20, + "confounders": 30, + "onchain": 45, + "belief_update": 75, + "fear_greed": 105, +} + def _tracked_job(job): """Track a scheduler job so shutdown can drain it without cancellation.""" @@ -394,10 +405,21 @@ async def _job_stats(self) -> None: def _setup_jobs(self) -> None: """Configure scheduled jobs.""" + anchor = datetime.now(timezone.utc) + + def staggered_interval(seconds: int, job_id: str) -> IntervalTrigger: + """Keep recurring SQLite writers from firing on the same second.""" + + return IntervalTrigger( + seconds=seconds, + start_date=anchor + + timedelta(seconds=seconds + JOB_STAGGER_SECONDS[job_id]), + ) + # Crawl job - most frequent self.scheduler.add_job( self._job_crawl, - IntervalTrigger(seconds=self.crawl_interval), + staggered_interval(self.crawl_interval, "crawl"), id="crawl", name="Bayesian Crawl", max_instances=1, @@ -406,7 +428,7 @@ def _setup_jobs(self) -> None: # Price job self.scheduler.add_job( self._job_price, - IntervalTrigger(seconds=self.price_interval), + staggered_interval(self.price_interval, "price"), id="price", name="Price Collection", max_instances=1, @@ -415,7 +437,7 @@ def _setup_jobs(self) -> None: # Evaluation job self.scheduler.add_job( self._job_evaluate, - IntervalTrigger(seconds=self.eval_interval), + staggered_interval(self.eval_interval, "evaluate"), id="evaluate", name="Outcome Evaluation", max_instances=1, @@ -424,7 +446,7 @@ def _setup_jobs(self) -> None: # Fear & Greed job self.scheduler.add_job( self._job_fear_greed, - IntervalTrigger(seconds=self.fear_greed_interval), + staggered_interval(self.fear_greed_interval, "fear_greed"), id="fear_greed", name="Fear & Greed", max_instances=1, @@ -433,7 +455,7 @@ def _setup_jobs(self) -> None: # Confounder collection job (for causal inference) self.scheduler.add_job( self._job_confounders, - IntervalTrigger(seconds=self.confounder_interval), + staggered_interval(self.confounder_interval, "confounders"), id="confounders", name="Confounder Collection", max_instances=1, @@ -442,7 +464,7 @@ def _setup_jobs(self) -> None: # On-chain metrics job self.scheduler.add_job( self._job_onchain, - IntervalTrigger(seconds=self.onchain_interval), + staggered_interval(self.onchain_interval, "onchain"), id="onchain", name="On-Chain Metrics", max_instances=1, @@ -451,7 +473,7 @@ def _setup_jobs(self) -> None: # Belief update + source_weights sync - every 30 minutes self.scheduler.add_job( self._job_belief_update, - IntervalTrigger(seconds=1800), + staggered_interval(1800, "belief_update"), id="belief_update", name="Belief Update & Weights Sync", max_instances=1, @@ -460,7 +482,7 @@ def _setup_jobs(self) -> None: # Stats job - every 10 minutes self.scheduler.add_job( self._job_stats, - IntervalTrigger(seconds=600), + staggered_interval(600, "stats"), id="stats", name="Statistics", max_instances=1, @@ -557,8 +579,8 @@ def signal_handler(): # Quick stats viewer async def view_live_stats() -> None: """View current stats from the database.""" - import sqlite3 import json + import sqlite3 db_path = Path("data/sentiment.db") state_path = Path("data/orchestrator_state.json") diff --git a/crypto_sentiment_crawler/signals/alerts.py b/crypto_sentiment_crawler/signals/alerts.py index c426d57..acb3a70 100644 --- a/crypto_sentiment_crawler/signals/alerts.py +++ b/crypto_sentiment_crawler/signals/alerts.py @@ -475,9 +475,11 @@ async def _handle_sources(self, chat_id: str) -> None: try: from ..analysis.source_weights import load_weights_from_db_sync - weight_data = load_weights_from_db_sync(self.db_path) + weight_data = await asyncio.to_thread( + load_weights_from_db_sync, + self.db_path, + ) weights = weight_data.get("weights", {}) - contrarian = weight_data.get("contrarian_sources", set()) if not weights: await self.channel.send_message( @@ -488,26 +490,35 @@ async def _handle_sources(self, chat_id: str) -> None: # Load accuracy from database import sqlite3 - conn = sqlite3.connect(self.db_path) - conn.row_factory = sqlite3.Row + from ..sqlite_utils import connect_sqlite + belief_version = weight_data.get("belief_version") - snapshot_columns = { - row[1] - for row in conn.execute("PRAGMA table_info(source_weight_snapshots)") - } - if belief_version is None or "belief_version" not in snapshot_columns: - rows = conn.execute( - "SELECT source, weight, accuracy, is_contrarian, sample_size " - "FROM source_weights ORDER BY weight DESC LIMIT 10" - ).fetchall() - else: - rows = conn.execute( - "SELECT source, weight, accuracy, is_contrarian, sample_size " - "FROM source_weight_snapshots WHERE belief_version = ? " - "ORDER BY weight DESC LIMIT 10", - (belief_version,), - ).fetchall() - conn.close() + + def load_rows(): + conn = connect_sqlite(self.db_path) + conn.row_factory = sqlite3.Row + try: + snapshot_columns = { + row[1] + for row in conn.execute( + "PRAGMA table_info(source_weight_snapshots)" + ) + } + if belief_version is None or "belief_version" not in snapshot_columns: + return conn.execute( + "SELECT source, weight, accuracy, is_contrarian, sample_size " + "FROM source_weights ORDER BY weight DESC LIMIT 10" + ).fetchall() + return conn.execute( + "SELECT source, weight, accuracy, is_contrarian, sample_size " + "FROM source_weight_snapshots WHERE belief_version = ? " + "ORDER BY weight DESC LIMIT 10", + (belief_version,), + ).fetchall() + finally: + conn.close() + + rows = await asyncio.to_thread(load_rows) if not rows: await self.channel.send_message( @@ -520,7 +531,6 @@ async def _handle_sources(self, chat_id: str) -> None: for row in rows: name = row["source"].replace("_", " ").title() acc = row["accuracy"] * 100 if row["accuracy"] else 0 - w = row["weight"] n = row["sample_size"] or 0 entry = f" `{name[:18]:<18}` {acc:.0f}% (n={n})" if row["is_contrarian"]: diff --git a/crypto_sentiment_crawler/signals/api.py b/crypto_sentiment_crawler/signals/api.py index 2b0b980..d485215 100644 --- a/crypto_sentiment_crawler/signals/api.py +++ b/crypto_sentiment_crawler/signals/api.py @@ -1,7 +1,6 @@ """REST API for the Contrarian Signal Service.""" import os -import sqlite3 from datetime import datetime from typing import Optional @@ -9,20 +8,19 @@ from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel -from .service import SignalService -from .subscriptions import SubscriptionManager, SubscriptionTier, TIERS from ..dashboard import router as dashboard_router +from ..sqlite_utils import sqlite_transaction from ..storage.db import SCHEMA +from .service import SignalService +from .subscriptions import TIERS, SubscriptionManager, SubscriptionTier def init_database_schema(db_path: str) -> None: """Initialize database schema on startup.""" from pathlib import Path Path(db_path).parent.mkdir(parents=True, exist_ok=True) - conn = sqlite3.connect(db_path) - conn.executescript(SCHEMA) - conn.commit() - conn.close() + with sqlite_transaction(db_path) as conn: + conn.executescript(SCHEMA) print(f"Database schema initialized: {db_path}") diff --git a/crypto_sentiment_crawler/signals/service.py b/crypto_sentiment_crawler/signals/service.py index c183cd3..591f875 100644 --- a/crypto_sentiment_crawler/signals/service.py +++ b/crypto_sentiment_crawler/signals/service.py @@ -6,9 +6,9 @@ from pathlib import Path from typing import Optional -from ..storage.db import Database from ..analysis.source_weights import load_weights_from_db_sync -from .alerts import AlertManager, TelegramChannel, TelegramBot +from ..storage.db import Database +from .alerts import AlertManager, TelegramBot, TelegramChannel from .detector import ContrarianSignalDetector from .models import Signal @@ -161,7 +161,7 @@ async def _load_data(self) -> tuple[list, list, dict]: async def check_signals(self) -> Optional[Signal]: """Check current conditions for signals.""" try: - self._load_source_weights() + await asyncio.to_thread(self._load_source_weights) sentiment_history, price_history, multi_dim = await self._load_data() if len(sentiment_history) < 10 or len(price_history) < 24: @@ -199,7 +199,7 @@ async def check_signals(self) -> Optional[Signal]: async def get_market_summary(self) -> dict: """Get current market summary without requiring a signal.""" - self._load_source_weights() + await asyncio.to_thread(self._load_source_weights) sentiment_history, price_history, multi_dim = await self._load_data() if len(sentiment_history) < 5 or len(price_history) < 2: diff --git a/crypto_sentiment_crawler/sqlite_utils.py b/crypto_sentiment_crawler/sqlite_utils.py new file mode 100644 index 0000000..ac1c3b7 --- /dev/null +++ b/crypto_sentiment_crawler/sqlite_utils.py @@ -0,0 +1,42 @@ +"""Shared SQLite connection settings for concurrent runtime services.""" + +import sqlite3 +from contextlib import contextmanager +from os import PathLike +from typing import Any, Iterator + +SQLITE_BUSY_TIMEOUT_SECONDS = 30.0 +SQLITE_BUSY_TIMEOUT_MS = int(SQLITE_BUSY_TIMEOUT_SECONDS * 1000) + + +def connect_sqlite( + database: str | bytes | PathLike[str] | PathLike[bytes], + **kwargs: Any, +) -> sqlite3.Connection: + """Open SQLite with a consistent bounded wait for concurrent writers.""" + + timeout_seconds = float(kwargs.setdefault("timeout", SQLITE_BUSY_TIMEOUT_SECONDS)) + connection = sqlite3.connect(database, **kwargs) + connection.execute(f"PRAGMA busy_timeout = {int(timeout_seconds * 1000)}") + return connection + + +@contextmanager +def sqlite_transaction( + database: str | bytes | PathLike[str] | PathLike[bytes], + **kwargs: Any, +) -> Iterator[sqlite3.Connection]: + """Open a short write transaction that always rolls back and closes safely.""" + + connection = connect_sqlite(database, **kwargs) + try: + yield connection + connection.commit() + except BaseException: + try: + connection.rollback() + except sqlite3.Error: + pass + raise + finally: + connection.close() diff --git a/crypto_sentiment_crawler/storage/db.py b/crypto_sentiment_crawler/storage/db.py index 679c410..d3a7efa 100644 --- a/crypto_sentiment_crawler/storage/db.py +++ b/crypto_sentiment_crawler/storage/db.py @@ -1,15 +1,20 @@ """Database operations using aiosqlite.""" +import asyncio import hashlib import json +import sqlite3 import uuid +from contextlib import asynccontextmanager from datetime import datetime, timedelta, timezone from pathlib import Path +from typing import AsyncIterator import aiosqlite from ..config import settings from ..logging_config import logger +from ..sqlite_utils import SQLITE_BUSY_TIMEOUT_MS, SQLITE_BUSY_TIMEOUT_SECONDS from .migrations import run_all_migrations from .models import OnChainMetric, PriceData, SentimentRaw @@ -211,12 +216,17 @@ class Database: def __init__(self, db_path: Path | None = None): self.db_path = db_path or settings.db_path self._connection: aiosqlite.Connection | None = None + self._write_lock = asyncio.Lock() async def connect(self) -> None: """Connect to the database and initialize schema.""" self.db_path.parent.mkdir(parents=True, exist_ok=True) - self._connection = await aiosqlite.connect(self.db_path) + self._connection = await aiosqlite.connect( + self.db_path, + timeout=SQLITE_BUSY_TIMEOUT_SECONDS, + ) self._connection.row_factory = aiosqlite.Row + await self._connection.execute(f"PRAGMA busy_timeout = {SQLITE_BUSY_TIMEOUT_MS}") await self._connection.executescript(SCHEMA) await self._connection.commit() @@ -225,8 +235,13 @@ async def connect(self) -> None: try: await self._connection.execute(migration) await self._connection.commit() + except sqlite3.OperationalError as exc: + await self._connection.rollback() + if "duplicate column name" not in str(exc).lower(): + raise except Exception: - pass # Column already exists + await self._connection.rollback() + raise # Run data migrations (one-time cleanups, etc.) await run_all_migrations(self._connection) @@ -282,23 +297,23 @@ async def record_heartbeat( last_error_at = now last_error_message = error_message - await self.conn.execute( - """ - INSERT INTO pipeline_heartbeats ( - component, last_success_at, last_error_at, last_error_message, - freshness_seconds, metadata - ) VALUES (?, ?, ?, ?, ?, ?) - """, - ( - component, - last_success_at.isoformat() if last_success_at else None, - last_error_at.isoformat() if last_error_at else None, - last_error_message, - freshness_seconds, - json.dumps(metadata) if metadata is not None else None, - ), - ) - await self.conn.commit() + async with self.write_transaction() as conn: + await conn.execute( + """ + INSERT INTO pipeline_heartbeats ( + component, last_success_at, last_error_at, last_error_message, + freshness_seconds, metadata + ) VALUES (?, ?, ?, ?, ?, ?) + """, + ( + component, + last_success_at.isoformat() if last_success_at else None, + last_error_at.isoformat() if last_error_at else None, + last_error_message, + freshness_seconds, + json.dumps(metadata) if metadata is not None else None, + ), + ) async def get_latest_heartbeat(self, component: str) -> dict | None: """Return the most recent heartbeat for one component.""" @@ -337,6 +352,21 @@ def conn(self) -> aiosqlite.Connection: raise RuntimeError("Database not connected. Call connect() first.") return self._connection + @asynccontextmanager + async def write_transaction(self) -> AsyncIterator[aiosqlite.Connection]: + """Serialize writes and guarantee rollback after a failed transaction.""" + + async with self._write_lock: + try: + yield self.conn + await self.conn.commit() + except BaseException: + try: + await self.conn.rollback() + except Exception: + logger.exception("Failed to roll back SQLite transaction") + raise + def _compute_content_hash(self, source: str, raw_data: dict) -> str: """Compute a hash for deduplication based on source and URL/title.""" # Use URL if available, otherwise use source + title @@ -359,57 +389,58 @@ async def insert_sentiment_raw(self, data: SentimentRaw) -> int: return 0 # Skip duplicate source = data.source.lower() - cursor = await self.conn.execute( - """ - INSERT INTO sentiment_raw (timestamp, source, coin, raw_data, content_hash) - VALUES (?, ?, ?, ?, ?) - """, - ( - data.timestamp.isoformat(), - source, - data.coin, - json.dumps(data.raw_data), - content_hash, - ), - ) - await self.conn.commit() + async with self.write_transaction() as conn: + cursor = await conn.execute( + """ + INSERT INTO sentiment_raw (timestamp, source, coin, raw_data, content_hash) + VALUES (?, ?, ?, ?, ?) + """, + ( + data.timestamp.isoformat(), + source, + data.coin, + json.dumps(data.raw_data), + content_hash, + ), + ) return cursor.lastrowid or 0 async def insert_price_data(self, data: PriceData) -> int: """Insert price data.""" - cursor = await self.conn.execute( - """ - INSERT INTO price_data (timestamp, coin, price_usd, volume_24h, market_cap, source) - VALUES (?, ?, ?, ?, ?, ?) - """, - ( - data.timestamp.isoformat(), - data.coin, - data.price_usd, - data.volume_24h, - data.market_cap, - data.source, - ), - ) - await self.conn.commit() + async with self.write_transaction() as conn: + cursor = await conn.execute( + """ + INSERT INTO price_data ( + timestamp, coin, price_usd, volume_24h, market_cap, source + ) VALUES (?, ?, ?, ?, ?, ?) + """, + ( + data.timestamp.isoformat(), + data.coin, + data.price_usd, + data.volume_24h, + data.market_cap, + data.source, + ), + ) return cursor.lastrowid or 0 async def insert_on_chain_metric(self, data: OnChainMetric) -> int: """Insert on-chain metric.""" - cursor = await self.conn.execute( - """ - INSERT INTO on_chain_metrics (timestamp, coin, metric_type, value, metadata) - VALUES (?, ?, ?, ?, ?) - """, - ( - data.timestamp.isoformat(), - data.coin, - data.metric_type, - data.value, - json.dumps(data.metadata) if data.metadata else None, - ), - ) - await self.conn.commit() + async with self.write_transaction() as conn: + cursor = await conn.execute( + """ + INSERT INTO on_chain_metrics (timestamp, coin, metric_type, value, metadata) + VALUES (?, ?, ?, ?, ?) + """, + ( + data.timestamp.isoformat(), + data.coin, + data.metric_type, + data.value, + json.dumps(data.metadata) if data.metadata else None, + ), + ) return cursor.lastrowid or 0 async def insert_sentiment_score(self, data) -> int: @@ -463,29 +494,29 @@ async def upsert_outcome( Returns the row id. """ - cursor = await self.conn.execute( - """ - INSERT INTO prediction_outcomes - (source, signal_timestamp, target_timestamp, calibrated_score, - price_before, price_before_timestamp, evaluator_version) - VALUES (?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(source, signal_timestamp, evaluator_version) DO UPDATE SET - target_timestamp = excluded.target_timestamp, - calibrated_score = excluded.calibrated_score, - price_before = excluded.price_before, - price_before_timestamp = excluded.price_before_timestamp - """, - ( - source.lower(), - signal_timestamp.isoformat(), - target_timestamp.isoformat(), - calibrated_score, - price_before, - price_before_timestamp.isoformat() if price_before_timestamp else None, - evaluator_version, - ), - ) - await self.conn.commit() + async with self.write_transaction() as conn: + cursor = await conn.execute( + """ + INSERT INTO prediction_outcomes + (source, signal_timestamp, target_timestamp, calibrated_score, + price_before, price_before_timestamp, evaluator_version) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(source, signal_timestamp, evaluator_version) DO UPDATE SET + target_timestamp = excluded.target_timestamp, + calibrated_score = excluded.calibrated_score, + price_before = excluded.price_before, + price_before_timestamp = excluded.price_before_timestamp + """, + ( + source.lower(), + signal_timestamp.isoformat(), + target_timestamp.isoformat(), + calibrated_score, + price_before, + price_before_timestamp.isoformat() if price_before_timestamp else None, + evaluator_version, + ), + ) return cursor.lastrowid or 0 async def claim_pending_outcomes( @@ -501,32 +532,32 @@ async def claim_pending_outcomes( """ now = now or datetime.now(timezone.utc) claim_token = f"claim:{now.isoformat()}:{uuid.uuid4().hex}" - await self.conn.execute( - """ - UPDATE prediction_outcomes - SET evaluated_at = ? - WHERE target_timestamp < ? - AND price_after IS NULL - AND abstained = FALSE - AND evaluated_at IS NULL - """, - (claim_token, now.isoformat()), - ) + async with self.write_transaction() as conn: + await conn.execute( + """ + UPDATE prediction_outcomes + SET evaluated_at = ? + WHERE target_timestamp < ? + AND price_after IS NULL + AND abstained = FALSE + AND evaluated_at IS NULL + """, + (claim_token, now.isoformat()), + ) - # Fetch only rows claimed by this worker. A generic "claiming" marker - # would allow a concurrent worker to process someone else's outcomes. - cursor = await self.conn.execute( - """ - SELECT id, source, signal_timestamp, target_timestamp, - calibrated_score, price_before, price_before_timestamp - FROM prediction_outcomes - WHERE evaluated_at = ? - ORDER BY signal_timestamp ASC - """, - (claim_token,), - ) - rows = await cursor.fetchall() - await self.conn.commit() + # Fetch only rows claimed by this worker. A generic "claiming" marker + # would allow a concurrent worker to process someone else's outcomes. + cursor = await conn.execute( + """ + SELECT id, source, signal_timestamp, target_timestamp, + calibrated_score, price_before, price_before_timestamp + FROM prediction_outcomes + WHERE evaluated_at = ? + ORDER BY signal_timestamp ASC + """, + (claim_token,), + ) + rows = await cursor.fetchall() return [dict(row) for row in rows] async def mark_outcome_evaluated( @@ -539,28 +570,28 @@ async def mark_outcome_evaluated( price_gap_seconds: float | None = None, ) -> None: """Update an outcome row with evaluation results.""" - await self.conn.execute( - """ - UPDATE prediction_outcomes - SET price_after = ?, - price_after_timestamp = ?, - correct = ?, - direction = ?, - price_gap_seconds = ?, - evaluated_at = ? - WHERE id = ? - """, - ( - price_after, - price_after_timestamp.isoformat(), - 1 if correct else 0, - direction, - price_gap_seconds, - datetime.now(timezone.utc).isoformat(), - outcome_id, - ), - ) - await self.conn.commit() + async with self.write_transaction() as conn: + await conn.execute( + """ + UPDATE prediction_outcomes + SET price_after = ?, + price_after_timestamp = ?, + correct = ?, + direction = ?, + price_gap_seconds = ?, + evaluated_at = ? + WHERE id = ? + """, + ( + price_after, + price_after_timestamp.isoformat(), + 1 if correct else 0, + direction, + price_gap_seconds, + datetime.now(timezone.utc).isoformat(), + outcome_id, + ), + ) async def get_source_performance( self, source: str, days: int = 30 diff --git a/pipeline.md b/pipeline.md index 013a575..61c8172 100644 --- a/pipeline.md +++ b/pipeline.md @@ -62,7 +62,7 @@ uv run python -m crypto_sentiment_crawler.taskmanager status ### Single Commands ```bash -# Run inference (price prediction) +# Run the experimental inference diagnostic (reads live user sentiment) uv run python -m crypto_sentiment_crawler.inference # Run signal check once @@ -365,7 +365,7 @@ sqlite3 data/sentiment.db "SELECT source, weight, accuracy, is_contrarian FROM s |------|---------| | `scheduler.py` | Background job scheduler | | `orchestrator.py` | Integration layer | -| `inference.py` | Price prediction with dynamic weights | +| `inference.py` | Experimental live-data inference diagnostic | | `taskmanager.py` | Task management CLI | | `bayesian/bandit.py` | Thompson Sampling | | `bayesian/beliefs.py` | Source belief model | diff --git a/tests/test_inference_current_data.py b/tests/test_inference_current_data.py new file mode 100644 index 0000000..3c68ea9 --- /dev/null +++ b/tests/test_inference_current_data.py @@ -0,0 +1,89 @@ +"""Regression tests for the experimental inference data source.""" + +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from crypto_sentiment_crawler.inference import PricePredictor, SentimentAnalyzer +from crypto_sentiment_crawler.sqlite_utils import connect_sqlite +from crypto_sentiment_crawler.storage.db import SCHEMA + + +def create_inference_database(path: Path, *, sentiment_age_hours: int | None) -> None: + now = datetime.now(timezone.utc) + connection = connect_sqlite(path) + connection.executescript(SCHEMA) + connection.execute( + """ + CREATE TABLE sentiment_scores ( + source TEXT, coin TEXT, score REAL, confidence REAL, + sample_size INTEGER, timestamp TEXT + ) + """ + ) + connection.execute( + "INSERT INTO sentiment_scores VALUES (?, ?, ?, ?, ?, ?)", + ("legacy_source", "BTC", 0.99, 1.0, 1, now.isoformat()), + ) + connection.execute( + "INSERT INTO confounders (timestamp, fear_greed_index) VALUES (?, ?)", + ((now - timedelta(minutes=5)).isoformat(), 20), + ) + connection.executemany( + """ + INSERT INTO price_data (timestamp, coin, price_usd, volume_24h) + VALUES (?, 'BTC', ?, 1000) + """, + [ + ((now - timedelta(hours=2)).isoformat(), 100.0), + ((now - timedelta(minutes=1)).isoformat(), 101.0), + ], + ) + + if sentiment_age_hours is not None: + cursor = connection.execute( + """ + INSERT INTO user_profiles (username, source, first_seen, last_seen) + VALUES ('alice', 'reddit_bitcoin', ?, ?) + """, + (now.isoformat(), now.isoformat()), + ) + connection.execute( + """ + INSERT INTO user_sentiment_scores ( + user_id, timestamp, coin, title_score, final_score, segments_scored + ) VALUES (?, ?, 'BTC', 0.6, 0.6, 4) + """, + ( + cursor.lastrowid, + (now - timedelta(hours=sentiment_age_hours)).isoformat(), + ), + ) + + connection.commit() + connection.close() + + +def test_inference_reads_current_tables_and_ignores_legacy_scores(tmp_path: Path) -> None: + db_path = tmp_path / "sentiment.db" + create_inference_database(db_path, sentiment_age_hours=1) + + analyzer = SentimentAnalyzer(str(db_path)) + frame = analyzer.get_recent_sentiment(hours=4, coin="BTC") + + assert set(frame["source"]) == {"reddit_bitcoin", "fear_greed"} + assert "legacy_source" not in set(frame["source"]) + assert frame.loc[frame["source"] == "reddit_bitcoin", "score"].iloc[0] == 0.6 + assert frame.loc[frame["source"] == "fear_greed", "score"].iloc[0] == -0.6 + assert analyzer.compute_aggregate_sentiment(hours=4, coin="BTC")["n_social_sources"] == 1 + + +def test_direction_is_suppressed_without_fresh_user_sentiment(tmp_path: Path) -> None: + db_path = tmp_path / "sentiment.db" + create_inference_database(db_path, sentiment_age_hours=48) + + prediction = PricePredictor(str(db_path)).predict(coin="BTC", lookback_hours=4) + + assert prediction.predicted_direction == "neutral" + assert prediction.confidence == 0.0 + assert prediction.signals["sentiment"]["n_social_sources"] == 0 + assert "suppressed" in prediction.reasoning diff --git a/tests/test_scheduler_belief_job.py b/tests/test_scheduler_belief_job.py index 79e5dde..47e012e 100644 --- a/tests/test_scheduler_belief_job.py +++ b/tests/test_scheduler_belief_job.py @@ -1,9 +1,10 @@ """Tests for the belief update scheduler job.""" import asyncio +from datetime import timedelta from unittest.mock import AsyncMock, patch -from crypto_sentiment_crawler.scheduler import CrawlerScheduler +from crypto_sentiment_crawler.scheduler import JOB_STAGGER_SECONDS, CrawlerScheduler def test_belief_update_job_success(): @@ -39,3 +40,45 @@ def test_belief_update_job_registered(): job = s.scheduler.get_job("belief_update") assert job is not None assert job.name == "Belief Update & Weights Sync" + + +def test_interval_jobs_use_one_staggered_anchor(): + """Recurring writers receive deterministic offsets from one anchor.""" + scheduler = CrawlerScheduler() + scheduler._setup_jobs() + + intervals = { + "crawl": scheduler.crawl_interval, + "price": scheduler.price_interval, + "evaluate": scheduler.eval_interval, + "fear_greed": scheduler.fear_greed_interval, + "confounders": scheduler.confounder_interval, + "onchain": scheduler.onchain_interval, + "belief_update": 1800, + "stats": 600, + } + inferred_anchors = { + scheduler.scheduler.get_job(job_id).trigger.start_date + - timedelta(seconds=interval + JOB_STAGGER_SECONDS[job_id]) + for job_id, interval in intervals.items() + } + + assert len(inferred_anchors) == 1 + starts = { + scheduler.scheduler.get_job(job_id).trigger.start_date + for job_id in intervals + } + assert len(starts) == len(intervals) + + anchor = inferred_anchors.pop() + scheduled_writes = {} + for job_id, interval in intervals.items(): + if job_id == "stats": + continue + fire_time = scheduler.scheduler.get_job(job_id).trigger.start_date + while fire_time <= anchor + timedelta(hours=8): + assert fire_time not in scheduled_writes, ( + f"{job_id} collides with {scheduled_writes.get(fire_time)}" + ) + scheduled_writes[fire_time] = job_id + fire_time += timedelta(seconds=interval) diff --git a/tests/test_semantic_runtime.py b/tests/test_semantic_runtime.py new file mode 100644 index 0000000..d292da2 --- /dev/null +++ b/tests/test_semantic_runtime.py @@ -0,0 +1,144 @@ +"""Tests for fail-closed and batched semantic scoring.""" + +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import numpy as np +import pytest + +from crypto_sentiment_crawler.processing.semantic_sentiment import ( + SemanticSentimentAnalyzer, +) +from crypto_sentiment_crawler.processing.user_sentiment import UserSentimentScorer + + +class DeterministicProvider: + """Small deterministic provider used without downloading a model.""" + + dim = 4 + + def __init__(self) -> None: + self.batch_calls: list[list[str]] = [] + self.single_calls: list[str] = [] + + @staticmethod + def _vector(text: str) -> np.ndarray: + values = np.array( + [ + len(text) + 1, + sum(ord(char) for char in text) % 97 + 1, + text.lower().count("a") + 1, + text.lower().count("e") + 1, + ], + dtype=np.float32, + ) + return values / np.linalg.norm(values) + + def encode(self, texts: list[str] | str, normalize: bool = True) -> np.ndarray: + values = [texts] if isinstance(texts, str) else texts + self.batch_calls.append(list(values)) + return np.stack([self._vector(text) for text in values]) + + def encode_single(self, text: str, normalize: bool = True) -> np.ndarray: + self.single_calls.append(text) + return self._vector(text) + + +def test_configured_openrouter_never_falls_back_to_local() -> None: + configured = SimpleNamespace( + embedding_backend="openrouter", + embedding_model="hosted/model", + openrouter_api_key="", + ) + + with ( + patch("crypto_sentiment_crawler.config.settings", configured), + patch( + "crypto_sentiment_crawler.processing.embedding_providers." + "OpenRouterEmbeddingProvider", + side_effect=RuntimeError("missing key"), + ), + patch( + "crypto_sentiment_crawler.processing.embedding_providers." + "LocalSentenceTransformerProvider" + ) as local_provider, + pytest.raises(RuntimeError, match="missing key"), + ): + SemanticSentimentAnalyzer() + + local_provider.assert_not_called() + + +def test_unknown_configured_backend_is_rejected() -> None: + configured = SimpleNamespace( + embedding_backend="openruter", + embedding_model="hosted/model", + openrouter_api_key="", + ) + + with ( + patch("crypto_sentiment_crawler.config.settings", configured), + patch( + "crypto_sentiment_crawler.processing.embedding_providers." + "LocalSentenceTransformerProvider" + ) as local_provider, + pytest.raises(ValueError, match="Unknown embedding backend"), + ): + SemanticSentimentAnalyzer() + + local_provider.assert_not_called() + + +@pytest.mark.parametrize("method", ["centroid", "top_k", "asymmetric"]) +def test_batch_analysis_matches_individual_scoring(method: str) -> None: + provider = DeterministicProvider() + analyzer = SemanticSentimentAnalyzer(provider=provider) + texts = ["Bitcoin looks strong", "The market may crash"] + + provider.batch_calls.clear() + individual = [analyzer.analyze(text, method=method) for text in texts] + provider.batch_calls.clear() + batched = analyzer.analyze_batch(texts, method=method) + + assert batched == individual + assert provider.batch_calls == [texts] + + +def test_user_scorer_batches_title_and_segments() -> None: + analyzer = MagicMock() + analyzer.analyze_batch.side_effect = lambda texts, method: [ + {"score": (index + 1) / 10} for index, _ in enumerate(texts) + ] + + with patch( + "crypto_sentiment_crawler.processing.semantic_sentiment." + "SemanticSentimentAnalyzer", + return_value=analyzer, + ): + scorer = UserSentimentScorer() + + raw_data = { + "author": "alice", + "title": "Bitcoin market outlook today", + "content": "The market has strong momentum today.", + "metadata": { + "comments": [ + {"author": f"user-{index}", "body": f"Distinct human comment number {index}."} + for index in range(4) + ] + }, + } + score = scorer.score_post( + raw_data, + raw_id=1, + timestamp="2026-08-14T12:00:00+00:00", + source="reddit_bitcoin", + ) + + assert score is not None + analyzer.analyze.assert_not_called() + analyzer.analyze_batch.assert_called_once() + texts, = analyzer.analyze_batch.call_args.args + assert texts[0] == raw_data["title"] + assert len(texts) == 6 + assert analyzer.analyze_batch.call_args.kwargs == {"method": "asymmetric"} diff --git a/tests/test_sqlite_runtime.py b/tests/test_sqlite_runtime.py new file mode 100644 index 0000000..742aff6 --- /dev/null +++ b/tests/test_sqlite_runtime.py @@ -0,0 +1,184 @@ +"""Tests for shared SQLite runtime settings.""" + +import sqlite3 +import threading +import time +from datetime import datetime, timezone +from pathlib import Path + +import pytest + +from crypto_sentiment_crawler.processing.user_sentiment import ( + PostScore, + SegmentScore, + UserSentimentScorer, +) +from crypto_sentiment_crawler.sqlite_utils import ( + SQLITE_BUSY_TIMEOUT_MS, + connect_sqlite, + sqlite_transaction, +) +from crypto_sentiment_crawler.storage.db import SCHEMA, Database +from crypto_sentiment_crawler.storage.models import PriceData + + +def test_sync_connection_sets_busy_timeout(tmp_path: Path) -> None: + connection = connect_sqlite(tmp_path / "sync.db") + try: + timeout = connection.execute("PRAGMA busy_timeout").fetchone()[0] + finally: + connection.close() + + assert timeout == SQLITE_BUSY_TIMEOUT_MS + + +def test_waiting_sync_writer_succeeds_after_lock_release(tmp_path: Path) -> None: + db_path = tmp_path / "wait.db" + with sqlite_transaction(db_path) as connection: + connection.execute("CREATE TABLE events (value INTEGER)") + + blocker = connect_sqlite(db_path) + blocker.execute("BEGIN IMMEDIATE") + blocker.execute("INSERT INTO events VALUES (1)") + started = threading.Event() + errors: list[Exception] = [] + + def write_from_thread() -> None: + started.set() + try: + with sqlite_transaction(db_path) as connection: + connection.execute("INSERT INTO events VALUES (2)") + except Exception as exc: # pragma: no cover - asserted through errors + errors.append(exc) + + worker = threading.Thread(target=write_from_thread) + worker.start() + assert started.wait(timeout=1) + time.sleep(0.05) + assert worker.is_alive() + blocker.commit() + blocker.close() + worker.join(timeout=2) + + assert not worker.is_alive() + assert errors == [] + connection = connect_sqlite(db_path) + try: + assert connection.execute("SELECT COUNT(*) FROM events").fetchone()[0] == 2 + finally: + connection.close() + + +def test_sync_transaction_rolls_back_on_error(tmp_path: Path) -> None: + db_path = tmp_path / "rollback.db" + with sqlite_transaction(db_path) as connection: + connection.execute("CREATE TABLE events (value INTEGER)") + + with pytest.raises(RuntimeError, match="abort"): + with sqlite_transaction(db_path) as connection: + connection.execute("INSERT INTO events VALUES (1)") + raise RuntimeError("abort") + + connection = connect_sqlite(db_path) + try: + assert connection.execute("SELECT COUNT(*) FROM events").fetchone()[0] == 0 + finally: + connection.close() + + +def test_user_score_writes_commit_through_safe_transactions(tmp_path: Path) -> None: + db_path = tmp_path / "user-score.db" + with sqlite_transaction(db_path) as connection: + connection.executescript(SCHEMA) + connection.execute( + """ + INSERT INTO sentiment_raw (id, timestamp, source, raw_data) + VALUES (1, '2026-08-14T12:00:00+00:00', 'reddit_bitcoin', '{}') + """ + ) + + scorer = object.__new__(UserSentimentScorer) + scorer.db_path = str(db_path) + post_score = PostScore( + raw_id=1, + timestamp="2026-08-14T12:00:00+00:00", + coin="BTC", + username="alice", + source="reddit_bitcoin", + title="Bitcoin outlook", + title_score=0.5, + body_score=0.25, + segment_scores=[SegmentScore("Bullish momentum", 0.25, 16)], + final_score=0.3, + aggregation_method="title_weighted", + pos_count=2, + neg_count=0, + neu_count=0, + segments_total=1, + segments_scored=1, + ) + + score_id = scorer.save_post_score(post_score) + assert score_id > 0 + assert scorer.save_post_score(post_score) == -1 + user_id = scorer.get_or_create_user( + post_score.username, + post_score.source, + post_score.timestamp, + ) + scorer.update_user_profile(user_id) + + connection = connect_sqlite(db_path) + try: + row = connection.execute( + "SELECT total_posts, avg_sentiment FROM user_profiles WHERE user_id = ?", + (user_id,), + ).fetchone() + finally: + connection.close() + + assert row == (1, 0.3) + + +@pytest.mark.asyncio +async def test_async_database_sets_busy_timeout(tmp_path: Path) -> None: + database = Database(tmp_path / "async.db") + await database.connect() + try: + cursor = await database.conn.execute("PRAGMA busy_timeout") + row = await cursor.fetchone() + finally: + await database.close() + + assert row[0] == SQLITE_BUSY_TIMEOUT_MS + + +@pytest.mark.asyncio +async def test_failed_commit_rolls_back_before_next_write(tmp_path: Path) -> None: + db_path = tmp_path / "locked.db" + database = Database(db_path) + await database.connect() + blocker = sqlite3.connect(db_path) + + try: + blocker.execute("BEGIN") + blocker.execute("SELECT * FROM price_data").fetchall() + await database.conn.execute("PRAGMA busy_timeout = 25") + + price = PriceData( + timestamp=datetime.now(timezone.utc), + coin="BTC", + price_usd=100.0, + ) + with pytest.raises(sqlite3.OperationalError, match="locked"): + await database.insert_price_data(price) + + assert not database.conn.in_transaction + + blocker.rollback() + await database.insert_price_data(price) + cursor = await database.conn.execute("SELECT COUNT(*) FROM price_data") + assert (await cursor.fetchone())[0] == 1 + finally: + blocker.close() + await database.close()