fix: resolve .env.example variable interpolation and alembic env var mismatch (#62 )

.env files don't support shell variable expansion, so DATABASE_URL and REDIS_URL contained literal ${...} strings. Also alembic/env.py read SQLALCHEMY_DATABASE_URL instead of DATABASE_URL. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-09 00:04:42 +00:00
13 changed files with 306 additions and 760 deletions
@@ -2,103 +2,80 @@

 **Detect informed money before the market moves.**

-[![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.
+[![Python 3.11+](https://img.shields.io/badge/python-3.11+-blue.svg)](https://www.python.org/downloads/)

 ---

-## Quick Start (< 2 minutes)
+## The Opportunity

-### 1. Install
+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.

-```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]"
-```
+> "You don't need to predict the future, you need to track suspicious behavior."
+> — [@DidiTrading](https://x.com/DidiTrading)

-### 2. Start infrastructure
+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.

-```bash
-docker compose up -d           # PostgreSQL 15 + Redis 7
-docker compose ps              # wait for healthy
-```
-
-### 3. Configure
-
-```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. 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
-```
+**This repository builds that tool.**

 ---

-## Environment Variables
+## What This Does

-| 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 |
+The Polymarket Insider Tracker monitors prediction market trading activity in real-time and identifies patterns that suggest informed trading:

-No API keys are needed for basic operation — the Polymarket WebSocket and CLOB REST APIs are public.
+| 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.

 ---

-## What It Detects
+## How It Works

-| 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 |
+```
+┌─────────────────┐     ┌──────────────────┐     ┌────────────────────┐
+│  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" │
+                   └─────────────────────┘
+```

-Risk scoring combines signals with configurable weights (default threshold: 0.6). Multi-signal bonuses: 2 signals +20%, 3+ signals +30%.
+### Detection Algorithms

-### Sample Alert
+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
+
+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. **Sniper Cluster Detection**
+   - Uses DBSCAN clustering to find wallets that consistently enter markets within minutes of creation
+   - Identifies coordinated behavior patterns
+
+4. **Event Correlation**
+   - Cross-references trading activity with news feeds
+   - Detects positions opened 1-4 hours before related news breaks
+
+---
+
+## Sample Alert

 ```
 SUSPICIOUS ACTIVITY DETECTED
@@ -122,62 +99,207 @@ Confidence: HIGH (3/4 signals triggered)

 ---

-## Architecture
+## Quick Start

-```
-Polymarket WebSocket ──> Ingestor ──> Profiler ──> Detector ──> Alerter
-(wss://ws-live-data)    (trades)    (on-chain)   (scoring)   (Discord/TG)
-                                        |
-                                   Polygon RPC
-```
+### Prerequisites

-### Components
+- 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))

-| 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
+### Installation

 ```bash
-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)
+# 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
 ```

 ### 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, `--profile tools`) |
-| RedisInsight | 5540 | Redis admin UI (optional, `--profile tools`) |
+| 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
+```

 ---

-## Troubleshooting
+## Project Structure

-**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).
+```
+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
+```

-**Connection timeout / DNS errors**
-Verify `wss://ws-live-data.polymarket.com` is reachable from your network. Some corporate firewalls block WebSocket connections.
+---

-**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.
+## Roadmap

-**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.).
+### 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/
+```

 ---

@@ -190,6 +312,20 @@ 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.
@@ -5,16 +5,16 @@ Revises:
 Create Date: 2026-01-04 00:00:00.000000+00:00
 """

-from collections.abc import Sequence
+from typing import Sequence, Union

 import sqlalchemy as sa
 from alembic import op

 # revision identifiers, used by Alembic.
 revision: str = "001_initial"
-down_revision: str | None = None
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
+down_revision: Union[str, None] = None
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None


 def upgrade() -> None:
@@ -1,3 +1,5 @@
+version: "3.8"
+
 services:
  postgres:
    image: postgres:15
@@ -1,113 +0,0 @@
-# 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,14 +290,12 @@ class AlertFormatter:
                    wallet_line += f" \\(Age: {age_hours:.0f}h\\)"
        lines.append(wallet_line)

-        # Risk score — every numeric literal here must be escaped because
-        # MarkdownV2 treats `.` as a special character and rejects unescaped
-        # ones with `Bad Request: can't parse entities`.
-        score_str = self._escape_telegram_markdown(f"{assessment.weighted_score:.2f}")
-        lines.append(f"*Risk Score:* {score_str} \\({risk_level}\\)")
+        # Risk score
+        lines.append(f"*Risk Score:* {assessment.weighted_score:.2f} \\({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']})")
@@ -305,16 +303,14 @@ class AlertFormatter:
            lines.append(f"*Market:* {market_title_escaped}")

        # Trade details
-        usdc_value = self._escape_telegram_markdown(format_usdc(trade.notional_value))
-        price_str = self._escape_telegram_markdown(f"{trade.price:.3f}")
-        side_escaped = self._escape_telegram_markdown(trade.side)
-        outcome_escaped = self._escape_telegram_markdown(trade.outcome)
-        lines.append(f"*Trade:* {side_escaped} {outcome_escaped} @ \\${price_str} \\| {usdc_value}")
+        usdc_value = format_usdc(trade.notional_value).replace("$", "\\$")
+        lines.append(
+            f"*Trade:* {trade.side} {trade.outcome} @ \\${trade.price:.3f} \\| {usdc_value}"
+        )

        # Signals
        if signals:
-            signals_escaped = [self._escape_telegram_markdown(s) for s in signals]
-            lines.append(f"*Signals:* {', '.join(signals_escaped)}")
+            lines.append(f"*Signals:* {', '.join(signals)}")

        # Links
        lines.append("")
@@ -18,9 +18,7 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
 class DatabaseSettings(BaseSettings):
    """Database connection settings."""

-    model_config = SettingsConfigDict(
-        env_prefix="", env_file=".env", env_file_encoding="utf-8", extra="ignore"
-    )
+    model_config = SettingsConfigDict(env_prefix="")

    url: str = Field(
        alias="DATABASE_URL",
@@ -39,9 +37,7 @@ class DatabaseSettings(BaseSettings):
 class RedisSettings(BaseSettings):
    """Redis connection settings."""

-    model_config = SettingsConfigDict(
-        env_prefix="", env_file=".env", env_file_encoding="utf-8", extra="ignore"
-    )
+    model_config = SettingsConfigDict(env_prefix="")

    url: str = Field(
        default="redis://localhost:6379",
@@ -61,9 +57,7 @@ class RedisSettings(BaseSettings):
 class PolygonSettings(BaseSettings):
    """Polygon blockchain RPC settings."""

-    model_config = SettingsConfigDict(
-        env_prefix="POLYGON_", env_file=".env", env_file_encoding="utf-8", extra="ignore"
-    )
+    model_config = SettingsConfigDict(env_prefix="POLYGON_")

    rpc_url: str = Field(
        default="https://polygon-rpc.com",
@@ -90,9 +84,7 @@ class PolygonSettings(BaseSettings):
 class PolymarketSettings(BaseSettings):
    """Polymarket API settings."""

-    model_config = SettingsConfigDict(
-        env_prefix="POLYMARKET_", env_file=".env", env_file_encoding="utf-8", extra="ignore"
-    )
+    model_config = SettingsConfigDict(env_prefix="POLYMARKET_")

    ws_url: str = Field(
        default="wss://ws-subscriptions-clob.polymarket.com/ws/market",
@@ -117,9 +109,7 @@ class PolymarketSettings(BaseSettings):
 class DiscordSettings(BaseSettings):
    """Discord notification settings."""

-    model_config = SettingsConfigDict(
-        env_prefix="DISCORD_", env_file=".env", env_file_encoding="utf-8", extra="ignore"
-    )
+    model_config = SettingsConfigDict(env_prefix="DISCORD_")

    webhook_url: SecretStr | None = Field(
        default=None,
@@ -130,15 +120,13 @@ class DiscordSettings(BaseSettings):
    @property
    def enabled(self) -> bool:
        """Check if Discord notifications are enabled."""
-        return self.webhook_url is not None and bool(self.webhook_url.get_secret_value().strip())
+        return self.webhook_url is not None


 class TelegramSettings(BaseSettings):
    """Telegram notification settings."""

-    model_config = SettingsConfigDict(
-        env_prefix="TELEGRAM_", env_file=".env", env_file_encoding="utf-8", extra="ignore"
-    )
+    model_config = SettingsConfigDict(env_prefix="TELEGRAM_")

    bot_token: SecretStr | None = Field(
        default=None,
@@ -154,12 +142,7 @@ class TelegramSettings(BaseSettings):
    @property
    def enabled(self) -> bool:
        """Check if Telegram notifications are enabled."""
-        return (
-            self.bot_token is not None
-            and bool(self.bot_token.get_secret_value().strip())
-            and self.chat_id is not None
-            and bool(self.chat_id.strip())
-        )
+        return self.bot_token is not None and self.chat_id is not None


 class Settings(BaseSettings):
@@ -11,7 +11,7 @@ import logging
 from collections.abc import Callable
 from dataclasses import dataclass
 from datetime import UTC, datetime
-from enum import StrEnum
+from enum import Enum

 from redis.asyncio import Redis

@@ -27,7 +27,7 @@ DEFAULT_CACHE_TTL_SECONDS = 600  # 10 minutes
 DEFAULT_REDIS_KEY_PREFIX = "polymarket:market:"


-class SyncState(StrEnum):
+class SyncState(str, Enum):
    """State of the metadata synchronizer."""

    STOPPED = "stopped"
@@ -86,7 +86,7 @@ class Orderbook:
    bids: tuple[OrderbookLevel, ...]
    asks: tuple[OrderbookLevel, ...]
    tick_size: Decimal
-    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
+    timestamp: datetime = field(default_factory=datetime.utcnow)

    @classmethod
    def from_clob_orderbook(cls, orderbook: Any) -> "Orderbook":
@@ -9,9 +9,8 @@ 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

@@ -150,14 +149,14 @@ class TradeStreamHandler:
        elif self._market_filter:
            subscription["filters"] = json.dumps({"market_slug": self._market_filter})

-        return {"action": "subscribe", "subscriptions": [subscription]}
+        return {"subscriptions": [subscription]}

    async def _connect(self) -> ClientConnection:
        """Establish WebSocket connection."""
        await self._set_state(ConnectionState.CONNECTING)

        try:
-            ws = await ws_connect(
+            ws = await websockets.connect(
                self._host,
                ping_interval=self._ping_interval,
                ping_timeout=self._ping_interval * 2,
@@ -183,13 +182,12 @@ class TradeStreamHandler:
        try:
            data = json.loads(message)

-            # 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
-            ):
+            # 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", {})
                trade = TradeEvent.from_websocket_message(payload)

                self._stats.trades_received += 1
@@ -209,7 +207,8 @@ class TradeStreamHandler:
                    logger.error("Error in trade callback: %s", e)

            else:
-                logger.debug("Received non-trade message: %s", str(data)[:120])
+                # Log other message types for debugging
+                logger.debug("Received message: topic=%s type=%s", topic, msg_type)

        except json.JSONDecodeError as e:
            logger.warning("Invalid JSON message: %s", e)
@@ -228,7 +227,7 @@ class TradeStreamHandler:
                else:
                    logger.debug("Received binary message (%d bytes)", len(message))

-        except ConnectionClosed as e:
+        except websockets.ConnectionClosed as e:
            logger.warning("Connection closed: %s", e)
            raise
        except Exception as e:
@@ -285,7 +284,7 @@ class TradeStreamHandler:
            while self._running:
                try:
                    await self._listen(self._ws)
-                except (ConnectionClosed, Exception) as e:
+                except (websockets.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 StrEnum
+from enum import Enum
 from typing import TYPE_CHECKING

 from redis.asyncio import Redis
@@ -29,14 +29,7 @@ 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
@@ -50,7 +43,7 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)


-class PipelineState(StrEnum):
+class PipelineState(str, Enum):
    """Pipeline lifecycle states."""

    STOPPED = "stopped"
@@ -127,7 +120,6 @@ 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
@@ -241,10 +233,6 @@ 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)
@@ -377,10 +365,6 @@ 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,
@@ -398,63 +382,6 @@ 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:
@@ -409,27 +409,12 @@ 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, MarkdownV2-escaped."""
+        """Test that Telegram message includes risk score."""
        formatter = AlertFormatter()
        result = formatter.format(high_risk_assessment)
-        # MarkdownV2 requires `.` to be escaped, so 0.82 becomes 0\.82.
-        assert "0\\.82" in result.telegram_markdown
+        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()
@@ -84,10 +84,7 @@ class TestTradeStreamHandler:
        """Test building subscription message without filters."""
        msg = handler._build_subscription_message()

-        assert msg == {
-            "action": "subscribe",
-            "subscriptions": [{"topic": "activity", "type": "trades"}],
-        }
+        assert msg == {"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."""
@@ -113,10 +110,11 @@ class TestTradeStreamHandler:
    async def test_handle_message_trade(
        self, handler: TradeStreamHandler, on_trade_mock: AsyncMock
    ) -> None:
-        """Test handling a valid trade message (payload-based routing)."""
+        """Test handling a valid trade message."""
        message = json.dumps(
            {
-                "connection_id": "abc123",
+                "topic": "activity",
+                "type": "trades",
                "payload": {
                    "conditionId": "0xmarket",
                    "transactionHash": "0xtx",
@@ -144,10 +142,11 @@ class TestTradeStreamHandler:
    async def test_handle_message_non_trade(
        self, handler: TradeStreamHandler, on_trade_mock: AsyncMock
    ) -> None:
-        """Test handling a non-trade message (no transactionHash/proxyWallet)."""
+        """Test handling a non-trade message."""
        message = json.dumps(
            {
-                "connection_id": "abc123",
+                "topic": "comments",
+                "type": "comment_created",
                "payload": {"body": "Hello"},
            }
        )
@@ -157,35 +156,6 @@ 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
@@ -205,7 +175,8 @@ class TestTradeStreamHandler:

        message = json.dumps(
            {
-                "connection_id": "abc",
+                "topic": "activity",
+                "type": "trades",
                "payload": {
                    "conditionId": "0x",
                    "transactionHash": "0x",
@@ -260,18 +231,14 @@ class TestTradeStreamHandler:
        mock_ws = AsyncMock()
        mock_ws.send = AsyncMock()

-        with patch(
-            "polymarket_insider_tracker.ingestor.websocket.ws_connect",
-            AsyncMock(return_value=mock_ws),
-        ):
+        with patch("websockets.connect", AsyncMock(return_value=mock_ws)):
            ws = await handler._connect()

            assert ws is mock_ws
            mock_ws.send.assert_called_once()

-            # Verify subscription message includes action: subscribe
+            # Verify subscription message
            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"
@@ -322,7 +289,8 @@ class TestTradeStreamHandlerIntegration:

        trade_message = json.dumps(
            {
-                "connection_id": "test-conn",
+                "topic": "activity",
+                "type": "trades",
                "payload": {
                    "conditionId": "0xtest",
                    "transactionHash": "0xtx",
@@ -365,10 +333,7 @@ class TestTradeStreamHandlerIntegration:

        mock_ws = MockWebSocket(handler, trade_message)

-        with patch(
-            "polymarket_insider_tracker.ingestor.websocket.ws_connect",
-            AsyncMock(return_value=mock_ws),
-        ):
+        with patch("websockets.connect", AsyncMock(return_value=mock_ws)):
            # Run with timeout to prevent hanging
            try:
                await asyncio.wait_for(handler.start(), timeout=1.0)
@@ -1,334 +0,0 @@
-"""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