style: apply ruff formatting

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

.env.example

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

README.md

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

alembic/env.py

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

alembic/versions/20260104_0000_initial_schema.py

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

docker-compose.yml

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

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

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

src/polymarket_insider_tracker/config.py

@@ -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,12 @@ 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 Settings(BaseSettings):

src/polymarket_insider_tracker/ingestor/metadata_sync.py

@@ -11,7 +11,7 @@ import logging
 from collections.abc import Callable
 from dataclasses import dataclass
 from datetime import UTC, datetime
-from enum import Enum
+from enum import StrEnum
 
 from redis.asyncio import Redis
 
@@ -27,7 +27,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"

src/polymarket_insider_tracker/ingestor/models.py

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

src/polymarket_insider_tracker/ingestor/websocket.py

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

src/polymarket_insider_tracker/pipeline.py

@@ -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,7 +29,14 @@ 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,
+    WalletProfileDTO,
+    WalletRepository,
+)
 
 if TYPE_CHECKING:
     from typing import Any
@@ -43,7 +50,7 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
-class PipelineState(str, Enum):
+class PipelineState(StrEnum):
     """Pipeline lifecycle states."""
 
     STOPPED = "stopped"
@@ -120,6 +127,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 +241,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)
@@ -365,6 +377,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 +398,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:

tests/ingestor/test_websocket.py

@@ -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",
-                "type": "trades",
+                "connection_id": "abc123",
                 "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",
-                "type": "comment_created",
+                "connection_id": "abc123",
                 "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",
-                "type": "trades",
+                "connection_id": "abc",
                 "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",
-                "type": "trades",
+                "connection_id": "test-conn",
                 "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)

tests/test_pipeline_persistence.py

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