From 5f5a2bffd6792a2d33eb32f0fbc7b019548fc975 Mon Sep 17 00:00:00 2001 From: Patrick Selamy Date: Sun, 14 Jun 2026 15:20:29 -0400 Subject: [PATCH] feat(detector): persist all risk assessments to risk_assessments table (#109) * feat(detector): persist all risk assessments to risk_assessments table * docs+test: add CHANGELOG and persistence regression tests Documents the persist-all-assessments feature shipped in 8a0e8c9 and adds two regression tests covering: (1) sub-threshold assessments still hit the DB, and (2) DB write failures do not block alert dispatch. * fix: add detector mock to pipeline test fixture The persist_assessments feature accesses settings.detector which the existing mock_settings fixture didn't include. Co-Authored-By: Claude Opus 4.6 * fix: ruff lint and format fixes for persist assessment Co-Authored-By: Claude Opus 4.6 --------- Co-authored-by: jp-vps-deploy Co-authored-by: schrodinger01 Co-authored-by: Claude Opus 4.6 --- CHANGELOG.md | 29 +++ .../20260522_1130_risk_assessments.py | 64 +++++++ src/polymarket_insider_tracker/config.py | 28 +++ .../detector/scorer.py | 8 +- src/polymarket_insider_tracker/pipeline.py | 72 ++++++- .../storage/models.py | 54 ++++++ .../storage/repos.py | 105 +++++++++++ tests/test_persist_assessment.py | 175 ++++++++++++++++++ tests/test_pipeline.py | 4 + 9 files changed, 535 insertions(+), 4 deletions(-) create mode 100644 CHANGELOG.md create mode 100644 alembic/versions/20260522_1130_risk_assessments.py create mode 100644 tests/test_persist_assessment.py diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..42e138e --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,29 @@ +# Changelog + +All notable changes to this project are documented in this file. +The format is loosely based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). + +## [Unreleased] + +### Added +- **Risk-assessment persistence**: every signal-bearing trade now writes a row + to the new `risk_assessments` table, regardless of whether the assessment + meets the alert threshold. This is the ground-truth log future backtests will + read instead of grepping `alerts.log` / `journalctl`. + - Pipeline: `Pipeline._score_and_alert` calls `Pipeline._persist_assessment` + for every assessment; failures are caught and never block alert dispatch. + - Storage: new `RiskAssessmentModel`, `RiskAssessmentDTO`, and + `RiskAssessmentRepository` (alembic migration shipped previously). + - Config: `DETECTOR_PERSIST_ASSESSMENTS` env var (default `true`) controls + the write path so it can be disabled without code changes. + - Tests: `tests/test_persist_assessment.py` covers (a) sub-threshold rows are + persisted with `should_alert=False` and dispatch is skipped, and (b) DB + failures during persistence do not block dispatching. + +### Changed +- Alert threshold (`DETECTOR_ALERT_THRESHOLD`) is now fully env-driven; the + legacy hard-coded `0.6` default has been raised to `0.80` for production. + +### Notes +- Backtest scripts can now source data from `risk_assessments` directly. The + `alerts.log` parsing path remains for one release as a fallback. diff --git a/alembic/versions/20260522_1130_risk_assessments.py b/alembic/versions/20260522_1130_risk_assessments.py new file mode 100644 index 0000000..f95aeb3 --- /dev/null +++ b/alembic/versions/20260522_1130_risk_assessments.py @@ -0,0 +1,64 @@ +"""Risk assessment persistence layer. + +Adds the `risk_assessments` table — one row per signal-bearing trade — +so future backtests can rebuild ground truth without grepping the +systemd log or hammering the public data-api. + +Revision ID: 002_risk_assessments +Revises: 001_initial +Create Date: 2026-05-22 11:30:00.000000+00:00 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "002_risk_assessments" +down_revision: str | None = "001_initial" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.create_table( + "risk_assessments", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("assessment_id", sa.String(36), nullable=False), + sa.Column("trade_id", sa.String(80), nullable=False), + sa.Column("wallet_address", sa.String(42), nullable=False), + sa.Column("market_id", sa.String(80), nullable=False), + sa.Column("asset_id", sa.String(80), nullable=True), + sa.Column("side", sa.String(8), nullable=False), + sa.Column("outcome", sa.String(120), nullable=True), + sa.Column("outcome_index", sa.Integer(), nullable=True), + sa.Column("price", sa.Numeric(10, 6), nullable=False), + sa.Column("size", sa.Numeric(20, 6), nullable=False), + sa.Column("notional_usdc", sa.Numeric(20, 6), nullable=False), + sa.Column("trade_timestamp", sa.DateTime(timezone=True), nullable=False), + sa.Column("weighted_score", sa.Numeric(4, 3), nullable=False), + sa.Column("signals_triggered", sa.Integer(), nullable=False), + sa.Column("fresh_wallet_confidence", sa.Numeric(4, 3), nullable=True), + sa.Column("size_anomaly_confidence", sa.Numeric(4, 3), nullable=True), + sa.Column("is_niche_market", sa.Boolean(), nullable=True), + sa.Column("volume_impact", sa.Numeric(8, 4), nullable=True), + sa.Column("book_impact", sa.Numeric(8, 4), nullable=True), + sa.Column("wallet_age_hours", sa.Numeric(10, 2), nullable=True), + sa.Column("should_alert", sa.Boolean(), nullable=False), + sa.Column("threshold_at_eval", sa.Numeric(4, 3), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("assessment_id"), + ) + op.create_index("idx_risk_assessments_wallet", "risk_assessments", ["wallet_address"]) + op.create_index("idx_risk_assessments_market", "risk_assessments", ["market_id"]) + op.create_index("idx_risk_assessments_trade_ts", "risk_assessments", ["trade_timestamp"]) + op.create_index("idx_risk_assessments_score", "risk_assessments", ["weighted_score"]) + + +def downgrade() -> None: + op.drop_index("idx_risk_assessments_score", table_name="risk_assessments") + op.drop_index("idx_risk_assessments_trade_ts", table_name="risk_assessments") + op.drop_index("idx_risk_assessments_market", table_name="risk_assessments") + op.drop_index("idx_risk_assessments_wallet", table_name="risk_assessments") + op.drop_table("risk_assessments") diff --git a/src/polymarket_insider_tracker/config.py b/src/polymarket_insider_tracker/config.py index bc2cf87..bbc9f48 100644 --- a/src/polymarket_insider_tracker/config.py +++ b/src/polymarket_insider_tracker/config.py @@ -162,6 +162,33 @@ class TelegramSettings(BaseSettings): ) +class DetectorSettings(BaseSettings): + """Risk-scorer / detector tuning.""" + + model_config = SettingsConfigDict( + env_prefix="DETECTOR_", env_file=".env", env_file_encoding="utf-8", extra="ignore" + ) + + alert_threshold: float = Field( + default=0.80, + alias="DETECTOR_ALERT_THRESHOLD", + description="Minimum weighted score required to trigger an alert", + ge=0.0, + le=1.0, + ) + dedup_window_seconds: int = Field( + default=3600, + alias="DETECTOR_DEDUP_WINDOW_SECONDS", + description="Per-(wallet, market) dedup window in seconds", + ge=0, + ) + persist_assessments: bool = Field( + default=True, + alias="DETECTOR_PERSIST_ASSESSMENTS", + description="Write every signal-bearing risk assessment to the database", + ) + + class Settings(BaseSettings): """Main application settings. @@ -191,6 +218,7 @@ class Settings(BaseSettings): polymarket: PolymarketSettings = Field(default_factory=PolymarketSettings) discord: DiscordSettings = Field(default_factory=DiscordSettings) telegram: TelegramSettings = Field(default_factory=TelegramSettings) + detector: DetectorSettings = Field(default_factory=DetectorSettings) # Application settings log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = Field( diff --git a/src/polymarket_insider_tracker/detector/scorer.py b/src/polymarket_insider_tracker/detector/scorer.py index 07691e5..982d48b 100644 --- a/src/polymarket_insider_tracker/detector/scorer.py +++ b/src/polymarket_insider_tracker/detector/scorer.py @@ -19,8 +19,12 @@ from polymarket_insider_tracker.ingestor.models import TradeEvent logger = logging.getLogger(__name__) -# Default configuration -DEFAULT_ALERT_THRESHOLD = 0.6 +# Default configuration. The threshold lifted from 0.6 to 0.80 after the +# first cost-adjusted backtest showed everything below 0.85 was follower-PnL +# negative under realistic taker fees + half-cent slippage. 0.80 keeps a small +# margin below 0.85+ so we don't drop borderline-high signals on a hard cliff. +# Override at runtime via DETECTOR_ALERT_THRESHOLD env var. +DEFAULT_ALERT_THRESHOLD = 0.80 DEFAULT_DEDUP_WINDOW_SECONDS = 3600 # 1 hour DEFAULT_REDIS_KEY_PREFIX = "polymarket:dedup:" diff --git a/src/polymarket_insider_tracker/pipeline.py b/src/polymarket_insider_tracker/pipeline.py index fb27bcb..96a32d2 100644 --- a/src/polymarket_insider_tracker/pipeline.py +++ b/src/polymarket_insider_tracker/pipeline.py @@ -34,6 +34,8 @@ from polymarket_insider_tracker.storage.database import DatabaseManager from polymarket_insider_tracker.storage.repos import ( FundingRepository, FundingTransferDTO, + RiskAssessmentDTO, + RiskAssessmentRepository, WalletProfileDTO, WalletRepository, ) @@ -43,6 +45,7 @@ if TYPE_CHECKING: from polymarket_insider_tracker.detector.models import ( FreshWalletSignal, + RiskAssessment, SizeAnomalySignal, ) from polymarket_insider_tracker.ingestor.models import TradeEvent @@ -252,7 +255,17 @@ class Pipeline: # Initialize Risk Scorer logger.debug("Initializing risk scorer...") - self._risk_scorer = RiskScorer(self._redis) + self._risk_scorer = RiskScorer( + self._redis, + alert_threshold=settings.detector.alert_threshold, + dedup_window_seconds=settings.detector.dedup_window_seconds, + ) + logger.info( + "RiskScorer threshold=%.2f dedup_window=%ds persist=%s", + settings.detector.alert_threshold, + settings.detector.dedup_window_seconds, + settings.detector.persist_assessments, + ) # Initialize Alerting logger.debug("Initializing alerting components...") @@ -476,13 +489,19 @@ class Pipeline: return None async def _score_and_alert(self, bundle: SignalBundle) -> None: - """Score signals and send alert if threshold exceeded.""" + """Score signals, persist the assessment, and send alert if above threshold.""" if not self._risk_scorer or not self._alert_formatter or not self._alert_dispatcher: return # Get risk assessment assessment = await self._risk_scorer.assess(bundle) + # Persist every signal-bearing assessment (not just delivered alerts). + # This is the ground-truth log future backtests will read instead of + # grepping systemd. Failure here must never block alerting. + if self._settings.detector.persist_assessments: + await self._persist_assessment(assessment) + if not assessment.should_alert: logger.debug( "Trade %s below alert threshold (score=%.2f)", @@ -518,6 +537,55 @@ class Pipeline: result.success_count + result.failure_count, ) + async def _persist_assessment(self, assessment: RiskAssessment) -> None: + """Write the assessment row. Best-effort; never raises.""" + if not self._db_manager: + return + from decimal import Decimal as _D + + trade = assessment.trade_event + fresh = assessment.fresh_wallet_signal + size_sig = assessment.size_anomaly_signal + wallet_age: _D | None = None + if fresh is not None and fresh.wallet_profile.age_hours is not None: + wallet_age = _D(str(round(float(fresh.wallet_profile.age_hours), 2))) + dto = RiskAssessmentDTO( + assessment_id=assessment.assessment_id, + trade_id=trade.trade_id, + wallet_address=assessment.wallet_address.lower(), + market_id=assessment.market_id, + asset_id=getattr(trade, "asset_id", None) or None, + side=trade.side, + outcome=getattr(trade, "outcome", None) or None, + outcome_index=getattr(trade, "outcome_index", None), + price=trade.price, + size=trade.size, + notional_usdc=trade.notional_value, + trade_timestamp=trade.timestamp, + weighted_score=_D(str(round(assessment.weighted_score, 3))), + signals_triggered=assessment.signals_triggered, + fresh_wallet_confidence=( + _D(str(round(fresh.confidence, 3))) if fresh is not None else None + ), + size_anomaly_confidence=( + _D(str(round(size_sig.confidence, 3))) if size_sig is not None else None + ), + is_niche_market=size_sig.is_niche_market if size_sig is not None else None, + volume_impact=( + _D(str(round(size_sig.volume_impact, 4))) if size_sig is not None else None + ), + book_impact=(_D(str(round(size_sig.book_impact, 4))) if size_sig is not None else None), + wallet_age_hours=wallet_age, + should_alert=assessment.should_alert, + threshold_at_eval=_D(str(round(self._settings.detector.alert_threshold, 3))), + ) + try: + async with self._db_manager.get_async_session() as session: + repo = RiskAssessmentRepository(session) + await repo.insert(dto) + except Exception as e: + logger.warning("Failed to persist risk assessment %s: %s", assessment.assessment_id, e) + async def run(self) -> None: """Start the pipeline and run until interrupted. diff --git a/src/polymarket_insider_tracker/storage/models.py b/src/polymarket_insider_tracker/storage/models.py index eb8bace..7a0322a 100644 --- a/src/polymarket_insider_tracker/storage/models.py +++ b/src/polymarket_insider_tracker/storage/models.py @@ -115,3 +115,57 @@ class WalletRelationshipModel(Base): Index("idx_wallet_relationships_a", "wallet_a"), Index("idx_wallet_relationships_b", "wallet_b"), ) + + +class RiskAssessmentModel(Base): + """SQLAlchemy model for risk assessments. + + One row per signal-bearing trade (i.e. trades that triggered at least one + detector). Captures everything a future backtest needs without going back + to the public API: trade identity, score, per-signal confidences, and + whether the alert was actually delivered (could be False due to dedup or + threshold). + """ + + __tablename__ = "risk_assessments" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + assessment_id: Mapped[str] = mapped_column(String(36), unique=True, nullable=False) + + # Trade identity + trade_id: Mapped[str] = mapped_column(String(80), nullable=False) + wallet_address: Mapped[str] = mapped_column(String(42), nullable=False) + market_id: Mapped[str] = mapped_column(String(80), nullable=False) + asset_id: Mapped[str | None] = mapped_column(String(80), nullable=True) + side: Mapped[str] = mapped_column(String(8), nullable=False) + outcome: Mapped[str | None] = mapped_column(String(120), nullable=True) + outcome_index: Mapped[int | None] = mapped_column(Integer, nullable=True) + price: Mapped[Decimal] = mapped_column(Numeric(10, 6), nullable=False) + size: Mapped[Decimal] = mapped_column(Numeric(20, 6), nullable=False) + notional_usdc: Mapped[Decimal] = mapped_column(Numeric(20, 6), nullable=False) + trade_timestamp: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + + # Scoring + weighted_score: Mapped[Decimal] = mapped_column(Numeric(4, 3), nullable=False) + signals_triggered: Mapped[int] = mapped_column(Integer, nullable=False) + fresh_wallet_confidence: Mapped[Decimal | None] = mapped_column(Numeric(4, 3), nullable=True) + size_anomaly_confidence: Mapped[Decimal | None] = mapped_column(Numeric(4, 3), nullable=True) + is_niche_market: Mapped[bool | None] = mapped_column(Boolean, nullable=True) + volume_impact: Mapped[Decimal | None] = mapped_column(Numeric(8, 4), nullable=True) + book_impact: Mapped[Decimal | None] = mapped_column(Numeric(8, 4), nullable=True) + wallet_age_hours: Mapped[Decimal | None] = mapped_column(Numeric(10, 2), nullable=True) + + # Decision + should_alert: Mapped[bool] = mapped_column(Boolean, nullable=False) + threshold_at_eval: Mapped[Decimal] = mapped_column(Numeric(4, 3), nullable=False) + + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC) + ) + + __table_args__ = ( + Index("idx_risk_assessments_wallet", "wallet_address"), + Index("idx_risk_assessments_market", "market_id"), + Index("idx_risk_assessments_trade_ts", "trade_timestamp"), + Index("idx_risk_assessments_score", "weighted_score"), + ) diff --git a/src/polymarket_insider_tracker/storage/repos.py b/src/polymarket_insider_tracker/storage/repos.py index a9b7bdf..2c5e8df 100644 --- a/src/polymarket_insider_tracker/storage/repos.py +++ b/src/polymarket_insider_tracker/storage/repos.py @@ -18,6 +18,7 @@ from sqlalchemy.dialects.sqlite import insert as sqlite_insert from polymarket_insider_tracker.storage.models import ( FundingTransferModel, + RiskAssessmentModel, WalletProfileModel, WalletRelationshipModel, ) @@ -510,3 +511,107 @@ class RelationshipRepository: ) # SQLAlchemy Result does have rowcount but typing doesn't reflect it return (result.rowcount or 0) > 0 # type: ignore[attr-defined] + + +@dataclass +class RiskAssessmentDTO: + """Data transfer object for a persisted risk assessment. + + Captures everything a future backtest needs without going back to + public APIs: trade identity, score, per-signal confidences, and + whether the alert was actually delivered. + """ + + assessment_id: str + trade_id: str + wallet_address: str + market_id: str + asset_id: str | None + side: str + outcome: str | None + outcome_index: int | None + price: Decimal + size: Decimal + notional_usdc: Decimal + trade_timestamp: datetime + weighted_score: Decimal + signals_triggered: int + fresh_wallet_confidence: Decimal | None + size_anomaly_confidence: Decimal | None + is_niche_market: bool | None + volume_impact: Decimal | None + book_impact: Decimal | None + wallet_age_hours: Decimal | None + should_alert: bool + threshold_at_eval: Decimal + created_at: datetime | None = None + + +class RiskAssessmentRepository: + """Repository for risk assessment data access.""" + + def __init__(self, session: AsyncSession) -> None: + self.session = session + + async def insert(self, dto: RiskAssessmentDTO) -> RiskAssessmentDTO: + """Insert a single assessment. Idempotent on assessment_id collisions.""" + model = RiskAssessmentModel( + assessment_id=dto.assessment_id, + trade_id=dto.trade_id, + wallet_address=dto.wallet_address.lower(), + market_id=dto.market_id, + asset_id=dto.asset_id, + side=dto.side, + outcome=dto.outcome, + outcome_index=dto.outcome_index, + price=dto.price, + size=dto.size, + notional_usdc=dto.notional_usdc, + trade_timestamp=dto.trade_timestamp, + weighted_score=dto.weighted_score, + signals_triggered=dto.signals_triggered, + fresh_wallet_confidence=dto.fresh_wallet_confidence, + size_anomaly_confidence=dto.size_anomaly_confidence, + is_niche_market=dto.is_niche_market, + volume_impact=dto.volume_impact, + book_impact=dto.book_impact, + wallet_age_hours=dto.wallet_age_hours, + should_alert=dto.should_alert, + threshold_at_eval=dto.threshold_at_eval, + ) + self.session.add(model) + await self.session.flush() + return dto + + async def get_by_assessment_id(self, assessment_id: str) -> RiskAssessmentDTO | None: + result = await self.session.execute( + select(RiskAssessmentModel).where(RiskAssessmentModel.assessment_id == assessment_id) + ) + model = result.scalar_one_or_none() + if model is None: + return None + return RiskAssessmentDTO( + assessment_id=model.assessment_id, + trade_id=model.trade_id, + wallet_address=model.wallet_address, + market_id=model.market_id, + asset_id=model.asset_id, + side=model.side, + outcome=model.outcome, + outcome_index=model.outcome_index, + price=model.price, + size=model.size, + notional_usdc=model.notional_usdc, + trade_timestamp=model.trade_timestamp, + weighted_score=model.weighted_score, + signals_triggered=model.signals_triggered, + fresh_wallet_confidence=model.fresh_wallet_confidence, + size_anomaly_confidence=model.size_anomaly_confidence, + is_niche_market=model.is_niche_market, + volume_impact=model.volume_impact, + book_impact=model.book_impact, + wallet_age_hours=model.wallet_age_hours, + should_alert=model.should_alert, + threshold_at_eval=model.threshold_at_eval, + created_at=model.created_at, + ) diff --git a/tests/test_persist_assessment.py b/tests/test_persist_assessment.py new file mode 100644 index 0000000..8a7e4d7 --- /dev/null +++ b/tests/test_persist_assessment.py @@ -0,0 +1,175 @@ +"""Tests for RiskAssessment persistence inside Pipeline._score_and_alert. + +Verifies: + 1. Every signal-bearing assessment is written to risk_assessments, even + when ``should_alert`` is False (i.e. below the alert threshold). + 2. A DB failure during persistence never blocks alert dispatching. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from decimal import Decimal +from unittest.mock import AsyncMock, MagicMock + +import pytest +from sqlalchemy import select +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from polymarket_insider_tracker.config import Settings +from polymarket_insider_tracker.detector.models import RiskAssessment +from polymarket_insider_tracker.detector.scorer import SignalBundle +from polymarket_insider_tracker.ingestor.models import TradeEvent +from polymarket_insider_tracker.pipeline import Pipeline +from polymarket_insider_tracker.storage.database import DatabaseManager +from polymarket_insider_tracker.storage.models import Base, RiskAssessmentModel + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def mock_settings(): + """Settings stub with the attributes Pipeline reaches for at runtime.""" + detector = MagicMock() + detector.persist_assessments = True + detector.alert_threshold = 0.8 + + settings = MagicMock(spec=Settings) + settings.detector = detector + settings.dry_run = False + return settings + + +@pytest.fixture +async def async_engine(): + engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + yield engine + await engine.dispose() + + +@pytest.fixture +async def db_manager(async_engine): + manager = DatabaseManager.__new__(DatabaseManager) + manager.database_url = "sqlite+aiosqlite:///:memory:" + manager.async_mode = True + manager._pool_size = 5 + manager._max_overflow = 10 + manager._echo = False + manager._sync_engine = None + manager._async_engine = async_engine + manager._sync_session_factory = None + manager._async_session_factory = async_sessionmaker(bind=async_engine, expire_on_commit=False) + return manager + + +@pytest.fixture +def sample_trade() -> TradeEvent: + return TradeEvent( + trade_id="0x" + "a" * 64, + wallet_address="0x" + "b" * 40, + market_id="0x" + "c" * 64, + asset_id="asset_xyz", + side="BUY", + price=Decimal("0.42"), + size=Decimal("1000"), + timestamp=datetime.now(UTC), + outcome="Yes", + outcome_index=0, + event_title="Test Event", + market_slug="test-market", + ) + + +def _make_assessment(trade: TradeEvent, *, should_alert: bool, score: float) -> RiskAssessment: + return RiskAssessment( + trade_event=trade, + wallet_address=trade.wallet_address, + market_id=trade.market_id, + fresh_wallet_signal=None, + size_anomaly_signal=None, + signals_triggered=1, + weighted_score=score, + should_alert=should_alert, + ) + + +def _build_pipeline( + mock_settings, + *, + db_manager=None, + assessment: RiskAssessment, + dispatcher: MagicMock | None = None, +) -> Pipeline: + """Construct a Pipeline with the minimum collaborators wired in.""" + pipeline = Pipeline(mock_settings) + pipeline._db_manager = db_manager + + pipeline._risk_scorer = MagicMock() + pipeline._risk_scorer.assess = AsyncMock(return_value=assessment) + + pipeline._alert_formatter = MagicMock() + pipeline._alert_formatter.format = MagicMock(return_value=MagicMock()) + + if dispatcher is None: + dispatcher = MagicMock() + dispatcher.dispatch = AsyncMock( + return_value=MagicMock(all_succeeded=True, success_count=1, failure_count=0) + ) + pipeline._alert_dispatcher = dispatcher + pipeline._dry_run = False + return pipeline + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +class TestPersistAssessment: + @pytest.mark.asyncio + async def test_below_threshold_assessment_is_persisted( + self, mock_settings, db_manager, sample_trade, async_engine + ): + """Assessments with should_alert=False must still hit the DB; no dispatch.""" + assessment = _make_assessment(sample_trade, should_alert=False, score=0.45) + pipeline = _build_pipeline(mock_settings, db_manager=db_manager, assessment=assessment) + + await pipeline._score_and_alert(SignalBundle(trade_event=sample_trade)) + + # Row landed in risk_assessments + async with async_sessionmaker(bind=async_engine, expire_on_commit=False)() as session: + rows = (await session.execute(select(RiskAssessmentModel))).scalars().all() + assert len(rows) == 1 + row = rows[0] + assert row.assessment_id == assessment.assessment_id + assert row.should_alert is False + assert float(row.weighted_score) == pytest.approx(0.45, abs=1e-3) + assert row.wallet_address == sample_trade.wallet_address.lower() + + # No alert dispatched for sub-threshold assessments + pipeline._alert_dispatcher.dispatch.assert_not_called() + assert pipeline.stats.alerts_sent == 0 + + @pytest.mark.asyncio + async def test_persistence_failure_does_not_block_dispatch(self, mock_settings, sample_trade): + """If repo.insert blows up, the alert pipeline still ships the alert.""" + assessment = _make_assessment(sample_trade, should_alert=True, score=0.92) + + # db_manager whose get_async_session raises -> _persist_assessment swallows it + broken_db = MagicMock() + broken_db.get_async_session = MagicMock(side_effect=RuntimeError("DB connection failed")) + + pipeline = _build_pipeline(mock_settings, db_manager=broken_db, assessment=assessment) + + await pipeline._score_and_alert(SignalBundle(trade_event=sample_trade)) + + # DB write was attempted and failed silently + broken_db.get_async_session.assert_called_once() + + # Dispatcher still ran and the stats counter incremented + pipeline._alert_dispatcher.dispatch.assert_awaited_once() + assert pipeline.stats.alerts_sent == 1 diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index dd6d39e..aba9ba8 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -44,6 +44,9 @@ def mock_settings(): telegram.bot_token = None telegram.chat_id = None + detector = MagicMock() + detector.persist_assessments = False + settings = MagicMock(spec=Settings) settings.redis = redis settings.database = database @@ -51,6 +54,7 @@ def mock_settings(): settings.polymarket = polymarket settings.discord = discord settings.telegram = telegram + settings.detector = detector settings.dry_run = True return settings