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

fix(detector): suppress niche-only signals under a min trade size (#100 )
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>
2026-06-18 00:21:59 -04:00 · 2026-06-17 15:48:01 -04:00 · 2026-06-14 15:20:29 -04:00 · 2026-06-14 15:17:57 -04:00 · 2026-06-14 15:17:46 -04:00 · 2026-06-14 15:17:39 -04:00
30 changed files with 2422 additions and 372 deletions
@@ -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
@@ -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.
@@ -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.
@@ -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."
+```bash
-> — [@DidiTrading](https://x.com/DidiTrading)
+# Requires: Python 3.11+, Docker
-
+git clone https://github.com/pselamy/polymarket-insider-tracker.git
-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.
+cd polymarket-insider-tracker
-
+uv sync --all-extras          # or: pip install -e ".[dev]"
 **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" │
                   └─────────────────────┘
 ```
-### Detection Algorithms
+### 2. Start infrastructure
-1. **Fresh Wallet Detection**
+```bash
-   - Checks wallet transaction history on Polygon
+docker compose up -d           # PostgreSQL 15 + Redis 7
-   - Flags wallets with fewer than 5 lifetime transactions making trades over $1,000
+docker compose ps              # wait for healthy
-   - Traces funding source to identify if connected to known entities
+```
-2. **Liquidity Impact Analysis**
+### 3. Configure
   - 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. **Sniper Cluster Detection**
+```bash
-   - Uses DBSCAN clustering to find wallets that consistently enter markets within minutes of creation
+cp .env.example .env
-   - Identifies coordinated behavior patterns
+# Edit .env — only DATABASE_URL and REDIS_URL are required for local dev
 # (defaults in .env.example work with docker compose)
 ```
-4. **Event Correlation**
+### 4. Run migrations + start
-   - Cross-references trading activity with news feeds
+
-   - Detects positions opened 1-4 hours before related news breaks
+```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+
+### Components
 - Docker and Docker Compose
 - Polygon RPC endpoint (Alchemy, QuickNode, or self-hosted)
 - Polymarket API key (free at [docs.polymarket.com](https://docs.polymarket.com))
-### 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
+uv run pytest                        # run tests
-git clone https://github.com/pselamy/polymarket-insider-tracker.git
+uv run ruff check src/ tests/        # lint
-cd polymarket-insider-tracker
+uv run ruff format src/ tests/       # format
-
+uv run mypy src/                     # type check (strict mode)
 # 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
 ```
 ### 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) |
+| Adminer | 8080 | Database admin UI (optional, `--profile tools`) |
-| RedisInsight | 5540 | Redis admin UI (optional) |
+| RedisInsight | 5540 | Redis admin UI (optional, `--profile tools`) |
 ```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
 ```
 ---
-## Project Structure
+## Troubleshooting
-```
+**No trades received / silent connection**
-polymarket-insider-tracker/
+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).
 ├── 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
 ```
---
+**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)
+**Rate limiting on Polygon RPC**
- [x] Project structure and documentation
+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.).
 - [ ] 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/
 ```
 ---
@@ -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.
@@ -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)
@@ -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
+down_revision: str | None = None
-branch_labels: Union[str, Sequence[str], None] = None
+branch_labels: str | Sequence[str] | None = None
-depends_on: Union[str, Sequence[str], None] = None
+depends_on: str | Sequence[str] | None = None
 def upgrade() -> None:
@@ -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")
@@ -1,5 +1,3 @@
 version: "3.8"
 services:
  postgres:
    image: postgres:15
@@ -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
@@ -290,12 +290,14 @@ class AlertFormatter:
                    wallet_line += f" \\(Age: {age_hours:.0f}h\\)"
        lines.append(wallet_line)
-        # Risk score
+        # Risk score — every numeric literal here must be escaped because
-        lines.append(f"*Risk Score:* {assessment.weighted_score:.2f} \\({risk_level}\\)")
+        # MarkdownV2 treats `.` as a special character and rejects unescaped
        # ones with `Bad Request: can't parse entities`.
        score_str = self._escape_telegram_markdown(f"{assessment.weighted_score:.2f}")
        lines.append(f"*Risk Score:* {score_str} \\({risk_level}\\)")
        # Market
        market_title = trade.event_title or trade.market_slug or "Unknown Market"
        # Escape special Telegram markdown characters
        market_title_escaped = self._escape_telegram_markdown(market_title)
        if "market" in links:
            lines.append(f"*Market:* [{market_title_escaped}]({links['market']})")
@@ -303,14 +305,16 @@ class AlertFormatter:
            lines.append(f"*Market:* {market_title_escaped}")
        # Trade details
-        usdc_value = format_usdc(trade.notional_value).replace("$", "\\$")
+        usdc_value = self._escape_telegram_markdown(format_usdc(trade.notional_value))
-        lines.append(
+        price_str = self._escape_telegram_markdown(f"{trade.price:.3f}")
-            f"*Trade:* {trade.side} {trade.outcome} @ \\${trade.price:.3f} \\| {usdc_value}"
+        side_escaped = self._escape_telegram_markdown(trade.side)
-        )
+        outcome_escaped = self._escape_telegram_markdown(trade.outcome)
        lines.append(f"*Trade:* {side_escaped} {outcome_escaped} @ \\${price_str} \\| {usdc_value}")
        # Signals
        if signals:
-            lines.append(f"*Signals:* {', '.join(signals)}")
+            signals_escaped = [self._escape_telegram_markdown(s) for s in signals]
            lines.append(f"*Signals:* {', '.join(signals_escaped)}")
        # Links
        lines.append("")
@@ -18,7 +18,9 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
 class DatabaseSettings(BaseSettings):
    """Database connection settings."""
-    model_config = SettingsConfigDict(env_prefix="")
+    model_config = SettingsConfigDict(
        env_prefix="", env_file=".env", env_file_encoding="utf-8", extra="ignore"
    )
    url: str = Field(
        alias="DATABASE_URL",
@@ -37,7 +39,9 @@ class DatabaseSettings(BaseSettings):
 class RedisSettings(BaseSettings):
    """Redis connection settings."""
-    model_config = SettingsConfigDict(env_prefix="")
+    model_config = SettingsConfigDict(
        env_prefix="", env_file=".env", env_file_encoding="utf-8", extra="ignore"
    )
    url: str = Field(
        default="redis://localhost:6379",
@@ -57,7 +61,9 @@ class RedisSettings(BaseSettings):
 class PolygonSettings(BaseSettings):
    """Polygon blockchain RPC settings."""
-    model_config = SettingsConfigDict(env_prefix="POLYGON_")
+    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",
@@ -84,7 +90,9 @@ class PolygonSettings(BaseSettings):
 class PolymarketSettings(BaseSettings):
    """Polymarket API settings."""
-    model_config = SettingsConfigDict(env_prefix="POLYMARKET_")
+    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",
@@ -109,7 +117,9 @@ class PolymarketSettings(BaseSettings):
 class DiscordSettings(BaseSettings):
    """Discord notification settings."""
-    model_config = SettingsConfigDict(env_prefix="DISCORD_")
+    model_config = SettingsConfigDict(
        env_prefix="DISCORD_", env_file=".env", env_file_encoding="utf-8", extra="ignore"
    )
    webhook_url: SecretStr | None = Field(
        default=None,
@@ -120,13 +130,15 @@ class DiscordSettings(BaseSettings):
    @property
    def enabled(self) -> bool:
        """Check if Discord notifications are enabled."""
-        return self.webhook_url is not None
+        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_")
+    model_config = SettingsConfigDict(
        env_prefix="TELEGRAM_", env_file=".env", env_file_encoding="utf-8", extra="ignore"
    )
    bot_token: SecretStr | None = Field(
        default=None,
@@ -142,7 +154,39 @@ class TelegramSettings(BaseSettings):
    @property
    def enabled(self) -> bool:
        """Check if Telegram notifications are enabled."""
-        return self.bot_token is not None and self.chat_id is not None
+        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):
@@ -174,6 +218,7 @@ class Settings(BaseSettings):
    polymarket: PolymarketSettings = Field(default_factory=PolymarketSettings)
    discord: DiscordSettings = Field(default_factory=DiscordSettings)
    telegram: TelegramSettings = Field(default_factory=TelegramSettings)
    detector: DetectorSettings = Field(default_factory=DetectorSettings)
    # Application settings
    log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = Field(
@@ -19,8 +19,12 @@ from polymarket_insider_tracker.ingestor.models import TradeEvent
 logger = logging.getLogger(__name__)
-# Default configuration
+# Default configuration. The threshold lifted from 0.6 to 0.80 after the
-DEFAULT_ALERT_THRESHOLD = 0.6
+# first cost-adjusted backtest showed everything below 0.85 was follower-PnL
 # negative under realistic taker fees + half-cent slippage. 0.80 keeps a small
 # margin below 0.85+ so we don't drop borderline-high signals on a hard cliff.
 # Override at runtime via DETECTOR_ALERT_THRESHOLD env var.
 DEFAULT_ALERT_THRESHOLD = 0.80
 DEFAULT_DEDUP_WINDOW_SECONDS = 3600  # 1 hour
 DEFAULT_REDIS_KEY_PREFIX = "polymarket:dedup:"
@@ -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",
@@ -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
@@ -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)
+            # Fetch CLOB markets and gamma volume snapshot in parallel
-            markets = await asyncio.to_thread(self._clob.get_markets, True)
+            markets, gamma_stats = await asyncio.gather(
                asyncio.to_thread(self._clob.get_markets, True),
                self._fetch_gamma_stats(),
            )
-            # Cache each market in Redis
+            # Cache each market in Redis, enriched with gamma volume/liquidity
            cached_count = 0
            enriched_count = 0
            for market in markets:
                try:
                    metadata = MarketMetadata.from_market(market)
                    stats = gamma_stats.get(metadata.condition_id)
                    if stats is not None:
                        metadata = replace(
                            metadata,
                            daily_volume=stats.daily_volume,
                            weekly_volume=stats.weekly_volume,
                            liquidity=stats.liquidity,
                        )
                        enriched_count += 1
                    await self._cache_market(metadata)
                    cached_count += 1
                except Exception as e:
@@ -246,7 +276,10 @@ class MarketMetadataSync:
            self._set_state(SyncState.IDLE)
            logger.info(
-                f"Synced {cached_count} markets in {self._stats.last_sync_duration_seconds:.2f}s"
+                "Synced %d markets (%d enriched with gamma volume) in %.2fs",
                cached_count,
                enriched_count,
                self._stats.last_sync_duration_seconds,
            )
            # Notify callback
@@ -86,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":
@@ -399,6 +399,12 @@ class MarketMetadata:
    # Derived metadata
    category: str = "other"
    # Liquidity/volume snapshot (from gamma-api). All optional — older
    # cache entries and CLOB-only sync results may not have these.
    daily_volume: Decimal | None = None
    weekly_volume: Decimal | None = None
    liquidity: Decimal | None = None
    # Cache metadata
    last_updated: datetime = field(default_factory=lambda: datetime.now(UTC))
@@ -446,6 +452,9 @@ class MarketMetadata:
            "active": self.active,
            "closed": self.closed,
            "category": self.category,
            "daily_volume": str(self.daily_volume) if self.daily_volume is not None else None,
            "weekly_volume": str(self.weekly_volume) if self.weekly_volume is not None else None,
            "liquidity": str(self.liquidity) if self.liquidity is not None else None,
            "last_updated": self.last_updated.isoformat(),
        }
@@ -477,6 +486,15 @@ class MarketMetadata:
        else:
            last_updated = datetime.now(UTC)
        def _opt_dec(key: str) -> Decimal | None:
            raw = data.get(key)
            if raw is None or raw == "":
                return None
            try:
                return Decimal(str(raw))
            except (ValueError, ArithmeticError):
                return None
        return cls(
            condition_id=str(data["condition_id"]),
            question=str(data.get("question", "")),
@@ -486,5 +504,8 @@ class MarketMetadata:
            active=bool(data.get("active", True)),
            closed=bool(data.get("closed", False)),
            category=str(data.get("category", "other")),
            daily_volume=_opt_dec("daily_volume"),
            weekly_volume=_opt_dec("weekly_volume"),
            liquidity=_opt_dec("liquidity"),
            last_updated=last_updated,
        )
@@ -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
+            # ws-live-data pushes {connection_id, payload:{...trade fields}}
-            topic = data.get("topic")
+            payload = data.get("payload")
-            msg_type = data.get("type")
+            if (
-
+                isinstance(payload, dict)
-            if topic == "activity" and msg_type == "trades":
+                and "transactionHash" in payload
-                payload = data.get("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 non-trade message: %s", str(data)[:120])
                logger.debug("Received message: topic=%s type=%s", topic, msg_type)
        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
@@ -11,7 +11,7 @@ import contextlib
 import logging
 from dataclasses import dataclass
 from datetime import UTC, datetime
-from enum import Enum
+from enum import StrEnum
 from typing import TYPE_CHECKING
 from redis.asyncio import Redis
@@ -29,13 +29,23 @@ 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
@@ -43,7 +53,7 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
-class PipelineState(str, Enum):
+class PipelineState(StrEnum):
    """Pipeline lifecycle states."""
    STOPPED = "stopped"
@@ -120,6 +130,7 @@ class Pipeline:
        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
@@ -233,6 +244,10 @@ class Pipeline:
            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)
@@ -240,7 +255,17 @@ class Pipeline:
        # Initialize Risk Scorer
        logger.debug("Initializing risk scorer...")
-        self._risk_scorer = RiskScorer(self._redis)
+        self._risk_scorer = RiskScorer(
            self._redis,
            alert_threshold=settings.detector.alert_threshold,
            dedup_window_seconds=settings.detector.dedup_window_seconds,
        )
        logger.info(
            "RiskScorer threshold=%.2f dedup_window=%ds persist=%s",
            settings.detector.alert_threshold,
            settings.detector.dedup_window_seconds,
            settings.detector.persist_assessments,
        )
        # Initialize Alerting
        logger.debug("Initializing alerting components...")
@@ -365,6 +390,10 @@ class Pipeline:
                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,
@@ -382,6 +411,63 @@ class Pipeline:
            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:
@@ -403,13 +489,19 @@ class Pipeline:
            return None
    async def _score_and_alert(self, bundle: SignalBundle) -> None:
-        """Score signals and send alert if threshold exceeded."""
+        """Score signals, persist the assessment, and send alert if above threshold."""
        if not self._risk_scorer or not self._alert_formatter or not self._alert_dispatcher:
            return
        # Get risk assessment
        assessment = await self._risk_scorer.assess(bundle)
        # Persist every signal-bearing assessment (not just delivered alerts).
        # This is the ground-truth log future backtests will read instead of
        # grepping systemd. Failure here must never block alerting.
        if self._settings.detector.persist_assessments:
            await self._persist_assessment(assessment)
        if not assessment.should_alert:
            logger.debug(
                "Trade %s below alert threshold (score=%.2f)",
@@ -445,6 +537,55 @@ class Pipeline:
                result.success_count + result.failure_count,
            )
    async def _persist_assessment(self, assessment: RiskAssessment) -> None:
        """Write the assessment row. Best-effort; never raises."""
        if not self._db_manager:
            return
        from decimal import Decimal as _D
        trade = assessment.trade_event
        fresh = assessment.fresh_wallet_signal
        size_sig = assessment.size_anomaly_signal
        wallet_age: _D | None = None
        if fresh is not None and fresh.wallet_profile.age_hours is not None:
            wallet_age = _D(str(round(float(fresh.wallet_profile.age_hours), 2)))
        dto = RiskAssessmentDTO(
            assessment_id=assessment.assessment_id,
            trade_id=trade.trade_id,
            wallet_address=assessment.wallet_address.lower(),
            market_id=assessment.market_id,
            asset_id=getattr(trade, "asset_id", None) or None,
            side=trade.side,
            outcome=getattr(trade, "outcome", None) or None,
            outcome_index=getattr(trade, "outcome_index", None),
            price=trade.price,
            size=trade.size,
            notional_usdc=trade.notional_value,
            trade_timestamp=trade.timestamp,
            weighted_score=_D(str(round(assessment.weighted_score, 3))),
            signals_triggered=assessment.signals_triggered,
            fresh_wallet_confidence=(
                _D(str(round(fresh.confidence, 3))) if fresh is not None else None
            ),
            size_anomaly_confidence=(
                _D(str(round(size_sig.confidence, 3))) if size_sig is not None else None
            ),
            is_niche_market=size_sig.is_niche_market if size_sig is not None else None,
            volume_impact=(
                _D(str(round(size_sig.volume_impact, 4))) if size_sig is not None else None
            ),
            book_impact=(_D(str(round(size_sig.book_impact, 4))) if size_sig is not None else None),
            wallet_age_hours=wallet_age,
            should_alert=assessment.should_alert,
            threshold_at_eval=_D(str(round(self._settings.detector.alert_threshold, 3))),
        )
        try:
            async with self._db_manager.get_async_session() as session:
                repo = RiskAssessmentRepository(session)
                await repo.insert(dto)
        except Exception as e:
            logger.warning("Failed to persist risk assessment %s: %s", assessment.assessment_id, e)
    async def run(self) -> None:
        """Start the pipeline and run until interrupted.
@@ -26,8 +26,46 @@ logger = logging.getLogger(__name__)
 USDC_BRIDGED = "0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174"
 USDC_NATIVE = "0x3c499c542cEF5E3811e1192ce70d8cC03d5c3359"
-# ERC20 Transfer event signature
+# ERC20 Transfer event signature. ``HexBytes.hex()`` returns a *bare* hex
 # string without the ``0x`` prefix; publicnode tolerates that, but stricter
 # providers (e.g. drpc — which we use as the fallback) reject it with
 # ``invalid argument 0: hex string without 0x prefix``. Always pass the
 # 0x-prefixed form to ``eth_getLogs``.
 TRANSFER_EVENT_SIGNATURE = AsyncWeb3.keccak(text="Transfer(address,address,uint256)")
 TRANSFER_EVENT_TOPIC = "0x" + TRANSFER_EVENT_SIGNATURE.hex().removeprefix("0x")
 # eth_getLogs block-range chunking. Most public Polygon RPCs (publicnode, ankr,
 # llamarpc) cap the range at 10_000 blocks per call; pick a window slightly
 # under the cap so off-by-one differences between providers don't trip us up.
 DEFAULT_CHUNK_SIZE_BLOCKS = 9_000
 # Polygon block time is ~2.0s. publicnode (the most common free RPC) prunes
 # log history aggressively — empirically only ~100k blocks (~55 hours) are
 # served before requests start returning "History has been pruned". We default
 # to 80k blocks (~44 hours), which is more than enough for fresh-wallet
 # funding traces (those wallets are by definition new) and fits comfortably
 # inside what most public providers retain.
 DEFAULT_MAX_LOOKBACK_BLOCKS = 80_000
 # Substrings that, when present in an RPC error, indicate the chunk we just
 # asked for is outside the provider's archive horizon. Walking further back
 # is futile, so we stop the trace early instead of hammering every chunk.
 _PRUNED_HISTORY_MARKERS: tuple[str, ...] = (
    "history has been pruned",
    "missing trie node",
    "older than",
 )
 def _is_pruned_history_error(err: BaseException) -> bool:
    """Return True if the RPC error indicates pruned history.
    Public Polygon nodes only retain a recent slice of log history. When we
    walk back through that slice in chunks and hit the cutoff, every further
    chunk will fail with the same message — so we stop early instead of
    burning quota on guaranteed failures.
    """
    text = str(err).lower()
    return any(marker in text for marker in _PRUNED_HISTORY_MARKERS)
 class FundingTracer:
@@ -50,6 +88,8 @@ class FundingTracer:
        *,
        max_hops: int = 3,
        usdc_addresses: list[str] | None = None,
        chunk_size_blocks: int = DEFAULT_CHUNK_SIZE_BLOCKS,
        max_lookback_blocks: int = DEFAULT_MAX_LOOKBACK_BLOCKS,
    ) -> None:
        """Initialize the funding tracer.
@@ -58,6 +98,11 @@ class FundingTracer:
            entity_registry: Registry for entity classification. Creates default if None.
            max_hops: Maximum hops to trace back (default 3).
            usdc_addresses: USDC contract addresses to track. Uses defaults if None.
            chunk_size_blocks: Block window size per eth_getLogs call. Public
                Polygon RPCs cap at 10_000 blocks; default leaves a safety margin.
            max_lookback_blocks: How far back to scan when caller passes
                ``from_block=0``. Default ~44 hours at 2s block time, which
                fits inside the pruned-history horizon of most public RPCs.
        """
        self.polygon_client = polygon_client
        self.entity_registry = entity_registry or EntityRegistry()
@@ -65,6 +110,8 @@ class FundingTracer:
        self._usdc_addresses = [
            addr.lower() for addr in (usdc_addresses or [USDC_BRIDGED, USDC_NATIVE])
        ]
        self._chunk_size_blocks = chunk_size_blocks
        self._max_lookback_blocks = max_lookback_blocks
    async def trace(
        self,
@@ -209,46 +256,144 @@ class FundingTracer:
    ) -> list[dict[str, Any]]:
        """Get ERC20 Transfer event logs.
        Public Polygon RPCs (publicnode, ankr, llamarpc) cap ``eth_getLogs`` at
        10_000 blocks per call. To work around this we resolve the requested
        range into a concrete block window (defaulting to the last
        ``max_lookback_blocks`` when caller passes ``from_block=0``) and walk
        the window in chunks of ``chunk_size_blocks``, oldest-first, stopping
        once ``limit`` matches are collected. Walking oldest-first preserves
        the "first transfer" semantics expected by the funding chain tracer.
        If a chunk comes back with a "history has been pruned" style error
        the rest of the walk is short-circuited — every subsequent chunk
        would hit the same archive cutoff and there's no point burning quota
        on guaranteed failures.
        Args:
            to_address: Filter by recipient address.
            token_address: ERC20 token contract address.
            limit: Maximum logs to return.
-            from_block: Starting block number.
+            from_block: Starting block number (0 means
-            to_block: Ending block number.
+                ``latest - max_lookback_blocks``).
            to_block: Ending block number ("latest" resolves to current head).
        Returns:
-            List of log dictionaries.
+            List of log dictionaries, oldest first, capped at ``limit``.
        """
        # Pad address to 32 bytes for topic filter
        padded_to = "0x" + to_address.lower().replace("0x", "").zfill(64)
        topics = [
            TRANSFER_EVENT_TOPIC,  # Transfer event (must be 0x-prefixed for drpc)
            None,  # from (any)
            padded_to,  # to (target address)
        ]
        contract_address = AsyncWeb3.to_checksum_address(token_address)
        start_block, end_block = await self._resolve_block_range(from_block, to_block)
        if start_block > end_block:
            return []
        results: list[dict[str, Any]] = []
        chunk_size = max(1, self._chunk_size_blocks)
        chunk_start = start_block
        while chunk_start <= end_block:
            chunk_end = min(chunk_start + chunk_size - 1, end_block)
            try:
                chunk_logs = await self._fetch_logs_chunk(
                    contract_address=contract_address,
                    topics=topics,
                    from_block=chunk_start,
                    to_block=chunk_end,
                )
            except Exception as e:
                if _is_pruned_history_error(e):
                    # The provider has dropped this slice of history. Walking
                    # further back will hit the same wall on every chunk;
                    # stop now and return what we already have.
                    logger.info(
                        "eth_getLogs chunk %d-%d outside archive horizon for %s; stopping trace",
                        chunk_start,
                        chunk_end,
                        to_address,
                    )
                    break
                logger.warning(
                    "eth_getLogs chunk %d-%d failed for %s: %s",
                    chunk_start,
                    chunk_end,
                    to_address,
                    e,
                )
                # Skip this window and keep walking — partial data is better
                # than aborting the whole trace on a single flaky chunk.
                chunk_start = chunk_end + 1
                continue
            for log in chunk_logs:
                results.append(dict(log))
                if len(results) >= limit:
                    return results
            chunk_start = chunk_end + 1
        return results
    async def _resolve_block_range(
        self,
        from_block: int | str,
        to_block: int | str,
    ) -> tuple[int, int]:
        """Resolve symbolic block params to concrete numeric bounds.
        ``from_block=0`` (the historical default) is rewritten to
        ``latest - max_lookback_blocks`` so we don't try to scan all of Polygon.
        """
        w3 = self._select_w3()
        if isinstance(to_block, str):
            await self.polygon_client._rate_limiter.acquire()
            head = int(await w3.eth.block_number)
            end = head
        else:
            end = int(to_block)
        if isinstance(from_block, str):
            # Treat any symbolic from-block (e.g. "earliest") as "go back
            # max_lookback_blocks from end"; that's what callers actually want.
            start = max(0, end - self._max_lookback_blocks)
        elif from_block == 0:
            start = max(0, end - self._max_lookback_blocks)
        else:
            start = int(from_block)
        return start, end
    async def _fetch_logs_chunk(
        self,
        contract_address: str,
        topics: list[Any],
        from_block: int,
        to_block: int,
    ) -> list[Any]:
        """Issue a single bounded ``eth_getLogs`` call."""
        await self.polygon_client._rate_limiter.acquire()
-
+        w3 = self._select_w3()
        # Use the web3 instance from polygon client
        w3 = (
            self.polygon_client._w3
            if self.polygon_client._primary_healthy
            else (self.polygon_client._w3_fallback or self.polygon_client._w3)
        )
        # Get logs with Transfer event filtering by recipient
        # Note: web3 typing is overly restrictive for block params
-        logs = await w3.eth.get_logs(
+        return await w3.eth.get_logs(
            {
-                "address": AsyncWeb3.to_checksum_address(token_address),
+                "address": contract_address,
-                "topics": [
+                "topics": topics,
                    TRANSFER_EVENT_SIGNATURE.hex(),  # Transfer event
                    None,  # from (any)
                    padded_to,  # to (target address)
                ],
                "fromBlock": from_block,  # type: ignore[typeddict-item]
                "toBlock": to_block,  # type: ignore[typeddict-item]
            }
        )
-        # Convert to list of dicts and limit
+    def _select_w3(self) -> AsyncWeb3:
-        result = [dict(log) for log in logs[:limit]]
+        """Pick primary or fallback web3 instance based on health."""
-        return result
+        if self.polygon_client._primary_healthy:
            return self.polygon_client._w3
        return self.polygon_client._w3_fallback or self.polygon_client._w3
    async def _log_to_funding_transfer(
        self,
@@ -115,3 +115,57 @@ class WalletRelationshipModel(Base):
        Index("idx_wallet_relationships_a", "wallet_a"),
        Index("idx_wallet_relationships_b", "wallet_b"),
    )
 class RiskAssessmentModel(Base):
    """SQLAlchemy model for risk assessments.
    One row per signal-bearing trade (i.e. trades that triggered at least one
    detector). Captures everything a future backtest needs without going back
    to the public API: trade identity, score, per-signal confidences, and
    whether the alert was actually delivered (could be False due to dedup or
    threshold).
    """
    __tablename__ = "risk_assessments"
    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    assessment_id: Mapped[str] = mapped_column(String(36), unique=True, nullable=False)
    # Trade identity
    trade_id: Mapped[str] = mapped_column(String(80), nullable=False)
    wallet_address: Mapped[str] = mapped_column(String(42), nullable=False)
    market_id: Mapped[str] = mapped_column(String(80), nullable=False)
    asset_id: Mapped[str | None] = mapped_column(String(80), nullable=True)
    side: Mapped[str] = mapped_column(String(8), nullable=False)
    outcome: Mapped[str | None] = mapped_column(String(120), nullable=True)
    outcome_index: Mapped[int | None] = mapped_column(Integer, nullable=True)
    price: Mapped[Decimal] = mapped_column(Numeric(10, 6), nullable=False)
    size: Mapped[Decimal] = mapped_column(Numeric(20, 6), nullable=False)
    notional_usdc: Mapped[Decimal] = mapped_column(Numeric(20, 6), nullable=False)
    trade_timestamp: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
    # Scoring
    weighted_score: Mapped[Decimal] = mapped_column(Numeric(4, 3), nullable=False)
    signals_triggered: Mapped[int] = mapped_column(Integer, nullable=False)
    fresh_wallet_confidence: Mapped[Decimal | None] = mapped_column(Numeric(4, 3), nullable=True)
    size_anomaly_confidence: Mapped[Decimal | None] = mapped_column(Numeric(4, 3), nullable=True)
    is_niche_market: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
    volume_impact: Mapped[Decimal | None] = mapped_column(Numeric(8, 4), nullable=True)
    book_impact: Mapped[Decimal | None] = mapped_column(Numeric(8, 4), nullable=True)
    wallet_age_hours: Mapped[Decimal | None] = mapped_column(Numeric(10, 2), nullable=True)
    # Decision
    should_alert: Mapped[bool] = mapped_column(Boolean, nullable=False)
    threshold_at_eval: Mapped[Decimal] = mapped_column(Numeric(4, 3), nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC)
    )
    __table_args__ = (
        Index("idx_risk_assessments_wallet", "wallet_address"),
        Index("idx_risk_assessments_market", "market_id"),
        Index("idx_risk_assessments_trade_ts", "trade_timestamp"),
        Index("idx_risk_assessments_score", "weighted_score"),
    )
@@ -18,6 +18,7 @@ from sqlalchemy.dialects.sqlite import insert as sqlite_insert
 from polymarket_insider_tracker.storage.models import (
    FundingTransferModel,
    RiskAssessmentModel,
    WalletProfileModel,
    WalletRelationshipModel,
 )
@@ -510,3 +511,107 @@ class RelationshipRepository:
        )
        # SQLAlchemy Result does have rowcount but typing doesn't reflect it
        return (result.rowcount or 0) > 0  # type: ignore[attr-defined]
@dataclass
 class RiskAssessmentDTO:
    """Data transfer object for a persisted risk assessment.
    Captures everything a future backtest needs without going back to
    public APIs: trade identity, score, per-signal confidences, and
    whether the alert was actually delivered.
    """
    assessment_id: str
    trade_id: str
    wallet_address: str
    market_id: str
    asset_id: str | None
    side: str
    outcome: str | None
    outcome_index: int | None
    price: Decimal
    size: Decimal
    notional_usdc: Decimal
    trade_timestamp: datetime
    weighted_score: Decimal
    signals_triggered: int
    fresh_wallet_confidence: Decimal | None
    size_anomaly_confidence: Decimal | None
    is_niche_market: bool | None
    volume_impact: Decimal | None
    book_impact: Decimal | None
    wallet_age_hours: Decimal | None
    should_alert: bool
    threshold_at_eval: Decimal
    created_at: datetime | None = None
 class RiskAssessmentRepository:
    """Repository for risk assessment data access."""
    def __init__(self, session: AsyncSession) -> None:
        self.session = session
    async def insert(self, dto: RiskAssessmentDTO) -> RiskAssessmentDTO:
        """Insert a single assessment. Idempotent on assessment_id collisions."""
        model = RiskAssessmentModel(
            assessment_id=dto.assessment_id,
            trade_id=dto.trade_id,
            wallet_address=dto.wallet_address.lower(),
            market_id=dto.market_id,
            asset_id=dto.asset_id,
            side=dto.side,
            outcome=dto.outcome,
            outcome_index=dto.outcome_index,
            price=dto.price,
            size=dto.size,
            notional_usdc=dto.notional_usdc,
            trade_timestamp=dto.trade_timestamp,
            weighted_score=dto.weighted_score,
            signals_triggered=dto.signals_triggered,
            fresh_wallet_confidence=dto.fresh_wallet_confidence,
            size_anomaly_confidence=dto.size_anomaly_confidence,
            is_niche_market=dto.is_niche_market,
            volume_impact=dto.volume_impact,
            book_impact=dto.book_impact,
            wallet_age_hours=dto.wallet_age_hours,
            should_alert=dto.should_alert,
            threshold_at_eval=dto.threshold_at_eval,
        )
        self.session.add(model)
        await self.session.flush()
        return dto
    async def get_by_assessment_id(self, assessment_id: str) -> RiskAssessmentDTO | None:
        result = await self.session.execute(
            select(RiskAssessmentModel).where(RiskAssessmentModel.assessment_id == assessment_id)
        )
        model = result.scalar_one_or_none()
        if model is None:
            return None
        return RiskAssessmentDTO(
            assessment_id=model.assessment_id,
            trade_id=model.trade_id,
            wallet_address=model.wallet_address,
            market_id=model.market_id,
            asset_id=model.asset_id,
            side=model.side,
            outcome=model.outcome,
            outcome_index=model.outcome_index,
            price=model.price,
            size=model.size,
            notional_usdc=model.notional_usdc,
            trade_timestamp=model.trade_timestamp,
            weighted_score=model.weighted_score,
            signals_triggered=model.signals_triggered,
            fresh_wallet_confidence=model.fresh_wallet_confidence,
            size_anomaly_confidence=model.size_anomaly_confidence,
            is_niche_market=model.is_niche_market,
            volume_impact=model.volume_impact,
            book_impact=model.book_impact,
            wallet_age_hours=model.wallet_age_hours,
            should_alert=model.should_alert,
            threshold_at_eval=model.threshold_at_eval,
            created_at=model.created_at,
        )
@@ -409,12 +409,27 @@ class TestTelegramMarkdown:
        assert "`0x1234...5678`" in result.telegram_markdown
    def test_telegram_includes_risk_score(self, high_risk_assessment: RiskAssessment) -> None:
-        """Test that Telegram message includes risk score."""
+        """Test that Telegram message includes risk score, MarkdownV2-escaped."""
        formatter = AlertFormatter()
        result = formatter.format(high_risk_assessment)
-        assert "0.82" in result.telegram_markdown
+        # MarkdownV2 requires `.` to be escaped, so 0.82 becomes 0\.82.
        assert "0\\.82" in result.telegram_markdown
        assert "HIGH" in result.telegram_markdown
    def test_telegram_escapes_all_decimals(self, high_risk_assessment: RiskAssessment) -> None:
        """Telegram MarkdownV2 rejects unescaped `.` in dynamic numeric
        fields (risk score, price, USDC amount). All must be present in
        their escaped form."""
        formatter = AlertFormatter()
        result = formatter.format(high_risk_assessment)
        md = result.telegram_markdown
        for unescaped in ("0.82", "0.075", "15,000.00"):
            assert unescaped not in md, (
                f"unescaped {unescaped!r} would be rejected by Telegram MarkdownV2: {md!r}"
            )
        for escaped in ("0\\.82", "0\\.075", "15,000\\.00"):
            assert escaped in md, f"missing escaped {escaped!r} in {md!r}"
    def test_telegram_includes_links(self, high_risk_assessment: RiskAssessment) -> None:
        """Test that Telegram message includes links."""
        formatter = AlertFormatter()
@@ -588,6 +588,93 @@ class TestAnalyzeMethod:
        assert signal.is_niche_market is True
        assert signal.confidence == 0.2  # niche_base
    @pytest.mark.asyncio
    async def test_analyze_niche_only_below_min_trade_size_skipped(
        self,
        mock_metadata_sync: AsyncMock,
        sample_metadata: MarketMetadata,
    ) -> None:
        """Niche-only trades below the min trade size are suppressed."""
        mock_metadata_sync.get_market.return_value = sample_metadata
        detector = SizeAnomalyDetector(mock_metadata_sync)
        tiny_trade = TradeEvent(
            market_id="market_abc123",
            trade_id="tx_tiny",
            wallet_address="0xabc",
            side="BUY",
            outcome="Yes",
            outcome_index=0,
            price=Decimal("0.5"),
            size=Decimal("100"),  # $50 notional, below default $500 floor
            timestamp=datetime.now(UTC),
            asset_id="token_123",
            event_title="Niche tiny trade",
        )
        signal = await detector.analyze(tiny_trade)
        assert signal is None
    @pytest.mark.asyncio
    async def test_analyze_niche_only_at_min_trade_size_emits(
        self,
        mock_metadata_sync: AsyncMock,
        sample_metadata: MarketMetadata,
    ) -> None:
        """Niche-only trades at or above the min trade size still emit."""
        mock_metadata_sync.get_market.return_value = sample_metadata
        detector = SizeAnomalyDetector(mock_metadata_sync)
        ok_trade = TradeEvent(
            market_id="market_abc123",
            trade_id="tx_ok",
            wallet_address="0xabc",
            side="BUY",
            outcome="Yes",
            outcome_index=0,
            price=Decimal("0.5"),
            size=Decimal("2000"),  # $1000 notional, above default $500 floor
            timestamp=datetime.now(UTC),
            asset_id="token_123",
            event_title="Niche ok trade",
        )
        signal = await detector.analyze(ok_trade)
        assert signal is not None
        assert signal.is_niche_market is True
        assert signal.confidence == 0.2
    @pytest.mark.asyncio
    async def test_niche_min_trade_size_does_not_block_real_anomalies(
        self,
        mock_metadata_sync: AsyncMock,
        sample_metadata: MarketMetadata,
    ) -> None:
        """A trade that exceeds volume/book thresholds is never blocked by the niche guard."""
        mock_metadata_sync.get_market.return_value = sample_metadata
        detector = SizeAnomalyDetector(mock_metadata_sync)
        small_but_high_impact_trade = TradeEvent(
            market_id="market_abc123",
            trade_id="tx_small_impact",
            wallet_address="0xabc",
            side="BUY",
            outcome="Yes",
            outcome_index=0,
            price=Decimal("0.5"),
            size=Decimal("200"),  # $100 notional, below niche floor
            timestamp=datetime.now(UTC),
            asset_id="token_123",
            event_title="Small but high-impact",
        )
        # Provide small daily_volume so volume_impact exceeds 2% threshold
        signal = await detector.analyze(
            small_but_high_impact_trade, daily_volume=Decimal("1000")
        )
        assert signal is not None
        assert signal.volume_impact > 0.02
    @pytest.mark.asyncio
    async def test_analyze_no_anomaly(
        self,
@@ -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)
@@ -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
@@ -84,7 +84,10 @@ class TestTradeStreamHandler:
        """Test building subscription message without filters."""
        msg = handler._build_subscription_message()
-        assert msg == {"subscriptions": [{"topic": "activity", "type": "trades"}]}
+        assert msg == {
            "action": "subscribe",
            "subscriptions": [{"topic": "activity", "type": "trades"}],
        }
    def test_build_subscription_message_with_event_filter(self, on_trade_mock: AsyncMock) -> None:
        """Test building subscription message with event filter."""
@@ -110,11 +113,10 @@ class TestTradeStreamHandler:
    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",
+                "connection_id": "abc123",
                "type": "trades",
                "payload": {
                    "conditionId": "0xmarket",
                    "transactionHash": "0xtx",
@@ -142,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",
+                "connection_id": "abc123",
                "type": "comment_created",
                "payload": {"body": "Hello"},
            }
        )
@@ -156,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
@@ -175,8 +205,7 @@ class TestTradeStreamHandler:
        message = json.dumps(
            {
-                "topic": "activity",
+                "connection_id": "abc",
                "type": "trades",
                "payload": {
                    "conditionId": "0x",
                    "transactionHash": "0x",
@@ -231,14 +260,18 @@ class TestTradeStreamHandler:
        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"
@@ -289,8 +322,7 @@ class TestTradeStreamHandlerIntegration:
        trade_message = json.dumps(
            {
-                "topic": "activity",
+                "connection_id": "test-conn",
                "type": "trades",
                "payload": {
                    "conditionId": "0xtest",
                    "transactionHash": "0xtx",
@@ -333,7 +365,10 @@ class TestTradeStreamHandlerIntegration:
        mock_ws = MockWebSocket(handler, trade_message)
-        with patch("websockets.connect", AsyncMock(return_value=mock_ws)):
+        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)
@@ -327,6 +327,10 @@ class TestGetTransferLogs:
        await funding_tracer._get_transfer_logs(
            to_address=TEST_WALLET,
            token_address=USDC_BRIDGED,
            # Explicit numeric range so we stay inside one chunk and skip
            # the "latest" → block_number resolution path.
            from_block=1,
            to_block=8_000,
        )
        mock_w3.eth.get_logs.assert_called_once()
@@ -334,10 +338,15 @@ class TestGetTransferLogs:
        # Verify topics structure
        assert len(call_args["topics"]) == 3
-        assert call_args["topics"][0] == TRANSFER_EVENT_SIGNATURE.hex()
+        # The Transfer event topic must be 0x-prefixed; drpc rejects bare hex.
        assert call_args["topics"][0] == "0x" + TRANSFER_EVENT_SIGNATURE.hex().removeprefix("0x")
        assert call_args["topics"][0].startswith("0x")
        assert call_args["topics"][1] is None  # from (any)
        # to address should be padded to 32 bytes
        assert call_args["topics"][2].endswith(TEST_WALLET.lower().replace("0x", ""))
        # And the chunk bounds match what we asked for.
        assert call_args["fromBlock"] == 1
        assert call_args["toBlock"] == 8_000
    @pytest.mark.asyncio
    async def test_get_transfer_logs_respects_limit(
@@ -355,6 +364,8 @@ class TestGetTransferLogs:
            to_address=TEST_WALLET,
            token_address=USDC_BRIDGED,
            limit=3,
            from_block=1,
            to_block=8_000,
        )
        assert len(result) == 3
@@ -374,10 +385,242 @@ class TestGetTransferLogs:
        await funding_tracer._get_transfer_logs(
            to_address=TEST_WALLET,
            token_address=USDC_BRIDGED,
            from_block=1,
            to_block=8_000,
        )
        mock_fallback.eth.get_logs.assert_called_once()
    @pytest.mark.asyncio
    async def test_get_transfer_logs_chunks_large_ranges(
        self,
        funding_tracer: FundingTracer,
        mock_polygon_client: MagicMock,
    ) -> None:
        """Ranges wider than chunk_size are split into multiple eth_getLogs calls.
        This is the regression guard for the publicnode 10_000-block cap that
        was rejecting every funding trace before chunking landed.
        """
        mock_w3 = MagicMock()
        mock_w3.eth.get_logs = AsyncMock(return_value=[])
        mock_polygon_client._w3 = mock_w3
        # 25_000 blocks at 9_000-per-chunk → 3 calls (9000 + 9000 + 7001).
        await funding_tracer._get_transfer_logs(
            to_address=TEST_WALLET,
            token_address=USDC_BRIDGED,
            from_block=1_000_000,
            to_block=1_025_000,
        )
        assert mock_w3.eth.get_logs.call_count == 3
        windows = [call[0][0] for call in mock_w3.eth.get_logs.call_args_list]
        assert windows[0]["fromBlock"] == 1_000_000
        assert windows[0]["toBlock"] == 1_008_999
        assert windows[1]["fromBlock"] == 1_009_000
        assert windows[1]["toBlock"] == 1_017_999
        assert windows[2]["fromBlock"] == 1_018_000
        assert windows[2]["toBlock"] == 1_025_000
        # No window exceeds the chunk size — that's what RPC providers reject.
        for win in windows:
            assert win["toBlock"] - win["fromBlock"] + 1 <= 9_000
    @pytest.mark.asyncio
    async def test_get_transfer_logs_stops_when_limit_hit_mid_walk(
        self,
        funding_tracer: FundingTracer,
        mock_polygon_client: MagicMock,
    ) -> None:
        """Walking should stop as soon as ``limit`` matches are gathered."""
        mock_w3 = MagicMock()
        # First chunk yields 5 logs, more than the limit, so subsequent chunks
        # must not be queried.
        mock_w3.eth.get_logs = AsyncMock(return_value=[MagicMock() for _ in range(5)])
        mock_polygon_client._w3 = mock_w3
        result = await funding_tracer._get_transfer_logs(
            to_address=TEST_WALLET,
            token_address=USDC_BRIDGED,
            limit=2,
            from_block=1_000_000,
            to_block=1_025_000,
        )
        assert len(result) == 2
        mock_w3.eth.get_logs.assert_called_once()
    @pytest.mark.asyncio
    async def test_get_transfer_logs_skips_failing_chunk(
        self,
        funding_tracer: FundingTracer,
        mock_polygon_client: MagicMock,
    ) -> None:
        """A flaky chunk must not abort the whole trace — we move on."""
        mock_w3 = MagicMock()
        good_log = MagicMock()
        responses: list[Any] = [
            RuntimeError("RPC hiccup"),
            [good_log],
        ]
        async def fake_get_logs(_params: dict[str, Any]) -> list[Any]:
            outcome = responses.pop(0)
            if isinstance(outcome, BaseException):
                raise outcome
            return outcome
        mock_w3.eth.get_logs = AsyncMock(side_effect=fake_get_logs)
        mock_polygon_client._w3 = mock_w3
        result = await funding_tracer._get_transfer_logs(
            to_address=TEST_WALLET,
            token_address=USDC_BRIDGED,
            from_block=1_000_000,
            to_block=1_018_000,  # forces 3 chunks; we exercise chunks 1+2
        )
        # The error chunk is skipped; the second chunk contributes one log.
        assert result == [dict(good_log)]
        assert mock_w3.eth.get_logs.call_count >= 2
    @pytest.mark.asyncio
    async def test_get_transfer_logs_resolves_latest_via_block_number(
        self,
        funding_tracer: FundingTracer,
        mock_polygon_client: MagicMock,
    ) -> None:
        """``to_block='latest'`` should resolve via ``eth.block_number``.
        And ``from_block=0`` should not become a full-history scan — it must
        be clamped to ``latest - max_lookback_blocks``.
        """
        async def _block_number_coro() -> int:
            return 5_000
        mock_eth = MagicMock()
        mock_eth.get_logs = AsyncMock(return_value=[])
        # Property-style awaitable: web3.py exposes block_number as a property
        # returning a coroutine, so each access must yield a fresh awaitable.
        type(mock_eth).block_number = property(  # type: ignore[misc]
            lambda _self: _block_number_coro()
        )
        mock_w3 = MagicMock()
        mock_w3.eth = mock_eth
        mock_polygon_client._w3 = mock_w3
        await funding_tracer._get_transfer_logs(
            to_address=TEST_WALLET,
            token_address=USDC_BRIDGED,
        )
        # block_number=5000 < chunk_size, so it's one chunk that bottoms at 0.
        mock_eth.get_logs.assert_called_once()
        call_args = mock_eth.get_logs.call_args[0][0]
        assert call_args["fromBlock"] == 0
        assert call_args["toBlock"] == 5_000
    @pytest.mark.asyncio
    async def test_get_transfer_logs_breaks_on_pruned_history(
        self,
        funding_tracer: FundingTracer,
        mock_polygon_client: MagicMock,
    ) -> None:
        """A pruned-history error must short-circuit the whole walk.
        Public Polygon RPCs prune log history. Once we walk past the cutoff,
        every subsequent chunk will raise the same error — keep walking and
        we just burn quota on guaranteed failures. The first such error must
        end the walk and return whatever we already collected.
        """
        mock_w3 = MagicMock()
        good_log = MagicMock()
        responses: list[Any] = [
            [good_log],
            RuntimeError(
                "{'code': -32701, 'message': 'History has been pruned for "
                "this block. To remove restrictions, order a dedicated full "
                "node here: https://www.allnodes.com/pol/host'}"
            ),
            # If the early-break logic is missing, this third chunk would
            # also be requested. The test asserts it isn't.
            [MagicMock()],
        ]
        async def fake_get_logs(_params: dict[str, Any]) -> list[Any]:
            outcome = responses.pop(0)
            if isinstance(outcome, BaseException):
                raise outcome
            return outcome
        mock_w3.eth.get_logs = AsyncMock(side_effect=fake_get_logs)
        mock_polygon_client._w3 = mock_w3
        # 3 chunks total. The pruned error fires on chunk #2; chunk #3 must
        # never be issued.
        result = await funding_tracer._get_transfer_logs(
            to_address=TEST_WALLET,
            token_address=USDC_BRIDGED,
            from_block=1_000_000,
            to_block=1_027_000,
        )
        assert result == [dict(good_log)]
        assert mock_w3.eth.get_logs.call_count == 2
    @pytest.mark.asyncio
    async def test_get_transfer_logs_default_lookback_fits_pruned_horizon(
        self,
    ) -> None:
        """Default ``max_lookback_blocks`` must stay inside what public RPCs serve.
        publicnode prunes after ~100k blocks. If we default to 1.3M, every
        funding trace blows through the archive horizon and produces nothing
        but pruned-history warnings. Pin the default at <= 100k as a
        regression guard.
        """
        from polymarket_insider_tracker.profiler.funding import (
            DEFAULT_MAX_LOOKBACK_BLOCKS,
        )
        assert DEFAULT_MAX_LOOKBACK_BLOCKS <= 100_000
    @pytest.mark.asyncio
    async def test_get_transfer_logs_topic_is_0x_prefixed(
        self,
        funding_tracer: FundingTracer,
        mock_polygon_client: MagicMock,
    ) -> None:
        """The Transfer event topic passed to ``eth_getLogs`` must begin with ``0x``.
        ``HexBytes.hex()`` returns a bare hex string. publicnode tolerates
        that, but stricter providers like drpc (our fallback) reject it with
        ``invalid argument 0: hex string without 0x prefix`` and every chunk
        in the trace fails. This guards against regressing back to the
        bare-hex form.
        """
        mock_w3 = MagicMock()
        mock_w3.eth.get_logs = AsyncMock(return_value=[])
        mock_polygon_client._w3 = mock_w3
        await funding_tracer._get_transfer_logs(
            to_address=TEST_WALLET,
            token_address=USDC_BRIDGED,
            from_block=1,
            to_block=8_000,
        )
        topics = mock_w3.eth.get_logs.call_args[0][0]["topics"]
        assert topics[0].startswith("0x")
        # And the topic also has to be 32 bytes (64 hex chars) as required by
        # the JSON-RPC spec.
        assert len(topics[0]) == 2 + 64
        # The padded `to` topic was already 0x-prefixed; double-check that
        # didn't regress either.
        assert topics[2].startswith("0x")
 class TestLogToFundingTransfer:
    """Tests for _log_to_funding_transfer method."""
@@ -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
@@ -44,6 +44,9 @@ def mock_settings():
    telegram.bot_token = None
    telegram.chat_id = None
    detector = MagicMock()
    detector.persist_assessments = False
    settings = MagicMock(spec=Settings)
    settings.redis = redis
    settings.database = database
@@ -51,6 +54,7 @@ def mock_settings():
    settings.polymarket = polymarket
    settings.discord = discord
    settings.telegram = telegram
    settings.detector = detector
    settings.dry_run = True
    return settings
@@ -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