from __future__ import annotations import hashlib import json import sqlite3 import threading import uuid from contextlib import closing from datetime import datetime, timedelta, timezone from decimal import Decimal, InvalidOperation from pathlib import Path from typing import Any ACTIVE = "ACTIVE" EXPIRED = "EXPIRED" INVALIDATED = "INVALIDATED" def _decimal(value: object) -> Decimal: try: return Decimal(str(value)) except (InvalidOperation, TypeError, ValueError): return Decimal("0") def _utc(value: datetime | None = None) -> datetime: current = value or datetime.now(timezone.utc) return current.astimezone(timezone.utc) class SignalStore: """Persistent lifecycle registry for sanitized, decision-only signals.""" def __init__( self, database_file: Path, *, ttl_minutes: int = 30, meaningful_price_change_percent: Decimal = Decimal("1"), meaningful_score_change: Decimal = Decimal("0.05"), ) -> None: self._database_file = database_file self._ttl_minutes = max(2, min(int(ttl_minutes), 240)) self._meaningful_price_change_percent = meaningful_price_change_percent self._meaningful_score_change = meaningful_score_change self._lock = threading.Lock() def _connect(self) -> sqlite3.Connection: connection = sqlite3.connect(self._database_file, timeout=15) connection.row_factory = sqlite3.Row connection.execute("PRAGMA foreign_keys = ON") connection.execute("PRAGMA busy_timeout = 15000") connection.execute("PRAGMA journal_mode = WAL") self._ensure_schema(connection) return connection @staticmethod def _ensure_schema(connection: sqlite3.Connection) -> None: connection.executescript( """ CREATE TABLE IF NOT EXISTS signals ( id TEXT PRIMARY KEY, symbol TEXT NOT NULL, fingerprint TEXT NOT NULL, status TEXT NOT NULL CHECK (status IN ('ACTIVE', 'EXPIRED', 'INVALIDATED')), created_at TEXT NOT NULL, updated_at TEXT NOT NULL, last_seen_at TEXT NOT NULL, expires_at TEXT NOT NULL, invalidated_at TEXT, invalidation_reason TEXT, entry_low TEXT NOT NULL, entry_high TEXT NOT NULL, stop_loss TEXT NOT NULL, target_1 TEXT NOT NULL, target_2 TEXT NOT NULL, current_price TEXT NOT NULL, combined_score TEXT NOT NULL, confidence_percent TEXT NOT NULL, snapshot_json TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_signals_symbol_created ON signals(symbol, created_at DESC); CREATE INDEX IF NOT EXISTS idx_signals_status ON signals(status); CREATE UNIQUE INDEX IF NOT EXISTS idx_signals_one_active_symbol ON signals(symbol) WHERE status = 'ACTIVE'; CREATE TABLE IF NOT EXISTS signal_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, signal_id TEXT NOT NULL, event_type TEXT NOT NULL, created_at TEXT NOT NULL, payload_json TEXT NOT NULL, FOREIGN KEY(signal_id) REFERENCES signals(id) ); CREATE INDEX IF NOT EXISTS idx_signal_events_created ON signal_events(created_at DESC); """ ) @staticmethod def _snapshot(analysis: dict[str, Any]) -> dict[str, Any]: levels = analysis.get("levels") if isinstance(analysis.get("levels"), dict) else {} return { "symbol": str(analysis.get("symbol", "")), "current_price": str(analysis.get("current_price", "0")), "combined_score": str(analysis.get("combined_score", "0")), "confidence_percent": str(analysis.get("confidence_percent", "0")), "entry_low": str(levels.get("entry_low", "0")), "entry_high": str(levels.get("entry_high", "0")), "stop_loss": str(levels.get("stop_loss", "0")), "target_1": str(levels.get("target_1", "0")), "target_2": str(levels.get("target_2", "0")), } @staticmethod def _fingerprint(snapshot: dict[str, Any]) -> str: stable = { key: snapshot[key] for key in ("symbol", "entry_low", "entry_high", "stop_loss", "target_1", "target_2") } encoded = json.dumps(stable, sort_keys=True, separators=(",", ":")).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def _meaningfully_changed(self, previous: sqlite3.Row, snapshot: dict[str, Any]) -> bool: previous_mid = (_decimal(previous["entry_low"]) + _decimal(previous["entry_high"])) / Decimal("2") current_mid = (_decimal(snapshot["entry_low"]) + _decimal(snapshot["entry_high"])) / Decimal("2") comparisons = ( (previous_mid, current_mid), (_decimal(previous["stop_loss"]), _decimal(snapshot["stop_loss"])), (_decimal(previous["target_1"]), _decimal(snapshot["target_1"])), ) for old, new in comparisons: if old > 0 and (abs(new - old) / old) * Decimal("100") >= self._meaningful_price_change_percent: return True return abs(_decimal(snapshot["combined_score"]) - _decimal(previous["combined_score"])) >= self._meaningful_score_change @staticmethod def _invalidation_reason(analysis: dict[str, Any] | None) -> str: if analysis is None: return "SYMBOL_NOT_ANALYZED" failed = analysis.get("quality_gates") if isinstance(failed, dict): names = [str(name) for name, passed in failed.items() if passed is False] if names: return "FAILED_GATES:" + ",".join(names) return "DECISION_CHANGED:" + str(analysis.get("decision", "UNKNOWN")) @staticmethod def _public_signal(row: sqlite3.Row) -> dict[str, Any]: return { "id": row["id"], "symbol": row["symbol"], "status": row["status"], "created_at": row["created_at"], "updated_at": row["updated_at"], "last_seen_at": row["last_seen_at"], "expires_at": row["expires_at"], "invalidation_reason": row["invalidation_reason"], "entry_low": row["entry_low"], "entry_high": row["entry_high"], "stop_loss": row["stop_loss"], "target_1": row["target_1"], "target_2": row["target_2"], "current_price": row["current_price"], "combined_score": row["combined_score"], "confidence_percent": row["confidence_percent"], } def process( self, analyses: list[dict[str, Any]], *, now: datetime | None = None, ttl_minutes: int | None = None, gatekeeper: Any = None, ) -> dict[str, Any]: """Update the signal lifecycle. ``gatekeeper`` is an optional callable ``(symbol) -> (allowed, reason)`` consulted for every BUY_SETUP candidate, including symbols that already hold an active signal. A blocked candidate is never created, and an active signal for a blocked symbol is invalidated with the given reason. Typical use: a beginner filter that rejects very small or very young projects. """ current_time = _utc(now) now_text = current_time.isoformat() effective_ttl = ( self._ttl_minutes if ttl_minutes is None else max(2, min(int(ttl_minutes), 240)) ) analyses_by_symbol = { str(item.get("symbol", "")): item for item in analyses if item.get("symbol") } qualified: dict[str, dict[str, Any]] = {} blocked: dict[str, str] = {} for symbol, item in analyses_by_symbol.items(): if item.get("decision") != "BUY_SETUP": continue if gatekeeper is not None: allowed, reason = gatekeeper(symbol) if not allowed: blocked[symbol] = str(reason or "BLOCKED") continue qualified[symbol] = item created_events: list[dict[str, Any]] = [] signal_for_symbol: dict[str, str] = {} with self._lock, closing(self._connect()) as connection: active_rows = connection.execute( "SELECT * FROM signals WHERE status = 'ACTIVE' ORDER BY created_at" ).fetchall() active_by_symbol = {str(row["symbol"]): row for row in active_rows} for symbol, row in active_by_symbol.items(): expires_at = datetime.fromisoformat(str(row["expires_at"])) if current_time >= expires_at: connection.execute( "UPDATE signals SET status = ?, updated_at = ? WHERE id = ?", (EXPIRED, now_text, row["id"]), ) event = {"signal_id": row["id"], "symbol": symbol, "event": "SIGNAL_EXPIRED"} connection.execute( "INSERT INTO signal_events(signal_id, event_type, created_at, payload_json) VALUES (?, ?, ?, ?)", (row["id"], event["event"], now_text, json.dumps(event, ensure_ascii=False)), ) created_events.append(event) continue analysis = qualified.get(symbol) if analysis is None: if symbol in blocked: reason = "BEGINNER_FILTER:" + blocked[symbol] else: reason = self._invalidation_reason(analyses_by_symbol.get(symbol)) connection.execute( "UPDATE signals SET status = ?, updated_at = ?, invalidated_at = ?, invalidation_reason = ? WHERE id = ?", (INVALIDATED, now_text, now_text, reason, row["id"]), ) event = { "signal_id": row["id"], "symbol": symbol, "event": "SIGNAL_INVALIDATED", "reason": reason, } connection.execute( "INSERT INTO signal_events(signal_id, event_type, created_at, payload_json) VALUES (?, ?, ?, ?)", (row["id"], event["event"], now_text, json.dumps(event, ensure_ascii=False)), ) created_events.append(event) continue snapshot = self._snapshot(analysis) connection.execute( "UPDATE signals SET updated_at = ?, last_seen_at = ?, current_price = ?, combined_score = ?, confidence_percent = ?, snapshot_json = ? WHERE id = ?", ( now_text, now_text, snapshot["current_price"], snapshot["combined_score"], snapshot["confidence_percent"], json.dumps(snapshot, ensure_ascii=False, separators=(",", ":")), row["id"], ), ) signal_for_symbol[symbol] = str(row["id"]) for symbol, analysis in qualified.items(): if symbol in signal_for_symbol: continue snapshot = self._snapshot(analysis) latest = connection.execute( "SELECT * FROM signals WHERE symbol = ? ORDER BY created_at DESC LIMIT 1", (symbol,), ).fetchone() if latest is not None and latest["status"] == EXPIRED and not self._meaningfully_changed(latest, snapshot): continue signal_id = uuid.uuid4().hex expires_at = current_time + timedelta(minutes=effective_ttl) fingerprint = self._fingerprint(snapshot) connection.execute( """ INSERT INTO signals( id, symbol, fingerprint, status, created_at, updated_at, last_seen_at, expires_at, entry_low, entry_high, stop_loss, target_1, target_2, current_price, combined_score, confidence_percent, snapshot_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( signal_id, symbol, fingerprint, ACTIVE, now_text, now_text, now_text, expires_at.isoformat(), snapshot["entry_low"], snapshot["entry_high"], snapshot["stop_loss"], snapshot["target_1"], snapshot["target_2"], snapshot["current_price"], snapshot["combined_score"], snapshot["confidence_percent"], json.dumps(snapshot, ensure_ascii=False, separators=(",", ":")), ), ) event = {"signal_id": signal_id, "symbol": symbol, "event": "NEW_STRONG_SIGNAL"} connection.execute( "INSERT INTO signal_events(signal_id, event_type, created_at, payload_json) VALUES (?, ?, ?, ?)", (signal_id, event["event"], now_text, json.dumps(event, ensure_ascii=False)), ) created_events.append(event) signal_for_symbol[symbol] = signal_id active = connection.execute( "SELECT * FROM signals WHERE status = 'ACTIVE' ORDER BY created_at DESC" ).fetchall() connection.commit() return { "events": created_events, "active_signals": [self._public_signal(row) for row in active], "signal_for_symbol": signal_for_symbol, "ttl_minutes": effective_ttl, "blocked_signals": [ { "symbol": symbol, "status": "BLOCKED", "reason": "BEGINNER_FILTER", "detail": reason, } for symbol, reason in sorted(blocked.items()) ], } def recent(self, limit: int = 100) -> list[dict[str, Any]]: safe_limit = max(1, min(int(limit), 500)) with self._lock, closing(self._connect()) as connection: rows = connection.execute( "SELECT * FROM signals ORDER BY created_at DESC LIMIT ?", (safe_limit,) ).fetchall() return [self._public_signal(row) for row in rows]