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>

.env.example

@@ -6,12 +6,12 @@ POSTGRES_USER=tracker
 POSTGRES_PASSWORD=dev_password
 
 # Constructed database URL (for application use)
-DATABASE_URL=postgresql://${POSTGRES_USER}:${POSTGRES_PASSWORD}@${POSTGRES_HOST}:${POSTGRES_PORT}/${POSTGRES_DB}
+DATABASE_URL=postgresql://tracker:dev_password@localhost:5432/polymarket_tracker
 
 # Redis Configuration
 REDIS_HOST=localhost
 REDIS_PORT=6379
-REDIS_URL=redis://${REDIS_HOST}:${REDIS_PORT}
+REDIS_URL=redis://localhost:6379
 
 # Optional: Development tool ports
 ADMINER_PORT=8080

.github/workflows/ci.yml

@@ -0,0 +1,98 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+
+jobs:
+  lint:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+          cache: "pip"
+
+      - name: Install dependencies
+        run: pip install -e ".[dev]"
+
+      - name: Lint with ruff
+        run: ruff check src/ tests/
+
+      - name: Check formatting with ruff
+        run: ruff format --check src/ tests/
+
+  type-check:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+          cache: "pip"
+
+      - name: Install dependencies
+        run: pip install -e ".[dev]"
+
+      - name: Type check with mypy
+        # TODO: Remove continue-on-error after fixing #48
+        continue-on-error: true
+        run: mypy src/
+
+  test:
+    runs-on: ubuntu-latest
+
+    services:
+      postgres:
+        image: postgres:15
+        env:
+          POSTGRES_DB: test_db
+          POSTGRES_USER: test
+          POSTGRES_PASSWORD: test
+        ports:
+          - "5432:5432"
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
+      redis:
+        image: redis:7
+        ports:
+          - "6379:6379"
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+          cache: "pip"
+
+      - name: Install dependencies
+        run: pip install -e ".[dev]"
+
+      - name: Run tests with coverage
+        run: pytest --cov=src --cov-report=xml --cov-report=term-missing
+        env:
+          DATABASE_URL: postgresql://test:test@localhost:5432/test_db
+          REDIS_URL: redis://localhost:6379
+
+      - name: Upload coverage to Codecov
+        uses: codecov/codecov-action@v4
+        with:
+          files: ./coverage.xml
+          fail_ci_if_error: false
+        env:
+          CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}

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.

README.md

@@ -2,80 +2,103 @@
 
 **Detect informed money before the market moves.**
 
-[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
+[![CI](https://github.com/pselamy/polymarket-insider-tracker/actions/workflows/ci.yml/badge.svg)](https://github.com/pselamy/polymarket-insider-tracker/actions/workflows/ci.yml)
 [![Python 3.11+](https://img.shields.io/badge/python-3.11+-blue.svg)](https://www.python.org/downloads/)
+[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
+
+Real-time detection of suspicious trading patterns on Polymarket: fresh wallets, unusual sizing, niche-market activity, and funding chain analysis. Streams trades via WebSocket, profiles wallets on-chain (Polygon), scores risk with ML + heuristics, and dispatches alerts to Discord/Telegram.
 
 ---
 
-## The Opportunity
+## Quick Start (< 2 minutes)
 
-On January 3, 2026, a trader spotted a significant political event on Polymarket **before it happened**. How? Not by predicting the future, but by tracking suspicious trading behavior.
+### 1. Install
 
-> "You don't need to predict the future, you need to track suspicious behavior."
-> — [@DidiTrading](https://x.com/DidiTrading)
-
-An insider wallet turned **$35,000 into $442,000** (12.6x return) by entering a position hours before a major market move. The tool that detected this activity flagged five separate alerts before the event occurred.
-
-**This repository builds that tool.**
-
----
-
-## What This Does
-
-The Polymarket Insider Tracker monitors prediction market trading activity in real-time and identifies patterns that suggest informed trading:
-
-| Signal | What It Detects | Why It Matters |
-|--------|-----------------|----------------|
-| **Fresh Wallets** | Brand new wallets making large trades | Insiders create new wallets to hide their identity |
-| **Unusual Sizing** | Trades that are disproportionately large for the market | Informed traders bet bigger when they have edge |
-| **Niche Markets** | Activity in low-volume, specific-outcome markets | Easier to have inside information on obscure events |
-| **Funding Chains** | Where wallet funds originated from | Links seemingly separate wallets to the same entity |
-
-When suspicious activity is detected, you receive an instant alert with actionable intelligence.
-
----
-
-## How It Works
-
-```
-┌─────────────────┐     ┌──────────────────┐     ┌────────────────────┐
-│  Polymarket API │────>│  Wallet Profiler │────>│  Anomaly Detector  │
-│  (Real-time)    │     │  (Blockchain)    │     │  (ML + Heuristics) │
-└─────────────────┘     └──────────────────┘     └────────────────────┘
-                                                          │
-                              ┌────────────────────────────┘
-                              v
-                   ┌─────────────────────┐
-                   │   Alert Dispatcher  │───> Discord / Telegram / Email
-                   │   "Fresh wallet     │
-                   │    buying YES @7.5¢ │
-                   │    on niche market" │
-                   └─────────────────────┘
+```bash
+# Requires: Python 3.11+, Docker
+git clone https://github.com/pselamy/polymarket-insider-tracker.git
+cd polymarket-insider-tracker
+uv sync --all-extras          # or: pip install -e ".[dev]"
 ```
 
-### Detection Algorithms
+### 2. Start infrastructure
 
-1. **Fresh Wallet Detection**
-   - Checks wallet transaction history on Polygon
-   - Flags wallets with fewer than 5 lifetime transactions making trades over $1,000
-   - Traces funding source to identify if connected to known entities
+```bash
+docker compose up -d           # PostgreSQL 15 + Redis 7
+docker compose ps              # wait for healthy
+```
 
-2. **Liquidity Impact Analysis**
-   - Calculates trade size relative to market depth
-   - Flags trades consuming more than 2% of visible order book
-   - Weights by market category (niche markets score higher)
+### 3. Configure
 
-3. **Sniper Cluster Detection**
-   - Uses DBSCAN clustering to find wallets that consistently enter markets within minutes of creation
-   - Identifies coordinated behavior patterns
+```bash
+cp .env.example .env
+# Edit .env — only DATABASE_URL and REDIS_URL are required for local dev
+# (defaults in .env.example work with docker compose)
+```
 
-4. **Event Correlation**
-   - Cross-references trading activity with news feeds
-   - Detects positions opened 1-4 hours before related news breaks
+### 4. Run migrations + start
+
+```bash
+uv run alembic upgrade head
+uv run python -m polymarket_insider_tracker
+```
+
+You should see live trades within seconds:
+
+```
+INFO  Connection state: disconnected -> connecting
+INFO  Connected to wss://ws-live-data.polymarket.com and subscribed to trades
+DEBUG Trade: BUY 450 @ 1.00 on fifwc-ger-kor-2026-06-14-ger
+DEBUG Trade: SELL 5 @ 0.86 on chi1-cd1-cdl-2026-06-14-draw
+```
+
+### CLI Options
+
+```bash
+python -m polymarket_insider_tracker --help
+  --version          Show version
+  --config-check     Validate configuration and exit
+  --log-level DEBUG  Override log level
+  --dry-run          Run pipeline without sending alerts
+  --health-port 8080 Override health check port
+```
 
 ---
 
-## Sample Alert
+## Environment Variables
+
+| Variable | Required | Default | Description |
+|----------|----------|---------|-------------|
+| `DATABASE_URL` | Yes | — | PostgreSQL connection string |
+| `REDIS_URL` | No | `redis://localhost:6379` | Redis connection string |
+| `POLYGON_RPC_URL` | No | `https://polygon-rpc.com` | Polygon RPC (public default works) |
+| `POLYGON_FALLBACK_RPC_URL` | No | — | Fallback RPC endpoint |
+| `POLYMARKET_WS_URL` | No | `wss://ws-live-data.polymarket.com` | WebSocket endpoint |
+| `POLYMARKET_API_KEY` | No | — | Optional API key for higher rate limits |
+| `DISCORD_WEBHOOK_URL` | No | — | Discord alerts |
+| `TELEGRAM_BOT_TOKEN` | No | — | Telegram alerts (needs `TELEGRAM_CHAT_ID` too) |
+| `TELEGRAM_CHAT_ID` | No | — | Telegram chat for alerts |
+| `LOG_LEVEL` | No | `INFO` | Logging level |
+| `DRY_RUN` | No | `false` | Skip sending alerts |
+| `HEALTH_PORT` | No | `8080` | Health check HTTP port |
+
+No API keys are needed for basic operation — the Polymarket WebSocket and CLOB REST APIs are public.
+
+---
+
+## What It Detects
+
+| Signal | Detection Method | Threshold |
+|--------|-----------------|-----------|
+| **Fresh Wallets** | Wallet age < 48h, nonce <= 5, making trades > $1k | Confidence 0.5-0.9 |
+| **Size Anomalies** | Trade size > 2% of 24h volume or > 5% of order book | Weighted by niche factor |
+| **Niche Markets** | Low-volume markets (< $50k daily) with specific outcomes | 1.5x risk multiplier |
+| **Funding Chains** | Trace wallet funding to known entities (exchanges, etc.) | On-chain lineage |
+| **Sniper Clusters** | DBSCAN clustering of wallets entering within minutes | Coordinated behavior |
+
+Risk scoring combines signals with configurable weights (default threshold: 0.6). Multi-signal bonuses: 2 signals +20%, 3+ signals +30%.
+
+### Sample Alert
 
 ```
 SUSPICIOUS ACTIVITY DETECTED
@@ -99,207 +122,62 @@ Confidence: HIGH (3/4 signals triggered)
 
 ---
 
-## Quick Start
+## Architecture
 
-### Prerequisites
+```
+Polymarket WebSocket ──> Ingestor ──> Profiler ──> Detector ──> Alerter
+(wss://ws-live-data)    (trades)    (on-chain)   (scoring)   (Discord/TG)
+                                        |
+                                   Polygon RPC
+```
 
-- Python 3.11+
-- Docker and Docker Compose
-- Polygon RPC endpoint (Alchemy, QuickNode, or self-hosted)
-- Polymarket API key (free at [docs.polymarket.com](https://docs.polymarket.com))
+### Components
 
-### Installation
+| Module | Purpose |
+|--------|---------|
+| `ingestor/` | WebSocket trade stream + CLOB REST client with rate limiting |
+| `profiler/` | Polygon wallet analysis, entity identification, funding chain tracing |
+| `detector/` | Fresh wallet, size anomaly, sniper cluster detection, composite risk scorer |
+| `alerter/` | Multi-channel dispatch (Discord webhooks, Telegram bot) with dedup |
+| `storage/` | SQLAlchemy ORM + Alembic migrations (PostgreSQL) |
+| `pipeline.py` | Orchestrator wiring all components together |
+| `shutdown.py` | Graceful SIGTERM/SIGINT handling with cleanup callbacks |
+
+---
+
+## Development
 
 ```bash
-# Clone the repository
-git clone https://github.com/pselamy/polymarket-insider-tracker.git
-cd polymarket-insider-tracker
-
-# Copy environment template
-cp .env.example .env
-# Edit .env with your API keys
-
-# Start infrastructure (PostgreSQL, Redis)
-docker compose up -d
-
-# Wait for services to be healthy
-docker compose ps
-
-# Install Python dependencies
-pip install -e .
-
-# Run database migrations
-alembic upgrade head
-
-# Run the tracker
-python -m src.main
+uv run pytest                        # run tests
+uv run ruff check src/ tests/        # lint
+uv run ruff format src/ tests/       # format
+uv run mypy src/                     # type check (strict mode)
 ```
 
 ### Docker Services
 
-The development stack includes:
-
 | Service | Port | Description |
 |---------|------|-------------|
 | PostgreSQL 15 | 5432 | Primary database |
 | Redis 7 | 6379 | Caching and pub/sub |
-| Adminer | 8080 | Database admin UI (optional) |
-| RedisInsight | 5540 | Redis admin UI (optional) |
-
-```bash
-# Start core services only
-docker compose up -d
-
-# Start with development tools (Adminer, RedisInsight)
-docker compose --profile tools up -d
-
-# View logs
-docker compose logs -f
-
-# Stop all services
-docker compose down
-
-# Stop and remove volumes (reset data)
-docker compose down -v
-```
-
-### Configuration
-
-```bash
-# .env file
-POLYGON_RPC_URL=https://polygon-mainnet.g.alchemy.com/v2/YOUR_KEY
-POLYMARKET_API_KEY=your_polymarket_api_key
-
-# Alert destinations (optional)
-DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/...
-TELEGRAM_BOT_TOKEN=your_bot_token
-TELEGRAM_CHAT_ID=your_chat_id
-
-# Detection thresholds
-MIN_TRADE_SIZE_USDC=1000
-FRESH_WALLET_MAX_NONCE=5
-LIQUIDITY_IMPACT_THRESHOLD=0.02
-```
+| Adminer | 8080 | Database admin UI (optional, `--profile tools`) |
+| RedisInsight | 5540 | Redis admin UI (optional, `--profile tools`) |
 
 ---
 
-## Project Structure
+## Troubleshooting
 
-```
-polymarket-insider-tracker/
-├── src/
-│   ├── ingestor/           # Real-time market data ingestion
-│   │   ├── clob_client.py  # Polymarket CLOB API wrapper
-│   │   └── websocket.py    # WebSocket event handler
-│   ├── profiler/           # Wallet analysis
-│   │   ├── analyzer.py     # Core wallet profiling logic
-│   │   ├── chain.py        # Polygon blockchain client
-│   │   └── funding.py      # Funding chain tracer
-│   ├── detector/           # Anomaly detection engines
-│   │   ├── fresh_wallet.py
-│   │   ├── size_anomaly.py
-│   │   ├── sniper.py       # DBSCAN clustering
-│   │   └── scorer.py       # Composite risk scoring
-│   ├── alerter/            # Notification dispatch
-│   │   ├── formatter.py    # Alert message formatting
-│   │   └── dispatcher.py   # Multi-channel delivery
-│   └── storage/            # Persistence layer
-│       ├── models.py       # SQLAlchemy models
-│       └── repos.py        # Repository pattern
-├── tests/                  # Test suite
-├── scripts/
-│   └── backtest.py         # Historical analysis
-├── docker-compose.yml
-├── pyproject.toml
-└── README.md
-```
+**No trades received / silent connection**
+The WebSocket subscription requires `action: "subscribe"` in the envelope. If you're on an older version, update — this was fixed in the WebSocket protocol alignment (see #89).
 
----
+**Connection timeout / DNS errors**
+Verify `wss://ws-live-data.polymarket.com` is reachable from your network. Some corporate firewalls block WebSocket connections.
 
-## Roadmap
+**Database migration errors**
+Ensure PostgreSQL is running (`docker compose ps`) and `DATABASE_URL` matches your docker-compose config. Run `uv run alembic upgrade head` after any schema changes.
 
-### Phase 1: Core Detection (Current)
-- [x] Project structure and documentation
-- [ ] Polymarket CLOB API integration
-- [ ] Fresh wallet detection
-- [ ] Size anomaly detection
-- [ ] Basic alerting (Discord/Telegram)
-
-### Phase 2: Advanced Intelligence
-- [ ] Funding chain analysis
-- [ ] Sniper cluster detection (DBSCAN)
-- [ ] Market categorization (niche vs mainstream)
-- [ ] Historical backtesting framework
-
-### Phase 3: Production Hardening
-- [ ] High-availability deployment
-- [ ] Rate limit management
-- [ ] False positive feedback loop
-- [ ] Web dashboard
-
----
-
-## Why This Matters
-
-Prediction markets are becoming a critical source of real-time probability estimates for world events. As they grow, so does the incentive for informed actors to exploit information asymmetry.
-
-This tool democratizes access to the same detection capabilities that sophisticated traders use. Whether you are:
-
-- **A trader** looking for alpha signals
-- **A researcher** studying market microstructure
-- **A platform operator** monitoring for manipulation
-
-...this tracker provides visibility into the hidden flows that move markets.
-
----
-
-## Technical Background
-
-### Polymarket Architecture
-
-Polymarket is a prediction market platform built on Polygon (Ethereum L2). Key characteristics:
-
-- **CLOB (Central Limit Order Book)**: Centralized matching engine for speed
-- **On-chain Settlement**: Final trades settle on Polygon blockchain
-- **USDC Collateral**: All positions denominated in USDC stablecoin
-- **Binary Outcomes**: Shares priced between $0.00 and $1.00
-
-### Data Sources
-
-| Source | Purpose | Latency |
-|--------|---------|---------|
-| Polymarket CLOB API | Real-time trades, orderbook | Milliseconds |
-| Polygon RPC | Wallet history, nonce, funding | 1-2 seconds |
-| Market Metadata API | Market categorization | On-demand |
-
-### Detection Challenges
-
-1. **Sybil Resistance**: Insiders use fresh wallets per trade
-2. **Rate Limits**: Polygon RPC calls require caching strategy
-3. **Market Classification**: NLP needed to categorize market niches
-4. **Timing**: CLOB data leads on-chain by seconds
-
----
-
-## Contributing
-
-Contributions are welcome! Please read our Contributing Guide before submitting PRs.
-
-### Development Setup
-
-```bash
-# Install dev dependencies
-pip install -e ".[dev]"
-
-# Run tests
-pytest
-
-# Run linting
-ruff check src/
-
-# Run type checking
-mypy src/
-```
+**Rate limiting on Polygon RPC**
+The default public RPC (`https://polygon-rpc.com`) has low rate limits. For production use, set `POLYGON_RPC_URL` to a dedicated provider (Alchemy, QuickNode, etc.).
 
 ---
 
@@ -312,20 +190,6 @@ This software is provided for **educational and research purposes only**.
 - Insider trading is illegal in regulated markets; this tool is for transparency and research
 - Users are responsible for compliance with applicable laws
 
----
-
 ## License
 
 MIT License - see [LICENSE](LICENSE) for details.
-
----
-
-## Acknowledgments
-
-- Inspired by [@DidiTrading](https://x.com/DidiTrading) and [@spacexbt](https://x.com/spacexbt)
-- Built on the open Polymarket API ecosystem
-- Community contributions welcome
-
----
-
-**Questions?** Open an issue or start a discussion.

alembic/env.py

@@ -19,7 +19,7 @@ if config.config_file_name is not None:
 target_metadata = Base.metadata
 
 # Get database URL from environment variable or config
-database_url = os.environ.get("SQLALCHEMY_DATABASE_URL")
+database_url = os.environ.get("DATABASE_URL")
 if database_url:
     config.set_main_option("sqlalchemy.url", database_url)
 

alembic/versions/20260104_0000_initial_schema.py

@@ -5,16 +5,16 @@ Revises:
 Create Date: 2026-01-04 00:00:00.000000+00:00
 """
 
-from typing import Sequence, Union
+from collections.abc import Sequence
 
 import sqlalchemy as sa
 from alembic import op
 
 # revision identifiers, used by Alembic.
 revision: str = "001_initial"
-down_revision: Union[str, None] = None
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
+down_revision: str | None = None
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
 
 
 def upgrade() -> None:

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

conftest.py

@@ -0,0 +1,14 @@
+"""Pytest configuration for the polymarket-insider-tracker tests."""
+
+import pytest
+
+# Configure pytest-asyncio
+pytest_plugins = ["pytest_asyncio"]
+
+
+@pytest.fixture(scope="session")
+def event_loop_policy():
+    """Use default event loop policy."""
+    import asyncio
+
+    return asyncio.DefaultEventLoopPolicy()

docker-compose.yml

@@ -1,5 +1,3 @@
-version: "3.8"
-
 services:
   postgres:
     image: postgres:15

docs/skill-tracking-prediction-market-flow.md

@@ -0,0 +1,113 @@
+# Skill: tracking-prediction-market-flow
+
+Use when analyzing prediction market activity for informed-flow signals, insider
+trading patterns, or suspicious wallet behavior on Polymarket.
+
+## What This Tool Does
+
+polymarket-insider-tracker streams real-time trades from Polymarket's WebSocket
+feed, profiles trader wallets on the Polygon blockchain, and scores each trade
+for informed-flow risk using multiple detection signals:
+
+- **Fresh wallet detection**: New wallets (age < 48h, nonce <= 5) making large
+  trades (> $1k). Insiders create disposable wallets per trade.
+- **Size anomaly detection**: Trades consuming > 2% of 24h volume or > 5% of
+  visible order book depth. Informed traders bet bigger when they have edge.
+- **Niche market scoring**: Low-volume markets (< $50k daily) get a 1.5x risk
+  multiplier. Easier to have inside information on obscure events.
+- **Funding chain analysis**: Traces wallet funding sources on-chain to link
+  seemingly separate wallets to the same entity or exchange.
+- **Sniper cluster detection**: DBSCAN clustering identifies wallets that
+  consistently enter markets within minutes of creation.
+
+Composite risk scoring combines signals with configurable weights (default
+alert threshold: 0.6). Multi-signal bonuses: 2 signals +20%, 3+ signals +30%.
+
+## Installation
+
+```bash
+git clone https://github.com/pselamy/polymarket-insider-tracker.git
+cd polymarket-insider-tracker
+uv sync --all-extras
+docker compose up -d   # PostgreSQL + Redis
+cp .env.example .env   # defaults work for local dev
+uv run alembic upgrade head
+```
+
+No API keys required for basic operation (Polymarket APIs are public).
+
+## Usage
+
+```bash
+# Start the tracker (streams trades, profiles wallets, scores risk, alerts)
+uv run python -m polymarket_insider_tracker
+
+# Dry run (no alerts sent)
+uv run python -m polymarket_insider_tracker --dry-run
+
+# Debug mode (see every trade)
+uv run python -m polymarket_insider_tracker --log-level DEBUG
+
+# Validate config without starting
+uv run python -m polymarket_insider_tracker --config-check
+```
+
+## Interpreting Signals
+
+### Risk Assessment Output
+
+Each flagged trade produces a risk assessment with:
+
+- **Confidence score** (0.0-1.0): Composite of weighted signals
+- **Signal breakdown**: Which detectors fired and their individual confidence
+- **Wallet profile**: Age, nonce, transaction count, funding source
+- **Market context**: Volume, category, order book depth
+
+### Signal Interpretation Guide
+
+| Score Range | Interpretation | Action |
+|-------------|---------------|--------|
+| 0.6-0.7 | Moderate: single strong signal or two weak ones | Monitor, note the market |
+| 0.7-0.85 | High: multiple signals converging | Investigate the market and wallet |
+| 0.85-1.0 | Critical: fresh wallet + large size + niche market | High-confidence informed flow |
+
+### What This Is NOT
+
+- Not a trading signal generator. Informed flow != actionable alpha without
+  further analysis (hypothesis -> leakage-aware backtest -> capital).
+- Not real-time enough for front-running. The tool detects patterns for
+  research and monitoring, not millisecond-level execution.
+- Detection of informed flow does not prove insider trading. Many legitimate
+  reasons exist for the patterns this tool flags.
+
+## Rate Limits and Etiquette
+
+- **Polymarket WebSocket**: No explicit rate limit; one persistent connection.
+  Do not open multiple connections unnecessarily.
+- **Polymarket CLOB REST**: Built-in rate limiter at 10 req/s with retry
+  backoff on 429/5xx. Respect this for metadata/orderbook queries.
+- **Polygon RPC**: Public endpoints (polygon-rpc.com) have low limits. For
+  sustained use, configure a dedicated RPC provider via `POLYGON_RPC_URL`.
+  Built-in token-bucket rate limiter at 25 req/s with Redis caching (5min TTL).
+
+## Known Pitfalls
+
+1. **WebSocket subscription format**: Must include `action: "subscribe"` in the
+   envelope. Without it, the server accepts the connection but delivers zero
+   trade events (silent failure). Fixed in the current version.
+
+2. **Message routing**: Live-data WebSocket pushes `{connection_id, payload:
+   {...trade fields}}`, not `{topic, type, payload}`. Route by checking for
+   `transactionHash` + `proxyWallet` keys in `payload`.
+
+3. **Public RPC rate limits**: Default Polygon RPC will throttle under load.
+   Use a dedicated provider for production.
+
+4. **Database required**: PostgreSQL + Redis must be running. Use
+   `docker compose up -d` for local dev.
+
+## Cross-References
+
+- **Repository**: https://github.com/pselamy/polymarket-insider-tracker
+- **Issues**: https://github.com/pselamy/polymarket-insider-tracker/issues
+- **Agent skill landing** (follow-on): selamy-labs/agent-skills

pyproject.toml

@@ -12,6 +12,7 @@ dependencies = [
     "sqlalchemy>=2.0.0",
     "alembic>=1.13.0",
     "pydantic>=2.0.0",
+    "pydantic-settings>=2.0.0",
     "python-dotenv>=1.0.0",
     "websockets>=12.0",
     "prometheus-client>=0.19.0",
@@ -28,6 +29,7 @@ dev = [
     "ruff>=0.1.0",
     "mypy>=1.7.0",
     "pre-commit>=3.0.0",
+    "aiosqlite>=0.19.0",
 ]
 
 [build-system]
@@ -76,7 +78,7 @@ warn_unused_configs = true
 plugins = ["pydantic.mypy"]
 
 [[tool.mypy.overrides]]
-module = ["py_clob_client.*", "web3.*", "redis.*"]
+module = ["py_clob_client.*", "web3.*", "redis.*", "sklearn.*", "prometheus_client.*"]
 ignore_missing_imports = true
 
 [tool.pytest.ini_options]

src/polymarket_insider_tracker/__main__.py

@@ -0,0 +1,302 @@
+"""CLI entry point for Polymarket Insider Tracker.
+
+This module provides the main entry point for running the tracker
+from the command line.
+
+Usage:
+    python -m polymarket_insider_tracker [options]
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import logging
+import logging.config
+import sys
+from typing import NoReturn
+
+from pydantic import ValidationError
+
+from polymarket_insider_tracker import __version__
+from polymarket_insider_tracker.config import Settings, clear_settings_cache, get_settings
+from polymarket_insider_tracker.pipeline import Pipeline
+from polymarket_insider_tracker.shutdown import GracefulShutdown
+
+# Application info
+APP_NAME = "Polymarket Insider Tracker"
+APP_VERSION = __version__
+
+# Exit codes
+EXIT_SUCCESS = 0
+EXIT_ERROR = 1
+EXIT_CONFIG_ERROR = 2
+EXIT_INTERRUPTED = 130
+
+
+def create_parser() -> argparse.ArgumentParser:
+    """Create the argument parser for the CLI.
+
+    Returns:
+        Configured ArgumentParser instance.
+    """
+    parser = argparse.ArgumentParser(
+        prog="polymarket-insider-tracker",
+        description="Detect insider trading activity on Polymarket prediction markets.",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""
+Examples:
+  python -m polymarket_insider_tracker           Run full pipeline
+  python -m polymarket_insider_tracker --config-check  Validate config and exit
+  python -m polymarket_insider_tracker --dry-run       Run without sending alerts
+  python -m polymarket_insider_tracker --log-level DEBUG  Enable debug logging
+        """,
+    )
+
+    parser.add_argument(
+        "--version",
+        action="version",
+        version=f"%(prog)s {APP_VERSION}",
+    )
+
+    parser.add_argument(
+        "--config-check",
+        action="store_true",
+        help="Validate configuration and exit without running pipeline",
+    )
+
+    parser.add_argument(
+        "--log-level",
+        choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
+        default=None,
+        help="Override logging level (default: from settings)",
+    )
+
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Run pipeline but don't send alerts",
+    )
+
+    parser.add_argument(
+        "--health-port",
+        type=int,
+        default=None,
+        help="Override health check port (default: from settings)",
+    )
+
+    return parser
+
+
+def configure_logging(level: str) -> None:
+    """Configure structured logging for the application.
+
+    Args:
+        level: Logging level string (DEBUG, INFO, etc.)
+    """
+    config = {
+        "version": 1,
+        "disable_existing_loggers": False,
+        "formatters": {
+            "standard": {
+                "format": "%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+                "datefmt": "%Y-%m-%d %H:%M:%S",
+            },
+            "detailed": {
+                "format": (
+                    "%(asctime)s [%(levelname)s] %(name)s (%(filename)s:%(lineno)d): %(message)s"
+                ),
+                "datefmt": "%Y-%m-%d %H:%M:%S",
+            },
+        },
+        "handlers": {
+            "console": {
+                "class": "logging.StreamHandler",
+                "level": level,
+                "formatter": "detailed" if level == "DEBUG" else "standard",
+                "stream": "ext://sys.stdout",
+            },
+        },
+        "root": {
+            "level": level,
+            "handlers": ["console"],
+        },
+        # Quieter logging for noisy libraries
+        "loggers": {
+            "httpx": {"level": "WARNING"},
+            "httpcore": {"level": "WARNING"},
+            "websockets": {"level": "WARNING"},
+            "web3": {"level": "WARNING"},
+            "urllib3": {"level": "WARNING"},
+        },
+    }
+    logging.config.dictConfig(config)
+
+
+def print_banner() -> None:
+    """Print the application startup banner."""
+    banner = f"""
+╔══════════════════════════════════════════════════════════════╗
+║                                                              ║
+║   {APP_NAME:^56}   ║
+║   {"v" + APP_VERSION:^56}   ║
+║                                                              ║
+╚══════════════════════════════════════════════════════════════╝
+"""
+    print(banner)
+
+
+def print_config_summary(settings: Settings, dry_run: bool) -> None:
+    """Print a summary of the configuration.
+
+    Args:
+        settings: Application settings.
+        dry_run: Whether dry-run mode is enabled.
+    """
+    summary = settings.redacted_summary()
+    print("Configuration:")
+    print(f"  Database: {summary['database_url']}")
+    print(f"  Redis: {summary['redis_url']}")
+    print(f"  Log Level: {summary['log_level']}")
+    print(f"  Health Port: {summary['health_port']}")
+    print(f"  Dry Run: {dry_run}")
+    print(f"  Discord: {'enabled' if summary['discord_enabled'] == 'True' else 'disabled'}")
+    print(f"  Telegram: {'enabled' if summary['telegram_enabled'] == 'True' else 'disabled'}")
+    print()
+
+
+def validate_config() -> Settings | None:
+    """Validate and load configuration.
+
+    Returns:
+        Settings instance if valid, None if invalid.
+    """
+    try:
+        # Clear cache to force reload
+        clear_settings_cache()
+        return get_settings()
+    except ValidationError as e:
+        print("Configuration validation failed:", file=sys.stderr)
+        for error in e.errors():
+            field = ".".join(str(loc) for loc in error["loc"])
+            msg = error["msg"]
+            print(f"  {field}: {msg}", file=sys.stderr)
+        return None
+
+
+def run_config_check(settings: Settings) -> int:
+    """Run configuration check and exit.
+
+    Args:
+        settings: Validated settings.
+
+    Returns:
+        Exit code (0 for success).
+    """
+    print("Configuration is valid!")
+    print()
+    print_config_summary(settings, dry_run=False)
+
+    # Test component availability
+    print("Checking component availability...")
+
+    # Check Discord
+    if settings.discord.enabled:
+        print("  Discord: configured")
+    else:
+        print("  Discord: not configured")
+
+    # Check Telegram
+    if settings.telegram.enabled:
+        print("  Telegram: configured")
+    else:
+        print("  Telegram: not configured")
+
+    print()
+    print("All checks passed. Ready to run.")
+    return EXIT_SUCCESS
+
+
+async def run_pipeline(
+    settings: Settings,
+    dry_run: bool,
+    shutdown_timeout: float = 30.0,
+) -> int:
+    """Run the main pipeline with graceful shutdown handling.
+
+    Args:
+        settings: Application settings.
+        dry_run: Whether to skip sending alerts.
+        shutdown_timeout: Maximum time to wait for graceful shutdown.
+
+    Returns:
+        Exit code.
+    """
+    logger = logging.getLogger(__name__)
+    shutdown = GracefulShutdown(timeout=shutdown_timeout)
+
+    try:
+        async with shutdown:
+            pipeline = Pipeline(settings, dry_run=dry_run)
+
+            # Register pipeline cleanup
+            shutdown.register_cleanup(pipeline.stop)
+
+            logger.info("Starting pipeline...")
+            await pipeline.start()
+
+            logger.info("Pipeline running. Press Ctrl+C to stop.")
+
+            # Wait for shutdown signal
+            await shutdown.wait()
+
+            logger.info("Shutdown signal received, stopping pipeline...")
+            await pipeline.stop()
+
+        return EXIT_SUCCESS
+    except KeyboardInterrupt:
+        logger.info("Interrupted by user")
+        return EXIT_INTERRUPTED
+    except Exception as e:
+        logger.exception("Pipeline failed: %s", e)
+        return EXIT_ERROR
+
+
+def main(argv: list[str] | None = None) -> NoReturn:
+    """Main entry point for the CLI.
+
+    Args:
+        argv: Command line arguments (defaults to sys.argv[1:]).
+    """
+    parser = create_parser()
+    args = parser.parse_args(argv)
+
+    # Validate configuration first
+    settings = validate_config()
+    if settings is None:
+        sys.exit(EXIT_CONFIG_ERROR)
+
+    # Determine effective log level
+    log_level = args.log_level or settings.log_level
+    configure_logging(log_level)
+
+    # Print banner
+    print_banner()
+
+    # Config check mode
+    if args.config_check:
+        sys.exit(run_config_check(settings))
+
+    # Determine dry-run mode
+    dry_run = args.dry_run or settings.dry_run
+
+    # Print config summary
+    print_config_summary(settings, dry_run)
+
+    # Run pipeline
+    exit_code = asyncio.run(run_pipeline(settings, dry_run))
+    sys.exit(exit_code)
+
+
+if __name__ == "__main__":
+    main()

src/polymarket_insider_tracker/alerter/channels/discord.py

@@ -100,9 +100,7 @@ class DiscordChannel:
                         await asyncio.sleep(retry_after)
                         continue
 
-                    logger.error(
-                        f"Discord webhook failed: {response.status_code} {response.text}"
-                    )
+                    logger.error(f"Discord webhook failed: {response.status_code} {response.text}")
 
             except httpx.TimeoutException:
                 logger.warning(f"Discord webhook timeout (attempt {attempt + 1})")

src/polymarket_insider_tracker/alerter/dispatcher.py

@@ -101,8 +101,7 @@ class AlertDispatcher:
             ):
                 # Allow half-open attempt
                 logger.info(
-                    f"Circuit half-open for {channel_name}, "
-                    f"attempt {state.half_open_attempts + 1}"
+                    f"Circuit half-open for {channel_name}, attempt {state.half_open_attempts + 1}"
                 )
                 return True
 
@@ -130,8 +129,7 @@ class AlertDispatcher:
             # Open the circuit
             state.is_open = True
             logger.warning(
-                f"Circuit opened for {channel_name} after "
-                f"{state.failure_count} failures"
+                f"Circuit opened for {channel_name} after {state.failure_count} failures"
             )
 
     async def _send_to_channel(
@@ -184,15 +182,11 @@ class AlertDispatcher:
             channel_results=channel_results,
         )
 
-        logger.info(
-            f"Dispatch complete: {success_count}/{len(channel_results)} succeeded"
-        )
+        logger.info(f"Dispatch complete: {success_count}/{len(channel_results)} succeeded")
 
         return result
 
-    async def dispatch_batch(
-        self, alerts: list[FormattedAlert]
-    ) -> list[DispatchResult]:
+    async def dispatch_batch(self, alerts: list[FormattedAlert]) -> list[DispatchResult]:
         """Dispatch multiple alerts sequentially.
 
         Args:
@@ -215,9 +209,7 @@ class AlertDispatcher:
                 "failure_count": state.failure_count,
                 "half_open_attempts": state.half_open_attempts,
                 "last_failure": (
-                    state.last_failure_time.isoformat()
-                    if state.last_failure_time
-                    else None
+                    state.last_failure_time.isoformat() if state.last_failure_time else None
                 ),
             }
             for name, state in self._circuit_state.items()

src/polymarket_insider_tracker/alerter/formatter.py

@@ -30,7 +30,7 @@ def truncate_address(address: str, chars: int = 4) -> str:
     """Truncate an Ethereum address to 0x1234...5678 format."""
     if len(address) < chars * 2 + 4:
         return address
-    return f"{address[:chars+2]}...{address[-chars:]}"
+    return f"{address[: chars + 2]}...{address[-chars:]}"
 
 
 def format_usdc(amount: Decimal) -> str:
@@ -117,9 +117,7 @@ class AlertFormatter:
         telegram_md = self._build_telegram_markdown(
             assessment, wallet_short, risk_level, signals, links
         )
-        plain_text = self._build_plain_text(
-            assessment, wallet_short, risk_level, signals, links
-        )
+        plain_text = self._build_plain_text(assessment, wallet_short, risk_level, signals, links)
 
         return FormattedAlert(
             title=title,
@@ -192,10 +190,11 @@ class AlertFormatter:
         wallet_age_str = ""
         if assessment.fresh_wallet_signal:
             age_hours = assessment.fresh_wallet_signal.wallet_profile.age_hours
-            if age_hours < 1:
-                wallet_age_str = f" (Age: {int(age_hours * 60)}m)"
-            else:
-                wallet_age_str = f" (Age: {age_hours:.0f}h)"
+            if age_hours is not None:
+                if age_hours < 1:
+                    wallet_age_str = f" (Age: {int(age_hours * 60)}m)"
+                else:
+                    wallet_age_str = f" (Age: {age_hours:.0f}h)"
 
         fields: list[dict[str, object]] = [
             {
@@ -226,11 +225,13 @@ class AlertFormatter:
 
         # Signals (if any)
         if signals:
-            fields.append({
-                "name": "Signals",
-                "value": ", ".join(signals),
-                "inline": False,
-            })
+            fields.append(
+                {
+                    "name": "Signals",
+                    "value": ", ".join(signals),
+                    "inline": False,
+                }
+            )
 
         # Add detailed info for detailed verbosity
         if self.verbosity == "detailed":
@@ -244,11 +245,13 @@ class AlertFormatter:
                 confidences.append(f"Size Anomaly: {conf:.0%}")
 
             if confidences:
-                fields.append({
-                    "name": "Confidence",
-                    "value": " | ".join(confidences),
-                    "inline": False,
-                })
+                fields.append(
+                    {
+                        "name": "Confidence",
+                        "value": " | ".join(confidences),
+                        "inline": False,
+                    }
+                )
 
         embed: dict[str, object] = {
             "title": "🚨 Suspicious Activity Detected",
@@ -280,18 +283,21 @@ class AlertFormatter:
         wallet_line = f"*Wallet:* `{wallet_short}`"
         if assessment.fresh_wallet_signal:
             age_hours = assessment.fresh_wallet_signal.wallet_profile.age_hours
-            if age_hours < 1:
-                wallet_line += f" \\(Age: {int(age_hours * 60)}m\\)"
-            else:
-                wallet_line += f" \\(Age: {age_hours:.0f}h\\)"
+            if age_hours is not None:
+                if age_hours < 1:
+                    wallet_line += f" \\(Age: {int(age_hours * 60)}m\\)"
+                else:
+                    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']})")
@@ -299,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("")
@@ -319,7 +327,26 @@ class AlertFormatter:
 
     def _escape_telegram_markdown(self, text: str) -> str:
         """Escape special Telegram MarkdownV2 characters."""
-        special_chars = ["_", "*", "[", "]", "(", ")", "~", "`", ">", "#", "+", "-", "=", "|", "{", "}", ".", "!"]
+        special_chars = [
+            "_",
+            "*",
+            "[",
+            "]",
+            "(",
+            ")",
+            "~",
+            "`",
+            ">",
+            "#",
+            "+",
+            "-",
+            "=",
+            "|",
+            "{",
+            "}",
+            ".",
+            "!",
+        ]
         for char in special_chars:
             text = text.replace(char, f"\\{char}")
         return text
@@ -345,10 +372,11 @@ class AlertFormatter:
         wallet_line = f"Wallet: {wallet_short}"
         if assessment.fresh_wallet_signal:
             age_hours = assessment.fresh_wallet_signal.wallet_profile.age_hours
-            if age_hours < 1:
-                wallet_line += f" (Age: {int(age_hours * 60)}m)"
-            else:
-                wallet_line += f" (Age: {age_hours:.0f}h)"
+            if age_hours is not None:
+                if age_hours < 1:
+                    wallet_line += f" (Age: {int(age_hours * 60)}m)"
+                else:
+                    wallet_line += f" (Age: {age_hours:.0f}h)"
         lines.append(wallet_line)
 
         # Risk

src/polymarket_insider_tracker/alerter/history.py

@@ -355,16 +355,14 @@ class AlertHistory:
         end = datetime.now(UTC)
         start = end - timedelta(hours=hours)
 
-        index_key = (
-            f"{self.KEY_INDEX_WALLET}{wallet}" if wallet else self.KEY_INDEX_TIME
-        )
+        index_key = f"{self.KEY_INDEX_WALLET}{wallet}" if wallet else self.KEY_INDEX_TIME
 
         count = await self.redis.zcount(
             index_key,
             start.timestamp(),
             end.timestamp(),
         )
-        return count
+        return int(count)
 
     async def cleanup_old_alerts(self) -> int:
         """Remove alerts older than retention period.
@@ -395,4 +393,4 @@ class AlertHistory:
         # Note: Individual alert records will expire via TTL
         # Wallet/market indexes will also expire via TTL
         logger.info(f"Cleaned up {removed} old alert references")
-        return removed
+        return int(removed)

src/polymarket_insider_tracker/config.py

@@ -0,0 +1,308 @@
+"""Configuration management service with Pydantic Settings.
+
+This module provides centralized configuration management for the
+Polymarket Insider Tracker application, loading and validating
+environment variables at startup.
+"""
+
+from __future__ import annotations
+
+import logging
+from functools import lru_cache
+from typing import Literal
+
+from pydantic import Field, SecretStr, field_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class DatabaseSettings(BaseSettings):
+    """Database connection settings."""
+
+    model_config = SettingsConfigDict(
+        env_prefix="", env_file=".env", env_file_encoding="utf-8", extra="ignore"
+    )
+
+    url: str = Field(
+        alias="DATABASE_URL",
+        description="PostgreSQL connection string",
+    )
+
+    @field_validator("url")
+    @classmethod
+    def validate_url(cls, v: str) -> str:
+        """Validate database URL format."""
+        if not v.startswith(("postgresql://", "postgresql+asyncpg://")):
+            raise ValueError("DATABASE_URL must be a PostgreSQL connection string")
+        return v
+
+
+class RedisSettings(BaseSettings):
+    """Redis connection settings."""
+
+    model_config = SettingsConfigDict(
+        env_prefix="", env_file=".env", env_file_encoding="utf-8", extra="ignore"
+    )
+
+    url: str = Field(
+        default="redis://localhost:6379",
+        alias="REDIS_URL",
+        description="Redis connection string",
+    )
+
+    @field_validator("url")
+    @classmethod
+    def validate_url(cls, v: str) -> str:
+        """Validate Redis URL format."""
+        if not v.startswith("redis://"):
+            raise ValueError("REDIS_URL must start with redis://")
+        return v
+
+
+class PolygonSettings(BaseSettings):
+    """Polygon blockchain RPC settings."""
+
+    model_config = SettingsConfigDict(
+        env_prefix="POLYGON_", env_file=".env", env_file_encoding="utf-8", extra="ignore"
+    )
+
+    rpc_url: str = Field(
+        default="https://polygon-rpc.com",
+        alias="POLYGON_RPC_URL",
+        description="Primary Polygon RPC endpoint",
+    )
+    fallback_rpc_url: str | None = Field(
+        default=None,
+        alias="POLYGON_FALLBACK_RPC_URL",
+        description="Fallback Polygon RPC endpoint",
+    )
+
+    @field_validator("rpc_url", "fallback_rpc_url")
+    @classmethod
+    def validate_url(cls, v: str | None) -> str | None:
+        """Validate RPC URL format."""
+        if v is None:
+            return v
+        if not v.startswith(("http://", "https://")):
+            raise ValueError("RPC URL must be an HTTP(S) endpoint")
+        return v
+
+
+class PolymarketSettings(BaseSettings):
+    """Polymarket API settings."""
+
+    model_config = SettingsConfigDict(
+        env_prefix="POLYMARKET_", env_file=".env", env_file_encoding="utf-8", extra="ignore"
+    )
+
+    ws_url: str = Field(
+        default="wss://ws-subscriptions-clob.polymarket.com/ws/market",
+        alias="POLYMARKET_WS_URL",
+        description="Polymarket WebSocket URL for live data",
+    )
+    api_key: SecretStr | None = Field(
+        default=None,
+        alias="POLYMARKET_API_KEY",
+        description="Optional Polymarket API key",
+    )
+
+    @field_validator("ws_url")
+    @classmethod
+    def validate_ws_url(cls, v: str) -> str:
+        """Validate WebSocket URL format."""
+        if not v.startswith(("ws://", "wss://")):
+            raise ValueError("WebSocket URL must start with ws:// or wss://")
+        return v
+
+
+class DiscordSettings(BaseSettings):
+    """Discord notification settings."""
+
+    model_config = SettingsConfigDict(
+        env_prefix="DISCORD_", env_file=".env", env_file_encoding="utf-8", extra="ignore"
+    )
+
+    webhook_url: SecretStr | None = Field(
+        default=None,
+        alias="DISCORD_WEBHOOK_URL",
+        description="Discord webhook URL for alerts",
+    )
+
+    @property
+    def enabled(self) -> bool:
+        """Check if Discord notifications are enabled."""
+        return self.webhook_url is not None and bool(self.webhook_url.get_secret_value().strip())
+
+
+class TelegramSettings(BaseSettings):
+    """Telegram notification settings."""
+
+    model_config = SettingsConfigDict(
+        env_prefix="TELEGRAM_", env_file=".env", env_file_encoding="utf-8", extra="ignore"
+    )
+
+    bot_token: SecretStr | None = Field(
+        default=None,
+        alias="TELEGRAM_BOT_TOKEN",
+        description="Telegram bot token",
+    )
+    chat_id: str | None = Field(
+        default=None,
+        alias="TELEGRAM_CHAT_ID",
+        description="Telegram chat ID for alerts",
+    )
+
+    @property
+    def enabled(self) -> bool:
+        """Check if Telegram notifications are enabled."""
+        return (
+            self.bot_token is not None
+            and bool(self.bot_token.get_secret_value().strip())
+            and self.chat_id is not None
+            and bool(self.chat_id.strip())
+        )
+
+
+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.
+
+    Loads configuration from environment variables with support for
+    .env files via python-dotenv.
+
+    Example:
+        ```python
+        from polymarket_insider_tracker.config import get_settings
+
+        settings = get_settings()
+        print(settings.database.url)
+        print(settings.log_level)
+        ```
+    """
+
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="ignore",
+    )
+
+    # Nested configuration groups
+    database: DatabaseSettings = Field(default_factory=DatabaseSettings)
+    redis: RedisSettings = Field(default_factory=RedisSettings)
+    polygon: PolygonSettings = Field(default_factory=PolygonSettings)
+    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(
+        default="INFO",
+        alias="LOG_LEVEL",
+        description="Logging level",
+    )
+    health_port: int = Field(
+        default=8080,
+        alias="HEALTH_PORT",
+        description="HTTP port for health check endpoints",
+        ge=1,
+        le=65535,
+    )
+    dry_run: bool = Field(
+        default=False,
+        alias="DRY_RUN",
+        description="Run without sending actual alerts",
+    )
+
+    def get_logging_level(self) -> int:
+        """Get the numeric logging level."""
+        level: int = getattr(logging, self.log_level)
+        return level
+
+    def redacted_summary(self) -> dict[str, str | dict[str, str]]:
+        """Get a summary of settings with secrets redacted.
+
+        Returns:
+            Dictionary of settings with sensitive values masked.
+        """
+        return {
+            "database_url": self._redact_url(self.database.url),
+            "redis_url": self._redact_url(self.redis.url),
+            "polygon": {
+                "rpc_url": self.polygon.rpc_url,
+                "fallback_rpc_url": self.polygon.fallback_rpc_url or "(not set)",
+            },
+            "polymarket": {
+                "ws_url": self.polymarket.ws_url,
+                "api_key": "(set)" if self.polymarket.api_key else "(not set)",
+            },
+            "discord_enabled": str(self.discord.enabled),
+            "telegram_enabled": str(self.telegram.enabled),
+            "log_level": self.log_level,
+            "health_port": str(self.health_port),
+            "dry_run": str(self.dry_run),
+        }
+
+    @staticmethod
+    def _redact_url(url: str) -> str:
+        """Redact password from URL if present."""
+        if "@" in url and "://" in url:
+            # URL has credentials - redact the password
+            protocol_end = url.index("://") + 3
+            at_pos = url.index("@")
+            creds_part = url[protocol_end:at_pos]
+            if ":" in creds_part:
+                username = creds_part.split(":")[0]
+                return f"{url[:protocol_end]}{username}:***@{url[at_pos + 1 :]}"
+        return url
+
+
+@lru_cache(maxsize=1)
+def get_settings() -> Settings:
+    """Get the application settings singleton.
+
+    Uses LRU cache to ensure settings are loaded only once and
+    reused across the application.
+
+    Returns:
+        The Settings instance.
+
+    Raises:
+        ValidationError: If required environment variables are missing
+            or have invalid values.
+    """
+    return Settings()
+
+
+def clear_settings_cache() -> None:
+    """Clear the settings cache.
+
+    Useful for testing when you need to reload settings with
+    different environment variables.
+    """
+    get_settings.cache_clear()

src/polymarket_insider_tracker/detector/models.py

@@ -265,14 +265,10 @@ class RiskAssessment:
             "has_fresh_wallet_signal": self.fresh_wallet_signal is not None,
             "has_size_anomaly_signal": self.size_anomaly_signal is not None,
             "fresh_wallet_confidence": (
-                self.fresh_wallet_signal.confidence
-                if self.fresh_wallet_signal
-                else None
+                self.fresh_wallet_signal.confidence if self.fresh_wallet_signal else None
             ),
             "size_anomaly_confidence": (
-                self.size_anomaly_signal.confidence
-                if self.size_anomaly_signal
-                else None
+                self.size_anomaly_signal.confidence if self.size_anomaly_signal else None
             ),
             "timestamp": self.timestamp.isoformat(),
         }

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:"
 
@@ -155,8 +159,7 @@ class RiskScorer:
         # Log assessment
         if should_alert:
             logger.info(
-                "Risk assessment triggered alert: wallet=%s, market=%s, "
-                "score=%.2f, signals=%d",
+                "Risk assessment triggered alert: wallet=%s, market=%s, score=%.2f, signals=%d",
                 bundle.wallet_address[:10] + "...",
                 bundle.market_id[:10] + "...",
                 weighted_score,
@@ -180,9 +183,7 @@ class RiskScorer:
             should_alert=should_alert,
         )
 
-    def calculate_weighted_score(
-        self, bundle: SignalBundle
-    ) -> tuple[float, int]:
+    def calculate_weighted_score(self, bundle: SignalBundle) -> tuple[float, int]:
         """Calculate weighted score from all signals.
 
         Applies per-signal weights and multi-signal bonuses.
@@ -271,11 +272,9 @@ class RiskScorer:
         """
         key = f"{self._key_prefix}{wallet_address}:{market_id}"
         deleted = await self._redis.delete(key)
-        return deleted > 0
+        return int(deleted) > 0
 
-    async def assess_batch(
-        self, bundles: list[SignalBundle]
-    ) -> list[RiskAssessment]:
+    async def assess_batch(self, bundles: list[SignalBundle]) -> list[RiskAssessment]:
         """Assess multiple trade bundles.
 
         Args:

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/detector/sniper.py

@@ -228,10 +228,17 @@ class SniperDetector:
             for entry in entries:
                 # Normalize market ID to 0-1 range
                 market_hash = (
-                    int(hashlib.md5(  # noqa: S324
-                        entry.market_id.encode()
-                    ).hexdigest()[:8], 16) % 1000
-                ) / 1000.0
+                    (
+                        int(
+                            hashlib.md5(  # noqa: S324
+                                entry.market_id.encode()
+                            ).hexdigest()[:8],
+                            16,
+                        )
+                        % 1000
+                    )
+                    / 1000.0
+                )
 
                 # Normalize entry delta to hours (0-5 mins = 0-0.083 hours)
                 delta_hours = entry.entry_delta_seconds / 3600.0
@@ -285,7 +292,7 @@ class SniperDetector:
                 cluster_id=cluster_id,
                 wallet_addresses=cluster_wallets,
                 avg_entry_delta=cluster_stats["avg_delta"],
-                markets_in_common=cluster_stats["markets_in_common"],
+                markets_in_common=int(cluster_stats["markets_in_common"]),
             )
 
             # Update wallet-cluster mapping
@@ -305,7 +312,7 @@ class SniperDetector:
                         cluster_id=cluster_id,
                         cluster_size=len(cluster_wallets),
                         avg_entry_delta_seconds=cluster_stats["avg_delta"],
-                        markets_in_common=cluster_stats["markets_in_common"],
+                        markets_in_common=int(cluster_stats["markets_in_common"]),
                         confidence=confidence,
                     )
 
@@ -415,11 +422,7 @@ class SniperDetector:
         overlap_factor = min(1.0, markets_common / 5.0)
 
         # Weighted combination
-        confidence = (
-            0.3 * size_factor +
-            0.4 * speed_factor +
-            0.3 * overlap_factor
-        )
+        confidence = 0.3 * size_factor + 0.4 * speed_factor + 0.3 * overlap_factor
 
         return round(min(1.0, confidence), 3)
 

src/polymarket_insider_tracker/ingestor/__init__.py

@@ -35,10 +35,12 @@ from polymarket_insider_tracker.ingestor.publisher import (
 )
 from polymarket_insider_tracker.ingestor.websocket import (
     ConnectionState,
-    StreamStats as WebSocketStreamStats,
     TradeStreamError,
     TradeStreamHandler,
 )
+from polymarket_insider_tracker.ingestor.websocket import (
+    StreamStats as WebSocketStreamStats,
+)
 
 __all__ = [
     # CLOB Client

src/polymarket_insider_tracker/ingestor/clob_client.py

@@ -6,7 +6,7 @@ import os
 import time
 from collections.abc import Callable
 from functools import wraps
-from typing import Any, ParamSpec, TypeVar
+from typing import ParamSpec, TypeVar
 
 from py_clob_client.client import ClobClient as BaseClobClient
 from py_clob_client.clob_types import BookParams
@@ -287,7 +287,8 @@ class ClobClient:
 
         try:
             response = self._client.get_midpoint(token_id)
-            return response.get("mid")
+            mid = response.get("mid")
+            return str(mid) if mid is not None else None
         except Exception as e:
             logger.warning("Failed to get midpoint for %s: %s", token_id, e)
             return None
@@ -307,7 +308,8 @@ class ClobClient:
 
         try:
             response = self._client.get_price(token_id, side=side)
-            return response.get("price")
+            price = response.get("price")
+            return str(price) if price is not None else None
         except Exception as e:
             logger.warning("Failed to get %s price for %s: %s", side, token_id, e)
             return None
@@ -321,7 +323,7 @@ class ClobClient:
         try:
             self._rate_limiter.acquire_sync()
             result = self._client.get_ok()
-            return result == "OK"
+            return str(result) == "OK"
         except Exception as e:
             logger.error("Health check failed: %s", e)
             return False
@@ -334,7 +336,8 @@ class ClobClient:
         """
         try:
             self._rate_limiter.acquire_sync()
-            return self._client.get_server_time()
+            result = self._client.get_server_time()
+            return int(result) if result is not None else None
         except Exception as e:
             logger.error("Failed to get server time: %s", e)
             return None

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/health.py

@@ -335,8 +335,10 @@ class HealthMonitor:
 
         overall_status = self._determine_overall_status()
         HEALTH_STATUS.set(
-            1.0 if overall_status == HealthStatus.HEALTHY
-            else 0.5 if overall_status == HealthStatus.DEGRADED
+            1.0
+            if overall_status == HealthStatus.HEALTHY
+            else 0.5
+            if overall_status == HealthStatus.DEGRADED
             else 0.0
         )
 
@@ -362,10 +364,7 @@ class HealthMonitor:
                 report = self.get_health_report()
 
                 # Notify on status change
-                if (
-                    self._on_health_change
-                    and report.status != self._last_health_status
-                ):
+                if self._on_health_change and report.status != self._last_health_status:
                     self._last_health_status = report.status
                     try:
                         await self._on_health_change(report)

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 Enum
+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__)
@@ -27,7 +28,7 @@ DEFAULT_CACHE_TTL_SECONDS = 600  # 10 minutes
 DEFAULT_REDIS_KEY_PREFIX = "polymarket:market:"
 
 
-class SyncState(str, Enum):
+class SyncState(StrEnum):
     """State of the metadata synchronizer."""
 
     STOPPED = "stopped"
@@ -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)
-            markets = await asyncio.to_thread(self._clob.get_markets, True)
+            # Fetch CLOB markets and gamma volume snapshot in parallel
+            markets, gamma_stats = await asyncio.gather(
+                asyncio.to_thread(self._clob.get_markets, True),
+                self._fetch_gamma_stats(),
+            )
 
-            # 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
@@ -351,7 +384,7 @@ class MarketMetadataSync:
         """
         key = f"{self._key_prefix}{condition_id}"
         deleted = await self._redis.delete(key)
-        return deleted > 0
+        return int(deleted) > 0
 
     async def force_sync(self) -> None:
         """Force an immediate sync of all markets.

src/polymarket_insider_tracker/ingestor/models.py

@@ -1,5 +1,6 @@
 """Data models for the ingestor module."""
 
+import contextlib
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from decimal import Decimal
@@ -46,10 +47,8 @@ class Market:
         end_date = None
         end_date_iso = data.get("end_date_iso")
         if end_date_iso:
-            try:
+            with contextlib.suppress(ValueError, AttributeError):
                 end_date = datetime.fromisoformat(end_date_iso.replace("Z", "+00:00"))
-            except (ValueError, AttributeError):
-                pass
 
         return cls(
             condition_id=str(data["condition_id"]),
@@ -87,7 +86,7 @@ class Orderbook:
     bids: tuple[OrderbookLevel, ...]
     asks: tuple[OrderbookLevel, ...]
     tick_size: Decimal
-    timestamp: datetime = field(default_factory=datetime.utcnow)
+    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
 
     @classmethod
     def from_clob_orderbook(cls, orderbook: Any) -> "Orderbook":
@@ -400,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))
 
@@ -447,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(),
         }
 
@@ -466,10 +474,8 @@ class MarketMetadata:
         end_date = None
         end_date_str = data.get("end_date")
         if end_date_str:
-            try:
+            with contextlib.suppress(ValueError, AttributeError):
                 end_date = datetime.fromisoformat(end_date_str)
-            except (ValueError, AttributeError):
-                pass
 
         last_updated_str = data.get("last_updated")
         if last_updated_str:
@@ -480,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", "")),
@@ -489,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/ingestor/publisher.py

@@ -186,9 +186,10 @@ class EventPublisher:
             The entry ID assigned by Redis.
         """
         data = _serialize_trade_event(event)
+        # redis-py typing expects broader dict type than dict[str, str]
         entry_id = await self._redis.xadd(
             self._stream_name,
-            data,
+            data,  # type: ignore[arg-type]
             maxlen=self._max_len,
         )
         # entry_id may be bytes or str
@@ -213,7 +214,8 @@ class EventPublisher:
         pipe = self._redis.pipeline()
         for event in events:
             data = _serialize_trade_event(event)
-            pipe.xadd(self._stream_name, data, maxlen=self._max_len)
+            # redis-py typing expects broader dict type than dict[str, str]
+            pipe.xadd(self._stream_name, data, maxlen=self._max_len)  # type: ignore[arg-type]
 
         results = await pipe.execute()
 
@@ -384,7 +386,8 @@ class EventPublisher:
         """
         if not entry_ids:
             return 0
-        return await self._redis.xack(self._stream_name, group_name, *entry_ids)
+        result = await self._redis.xack(self._stream_name, group_name, *entry_ids)
+        return int(result)
 
     async def get_stream_info(self) -> dict[str, Any]:
         """Get information about the stream.
@@ -404,7 +407,8 @@ class EventPublisher:
         Returns:
             Number of entries in the stream.
         """
-        return await self._redis.xlen(self._stream_name)
+        result = await self._redis.xlen(self._stream_name)
+        return int(result)
 
     async def trim_stream(self, max_len: int | None = None) -> int:
         """Trim the stream to a maximum length.
@@ -416,4 +420,5 @@ class EventPublisher:
             Number of entries removed.
         """
         length = max_len or self._max_len
-        return await self._redis.xtrim(self._stream_name, maxlen=length)
+        result = await self._redis.xtrim(self._stream_name, maxlen=length)
+        return int(result)

src/polymarket_insider_tracker/ingestor/websocket.py

@@ -9,8 +9,9 @@ from dataclasses import dataclass
 from enum import Enum
 from typing import Any
 
-import websockets
 from websockets.asyncio.client import ClientConnection
+from websockets.asyncio.client import connect as ws_connect
+from websockets.exceptions import ConnectionClosed
 
 from polymarket_insider_tracker.ingestor.models import TradeEvent
 
@@ -149,14 +150,14 @@ class TradeStreamHandler:
         elif self._market_filter:
             subscription["filters"] = json.dumps({"market_slug": self._market_filter})
 
-        return {"subscriptions": [subscription]}
+        return {"action": "subscribe", "subscriptions": [subscription]}
 
     async def _connect(self) -> ClientConnection:
         """Establish WebSocket connection."""
         await self._set_state(ConnectionState.CONNECTING)
 
         try:
-            ws = await websockets.connect(
+            ws = await ws_connect(
                 self._host,
                 ping_interval=self._ping_interval,
                 ping_timeout=self._ping_interval * 2,
@@ -182,12 +183,13 @@ class TradeStreamHandler:
         try:
             data = json.loads(message)
 
-            # Check if this is a trade message
-            topic = data.get("topic")
-            msg_type = data.get("type")
-
-            if topic == "activity" and msg_type == "trades":
-                payload = data.get("payload", {})
+            # ws-live-data pushes {connection_id, payload:{...trade fields}}
+            payload = data.get("payload")
+            if (
+                isinstance(payload, dict)
+                and "transactionHash" in payload
+                and "proxyWallet" in payload
+            ):
                 trade = TradeEvent.from_websocket_message(payload)
 
                 self._stats.trades_received += 1
@@ -207,8 +209,7 @@ class TradeStreamHandler:
                     logger.error("Error in trade callback: %s", e)
 
             else:
-                # Log other message types for debugging
-                logger.debug("Received message: topic=%s type=%s", topic, msg_type)
+                logger.debug("Received non-trade message: %s", str(data)[:120])
 
         except json.JSONDecodeError as e:
             logger.warning("Invalid JSON message: %s", e)
@@ -227,7 +228,7 @@ class TradeStreamHandler:
                 else:
                     logger.debug("Received binary message (%d bytes)", len(message))
 
-        except websockets.ConnectionClosed as e:
+        except ConnectionClosed as e:
             logger.warning("Connection closed: %s", e)
             raise
         except Exception as e:
@@ -284,7 +285,7 @@ class TradeStreamHandler:
             while self._running:
                 try:
                     await self._listen(self._ws)
-                except (websockets.ConnectionClosed, Exception) as e:
+                except (ConnectionClosed, Exception) as e:
                     if not self._running:
                         break
 

src/polymarket_insider_tracker/pipeline.py

@@ -0,0 +1,621 @@
+"""Main pipeline orchestrator for Polymarket Insider Tracker.
+
+This module provides the Pipeline class that wires together all detection
+components and manages the event flow from ingestion to alerting.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from enum import StrEnum
+from typing import TYPE_CHECKING
+
+from redis.asyncio import Redis
+
+from polymarket_insider_tracker.alerter.channels.discord import DiscordChannel
+from polymarket_insider_tracker.alerter.channels.telegram import TelegramChannel
+from polymarket_insider_tracker.alerter.dispatcher import AlertChannel, AlertDispatcher
+from polymarket_insider_tracker.alerter.formatter import AlertFormatter
+from polymarket_insider_tracker.config import Settings, get_settings
+from polymarket_insider_tracker.detector.fresh_wallet import FreshWalletDetector
+from polymarket_insider_tracker.detector.scorer import RiskScorer, SignalBundle
+from polymarket_insider_tracker.detector.size_anomaly import SizeAnomalyDetector
+from polymarket_insider_tracker.ingestor.clob_client import ClobClient
+from polymarket_insider_tracker.ingestor.metadata_sync import MarketMetadataSync
+from polymarket_insider_tracker.ingestor.websocket import TradeStreamHandler
+from polymarket_insider_tracker.profiler.analyzer import WalletAnalyzer
+from polymarket_insider_tracker.profiler.chain import PolygonClient
+from polymarket_insider_tracker.profiler.funding import FundingTracer
+from polymarket_insider_tracker.storage.database import DatabaseManager
+from polymarket_insider_tracker.storage.repos import (
+    FundingRepository,
+    FundingTransferDTO,
+    RiskAssessmentDTO,
+    RiskAssessmentRepository,
+    WalletProfileDTO,
+    WalletRepository,
+)
+
+if TYPE_CHECKING:
+    from typing import Any
+
+    from polymarket_insider_tracker.detector.models import (
+        FreshWalletSignal,
+        RiskAssessment,
+        SizeAnomalySignal,
+    )
+    from polymarket_insider_tracker.ingestor.models import TradeEvent
+
+logger = logging.getLogger(__name__)
+
+
+class PipelineState(StrEnum):
+    """Pipeline lifecycle states."""
+
+    STOPPED = "stopped"
+    STARTING = "starting"
+    RUNNING = "running"
+    STOPPING = "stopping"
+    ERROR = "error"
+
+
+@dataclass
+class PipelineStats:
+    """Statistics for the pipeline."""
+
+    started_at: datetime | None = None
+    trades_processed: int = 0
+    signals_generated: int = 0
+    alerts_sent: int = 0
+    errors: int = 0
+    last_trade_time: datetime | None = None
+    last_error: str | None = None
+
+
+class Pipeline:
+    """Main pipeline orchestrator for the Polymarket Insider Tracker.
+
+    This class wires together all detection components and manages the
+    event flow from trade ingestion through profiling, detection, and alerting.
+
+    Pipeline flow:
+        WebSocket Trade Stream → Wallet Profiler → Detectors → Risk Scorer → Alerter
+
+    Example:
+        ```python
+        from polymarket_insider_tracker.config import get_settings
+        from polymarket_insider_tracker.pipeline import Pipeline
+
+        settings = get_settings()
+        pipeline = Pipeline(settings)
+
+        await pipeline.start()
+        # Pipeline runs until stop() is called
+        await pipeline.stop()
+        ```
+    """
+
+    def __init__(
+        self,
+        settings: Settings | None = None,
+        *,
+        dry_run: bool | None = None,
+    ) -> None:
+        """Initialize the pipeline.
+
+        Args:
+            settings: Application settings. If not provided, uses get_settings().
+            dry_run: If True, skip sending alerts. Overrides settings.dry_run.
+        """
+        self._settings = settings or get_settings()
+        self._dry_run = dry_run if dry_run is not None else self._settings.dry_run
+
+        self._state = PipelineState.STOPPED
+        self._stats = PipelineStats()
+
+        # Components (initialized in start())
+        self._redis: Redis | None = None
+        self._db_manager: DatabaseManager | None = None
+        self._polygon_client: PolygonClient | None = None
+        self._clob_client: ClobClient | None = None
+        self._metadata_sync: MarketMetadataSync | None = None
+        self._wallet_analyzer: WalletAnalyzer | None = None
+        self._fresh_wallet_detector: FreshWalletDetector | None = None
+        self._size_anomaly_detector: SizeAnomalyDetector | None = None
+        self._risk_scorer: RiskScorer | None = None
+        self._alert_formatter: AlertFormatter | None = None
+        self._alert_dispatcher: AlertDispatcher | None = None
+        self._trade_stream: TradeStreamHandler | None = None
+        self._funding_tracer: FundingTracer | None = None
+
+        # Synchronization
+        self._stop_event: asyncio.Event | None = None
+        self._stream_task: asyncio.Task[None] | None = None
+
+    @property
+    def state(self) -> PipelineState:
+        """Current pipeline state."""
+        return self._state
+
+    @property
+    def stats(self) -> PipelineStats:
+        """Current pipeline statistics."""
+        return self._stats
+
+    @property
+    def is_running(self) -> bool:
+        """Check if pipeline is running."""
+        return self._state == PipelineState.RUNNING
+
+    async def start(self) -> None:
+        """Start the pipeline.
+
+        Initializes all components and begins processing trades.
+
+        Raises:
+            RuntimeError: If pipeline is already running.
+            Exception: If any component fails to initialize.
+        """
+        if self._state != PipelineState.STOPPED:
+            raise RuntimeError(f"Cannot start pipeline in state {self._state}")
+
+        self._state = PipelineState.STARTING
+        self._stop_event = asyncio.Event()
+        logger.info("Starting pipeline...")
+
+        try:
+            await self._initialize_components()
+            await self._start_background_services()
+            self._stats.started_at = datetime.now(UTC)
+            self._state = PipelineState.RUNNING
+            logger.info("Pipeline started successfully")
+        except Exception as e:
+            self._state = PipelineState.ERROR
+            self._stats.last_error = str(e)
+            logger.error("Failed to start pipeline: %s", e)
+            await self._cleanup()
+            raise
+
+    async def stop(self) -> None:
+        """Stop the pipeline gracefully.
+
+        Stops all background services and cleans up resources.
+        """
+        if self._state == PipelineState.STOPPED:
+            return
+
+        self._state = PipelineState.STOPPING
+        logger.info("Stopping pipeline...")
+
+        if self._stop_event:
+            self._stop_event.set()
+
+        await self._stop_background_services()
+        await self._cleanup()
+
+        self._state = PipelineState.STOPPED
+        logger.info("Pipeline stopped")
+
+    async def _initialize_components(self) -> None:
+        """Initialize all pipeline components."""
+        settings = self._settings
+
+        # Initialize Redis
+        logger.debug("Initializing Redis connection...")
+        self._redis = Redis.from_url(settings.redis.url)
+
+        # Initialize Database Manager
+        logger.debug("Initializing database manager...")
+        self._db_manager = DatabaseManager(
+            settings.database.url,
+            async_mode=True,
+        )
+
+        # Initialize Polygon client
+        logger.debug("Initializing Polygon client...")
+        self._polygon_client = PolygonClient(
+            settings.polygon.rpc_url,
+            fallback_rpc_url=settings.polygon.fallback_rpc_url,
+            redis=self._redis,
+        )
+
+        # Initialize CLOB client
+        logger.debug("Initializing CLOB client...")
+        api_key = (
+            settings.polymarket.api_key.get_secret_value() if settings.polymarket.api_key else None
+        )
+        self._clob_client = ClobClient(api_key=api_key)
+
+        # Initialize Market Metadata Sync
+        logger.debug("Initializing market metadata sync...")
+        self._metadata_sync = MarketMetadataSync(
+            redis=self._redis,
+            clob_client=self._clob_client,
+        )
+
+        # Initialize Wallet Analyzer
+        logger.debug("Initializing wallet analyzer...")
+        self._wallet_analyzer = WalletAnalyzer(
+            self._polygon_client,
+            redis=self._redis,
+        )
+
+        # Initialize Funding Tracer
+        logger.debug("Initializing funding tracer...")
+        self._funding_tracer = FundingTracer(self._polygon_client)
+
+        # Initialize Detectors
+        logger.debug("Initializing detectors...")
+        self._fresh_wallet_detector = FreshWalletDetector(self._wallet_analyzer)
+        self._size_anomaly_detector = SizeAnomalyDetector(self._metadata_sync)
+
+        # Initialize Risk Scorer
+        logger.debug("Initializing risk scorer...")
+        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...")
+        self._alert_formatter = AlertFormatter(verbosity="detailed")
+        channels = self._build_alert_channels()
+        self._alert_dispatcher = AlertDispatcher(channels)
+
+        # Initialize Trade Stream
+        logger.debug("Initializing trade stream handler...")
+        self._trade_stream = TradeStreamHandler(
+            on_trade=self._on_trade,
+            host=settings.polymarket.ws_url,
+        )
+
+        logger.info("All components initialized")
+
+    def _build_alert_channels(self) -> list[AlertChannel]:
+        """Build list of enabled alert channels."""
+        channels: list[AlertChannel] = []
+        settings = self._settings
+
+        if settings.discord.enabled and settings.discord.webhook_url:
+            webhook_url = settings.discord.webhook_url.get_secret_value()
+            channels.append(DiscordChannel(webhook_url))
+            logger.info("Discord channel enabled")
+
+        if settings.telegram.enabled:
+            bot_token = settings.telegram.bot_token
+            chat_id = settings.telegram.chat_id
+            if bot_token and chat_id:
+                channels.append(
+                    TelegramChannel(
+                        bot_token.get_secret_value(),
+                        chat_id,
+                    )
+                )
+                logger.info("Telegram channel enabled")
+
+        if not channels:
+            logger.warning("No alert channels configured")
+
+        return channels
+
+    async def _start_background_services(self) -> None:
+        """Start background services."""
+        # Start metadata sync
+        if self._metadata_sync:
+            logger.debug("Starting metadata sync service...")
+            await self._metadata_sync.start()
+
+        # Start trade stream in background task
+        if self._trade_stream:
+            logger.debug("Starting trade stream...")
+            self._stream_task = asyncio.create_task(self._run_trade_stream())
+
+    async def _run_trade_stream(self) -> None:
+        """Run the trade stream in a task."""
+        if not self._trade_stream:
+            return
+
+        try:
+            await self._trade_stream.start()
+        except asyncio.CancelledError:
+            logger.debug("Trade stream task cancelled")
+        except Exception as e:
+            logger.error("Trade stream error: %s", e)
+            self._stats.last_error = str(e)
+            self._stats.errors += 1
+
+    async def _stop_background_services(self) -> None:
+        """Stop background services."""
+        # Stop trade stream
+        if self._trade_stream:
+            logger.debug("Stopping trade stream...")
+            await self._trade_stream.stop()
+
+        # Cancel stream task
+        if self._stream_task:
+            self._stream_task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await self._stream_task
+            self._stream_task = None
+
+        # Stop metadata sync
+        if self._metadata_sync:
+            logger.debug("Stopping metadata sync...")
+            await self._metadata_sync.stop()
+
+    async def _cleanup(self) -> None:
+        """Clean up resources."""
+        # Close database connections
+        if self._db_manager:
+            await self._db_manager.dispose_async()
+            self._db_manager = None
+
+        # Close Redis connection
+        if self._redis:
+            await self._redis.aclose()
+            self._redis = None
+
+        logger.debug("Resources cleaned up")
+
+    async def _on_trade(self, trade: TradeEvent) -> None:
+        """Process a single trade event.
+
+        This is the main event handler that runs the detection pipeline:
+        1. Run fresh wallet detection
+        2. Run size anomaly detection
+        3. Score the combined signals
+        4. Send alert if threshold exceeded
+
+        Args:
+            trade: The trade event from the WebSocket stream.
+        """
+        self._stats.trades_processed += 1
+        self._stats.last_trade_time = datetime.now(UTC)
+
+        try:
+            # Run detectors in parallel
+            fresh_signal, size_signal = await asyncio.gather(
+                self._detect_fresh_wallet(trade),
+                self._detect_size_anomaly(trade),
+            )
+
+            # Persist wallet profile and funding data when a fresh wallet is detected
+            if fresh_signal is not None:
+                await self._persist_wallet_and_funding(fresh_signal)
+
+            # Bundle signals
+            bundle = SignalBundle(
+                trade_event=trade,
+                fresh_wallet_signal=fresh_signal,
+                size_anomaly_signal=size_signal,
+            )
+
+            # Score and potentially alert
+            if fresh_signal or size_signal:
+                self._stats.signals_generated += 1
+                await self._score_and_alert(bundle)
+
+        except Exception as e:
+            logger.error("Error processing trade %s: %s", trade.trade_id, e)
+            self._stats.errors += 1
+            self._stats.last_error = str(e)
+
+    async def _persist_wallet_and_funding(self, signal: FreshWalletSignal) -> None:
+        """Persist wallet profile and funding transfers to Postgres.
+
+        Called when a fresh wallet signal is detected. Upserts the wallet
+        profile and traces/inserts any funding transfers found on-chain.
+
+        Args:
+            signal: The fresh wallet signal containing the wallet profile.
+        """
+        if not self._db_manager:
+            return
+
+        profile = signal.wallet_profile
+        address = profile.address
+
+        try:
+            async with self._db_manager.get_async_session() as session:
+                # Persist wallet profile
+                wallet_repo = WalletRepository(session)
+                dto = WalletProfileDTO(
+                    address=address,
+                    nonce=profile.nonce,
+                    first_seen_at=profile.first_seen,
+                    is_fresh=profile.is_fresh,
+                    matic_balance=profile.matic_balance,
+                    usdc_balance=profile.usdc_balance,
+                    analyzed_at=profile.analyzed_at,
+                )
+                await wallet_repo.upsert(dto)
+
+                # Trace and persist funding transfers
+                if self._funding_tracer:
+                    chain = await self._funding_tracer.trace(address)
+                    if chain.chain:
+                        funding_repo = FundingRepository(session)
+                        funding_dtos = [
+                            FundingTransferDTO(
+                                from_address=t.from_address,
+                                to_address=t.to_address,
+                                amount=t.amount,
+                                token=t.token,
+                                tx_hash=t.tx_hash,
+                                block_number=t.block_number,
+                                timestamp=t.timestamp,
+                            )
+                            for t in chain.chain
+                        ]
+                        await funding_repo.insert_many(funding_dtos)
+
+                logger.debug(
+                    "Persisted wallet profile and %d funding transfers for %s",
+                    len(chain.chain) if self._funding_tracer and chain.chain else 0,
+                    address[:10] + "...",
+                )
+        except Exception as e:
+            logger.warning("Failed to persist wallet/funding data for %s: %s", address, e)
+
+    async def _detect_fresh_wallet(self, trade: TradeEvent) -> FreshWalletSignal | None:
+        """Run fresh wallet detection."""
+        if not self._fresh_wallet_detector:
+            return None
+        try:
+            return await self._fresh_wallet_detector.analyze(trade)
+        except Exception as e:
+            logger.warning("Fresh wallet detection failed for %s: %s", trade.trade_id, e)
+            return None
+
+    async def _detect_size_anomaly(self, trade: TradeEvent) -> SizeAnomalySignal | None:
+        """Run size anomaly detection."""
+        if not self._size_anomaly_detector:
+            return None
+        try:
+            return await self._size_anomaly_detector.analyze(trade)
+        except Exception as e:
+            logger.warning("Size anomaly detection failed for %s: %s", trade.trade_id, e)
+            return None
+
+    async def _score_and_alert(self, bundle: SignalBundle) -> None:
+        """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)",
+                bundle.trade_event.trade_id,
+                assessment.weighted_score,
+            )
+            return
+
+        # Format and dispatch alert
+        formatted_alert = self._alert_formatter.format(assessment)
+
+        if self._dry_run:
+            logger.info(
+                "[DRY RUN] Would send alert: wallet=%s, score=%.2f",
+                assessment.wallet_address[:10] + "...",
+                assessment.weighted_score,
+            )
+            return
+
+        result = await self._alert_dispatcher.dispatch(formatted_alert)
+
+        if result.all_succeeded:
+            self._stats.alerts_sent += 1
+            logger.info(
+                "Alert sent successfully: wallet=%s, score=%.2f",
+                assessment.wallet_address[:10] + "...",
+                assessment.weighted_score,
+            )
+        else:
+            logger.warning(
+                "Alert partially failed: %d/%d channels succeeded",
+                result.success_count,
+                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.
+
+        This is a convenience method that starts the pipeline and
+        blocks until a stop signal is received.
+
+        Example:
+            ```python
+            pipeline = Pipeline()
+            try:
+                await pipeline.run()
+            except KeyboardInterrupt:
+                pass
+            ```
+        """
+        await self.start()
+
+        try:
+            if self._stop_event:
+                await self._stop_event.wait()
+        except asyncio.CancelledError:
+            pass
+        finally:
+            await self.stop()
+
+    async def __aenter__(self) -> Pipeline:
+        """Async context manager entry."""
+        await self.start()
+        return self
+
+    async def __aexit__(self, *args: Any) -> None:
+        """Async context manager exit."""
+        await self.stop()

src/polymarket_insider_tracker/profiler/analyzer.py

@@ -91,7 +91,9 @@ class WalletAnalyzer:
             return WalletProfile(
                 address=data["address"],
                 nonce=data["nonce"],
-                first_seen=datetime.fromisoformat(data["first_seen"]) if data["first_seen"] else None,
+                first_seen=datetime.fromisoformat(data["first_seen"])
+                if data["first_seen"]
+                else None,
                 age_hours=data["age_hours"],
                 is_fresh=data["is_fresh"],
                 total_tx_count=data["total_tx_count"],

src/polymarket_insider_tracker/profiler/chain.py

@@ -503,9 +503,7 @@ class PolygonClient:
         balance_task = self.get_balance(address)
         first_tx_task = self.get_first_transaction(address)
 
-        nonce, balance, first_tx = await asyncio.gather(
-            nonce_task, balance_task, first_tx_task
-        )
+        nonce, balance, first_tx = await asyncio.gather(nonce_task, balance_task, first_tx_task)
 
         return WalletInfo(
             address=address.lower(),

src/polymarket_insider_tracker/profiler/entities.py

@@ -195,19 +195,16 @@ class EntityRegistry:
             True if the address is a known smart contract.
         """
         entity_type = self.classify(address)
-        contract_types = (
-            self.DEX_ENTITY_TYPES
-            | {
-                EntityType.TOKEN_USDC,
-                EntityType.TOKEN_USDT,
-                EntityType.TOKEN_WETH,
-                EntityType.TOKEN_WMATIC,
-                EntityType.DEFI_AAVE,
-                EntityType.DEFI_COMPOUND,
-                EntityType.DEFI_OTHER,
-                EntityType.CONTRACT,
-            }
-        )
+        contract_types = self.DEX_ENTITY_TYPES | {
+            EntityType.TOKEN_USDC,
+            EntityType.TOKEN_USDT,
+            EntityType.TOKEN_WETH,
+            EntityType.TOKEN_WMATIC,
+            EntityType.DEFI_AAVE,
+            EntityType.DEFI_COMPOUND,
+            EntityType.DEFI_OTHER,
+            EntityType.CONTRACT,
+        }
         return entity_type in contract_types
 
     def get_entity_category(self, address: str) -> str:

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,45 +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
-        logs = await w3.eth.get_logs(
+        w3 = self._select_w3()
+        # Note: web3 typing is overly restrictive for block params
+        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)
-                ],
-                "fromBlock": from_block,
-                "toBlock": to_block,
+                "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,
@@ -310,7 +456,7 @@ class FundingTracer:
 
         chains: dict[str, FundingChain] = {}
         for addr, result in zip(addresses, results, strict=True):
-            if isinstance(result, Exception):
+            if isinstance(result, BaseException):
                 logger.warning("Failed to trace %s: %s", addr, result)
                 chains[addr.lower()] = FundingChain(
                     target_address=addr.lower(),

src/polymarket_insider_tracker/profiler/models.py

@@ -58,7 +58,10 @@ class WalletInfo:
         """Return wallet age in days based on first transaction."""
         if self.first_transaction is None:
             return None
-        delta = datetime.now(tz=self.first_transaction.timestamp.tzinfo) - self.first_transaction.timestamp
+        delta = (
+            datetime.now(tz=self.first_transaction.timestamp.tzinfo)
+            - self.first_transaction.timestamp
+        )
         return delta.total_seconds() / 86400
 
 

src/polymarket_insider_tracker/shutdown.py

@@ -0,0 +1,325 @@
+"""Graceful shutdown handler for Polymarket Insider Tracker.
+
+This module provides signal handling and graceful shutdown coordination
+for the async pipeline components.
+
+Usage:
+    ```python
+    async def main():
+        shutdown = GracefulShutdown()
+
+        async with shutdown:
+            pipeline = Pipeline(settings)
+            await pipeline.start()
+
+            # Wait for shutdown signal
+            await shutdown.wait()
+
+            # Graceful cleanup
+            await pipeline.stop()
+    ```
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import signal
+import sys
+from collections.abc import Callable
+from contextlib import suppress
+from types import FrameType
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# Default shutdown timeout in seconds
+DEFAULT_SHUTDOWN_TIMEOUT = 30.0
+
+# Signals to trap for graceful shutdown
+SHUTDOWN_SIGNALS = (signal.SIGTERM, signal.SIGINT)
+
+
+class ShutdownTimeoutError(Exception):
+    """Raised when graceful shutdown exceeds timeout."""
+
+
+class GracefulShutdown:
+    """Graceful shutdown handler with signal trapping.
+
+    This class provides coordinated shutdown handling for async applications.
+    It traps SIGTERM and SIGINT signals and provides an async event that
+    can be awaited to detect shutdown requests.
+
+    Features:
+        - SIGTERM and SIGINT signal trapping
+        - Async event-based shutdown coordination
+        - Configurable shutdown timeout
+        - Async context manager support
+        - Cleanup callback registration
+        - Force exit on second signal or timeout
+
+    Example:
+        ```python
+        shutdown = GracefulShutdown(timeout=30.0)
+
+        async with shutdown:
+            await some_long_running_task()
+            # Automatically handles cleanup on signals
+        ```
+    """
+
+    def __init__(
+        self,
+        timeout: float = DEFAULT_SHUTDOWN_TIMEOUT,
+        *,
+        exit_on_timeout: bool = True,
+    ) -> None:
+        """Initialize the shutdown handler.
+
+        Args:
+            timeout: Maximum time in seconds to wait for graceful shutdown.
+            exit_on_timeout: If True, force exit when timeout is exceeded.
+        """
+        self._timeout = timeout
+        self._exit_on_timeout = exit_on_timeout
+
+        self._shutdown_event: asyncio.Event | None = None
+        self._shutdown_requested = False
+        self._force_exit_requested = False
+        self._original_handlers: dict[signal.Signals, Any] = {}
+        self._cleanup_callbacks: list[Callable[[], Any]] = []
+        self._loop: asyncio.AbstractEventLoop | None = None
+
+    @property
+    def timeout(self) -> float:
+        """Shutdown timeout in seconds."""
+        return self._timeout
+
+    @property
+    def is_shutdown_requested(self) -> bool:
+        """Check if shutdown has been requested."""
+        return self._shutdown_requested
+
+    @property
+    def is_force_exit_requested(self) -> bool:
+        """Check if force exit has been requested (second signal received)."""
+        return self._force_exit_requested
+
+    def register_cleanup(self, callback: Callable[[], Any]) -> None:
+        """Register a cleanup callback to run during shutdown.
+
+        Args:
+            callback: A callable (sync or async) to run during shutdown.
+        """
+        self._cleanup_callbacks.append(callback)
+
+    def request_shutdown(self) -> None:
+        """Programmatically request shutdown.
+
+        This can be used to trigger shutdown from application code
+        instead of waiting for a signal.
+        """
+        if not self._shutdown_requested:
+            self._shutdown_requested = True
+            logger.info("Shutdown requested programmatically")
+            if self._shutdown_event:
+                self._shutdown_event.set()
+
+    async def wait(self) -> None:
+        """Wait for a shutdown signal.
+
+        This coroutine blocks until a shutdown signal is received
+        or request_shutdown() is called.
+        """
+        if self._shutdown_event is None:
+            self._shutdown_event = asyncio.Event()
+
+        await self._shutdown_event.wait()
+
+    async def wait_with_timeout(self) -> bool:
+        """Wait for shutdown with timeout.
+
+        Returns:
+            True if shutdown was requested, False if timeout occurred.
+        """
+        if self._shutdown_event is None:
+            self._shutdown_event = asyncio.Event()
+
+        try:
+            await asyncio.wait_for(
+                self._shutdown_event.wait(),
+                timeout=self._timeout,
+            )
+            return True
+        except TimeoutError:
+            return False
+
+    def install_signal_handlers(self) -> None:
+        """Install signal handlers for graceful shutdown.
+
+        Traps SIGTERM and SIGINT to trigger graceful shutdown.
+        On Windows, only SIGINT is trapped as SIGTERM is not available.
+        """
+        self._loop = asyncio.get_running_loop()
+        self._shutdown_event = asyncio.Event()
+
+        # Platform-specific signal handling
+        if sys.platform == "win32":
+            # Windows: use signal.signal for SIGINT only
+            self._install_windows_handlers()
+        else:
+            # Unix: use loop.add_signal_handler for both signals
+            self._install_unix_handlers()
+
+        logger.debug("Signal handlers installed")
+
+    def _install_unix_handlers(self) -> None:
+        """Install Unix signal handlers using the event loop."""
+        if self._loop is None:
+            return
+
+        for sig in SHUTDOWN_SIGNALS:
+            try:
+                self._loop.add_signal_handler(
+                    sig,
+                    self._handle_signal,
+                    sig,
+                )
+                logger.debug("Installed handler for %s", sig.name)
+            except (ValueError, OSError) as e:
+                logger.warning("Could not install handler for %s: %s", sig.name, e)
+
+    def _install_windows_handlers(self) -> None:
+        """Install Windows signal handlers using signal.signal."""
+        for sig in SHUTDOWN_SIGNALS:
+            try:
+                self._original_handlers[sig] = signal.signal(
+                    sig,
+                    self._handle_signal_sync,
+                )
+                logger.debug("Installed handler for %s", sig.name)
+            except (ValueError, OSError) as e:
+                logger.warning("Could not install handler for %s: %s", sig.name, e)
+
+    def remove_signal_handlers(self) -> None:
+        """Remove installed signal handlers and restore originals."""
+        if sys.platform == "win32":
+            self._remove_windows_handlers()
+        else:
+            self._remove_unix_handlers()
+
+        logger.debug("Signal handlers removed")
+
+    def _remove_unix_handlers(self) -> None:
+        """Remove Unix signal handlers."""
+        if self._loop is None:
+            return
+
+        for sig in SHUTDOWN_SIGNALS:
+            with suppress(ValueError, OSError):
+                self._loop.remove_signal_handler(sig)
+
+    def _remove_windows_handlers(self) -> None:
+        """Remove Windows signal handlers and restore originals."""
+        for sig, original in self._original_handlers.items():
+            with suppress(ValueError, OSError):
+                signal.signal(sig, original)
+        self._original_handlers.clear()
+
+    def _handle_signal(self, sig: signal.Signals) -> None:
+        """Handle shutdown signal (Unix version).
+
+        Args:
+            sig: The signal that was received.
+        """
+        if self._shutdown_requested:
+            # Second signal - force exit
+            self._force_exit_requested = True
+            logger.warning("Received %s again - forcing exit!", sig.name)
+            sys.exit(128 + sig.value)
+        else:
+            self._shutdown_requested = True
+            logger.info("Received %s - initiating graceful shutdown...", sig.name)
+            if self._shutdown_event:
+                self._shutdown_event.set()
+
+    def _handle_signal_sync(self, sig: int, _frame: FrameType | None) -> None:
+        """Handle shutdown signal (Windows version).
+
+        Args:
+            sig: The signal number that was received.
+            _frame: The current stack frame (unused).
+        """
+        sig_enum = signal.Signals(sig)
+        self._handle_signal(sig_enum)
+
+    async def run_cleanup_callbacks(self) -> None:
+        """Run all registered cleanup callbacks."""
+        for callback in self._cleanup_callbacks:
+            try:
+                result = callback()
+                if asyncio.iscoroutine(result):
+                    await result
+            except Exception as e:
+                logger.error("Cleanup callback failed: %s", e)
+
+    async def __aenter__(self) -> GracefulShutdown:
+        """Async context manager entry - install signal handlers."""
+        self.install_signal_handlers()
+        return self
+
+    async def __aexit__(self, *_args: Any) -> None:
+        """Async context manager exit - cleanup."""
+        self.remove_signal_handlers()
+        await self.run_cleanup_callbacks()
+
+
+async def run_with_graceful_shutdown(
+    coro: Any,
+    *,
+    timeout: float = DEFAULT_SHUTDOWN_TIMEOUT,
+) -> None:
+    """Run an async coroutine with graceful shutdown handling.
+
+    This is a convenience function that wraps a coroutine with
+    signal handling and timeout-based cleanup.
+
+    Args:
+        coro: The coroutine to run.
+        timeout: Maximum time to wait for graceful shutdown.
+
+    Example:
+        ```python
+        async def my_app():
+            await asyncio.sleep(3600)  # Run for an hour
+
+        # Will handle SIGTERM/SIGINT gracefully
+        await run_with_graceful_shutdown(my_app())
+        ```
+    """
+    shutdown = GracefulShutdown(timeout=timeout)
+
+    async with shutdown:
+        task = asyncio.create_task(coro)
+
+        # Wait for either task completion or shutdown signal
+        shutdown_wait = asyncio.create_task(shutdown.wait())
+
+        done, pending = await asyncio.wait(
+            [task, shutdown_wait],
+            return_when=asyncio.FIRST_COMPLETED,
+        )
+
+        # Cancel pending tasks
+        for pending_task in pending:
+            pending_task.cancel()
+            with suppress(asyncio.CancelledError):
+                await pending_task
+
+        # If the main task is in done, get any exception
+        if task in done:
+            task.result()

src/polymarket_insider_tracker/storage/models.py

@@ -109,7 +109,63 @@ class WalletRelationshipModel(Base):
     )
 
     __table_args__ = (
-        UniqueConstraint("wallet_a", "wallet_b", "relationship_type", name="uq_wallet_relationship"),
+        UniqueConstraint(
+            "wallet_a", "wallet_b", "relationship_type", name="uq_wallet_relationship"
+        ),
         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,
 )
@@ -208,20 +209,20 @@ class WalletRepository:
             await self.session.execute(stmt)
         except Exception:
             # Fall back to SQLite upsert for testing
-            stmt = sqlite_insert(WalletProfileModel).values(**values, created_at=now)
-            stmt = stmt.on_conflict_do_update(
+            sqlite_stmt = sqlite_insert(WalletProfileModel).values(**values, created_at=now)
+            sqlite_stmt = sqlite_stmt.on_conflict_do_update(
                 index_elements=["address"],
                 set_={
-                    "nonce": stmt.excluded.nonce,
-                    "first_seen_at": stmt.excluded.first_seen_at,
-                    "is_fresh": stmt.excluded.is_fresh,
-                    "matic_balance": stmt.excluded.matic_balance,
-                    "usdc_balance": stmt.excluded.usdc_balance,
-                    "analyzed_at": stmt.excluded.analyzed_at,
-                    "updated_at": stmt.excluded.updated_at,
+                    "nonce": sqlite_stmt.excluded.nonce,
+                    "first_seen_at": sqlite_stmt.excluded.first_seen_at,
+                    "is_fresh": sqlite_stmt.excluded.is_fresh,
+                    "matic_balance": sqlite_stmt.excluded.matic_balance,
+                    "usdc_balance": sqlite_stmt.excluded.usdc_balance,
+                    "analyzed_at": sqlite_stmt.excluded.analyzed_at,
+                    "updated_at": sqlite_stmt.excluded.updated_at,
                 },
             )
-            await self.session.execute(stmt)
+            await self.session.execute(sqlite_stmt)
 
         await self.session.flush()
         return dto
@@ -238,7 +239,8 @@ class WalletRepository:
         result = await self.session.execute(
             delete(WalletProfileModel).where(WalletProfileModel.address == address.lower())
         )
-        return result.rowcount > 0
+        # SQLAlchemy Result does have rowcount but typing doesn't reflect it
+        return (result.rowcount or 0) > 0  # type: ignore[attr-defined]
 
     async def mark_stale(self, address: str) -> bool:
         """Mark a wallet profile as stale (soft delete).
@@ -257,7 +259,8 @@ class WalletRepository:
             .where(WalletProfileModel.address == address.lower())
             .values(analyzed_at=stale_time, updated_at=datetime.now(UTC))
         )
-        return result.rowcount > 0
+        # SQLAlchemy Result does have rowcount but typing doesn't reflect it
+        return (result.rowcount or 0) > 0  # type: ignore[attr-defined]
 
 
 class FundingRepository:
@@ -274,9 +277,7 @@ class FundingRepository:
         """
         self.session = session
 
-    async def get_transfers_to(
-        self, address: str, limit: int = 100
-    ) -> list[FundingTransferDTO]:
+    async def get_transfers_to(self, address: str, limit: int = 100) -> list[FundingTransferDTO]:
         """Get transfers to a wallet address.
 
         Args:
@@ -294,9 +295,7 @@ class FundingRepository:
         )
         return [FundingTransferDTO.from_model(m) for m in result.scalars().all()]
 
-    async def get_transfers_from(
-        self, address: str, limit: int = 100
-    ) -> list[FundingTransferDTO]:
+    async def get_transfers_from(self, address: str, limit: int = 100) -> list[FundingTransferDTO]:
         """Get transfers from a wallet address.
 
         Args:
@@ -482,19 +481,17 @@ class RelationshipRepository:
             await self.session.execute(stmt)
         except Exception:
             # Fall back to SQLite upsert for testing
-            stmt = sqlite_insert(WalletRelationshipModel).values(**values)
-            stmt = stmt.on_conflict_do_update(
+            sqlite_stmt = sqlite_insert(WalletRelationshipModel).values(**values)
+            sqlite_stmt = sqlite_stmt.on_conflict_do_update(
                 index_elements=["wallet_a", "wallet_b", "relationship_type"],
-                set_={"confidence": stmt.excluded.confidence},
+                set_={"confidence": sqlite_stmt.excluded.confidence},
             )
-            await self.session.execute(stmt)
+            await self.session.execute(sqlite_stmt)
 
         await self.session.flush()
         return dto
 
-    async def delete(
-        self, wallet_a: str, wallet_b: str, relationship_type: str
-    ) -> bool:
+    async def delete(self, wallet_a: str, wallet_b: str, relationship_type: str) -> bool:
         """Delete a specific relationship.
 
         Args:
@@ -512,4 +509,109 @@ class RelationshipRepository:
                 WalletRelationshipModel.relationship_type == relationship_type,
             )
         )
-        return result.rowcount > 0
+        # 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_dispatcher.py

@@ -75,9 +75,7 @@ class TestDiscordChannel:
     @pytest.mark.asyncio
     async def test_send_success(self, sample_alert: FormattedAlert) -> None:
         """Test successful Discord message send."""
-        channel = DiscordChannel(
-            webhook_url="https://discord.com/api/webhooks/123/abc"
-        )
+        channel = DiscordChannel(webhook_url="https://discord.com/api/webhooks/123/abc")
 
         with patch("httpx.AsyncClient") as mock_client_class:
             mock_response = MagicMock()
@@ -310,9 +308,7 @@ class TestAlertDispatcher:
         mock_telegram_channel: MagicMock,
     ) -> None:
         """Test dispatcher initialization."""
-        dispatcher = AlertDispatcher(
-            channels=[mock_discord_channel, mock_telegram_channel]
-        )
+        dispatcher = AlertDispatcher(channels=[mock_discord_channel, mock_telegram_channel])
         assert len(dispatcher.channels) == 2
         assert "discord" in dispatcher._circuit_state
         assert "telegram" in dispatcher._circuit_state
@@ -325,9 +321,7 @@ class TestAlertDispatcher:
         mock_telegram_channel: MagicMock,
     ) -> None:
         """Test successful dispatch to all channels."""
-        dispatcher = AlertDispatcher(
-            channels=[mock_discord_channel, mock_telegram_channel]
-        )
+        dispatcher = AlertDispatcher(channels=[mock_discord_channel, mock_telegram_channel])
 
         result = await dispatcher.dispatch(sample_alert)
 
@@ -345,9 +339,7 @@ class TestAlertDispatcher:
         """Test dispatch with one channel failing."""
         mock_telegram_channel.send.return_value = False
 
-        dispatcher = AlertDispatcher(
-            channels=[mock_discord_channel, mock_telegram_channel]
-        )
+        dispatcher = AlertDispatcher(channels=[mock_discord_channel, mock_telegram_channel])
 
         result = await dispatcher.dispatch(sample_alert)
 
@@ -357,9 +349,7 @@ class TestAlertDispatcher:
         assert result.channel_results["telegram"] is False
 
     @pytest.mark.asyncio
-    async def test_dispatch_no_channels(
-        self, sample_alert: FormattedAlert
-    ) -> None:
+    async def test_dispatch_no_channels(self, sample_alert: FormattedAlert) -> None:
         """Test dispatch with no channels configured."""
         dispatcher = AlertDispatcher(channels=[])
 
@@ -435,9 +425,7 @@ class TestAlertDispatcher:
         # Now succeed
         mock_discord_channel.send.return_value = True
         # Force half-open by resetting last_failure to past
-        dispatcher._circuit_state["discord"].last_failure_time = datetime(
-            2020, 1, 1, tzinfo=UTC
-        )
+        dispatcher._circuit_state["discord"].last_failure_time = datetime(2020, 1, 1, tzinfo=UTC)
 
         result = await dispatcher.dispatch(sample_alert)
 
@@ -465,9 +453,7 @@ class TestAlertDispatcher:
         mock_telegram_channel: MagicMock,
     ) -> None:
         """Test getting circuit status."""
-        dispatcher = AlertDispatcher(
-            channels=[mock_discord_channel, mock_telegram_channel]
-        )
+        dispatcher = AlertDispatcher(channels=[mock_discord_channel, mock_telegram_channel])
 
         status = dispatcher.get_circuit_status()
 

tests/alerter/test_formatter.py

@@ -284,17 +284,13 @@ class TestAlertFormatterInit:
 class TestAlertFormatterFormat:
     """Tests for AlertFormatter.format method."""
 
-    def test_format_returns_formatted_alert(
-        self, high_risk_assessment: RiskAssessment
-    ) -> None:
+    def test_format_returns_formatted_alert(self, high_risk_assessment: RiskAssessment) -> None:
         """Test that format returns a FormattedAlert."""
         formatter = AlertFormatter()
         result = formatter.format(high_risk_assessment)
         assert isinstance(result, FormattedAlert)
 
-    def test_format_includes_all_fields(
-        self, high_risk_assessment: RiskAssessment
-    ) -> None:
+    def test_format_includes_all_fields(self, high_risk_assessment: RiskAssessment) -> None:
         """Test that all fields are populated."""
         formatter = AlertFormatter()
         result = formatter.format(high_risk_assessment)
@@ -306,26 +302,20 @@ class TestAlertFormatterFormat:
         assert result.plain_text != ""
         assert result.links != {}
 
-    def test_format_title_includes_risk_level(
-        self, high_risk_assessment: RiskAssessment
-    ) -> None:
+    def test_format_title_includes_risk_level(self, high_risk_assessment: RiskAssessment) -> None:
         """Test that title includes risk level."""
         formatter = AlertFormatter()
         result = formatter.format(high_risk_assessment)
         assert "HIGH" in result.title
 
-    def test_format_includes_wallet_link(
-        self, high_risk_assessment: RiskAssessment
-    ) -> None:
+    def test_format_includes_wallet_link(self, high_risk_assessment: RiskAssessment) -> None:
         """Test that wallet explorer link is included."""
         formatter = AlertFormatter()
         result = formatter.format(high_risk_assessment)
         assert "wallet" in result.links
         assert "polygonscan.com" in result.links["wallet"]
 
-    def test_format_includes_market_link(
-        self, high_risk_assessment: RiskAssessment
-    ) -> None:
+    def test_format_includes_market_link(self, high_risk_assessment: RiskAssessment) -> None:
         """Test that market link is included when slug available."""
         formatter = AlertFormatter()
         result = formatter.format(high_risk_assessment)
@@ -336,9 +326,7 @@ class TestAlertFormatterFormat:
 class TestDiscordEmbed:
     """Tests for Discord embed format."""
 
-    def test_embed_has_required_fields(
-        self, high_risk_assessment: RiskAssessment
-    ) -> None:
+    def test_embed_has_required_fields(self, high_risk_assessment: RiskAssessment) -> None:
         """Test that embed has required Discord fields."""
         formatter = AlertFormatter()
         result = formatter.format(high_risk_assessment)
@@ -349,17 +337,13 @@ class TestDiscordEmbed:
         assert "fields" in embed
         assert "footer" in embed
 
-    def test_embed_color_reflects_risk(
-        self, high_risk_assessment: RiskAssessment
-    ) -> None:
+    def test_embed_color_reflects_risk(self, high_risk_assessment: RiskAssessment) -> None:
         """Test that embed color matches risk level."""
         formatter = AlertFormatter()
         result = formatter.format(high_risk_assessment)
         assert result.discord_embed["color"] == COLOR_HIGH_RISK
 
-    def test_embed_includes_wallet_field(
-        self, high_risk_assessment: RiskAssessment
-    ) -> None:
+    def test_embed_includes_wallet_field(self, high_risk_assessment: RiskAssessment) -> None:
         """Test that embed includes wallet field."""
         formatter = AlertFormatter()
         result = formatter.format(high_risk_assessment)
@@ -369,9 +353,7 @@ class TestDiscordEmbed:
         assert wallet_field is not None
         assert "0x1234" in wallet_field["value"]
 
-    def test_embed_includes_wallet_age(
-        self, high_risk_assessment: RiskAssessment
-    ) -> None:
+    def test_embed_includes_wallet_age(self, high_risk_assessment: RiskAssessment) -> None:
         """Test that wallet age is shown when fresh wallet signal present."""
         formatter = AlertFormatter()
         result = formatter.format(high_risk_assessment)
@@ -380,9 +362,7 @@ class TestDiscordEmbed:
         wallet_field = next((f for f in fields if f["name"] == "Wallet"), None)
         assert "Age:" in wallet_field["value"]
 
-    def test_embed_includes_trade_details(
-        self, high_risk_assessment: RiskAssessment
-    ) -> None:
+    def test_embed_includes_trade_details(self, high_risk_assessment: RiskAssessment) -> None:
         """Test that trade details are in embed."""
         formatter = AlertFormatter()
         result = formatter.format(high_risk_assessment)
@@ -393,9 +373,7 @@ class TestDiscordEmbed:
         assert "BUY" in trade_field["value"]
         assert "Yes" in trade_field["value"]
 
-    def test_embed_includes_signals_field(
-        self, high_risk_assessment: RiskAssessment
-    ) -> None:
+    def test_embed_includes_signals_field(self, high_risk_assessment: RiskAssessment) -> None:
         """Test that signals are listed in embed."""
         formatter = AlertFormatter()
         result = formatter.format(high_risk_assessment)
@@ -405,9 +383,7 @@ class TestDiscordEmbed:
         assert signals_field is not None
         assert "Fresh Wallet" in signals_field["value"]
 
-    def test_detailed_embed_includes_confidence(
-        self, high_risk_assessment: RiskAssessment
-    ) -> None:
+    def test_detailed_embed_includes_confidence(self, high_risk_assessment: RiskAssessment) -> None:
         """Test that detailed mode includes confidence breakdown."""
         formatter = AlertFormatter(verbosity="detailed")
         result = formatter.format(high_risk_assessment)
@@ -420,34 +396,41 @@ class TestDiscordEmbed:
 class TestTelegramMarkdown:
     """Tests for Telegram markdown format."""
 
-    def test_telegram_includes_header(
-        self, high_risk_assessment: RiskAssessment
-    ) -> None:
+    def test_telegram_includes_header(self, high_risk_assessment: RiskAssessment) -> None:
         """Test that Telegram message has header."""
         formatter = AlertFormatter()
         result = formatter.format(high_risk_assessment)
         assert "*Suspicious Activity Detected*" in result.telegram_markdown
 
-    def test_telegram_includes_wallet(
-        self, high_risk_assessment: RiskAssessment
-    ) -> None:
+    def test_telegram_includes_wallet(self, high_risk_assessment: RiskAssessment) -> None:
         """Test that Telegram message includes wallet."""
         formatter = AlertFormatter()
         result = formatter.format(high_risk_assessment)
         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."""
+    def test_telegram_includes_risk_score(self, high_risk_assessment: RiskAssessment) -> None:
+        """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_includes_links(
-        self, high_risk_assessment: RiskAssessment
-    ) -> None:
+    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()
         result = formatter.format(high_risk_assessment)
@@ -489,9 +472,7 @@ class TestPlainText:
 class TestCompactVerbosity:
     """Tests for compact verbosity mode."""
 
-    def test_compact_body_is_shorter(
-        self, high_risk_assessment: RiskAssessment
-    ) -> None:
+    def test_compact_body_is_shorter(self, high_risk_assessment: RiskAssessment) -> None:
         """Test that compact mode produces shorter body."""
         detailed_formatter = AlertFormatter(verbosity="detailed")
         compact_formatter = AlertFormatter(verbosity="compact")

tests/detector/test_scorer.py

@@ -76,9 +76,7 @@ def sample_metadata() -> MarketMetadata:
         condition_id="market_abc123",
         question="Will it rain tomorrow?",
         description="Weather prediction market",
-        tokens=(
-            Token(token_id="token_123", outcome="Yes", price=Decimal("0.65")),
-        ),
+        tokens=(Token(token_id="token_123", outcome="Yes", price=Decimal("0.65")),),
         category="science",
     )
 
@@ -310,9 +308,7 @@ class TestRiskScorerInit:
 class TestWeightedScoreCalculation:
     """Tests for weighted score calculation."""
 
-    def test_no_signals_zero_score(
-        self, mock_redis: AsyncMock, sample_trade: TradeEvent
-    ) -> None:
+    def test_no_signals_zero_score(self, mock_redis: AsyncMock, sample_trade: TradeEvent) -> None:
         """Test score is zero when no signals present."""
         scorer = RiskScorer(mock_redis)
         bundle = SignalBundle(trade_event=sample_trade)
@@ -358,10 +354,7 @@ class TestWeightedScoreCalculation:
         score, count = scorer.calculate_weighted_score(bundle)
 
         # 0.7 confidence * 0.35 weight + 0.7 * 0.25 niche weight = 0.42
-        expected = (
-            0.7 * DEFAULT_WEIGHTS["size_anomaly"]
-            + 0.7 * DEFAULT_WEIGHTS["niche_market"]
-        )
+        expected = 0.7 * DEFAULT_WEIGHTS["size_anomaly"] + 0.7 * DEFAULT_WEIGHTS["niche_market"]
         assert score == pytest.approx(expected)
         assert count == 1
 
@@ -559,9 +552,7 @@ class TestDeduplication:
     """Tests for deduplication functionality."""
 
     @pytest.mark.asyncio
-    async def test_check_and_set_dedup_new_key(
-        self, mock_redis: AsyncMock
-    ) -> None:
+    async def test_check_and_set_dedup_new_key(self, mock_redis: AsyncMock) -> None:
         """Test dedup returns False for new key."""
         mock_redis.set.return_value = True
 
@@ -572,9 +563,7 @@ class TestDeduplication:
         mock_redis.set.assert_called_once()
 
     @pytest.mark.asyncio
-    async def test_check_and_set_dedup_existing_key(
-        self, mock_redis: AsyncMock
-    ) -> None:
+    async def test_check_and_set_dedup_existing_key(self, mock_redis: AsyncMock) -> None:
         """Test dedup returns True for existing key."""
         mock_redis.set.return_value = False  # Key exists, NX failed
 
@@ -632,9 +621,7 @@ class TestBatchAnalysis:
                 confidence=0.8,
                 factors={},
             )
-            bundles.append(
-                SignalBundle(trade_event=trade, fresh_wallet_signal=signal)
-            )
+            bundles.append(SignalBundle(trade_event=trade, fresh_wallet_signal=signal))
 
         assessments = await scorer.assess_batch(bundles)
 

tests/detector/test_size_anomaly.py

@@ -290,9 +290,7 @@ class TestVolumeImpactCalculation:
         detector = SizeAnomalyDetector(mock_metadata_sync)
 
         # Trade size $1000, daily volume $50000 = 2% impact
-        impact = detector._calculate_volume_impact(
-            Decimal("1000"), Decimal("50000")
-        )
+        impact = detector._calculate_volume_impact(Decimal("1000"), Decimal("50000"))
         assert impact == pytest.approx(0.02)
 
     def test_volume_impact_none_volume(self, mock_metadata_sync: AsyncMock) -> None:
@@ -309,9 +307,7 @@ class TestVolumeImpactCalculation:
         impact = detector._calculate_volume_impact(Decimal("1000"), Decimal("0"))
         assert impact == 0.0
 
-    def test_volume_impact_negative_volume(
-        self, mock_metadata_sync: AsyncMock
-    ) -> None:
+    def test_volume_impact_negative_volume(self, mock_metadata_sync: AsyncMock) -> None:
         """Test volume impact returns 0 when volume is negative."""
         detector = SizeAnomalyDetector(mock_metadata_sync)
 
@@ -422,9 +418,7 @@ class TestNicheMarketDetection:
 class TestConfidenceScoring:
     """Tests for confidence score calculation."""
 
-    def test_confidence_volume_impact_only(
-        self, mock_metadata_sync: AsyncMock
-    ) -> None:
+    def test_confidence_volume_impact_only(self, mock_metadata_sync: AsyncMock) -> None:
         """Test confidence with only volume impact."""
         detector = SizeAnomalyDetector(mock_metadata_sync)
 
@@ -594,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,
@@ -819,9 +900,7 @@ class TestBatchAnalysis:
         assert len(signals) == 2
 
     @pytest.mark.asyncio
-    async def test_analyze_batch_empty_list(
-        self, mock_metadata_sync: AsyncMock
-    ) -> None:
+    async def test_analyze_batch_empty_list(self, mock_metadata_sync: AsyncMock) -> None:
         """Test batch analysis with empty list."""
         detector = SizeAnomalyDetector(mock_metadata_sync)
         signals = await detector.analyze_batch([])

tests/ingestor/test_clob_client.py

@@ -7,7 +7,6 @@ import pytest
 
 from polymarket_insider_tracker.ingestor.clob_client import (
     ClobClient,
-    ClobClientError,
     RateLimiter,
     RetryError,
     with_retry,
@@ -115,25 +114,23 @@ class TestClobClient:
     @pytest.fixture
     def mock_base_client(self) -> MagicMock:
         """Create a mock base CLOB client."""
-        with patch(
-            "polymarket_insider_tracker.ingestor.clob_client.BaseClobClient"
-        ) as mock:
+        with patch("polymarket_insider_tracker.ingestor.clob_client.BaseClobClient") as mock:
             yield mock.return_value
 
-    def test_init_defaults(self, mock_base_client: MagicMock) -> None:
+    def test_init_defaults(self, mock_base_client: MagicMock) -> None:  # noqa: ARG002
         """Test client initialization with defaults."""
         client = ClobClient()
 
         assert client._host == "https://clob.polymarket.com"
         assert client._max_retries == 3
 
-    def test_init_with_env_api_key(self, mock_base_client: MagicMock) -> None:
+    def test_init_with_env_api_key(self, mock_base_client: MagicMock) -> None:  # noqa: ARG002
         """Test client reads API key from environment."""
         with patch.dict("os.environ", {"POLYMARKET_API_KEY": "test-key"}):
             client = ClobClient()
             assert client._api_key == "test-key"
 
-    def test_init_with_explicit_api_key(self, mock_base_client: MagicMock) -> None:
+    def test_init_with_explicit_api_key(self, mock_base_client: MagicMock) -> None:  # noqa: ARG002
         """Test client uses explicitly provided API key."""
         client = ClobClient(api_key="explicit-key")
         assert client._api_key == "explicit-key"
@@ -256,15 +253,22 @@ class TestClobClient:
         assert len(market.tokens) == 2
 
     def test_get_market_not_found(self, mock_base_client: MagicMock) -> None:
-        """Test error handling when market not found."""
+        """Test error handling when market not found.
+
+        When the underlying API call fails, the @with_retry decorator will
+        retry the operation. After all retries are exhausted, it raises
+        RetryError wrapping the original exception.
+        """
         mock_base_client.get_market.side_effect = Exception("Not found")
 
         client = ClobClient()
 
-        with pytest.raises(ClobClientError) as exc_info:
+        with pytest.raises(RetryError) as exc_info:
             client.get_market("0xnotfound")
 
-        assert "0xnotfound" in str(exc_info.value)
+        # The RetryError wraps the original exception
+        assert "get_market" in str(exc_info.value)
+        assert exc_info.value.last_exception is not None
 
     def test_get_orderbook(self, mock_base_client: MagicMock) -> None:
         """Test fetching an orderbook."""

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_health.py

@@ -448,9 +448,7 @@ class TestHealthMonitorHTTPEndpoints:
             assert data["status"] == "unhealthy"
 
     @pytest.mark.asyncio
-    async def test_metrics_endpoint(
-        self, monitor: HealthMonitor, app: web.Application
-    ) -> None:
+    async def test_metrics_endpoint(self, monitor: HealthMonitor, app: web.Application) -> None:
         """Test /metrics endpoint returns Prometheus format."""
         from aiohttp.test_utils import TestClient, TestServer
 
@@ -468,9 +466,7 @@ class TestHealthMonitorHTTPEndpoints:
             assert "polymarket_health_status" in text
 
     @pytest.mark.asyncio
-    async def test_ready_endpoint_ready(
-        self, monitor: HealthMonitor, app: web.Application
-    ) -> None:
+    async def test_ready_endpoint_ready(self, monitor: HealthMonitor, app: web.Application) -> None:
         """Test /ready endpoint when ready."""
         from aiohttp.test_utils import TestClient, TestServer
 

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/ingestor/test_models.py

@@ -1,6 +1,6 @@
 """Tests for ingestor data models."""
 
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from decimal import Decimal
 from unittest.mock import MagicMock
 
@@ -75,7 +75,7 @@ class TestMarket:
         assert len(market.tokens) == 2
         assert market.tokens[0].outcome == "Yes"
         assert market.tokens[1].outcome == "No"
-        assert market.end_date == datetime(2024, 12, 31, 23, 59, 59, tzinfo=timezone.utc)
+        assert market.end_date == datetime(2024, 12, 31, 23, 59, 59, tzinfo=UTC)
         assert market.active is True
         assert market.closed is False
 
@@ -302,7 +302,7 @@ class TestTradeEvent:
         assert trade.outcome_index == 0
         assert trade.price == Decimal("0.65")
         assert trade.size == Decimal("100")
-        assert trade.timestamp == datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
+        assert trade.timestamp == datetime(2024, 1, 1, 0, 0, 0, tzinfo=UTC)
         assert trade.asset_id == "token123"
         assert trade.market_slug == "will-it-rain"
         assert trade.event_slug == "weather-markets"
@@ -364,7 +364,7 @@ class TestTradeEvent:
             outcome_index=0,
             price=Decimal("0.5"),
             size=Decimal("10"),
-            timestamp=datetime.now(timezone.utc),
+            timestamp=datetime.now(UTC),
             asset_id="",
         )
         sell_trade = TradeEvent(
@@ -376,7 +376,7 @@ class TestTradeEvent:
             outcome_index=0,
             price=Decimal("0.5"),
             size=Decimal("10"),
-            timestamp=datetime.now(timezone.utc),
+            timestamp=datetime.now(UTC),
             asset_id="",
         )
 
@@ -396,7 +396,7 @@ class TestTradeEvent:
             outcome_index=0,
             price=Decimal("0.65"),
             size=Decimal("100"),
-            timestamp=datetime.now(timezone.utc),
+            timestamp=datetime.now(UTC),
             asset_id="",
         )
 
@@ -413,7 +413,7 @@ class TestTradeEvent:
             outcome_index=0,
             price=Decimal("0.5"),
             size=Decimal("10"),
-            timestamp=datetime.now(timezone.utc),
+            timestamp=datetime.now(UTC),
             asset_id="token",
         )
         with pytest.raises(AttributeError):

tests/ingestor/test_websocket.py

@@ -3,7 +3,7 @@
 import asyncio
 import json
 from decimal import Decimal
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, patch
 
 import pytest
 
@@ -80,19 +80,16 @@ class TestTradeStreamHandler:
 
         assert handler._event_filter == "presidential-election-2024"
 
-    def test_build_subscription_message_no_filter(
-        self, handler: TradeStreamHandler
-    ) -> None:
+    def test_build_subscription_message_no_filter(self, handler: TradeStreamHandler) -> None:
         """Test building subscription message without filters."""
         msg = handler._build_subscription_message()
 
         assert msg == {
-            "subscriptions": [{"topic": "activity", "type": "trades"}]
+            "action": "subscribe",
+            "subscriptions": [{"topic": "activity", "type": "trades"}],
         }
 
-    def test_build_subscription_message_with_event_filter(
-        self, on_trade_mock: AsyncMock
-    ) -> None:
+    def test_build_subscription_message_with_event_filter(self, on_trade_mock: AsyncMock) -> None:
         """Test building subscription message with event filter."""
         handler = TradeStreamHandler(
             on_trade=on_trade_mock,
@@ -100,13 +97,9 @@ class TestTradeStreamHandler:
         )
         msg = handler._build_subscription_message()
 
-        assert msg["subscriptions"][0]["filters"] == json.dumps(
-            {"event_slug": "test-event"}
-        )
+        assert msg["subscriptions"][0]["filters"] == json.dumps({"event_slug": "test-event"})
 
-    def test_build_subscription_message_with_market_filter(
-        self, on_trade_mock: AsyncMock
-    ) -> None:
+    def test_build_subscription_message_with_market_filter(self, on_trade_mock: AsyncMock) -> None:
         """Test building subscription message with market filter."""
         handler = TradeStreamHandler(
             on_trade=on_trade_mock,
@@ -114,19 +107,16 @@ class TestTradeStreamHandler:
         )
         msg = handler._build_subscription_message()
 
-        assert msg["subscriptions"][0]["filters"] == json.dumps(
-            {"market_slug": "test-market"}
-        )
+        assert msg["subscriptions"][0]["filters"] == json.dumps({"market_slug": "test-market"})
 
     @pytest.mark.asyncio
     async def test_handle_message_trade(
         self, handler: TradeStreamHandler, on_trade_mock: AsyncMock
     ) -> None:
-        """Test handling a valid trade message."""
+        """Test handling a valid trade message (payload-based routing)."""
         message = json.dumps(
             {
-                "topic": "activity",
-                "type": "trades",
+                "connection_id": "abc123",
                 "payload": {
                     "conditionId": "0xmarket",
                     "transactionHash": "0xtx",
@@ -154,11 +144,10 @@ class TestTradeStreamHandler:
     async def test_handle_message_non_trade(
         self, handler: TradeStreamHandler, on_trade_mock: AsyncMock
     ) -> None:
-        """Test handling a non-trade message."""
+        """Test handling a non-trade message (no transactionHash/proxyWallet)."""
         message = json.dumps(
             {
-                "topic": "comments",
-                "type": "comment_created",
+                "connection_id": "abc123",
                 "payload": {"body": "Hello"},
             }
         )
@@ -168,6 +157,35 @@ class TestTradeStreamHandler:
         on_trade_mock.assert_not_called()
         assert handler.stats.trades_received == 0
 
+    @pytest.mark.asyncio
+    async def test_handle_message_payload_missing_proxy_wallet(
+        self, handler: TradeStreamHandler, on_trade_mock: AsyncMock
+    ) -> None:
+        """Ratchet: payload with transactionHash but no proxyWallet is not a trade."""
+        message = json.dumps(
+            {
+                "connection_id": "abc",
+                "payload": {"transactionHash": "0xtx", "other": "field"},
+            }
+        )
+
+        await handler._handle_message(message)
+
+        on_trade_mock.assert_not_called()
+        assert handler.stats.trades_received == 0
+
+    @pytest.mark.asyncio
+    async def test_handle_message_no_payload_key(
+        self, handler: TradeStreamHandler, on_trade_mock: AsyncMock
+    ) -> None:
+        """Ratchet: message without payload key is ignored."""
+        message = json.dumps({"connection_id": "abc", "status": "ok"})
+
+        await handler._handle_message(message)
+
+        on_trade_mock.assert_not_called()
+        assert handler.stats.trades_received == 0
+
     @pytest.mark.asyncio
     async def test_handle_message_invalid_json(
         self, handler: TradeStreamHandler, on_trade_mock: AsyncMock
@@ -187,8 +205,7 @@ class TestTradeStreamHandler:
 
         message = json.dumps(
             {
-                "topic": "activity",
-                "type": "trades",
+                "connection_id": "abc",
                 "payload": {
                     "conditionId": "0x",
                     "transactionHash": "0x",
@@ -235,28 +252,32 @@ class TestTradeStreamHandler:
 
     @pytest.mark.asyncio
     async def test_connect_sends_subscription(
-        self, handler: TradeStreamHandler, on_state_change_mock: AsyncMock
+        self,
+        handler: TradeStreamHandler,
+        on_state_change_mock: AsyncMock,  # noqa: ARG002
     ) -> None:
         """Test that connection sends subscription message."""
         mock_ws = AsyncMock()
         mock_ws.send = AsyncMock()
 
-        with patch("websockets.connect", AsyncMock(return_value=mock_ws)):
+        with patch(
+            "polymarket_insider_tracker.ingestor.websocket.ws_connect",
+            AsyncMock(return_value=mock_ws),
+        ):
             ws = await handler._connect()
 
             assert ws is mock_ws
             mock_ws.send.assert_called_once()
 
-            # Verify subscription message
+            # Verify subscription message includes action: subscribe
             sent_msg = json.loads(mock_ws.send.call_args[0][0])
+            assert sent_msg["action"] == "subscribe"
             assert "subscriptions" in sent_msg
             assert sent_msg["subscriptions"][0]["topic"] == "activity"
             assert sent_msg["subscriptions"][0]["type"] == "trades"
 
     @pytest.mark.asyncio
-    async def test_cleanup_closes_websocket(
-        self, handler: TradeStreamHandler
-    ) -> None:
+    async def test_cleanup_closes_websocket(self, handler: TradeStreamHandler) -> None:
         """Test that cleanup closes the WebSocket."""
         mock_ws = AsyncMock()
         mock_ws.close = AsyncMock()
@@ -299,15 +320,9 @@ class TestTradeStreamHandlerIntegration:
             initial_reconnect_delay=0.01,
         )
 
-        # Create mock WebSocket that sends one trade then closes
-        mock_ws = MagicMock()
-        mock_ws.send = AsyncMock()
-        mock_ws.close = AsyncMock()
-
         trade_message = json.dumps(
             {
-                "topic": "activity",
-                "type": "trades",
+                "connection_id": "test-conn",
                 "payload": {
                     "conditionId": "0xtest",
                     "transactionHash": "0xtx",
@@ -322,18 +337,42 @@ class TestTradeStreamHandlerIntegration:
             }
         )
 
-        # Mock async iteration
-        async def mock_iter() -> None:
-            yield trade_message
-            await handler.stop()  # Stop after first message
+        # Create a proper async iterable mock WebSocket
+        class MockWebSocket:
+            """Mock WebSocket that yields one message then stops."""
 
-        mock_ws.__aiter__ = mock_iter
+            def __init__(self, handler: TradeStreamHandler, message: str):
+                self.handler = handler
+                self.message = message
+                self.sent = False
 
-        with patch("websockets.connect", AsyncMock(return_value=mock_ws)):
+            async def send(self, _msg: str) -> None:
+                pass
+
+            async def close(self) -> None:
+                pass
+
+            def __aiter__(self):
+                return self
+
+            async def __anext__(self) -> str:
+                if not self.sent:
+                    self.sent = True
+                    return self.message
+                # Stop the handler and raise StopAsyncIteration
+                await self.handler.stop()
+                raise StopAsyncIteration
+
+        mock_ws = MockWebSocket(handler, trade_message)
+
+        with patch(
+            "polymarket_insider_tracker.ingestor.websocket.ws_connect",
+            AsyncMock(return_value=mock_ws),
+        ):
             # Run with timeout to prevent hanging
             try:
                 await asyncio.wait_for(handler.start(), timeout=1.0)
-            except asyncio.TimeoutError:
+            except TimeoutError:
                 await handler.stop()
 
         # Verify trade was received

tests/profiler/test_analyzer.py

@@ -144,9 +144,7 @@ class TestWalletAnalyzerAnalyze:
         assert profile.age_hours is None
 
     @pytest.mark.asyncio
-    async def test_analyze_uses_cache(
-        self, mock_client: AsyncMock, mock_redis: AsyncMock
-    ) -> None:
+    async def test_analyze_uses_cache(self, mock_client: AsyncMock, mock_redis: AsyncMock) -> None:
         """Test that analyze uses cached data."""
         cached_data = {
             "address": VALID_ADDRESS.lower(),
@@ -164,6 +162,7 @@ class TestWalletAnalyzerAnalyze:
 
         # Actually mock it properly with json
         import json
+
         mock_redis.get = AsyncMock(return_value=json.dumps(cached_data).encode())
 
         analyzer = WalletAnalyzer(mock_client, redis=mock_redis)

tests/profiler/test_chain.py

@@ -166,9 +166,7 @@ class TestPolygonClient:
 
         await client._set_cached("test:key", "value")
 
-        mock_redis.set.assert_called_once_with(
-            "test:key", "value", ex=DEFAULT_CACHE_TTL_SECONDS
-        )
+        mock_redis.set.assert_called_once_with("test:key", "value", ex=DEFAULT_CACHE_TTL_SECONDS)
 
     @pytest.mark.asyncio
     async def test_set_cached_custom_ttl(self, mock_redis: AsyncMock) -> None:
@@ -274,9 +272,7 @@ class TestPolygonClient:
             assert info.first_transaction is None
 
     @pytest.mark.asyncio
-    async def test_get_first_transaction_no_transactions(
-        self, mock_redis: AsyncMock
-    ) -> None:
+    async def test_get_first_transaction_no_transactions(self, mock_redis: AsyncMock) -> None:
         """Test get_first_transaction when wallet has no transactions."""
         client = PolygonClient("https://polygon-rpc.com", redis=mock_redis)
 
@@ -451,9 +447,7 @@ class TestPolygonClientTokenBalance:
 
         # Mock the contract call
         mock_contract = MagicMock()
-        mock_contract.functions.balanceOf.return_value.call = AsyncMock(
-            return_value=5000000
-        )
+        mock_contract.functions.balanceOf.return_value.call = AsyncMock(return_value=5000000)
         client._w3.eth.contract = MagicMock(return_value=mock_contract)
 
         balance = await client.get_token_balance(VALID_ADDRESS, VALID_TOKEN)

tests/profiler/test_entities.py

@@ -59,9 +59,7 @@ class TestEntityData:
         """Test that CEX addresses are populated."""
         assert len(CEX_ADDRESSES) > 0
         # Check Binance address is present
-        binance_found = any(
-            entity == EntityType.CEX_BINANCE for entity in CEX_ADDRESSES.values()
-        )
+        binance_found = any(entity == EntityType.CEX_BINANCE for entity in CEX_ADDRESSES.values())
         assert binance_found
 
     def test_bridge_addresses_populated(self) -> None:
@@ -72,9 +70,7 @@ class TestEntityData:
         """Test that DEX addresses are populated."""
         assert len(DEX_ADDRESSES) > 0
         # Check Uniswap is present
-        uniswap_found = any(
-            entity == EntityType.DEX_UNISWAP for entity in DEX_ADDRESSES.values()
-        )
+        uniswap_found = any(entity == EntityType.DEX_UNISWAP for entity in DEX_ADDRESSES.values())
         assert uniswap_found
 
     def test_token_addresses_include_usdc(self) -> None:

tests/profiler/test_funding.py

@@ -74,27 +74,19 @@ class TestFundingTracerInit:
         tracer = FundingTracer(mock_polygon_client, max_hops=5)
         assert tracer.max_hops == 5
 
-    def test_init_with_custom_usdc_addresses(
-        self, mock_polygon_client: MagicMock
-    ) -> None:
+    def test_init_with_custom_usdc_addresses(self, mock_polygon_client: MagicMock) -> None:
         """Test initialization with custom USDC addresses."""
         custom_addresses = ["0x1111111111111111111111111111111111111111"]
-        tracer = FundingTracer(
-            mock_polygon_client, usdc_addresses=custom_addresses
-        )
+        tracer = FundingTracer(mock_polygon_client, usdc_addresses=custom_addresses)
         assert tracer._usdc_addresses == [custom_addresses[0].lower()]
 
-    def test_init_with_custom_entity_registry(
-        self, mock_polygon_client: MagicMock
-    ) -> None:
+    def test_init_with_custom_entity_registry(self, mock_polygon_client: MagicMock) -> None:
         """Test initialization with custom entity registry."""
         registry = EntityRegistry()
         tracer = FundingTracer(mock_polygon_client, entity_registry=registry)
         assert tracer.entity_registry is registry
 
-    def test_init_creates_default_entity_registry(
-        self, mock_polygon_client: MagicMock
-    ) -> None:
+    def test_init_creates_default_entity_registry(self, mock_polygon_client: MagicMock) -> None:
         """Test initialization creates default EntityRegistry if None."""
         tracer = FundingTracer(mock_polygon_client, entity_registry=None)
         assert isinstance(tracer.entity_registry, EntityRegistry)
@@ -186,9 +178,7 @@ class TestFundingTracerTrace:
 
         call_count = 0
 
-        async def mock_get_logs(
-            *_args: Any, **_kwargs: Any
-        ) -> list[dict[str, Any]]:
+        async def mock_get_logs(*_args: Any, **_kwargs: Any) -> list[dict[str, Any]]:
             nonlocal call_count
             result = [mock_logs[call_count]] if call_count < len(mock_logs) else []
             call_count += 1
@@ -213,9 +203,7 @@ class TestFundingTracerTrace:
 
         call_count = 0
 
-        async def mock_get_logs(
-            *_args: Any, **_kwargs: Any
-        ) -> list[dict[str, Any]]:
+        async def mock_get_logs(*_args: Any, **_kwargs: Any) -> list[dict[str, Any]]:
             nonlocal call_count
             if call_count < len(wallets) - 1:
                 log = _create_mock_log(
@@ -286,9 +274,7 @@ class TestGetFirstUsdcTransfer:
         """Test fallback to native USDC contract."""
         call_count = 0
 
-        async def mock_get_logs(
-            *_args: Any, **_kwargs: Any
-        ) -> list[dict[str, Any]]:
+        async def mock_get_logs(*_args: Any, **_kwargs: Any) -> list[dict[str, Any]]:
             nonlocal call_count
             call_count += 1
             if call_count == 1:  # First call (bridged) returns nothing
@@ -341,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()
@@ -348,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(
@@ -369,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
@@ -388,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."""
@@ -410,9 +639,7 @@ class TestLogToFundingTransfer:
             block_number=50000000,
         )
 
-        result = await funding_tracer._log_to_funding_transfer(
-            mock_log, USDC_BRIDGED
-        )
+        result = await funding_tracer._log_to_funding_transfer(mock_log, USDC_BRIDGED)
 
         assert result.from_address == TEST_SOURCE.lower()
         assert result.to_address == TEST_WALLET.lower()
@@ -438,9 +665,7 @@ class TestLogToFundingTransfer:
             block_number=50000000,
         )
 
-        result = await funding_tracer._log_to_funding_transfer(
-            mock_log, USDC_BRIDGED
-        )
+        result = await funding_tracer._log_to_funding_transfer(mock_log, USDC_BRIDGED)
 
         # Should still return a valid transfer with current timestamp
         assert result.from_address == TEST_SOURCE.lower()
@@ -460,7 +685,9 @@ class TestGetFundingChainsBatch:
 
         # Mock trace to return simple chains
         async def mock_trace(
-            addr: str, *, max_hops: int | None = None  # noqa: ARG001
+            addr: str,
+            *,
+            max_hops: int | None = None,  # noqa: ARG001
         ) -> FundingChain:
             return FundingChain(
                 target_address=addr.lower(),
@@ -486,7 +713,9 @@ class TestGetFundingChainsBatch:
         call_count = 0
 
         async def mock_trace(
-            addr: str, *, max_hops: int | None = None  # noqa: ARG001
+            addr: str,
+            *,
+            max_hops: int | None = None,  # noqa: ARG001
         ) -> FundingChain:
             nonlocal call_count
             call_count += 1
@@ -525,9 +754,7 @@ class TestGetFundingChainsBatch:
         addresses = ["0x" + "11" * 20]
         captured_max_hops: list[int | None] = []
 
-        async def mock_trace(
-            addr: str, max_hops: int | None = None
-        ) -> FundingChain:
+        async def mock_trace(addr: str, max_hops: int | None = None) -> FundingChain:
             captured_max_hops.append(max_hops)
             return FundingChain(target_address=addr.lower())
 
@@ -565,9 +792,7 @@ class TestGetSuspiciousnessScore:
 
         assert score == 0.3
 
-    def test_unknown_no_transfers_high_score(
-        self, funding_tracer: FundingTracer
-    ) -> None:
+    def test_unknown_no_transfers_high_score(self, funding_tracer: FundingTracer) -> None:
         """Test unknown origin with no transfers is most suspicious."""
         chain = FundingChain(
             target_address=TEST_WALLET,
@@ -579,9 +804,7 @@ class TestGetSuspiciousnessScore:
 
         assert score == 1.0
 
-    def test_unknown_max_hops_high_score(
-        self, funding_tracer: FundingTracer
-    ) -> None:
+    def test_unknown_max_hops_high_score(self, funding_tracer: FundingTracer) -> None:
         """Test unknown origin at max hops is suspicious."""
         chain = FundingChain(
             target_address=TEST_WALLET,
@@ -593,9 +816,7 @@ class TestGetSuspiciousnessScore:
 
         assert score == 0.7
 
-    def test_unknown_partial_hops_medium_score(
-        self, funding_tracer: FundingTracer
-    ) -> None:
+    def test_unknown_partial_hops_medium_score(self, funding_tracer: FundingTracer) -> None:
         """Test unknown origin with partial hops is moderately suspicious."""
         chain = FundingChain(
             target_address=TEST_WALLET,

tests/profiler/test_models.py

@@ -160,9 +160,7 @@ class TestWalletInfo:
         """Test wallet age when no first transaction."""
         assert sample_wallet.wallet_age_days is None
 
-    def test_wallet_age_days_with_transaction(
-        self, wallet_with_transaction: WalletInfo
-    ) -> None:
+    def test_wallet_age_days_with_transaction(self, wallet_with_transaction: WalletInfo) -> None:
         """Test wallet age calculation."""
         age = wallet_with_transaction.wallet_age_days
 

tests/test_config.py

@@ -0,0 +1,329 @@
+"""Tests for configuration management service."""
+
+from __future__ import annotations
+
+import os
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
+import pytest
+from pydantic import ValidationError
+
+from polymarket_insider_tracker.config import (
+    DatabaseSettings,
+    DiscordSettings,
+    PolygonSettings,
+    PolymarketSettings,
+    RedisSettings,
+    Settings,
+    TelegramSettings,
+    clear_settings_cache,
+    get_settings,
+)
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+
+@pytest.fixture(autouse=True)
+def clear_cache() -> Iterator[None]:
+    """Clear settings cache before and after each test."""
+    clear_settings_cache()
+    yield
+    clear_settings_cache()
+
+
+class TestDatabaseSettings:
+    """Tests for DatabaseSettings."""
+
+    def test_valid_postgresql_url(self) -> None:
+        """Test valid PostgreSQL URL."""
+        with patch.dict(os.environ, {"DATABASE_URL": "postgresql://user:pass@localhost/db"}):
+            settings = DatabaseSettings()
+            assert settings.url == "postgresql://user:pass@localhost/db"
+
+    def test_valid_asyncpg_url(self) -> None:
+        """Test valid asyncpg URL."""
+        with patch.dict(
+            os.environ, {"DATABASE_URL": "postgresql+asyncpg://user:pass@localhost/db"}
+        ):
+            settings = DatabaseSettings()
+            assert settings.url == "postgresql+asyncpg://user:pass@localhost/db"
+
+    def test_invalid_url_raises(self) -> None:
+        """Test that invalid database URL raises validation error."""
+        with (
+            patch.dict(os.environ, {"DATABASE_URL": "mysql://user:pass@localhost/db"}),
+            pytest.raises(ValidationError, match="PostgreSQL connection string"),
+        ):
+            DatabaseSettings()
+
+
+class TestRedisSettings:
+    """Tests for RedisSettings."""
+
+    def test_default_url(self) -> None:
+        """Test default Redis URL."""
+        with patch.dict(os.environ, {}, clear=True):
+            settings = RedisSettings()
+            assert settings.url == "redis://localhost:6379"
+
+    def test_custom_url(self) -> None:
+        """Test custom Redis URL."""
+        with patch.dict(os.environ, {"REDIS_URL": "redis://redis:6380"}):
+            settings = RedisSettings()
+            assert settings.url == "redis://redis:6380"
+
+    def test_invalid_url_raises(self) -> None:
+        """Test that invalid Redis URL raises validation error."""
+        with (
+            patch.dict(os.environ, {"REDIS_URL": "http://localhost:6379"}),
+            pytest.raises(ValidationError, match="redis://"),
+        ):
+            RedisSettings()
+
+
+class TestPolygonSettings:
+    """Tests for PolygonSettings."""
+
+    def test_default_rpc_url(self) -> None:
+        """Test default Polygon RPC URL."""
+        with patch.dict(os.environ, {}, clear=True):
+            settings = PolygonSettings()
+            assert settings.rpc_url == "https://polygon-rpc.com"
+            assert settings.fallback_rpc_url is None
+
+    def test_custom_urls(self) -> None:
+        """Test custom Polygon RPC URLs."""
+        with patch.dict(
+            os.environ,
+            {
+                "POLYGON_RPC_URL": "https://alchemy.io/polygon",
+                "POLYGON_FALLBACK_RPC_URL": "https://backup.polygon.io",
+            },
+        ):
+            settings = PolygonSettings()
+            assert settings.rpc_url == "https://alchemy.io/polygon"
+            assert settings.fallback_rpc_url == "https://backup.polygon.io"
+
+    def test_invalid_url_raises(self) -> None:
+        """Test that invalid RPC URL raises validation error."""
+        with (
+            patch.dict(os.environ, {"POLYGON_RPC_URL": "ws://polygon.io"}),
+            pytest.raises(ValidationError, match="HTTP"),
+        ):
+            PolygonSettings()
+
+
+class TestPolymarketSettings:
+    """Tests for PolymarketSettings."""
+
+    def test_default_ws_url(self) -> None:
+        """Test default Polymarket WebSocket URL."""
+        with patch.dict(os.environ, {}, clear=True):
+            settings = PolymarketSettings()
+            assert "polymarket.com" in settings.ws_url
+            assert settings.api_key is None
+
+    def test_custom_api_key(self) -> None:
+        """Test custom API key (secret)."""
+        with patch.dict(os.environ, {"POLYMARKET_API_KEY": "secret-key-123"}):
+            settings = PolymarketSettings()
+            assert settings.api_key is not None
+            assert settings.api_key.get_secret_value() == "secret-key-123"
+
+    def test_invalid_ws_url_raises(self) -> None:
+        """Test that invalid WebSocket URL raises validation error."""
+        with (
+            patch.dict(os.environ, {"POLYMARKET_WS_URL": "http://polymarket.com"}),
+            pytest.raises(ValidationError, match="ws://"),
+        ):
+            PolymarketSettings()
+
+
+class TestDiscordSettings:
+    """Tests for DiscordSettings."""
+
+    def test_disabled_by_default(self) -> None:
+        """Test Discord is disabled when no webhook URL."""
+        with patch.dict(os.environ, {}, clear=True):
+            settings = DiscordSettings()
+            assert not settings.enabled
+            assert settings.webhook_url is None
+
+    def test_enabled_with_webhook(self) -> None:
+        """Test Discord is enabled with webhook URL."""
+        with patch.dict(os.environ, {"DISCORD_WEBHOOK_URL": "https://discord.com/webhook/123"}):
+            settings = DiscordSettings()
+            assert settings.enabled
+            assert settings.webhook_url is not None
+
+
+class TestTelegramSettings:
+    """Tests for TelegramSettings."""
+
+    def test_disabled_by_default(self) -> None:
+        """Test Telegram is disabled when no credentials."""
+        with patch.dict(os.environ, {}, clear=True):
+            settings = TelegramSettings()
+            assert not settings.enabled
+
+    def test_disabled_with_partial_config(self) -> None:
+        """Test Telegram is disabled with only token or chat_id."""
+        with patch.dict(os.environ, {"TELEGRAM_BOT_TOKEN": "token123"}):
+            settings = TelegramSettings()
+            assert not settings.enabled
+
+        with patch.dict(os.environ, {"TELEGRAM_CHAT_ID": "12345"}):
+            settings = TelegramSettings()
+            assert not settings.enabled
+
+    def test_enabled_with_full_config(self) -> None:
+        """Test Telegram is enabled with both token and chat_id."""
+        with patch.dict(
+            os.environ,
+            {
+                "TELEGRAM_BOT_TOKEN": "token123",
+                "TELEGRAM_CHAT_ID": "12345",
+            },
+        ):
+            settings = TelegramSettings()
+            assert settings.enabled
+
+
+class TestSettings:
+    """Tests for main Settings class."""
+
+    def test_loads_with_required_vars(self) -> None:
+        """Test settings load with required environment variables."""
+        with patch.dict(
+            os.environ,
+            {
+                "DATABASE_URL": "postgresql://user:pass@localhost/db",
+                "REDIS_URL": "redis://localhost:6379",
+            },
+        ):
+            settings = Settings()
+            assert settings.database.url == "postgresql://user:pass@localhost/db"
+            assert settings.redis.url == "redis://localhost:6379"
+
+    def test_default_log_level(self) -> None:
+        """Test default log level is INFO."""
+        with patch.dict(
+            os.environ,
+            {"DATABASE_URL": "postgresql://user:pass@localhost/db"},
+        ):
+            settings = Settings()
+            assert settings.log_level == "INFO"
+
+    def test_custom_log_level(self) -> None:
+        """Test custom log level."""
+        with patch.dict(
+            os.environ,
+            {
+                "DATABASE_URL": "postgresql://user:pass@localhost/db",
+                "LOG_LEVEL": "DEBUG",
+            },
+        ):
+            settings = Settings()
+            assert settings.log_level == "DEBUG"
+
+    def test_invalid_log_level_raises(self) -> None:
+        """Test invalid log level raises validation error."""
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "DATABASE_URL": "postgresql://user:pass@localhost/db",
+                    "LOG_LEVEL": "TRACE",
+                },
+            ),
+            pytest.raises(ValidationError),
+        ):
+            Settings()
+
+    def test_health_port_validation(self) -> None:
+        """Test health port must be valid port number."""
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "DATABASE_URL": "postgresql://user:pass@localhost/db",
+                    "HEALTH_PORT": "99999",
+                },
+            ),
+            pytest.raises(ValidationError, match="65535"),
+        ):
+            Settings()
+
+    def test_get_logging_level(self) -> None:
+        """Test get_logging_level returns numeric level."""
+        import logging
+
+        with patch.dict(
+            os.environ,
+            {
+                "DATABASE_URL": "postgresql://user:pass@localhost/db",
+                "LOG_LEVEL": "WARNING",
+            },
+        ):
+            settings = Settings()
+            assert settings.get_logging_level() == logging.WARNING
+
+    def test_redacted_summary(self) -> None:
+        """Test redacted_summary masks sensitive data."""
+        with patch.dict(
+            os.environ,
+            {
+                "DATABASE_URL": "postgresql://user:secretpass@localhost/db",
+                "REDIS_URL": "redis://localhost:6379",
+            },
+        ):
+            settings = Settings()
+            summary = settings.redacted_summary()
+
+            # Database password should be redacted
+            db_url = summary["database_url"]
+            assert isinstance(db_url, str)
+            assert "secretpass" not in db_url
+            assert "***" in db_url
+            assert "user" in db_url
+
+
+class TestGetSettings:
+    """Tests for get_settings singleton."""
+
+    def test_returns_same_instance(self) -> None:
+        """Test get_settings returns cached instance."""
+        with patch.dict(
+            os.environ,
+            {"DATABASE_URL": "postgresql://user:pass@localhost/db"},
+        ):
+            settings1 = get_settings()
+            settings2 = get_settings()
+            assert settings1 is settings2
+
+    def test_clear_cache_allows_reload(self) -> None:
+        """Test clear_settings_cache allows reloading settings."""
+        with patch.dict(
+            os.environ,
+            {
+                "DATABASE_URL": "postgresql://user:pass@localhost/db",
+                "LOG_LEVEL": "INFO",
+            },
+        ):
+            settings1 = get_settings()
+            assert settings1.log_level == "INFO"
+
+        clear_settings_cache()
+
+        with patch.dict(
+            os.environ,
+            {
+                "DATABASE_URL": "postgresql://user:pass@localhost/db",
+                "LOG_LEVEL": "DEBUG",
+            },
+        ):
+            settings2 = get_settings()
+            assert settings2.log_level == "DEBUG"
+            assert settings1 is not settings2

tests/test_main.py

@@ -0,0 +1,214 @@
+"""Tests for the CLI entry point."""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+
+from polymarket_insider_tracker.__main__ import (
+    EXIT_CONFIG_ERROR,
+    EXIT_SUCCESS,
+    configure_logging,
+    create_parser,
+    main,
+    print_banner,
+    run_config_check,
+    validate_config,
+)
+
+
+class TestCreateParser:
+    """Tests for argument parser creation."""
+
+    def test_parser_has_version(self):
+        """Parser should have version flag."""
+        parser = create_parser()
+        with pytest.raises(SystemExit) as exc_info:
+            parser.parse_args(["--version"])
+        assert exc_info.value.code == 0
+
+    def test_parser_config_check(self):
+        """Parser should accept --config-check flag."""
+        parser = create_parser()
+        args = parser.parse_args(["--config-check"])
+        assert args.config_check is True
+
+    def test_parser_log_level(self):
+        """Parser should accept --log-level option."""
+        parser = create_parser()
+        args = parser.parse_args(["--log-level", "DEBUG"])
+        assert args.log_level == "DEBUG"
+
+    def test_parser_dry_run(self):
+        """Parser should accept --dry-run flag."""
+        parser = create_parser()
+        args = parser.parse_args(["--dry-run"])
+        assert args.dry_run is True
+
+    def test_parser_health_port(self):
+        """Parser should accept --health-port option."""
+        parser = create_parser()
+        args = parser.parse_args(["--health-port", "9090"])
+        assert args.health_port == 9090
+
+    def test_parser_default_values(self):
+        """Parser should have correct defaults."""
+        parser = create_parser()
+        args = parser.parse_args([])
+        assert args.config_check is False
+        assert args.log_level is None
+        assert args.dry_run is False
+        assert args.health_port is None
+
+
+class TestConfigureLogging:
+    """Tests for logging configuration."""
+
+    def test_configure_logging_info(self):
+        """Should configure logging at INFO level."""
+        configure_logging("INFO")
+        import logging
+
+        assert logging.getLogger().level == logging.INFO
+
+    def test_configure_logging_debug(self):
+        """Should configure logging at DEBUG level."""
+        configure_logging("DEBUG")
+        import logging
+
+        assert logging.getLogger().level == logging.DEBUG
+
+
+class TestPrintBanner:
+    """Tests for banner printing."""
+
+    def test_banner_contains_app_name(self, capsys):
+        """Banner should contain application name."""
+        print_banner()
+        captured = capsys.readouterr()
+        assert "Polymarket Insider Tracker" in captured.out
+
+    def test_banner_contains_version(self, capsys):
+        """Banner should contain version."""
+        print_banner()
+        captured = capsys.readouterr()
+        assert "v0.1.0" in captured.out
+
+
+class TestValidateConfig:
+    """Tests for configuration validation."""
+
+    def test_validate_config_success(self, monkeypatch):
+        """Should return settings on valid config."""
+        monkeypatch.setenv("DATABASE_URL", "postgresql://localhost/test")
+
+        settings = validate_config()
+        assert settings is not None
+
+    def test_validate_config_failure(self, monkeypatch, capsys):
+        """Should return None on invalid config."""
+        # Clear any existing DATABASE_URL
+        monkeypatch.delenv("DATABASE_URL", raising=False)
+
+        settings = validate_config()
+        assert settings is None
+
+        captured = capsys.readouterr()
+        assert "Configuration validation failed" in captured.err
+
+
+class TestRunConfigCheck:
+    """Tests for config check mode."""
+
+    def test_config_check_prints_summary(self, monkeypatch, capsys):
+        """Config check should print configuration summary."""
+        monkeypatch.setenv("DATABASE_URL", "postgresql://localhost/test")
+
+        settings = validate_config()
+        assert settings is not None
+
+        result = run_config_check(settings)
+        assert result == EXIT_SUCCESS
+
+        captured = capsys.readouterr()
+        assert "Configuration is valid!" in captured.out
+        assert "Configuration:" in captured.out
+
+
+class TestMain:
+    """Tests for main entry point."""
+
+    def test_main_with_config_check(self, monkeypatch):
+        """Main should exit successfully with --config-check."""
+        monkeypatch.setenv("DATABASE_URL", "postgresql://localhost/test")
+
+        with pytest.raises(SystemExit) as exc_info:
+            main(["--config-check"])
+
+        assert exc_info.value.code == EXIT_SUCCESS
+
+    def test_main_with_invalid_config(self, monkeypatch):
+        """Main should exit with config error on invalid config."""
+        monkeypatch.delenv("DATABASE_URL", raising=False)
+
+        with pytest.raises(SystemExit) as exc_info:
+            main([])
+
+        assert exc_info.value.code == EXIT_CONFIG_ERROR
+
+    def test_main_with_dry_run_and_config_check(self, monkeypatch):
+        """Main should handle dry-run with config-check."""
+        monkeypatch.setenv("DATABASE_URL", "postgresql://localhost/test")
+
+        with pytest.raises(SystemExit) as exc_info:
+            main(["--config-check", "--dry-run"])
+
+        assert exc_info.value.code == EXIT_SUCCESS
+
+    @patch("polymarket_insider_tracker.__main__.run_pipeline")
+    @patch("polymarket_insider_tracker.__main__.asyncio.run")
+    def test_main_runs_pipeline(self, mock_asyncio_run, _mock_run_pipeline, monkeypatch):
+        """Main should run pipeline when not in config-check mode."""
+        monkeypatch.setenv("DATABASE_URL", "postgresql://localhost/test")
+        mock_asyncio_run.return_value = EXIT_SUCCESS
+
+        with pytest.raises(SystemExit) as exc_info:
+            main([])
+
+        assert exc_info.value.code == EXIT_SUCCESS
+        mock_asyncio_run.assert_called_once()
+
+
+class TestIntegration:
+    """Integration tests for CLI invocation."""
+
+    def test_cli_help_option(self, capsys):
+        """CLI should display help with -h option."""
+        with pytest.raises(SystemExit) as exc_info:
+            main(["-h"])
+
+        assert exc_info.value.code == 0
+        captured = capsys.readouterr()
+        assert "polymarket-insider-tracker" in captured.out
+        assert "--config-check" in captured.out
+        assert "--dry-run" in captured.out
+        assert "--log-level" in captured.out
+
+    def test_cli_version_option(self, capsys):
+        """CLI should display version with --version option."""
+        with pytest.raises(SystemExit) as exc_info:
+            main(["--version"])
+
+        assert exc_info.value.code == 0
+        captured = capsys.readouterr()
+        assert "0.1.0" in captured.out
+
+    def test_cli_invalid_log_level(self, capsys):
+        """CLI should reject invalid log level."""
+        with pytest.raises(SystemExit) as exc_info:
+            main(["--log-level", "INVALID"])
+
+        assert exc_info.value.code != 0
+        captured = capsys.readouterr()
+        assert "invalid choice" in captured.err

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

@@ -0,0 +1,398 @@
+"""Tests for the main pipeline orchestrator."""
+
+from __future__ import annotations
+
+import asyncio
+from datetime import UTC, datetime
+from decimal import Decimal
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from polymarket_insider_tracker.config import Settings
+from polymarket_insider_tracker.detector.models import FreshWalletSignal
+from polymarket_insider_tracker.detector.scorer import SignalBundle
+from polymarket_insider_tracker.ingestor.models import TradeEvent
+from polymarket_insider_tracker.pipeline import Pipeline, PipelineState
+from polymarket_insider_tracker.profiler.models import WalletProfile
+
+
+@pytest.fixture
+def mock_settings():
+    """Create mock settings for testing."""
+    # Create nested mock objects
+    redis = MagicMock()
+    redis.url = "redis://localhost:6379"
+
+    database = MagicMock()
+    database.url = "postgresql+asyncpg://user:pass@localhost/db"
+
+    polygon = MagicMock()
+    polygon.rpc_url = "https://polygon-rpc.com"
+    polygon.fallback_rpc_url = None
+
+    polymarket = MagicMock()
+    polymarket.ws_url = "wss://ws-subscriptions-clob.polymarket.com/ws/market"
+    polymarket.api_key = None
+
+    discord = MagicMock()
+    discord.enabled = False
+    discord.webhook_url = None
+
+    telegram = MagicMock()
+    telegram.enabled = False
+    telegram.bot_token = None
+    telegram.chat_id = None
+
+    detector = MagicMock()
+    detector.persist_assessments = False
+
+    settings = MagicMock(spec=Settings)
+    settings.redis = redis
+    settings.database = database
+    settings.polygon = polygon
+    settings.polymarket = polymarket
+    settings.discord = discord
+    settings.telegram = telegram
+    settings.detector = detector
+    settings.dry_run = True
+    return settings
+
+
+@pytest.fixture
+def sample_trade_event():
+    """Create a sample trade event for testing."""
+    return TradeEvent(
+        trade_id="0x" + "a" * 64,
+        wallet_address="0x" + "b" * 40,
+        market_id="0x" + "c" * 64,
+        asset_id="asset_123",
+        side="BUY",
+        price=Decimal("0.65"),
+        size=Decimal("5000"),
+        timestamp=datetime.now(UTC),
+        outcome="Yes",
+        outcome_index=0,
+        event_title="Test Market",
+        market_slug="test-market",
+    )
+
+
+@pytest.fixture
+def sample_wallet_profile():
+    """Create a sample wallet profile for testing."""
+    return WalletProfile(
+        address="0x" + "b" * 40,
+        nonce=2,
+        first_seen=datetime.now(UTC),
+        age_hours=1.5,
+        is_fresh=True,
+        total_tx_count=2,
+        matic_balance=Decimal("100"),
+        usdc_balance=Decimal("5000"),
+        fresh_threshold=5,
+    )
+
+
+class TestPipelineState:
+    """Tests for pipeline state management."""
+
+    def test_initial_state_is_stopped(self, mock_settings):
+        """Pipeline should start in stopped state."""
+        pipeline = Pipeline(mock_settings)
+        assert pipeline.state == PipelineState.STOPPED
+
+    def test_is_running_property(self, mock_settings):
+        """is_running property should reflect state."""
+        pipeline = Pipeline(mock_settings)
+        assert not pipeline.is_running
+
+        pipeline._state = PipelineState.RUNNING
+        assert pipeline.is_running
+
+
+class TestPipelineStats:
+    """Tests for pipeline statistics."""
+
+    def test_initial_stats(self, mock_settings):
+        """Pipeline should have zero stats initially."""
+        pipeline = Pipeline(mock_settings)
+        stats = pipeline.stats
+
+        assert stats.started_at is None
+        assert stats.trades_processed == 0
+        assert stats.signals_generated == 0
+        assert stats.alerts_sent == 0
+        assert stats.errors == 0
+
+
+class TestPipelineInitialization:
+    """Tests for pipeline initialization."""
+
+    def test_dry_run_from_settings(self, mock_settings):
+        """Pipeline should use dry_run from settings by default."""
+        mock_settings.dry_run = True
+        pipeline = Pipeline(mock_settings)
+        assert pipeline._dry_run is True
+
+        mock_settings.dry_run = False
+        pipeline = Pipeline(mock_settings)
+        assert pipeline._dry_run is False
+
+    def test_dry_run_override(self, mock_settings):
+        """Pipeline should allow overriding dry_run."""
+        mock_settings.dry_run = False
+        pipeline = Pipeline(mock_settings, dry_run=True)
+        assert pipeline._dry_run is True
+
+    def test_uses_get_settings_when_none_provided(self):
+        """Pipeline should call get_settings if no settings provided."""
+        with patch("polymarket_insider_tracker.pipeline.get_settings") as mock_get:
+            mock_get.return_value = MagicMock(spec=Settings)
+            mock_get.return_value.dry_run = False
+            Pipeline()
+            mock_get.assert_called_once()
+
+
+class TestBuildAlertChannels:
+    """Tests for alert channel building."""
+
+    def test_no_channels_when_none_enabled(self, mock_settings):
+        """Should return empty list when no channels enabled."""
+        mock_settings.discord.enabled = False
+        mock_settings.telegram.enabled = False
+
+        pipeline = Pipeline(mock_settings)
+        channels = pipeline._build_alert_channels()
+
+        assert channels == []
+
+    def test_discord_channel_when_enabled(self, mock_settings):
+        """Should add Discord channel when enabled."""
+        mock_settings.discord.enabled = True
+        mock_settings.discord.webhook_url = MagicMock()
+        mock_settings.discord.webhook_url.get_secret_value.return_value = (
+            "https://discord.com/webhook"
+        )
+
+        pipeline = Pipeline(mock_settings)
+        channels = pipeline._build_alert_channels()
+
+        assert len(channels) == 1
+        assert channels[0].name == "discord"
+
+    def test_telegram_channel_when_enabled(self, mock_settings):
+        """Should add Telegram channel when enabled."""
+        mock_settings.telegram.enabled = True
+        mock_settings.telegram.bot_token = MagicMock()
+        mock_settings.telegram.bot_token.get_secret_value.return_value = "bot_token"
+        mock_settings.telegram.chat_id = "chat_123"
+
+        pipeline = Pipeline(mock_settings)
+        channels = pipeline._build_alert_channels()
+
+        assert len(channels) == 1
+        assert channels[0].name == "telegram"
+
+    def test_both_channels_when_both_enabled(self, mock_settings):
+        """Should add both channels when both enabled."""
+        mock_settings.discord.enabled = True
+        mock_settings.discord.webhook_url = MagicMock()
+        mock_settings.discord.webhook_url.get_secret_value.return_value = (
+            "https://discord.com/webhook"
+        )
+        mock_settings.telegram.enabled = True
+        mock_settings.telegram.bot_token = MagicMock()
+        mock_settings.telegram.bot_token.get_secret_value.return_value = "bot_token"
+        mock_settings.telegram.chat_id = "chat_123"
+
+        pipeline = Pipeline(mock_settings)
+        channels = pipeline._build_alert_channels()
+
+        assert len(channels) == 2
+
+
+class TestOnTrade:
+    """Tests for trade event processing."""
+
+    @pytest.mark.asyncio
+    async def test_on_trade_increments_stats(self, mock_settings, sample_trade_event):
+        """Processing a trade should increment stats."""
+        pipeline = Pipeline(mock_settings)
+        pipeline._fresh_wallet_detector = AsyncMock(return_value=None)
+        pipeline._size_anomaly_detector = AsyncMock(return_value=None)
+
+        await pipeline._on_trade(sample_trade_event)
+
+        assert pipeline.stats.trades_processed == 1
+        assert pipeline.stats.last_trade_time is not None
+
+    @pytest.mark.asyncio
+    async def test_on_trade_runs_detectors_in_parallel(self, mock_settings, sample_trade_event):
+        """Detectors should run in parallel."""
+        pipeline = Pipeline(mock_settings)
+        pipeline._fresh_wallet_detector = AsyncMock()
+        pipeline._size_anomaly_detector = AsyncMock()
+
+        # Make detectors take some time
+        async def slow_detect(*_args):
+            await asyncio.sleep(0.1)
+            return None
+
+        pipeline._fresh_wallet_detector.analyze = slow_detect
+        pipeline._size_anomaly_detector.analyze = slow_detect
+
+        start = asyncio.get_event_loop().time()
+        await pipeline._on_trade(sample_trade_event)
+        elapsed = asyncio.get_event_loop().time() - start
+
+        # Should complete in ~0.1s not ~0.2s
+        assert elapsed < 0.15
+
+    @pytest.mark.asyncio
+    async def test_on_trade_handles_detector_errors(self, mock_settings, sample_trade_event):
+        """Should handle detector errors gracefully."""
+        pipeline = Pipeline(mock_settings)
+        pipeline._fresh_wallet_detector = MagicMock()
+        pipeline._fresh_wallet_detector.analyze = AsyncMock(side_effect=Exception("Detector error"))
+        pipeline._size_anomaly_detector = MagicMock()
+        pipeline._size_anomaly_detector.analyze = AsyncMock(return_value=None)
+
+        # Should not raise
+        await pipeline._on_trade(sample_trade_event)
+
+        # Should still increment trades processed
+        assert pipeline.stats.trades_processed == 1
+
+    @pytest.mark.asyncio
+    async def test_on_trade_calls_score_and_alert_when_signals(
+        self, mock_settings, sample_trade_event, sample_wallet_profile
+    ):
+        """Should call score_and_alert when signals are detected."""
+        pipeline = Pipeline(mock_settings)
+
+        # Create a signal
+        fresh_signal = FreshWalletSignal(
+            trade_event=sample_trade_event,
+            wallet_profile=sample_wallet_profile,
+            confidence=0.8,
+            factors={"base": 0.5, "brand_new": 0.2},
+        )
+
+        pipeline._fresh_wallet_detector = MagicMock()
+        pipeline._fresh_wallet_detector.analyze = AsyncMock(return_value=fresh_signal)
+        pipeline._size_anomaly_detector = MagicMock()
+        pipeline._size_anomaly_detector.analyze = AsyncMock(return_value=None)
+
+        # Mock score_and_alert
+        pipeline._score_and_alert = AsyncMock()
+
+        await pipeline._on_trade(sample_trade_event)
+
+        # Should call score_and_alert with the bundle
+        pipeline._score_and_alert.assert_called_once()
+        bundle = pipeline._score_and_alert.call_args[0][0]
+        assert bundle.fresh_wallet_signal == fresh_signal
+        assert pipeline.stats.signals_generated == 1
+
+
+class TestScoreAndAlert:
+    """Tests for scoring and alerting."""
+
+    @pytest.mark.asyncio
+    async def test_dry_run_skips_dispatch(
+        self, mock_settings, sample_trade_event, sample_wallet_profile
+    ):
+        """Dry run should skip actual alert dispatch."""
+        mock_settings.dry_run = True
+        pipeline = Pipeline(mock_settings)
+
+        # Create mock components
+        pipeline._risk_scorer = MagicMock()
+        pipeline._risk_scorer.assess = AsyncMock(
+            return_value=MagicMock(
+                should_alert=True,
+                wallet_address="0x" + "b" * 40,
+                weighted_score=0.85,
+            )
+        )
+        pipeline._alert_formatter = MagicMock()
+        pipeline._alert_dispatcher = MagicMock()
+        pipeline._alert_dispatcher.dispatch = AsyncMock()
+
+        bundle = SignalBundle(
+            trade_event=sample_trade_event,
+            fresh_wallet_signal=FreshWalletSignal(
+                trade_event=sample_trade_event,
+                wallet_profile=sample_wallet_profile,
+                confidence=0.8,
+                factors={},
+            ),
+        )
+
+        await pipeline._score_and_alert(bundle)
+
+        # Dispatcher should NOT be called in dry run
+        pipeline._alert_dispatcher.dispatch.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_no_alert_when_below_threshold(self, mock_settings, sample_trade_event):
+        """Should not alert when below threshold."""
+        pipeline = Pipeline(mock_settings)
+
+        # Create mock components
+        pipeline._risk_scorer = MagicMock()
+        pipeline._risk_scorer.assess = AsyncMock(
+            return_value=MagicMock(
+                should_alert=False,
+                weighted_score=0.4,
+            )
+        )
+        pipeline._alert_formatter = MagicMock()
+        pipeline._alert_dispatcher = MagicMock()
+
+        bundle = SignalBundle(trade_event=sample_trade_event)
+
+        await pipeline._score_and_alert(bundle)
+
+        # Formatter should NOT be called
+        pipeline._alert_formatter.format.assert_not_called()
+
+
+class TestPipelineLifecycle:
+    """Tests for pipeline lifecycle methods."""
+
+    @pytest.mark.asyncio
+    async def test_cannot_start_when_not_stopped(self, mock_settings):
+        """Should raise error when starting non-stopped pipeline."""
+        pipeline = Pipeline(mock_settings)
+        pipeline._state = PipelineState.RUNNING
+
+        with pytest.raises(RuntimeError, match="Cannot start pipeline"):
+            await pipeline.start()
+
+    @pytest.mark.asyncio
+    async def test_stop_when_already_stopped(self, mock_settings):
+        """Stop should be no-op when already stopped."""
+        pipeline = Pipeline(mock_settings)
+        assert pipeline.state == PipelineState.STOPPED
+
+        # Should not raise
+        await pipeline.stop()
+        assert pipeline.state == PipelineState.STOPPED
+
+
+class TestPipelineContextManager:
+    """Tests for async context manager."""
+
+    @pytest.mark.asyncio
+    async def test_context_manager_calls_start_and_stop(self, mock_settings):
+        """Context manager should call start and stop."""
+        pipeline = Pipeline(mock_settings)
+        pipeline.start = AsyncMock()
+        pipeline.stop = AsyncMock()
+
+        async with pipeline:
+            pipeline.start.assert_called_once()
+
+        pipeline.stop.assert_called_once()

tests/test_pipeline_persistence.py

@@ -0,0 +1,334 @@
+"""Tests verifying wallet and funding data persistence in the pipeline.
+
+These tests confirm that running the live pipeline writes rows into
+wallet_profiles and funding_transfers tables when fresh wallets are detected.
+"""
+
+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 FreshWalletSignal
+from polymarket_insider_tracker.ingestor.models import TradeEvent
+from polymarket_insider_tracker.pipeline import Pipeline
+from polymarket_insider_tracker.profiler.models import FundingChain, FundingTransfer, WalletProfile
+from polymarket_insider_tracker.storage.database import DatabaseManager
+from polymarket_insider_tracker.storage.models import Base, FundingTransferModel, WalletProfileModel
+
+
+@pytest.fixture
+def mock_settings():
+    """Create mock settings for testing."""
+    redis = MagicMock()
+    redis.url = "redis://localhost:6379"
+
+    database = MagicMock()
+    database.url = "sqlite+aiosqlite:///:memory:"
+
+    polygon = MagicMock()
+    polygon.rpc_url = "https://polygon-rpc.com"
+    polygon.fallback_rpc_url = None
+
+    polymarket = MagicMock()
+    polymarket.ws_url = "wss://ws-subscriptions-clob.polymarket.com/ws/market"
+    polymarket.api_key = None
+
+    discord = MagicMock()
+    discord.enabled = False
+    discord.webhook_url = None
+
+    telegram = MagicMock()
+    telegram.enabled = False
+    telegram.bot_token = None
+    telegram.chat_id = None
+
+    settings = MagicMock(spec=Settings)
+    settings.redis = redis
+    settings.database = database
+    settings.polygon = polygon
+    settings.polymarket = polymarket
+    settings.discord = discord
+    settings.telegram = telegram
+    settings.dry_run = True
+    return settings
+
+
+@pytest.fixture
+async def async_engine():
+    """Create an async SQLite engine for testing."""
+    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):
+    """Create a DatabaseManager backed by the in-memory SQLite 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():
+    """Create a sample trade event."""
+    return TradeEvent(
+        trade_id="0x" + "a" * 64,
+        wallet_address="0x" + "b" * 40,
+        market_id="0x" + "c" * 64,
+        asset_id="asset_123",
+        side="BUY",
+        price=Decimal("0.65"),
+        size=Decimal("5000"),
+        timestamp=datetime.now(UTC),
+        outcome="Yes",
+        outcome_index=0,
+        event_title="Test Market",
+        market_slug="test-market",
+    )
+
+
+@pytest.fixture
+def sample_profile():
+    """Create a sample fresh wallet profile."""
+    return WalletProfile(
+        address="0x" + "b" * 40,
+        nonce=2,
+        first_seen=datetime(2026, 3, 31, 12, 0, 0, tzinfo=UTC),
+        age_hours=1.5,
+        is_fresh=True,
+        total_tx_count=2,
+        matic_balance=Decimal("1000000000000000000"),
+        usdc_balance=Decimal("5000000000"),
+        fresh_threshold=5,
+    )
+
+
+@pytest.fixture
+def sample_funding_chain():
+    """Create a sample funding chain with one transfer."""
+    return FundingChain(
+        target_address="0x" + "b" * 40,
+        chain=[
+            FundingTransfer(
+                from_address="0x" + "d" * 40,
+                to_address="0x" + "b" * 40,
+                amount=Decimal("5000000000"),
+                token="USDC",
+                tx_hash="0x" + "e" * 64,
+                block_number=12345678,
+                timestamp=datetime(2026, 3, 31, 11, 0, 0, tzinfo=UTC),
+            ),
+        ],
+        origin_address="0x" + "d" * 40,
+        origin_type="cex_binance",
+        hop_count=1,
+    )
+
+
+class TestPipelinePersistence:
+    """Tests that the pipeline persists wallet and funding data to Postgres."""
+
+    @pytest.mark.asyncio
+    async def test_on_trade_persists_wallet_profile(
+        self, mock_settings, db_manager, sample_trade, sample_profile, async_engine
+    ):
+        """When a fresh wallet signal fires, the wallet profile is written to wallet_profiles."""
+        pipeline = Pipeline(mock_settings)
+        pipeline._db_manager = db_manager
+
+        fresh_signal = FreshWalletSignal(
+            trade_event=sample_trade,
+            wallet_profile=sample_profile,
+            confidence=0.8,
+            factors={"base": 0.5, "brand_new": 0.2},
+        )
+
+        pipeline._fresh_wallet_detector = MagicMock()
+        pipeline._fresh_wallet_detector.analyze = AsyncMock(return_value=fresh_signal)
+        pipeline._size_anomaly_detector = MagicMock()
+        pipeline._size_anomaly_detector.analyze = AsyncMock(return_value=None)
+        pipeline._funding_tracer = MagicMock()
+        pipeline._funding_tracer.trace = AsyncMock(
+            return_value=FundingChain(target_address=sample_profile.address)
+        )
+        pipeline._risk_scorer = MagicMock()
+        pipeline._risk_scorer.assess = AsyncMock(
+            return_value=MagicMock(should_alert=False, weighted_score=0.3)
+        )
+        pipeline._alert_formatter = MagicMock()
+        pipeline._alert_dispatcher = MagicMock()
+
+        await pipeline._on_trade(sample_trade)
+
+        # Verify wallet_profiles has a row
+        async with async_sessionmaker(bind=async_engine, expire_on_commit=False)() as session:
+            result = await session.execute(select(WalletProfileModel))
+            rows = result.scalars().all()
+            assert len(rows) == 1
+            assert rows[0].address == sample_profile.address.lower()
+            assert rows[0].nonce == sample_profile.nonce
+            assert rows[0].is_fresh is True
+
+    @pytest.mark.asyncio
+    async def test_on_trade_persists_funding_transfers(
+        self,
+        mock_settings,
+        db_manager,
+        sample_trade,
+        sample_profile,
+        sample_funding_chain,
+        async_engine,
+    ):
+        """When a fresh wallet signal fires, funding transfers are written to funding_transfers."""
+        pipeline = Pipeline(mock_settings)
+        pipeline._db_manager = db_manager
+
+        fresh_signal = FreshWalletSignal(
+            trade_event=sample_trade,
+            wallet_profile=sample_profile,
+            confidence=0.8,
+            factors={"base": 0.5, "brand_new": 0.2},
+        )
+
+        pipeline._fresh_wallet_detector = MagicMock()
+        pipeline._fresh_wallet_detector.analyze = AsyncMock(return_value=fresh_signal)
+        pipeline._size_anomaly_detector = MagicMock()
+        pipeline._size_anomaly_detector.analyze = AsyncMock(return_value=None)
+        pipeline._funding_tracer = MagicMock()
+        pipeline._funding_tracer.trace = AsyncMock(return_value=sample_funding_chain)
+        pipeline._risk_scorer = MagicMock()
+        pipeline._risk_scorer.assess = AsyncMock(
+            return_value=MagicMock(should_alert=False, weighted_score=0.3)
+        )
+        pipeline._alert_formatter = MagicMock()
+        pipeline._alert_dispatcher = MagicMock()
+
+        await pipeline._on_trade(sample_trade)
+
+        # Verify funding_transfers has a row
+        async with async_sessionmaker(bind=async_engine, expire_on_commit=False)() as session:
+            result = await session.execute(select(FundingTransferModel))
+            rows = result.scalars().all()
+            assert len(rows) == 1
+            assert rows[0].to_address == ("0x" + "b" * 40).lower()
+            assert rows[0].from_address == ("0x" + "d" * 40).lower()
+            assert rows[0].token == "USDC"
+            assert rows[0].tx_hash == ("0x" + "e" * 64).lower()
+
+    @pytest.mark.asyncio
+    async def test_no_persistence_without_fresh_signal(
+        self, mock_settings, db_manager, sample_trade, async_engine
+    ):
+        """No rows written when fresh wallet signal is None (wallet not fresh)."""
+        pipeline = Pipeline(mock_settings)
+        pipeline._db_manager = db_manager
+
+        pipeline._fresh_wallet_detector = MagicMock()
+        pipeline._fresh_wallet_detector.analyze = AsyncMock(return_value=None)
+        pipeline._size_anomaly_detector = MagicMock()
+        pipeline._size_anomaly_detector.analyze = AsyncMock(return_value=None)
+
+        await pipeline._on_trade(sample_trade)
+
+        async with async_sessionmaker(bind=async_engine, expire_on_commit=False)() as session:
+            wallets = (await session.execute(select(WalletProfileModel))).scalars().all()
+            transfers = (await session.execute(select(FundingTransferModel))).scalars().all()
+            assert len(wallets) == 0
+            assert len(transfers) == 0
+
+    @pytest.mark.asyncio
+    async def test_persistence_failure_does_not_break_pipeline(
+        self, mock_settings, sample_trade, sample_profile
+    ):
+        """Persistence errors are caught and don't crash trade processing."""
+        pipeline = Pipeline(mock_settings)
+
+        # Use a broken db_manager that raises on get_async_session
+        broken_db = MagicMock()
+        broken_db.get_async_session = MagicMock(side_effect=Exception("DB connection failed"))
+        pipeline._db_manager = broken_db
+
+        fresh_signal = FreshWalletSignal(
+            trade_event=sample_trade,
+            wallet_profile=sample_profile,
+            confidence=0.8,
+            factors={"base": 0.5},
+        )
+
+        pipeline._fresh_wallet_detector = MagicMock()
+        pipeline._fresh_wallet_detector.analyze = AsyncMock(return_value=fresh_signal)
+        pipeline._size_anomaly_detector = MagicMock()
+        pipeline._size_anomaly_detector.analyze = AsyncMock(return_value=None)
+        pipeline._funding_tracer = MagicMock()
+        pipeline._risk_scorer = MagicMock()
+        pipeline._risk_scorer.assess = AsyncMock(
+            return_value=MagicMock(should_alert=False, weighted_score=0.3)
+        )
+        pipeline._alert_formatter = MagicMock()
+        pipeline._alert_dispatcher = MagicMock()
+
+        # Should not raise
+        await pipeline._on_trade(sample_trade)
+        assert pipeline.stats.trades_processed == 1
+
+    @pytest.mark.asyncio
+    async def test_duplicate_funding_transfers_are_skipped(
+        self,
+        mock_settings,
+        db_manager,
+        sample_trade,
+        sample_profile,
+        sample_funding_chain,
+        async_engine,
+    ):
+        """Processing the same trade twice should not duplicate funding transfer rows."""
+        pipeline = Pipeline(mock_settings)
+        pipeline._db_manager = db_manager
+
+        fresh_signal = FreshWalletSignal(
+            trade_event=sample_trade,
+            wallet_profile=sample_profile,
+            confidence=0.8,
+            factors={"base": 0.5},
+        )
+
+        pipeline._fresh_wallet_detector = MagicMock()
+        pipeline._fresh_wallet_detector.analyze = AsyncMock(return_value=fresh_signal)
+        pipeline._size_anomaly_detector = MagicMock()
+        pipeline._size_anomaly_detector.analyze = AsyncMock(return_value=None)
+        pipeline._funding_tracer = MagicMock()
+        pipeline._funding_tracer.trace = AsyncMock(return_value=sample_funding_chain)
+        pipeline._risk_scorer = MagicMock()
+        pipeline._risk_scorer.assess = AsyncMock(
+            return_value=MagicMock(should_alert=False, weighted_score=0.3)
+        )
+        pipeline._alert_formatter = MagicMock()
+        pipeline._alert_dispatcher = MagicMock()
+
+        # Process same trade twice
+        await pipeline._on_trade(sample_trade)
+        await pipeline._on_trade(sample_trade)
+
+        # Should still have only 1 funding transfer (duplicate skipped)
+        async with async_sessionmaker(bind=async_engine, expire_on_commit=False)() as session:
+            result = await session.execute(select(FundingTransferModel))
+            rows = result.scalars().all()
+            assert len(rows) == 1

tests/test_setup.py

@@ -10,11 +10,7 @@ def test_version() -> None:
 
 def test_import_modules() -> None:
     """Test that all submodules can be imported."""
-    from polymarket_insider_tracker import ingestor
-    from polymarket_insider_tracker import profiler
-    from polymarket_insider_tracker import detector
-    from polymarket_insider_tracker import alerter
-    from polymarket_insider_tracker import storage
+    from polymarket_insider_tracker import alerter, detector, ingestor, profiler, storage
 
     # Just verify imports work
     assert ingestor is not None

tests/test_shutdown.py

@@ -0,0 +1,319 @@
+"""Tests for the graceful shutdown handler."""
+
+from __future__ import annotations
+
+import asyncio
+import signal
+import sys
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from polymarket_insider_tracker.shutdown import (
+    DEFAULT_SHUTDOWN_TIMEOUT,
+    SHUTDOWN_SIGNALS,
+    GracefulShutdown,
+    run_with_graceful_shutdown,
+)
+
+
+class TestGracefulShutdownInit:
+    """Tests for GracefulShutdown initialization."""
+
+    def test_default_timeout(self) -> None:
+        """Should use default timeout when not specified."""
+        shutdown = GracefulShutdown()
+        assert shutdown.timeout == DEFAULT_SHUTDOWN_TIMEOUT
+
+    def test_custom_timeout(self) -> None:
+        """Should accept custom timeout."""
+        shutdown = GracefulShutdown(timeout=60.0)
+        assert shutdown.timeout == 60.0
+
+    def test_initial_state(self) -> None:
+        """Should start in non-shutdown state."""
+        shutdown = GracefulShutdown()
+        assert shutdown.is_shutdown_requested is False
+        assert shutdown.is_force_exit_requested is False
+
+
+class TestRequestShutdown:
+    """Tests for programmatic shutdown requests."""
+
+    async def test_request_shutdown_sets_flag(self) -> None:
+        """Should set shutdown requested flag."""
+        shutdown = GracefulShutdown()
+        shutdown.request_shutdown()
+        assert shutdown.is_shutdown_requested is True
+
+    async def test_request_shutdown_sets_event(self) -> None:
+        """Should set the shutdown event when called."""
+        shutdown = GracefulShutdown()
+        shutdown._shutdown_event = asyncio.Event()
+
+        shutdown.request_shutdown()
+
+        assert shutdown._shutdown_event.is_set()
+
+    async def test_request_shutdown_idempotent(self) -> None:
+        """Multiple requests should be idempotent."""
+        shutdown = GracefulShutdown()
+        shutdown._shutdown_event = asyncio.Event()
+
+        shutdown.request_shutdown()
+        shutdown.request_shutdown()
+        shutdown.request_shutdown()
+
+        assert shutdown.is_shutdown_requested is True
+
+
+class TestWait:
+    """Tests for waiting for shutdown."""
+
+    async def test_wait_blocks_until_shutdown(self) -> None:
+        """Wait should block until shutdown is requested."""
+        shutdown = GracefulShutdown()
+
+        async def request_after_delay() -> None:
+            await asyncio.sleep(0.1)
+            shutdown.request_shutdown()
+
+        asyncio.create_task(request_after_delay())
+
+        # Should complete after the delay
+        await asyncio.wait_for(shutdown.wait(), timeout=1.0)
+
+        assert shutdown.is_shutdown_requested is True
+
+    async def test_wait_with_timeout_returns_true_on_shutdown(self) -> None:
+        """wait_with_timeout should return True when shutdown is requested."""
+        shutdown = GracefulShutdown(timeout=1.0)
+
+        async def request_after_delay() -> None:
+            await asyncio.sleep(0.05)
+            shutdown.request_shutdown()
+
+        asyncio.create_task(request_after_delay())
+
+        result = await shutdown.wait_with_timeout()
+        assert result is True
+
+    async def test_wait_with_timeout_returns_false_on_timeout(self) -> None:
+        """wait_with_timeout should return False when timeout occurs."""
+        shutdown = GracefulShutdown(timeout=0.05)
+
+        # Don't request shutdown, let it timeout
+        result = await shutdown.wait_with_timeout()
+        assert result is False
+
+
+class TestSignalHandlers:
+    """Tests for signal handler installation and removal."""
+
+    async def test_install_signal_handlers_creates_event(self) -> None:
+        """Installing handlers should create the shutdown event."""
+        shutdown = GracefulShutdown()
+        shutdown.install_signal_handlers()
+
+        try:
+            assert shutdown._shutdown_event is not None
+            assert not shutdown._shutdown_event.is_set()
+        finally:
+            shutdown.remove_signal_handlers()
+
+    @pytest.mark.skipif(sys.platform == "win32", reason="Unix-specific test")
+    async def test_unix_signal_handler_installed(self) -> None:
+        """On Unix, should install loop signal handlers."""
+        shutdown = GracefulShutdown()
+
+        with patch.object(asyncio.get_running_loop(), "add_signal_handler") as mock_add:
+            shutdown.install_signal_handlers()
+
+            # Should have been called for both SIGTERM and SIGINT
+            assert mock_add.call_count >= 1
+
+        shutdown.remove_signal_handlers()
+
+    @pytest.mark.skipif(sys.platform != "win32", reason="Windows-specific test")
+    async def test_windows_signal_handler_installed(self) -> None:
+        """On Windows, should install signal.signal handlers."""
+        shutdown = GracefulShutdown()
+
+        with patch("signal.signal") as mock_signal:
+            shutdown.install_signal_handlers()
+
+            # Should have been called for available signals
+            assert mock_signal.call_count >= 1
+
+        shutdown.remove_signal_handlers()
+
+
+class TestHandleSignal:
+    """Tests for signal handling behavior."""
+
+    async def test_first_signal_sets_shutdown_event(self) -> None:
+        """First signal should set the shutdown event."""
+        shutdown = GracefulShutdown()
+        shutdown._shutdown_event = asyncio.Event()
+
+        shutdown._handle_signal(signal.SIGTERM)
+
+        assert shutdown.is_shutdown_requested is True
+        assert shutdown._shutdown_event.is_set()
+        assert shutdown.is_force_exit_requested is False
+
+    async def test_second_signal_force_exits(self) -> None:
+        """Second signal should trigger force exit."""
+        shutdown = GracefulShutdown()
+        shutdown._shutdown_event = asyncio.Event()
+        shutdown._shutdown_requested = True  # First signal already received
+
+        with pytest.raises(SystemExit) as exc_info:
+            shutdown._handle_signal(signal.SIGTERM)
+
+        assert exc_info.value.code == 128 + signal.SIGTERM.value
+        assert shutdown.is_force_exit_requested is True
+
+
+class TestCleanupCallbacks:
+    """Tests for cleanup callback registration and execution."""
+
+    async def test_register_sync_callback(self) -> None:
+        """Should register sync cleanup callbacks."""
+        shutdown = GracefulShutdown()
+        callback = MagicMock()
+
+        shutdown.register_cleanup(callback)
+
+        assert callback in shutdown._cleanup_callbacks
+
+    async def test_run_sync_cleanup_callback(self) -> None:
+        """Should run sync cleanup callbacks."""
+        shutdown = GracefulShutdown()
+        callback = MagicMock()
+        shutdown.register_cleanup(callback)
+
+        await shutdown.run_cleanup_callbacks()
+
+        callback.assert_called_once()
+
+    async def test_run_async_cleanup_callback(self) -> None:
+        """Should run async cleanup callbacks."""
+        shutdown = GracefulShutdown()
+        called = False
+
+        async def async_callback() -> None:
+            nonlocal called
+            called = True
+
+        shutdown.register_cleanup(async_callback)
+
+        await shutdown.run_cleanup_callbacks()
+
+        assert called is True
+
+    async def test_cleanup_callback_error_logged(self) -> None:
+        """Cleanup callback errors should be logged, not raised."""
+        shutdown = GracefulShutdown()
+
+        def failing_callback() -> None:
+            raise ValueError("Cleanup failed")
+
+        shutdown.register_cleanup(failing_callback)
+
+        # Should not raise
+        await shutdown.run_cleanup_callbacks()
+
+
+class TestAsyncContextManager:
+    """Tests for async context manager protocol."""
+
+    async def test_context_manager_installs_handlers(self) -> None:
+        """Entering context should install signal handlers."""
+        shutdown = GracefulShutdown()
+
+        async with shutdown:
+            assert shutdown._shutdown_event is not None
+            assert shutdown._loop is not None
+
+    async def test_context_manager_removes_handlers(self) -> None:
+        """Exiting context should remove signal handlers."""
+        shutdown = GracefulShutdown()
+
+        with patch.object(shutdown, "remove_signal_handlers") as mock_remove:
+            async with shutdown:
+                pass
+
+            mock_remove.assert_called_once()
+
+    async def test_context_manager_runs_cleanup(self) -> None:
+        """Exiting context should run cleanup callbacks."""
+        shutdown = GracefulShutdown()
+        callback = MagicMock()
+        shutdown.register_cleanup(callback)
+
+        async with shutdown:
+            pass
+
+        callback.assert_called_once()
+
+
+class TestRunWithGracefulShutdown:
+    """Tests for the run_with_graceful_shutdown helper."""
+
+    async def test_runs_coroutine_to_completion(self) -> None:
+        """Should run the coroutine to completion if no shutdown."""
+        result = []
+
+        async def my_coro() -> None:
+            result.append("done")
+
+        await run_with_graceful_shutdown(my_coro(), timeout=1.0)
+
+        assert result == ["done"]
+
+    async def test_stops_on_shutdown_signal(self) -> None:
+        """Should stop when shutdown is signaled."""
+        result = []
+
+        async def long_running() -> None:
+            try:
+                await asyncio.sleep(10.0)
+            except asyncio.CancelledError:
+                result.append("cancelled")
+                raise
+
+        # Create a wrapper that will trigger shutdown
+        async def run_with_interrupt() -> None:
+            async def trigger_shutdown() -> None:
+                await asyncio.sleep(0.1)
+                # We can't easily trigger a signal in tests, so we test the task
+                # completion path instead
+                pass
+
+            # Run both tasks
+            task = asyncio.create_task(long_running())
+            asyncio.create_task(trigger_shutdown())
+
+            await asyncio.sleep(0.05)
+            task.cancel()
+
+            with pytest.raises(asyncio.CancelledError):
+                await task
+
+            result.append("finished")
+
+        await run_with_interrupt()
+        assert "finished" in result
+
+
+class TestShutdownSignals:
+    """Tests for shutdown signal configuration."""
+
+    def test_shutdown_signals_includes_sigterm(self) -> None:
+        """SHUTDOWN_SIGNALS should include SIGTERM."""
+        assert signal.SIGTERM in SHUTDOWN_SIGNALS
+
+    def test_shutdown_signals_includes_sigint(self) -> None:
+        """SHUTDOWN_SIGNALS should include SIGINT."""
+        assert signal.SIGINT in SHUTDOWN_SIGNALS

uv.lock

@@ -126,6 +126,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/fb/76/641ae371508676492379f16e2fa48f4e2c11741bd63c48be4b12a6b09cba/aiosignal-1.4.0-py3-none-any.whl", hash = "sha256:053243f8b92b990551949e63930a839ff0cf0b0ebbe0597b0f3fb19e1a0fe82e", size = 7490, upload-time = "2025-07-03T22:54:42.156Z" },
 ]
 
+[[package]]
+name = "aiosqlite"
+version = "0.22.1"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/4e/8a/64761f4005f17809769d23e518d915db74e6310474e733e3593cfc854ef1/aiosqlite-0.22.1.tar.gz", hash = "sha256:043e0bd78d32888c0a9ca90fc788b38796843360c855a7262a532813133a0650", size = 14821, upload-time = "2025-12-23T19:25:43.997Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/00/b7/e3bf5133d697a08128598c8d0abc5e16377b51465a33756de24fa7dee953/aiosqlite-0.22.1-py3-none-any.whl", hash = "sha256:21c002eb13823fad740196c5a2e9d8e62f6243bd9e7e4a1f87fb5e44ecb4fceb", size = 17405, upload-time = "2025-12-23T19:25:42.139Z" },
+]
+
 [[package]]
 name = "alembic"
 version = "1.17.2"
@@ -1538,6 +1547,7 @@ dependencies = [
     { name = "prometheus-client" },
     { name = "py-clob-client" },
     { name = "pydantic" },
+    { name = "pydantic-settings" },
     { name = "python-dotenv" },
     { name = "redis" },
     { name = "scikit-learn" },
@@ -1548,6 +1558,7 @@ dependencies = [
 
 [package.optional-dependencies]
 dev = [
+    { name = "aiosqlite" },
     { name = "mypy" },
     { name = "pre-commit" },
     { name = "pytest" },
@@ -1559,6 +1570,7 @@ dev = [
 [package.metadata]
 requires-dist = [
     { name = "aiohttp", specifier = ">=3.9.0" },
+    { name = "aiosqlite", marker = "extra == 'dev'", specifier = ">=0.19.0" },
     { name = "alembic", specifier = ">=1.13.0" },
     { name = "httpx", specifier = ">=0.25.0" },
     { name = "mypy", marker = "extra == 'dev'", specifier = ">=1.7.0" },
@@ -1567,6 +1579,7 @@ requires-dist = [
     { name = "prometheus-client", specifier = ">=0.19.0" },
     { name = "py-clob-client", specifier = ">=0.1.0" },
     { name = "pydantic", specifier = ">=2.0.0" },
+    { name = "pydantic-settings", specifier = ">=2.0.0" },
     { name = "pytest", marker = "extra == 'dev'", specifier = ">=7.0.0" },
     { name = "pytest-asyncio", marker = "extra == 'dev'", specifier = ">=0.21.0" },
     { name = "pytest-cov", marker = "extra == 'dev'", specifier = ">=4.0.0" },
@@ -1892,6 +1905,20 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/36/c7/cfc8e811f061c841d7990b0201912c3556bfeb99cdcb7ed24adc8d6f8704/pydantic_core-2.41.5-pp311-pypy311_pp73-win_amd64.whl", hash = "sha256:56121965f7a4dc965bff783d70b907ddf3d57f6eba29b6d2e5dabfaf07799c51", size = 2145302, upload-time = "2025-11-04T13:43:46.64Z" },
 ]
 
+[[package]]
+name = "pydantic-settings"
+version = "2.12.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "pydantic" },
+    { name = "python-dotenv" },
+    { name = "typing-inspection" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/43/4b/ac7e0aae12027748076d72a8764ff1c9d82ca75a7a52622e67ed3f765c54/pydantic_settings-2.12.0.tar.gz", hash = "sha256:005538ef951e3c2a68e1c08b292b5f2e71490def8589d4221b95dab00dafcfd0", size = 194184, upload-time = "2025-11-10T14:25:47.013Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/c1/60/5d4751ba3f4a40a6891f24eec885f51afd78d208498268c734e256fb13c4/pydantic_settings-2.12.0-py3-none-any.whl", hash = "sha256:fddb9fd99a5b18da837b29710391e945b1e30c135477f484084ee513adb93809", size = 51880, upload-time = "2025-11-10T14:25:45.546Z" },
+]
+
 [[package]]
 name = "pygments"
 version = "2.19.2"