chore: add MIT LICENSE (#711 portfolio bar — 164-star repo had none) (#110)

The size anomaly detector currently emits a base 0.2 confidence signal
whenever a market is "niche" (low volume OR niche-prone category with
unknown volume) — even for $5 trades. In production this floods the
alert pipeline with nothing-burgers, because the niche-prone category
set covers `science / tech / finance / other`, which matches a huge
chunk of Polymarket's long tail.

This adds a `DEFAULT_NICHE_MIN_TRADE_SIZE = $500` floor that applies
ONLY to the niche-only path: if a trade exceeds the volume or book
thresholds, the guard does not block it. So real anomalies still come
through, but tiny niche trades get filtered out before reaching the
risk scorer.

The threshold is configurable via `niche_min_trade_size` in
`SizeAnomalyDetector.__init__`.

Tests added:
- niche-only below floor → suppressed
- niche-only at/above floor → still emits 0.2 base
- small trade with high volume_impact → guard does not block

Co-authored-by: schrodinger01 <schrodinger01@users.noreply.github.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.

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Patrick Selamy
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

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/alerter/formatter.py

@@ -290,12 +290,14 @@ class AlertFormatter:
                     wallet_line += f" \\(Age: {age_hours:.0f}h\\)"
         lines.append(wallet_line)
 
-        # Risk score
+        # Risk score — every numeric literal here must be escaped because
-        lines.append(f"*Risk Score:* {assessment.weighted_score:.2f} \\({risk_level}\\)")
+        # MarkdownV2 treats `.` as a special character and rejects unescaped
+        # ones with `Bad Request: can't parse entities`.
+        score_str = self._escape_telegram_markdown(f"{assessment.weighted_score:.2f}")
+        lines.append(f"*Risk Score:* {score_str} \\({risk_level}\\)")
 
         # Market
         market_title = trade.event_title or trade.market_slug or "Unknown Market"
-        # Escape special Telegram markdown characters
         market_title_escaped = self._escape_telegram_markdown(market_title)
         if "market" in links:
             lines.append(f"*Market:* [{market_title_escaped}]({links['market']})")
@@ -303,14 +305,16 @@ class AlertFormatter:
             lines.append(f"*Market:* {market_title_escaped}")
 
         # Trade details
-        usdc_value = format_usdc(trade.notional_value).replace("$", "\\$")
+        usdc_value = self._escape_telegram_markdown(format_usdc(trade.notional_value))
-        lines.append(
+        price_str = self._escape_telegram_markdown(f"{trade.price:.3f}")
-            f"*Trade:* {trade.side} {trade.outcome} @ \\${trade.price:.3f} \\| {usdc_value}"
+        side_escaped = self._escape_telegram_markdown(trade.side)
-        )
+        outcome_escaped = self._escape_telegram_markdown(trade.outcome)
+        lines.append(f"*Trade:* {side_escaped} {outcome_escaped} @ \\${price_str} \\| {usdc_value}")
 
         # Signals
         if signals:
-            lines.append(f"*Signals:* {', '.join(signals)}")
+            signals_escaped = [self._escape_telegram_markdown(s) for s in signals]
+            lines.append(f"*Signals:* {', '.join(signals_escaped)}")
 
         # Links
         lines.append("")

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 configuration. The threshold lifted from 0.6 to 0.80 after the
-DEFAULT_ALERT_THRESHOLD = 0.6
+# 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/detector/size_anomaly.py

@@ -17,6 +17,9 @@ logger = logging.getLogger(__name__)
 DEFAULT_VOLUME_THRESHOLD = 0.02  # 2% of daily volume
 DEFAULT_BOOK_THRESHOLD = 0.05  # 5% of order book depth
 DEFAULT_NICHE_VOLUME_THRESHOLD = Decimal("50000")  # $50k daily volume
+# Niche-only path requires real trade size; below this, suppress the base
+# 0.2 confidence so we don't flood alerts with low-value niche trades.
+DEFAULT_NICHE_MIN_TRADE_SIZE = Decimal("500")  # USDC notional
 
 # Niche market categories - markets in these categories with low specificity
 # are more likely to have insider information value
@@ -59,6 +62,7 @@ class SizeAnomalyDetector:
         volume_threshold: float = DEFAULT_VOLUME_THRESHOLD,
         book_threshold: float = DEFAULT_BOOK_THRESHOLD,
         niche_volume_threshold: Decimal = DEFAULT_NICHE_VOLUME_THRESHOLD,
+        niche_min_trade_size: Decimal = DEFAULT_NICHE_MIN_TRADE_SIZE,
     ) -> None:
         """Initialize the size anomaly detector.
 
@@ -67,11 +71,15 @@ class SizeAnomalyDetector:
             volume_threshold: Threshold for volume impact (default 0.02 = 2%).
             book_threshold: Threshold for book impact (default 0.05 = 5%).
             niche_volume_threshold: Volume below which market is niche ($50k).
+            niche_min_trade_size: Minimum trade notional ($) to allow a
+                niche-only signal. Below this, niche-only path is suppressed
+                to avoid flooding alerts with low-value trades.
         """
         self._metadata_sync = metadata_sync
         self._volume_threshold = volume_threshold
         self._book_threshold = book_threshold
         self._niche_volume_threshold = niche_volume_threshold
+        self._niche_min_trade_size = niche_min_trade_size
 
     async def analyze(
         self,
@@ -129,6 +137,18 @@ class SizeAnomalyDetector:
         exceeds_volume = volume_impact > self._volume_threshold
         exceeds_book = book_impact > self._book_threshold
 
+        # Niche-only signals require a minimum trade size; otherwise we'd
+        # flood alerts with every tiny trade in any niche-prone category.
+        niche_only = is_niche and not exceeds_volume and not exceeds_book
+        if niche_only and trade_size < self._niche_min_trade_size:
+            logger.debug(
+                "Trade %s niche-only but size %s < min %s, skipping",
+                trade.trade_id,
+                trade_size,
+                self._niche_min_trade_size,
+            )
+            return None
+
         if not exceeds_volume and not exceeds_book and not is_niche:
             logger.debug(
                 "Trade %s does not exceed thresholds: volume=%.4f, book=%.4f",

src/polymarket_insider_tracker/ingestor/gamma_client.py

@@ -0,0 +1,205 @@
+"""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,13 +9,14 @@ import contextlib
 import json
 import logging
 from collections.abc import Callable
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 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__)
@@ -91,6 +92,7 @@ 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,
@@ -102,6 +104,8 @@ 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.
@@ -110,6 +114,7 @@ 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
@@ -216,6 +221,18 @@ 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)
@@ -223,14 +240,27 @@ class MarketMetadataSync:
         self._stats.total_syncs += 1
 
         try:
-            # Fetch markets from CLOB API (runs in thread pool for sync API)
+            # Fetch CLOB markets and gamma volume snapshot in parallel
-            markets = await asyncio.to_thread(self._clob.get_markets, True)
+            markets, gamma_stats = await asyncio.gather(
+                asyncio.to_thread(self._clob.get_markets, True),
+                self._fetch_gamma_stats(),
+            )
 
-            # Cache each market in Redis
+            # Cache each market in Redis, enriched with gamma volume/liquidity
             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:
@@ -246,7 +276,10 @@ class MarketMetadataSync:
 
             self._set_state(SyncState.IDLE)
             logger.info(
-                f"Synced {cached_count} markets in {self._stats.last_sync_duration_seconds:.2f}s"
+                "Synced %d markets (%d enriched with gamma volume) in %.2fs",
+                cached_count,
+                enriched_count,
+                self._stats.last_sync_duration_seconds,
             )
 
             # Notify callback

src/polymarket_insider_tracker/ingestor/models.py

@@ -399,6 +399,12 @@ 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))
 
@@ -446,6 +452,9 @@ 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(),
         }
 
@@ -477,6 +486,15 @@ 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", "")),
@@ -486,5 +504,8 @@ 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/profiler/funding.py

@@ -26,8 +26,46 @@ logger = logging.getLogger(__name__)
 USDC_BRIDGED = "0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174"
 USDC_NATIVE = "0x3c499c542cEF5E3811e1192ce70d8cC03d5c3359"
 
-# ERC20 Transfer event signature
+# ERC20 Transfer event signature. ``HexBytes.hex()`` returns a *bare* hex
+# string without the ``0x`` prefix; publicnode tolerates that, but stricter
+# providers (e.g. drpc — which we use as the fallback) reject it with
+# ``invalid argument 0: hex string without 0x prefix``. Always pass the
+# 0x-prefixed form to ``eth_getLogs``.
 TRANSFER_EVENT_SIGNATURE = AsyncWeb3.keccak(text="Transfer(address,address,uint256)")
+TRANSFER_EVENT_TOPIC = "0x" + TRANSFER_EVENT_SIGNATURE.hex().removeprefix("0x")
+
+# eth_getLogs block-range chunking. Most public Polygon RPCs (publicnode, ankr,
+# llamarpc) cap the range at 10_000 blocks per call; pick a window slightly
+# under the cap so off-by-one differences between providers don't trip us up.
+DEFAULT_CHUNK_SIZE_BLOCKS = 9_000
+# Polygon block time is ~2.0s. publicnode (the most common free RPC) prunes
+# log history aggressively — empirically only ~100k blocks (~55 hours) are
+# served before requests start returning "History has been pruned". We default
+# to 80k blocks (~44 hours), which is more than enough for fresh-wallet
+# funding traces (those wallets are by definition new) and fits comfortably
+# inside what most public providers retain.
+DEFAULT_MAX_LOOKBACK_BLOCKS = 80_000
+
+# Substrings that, when present in an RPC error, indicate the chunk we just
+# asked for is outside the provider's archive horizon. Walking further back
+# is futile, so we stop the trace early instead of hammering every chunk.
+_PRUNED_HISTORY_MARKERS: tuple[str, ...] = (
+    "history has been pruned",
+    "missing trie node",
+    "older than",
+)
+
+
+def _is_pruned_history_error(err: BaseException) -> bool:
+    """Return True if the RPC error indicates pruned history.
+
+    Public Polygon nodes only retain a recent slice of log history. When we
+    walk back through that slice in chunks and hit the cutoff, every further
+    chunk will fail with the same message — so we stop early instead of
+    burning quota on guaranteed failures.
+    """
+    text = str(err).lower()
+    return any(marker in text for marker in _PRUNED_HISTORY_MARKERS)
 
 
 class FundingTracer:
@@ -50,6 +88,8 @@ class FundingTracer:
         *,
         max_hops: int = 3,
         usdc_addresses: list[str] | None = None,
+        chunk_size_blocks: int = DEFAULT_CHUNK_SIZE_BLOCKS,
+        max_lookback_blocks: int = DEFAULT_MAX_LOOKBACK_BLOCKS,
     ) -> None:
         """Initialize the funding tracer.
 
@@ -58,6 +98,11 @@ class FundingTracer:
             entity_registry: Registry for entity classification. Creates default if None.
             max_hops: Maximum hops to trace back (default 3).
             usdc_addresses: USDC contract addresses to track. Uses defaults if None.
+            chunk_size_blocks: Block window size per eth_getLogs call. Public
+                Polygon RPCs cap at 10_000 blocks; default leaves a safety margin.
+            max_lookback_blocks: How far back to scan when caller passes
+                ``from_block=0``. Default ~44 hours at 2s block time, which
+                fits inside the pruned-history horizon of most public RPCs.
         """
         self.polygon_client = polygon_client
         self.entity_registry = entity_registry or EntityRegistry()
@@ -65,6 +110,8 @@ class FundingTracer:
         self._usdc_addresses = [
             addr.lower() for addr in (usdc_addresses or [USDC_BRIDGED, USDC_NATIVE])
         ]
+        self._chunk_size_blocks = chunk_size_blocks
+        self._max_lookback_blocks = max_lookback_blocks
 
     async def trace(
         self,
@@ -209,46 +256,144 @@ class FundingTracer:
     ) -> list[dict[str, Any]]:
         """Get ERC20 Transfer event logs.
 
+        Public Polygon RPCs (publicnode, ankr, llamarpc) cap ``eth_getLogs`` at
+        10_000 blocks per call. To work around this we resolve the requested
+        range into a concrete block window (defaulting to the last
+        ``max_lookback_blocks`` when caller passes ``from_block=0``) and walk
+        the window in chunks of ``chunk_size_blocks``, oldest-first, stopping
+        once ``limit`` matches are collected. Walking oldest-first preserves
+        the "first transfer" semantics expected by the funding chain tracer.
+
+        If a chunk comes back with a "history has been pruned" style error
+        the rest of the walk is short-circuited — every subsequent chunk
+        would hit the same archive cutoff and there's no point burning quota
+        on guaranteed failures.
+
         Args:
             to_address: Filter by recipient address.
             token_address: ERC20 token contract address.
             limit: Maximum logs to return.
-            from_block: Starting block number.
+            from_block: Starting block number (0 means
-            to_block: Ending block number.
+                ``latest - max_lookback_blocks``).
+            to_block: Ending block number ("latest" resolves to current head).
 
         Returns:
-            List of log dictionaries.
+            List of log dictionaries, oldest first, capped at ``limit``.
         """
         # Pad address to 32 bytes for topic filter
         padded_to = "0x" + to_address.lower().replace("0x", "").zfill(64)
+        topics = [
+            TRANSFER_EVENT_TOPIC,  # Transfer event (must be 0x-prefixed for drpc)
+            None,  # from (any)
+            padded_to,  # to (target address)
+        ]
+        contract_address = AsyncWeb3.to_checksum_address(token_address)
 
+        start_block, end_block = await self._resolve_block_range(from_block, to_block)
+        if start_block > end_block:
+            return []
+
+        results: list[dict[str, Any]] = []
+        chunk_size = max(1, self._chunk_size_blocks)
+        chunk_start = start_block
+
+        while chunk_start <= end_block:
+            chunk_end = min(chunk_start + chunk_size - 1, end_block)
+            try:
+                chunk_logs = await self._fetch_logs_chunk(
+                    contract_address=contract_address,
+                    topics=topics,
+                    from_block=chunk_start,
+                    to_block=chunk_end,
+                )
+            except Exception as e:
+                if _is_pruned_history_error(e):
+                    # The provider has dropped this slice of history. Walking
+                    # further back will hit the same wall on every chunk;
+                    # stop now and return what we already have.
+                    logger.info(
+                        "eth_getLogs chunk %d-%d outside archive horizon for %s; stopping trace",
+                        chunk_start,
+                        chunk_end,
+                        to_address,
+                    )
+                    break
+                logger.warning(
+                    "eth_getLogs chunk %d-%d failed for %s: %s",
+                    chunk_start,
+                    chunk_end,
+                    to_address,
+                    e,
+                )
+                # Skip this window and keep walking — partial data is better
+                # than aborting the whole trace on a single flaky chunk.
+                chunk_start = chunk_end + 1
+                continue
+
+            for log in chunk_logs:
+                results.append(dict(log))
+                if len(results) >= limit:
+                    return results
+
+            chunk_start = chunk_end + 1
+
+        return results
+
+    async def _resolve_block_range(
+        self,
+        from_block: int | str,
+        to_block: int | str,
+    ) -> tuple[int, int]:
+        """Resolve symbolic block params to concrete numeric bounds.
+
+        ``from_block=0`` (the historical default) is rewritten to
+        ``latest - max_lookback_blocks`` so we don't try to scan all of Polygon.
+        """
+        w3 = self._select_w3()
+
+        if isinstance(to_block, str):
+            await self.polygon_client._rate_limiter.acquire()
+            head = int(await w3.eth.block_number)
+            end = head
+        else:
+            end = int(to_block)
+
+        if isinstance(from_block, str):
+            # Treat any symbolic from-block (e.g. "earliest") as "go back
+            # max_lookback_blocks from end"; that's what callers actually want.
+            start = max(0, end - self._max_lookback_blocks)
+        elif from_block == 0:
+            start = max(0, end - self._max_lookback_blocks)
+        else:
+            start = int(from_block)
+
+        return start, end
+
+    async def _fetch_logs_chunk(
+        self,
+        contract_address: str,
+        topics: list[Any],
+        from_block: int,
+        to_block: int,
+    ) -> list[Any]:
+        """Issue a single bounded ``eth_getLogs`` call."""
         await self.polygon_client._rate_limiter.acquire()
-
+        w3 = self._select_w3()
-        # Use the web3 instance from polygon client
-        w3 = (
-            self.polygon_client._w3
-            if self.polygon_client._primary_healthy
-            else (self.polygon_client._w3_fallback or self.polygon_client._w3)
-        )
-
-        # Get logs with Transfer event filtering by recipient
         # Note: web3 typing is overly restrictive for block params
-        logs = await w3.eth.get_logs(
+        return await w3.eth.get_logs(
             {
-                "address": AsyncWeb3.to_checksum_address(token_address),
+                "address": contract_address,
-                "topics": [
+                "topics": topics,
-                    TRANSFER_EVENT_SIGNATURE.hex(),  # Transfer event
-                    None,  # from (any)
-                    padded_to,  # to (target address)
-                ],
                 "fromBlock": from_block,  # type: ignore[typeddict-item]
                 "toBlock": to_block,  # type: ignore[typeddict-item]
             }
         )
 
-        # Convert to list of dicts and limit
+    def _select_w3(self) -> AsyncWeb3:
-        result = [dict(log) for log in logs[:limit]]
+        """Pick primary or fallback web3 instance based on health."""
-        return result
+        if self.polygon_client._primary_healthy:
+            return self.polygon_client._w3
+        return self.polygon_client._w3_fallback or self.polygon_client._w3
 
     async def _log_to_funding_transfer(
         self,

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/alerter/test_formatter.py

@@ -409,12 +409,27 @@ class TestTelegramMarkdown:
         assert "`0x1234...5678`" in result.telegram_markdown
 
     def test_telegram_includes_risk_score(self, high_risk_assessment: RiskAssessment) -> None:
-        """Test that Telegram message includes risk score."""
+        """Test that Telegram message includes risk score, MarkdownV2-escaped."""
         formatter = AlertFormatter()
         result = formatter.format(high_risk_assessment)
-        assert "0.82" in result.telegram_markdown
+        # MarkdownV2 requires `.` to be escaped, so 0.82 becomes 0\.82.
+        assert "0\\.82" in result.telegram_markdown
         assert "HIGH" in result.telegram_markdown
 
+    def test_telegram_escapes_all_decimals(self, high_risk_assessment: RiskAssessment) -> None:
+        """Telegram MarkdownV2 rejects unescaped `.` in dynamic numeric
+        fields (risk score, price, USDC amount). All must be present in
+        their escaped form."""
+        formatter = AlertFormatter()
+        result = formatter.format(high_risk_assessment)
+        md = result.telegram_markdown
+        for unescaped in ("0.82", "0.075", "15,000.00"):
+            assert unescaped not in md, (
+                f"unescaped {unescaped!r} would be rejected by Telegram MarkdownV2: {md!r}"
+            )
+        for escaped in ("0\\.82", "0\\.075", "15,000\\.00"):
+            assert escaped in md, f"missing escaped {escaped!r} in {md!r}"
+
     def test_telegram_includes_links(self, high_risk_assessment: RiskAssessment) -> None:
         """Test that Telegram message includes links."""
         formatter = AlertFormatter()

tests/detector/test_size_anomaly.py

@@ -588,6 +588,93 @@ class TestAnalyzeMethod:
         assert signal.is_niche_market is True
         assert signal.confidence == 0.2  # niche_base
 
+    @pytest.mark.asyncio
+    async def test_analyze_niche_only_below_min_trade_size_skipped(
+        self,
+        mock_metadata_sync: AsyncMock,
+        sample_metadata: MarketMetadata,
+    ) -> None:
+        """Niche-only trades below the min trade size are suppressed."""
+        mock_metadata_sync.get_market.return_value = sample_metadata
+        detector = SizeAnomalyDetector(mock_metadata_sync)
+
+        tiny_trade = TradeEvent(
+            market_id="market_abc123",
+            trade_id="tx_tiny",
+            wallet_address="0xabc",
+            side="BUY",
+            outcome="Yes",
+            outcome_index=0,
+            price=Decimal("0.5"),
+            size=Decimal("100"),  # $50 notional, below default $500 floor
+            timestamp=datetime.now(UTC),
+            asset_id="token_123",
+            event_title="Niche tiny trade",
+        )
+
+        signal = await detector.analyze(tiny_trade)
+        assert signal is None
+
+    @pytest.mark.asyncio
+    async def test_analyze_niche_only_at_min_trade_size_emits(
+        self,
+        mock_metadata_sync: AsyncMock,
+        sample_metadata: MarketMetadata,
+    ) -> None:
+        """Niche-only trades at or above the min trade size still emit."""
+        mock_metadata_sync.get_market.return_value = sample_metadata
+        detector = SizeAnomalyDetector(mock_metadata_sync)
+
+        ok_trade = TradeEvent(
+            market_id="market_abc123",
+            trade_id="tx_ok",
+            wallet_address="0xabc",
+            side="BUY",
+            outcome="Yes",
+            outcome_index=0,
+            price=Decimal("0.5"),
+            size=Decimal("2000"),  # $1000 notional, above default $500 floor
+            timestamp=datetime.now(UTC),
+            asset_id="token_123",
+            event_title="Niche ok trade",
+        )
+
+        signal = await detector.analyze(ok_trade)
+        assert signal is not None
+        assert signal.is_niche_market is True
+        assert signal.confidence == 0.2
+
+    @pytest.mark.asyncio
+    async def test_niche_min_trade_size_does_not_block_real_anomalies(
+        self,
+        mock_metadata_sync: AsyncMock,
+        sample_metadata: MarketMetadata,
+    ) -> None:
+        """A trade that exceeds volume/book thresholds is never blocked by the niche guard."""
+        mock_metadata_sync.get_market.return_value = sample_metadata
+        detector = SizeAnomalyDetector(mock_metadata_sync)
+
+        small_but_high_impact_trade = TradeEvent(
+            market_id="market_abc123",
+            trade_id="tx_small_impact",
+            wallet_address="0xabc",
+            side="BUY",
+            outcome="Yes",
+            outcome_index=0,
+            price=Decimal("0.5"),
+            size=Decimal("200"),  # $100 notional, below niche floor
+            timestamp=datetime.now(UTC),
+            asset_id="token_123",
+            event_title="Small but high-impact",
+        )
+
+        # Provide small daily_volume so volume_impact exceeds 2% threshold
+        signal = await detector.analyze(
+            small_but_high_impact_trade, daily_volume=Decimal("1000")
+        )
+        assert signal is not None
+        assert signal.volume_impact > 0.02
+
     @pytest.mark.asyncio
     async def test_analyze_no_anomaly(
         self,

tests/ingestor/test_gamma_client.py

@@ -0,0 +1,249 @@
+"""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,6 +8,7 @@ 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,
@@ -72,6 +73,18 @@ 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."""
 
@@ -195,9 +208,9 @@ class TestSyncStats:
 class TestMarketMetadataSync:
     """Tests for the MarketMetadataSync class."""
 
-    def test_init(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
+    def test_init(self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock) -> None:
         """Test initialization."""
-        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
 
         assert sync.state == SyncState.STOPPED
         assert sync.stats.total_syncs == 0
@@ -205,11 +218,14 @@ 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) -> None:
+    def test_init_custom_config(
+        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: 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:",
@@ -220,9 +236,11 @@ class TestMarketMetadataSync:
         assert sync._key_prefix == "custom:"
 
     @pytest.mark.asyncio
-    async def test_start_stop(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
+    async def test_start_stop(
+        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
+    ) -> None:
         """Test starting and stopping the sync service."""
-        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
 
         # Start
         await sync.start()
@@ -236,10 +254,10 @@ class TestMarketMetadataSync:
 
     @pytest.mark.asyncio
     async def test_start_performs_initial_sync(
-        self, mock_redis: AsyncMock, mock_clob: MagicMock
+        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
     ) -> None:
         """Test that start performs an initial sync."""
-        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
 
         await sync.start()
 
@@ -252,10 +270,12 @@ class TestMarketMetadataSync:
         await sync.stop()
 
     @pytest.mark.asyncio
-    async def test_start_failure(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
+    async def test_start_failure(
+        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
+    ) -> None:
         """Test start failure handling."""
         mock_clob.get_markets.side_effect = Exception("API error")
-        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
 
         with pytest.raises(MetadataSyncError, match="initial sync failed"):
             await sync.start()
@@ -268,6 +288,7 @@ class TestMarketMetadataSync:
         self,
         mock_redis: AsyncMock,
         mock_clob: MagicMock,
+        mock_gamma: MagicMock,
         sample_metadata: MarketMetadata,
     ) -> None:
         """Test get_market with cache hit."""
@@ -275,7 +296,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)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
         await sync.start()
 
         result = await sync.get_market("cond123")
@@ -288,12 +309,14 @@ class TestMarketMetadataSync:
         await sync.stop()
 
     @pytest.mark.asyncio
-    async def test_get_market_cache_miss(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
+    async def test_get_market_cache_miss(
+        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: 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)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
         await sync.start()
 
         result = await sync.get_market("cond123")
@@ -308,12 +331,14 @@ class TestMarketMetadataSync:
         await sync.stop()
 
     @pytest.mark.asyncio
-    async def test_get_market_not_found(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
+    async def test_get_market_not_found(
+        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: 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)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
         await sync.start()
 
         result = await sync.get_market("nonexistent")
@@ -323,9 +348,11 @@ class TestMarketMetadataSync:
         await sync.stop()
 
     @pytest.mark.asyncio
-    async def test_invalidate_market(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
+    async def test_invalidate_market(
+        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
+    ) -> None:
         """Test cache invalidation."""
-        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
         await sync.start()
 
         result = await sync.invalidate_market("cond123")
@@ -336,9 +363,11 @@ class TestMarketMetadataSync:
         await sync.stop()
 
     @pytest.mark.asyncio
-    async def test_force_sync(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
+    async def test_force_sync(
+        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
+    ) -> None:
         """Test forced sync."""
-        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
         await sync.start()
 
         # Initial sync
@@ -353,7 +382,9 @@ class TestMarketMetadataSync:
         await sync.stop()
 
     @pytest.mark.asyncio
-    async def test_state_change_callback(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
+    async def test_state_change_callback(
+        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
+    ) -> None:
         """Test state change callback."""
         states: list[SyncState] = []
 
@@ -363,6 +394,7 @@ class TestMarketMetadataSync:
         sync = MarketMetadataSync(
             redis=mock_redis,
             clob_client=mock_clob,
+            gamma_client=mock_gamma,
             on_state_change=on_state_change,
         )
 
@@ -377,7 +409,7 @@ class TestMarketMetadataSync:
 
     @pytest.mark.asyncio
     async def test_sync_complete_callback(
-        self, mock_redis: AsyncMock, mock_clob: MagicMock
+        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
     ) -> None:
         """Test sync complete callback."""
         sync_stats: list[SyncStats] = []
@@ -388,6 +420,7 @@ class TestMarketMetadataSync:
         sync = MarketMetadataSync(
             redis=mock_redis,
             clob_client=mock_clob,
+            gamma_client=mock_gamma,
             on_sync_complete=on_sync_complete,
         )
 
@@ -401,7 +434,11 @@ class TestMarketMetadataSync:
 
     @pytest.mark.asyncio
     async def test_get_markets_by_category(
-        self, mock_redis: AsyncMock, mock_clob: MagicMock, sample_metadata: MarketMetadata
+        self,
+        mock_redis: AsyncMock,
+        mock_clob: MagicMock,
+        mock_gamma: MagicMock,
+        sample_metadata: MarketMetadata,
     ) -> None:
         """Test getting markets by category."""
         # Setup scan to return keys
@@ -412,7 +449,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)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
         # Don't start to avoid initial sync complexity
         sync._state = SyncState.IDLE
 
@@ -422,9 +459,11 @@ class TestMarketMetadataSync:
         assert results[0].category == "crypto"
 
     @pytest.mark.asyncio
-    async def test_cannot_start_twice(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
+    async def test_cannot_start_twice(
+        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
+    ) -> None:
         """Test that starting twice doesn't double-start."""
-        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
 
         await sync.start()
         await sync.start()  # Should be a no-op
@@ -434,9 +473,11 @@ class TestMarketMetadataSync:
         await sync.stop()
 
     @pytest.mark.asyncio
-    async def test_stop_when_stopped(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
+    async def test_stop_when_stopped(
+        self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
+    ) -> None:
         """Test stopping when already stopped."""
-        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
 
         await sync.stop()  # Should be a no-op
 

tests/profiler/test_funding.py

@@ -327,6 +327,10 @@ class TestGetTransferLogs:
         await funding_tracer._get_transfer_logs(
             to_address=TEST_WALLET,
             token_address=USDC_BRIDGED,
+            # Explicit numeric range so we stay inside one chunk and skip
+            # the "latest" → block_number resolution path.
+            from_block=1,
+            to_block=8_000,
         )
 
         mock_w3.eth.get_logs.assert_called_once()
@@ -334,10 +338,15 @@ class TestGetTransferLogs:
 
         # Verify topics structure
         assert len(call_args["topics"]) == 3
-        assert call_args["topics"][0] == TRANSFER_EVENT_SIGNATURE.hex()
+        # The Transfer event topic must be 0x-prefixed; drpc rejects bare hex.
+        assert call_args["topics"][0] == "0x" + TRANSFER_EVENT_SIGNATURE.hex().removeprefix("0x")
+        assert call_args["topics"][0].startswith("0x")
         assert call_args["topics"][1] is None  # from (any)
         # to address should be padded to 32 bytes
         assert call_args["topics"][2].endswith(TEST_WALLET.lower().replace("0x", ""))
+        # And the chunk bounds match what we asked for.
+        assert call_args["fromBlock"] == 1
+        assert call_args["toBlock"] == 8_000
 
     @pytest.mark.asyncio
     async def test_get_transfer_logs_respects_limit(
@@ -355,6 +364,8 @@ class TestGetTransferLogs:
             to_address=TEST_WALLET,
             token_address=USDC_BRIDGED,
             limit=3,
+            from_block=1,
+            to_block=8_000,
         )
 
         assert len(result) == 3
@@ -374,10 +385,242 @@ class TestGetTransferLogs:
         await funding_tracer._get_transfer_logs(
             to_address=TEST_WALLET,
             token_address=USDC_BRIDGED,
+            from_block=1,
+            to_block=8_000,
         )
 
         mock_fallback.eth.get_logs.assert_called_once()
 
+    @pytest.mark.asyncio
+    async def test_get_transfer_logs_chunks_large_ranges(
+        self,
+        funding_tracer: FundingTracer,
+        mock_polygon_client: MagicMock,
+    ) -> None:
+        """Ranges wider than chunk_size are split into multiple eth_getLogs calls.
+
+        This is the regression guard for the publicnode 10_000-block cap that
+        was rejecting every funding trace before chunking landed.
+        """
+        mock_w3 = MagicMock()
+        mock_w3.eth.get_logs = AsyncMock(return_value=[])
+        mock_polygon_client._w3 = mock_w3
+
+        # 25_000 blocks at 9_000-per-chunk → 3 calls (9000 + 9000 + 7001).
+        await funding_tracer._get_transfer_logs(
+            to_address=TEST_WALLET,
+            token_address=USDC_BRIDGED,
+            from_block=1_000_000,
+            to_block=1_025_000,
+        )
+
+        assert mock_w3.eth.get_logs.call_count == 3
+        windows = [call[0][0] for call in mock_w3.eth.get_logs.call_args_list]
+        assert windows[0]["fromBlock"] == 1_000_000
+        assert windows[0]["toBlock"] == 1_008_999
+        assert windows[1]["fromBlock"] == 1_009_000
+        assert windows[1]["toBlock"] == 1_017_999
+        assert windows[2]["fromBlock"] == 1_018_000
+        assert windows[2]["toBlock"] == 1_025_000
+        # No window exceeds the chunk size — that's what RPC providers reject.
+        for win in windows:
+            assert win["toBlock"] - win["fromBlock"] + 1 <= 9_000
+
+    @pytest.mark.asyncio
+    async def test_get_transfer_logs_stops_when_limit_hit_mid_walk(
+        self,
+        funding_tracer: FundingTracer,
+        mock_polygon_client: MagicMock,
+    ) -> None:
+        """Walking should stop as soon as ``limit`` matches are gathered."""
+        mock_w3 = MagicMock()
+        # First chunk yields 5 logs, more than the limit, so subsequent chunks
+        # must not be queried.
+        mock_w3.eth.get_logs = AsyncMock(return_value=[MagicMock() for _ in range(5)])
+        mock_polygon_client._w3 = mock_w3
+
+        result = await funding_tracer._get_transfer_logs(
+            to_address=TEST_WALLET,
+            token_address=USDC_BRIDGED,
+            limit=2,
+            from_block=1_000_000,
+            to_block=1_025_000,
+        )
+
+        assert len(result) == 2
+        mock_w3.eth.get_logs.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_get_transfer_logs_skips_failing_chunk(
+        self,
+        funding_tracer: FundingTracer,
+        mock_polygon_client: MagicMock,
+    ) -> None:
+        """A flaky chunk must not abort the whole trace — we move on."""
+        mock_w3 = MagicMock()
+        good_log = MagicMock()
+        responses: list[Any] = [
+            RuntimeError("RPC hiccup"),
+            [good_log],
+        ]
+
+        async def fake_get_logs(_params: dict[str, Any]) -> list[Any]:
+            outcome = responses.pop(0)
+            if isinstance(outcome, BaseException):
+                raise outcome
+            return outcome
+
+        mock_w3.eth.get_logs = AsyncMock(side_effect=fake_get_logs)
+        mock_polygon_client._w3 = mock_w3
+
+        result = await funding_tracer._get_transfer_logs(
+            to_address=TEST_WALLET,
+            token_address=USDC_BRIDGED,
+            from_block=1_000_000,
+            to_block=1_018_000,  # forces 3 chunks; we exercise chunks 1+2
+        )
+
+        # The error chunk is skipped; the second chunk contributes one log.
+        assert result == [dict(good_log)]
+        assert mock_w3.eth.get_logs.call_count >= 2
+
+    @pytest.mark.asyncio
+    async def test_get_transfer_logs_resolves_latest_via_block_number(
+        self,
+        funding_tracer: FundingTracer,
+        mock_polygon_client: MagicMock,
+    ) -> None:
+        """``to_block='latest'`` should resolve via ``eth.block_number``.
+
+        And ``from_block=0`` should not become a full-history scan — it must
+        be clamped to ``latest - max_lookback_blocks``.
+        """
+
+        async def _block_number_coro() -> int:
+            return 5_000
+
+        mock_eth = MagicMock()
+        mock_eth.get_logs = AsyncMock(return_value=[])
+        # Property-style awaitable: web3.py exposes block_number as a property
+        # returning a coroutine, so each access must yield a fresh awaitable.
+        type(mock_eth).block_number = property(  # type: ignore[misc]
+            lambda _self: _block_number_coro()
+        )
+        mock_w3 = MagicMock()
+        mock_w3.eth = mock_eth
+        mock_polygon_client._w3 = mock_w3
+
+        await funding_tracer._get_transfer_logs(
+            to_address=TEST_WALLET,
+            token_address=USDC_BRIDGED,
+        )
+
+        # block_number=5000 < chunk_size, so it's one chunk that bottoms at 0.
+        mock_eth.get_logs.assert_called_once()
+        call_args = mock_eth.get_logs.call_args[0][0]
+        assert call_args["fromBlock"] == 0
+        assert call_args["toBlock"] == 5_000
+
+    @pytest.mark.asyncio
+    async def test_get_transfer_logs_breaks_on_pruned_history(
+        self,
+        funding_tracer: FundingTracer,
+        mock_polygon_client: MagicMock,
+    ) -> None:
+        """A pruned-history error must short-circuit the whole walk.
+
+        Public Polygon RPCs prune log history. Once we walk past the cutoff,
+        every subsequent chunk will raise the same error — keep walking and
+        we just burn quota on guaranteed failures. The first such error must
+        end the walk and return whatever we already collected.
+        """
+        mock_w3 = MagicMock()
+        good_log = MagicMock()
+
+        responses: list[Any] = [
+            [good_log],
+            RuntimeError(
+                "{'code': -32701, 'message': 'History has been pruned for "
+                "this block. To remove restrictions, order a dedicated full "
+                "node here: https://www.allnodes.com/pol/host'}"
+            ),
+            # If the early-break logic is missing, this third chunk would
+            # also be requested. The test asserts it isn't.
+            [MagicMock()],
+        ]
+
+        async def fake_get_logs(_params: dict[str, Any]) -> list[Any]:
+            outcome = responses.pop(0)
+            if isinstance(outcome, BaseException):
+                raise outcome
+            return outcome
+
+        mock_w3.eth.get_logs = AsyncMock(side_effect=fake_get_logs)
+        mock_polygon_client._w3 = mock_w3
+
+        # 3 chunks total. The pruned error fires on chunk #2; chunk #3 must
+        # never be issued.
+        result = await funding_tracer._get_transfer_logs(
+            to_address=TEST_WALLET,
+            token_address=USDC_BRIDGED,
+            from_block=1_000_000,
+            to_block=1_027_000,
+        )
+
+        assert result == [dict(good_log)]
+        assert mock_w3.eth.get_logs.call_count == 2
+
+    @pytest.mark.asyncio
+    async def test_get_transfer_logs_default_lookback_fits_pruned_horizon(
+        self,
+    ) -> None:
+        """Default ``max_lookback_blocks`` must stay inside what public RPCs serve.
+
+        publicnode prunes after ~100k blocks. If we default to 1.3M, every
+        funding trace blows through the archive horizon and produces nothing
+        but pruned-history warnings. Pin the default at <= 100k as a
+        regression guard.
+        """
+        from polymarket_insider_tracker.profiler.funding import (
+            DEFAULT_MAX_LOOKBACK_BLOCKS,
+        )
+
+        assert DEFAULT_MAX_LOOKBACK_BLOCKS <= 100_000
+
+    @pytest.mark.asyncio
+    async def test_get_transfer_logs_topic_is_0x_prefixed(
+        self,
+        funding_tracer: FundingTracer,
+        mock_polygon_client: MagicMock,
+    ) -> None:
+        """The Transfer event topic passed to ``eth_getLogs`` must begin with ``0x``.
+
+        ``HexBytes.hex()`` returns a bare hex string. publicnode tolerates
+        that, but stricter providers like drpc (our fallback) reject it with
+        ``invalid argument 0: hex string without 0x prefix`` and every chunk
+        in the trace fails. This guards against regressing back to the
+        bare-hex form.
+        """
+        mock_w3 = MagicMock()
+        mock_w3.eth.get_logs = AsyncMock(return_value=[])
+        mock_polygon_client._w3 = mock_w3
+
+        await funding_tracer._get_transfer_logs(
+            to_address=TEST_WALLET,
+            token_address=USDC_BRIDGED,
+            from_block=1,
+            to_block=8_000,
+        )
+
+        topics = mock_w3.eth.get_logs.call_args[0][0]["topics"]
+        assert topics[0].startswith("0x")
+        # And the topic also has to be 32 bytes (64 hex chars) as required by
+        # the JSON-RPC spec.
+        assert len(topics[0]) == 2 + 64
+        # The padded `to` topic was already 0x-prefixed; double-check that
+        # didn't regress either.
+        assert topics[2].startswith("0x")
+
 
 class TestLogToFundingTransfer:
     """Tests for _log_to_funding_transfer method."""

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