fix: ruff lint and format fixes for persist assessment

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

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.

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")

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(

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:"
 

src/polymarket_insider_tracker/ingestor/gamma_client.py

@@ -1,205 +0,0 @@
-"""Gamma API client for Polymarket market volume / liquidity data.
-
-The CLOB API does not expose 24h volume or liquidity. The public Gamma API
-(https://gamma-api.polymarket.com) does, with no auth required. This module
-fetches the volume/liquidity snapshot keyed by condition_id so the
-size_anomaly detector can do real ratio math instead of falling back to
-the niche-base 0.2 confidence floor.
-"""
-
-from __future__ import annotations
-
-import asyncio
-import logging
-from dataclasses import dataclass
-from decimal import Decimal, InvalidOperation
-
-import httpx
-
-logger = logging.getLogger(__name__)
-
-DEFAULT_HOST = "https://gamma-api.polymarket.com"
-DEFAULT_TIMEOUT_SECONDS = 15.0
-# Gamma /markets enforces a server-side max of 100 per page even when a
-# higher `limit` is sent. Using 100 lines our page size up with the actual
-# response so pagination doesn't bail out after the first page.
-DEFAULT_PAGE_LIMIT = 100
-# Gamma also caps `offset` around 10000 for this collection. Combined with
-# the 100/page limit that gives ~10k markets max, sequential — way too slow
-# at default sync interval. We sort by 24h volume desc and only walk the
-# top N pages, since markets with zero recent volume don't need a real
-# ratio anyway (the niche path handles them).
-DEFAULT_MAX_PAGES = 50  # 50 * 100 = 5000 most-traded markets per sync
-DEFAULT_PAGE_CONCURRENCY = 5
-DEFAULT_MAX_RETRIES = 3
-DEFAULT_RETRY_BASE_DELAY_SECONDS = 1.0
-
-
-@dataclass(frozen=True)
-class GammaMarketStats:
-    """Volume / liquidity snapshot for a single market from gamma-api."""
-
-    condition_id: str
-    daily_volume: Decimal | None
-    weekly_volume: Decimal | None
-    monthly_volume: Decimal | None
-    total_volume: Decimal | None
-    liquidity: Decimal | None
-
-
-def _to_decimal(value: object) -> Decimal | None:
-    if value is None or value == "":
-        return None
-    try:
-        return Decimal(str(value))
-    except (InvalidOperation, ValueError, TypeError):
-        return None
-
-
-def _parse_market(raw: dict[str, object]) -> GammaMarketStats | None:
-    cid = raw.get("conditionId")
-    if not cid or not isinstance(cid, str):
-        return None
-    return GammaMarketStats(
-        condition_id=cid,
-        daily_volume=_to_decimal(raw.get("volume24hr")),
-        weekly_volume=_to_decimal(raw.get("volume1wk")),
-        monthly_volume=_to_decimal(raw.get("volume1mo")),
-        total_volume=_to_decimal(raw.get("volumeNum") or raw.get("volume")),
-        liquidity=_to_decimal(raw.get("liquidityNum") or raw.get("liquidity")),
-    )
-
-
-class GammaClientError(Exception):
-    """Raised when gamma-api returns an unrecoverable error."""
-
-
-class GammaClient:
-    """Async client for the public gamma-api markets endpoint.
-
-    Provides batched, paginated reads of every active market with their
-    24h/weekly/monthly volume and current liquidity. Designed to be called
-    from MarketMetadataSync once per sync interval; results are merged into
-    Redis-cached MarketMetadata objects.
-    """
-
-    def __init__(
-        self,
-        *,
-        host: str = DEFAULT_HOST,
-        timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
-        page_limit: int = DEFAULT_PAGE_LIMIT,
-        max_pages: int = DEFAULT_MAX_PAGES,
-        page_concurrency: int = DEFAULT_PAGE_CONCURRENCY,
-        max_retries: int = DEFAULT_MAX_RETRIES,
-        retry_base_delay_seconds: float = DEFAULT_RETRY_BASE_DELAY_SECONDS,
-    ) -> None:
-        self._host = host.rstrip("/")
-        self._timeout = timeout_seconds
-        self._page_limit = page_limit
-        self._max_pages = max_pages
-        self._page_concurrency = page_concurrency
-        self._max_retries = max_retries
-        self._retry_base = retry_base_delay_seconds
-
-    async def _get_with_retry(
-        self,
-        client: httpx.AsyncClient,
-        path: str,
-        params: dict[str, object],
-    ) -> list[dict[str, object]]:
-        last_exc: Exception | None = None
-        delay = self._retry_base
-        for attempt in range(self._max_retries):
-            try:
-                resp = await client.get(path, params=params)
-                resp.raise_for_status()
-                payload = resp.json()
-                if not isinstance(payload, list):
-                    raise GammaClientError(
-                        f"Unexpected gamma response shape for {path}: {type(payload).__name__}"
-                    )
-                return payload
-            except (httpx.HTTPError, ValueError) as exc:
-                last_exc = exc
-                logger.warning(
-                    "gamma %s attempt %d/%d failed: %s",
-                    path,
-                    attempt + 1,
-                    self._max_retries,
-                    exc,
-                )
-                if attempt < self._max_retries - 1:
-                    await asyncio.sleep(delay)
-                    delay *= 2
-        raise GammaClientError(
-            f"gamma {path} failed after {self._max_retries} attempts: {last_exc}"
-        )
-
-    async def get_active_market_stats(self) -> dict[str, GammaMarketStats]:
-        """Fetch volume/liquidity for the most-traded active markets.
-
-        Walks up to `max_pages` pages of `page_limit` markets each, sorted
-        by 24h volume descending, with bounded concurrency. Markets beyond
-        that window have effectively zero recent volume — the size_anomaly
-        niche path handles them without needing a ratio.
-
-        Returns:
-            Mapping condition_id -> GammaMarketStats.
-        """
-        results: dict[str, GammaMarketStats] = {}
-        sem = asyncio.Semaphore(self._page_concurrency)
-        stop = asyncio.Event()
-
-        async with httpx.AsyncClient(
-            base_url=self._host,
-            timeout=self._timeout,
-            headers={"User-Agent": "polymarket-insider-tracker/0.1"},
-        ) as client:
-
-            async def fetch_page(page_index: int) -> list[dict[str, object]]:
-                if stop.is_set():
-                    return []
-                params = {
-                    "limit": self._page_limit,
-                    "offset": page_index * self._page_limit,
-                    "active": "true",
-                    "closed": "false",
-                    "order": "volume24hr",
-                    "ascending": "false",
-                }
-                async with sem:
-                    if stop.is_set():
-                        return []
-                    try:
-                        return await self._get_with_retry(client, "/markets", params)
-                    except GammaClientError as exc:
-                        # Gamma rejects offsets past its hard cap with a
-                        # validation error; treat that as a clean stop.
-                        logger.debug("gamma stop at page %d: %s", page_index, exc)
-                        stop.set()
-                        return []
-
-            tasks = [asyncio.create_task(fetch_page(i)) for i in range(self._max_pages)]
-            pages = await asyncio.gather(*tasks)
-
-        empty_streak = 0
-        for page in pages:
-            if not page:
-                empty_streak += 1
-                continue
-            empty_streak = 0
-            for raw in page:
-                if not isinstance(raw, dict):
-                    continue
-                parsed = _parse_market(raw)
-                if parsed is not None:
-                    results[parsed.condition_id] = parsed
-            if len(page) < self._page_limit:
-                # short page — we walked past the end of the active set
-                empty_streak += 1
-            if empty_streak >= 2:
-                break
-
-        logger.info("gamma sync: fetched stats for %d active markets", len(results))
-        return results

src/polymarket_insider_tracker/ingestor/metadata_sync.py

@@ -9,14 +9,13 @@ import contextlib
 import json
 import logging
 from collections.abc import Callable
-from dataclasses import dataclass, replace
+from dataclasses import dataclass
 from datetime import UTC, datetime
 from enum import StrEnum
 
 from redis.asyncio import Redis
 
 from .clob_client import ClobClient
-from .gamma_client import GammaClient, GammaClientError, GammaMarketStats
 from .models import MarketMetadata
 
 logger = logging.getLogger(__name__)
@@ -92,7 +91,6 @@ class MarketMetadataSync:
         redis: Redis,
         clob_client: ClobClient,
         *,
-        gamma_client: GammaClient | None = None,
         sync_interval_seconds: int = DEFAULT_SYNC_INTERVAL_SECONDS,
         cache_ttl_seconds: int = DEFAULT_CACHE_TTL_SECONDS,
         key_prefix: str = DEFAULT_REDIS_KEY_PREFIX,
@@ -104,8 +102,6 @@ class MarketMetadataSync:
         Args:
             redis: Redis async client for caching.
             clob_client: CLOB API client for fetching markets.
-            gamma_client: Optional gamma-api client for volume/liquidity
-                enrichment. Defaults to a fresh GammaClient() instance.
             sync_interval_seconds: Interval between syncs (default: 300 / 5 min).
             cache_ttl_seconds: TTL for cached entries (default: 600 / 10 min).
             key_prefix: Redis key prefix for market data.
@@ -114,7 +110,6 @@ class MarketMetadataSync:
         """
         self._redis = redis
         self._clob = clob_client
-        self._gamma = gamma_client or GammaClient()
         self._sync_interval = sync_interval_seconds
         self._cache_ttl = cache_ttl_seconds
         self._key_prefix = key_prefix
@@ -221,18 +216,6 @@ class MarketMetadataSync:
                 self._set_state(SyncState.ERROR)
                 # Continue running - will retry on next interval
 
-    async def _fetch_gamma_stats(self) -> dict[str, GammaMarketStats]:
-        """Fetch volume/liquidity stats from gamma-api.
-
-        Returns an empty dict on failure so a degraded gamma endpoint
-        does not stop CLOB metadata from being cached.
-        """
-        try:
-            return await self._gamma.get_active_market_stats()
-        except (GammaClientError, Exception) as e:
-            logger.warning("gamma stats fetch failed (continuing without volume): %s", e)
-            return {}
-
     async def _sync_all_markets(self) -> None:
         """Fetch all markets and cache them in Redis."""
         self._set_state(SyncState.SYNCING)
@@ -240,27 +223,14 @@ class MarketMetadataSync:
         self._stats.total_syncs += 1
 
         try:
-            # Fetch CLOB markets and gamma volume snapshot in parallel
-            markets, gamma_stats = await asyncio.gather(
-                asyncio.to_thread(self._clob.get_markets, True),
-                self._fetch_gamma_stats(),
-            )
+            # Fetch markets from CLOB API (runs in thread pool for sync API)
+            markets = await asyncio.to_thread(self._clob.get_markets, True)
 
-            # Cache each market in Redis, enriched with gamma volume/liquidity
+            # Cache each market in Redis
             cached_count = 0
-            enriched_count = 0
             for market in markets:
                 try:
                     metadata = MarketMetadata.from_market(market)
-                    stats = gamma_stats.get(metadata.condition_id)
-                    if stats is not None:
-                        metadata = replace(
-                            metadata,
-                            daily_volume=stats.daily_volume,
-                            weekly_volume=stats.weekly_volume,
-                            liquidity=stats.liquidity,
-                        )
-                        enriched_count += 1
                     await self._cache_market(metadata)
                     cached_count += 1
                 except Exception as e:
@@ -276,10 +246,7 @@ class MarketMetadataSync:
 
             self._set_state(SyncState.IDLE)
             logger.info(
-                "Synced %d markets (%d enriched with gamma volume) in %.2fs",
-                cached_count,
-                enriched_count,
-                self._stats.last_sync_duration_seconds,
+                f"Synced {cached_count} markets in {self._stats.last_sync_duration_seconds:.2f}s"
             )
 
             # Notify callback

src/polymarket_insider_tracker/ingestor/models.py

@@ -399,12 +399,6 @@ class MarketMetadata:
     # Derived metadata
     category: str = "other"
 
-    # Liquidity/volume snapshot (from gamma-api). All optional — older
-    # cache entries and CLOB-only sync results may not have these.
-    daily_volume: Decimal | None = None
-    weekly_volume: Decimal | None = None
-    liquidity: Decimal | None = None
-
     # Cache metadata
     last_updated: datetime = field(default_factory=lambda: datetime.now(UTC))
 
@@ -452,9 +446,6 @@ class MarketMetadata:
             "active": self.active,
             "closed": self.closed,
             "category": self.category,
-            "daily_volume": str(self.daily_volume) if self.daily_volume is not None else None,
-            "weekly_volume": str(self.weekly_volume) if self.weekly_volume is not None else None,
-            "liquidity": str(self.liquidity) if self.liquidity is not None else None,
             "last_updated": self.last_updated.isoformat(),
         }
 
@@ -486,15 +477,6 @@ class MarketMetadata:
         else:
             last_updated = datetime.now(UTC)
 
-        def _opt_dec(key: str) -> Decimal | None:
-            raw = data.get(key)
-            if raw is None or raw == "":
-                return None
-            try:
-                return Decimal(str(raw))
-            except (ValueError, ArithmeticError):
-                return None
-
         return cls(
             condition_id=str(data["condition_id"]),
             question=str(data.get("question", "")),
@@ -504,8 +486,5 @@ class MarketMetadata:
             active=bool(data.get("active", True)),
             closed=bool(data.get("closed", False)),
             category=str(data.get("category", "other")),
-            daily_volume=_opt_dec("daily_volume"),
-            weekly_volume=_opt_dec("weekly_volume"),
-            liquidity=_opt_dec("liquidity"),
             last_updated=last_updated,
         )

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.
 

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"),
+    )

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,
+        )

tests/ingestor/test_gamma_client.py

@@ -1,249 +0,0 @@
-"""Tests for the gamma-api client."""
-
-from __future__ import annotations
-
-from decimal import Decimal
-
-import httpx
-import pytest
-
-from polymarket_insider_tracker.ingestor import gamma_client as gamma_module
-from polymarket_insider_tracker.ingestor.gamma_client import (
-    GammaClient,
-    GammaClientError,
-    GammaMarketStats,
-    _parse_market,
-)
-
-
-class TestParseMarket:
-    def test_parses_full_payload(self) -> None:
-        raw = {
-            "conditionId": "0xabc",
-            "volume24hr": "12345.67",
-            "volume1wk": "100000",
-            "volume1mo": "500000",
-            "volumeNum": "999999.5",
-            "liquidityNum": "42000",
-        }
-        stats = _parse_market(raw)
-        assert stats is not None
-        assert stats.condition_id == "0xabc"
-        assert stats.daily_volume == Decimal("12345.67")
-        assert stats.weekly_volume == Decimal("100000")
-        assert stats.monthly_volume == Decimal("500000")
-        assert stats.total_volume == Decimal("999999.5")
-        assert stats.liquidity == Decimal("42000")
-
-    def test_falls_back_to_alternative_keys(self) -> None:
-        raw = {
-            "conditionId": "0x1",
-            "volume24hr": "1",
-            "volume": "777",
-            "liquidity": "55",
-        }
-        stats = _parse_market(raw)
-        assert stats is not None
-        assert stats.total_volume == Decimal("777")
-        assert stats.liquidity == Decimal("55")
-
-    def test_handles_missing_numeric_fields(self) -> None:
-        stats = _parse_market({"conditionId": "0x2"})
-        assert stats is not None
-        assert stats.daily_volume is None
-        assert stats.liquidity is None
-
-    def test_drops_garbage_decimals(self) -> None:
-        stats = _parse_market(
-            {"conditionId": "0x3", "volume24hr": "not-a-number", "liquidityNum": ""}
-        )
-        assert stats is not None
-        assert stats.daily_volume is None
-        assert stats.liquidity is None
-
-    def test_rejects_missing_condition_id(self) -> None:
-        assert _parse_market({"volume24hr": "1"}) is None
-        assert _parse_market({"conditionId": ""}) is None
-        assert _parse_market({"conditionId": 123}) is None  # type: ignore[arg-type]
-
-
-def _make_client(
-    _handler: httpx.MockTransport,
-    *,
-    page_limit: int = 100,
-    max_pages: int = 5,
-    page_concurrency: int = 5,
-    max_retries: int = 1,
-) -> GammaClient:
-    """Build a GammaClient that constructs httpx.AsyncClient with the given transport.
-
-    GammaClient creates its own AsyncClient inside `get_active_market_stats`,
-    so we monkeypatch the AsyncClient factory in the module to inject the mock
-    transport.
-    """
-    return GammaClient(
-        page_limit=page_limit,
-        max_pages=max_pages,
-        page_concurrency=page_concurrency,
-        max_retries=max_retries,
-        retry_base_delay_seconds=0.0,
-    )
-
-
-@pytest.fixture
-def patch_async_client(monkeypatch: pytest.MonkeyPatch):
-    """Replace the AsyncClient used by gamma_client with one bound to a MockTransport."""
-
-    def _apply(handler: httpx.MockTransport) -> None:
-        original = gamma_module.httpx.AsyncClient
-
-        def factory(*args: object, **kwargs: object) -> httpx.AsyncClient:
-            kwargs["transport"] = handler  # type: ignore[index]
-            return original(*args, **kwargs)  # type: ignore[arg-type]
-
-        monkeypatch.setattr(gamma_module.httpx, "AsyncClient", factory)
-
-    return _apply
-
-
-@pytest.mark.asyncio
-async def test_get_active_market_stats_single_page(patch_async_client) -> None:
-    page_one = [
-        {"conditionId": "0xa", "volume24hr": "100", "liquidityNum": "10"},
-        {"conditionId": "0xb", "volume24hr": "200", "liquidityNum": "20"},
-    ]
-    calls: list[dict[str, str]] = []
-
-    def handler(request: httpx.Request) -> httpx.Response:
-        calls.append(dict(request.url.params))
-        offset = int(request.url.params.get("offset", "0"))
-        if offset == 0:
-            return httpx.Response(200, json=page_one)
-        return httpx.Response(200, json=[])
-
-    transport = httpx.MockTransport(handler)
-    patch_async_client(transport)
-    client = _make_client(transport, page_limit=2, max_pages=3)
-
-    result = await client.get_active_market_stats()
-
-    assert set(result.keys()) == {"0xa", "0xb"}
-    assert isinstance(result["0xa"], GammaMarketStats)
-    assert result["0xa"].daily_volume == Decimal("100")
-    assert calls[0]["limit"] == "2"
-    assert calls[0]["order"] == "volume24hr"
-    assert calls[0]["ascending"] == "false"
-
-
-@pytest.mark.asyncio
-async def test_get_active_market_stats_short_page_stops(patch_async_client) -> None:
-    """A page shorter than page_limit signals end-of-data after a small empty streak."""
-    page_zero = [{"conditionId": f"0x{i}", "volume24hr": str(i)} for i in range(5)]
-    page_one_short = [{"conditionId": "0xshort", "volume24hr": "1"}]
-
-    def handler(request: httpx.Request) -> httpx.Response:
-        offset = int(request.url.params.get("offset", "0"))
-        if offset == 0:
-            return httpx.Response(200, json=page_zero)
-        if offset == 5:
-            return httpx.Response(200, json=page_one_short)
-        return httpx.Response(200, json=[])
-
-    transport = httpx.MockTransport(handler)
-    patch_async_client(transport)
-    client = _make_client(transport, page_limit=5, max_pages=10)
-
-    result = await client.get_active_market_stats()
-
-    assert "0xshort" in result
-    assert len(result) == 6
-
-
-@pytest.mark.asyncio
-async def test_get_active_market_stats_offset_cap_clean_stop(
-    patch_async_client,
-) -> None:
-    """Gamma rejects offsets past its hard cap; that error is swallowed cleanly."""
-
-    def handler(request: httpx.Request) -> httpx.Response:
-        offset = int(request.url.params.get("offset", "0"))
-        if offset == 0:
-            return httpx.Response(200, json=[{"conditionId": "0xa", "volume24hr": "1"}])
-        return httpx.Response(400, json={"error": "offset too large"})
-
-    transport = httpx.MockTransport(handler)
-    patch_async_client(transport)
-    client = _make_client(transport, page_limit=1, max_pages=4, max_retries=1)
-
-    result = await client.get_active_market_stats()
-    assert "0xa" in result
-
-
-@pytest.mark.asyncio
-async def test_get_with_retry_recovers_after_transient_error(
-    patch_async_client,
-) -> None:
-    """Transient HTTP errors retry up to max_retries before giving up."""
-    state = {"attempts": 0}
-
-    def handler(request: httpx.Request) -> httpx.Response:
-        offset = int(request.url.params.get("offset", "0"))
-        if offset == 0:
-            state["attempts"] += 1
-            if state["attempts"] < 2:
-                return httpx.Response(503, json={"error": "transient"})
-            return httpx.Response(200, json=[{"conditionId": "0xrecover", "volume24hr": "1"}])
-        return httpx.Response(200, json=[])
-
-    transport = httpx.MockTransport(handler)
-    patch_async_client(transport)
-    client = _make_client(transport, page_limit=1, max_pages=2, max_retries=3)
-
-    result = await client.get_active_market_stats()
-    assert "0xrecover" in result
-    assert state["attempts"] == 2
-
-
-@pytest.mark.asyncio
-async def test_unexpected_response_shape_is_handled(patch_async_client) -> None:
-    """A non-list payload becomes a clean stop, not a crash."""
-
-    def handler(_request: httpx.Request) -> httpx.Response:
-        return httpx.Response(200, json={"unexpected": "shape"})
-
-    transport = httpx.MockTransport(handler)
-    patch_async_client(transport)
-    client = _make_client(transport, page_limit=1, max_pages=2, max_retries=1)
-
-    result = await client.get_active_market_stats()
-    assert result == {}
-
-
-@pytest.mark.asyncio
-async def test_skips_non_dict_entries(patch_async_client) -> None:
-    """Defensive: server returning mixed-type list items shouldn't crash."""
-
-    def handler(request: httpx.Request) -> httpx.Response:
-        offset = int(request.url.params.get("offset", "0"))
-        if offset == 0:
-            return httpx.Response(
-                200,
-                json=[
-                    {"conditionId": "0xa", "volume24hr": "1"},
-                    "garbage",
-                    None,
-                    42,
-                ],
-            )
-        return httpx.Response(200, json=[])
-
-    transport = httpx.MockTransport(handler)
-    patch_async_client(transport)
-    client = _make_client(transport, page_limit=4, max_pages=2)
-
-    result = await client.get_active_market_stats()
-    assert list(result.keys()) == ["0xa"]
-
-
-def test_gamma_client_error_inherits_exception() -> None:
-    assert issubclass(GammaClientError, Exception)

tests/ingestor/test_metadata_sync.py

@@ -8,7 +8,6 @@ from unittest.mock import AsyncMock, MagicMock
 import pytest
 
 from polymarket_insider_tracker.ingestor.clob_client import ClobClient
-from polymarket_insider_tracker.ingestor.gamma_client import GammaClient
 from polymarket_insider_tracker.ingestor.metadata_sync import (
     DEFAULT_CACHE_TTL_SECONDS,
     DEFAULT_REDIS_KEY_PREFIX,
@@ -73,18 +72,6 @@ def mock_clob(sample_market: Market) -> MagicMock:
     return clob
 
 
-@pytest.fixture
-def mock_gamma() -> MagicMock:
-    """Create a mock GammaClient that returns empty volume stats.
-
-    Without this, MarketMetadataSync would instantiate a default
-    GammaClient and hit the real gamma-api over HTTP during unit tests.
-    """
-    gamma = MagicMock(spec=GammaClient)
-    gamma.get_active_market_stats = AsyncMock(return_value={})
-    return gamma
-
-
 class TestDeriveCategory:
     """Tests for the derive_category function."""
 
@@ -208,9 +195,9 @@ class TestSyncStats:
 class TestMarketMetadataSync:
     """Tests for the MarketMetadataSync class."""
 
-    def test_init(self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock) -> None:
+    def test_init(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
         """Test initialization."""
-        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
 
         assert sync.state == SyncState.STOPPED
         assert sync.stats.total_syncs == 0
@@ -218,14 +205,11 @@ class TestMarketMetadataSync:
         assert sync._cache_ttl == DEFAULT_CACHE_TTL_SECONDS
         assert sync._key_prefix == DEFAULT_REDIS_KEY_PREFIX
 
-    def test_init_custom_config(
-        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
-    ) -> None:
+    def test_init_custom_config(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
         """Test initialization with custom config."""
         sync = MarketMetadataSync(
             redis=mock_redis,
             clob_client=mock_clob,
-            gamma_client=mock_gamma,
             sync_interval_seconds=60,
             cache_ttl_seconds=120,
             key_prefix="custom:",
@@ -236,11 +220,9 @@ class TestMarketMetadataSync:
         assert sync._key_prefix == "custom:"
 
     @pytest.mark.asyncio
-    async def test_start_stop(
-        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
-    ) -> None:
+    async def test_start_stop(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
         """Test starting and stopping the sync service."""
-        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
 
         # Start
         await sync.start()
@@ -254,10 +236,10 @@ class TestMarketMetadataSync:
 
     @pytest.mark.asyncio
     async def test_start_performs_initial_sync(
-        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
+        self, mock_redis: AsyncMock, mock_clob: MagicMock
     ) -> None:
         """Test that start performs an initial sync."""
-        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
 
         await sync.start()
 
@@ -270,12 +252,10 @@ class TestMarketMetadataSync:
         await sync.stop()
 
     @pytest.mark.asyncio
-    async def test_start_failure(
-        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
-    ) -> None:
+    async def test_start_failure(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
         """Test start failure handling."""
         mock_clob.get_markets.side_effect = Exception("API error")
-        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
 
         with pytest.raises(MetadataSyncError, match="initial sync failed"):
             await sync.start()
@@ -288,7 +268,6 @@ class TestMarketMetadataSync:
         self,
         mock_redis: AsyncMock,
         mock_clob: MagicMock,
-        mock_gamma: MagicMock,
         sample_metadata: MarketMetadata,
     ) -> None:
         """Test get_market with cache hit."""
@@ -296,7 +275,7 @@ class TestMarketMetadataSync:
         cached_data = json.dumps(sample_metadata.to_dict())
         mock_redis.get = AsyncMock(return_value=cached_data)
 
-        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
         await sync.start()
 
         result = await sync.get_market("cond123")
@@ -309,14 +288,12 @@ class TestMarketMetadataSync:
         await sync.stop()
 
     @pytest.mark.asyncio
-    async def test_get_market_cache_miss(
-        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
-    ) -> None:
+    async def test_get_market_cache_miss(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
         """Test get_market with cache miss."""
         # Setup cache miss
         mock_redis.get = AsyncMock(return_value=None)
 
-        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
         await sync.start()
 
         result = await sync.get_market("cond123")
@@ -331,14 +308,12 @@ class TestMarketMetadataSync:
         await sync.stop()
 
     @pytest.mark.asyncio
-    async def test_get_market_not_found(
-        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
-    ) -> None:
+    async def test_get_market_not_found(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
         """Test get_market when market doesn't exist."""
         mock_redis.get = AsyncMock(return_value=None)
         mock_clob.get_market.return_value = None
 
-        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
         await sync.start()
 
         result = await sync.get_market("nonexistent")
@@ -348,11 +323,9 @@ class TestMarketMetadataSync:
         await sync.stop()
 
     @pytest.mark.asyncio
-    async def test_invalidate_market(
-        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
-    ) -> None:
+    async def test_invalidate_market(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
         """Test cache invalidation."""
-        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
         await sync.start()
 
         result = await sync.invalidate_market("cond123")
@@ -363,11 +336,9 @@ class TestMarketMetadataSync:
         await sync.stop()
 
     @pytest.mark.asyncio
-    async def test_force_sync(
-        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
-    ) -> None:
+    async def test_force_sync(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
         """Test forced sync."""
-        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
         await sync.start()
 
         # Initial sync
@@ -382,9 +353,7 @@ class TestMarketMetadataSync:
         await sync.stop()
 
     @pytest.mark.asyncio
-    async def test_state_change_callback(
-        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
-    ) -> None:
+    async def test_state_change_callback(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
         """Test state change callback."""
         states: list[SyncState] = []
 
@@ -394,7 +363,6 @@ class TestMarketMetadataSync:
         sync = MarketMetadataSync(
             redis=mock_redis,
             clob_client=mock_clob,
-            gamma_client=mock_gamma,
             on_state_change=on_state_change,
         )
 
@@ -409,7 +377,7 @@ class TestMarketMetadataSync:
 
     @pytest.mark.asyncio
     async def test_sync_complete_callback(
-        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
+        self, mock_redis: AsyncMock, mock_clob: MagicMock
     ) -> None:
         """Test sync complete callback."""
         sync_stats: list[SyncStats] = []
@@ -420,7 +388,6 @@ class TestMarketMetadataSync:
         sync = MarketMetadataSync(
             redis=mock_redis,
             clob_client=mock_clob,
-            gamma_client=mock_gamma,
             on_sync_complete=on_sync_complete,
         )
 
@@ -434,11 +401,7 @@ class TestMarketMetadataSync:
 
     @pytest.mark.asyncio
     async def test_get_markets_by_category(
-        self,
-        mock_redis: AsyncMock,
-        mock_clob: MagicMock,
-        mock_gamma: MagicMock,
-        sample_metadata: MarketMetadata,
+        self, mock_redis: AsyncMock, mock_clob: MagicMock, sample_metadata: MarketMetadata
     ) -> None:
         """Test getting markets by category."""
         # Setup scan to return keys
@@ -449,7 +412,7 @@ class TestMarketMetadataSync:
         cached_data = json.dumps(sample_metadata.to_dict())
         mock_redis.get = AsyncMock(return_value=cached_data)
 
-        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
         # Don't start to avoid initial sync complexity
         sync._state = SyncState.IDLE
 
@@ -459,11 +422,9 @@ class TestMarketMetadataSync:
         assert results[0].category == "crypto"
 
     @pytest.mark.asyncio
-    async def test_cannot_start_twice(
-        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
-    ) -> None:
+    async def test_cannot_start_twice(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
         """Test that starting twice doesn't double-start."""
-        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
 
         await sync.start()
         await sync.start()  # Should be a no-op
@@ -473,11 +434,9 @@ class TestMarketMetadataSync:
         await sync.stop()
 
     @pytest.mark.asyncio
-    async def test_stop_when_stopped(
-        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
-    ) -> None:
+    async def test_stop_when_stopped(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
         """Test stopping when already stopped."""
-        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
 
         await sync.stop()  # Should be a no-op
 

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

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