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
-        lines.append(f"*Risk Score:* {assessment.weighted_score:.2f} \\({risk_level}\\)")
+        # Risk score — every numeric literal here must be escaped because
+        # 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("$", "\\$")
-        lines.append(
-            f"*Trade:* {trade.side} {trade.outcome} @ \\${trade.price:.3f} \\| {usdc_value}"
-        )
+        usdc_value = self._escape_telegram_markdown(format_usdc(trade.notional_value))
+        price_str = self._escape_telegram_markdown(f"{trade.price:.3f}")
+        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_ALERT_THRESHOLD = 0.6
+# Default configuration. The threshold lifted from 0.6 to 0.80 after the
+# first cost-adjusted backtest showed everything below 0.85 was follower-PnL
+# negative under realistic taker fees + half-cent slippage. 0.80 keeps a small
+# margin below 0.85+ so we don't drop borderline-high signals on a hard cliff.
+# Override at runtime via DETECTOR_ALERT_THRESHOLD env var.
+DEFAULT_ALERT_THRESHOLD = 0.80
 DEFAULT_DEDUP_WINDOW_SECONDS = 3600  # 1 hour
 DEFAULT_REDIS_KEY_PREFIX = "polymarket:dedup:"
 

src/polymarket_insider_tracker/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/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.
-            to_block: Ending block number.
+            from_block: Starting block number (0 means
+                ``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()
-
-        # 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
+        w3 = self._select_w3()
         # 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),
-                "topics": [
-                    TRANSFER_EVENT_SIGNATURE.hex(),  # Transfer event
-                    None,  # from (any)
-                    padded_to,  # to (target address)
-                ],
+                "address": contract_address,
+                "topics": topics,
                 "fromBlock": from_block,  # type: ignore[typeddict-item]
                 "toBlock": to_block,  # type: ignore[typeddict-item]
             }
         )
 
-        # Convert to list of dicts and limit
-        result = [dict(log) for log in logs[:limit]]
-        return result
+    def _select_w3(self) -> AsyncWeb3:
+        """Pick primary or fallback web3 instance based on health."""
+        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/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