Compare commits

..

1 Commits

Author SHA1 Message Date
pselamy 12bd40fa69 fix: replace (str, Enum) with StrEnum to resolve ruff UP042
StrEnum (Python 3.11+) is the modern replacement for the (str, Enum)
pattern. Updates SyncState and PipelineState enums.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-10 01:17:34 +00:00
30 changed files with 364 additions and 2414 deletions
+2 -2
View File
@@ -6,12 +6,12 @@ POSTGRES_USER=tracker
POSTGRES_PASSWORD=dev_password
# Constructed database URL (for application use)
DATABASE_URL=postgresql://tracker:dev_password@localhost:5432/polymarket_tracker
DATABASE_URL=postgresql://${POSTGRES_USER}:${POSTGRES_PASSWORD}@${POSTGRES_HOST}:${POSTGRES_PORT}/${POSTGRES_DB}
# Redis Configuration
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_URL=redis://localhost:6379
REDIS_URL=redis://${REDIS_HOST}:${REDIS_PORT}
# Optional: Development tool ports
ADMINER_PORT=8080
-29
View File
@@ -1,29 +0,0 @@
# Changelog
All notable changes to this project are documented in this file.
The format is loosely based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
## [Unreleased]
### Added
- **Risk-assessment persistence**: every signal-bearing trade now writes a row
to the new `risk_assessments` table, regardless of whether the assessment
meets the alert threshold. This is the ground-truth log future backtests will
read instead of grepping `alerts.log` / `journalctl`.
- Pipeline: `Pipeline._score_and_alert` calls `Pipeline._persist_assessment`
for every assessment; failures are caught and never block alert dispatch.
- Storage: new `RiskAssessmentModel`, `RiskAssessmentDTO`, and
`RiskAssessmentRepository` (alembic migration shipped previously).
- Config: `DETECTOR_PERSIST_ASSESSMENTS` env var (default `true`) controls
the write path so it can be disabled without code changes.
- Tests: `tests/test_persist_assessment.py` covers (a) sub-threshold rows are
persisted with `should_alert=False` and dispatch is skipped, and (b) DB
failures during persistence do not block dispatching.
### Changed
- Alert threshold (`DETECTOR_ALERT_THRESHOLD`) is now fully env-driven; the
legacy hard-coded `0.6` default has been raised to `0.80` for production.
### Notes
- Backtest scripts can now source data from `risk_assessments` directly. The
`alerts.log` parsing path remains for one release as a fallback.
-21
View File
@@ -1,21 +0,0 @@
MIT License
Copyright (c) 2026 Patrick Selamy
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
+250 -114
View File
@@ -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.
+1 -1
View File
@@ -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("DATABASE_URL")
database_url = os.environ.get("SQLALCHEMY_DATABASE_URL")
if database_url:
config.set_main_option("sqlalchemy.url", database_url)
@@ -5,16 +5,16 @@ Revises:
Create Date: 2026-01-04 00:00:00.000000+00:00
"""
from 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,64 +0,0 @@
"""Risk assessment persistence layer.
Adds the `risk_assessments` table — one row per signal-bearing trade —
so future backtests can rebuild ground truth without grepping the
systemd log or hammering the public data-api.
Revision ID: 002_risk_assessments
Revises: 001_initial
Create Date: 2026-05-22 11:30:00.000000+00:00
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
revision: str = "002_risk_assessments"
down_revision: str | None = "001_initial"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
op.create_table(
"risk_assessments",
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
sa.Column("assessment_id", sa.String(36), nullable=False),
sa.Column("trade_id", sa.String(80), nullable=False),
sa.Column("wallet_address", sa.String(42), nullable=False),
sa.Column("market_id", sa.String(80), nullable=False),
sa.Column("asset_id", sa.String(80), nullable=True),
sa.Column("side", sa.String(8), nullable=False),
sa.Column("outcome", sa.String(120), nullable=True),
sa.Column("outcome_index", sa.Integer(), nullable=True),
sa.Column("price", sa.Numeric(10, 6), nullable=False),
sa.Column("size", sa.Numeric(20, 6), nullable=False),
sa.Column("notional_usdc", sa.Numeric(20, 6), nullable=False),
sa.Column("trade_timestamp", sa.DateTime(timezone=True), nullable=False),
sa.Column("weighted_score", sa.Numeric(4, 3), nullable=False),
sa.Column("signals_triggered", sa.Integer(), nullable=False),
sa.Column("fresh_wallet_confidence", sa.Numeric(4, 3), nullable=True),
sa.Column("size_anomaly_confidence", sa.Numeric(4, 3), nullable=True),
sa.Column("is_niche_market", sa.Boolean(), nullable=True),
sa.Column("volume_impact", sa.Numeric(8, 4), nullable=True),
sa.Column("book_impact", sa.Numeric(8, 4), nullable=True),
sa.Column("wallet_age_hours", sa.Numeric(10, 2), nullable=True),
sa.Column("should_alert", sa.Boolean(), nullable=False),
sa.Column("threshold_at_eval", sa.Numeric(4, 3), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("assessment_id"),
)
op.create_index("idx_risk_assessments_wallet", "risk_assessments", ["wallet_address"])
op.create_index("idx_risk_assessments_market", "risk_assessments", ["market_id"])
op.create_index("idx_risk_assessments_trade_ts", "risk_assessments", ["trade_timestamp"])
op.create_index("idx_risk_assessments_score", "risk_assessments", ["weighted_score"])
def downgrade() -> None:
op.drop_index("idx_risk_assessments_score", table_name="risk_assessments")
op.drop_index("idx_risk_assessments_trade_ts", table_name="risk_assessments")
op.drop_index("idx_risk_assessments_market", table_name="risk_assessments")
op.drop_index("idx_risk_assessments_wallet", table_name="risk_assessments")
op.drop_table("risk_assessments")
+2
View File
@@ -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("")
+8 -53
View File
@@ -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,39 +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())
)
class DetectorSettings(BaseSettings):
"""Risk-scorer / detector tuning."""
model_config = SettingsConfigDict(
env_prefix="DETECTOR_", env_file=".env", env_file_encoding="utf-8", extra="ignore"
)
alert_threshold: float = Field(
default=0.80,
alias="DETECTOR_ALERT_THRESHOLD",
description="Minimum weighted score required to trigger an alert",
ge=0.0,
le=1.0,
)
dedup_window_seconds: int = Field(
default=3600,
alias="DETECTOR_DEDUP_WINDOW_SECONDS",
description="Per-(wallet, market) dedup window in seconds",
ge=0,
)
persist_assessments: bool = Field(
default=True,
alias="DETECTOR_PERSIST_ASSESSMENTS",
description="Write every signal-bearing risk assessment to the database",
)
return self.bot_token is not None and self.chat_id is not None
class Settings(BaseSettings):
@@ -218,7 +174,6 @@ class Settings(BaseSettings):
polymarket: PolymarketSettings = Field(default_factory=PolymarketSettings)
discord: DiscordSettings = Field(default_factory=DiscordSettings)
telegram: TelegramSettings = Field(default_factory=TelegramSettings)
detector: DetectorSettings = Field(default_factory=DetectorSettings)
# Application settings
log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = Field(
@@ -19,12 +19,8 @@ from polymarket_insider_tracker.ingestor.models import TradeEvent
logger = logging.getLogger(__name__)
# Default configuration. The threshold lifted from 0.6 to 0.80 after the
# first cost-adjusted backtest showed everything below 0.85 was follower-PnL
# negative under realistic taker fees + half-cent slippage. 0.80 keeps a small
# margin below 0.85+ so we don't drop borderline-high signals on a hard cliff.
# Override at runtime via DETECTOR_ALERT_THRESHOLD env var.
DEFAULT_ALERT_THRESHOLD = 0.80
# Default configuration
DEFAULT_ALERT_THRESHOLD = 0.6
DEFAULT_DEDUP_WINDOW_SECONDS = 3600 # 1 hour
DEFAULT_REDIS_KEY_PREFIX = "polymarket:dedup:"
@@ -17,9 +17,6 @@ logger = logging.getLogger(__name__)
DEFAULT_VOLUME_THRESHOLD = 0.02 # 2% of daily volume
DEFAULT_BOOK_THRESHOLD = 0.05 # 5% of order book depth
DEFAULT_NICHE_VOLUME_THRESHOLD = Decimal("50000") # $50k daily volume
# Niche-only path requires real trade size; below this, suppress the base
# 0.2 confidence so we don't flood alerts with low-value niche trades.
DEFAULT_NICHE_MIN_TRADE_SIZE = Decimal("500") # USDC notional
# Niche market categories - markets in these categories with low specificity
# are more likely to have insider information value
@@ -62,7 +59,6 @@ class SizeAnomalyDetector:
volume_threshold: float = DEFAULT_VOLUME_THRESHOLD,
book_threshold: float = DEFAULT_BOOK_THRESHOLD,
niche_volume_threshold: Decimal = DEFAULT_NICHE_VOLUME_THRESHOLD,
niche_min_trade_size: Decimal = DEFAULT_NICHE_MIN_TRADE_SIZE,
) -> None:
"""Initialize the size anomaly detector.
@@ -71,15 +67,11 @@ class SizeAnomalyDetector:
volume_threshold: Threshold for volume impact (default 0.02 = 2%).
book_threshold: Threshold for book impact (default 0.05 = 5%).
niche_volume_threshold: Volume below which market is niche ($50k).
niche_min_trade_size: Minimum trade notional ($) to allow a
niche-only signal. Below this, niche-only path is suppressed
to avoid flooding alerts with low-value trades.
"""
self._metadata_sync = metadata_sync
self._volume_threshold = volume_threshold
self._book_threshold = book_threshold
self._niche_volume_threshold = niche_volume_threshold
self._niche_min_trade_size = niche_min_trade_size
async def analyze(
self,
@@ -137,18 +129,6 @@ class SizeAnomalyDetector:
exceeds_volume = volume_impact > self._volume_threshold
exceeds_book = book_impact > self._book_threshold
# Niche-only signals require a minimum trade size; otherwise we'd
# flood alerts with every tiny trade in any niche-prone category.
niche_only = is_niche and not exceeds_volume and not exceeds_book
if niche_only and trade_size < self._niche_min_trade_size:
logger.debug(
"Trade %s niche-only but size %s < min %s, skipping",
trade.trade_id,
trade_size,
self._niche_min_trade_size,
)
return None
if not exceeds_volume and not exceeds_book and not is_niche:
logger.debug(
"Trade %s does not exceed thresholds: volume=%.4f, book=%.4f",
@@ -1,205 +0,0 @@
"""Gamma API client for Polymarket market volume / liquidity data.
The CLOB API does not expose 24h volume or liquidity. The public Gamma API
(https://gamma-api.polymarket.com) does, with no auth required. This module
fetches the volume/liquidity snapshot keyed by condition_id so the
size_anomaly detector can do real ratio math instead of falling back to
the niche-base 0.2 confidence floor.
"""
from __future__ import annotations
import asyncio
import logging
from dataclasses import dataclass
from decimal import Decimal, InvalidOperation
import httpx
logger = logging.getLogger(__name__)
DEFAULT_HOST = "https://gamma-api.polymarket.com"
DEFAULT_TIMEOUT_SECONDS = 15.0
# Gamma /markets enforces a server-side max of 100 per page even when a
# higher `limit` is sent. Using 100 lines our page size up with the actual
# response so pagination doesn't bail out after the first page.
DEFAULT_PAGE_LIMIT = 100
# Gamma also caps `offset` around 10000 for this collection. Combined with
# the 100/page limit that gives ~10k markets max, sequential — way too slow
# at default sync interval. We sort by 24h volume desc and only walk the
# top N pages, since markets with zero recent volume don't need a real
# ratio anyway (the niche path handles them).
DEFAULT_MAX_PAGES = 50 # 50 * 100 = 5000 most-traded markets per sync
DEFAULT_PAGE_CONCURRENCY = 5
DEFAULT_MAX_RETRIES = 3
DEFAULT_RETRY_BASE_DELAY_SECONDS = 1.0
@dataclass(frozen=True)
class GammaMarketStats:
"""Volume / liquidity snapshot for a single market from gamma-api."""
condition_id: str
daily_volume: Decimal | None
weekly_volume: Decimal | None
monthly_volume: Decimal | None
total_volume: Decimal | None
liquidity: Decimal | None
def _to_decimal(value: object) -> Decimal | None:
if value is None or value == "":
return None
try:
return Decimal(str(value))
except (InvalidOperation, ValueError, TypeError):
return None
def _parse_market(raw: dict[str, object]) -> GammaMarketStats | None:
cid = raw.get("conditionId")
if not cid or not isinstance(cid, str):
return None
return GammaMarketStats(
condition_id=cid,
daily_volume=_to_decimal(raw.get("volume24hr")),
weekly_volume=_to_decimal(raw.get("volume1wk")),
monthly_volume=_to_decimal(raw.get("volume1mo")),
total_volume=_to_decimal(raw.get("volumeNum") or raw.get("volume")),
liquidity=_to_decimal(raw.get("liquidityNum") or raw.get("liquidity")),
)
class GammaClientError(Exception):
"""Raised when gamma-api returns an unrecoverable error."""
class GammaClient:
"""Async client for the public gamma-api markets endpoint.
Provides batched, paginated reads of every active market with their
24h/weekly/monthly volume and current liquidity. Designed to be called
from MarketMetadataSync once per sync interval; results are merged into
Redis-cached MarketMetadata objects.
"""
def __init__(
self,
*,
host: str = DEFAULT_HOST,
timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
page_limit: int = DEFAULT_PAGE_LIMIT,
max_pages: int = DEFAULT_MAX_PAGES,
page_concurrency: int = DEFAULT_PAGE_CONCURRENCY,
max_retries: int = DEFAULT_MAX_RETRIES,
retry_base_delay_seconds: float = DEFAULT_RETRY_BASE_DELAY_SECONDS,
) -> None:
self._host = host.rstrip("/")
self._timeout = timeout_seconds
self._page_limit = page_limit
self._max_pages = max_pages
self._page_concurrency = page_concurrency
self._max_retries = max_retries
self._retry_base = retry_base_delay_seconds
async def _get_with_retry(
self,
client: httpx.AsyncClient,
path: str,
params: dict[str, object],
) -> list[dict[str, object]]:
last_exc: Exception | None = None
delay = self._retry_base
for attempt in range(self._max_retries):
try:
resp = await client.get(path, params=params)
resp.raise_for_status()
payload = resp.json()
if not isinstance(payload, list):
raise GammaClientError(
f"Unexpected gamma response shape for {path}: {type(payload).__name__}"
)
return payload
except (httpx.HTTPError, ValueError) as exc:
last_exc = exc
logger.warning(
"gamma %s attempt %d/%d failed: %s",
path,
attempt + 1,
self._max_retries,
exc,
)
if attempt < self._max_retries - 1:
await asyncio.sleep(delay)
delay *= 2
raise GammaClientError(
f"gamma {path} failed after {self._max_retries} attempts: {last_exc}"
)
async def get_active_market_stats(self) -> dict[str, GammaMarketStats]:
"""Fetch volume/liquidity for the most-traded active markets.
Walks up to `max_pages` pages of `page_limit` markets each, sorted
by 24h volume descending, with bounded concurrency. Markets beyond
that window have effectively zero recent volume — the size_anomaly
niche path handles them without needing a ratio.
Returns:
Mapping condition_id -> GammaMarketStats.
"""
results: dict[str, GammaMarketStats] = {}
sem = asyncio.Semaphore(self._page_concurrency)
stop = asyncio.Event()
async with httpx.AsyncClient(
base_url=self._host,
timeout=self._timeout,
headers={"User-Agent": "polymarket-insider-tracker/0.1"},
) as client:
async def fetch_page(page_index: int) -> list[dict[str, object]]:
if stop.is_set():
return []
params = {
"limit": self._page_limit,
"offset": page_index * self._page_limit,
"active": "true",
"closed": "false",
"order": "volume24hr",
"ascending": "false",
}
async with sem:
if stop.is_set():
return []
try:
return await self._get_with_retry(client, "/markets", params)
except GammaClientError as exc:
# Gamma rejects offsets past its hard cap with a
# validation error; treat that as a clean stop.
logger.debug("gamma stop at page %d: %s", page_index, exc)
stop.set()
return []
tasks = [asyncio.create_task(fetch_page(i)) for i in range(self._max_pages)]
pages = await asyncio.gather(*tasks)
empty_streak = 0
for page in pages:
if not page:
empty_streak += 1
continue
empty_streak = 0
for raw in page:
if not isinstance(raw, dict):
continue
parsed = _parse_market(raw)
if parsed is not None:
results[parsed.condition_id] = parsed
if len(page) < self._page_limit:
# short page — we walked past the end of the active set
empty_streak += 1
if empty_streak >= 2:
break
logger.info("gamma sync: fetched stats for %d active markets", len(results))
return results
@@ -9,14 +9,13 @@ import contextlib
import json
import logging
from collections.abc import Callable
from dataclasses import dataclass, replace
from dataclasses import dataclass
from datetime import UTC, datetime
from enum import StrEnum
from redis.asyncio import Redis
from .clob_client import ClobClient
from .gamma_client import GammaClient, GammaClientError, GammaMarketStats
from .models import MarketMetadata
logger = logging.getLogger(__name__)
@@ -92,7 +91,6 @@ class MarketMetadataSync:
redis: Redis,
clob_client: ClobClient,
*,
gamma_client: GammaClient | None = None,
sync_interval_seconds: int = DEFAULT_SYNC_INTERVAL_SECONDS,
cache_ttl_seconds: int = DEFAULT_CACHE_TTL_SECONDS,
key_prefix: str = DEFAULT_REDIS_KEY_PREFIX,
@@ -104,8 +102,6 @@ class MarketMetadataSync:
Args:
redis: Redis async client for caching.
clob_client: CLOB API client for fetching markets.
gamma_client: Optional gamma-api client for volume/liquidity
enrichment. Defaults to a fresh GammaClient() instance.
sync_interval_seconds: Interval between syncs (default: 300 / 5 min).
cache_ttl_seconds: TTL for cached entries (default: 600 / 10 min).
key_prefix: Redis key prefix for market data.
@@ -114,7 +110,6 @@ class MarketMetadataSync:
"""
self._redis = redis
self._clob = clob_client
self._gamma = gamma_client or GammaClient()
self._sync_interval = sync_interval_seconds
self._cache_ttl = cache_ttl_seconds
self._key_prefix = key_prefix
@@ -221,18 +216,6 @@ class MarketMetadataSync:
self._set_state(SyncState.ERROR)
# Continue running - will retry on next interval
async def _fetch_gamma_stats(self) -> dict[str, GammaMarketStats]:
"""Fetch volume/liquidity stats from gamma-api.
Returns an empty dict on failure so a degraded gamma endpoint
does not stop CLOB metadata from being cached.
"""
try:
return await self._gamma.get_active_market_stats()
except (GammaClientError, Exception) as e:
logger.warning("gamma stats fetch failed (continuing without volume): %s", e)
return {}
async def _sync_all_markets(self) -> None:
"""Fetch all markets and cache them in Redis."""
self._set_state(SyncState.SYNCING)
@@ -240,27 +223,14 @@ class MarketMetadataSync:
self._stats.total_syncs += 1
try:
# Fetch CLOB markets and gamma volume snapshot in parallel
markets, gamma_stats = await asyncio.gather(
asyncio.to_thread(self._clob.get_markets, True),
self._fetch_gamma_stats(),
)
# Fetch markets from CLOB API (runs in thread pool for sync API)
markets = await asyncio.to_thread(self._clob.get_markets, True)
# Cache each market in Redis, enriched with gamma volume/liquidity
# Cache each market in Redis
cached_count = 0
enriched_count = 0
for market in markets:
try:
metadata = MarketMetadata.from_market(market)
stats = gamma_stats.get(metadata.condition_id)
if stats is not None:
metadata = replace(
metadata,
daily_volume=stats.daily_volume,
weekly_volume=stats.weekly_volume,
liquidity=stats.liquidity,
)
enriched_count += 1
await self._cache_market(metadata)
cached_count += 1
except Exception as e:
@@ -276,10 +246,7 @@ class MarketMetadataSync:
self._set_state(SyncState.IDLE)
logger.info(
"Synced %d markets (%d enriched with gamma volume) in %.2fs",
cached_count,
enriched_count,
self._stats.last_sync_duration_seconds,
f"Synced {cached_count} markets in {self._stats.last_sync_duration_seconds:.2f}s"
)
# Notify callback
@@ -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":
@@ -399,12 +399,6 @@ class MarketMetadata:
# Derived metadata
category: str = "other"
# Liquidity/volume snapshot (from gamma-api). All optional — older
# cache entries and CLOB-only sync results may not have these.
daily_volume: Decimal | None = None
weekly_volume: Decimal | None = None
liquidity: Decimal | None = None
# Cache metadata
last_updated: datetime = field(default_factory=lambda: datetime.now(UTC))
@@ -452,9 +446,6 @@ class MarketMetadata:
"active": self.active,
"closed": self.closed,
"category": self.category,
"daily_volume": str(self.daily_volume) if self.daily_volume is not None else None,
"weekly_volume": str(self.weekly_volume) if self.weekly_volume is not None else None,
"liquidity": str(self.liquidity) if self.liquidity is not None else None,
"last_updated": self.last_updated.isoformat(),
}
@@ -486,15 +477,6 @@ class MarketMetadata:
else:
last_updated = datetime.now(UTC)
def _opt_dec(key: str) -> Decimal | None:
raw = data.get(key)
if raw is None or raw == "":
return None
try:
return Decimal(str(raw))
except (ValueError, ArithmeticError):
return None
return cls(
condition_id=str(data["condition_id"]),
question=str(data.get("question", "")),
@@ -504,8 +486,5 @@ class MarketMetadata:
active=bool(data.get("active", True)),
closed=bool(data.get("closed", False)),
category=str(data.get("category", "other")),
daily_volume=_opt_dec("daily_volume"),
weekly_volume=_opt_dec("weekly_volume"),
liquidity=_opt_dec("liquidity"),
last_updated=last_updated,
)
@@ -9,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
+2 -143
View File
@@ -29,23 +29,13 @@ from polymarket_insider_tracker.ingestor.metadata_sync import MarketMetadataSync
from polymarket_insider_tracker.ingestor.websocket import TradeStreamHandler
from polymarket_insider_tracker.profiler.analyzer import WalletAnalyzer
from polymarket_insider_tracker.profiler.chain import PolygonClient
from polymarket_insider_tracker.profiler.funding import FundingTracer
from polymarket_insider_tracker.storage.database import DatabaseManager
from polymarket_insider_tracker.storage.repos import (
FundingRepository,
FundingTransferDTO,
RiskAssessmentDTO,
RiskAssessmentRepository,
WalletProfileDTO,
WalletRepository,
)
if TYPE_CHECKING:
from typing import Any
from polymarket_insider_tracker.detector.models import (
FreshWalletSignal,
RiskAssessment,
SizeAnomalySignal,
)
from polymarket_insider_tracker.ingestor.models import TradeEvent
@@ -130,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
@@ -244,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)
@@ -255,17 +240,7 @@ class Pipeline:
# Initialize Risk Scorer
logger.debug("Initializing risk scorer...")
self._risk_scorer = RiskScorer(
self._redis,
alert_threshold=settings.detector.alert_threshold,
dedup_window_seconds=settings.detector.dedup_window_seconds,
)
logger.info(
"RiskScorer threshold=%.2f dedup_window=%ds persist=%s",
settings.detector.alert_threshold,
settings.detector.dedup_window_seconds,
settings.detector.persist_assessments,
)
self._risk_scorer = RiskScorer(self._redis)
# Initialize Alerting
logger.debug("Initializing alerting components...")
@@ -390,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,
@@ -411,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:
@@ -489,19 +403,13 @@ class Pipeline:
return None
async def _score_and_alert(self, bundle: SignalBundle) -> None:
"""Score signals, persist the assessment, and send alert if above threshold."""
"""Score signals and send alert if threshold exceeded."""
if not self._risk_scorer or not self._alert_formatter or not self._alert_dispatcher:
return
# Get risk assessment
assessment = await self._risk_scorer.assess(bundle)
# Persist every signal-bearing assessment (not just delivered alerts).
# This is the ground-truth log future backtests will read instead of
# grepping systemd. Failure here must never block alerting.
if self._settings.detector.persist_assessments:
await self._persist_assessment(assessment)
if not assessment.should_alert:
logger.debug(
"Trade %s below alert threshold (score=%.2f)",
@@ -537,55 +445,6 @@ class Pipeline:
result.success_count + result.failure_count,
)
async def _persist_assessment(self, assessment: RiskAssessment) -> None:
"""Write the assessment row. Best-effort; never raises."""
if not self._db_manager:
return
from decimal import Decimal as _D
trade = assessment.trade_event
fresh = assessment.fresh_wallet_signal
size_sig = assessment.size_anomaly_signal
wallet_age: _D | None = None
if fresh is not None and fresh.wallet_profile.age_hours is not None:
wallet_age = _D(str(round(float(fresh.wallet_profile.age_hours), 2)))
dto = RiskAssessmentDTO(
assessment_id=assessment.assessment_id,
trade_id=trade.trade_id,
wallet_address=assessment.wallet_address.lower(),
market_id=assessment.market_id,
asset_id=getattr(trade, "asset_id", None) or None,
side=trade.side,
outcome=getattr(trade, "outcome", None) or None,
outcome_index=getattr(trade, "outcome_index", None),
price=trade.price,
size=trade.size,
notional_usdc=trade.notional_value,
trade_timestamp=trade.timestamp,
weighted_score=_D(str(round(assessment.weighted_score, 3))),
signals_triggered=assessment.signals_triggered,
fresh_wallet_confidence=(
_D(str(round(fresh.confidence, 3))) if fresh is not None else None
),
size_anomaly_confidence=(
_D(str(round(size_sig.confidence, 3))) if size_sig is not None else None
),
is_niche_market=size_sig.is_niche_market if size_sig is not None else None,
volume_impact=(
_D(str(round(size_sig.volume_impact, 4))) if size_sig is not None else None
),
book_impact=(_D(str(round(size_sig.book_impact, 4))) if size_sig is not None else None),
wallet_age_hours=wallet_age,
should_alert=assessment.should_alert,
threshold_at_eval=_D(str(round(self._settings.detector.alert_threshold, 3))),
)
try:
async with self._db_manager.get_async_session() as session:
repo = RiskAssessmentRepository(session)
await repo.insert(dto)
except Exception as e:
logger.warning("Failed to persist risk assessment %s: %s", assessment.assessment_id, e)
async def run(self) -> None:
"""Start the pipeline and run until interrupted.
@@ -26,46 +26,8 @@ logger = logging.getLogger(__name__)
USDC_BRIDGED = "0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174"
USDC_NATIVE = "0x3c499c542cEF5E3811e1192ce70d8cC03d5c3359"
# ERC20 Transfer event signature. ``HexBytes.hex()`` returns a *bare* hex
# string without the ``0x`` prefix; publicnode tolerates that, but stricter
# providers (e.g. drpc — which we use as the fallback) reject it with
# ``invalid argument 0: hex string without 0x prefix``. Always pass the
# 0x-prefixed form to ``eth_getLogs``.
# ERC20 Transfer event signature
TRANSFER_EVENT_SIGNATURE = AsyncWeb3.keccak(text="Transfer(address,address,uint256)")
TRANSFER_EVENT_TOPIC = "0x" + TRANSFER_EVENT_SIGNATURE.hex().removeprefix("0x")
# eth_getLogs block-range chunking. Most public Polygon RPCs (publicnode, ankr,
# llamarpc) cap the range at 10_000 blocks per call; pick a window slightly
# under the cap so off-by-one differences between providers don't trip us up.
DEFAULT_CHUNK_SIZE_BLOCKS = 9_000
# Polygon block time is ~2.0s. publicnode (the most common free RPC) prunes
# log history aggressively — empirically only ~100k blocks (~55 hours) are
# served before requests start returning "History has been pruned". We default
# to 80k blocks (~44 hours), which is more than enough for fresh-wallet
# funding traces (those wallets are by definition new) and fits comfortably
# inside what most public providers retain.
DEFAULT_MAX_LOOKBACK_BLOCKS = 80_000
# Substrings that, when present in an RPC error, indicate the chunk we just
# asked for is outside the provider's archive horizon. Walking further back
# is futile, so we stop the trace early instead of hammering every chunk.
_PRUNED_HISTORY_MARKERS: tuple[str, ...] = (
"history has been pruned",
"missing trie node",
"older than",
)
def _is_pruned_history_error(err: BaseException) -> bool:
"""Return True if the RPC error indicates pruned history.
Public Polygon nodes only retain a recent slice of log history. When we
walk back through that slice in chunks and hit the cutoff, every further
chunk will fail with the same message so we stop early instead of
burning quota on guaranteed failures.
"""
text = str(err).lower()
return any(marker in text for marker in _PRUNED_HISTORY_MARKERS)
class FundingTracer:
@@ -88,8 +50,6 @@ class FundingTracer:
*,
max_hops: int = 3,
usdc_addresses: list[str] | None = None,
chunk_size_blocks: int = DEFAULT_CHUNK_SIZE_BLOCKS,
max_lookback_blocks: int = DEFAULT_MAX_LOOKBACK_BLOCKS,
) -> None:
"""Initialize the funding tracer.
@@ -98,11 +58,6 @@ class FundingTracer:
entity_registry: Registry for entity classification. Creates default if None.
max_hops: Maximum hops to trace back (default 3).
usdc_addresses: USDC contract addresses to track. Uses defaults if None.
chunk_size_blocks: Block window size per eth_getLogs call. Public
Polygon RPCs cap at 10_000 blocks; default leaves a safety margin.
max_lookback_blocks: How far back to scan when caller passes
``from_block=0``. Default ~44 hours at 2s block time, which
fits inside the pruned-history horizon of most public RPCs.
"""
self.polygon_client = polygon_client
self.entity_registry = entity_registry or EntityRegistry()
@@ -110,8 +65,6 @@ class FundingTracer:
self._usdc_addresses = [
addr.lower() for addr in (usdc_addresses or [USDC_BRIDGED, USDC_NATIVE])
]
self._chunk_size_blocks = chunk_size_blocks
self._max_lookback_blocks = max_lookback_blocks
async def trace(
self,
@@ -256,144 +209,46 @@ class FundingTracer:
) -> list[dict[str, Any]]:
"""Get ERC20 Transfer event logs.
Public Polygon RPCs (publicnode, ankr, llamarpc) cap ``eth_getLogs`` at
10_000 blocks per call. To work around this we resolve the requested
range into a concrete block window (defaulting to the last
``max_lookback_blocks`` when caller passes ``from_block=0``) and walk
the window in chunks of ``chunk_size_blocks``, oldest-first, stopping
once ``limit`` matches are collected. Walking oldest-first preserves
the "first transfer" semantics expected by the funding chain tracer.
If a chunk comes back with a "history has been pruned" style error
the rest of the walk is short-circuited every subsequent chunk
would hit the same archive cutoff and there's no point burning quota
on guaranteed failures.
Args:
to_address: Filter by recipient address.
token_address: ERC20 token contract address.
limit: Maximum logs to return.
from_block: Starting block number (0 means
``latest - max_lookback_blocks``).
to_block: Ending block number ("latest" resolves to current head).
from_block: Starting block number.
to_block: Ending block number.
Returns:
List of log dictionaries, oldest first, capped at ``limit``.
List of log dictionaries.
"""
# Pad address to 32 bytes for topic filter
padded_to = "0x" + to_address.lower().replace("0x", "").zfill(64)
topics = [
TRANSFER_EVENT_TOPIC, # Transfer event (must be 0x-prefixed for drpc)
None, # from (any)
padded_to, # to (target address)
]
contract_address = AsyncWeb3.to_checksum_address(token_address)
start_block, end_block = await self._resolve_block_range(from_block, to_block)
if start_block > end_block:
return []
results: list[dict[str, Any]] = []
chunk_size = max(1, self._chunk_size_blocks)
chunk_start = start_block
while chunk_start <= end_block:
chunk_end = min(chunk_start + chunk_size - 1, end_block)
try:
chunk_logs = await self._fetch_logs_chunk(
contract_address=contract_address,
topics=topics,
from_block=chunk_start,
to_block=chunk_end,
)
except Exception as e:
if _is_pruned_history_error(e):
# The provider has dropped this slice of history. Walking
# further back will hit the same wall on every chunk;
# stop now and return what we already have.
logger.info(
"eth_getLogs chunk %d-%d outside archive horizon for %s; stopping trace",
chunk_start,
chunk_end,
to_address,
)
break
logger.warning(
"eth_getLogs chunk %d-%d failed for %s: %s",
chunk_start,
chunk_end,
to_address,
e,
)
# Skip this window and keep walking — partial data is better
# than aborting the whole trace on a single flaky chunk.
chunk_start = chunk_end + 1
continue
for log in chunk_logs:
results.append(dict(log))
if len(results) >= limit:
return results
chunk_start = chunk_end + 1
return results
async def _resolve_block_range(
self,
from_block: int | str,
to_block: int | str,
) -> tuple[int, int]:
"""Resolve symbolic block params to concrete numeric bounds.
``from_block=0`` (the historical default) is rewritten to
``latest - max_lookback_blocks`` so we don't try to scan all of Polygon.
"""
w3 = self._select_w3()
if isinstance(to_block, str):
await self.polygon_client._rate_limiter.acquire()
head = int(await w3.eth.block_number)
end = head
else:
end = int(to_block)
if isinstance(from_block, str):
# Treat any symbolic from-block (e.g. "earliest") as "go back
# max_lookback_blocks from end"; that's what callers actually want.
start = max(0, end - self._max_lookback_blocks)
elif from_block == 0:
start = max(0, end - self._max_lookback_blocks)
else:
start = int(from_block)
return start, end
async def _fetch_logs_chunk(
self,
contract_address: str,
topics: list[Any],
from_block: int,
to_block: int,
) -> list[Any]:
"""Issue a single bounded ``eth_getLogs`` call."""
await self.polygon_client._rate_limiter.acquire()
w3 = self._select_w3()
# Use the web3 instance from polygon client
w3 = (
self.polygon_client._w3
if self.polygon_client._primary_healthy
else (self.polygon_client._w3_fallback or self.polygon_client._w3)
)
# Get logs with Transfer event filtering by recipient
# Note: web3 typing is overly restrictive for block params
return await w3.eth.get_logs(
logs = await w3.eth.get_logs(
{
"address": contract_address,
"topics": topics,
"address": AsyncWeb3.to_checksum_address(token_address),
"topics": [
TRANSFER_EVENT_SIGNATURE.hex(), # Transfer event
None, # from (any)
padded_to, # to (target address)
],
"fromBlock": from_block, # type: ignore[typeddict-item]
"toBlock": to_block, # type: ignore[typeddict-item]
}
)
def _select_w3(self) -> AsyncWeb3:
"""Pick primary or fallback web3 instance based on health."""
if self.polygon_client._primary_healthy:
return self.polygon_client._w3
return self.polygon_client._w3_fallback or self.polygon_client._w3
# Convert to list of dicts and limit
result = [dict(log) for log in logs[:limit]]
return result
async def _log_to_funding_transfer(
self,
@@ -115,57 +115,3 @@ class WalletRelationshipModel(Base):
Index("idx_wallet_relationships_a", "wallet_a"),
Index("idx_wallet_relationships_b", "wallet_b"),
)
class RiskAssessmentModel(Base):
"""SQLAlchemy model for risk assessments.
One row per signal-bearing trade (i.e. trades that triggered at least one
detector). Captures everything a future backtest needs without going back
to the public API: trade identity, score, per-signal confidences, and
whether the alert was actually delivered (could be False due to dedup or
threshold).
"""
__tablename__ = "risk_assessments"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
assessment_id: Mapped[str] = mapped_column(String(36), unique=True, nullable=False)
# Trade identity
trade_id: Mapped[str] = mapped_column(String(80), nullable=False)
wallet_address: Mapped[str] = mapped_column(String(42), nullable=False)
market_id: Mapped[str] = mapped_column(String(80), nullable=False)
asset_id: Mapped[str | None] = mapped_column(String(80), nullable=True)
side: Mapped[str] = mapped_column(String(8), nullable=False)
outcome: Mapped[str | None] = mapped_column(String(120), nullable=True)
outcome_index: Mapped[int | None] = mapped_column(Integer, nullable=True)
price: Mapped[Decimal] = mapped_column(Numeric(10, 6), nullable=False)
size: Mapped[Decimal] = mapped_column(Numeric(20, 6), nullable=False)
notional_usdc: Mapped[Decimal] = mapped_column(Numeric(20, 6), nullable=False)
trade_timestamp: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
# Scoring
weighted_score: Mapped[Decimal] = mapped_column(Numeric(4, 3), nullable=False)
signals_triggered: Mapped[int] = mapped_column(Integer, nullable=False)
fresh_wallet_confidence: Mapped[Decimal | None] = mapped_column(Numeric(4, 3), nullable=True)
size_anomaly_confidence: Mapped[Decimal | None] = mapped_column(Numeric(4, 3), nullable=True)
is_niche_market: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
volume_impact: Mapped[Decimal | None] = mapped_column(Numeric(8, 4), nullable=True)
book_impact: Mapped[Decimal | None] = mapped_column(Numeric(8, 4), nullable=True)
wallet_age_hours: Mapped[Decimal | None] = mapped_column(Numeric(10, 2), nullable=True)
# Decision
should_alert: Mapped[bool] = mapped_column(Boolean, nullable=False)
threshold_at_eval: Mapped[Decimal] = mapped_column(Numeric(4, 3), nullable=False)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC)
)
__table_args__ = (
Index("idx_risk_assessments_wallet", "wallet_address"),
Index("idx_risk_assessments_market", "market_id"),
Index("idx_risk_assessments_trade_ts", "trade_timestamp"),
Index("idx_risk_assessments_score", "weighted_score"),
)
@@ -18,7 +18,6 @@ from sqlalchemy.dialects.sqlite import insert as sqlite_insert
from polymarket_insider_tracker.storage.models import (
FundingTransferModel,
RiskAssessmentModel,
WalletProfileModel,
WalletRelationshipModel,
)
@@ -511,107 +510,3 @@ class RelationshipRepository:
)
# SQLAlchemy Result does have rowcount but typing doesn't reflect it
return (result.rowcount or 0) > 0 # type: ignore[attr-defined]
@dataclass
class RiskAssessmentDTO:
"""Data transfer object for a persisted risk assessment.
Captures everything a future backtest needs without going back to
public APIs: trade identity, score, per-signal confidences, and
whether the alert was actually delivered.
"""
assessment_id: str
trade_id: str
wallet_address: str
market_id: str
asset_id: str | None
side: str
outcome: str | None
outcome_index: int | None
price: Decimal
size: Decimal
notional_usdc: Decimal
trade_timestamp: datetime
weighted_score: Decimal
signals_triggered: int
fresh_wallet_confidence: Decimal | None
size_anomaly_confidence: Decimal | None
is_niche_market: bool | None
volume_impact: Decimal | None
book_impact: Decimal | None
wallet_age_hours: Decimal | None
should_alert: bool
threshold_at_eval: Decimal
created_at: datetime | None = None
class RiskAssessmentRepository:
"""Repository for risk assessment data access."""
def __init__(self, session: AsyncSession) -> None:
self.session = session
async def insert(self, dto: RiskAssessmentDTO) -> RiskAssessmentDTO:
"""Insert a single assessment. Idempotent on assessment_id collisions."""
model = RiskAssessmentModel(
assessment_id=dto.assessment_id,
trade_id=dto.trade_id,
wallet_address=dto.wallet_address.lower(),
market_id=dto.market_id,
asset_id=dto.asset_id,
side=dto.side,
outcome=dto.outcome,
outcome_index=dto.outcome_index,
price=dto.price,
size=dto.size,
notional_usdc=dto.notional_usdc,
trade_timestamp=dto.trade_timestamp,
weighted_score=dto.weighted_score,
signals_triggered=dto.signals_triggered,
fresh_wallet_confidence=dto.fresh_wallet_confidence,
size_anomaly_confidence=dto.size_anomaly_confidence,
is_niche_market=dto.is_niche_market,
volume_impact=dto.volume_impact,
book_impact=dto.book_impact,
wallet_age_hours=dto.wallet_age_hours,
should_alert=dto.should_alert,
threshold_at_eval=dto.threshold_at_eval,
)
self.session.add(model)
await self.session.flush()
return dto
async def get_by_assessment_id(self, assessment_id: str) -> RiskAssessmentDTO | None:
result = await self.session.execute(
select(RiskAssessmentModel).where(RiskAssessmentModel.assessment_id == assessment_id)
)
model = result.scalar_one_or_none()
if model is None:
return None
return RiskAssessmentDTO(
assessment_id=model.assessment_id,
trade_id=model.trade_id,
wallet_address=model.wallet_address,
market_id=model.market_id,
asset_id=model.asset_id,
side=model.side,
outcome=model.outcome,
outcome_index=model.outcome_index,
price=model.price,
size=model.size,
notional_usdc=model.notional_usdc,
trade_timestamp=model.trade_timestamp,
weighted_score=model.weighted_score,
signals_triggered=model.signals_triggered,
fresh_wallet_confidence=model.fresh_wallet_confidence,
size_anomaly_confidence=model.size_anomaly_confidence,
is_niche_market=model.is_niche_market,
volume_impact=model.volume_impact,
book_impact=model.book_impact,
wallet_age_hours=model.wallet_age_hours,
should_alert=model.should_alert,
threshold_at_eval=model.threshold_at_eval,
created_at=model.created_at,
)
+2 -17
View File
@@ -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()
-87
View File
@@ -588,93 +588,6 @@ class TestAnalyzeMethod:
assert signal.is_niche_market is True
assert signal.confidence == 0.2 # niche_base
@pytest.mark.asyncio
async def test_analyze_niche_only_below_min_trade_size_skipped(
self,
mock_metadata_sync: AsyncMock,
sample_metadata: MarketMetadata,
) -> None:
"""Niche-only trades below the min trade size are suppressed."""
mock_metadata_sync.get_market.return_value = sample_metadata
detector = SizeAnomalyDetector(mock_metadata_sync)
tiny_trade = TradeEvent(
market_id="market_abc123",
trade_id="tx_tiny",
wallet_address="0xabc",
side="BUY",
outcome="Yes",
outcome_index=0,
price=Decimal("0.5"),
size=Decimal("100"), # $50 notional, below default $500 floor
timestamp=datetime.now(UTC),
asset_id="token_123",
event_title="Niche tiny trade",
)
signal = await detector.analyze(tiny_trade)
assert signal is None
@pytest.mark.asyncio
async def test_analyze_niche_only_at_min_trade_size_emits(
self,
mock_metadata_sync: AsyncMock,
sample_metadata: MarketMetadata,
) -> None:
"""Niche-only trades at or above the min trade size still emit."""
mock_metadata_sync.get_market.return_value = sample_metadata
detector = SizeAnomalyDetector(mock_metadata_sync)
ok_trade = TradeEvent(
market_id="market_abc123",
trade_id="tx_ok",
wallet_address="0xabc",
side="BUY",
outcome="Yes",
outcome_index=0,
price=Decimal("0.5"),
size=Decimal("2000"), # $1000 notional, above default $500 floor
timestamp=datetime.now(UTC),
asset_id="token_123",
event_title="Niche ok trade",
)
signal = await detector.analyze(ok_trade)
assert signal is not None
assert signal.is_niche_market is True
assert signal.confidence == 0.2
@pytest.mark.asyncio
async def test_niche_min_trade_size_does_not_block_real_anomalies(
self,
mock_metadata_sync: AsyncMock,
sample_metadata: MarketMetadata,
) -> None:
"""A trade that exceeds volume/book thresholds is never blocked by the niche guard."""
mock_metadata_sync.get_market.return_value = sample_metadata
detector = SizeAnomalyDetector(mock_metadata_sync)
small_but_high_impact_trade = TradeEvent(
market_id="market_abc123",
trade_id="tx_small_impact",
wallet_address="0xabc",
side="BUY",
outcome="Yes",
outcome_index=0,
price=Decimal("0.5"),
size=Decimal("200"), # $100 notional, below niche floor
timestamp=datetime.now(UTC),
asset_id="token_123",
event_title="Small but high-impact",
)
# Provide small daily_volume so volume_impact exceeds 2% threshold
signal = await detector.analyze(
small_but_high_impact_trade, daily_volume=Decimal("1000")
)
assert signal is not None
assert signal.volume_impact > 0.02
@pytest.mark.asyncio
async def test_analyze_no_anomaly(
self,
-249
View File
@@ -1,249 +0,0 @@
"""Tests for the gamma-api client."""
from __future__ import annotations
from decimal import Decimal
import httpx
import pytest
from polymarket_insider_tracker.ingestor import gamma_client as gamma_module
from polymarket_insider_tracker.ingestor.gamma_client import (
GammaClient,
GammaClientError,
GammaMarketStats,
_parse_market,
)
class TestParseMarket:
def test_parses_full_payload(self) -> None:
raw = {
"conditionId": "0xabc",
"volume24hr": "12345.67",
"volume1wk": "100000",
"volume1mo": "500000",
"volumeNum": "999999.5",
"liquidityNum": "42000",
}
stats = _parse_market(raw)
assert stats is not None
assert stats.condition_id == "0xabc"
assert stats.daily_volume == Decimal("12345.67")
assert stats.weekly_volume == Decimal("100000")
assert stats.monthly_volume == Decimal("500000")
assert stats.total_volume == Decimal("999999.5")
assert stats.liquidity == Decimal("42000")
def test_falls_back_to_alternative_keys(self) -> None:
raw = {
"conditionId": "0x1",
"volume24hr": "1",
"volume": "777",
"liquidity": "55",
}
stats = _parse_market(raw)
assert stats is not None
assert stats.total_volume == Decimal("777")
assert stats.liquidity == Decimal("55")
def test_handles_missing_numeric_fields(self) -> None:
stats = _parse_market({"conditionId": "0x2"})
assert stats is not None
assert stats.daily_volume is None
assert stats.liquidity is None
def test_drops_garbage_decimals(self) -> None:
stats = _parse_market(
{"conditionId": "0x3", "volume24hr": "not-a-number", "liquidityNum": ""}
)
assert stats is not None
assert stats.daily_volume is None
assert stats.liquidity is None
def test_rejects_missing_condition_id(self) -> None:
assert _parse_market({"volume24hr": "1"}) is None
assert _parse_market({"conditionId": ""}) is None
assert _parse_market({"conditionId": 123}) is None # type: ignore[arg-type]
def _make_client(
_handler: httpx.MockTransport,
*,
page_limit: int = 100,
max_pages: int = 5,
page_concurrency: int = 5,
max_retries: int = 1,
) -> GammaClient:
"""Build a GammaClient that constructs httpx.AsyncClient with the given transport.
GammaClient creates its own AsyncClient inside `get_active_market_stats`,
so we monkeypatch the AsyncClient factory in the module to inject the mock
transport.
"""
return GammaClient(
page_limit=page_limit,
max_pages=max_pages,
page_concurrency=page_concurrency,
max_retries=max_retries,
retry_base_delay_seconds=0.0,
)
@pytest.fixture
def patch_async_client(monkeypatch: pytest.MonkeyPatch):
"""Replace the AsyncClient used by gamma_client with one bound to a MockTransport."""
def _apply(handler: httpx.MockTransport) -> None:
original = gamma_module.httpx.AsyncClient
def factory(*args: object, **kwargs: object) -> httpx.AsyncClient:
kwargs["transport"] = handler # type: ignore[index]
return original(*args, **kwargs) # type: ignore[arg-type]
monkeypatch.setattr(gamma_module.httpx, "AsyncClient", factory)
return _apply
@pytest.mark.asyncio
async def test_get_active_market_stats_single_page(patch_async_client) -> None:
page_one = [
{"conditionId": "0xa", "volume24hr": "100", "liquidityNum": "10"},
{"conditionId": "0xb", "volume24hr": "200", "liquidityNum": "20"},
]
calls: list[dict[str, str]] = []
def handler(request: httpx.Request) -> httpx.Response:
calls.append(dict(request.url.params))
offset = int(request.url.params.get("offset", "0"))
if offset == 0:
return httpx.Response(200, json=page_one)
return httpx.Response(200, json=[])
transport = httpx.MockTransport(handler)
patch_async_client(transport)
client = _make_client(transport, page_limit=2, max_pages=3)
result = await client.get_active_market_stats()
assert set(result.keys()) == {"0xa", "0xb"}
assert isinstance(result["0xa"], GammaMarketStats)
assert result["0xa"].daily_volume == Decimal("100")
assert calls[0]["limit"] == "2"
assert calls[0]["order"] == "volume24hr"
assert calls[0]["ascending"] == "false"
@pytest.mark.asyncio
async def test_get_active_market_stats_short_page_stops(patch_async_client) -> None:
"""A page shorter than page_limit signals end-of-data after a small empty streak."""
page_zero = [{"conditionId": f"0x{i}", "volume24hr": str(i)} for i in range(5)]
page_one_short = [{"conditionId": "0xshort", "volume24hr": "1"}]
def handler(request: httpx.Request) -> httpx.Response:
offset = int(request.url.params.get("offset", "0"))
if offset == 0:
return httpx.Response(200, json=page_zero)
if offset == 5:
return httpx.Response(200, json=page_one_short)
return httpx.Response(200, json=[])
transport = httpx.MockTransport(handler)
patch_async_client(transport)
client = _make_client(transport, page_limit=5, max_pages=10)
result = await client.get_active_market_stats()
assert "0xshort" in result
assert len(result) == 6
@pytest.mark.asyncio
async def test_get_active_market_stats_offset_cap_clean_stop(
patch_async_client,
) -> None:
"""Gamma rejects offsets past its hard cap; that error is swallowed cleanly."""
def handler(request: httpx.Request) -> httpx.Response:
offset = int(request.url.params.get("offset", "0"))
if offset == 0:
return httpx.Response(200, json=[{"conditionId": "0xa", "volume24hr": "1"}])
return httpx.Response(400, json={"error": "offset too large"})
transport = httpx.MockTransport(handler)
patch_async_client(transport)
client = _make_client(transport, page_limit=1, max_pages=4, max_retries=1)
result = await client.get_active_market_stats()
assert "0xa" in result
@pytest.mark.asyncio
async def test_get_with_retry_recovers_after_transient_error(
patch_async_client,
) -> None:
"""Transient HTTP errors retry up to max_retries before giving up."""
state = {"attempts": 0}
def handler(request: httpx.Request) -> httpx.Response:
offset = int(request.url.params.get("offset", "0"))
if offset == 0:
state["attempts"] += 1
if state["attempts"] < 2:
return httpx.Response(503, json={"error": "transient"})
return httpx.Response(200, json=[{"conditionId": "0xrecover", "volume24hr": "1"}])
return httpx.Response(200, json=[])
transport = httpx.MockTransport(handler)
patch_async_client(transport)
client = _make_client(transport, page_limit=1, max_pages=2, max_retries=3)
result = await client.get_active_market_stats()
assert "0xrecover" in result
assert state["attempts"] == 2
@pytest.mark.asyncio
async def test_unexpected_response_shape_is_handled(patch_async_client) -> None:
"""A non-list payload becomes a clean stop, not a crash."""
def handler(_request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json={"unexpected": "shape"})
transport = httpx.MockTransport(handler)
patch_async_client(transport)
client = _make_client(transport, page_limit=1, max_pages=2, max_retries=1)
result = await client.get_active_market_stats()
assert result == {}
@pytest.mark.asyncio
async def test_skips_non_dict_entries(patch_async_client) -> None:
"""Defensive: server returning mixed-type list items shouldn't crash."""
def handler(request: httpx.Request) -> httpx.Response:
offset = int(request.url.params.get("offset", "0"))
if offset == 0:
return httpx.Response(
200,
json=[
{"conditionId": "0xa", "volume24hr": "1"},
"garbage",
None,
42,
],
)
return httpx.Response(200, json=[])
transport = httpx.MockTransport(handler)
patch_async_client(transport)
client = _make_client(transport, page_limit=4, max_pages=2)
result = await client.get_active_market_stats()
assert list(result.keys()) == ["0xa"]
def test_gamma_client_error_inherits_exception() -> None:
assert issubclass(GammaClientError, Exception)
+26 -67
View File
@@ -8,7 +8,6 @@ from unittest.mock import AsyncMock, MagicMock
import pytest
from polymarket_insider_tracker.ingestor.clob_client import ClobClient
from polymarket_insider_tracker.ingestor.gamma_client import GammaClient
from polymarket_insider_tracker.ingestor.metadata_sync import (
DEFAULT_CACHE_TTL_SECONDS,
DEFAULT_REDIS_KEY_PREFIX,
@@ -73,18 +72,6 @@ def mock_clob(sample_market: Market) -> MagicMock:
return clob
@pytest.fixture
def mock_gamma() -> MagicMock:
"""Create a mock GammaClient that returns empty volume stats.
Without this, MarketMetadataSync would instantiate a default
GammaClient and hit the real gamma-api over HTTP during unit tests.
"""
gamma = MagicMock(spec=GammaClient)
gamma.get_active_market_stats = AsyncMock(return_value={})
return gamma
class TestDeriveCategory:
"""Tests for the derive_category function."""
@@ -208,9 +195,9 @@ class TestSyncStats:
class TestMarketMetadataSync:
"""Tests for the MarketMetadataSync class."""
def test_init(self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock) -> None:
def test_init(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
"""Test initialization."""
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
assert sync.state == SyncState.STOPPED
assert sync.stats.total_syncs == 0
@@ -218,14 +205,11 @@ class TestMarketMetadataSync:
assert sync._cache_ttl == DEFAULT_CACHE_TTL_SECONDS
assert sync._key_prefix == DEFAULT_REDIS_KEY_PREFIX
def test_init_custom_config(
self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
) -> None:
def test_init_custom_config(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
"""Test initialization with custom config."""
sync = MarketMetadataSync(
redis=mock_redis,
clob_client=mock_clob,
gamma_client=mock_gamma,
sync_interval_seconds=60,
cache_ttl_seconds=120,
key_prefix="custom:",
@@ -236,11 +220,9 @@ class TestMarketMetadataSync:
assert sync._key_prefix == "custom:"
@pytest.mark.asyncio
async def test_start_stop(
self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
) -> None:
async def test_start_stop(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
"""Test starting and stopping the sync service."""
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
# Start
await sync.start()
@@ -254,10 +236,10 @@ class TestMarketMetadataSync:
@pytest.mark.asyncio
async def test_start_performs_initial_sync(
self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
self, mock_redis: AsyncMock, mock_clob: MagicMock
) -> None:
"""Test that start performs an initial sync."""
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
await sync.start()
@@ -270,12 +252,10 @@ class TestMarketMetadataSync:
await sync.stop()
@pytest.mark.asyncio
async def test_start_failure(
self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
) -> None:
async def test_start_failure(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
"""Test start failure handling."""
mock_clob.get_markets.side_effect = Exception("API error")
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
with pytest.raises(MetadataSyncError, match="initial sync failed"):
await sync.start()
@@ -288,7 +268,6 @@ class TestMarketMetadataSync:
self,
mock_redis: AsyncMock,
mock_clob: MagicMock,
mock_gamma: MagicMock,
sample_metadata: MarketMetadata,
) -> None:
"""Test get_market with cache hit."""
@@ -296,7 +275,7 @@ class TestMarketMetadataSync:
cached_data = json.dumps(sample_metadata.to_dict())
mock_redis.get = AsyncMock(return_value=cached_data)
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
await sync.start()
result = await sync.get_market("cond123")
@@ -309,14 +288,12 @@ class TestMarketMetadataSync:
await sync.stop()
@pytest.mark.asyncio
async def test_get_market_cache_miss(
self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
) -> None:
async def test_get_market_cache_miss(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
"""Test get_market with cache miss."""
# Setup cache miss
mock_redis.get = AsyncMock(return_value=None)
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
await sync.start()
result = await sync.get_market("cond123")
@@ -331,14 +308,12 @@ class TestMarketMetadataSync:
await sync.stop()
@pytest.mark.asyncio
async def test_get_market_not_found(
self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
) -> None:
async def test_get_market_not_found(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
"""Test get_market when market doesn't exist."""
mock_redis.get = AsyncMock(return_value=None)
mock_clob.get_market.return_value = None
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
await sync.start()
result = await sync.get_market("nonexistent")
@@ -348,11 +323,9 @@ class TestMarketMetadataSync:
await sync.stop()
@pytest.mark.asyncio
async def test_invalidate_market(
self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
) -> None:
async def test_invalidate_market(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
"""Test cache invalidation."""
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
await sync.start()
result = await sync.invalidate_market("cond123")
@@ -363,11 +336,9 @@ class TestMarketMetadataSync:
await sync.stop()
@pytest.mark.asyncio
async def test_force_sync(
self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
) -> None:
async def test_force_sync(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
"""Test forced sync."""
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
await sync.start()
# Initial sync
@@ -382,9 +353,7 @@ class TestMarketMetadataSync:
await sync.stop()
@pytest.mark.asyncio
async def test_state_change_callback(
self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
) -> None:
async def test_state_change_callback(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
"""Test state change callback."""
states: list[SyncState] = []
@@ -394,7 +363,6 @@ class TestMarketMetadataSync:
sync = MarketMetadataSync(
redis=mock_redis,
clob_client=mock_clob,
gamma_client=mock_gamma,
on_state_change=on_state_change,
)
@@ -409,7 +377,7 @@ class TestMarketMetadataSync:
@pytest.mark.asyncio
async def test_sync_complete_callback(
self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
self, mock_redis: AsyncMock, mock_clob: MagicMock
) -> None:
"""Test sync complete callback."""
sync_stats: list[SyncStats] = []
@@ -420,7 +388,6 @@ class TestMarketMetadataSync:
sync = MarketMetadataSync(
redis=mock_redis,
clob_client=mock_clob,
gamma_client=mock_gamma,
on_sync_complete=on_sync_complete,
)
@@ -434,11 +401,7 @@ class TestMarketMetadataSync:
@pytest.mark.asyncio
async def test_get_markets_by_category(
self,
mock_redis: AsyncMock,
mock_clob: MagicMock,
mock_gamma: MagicMock,
sample_metadata: MarketMetadata,
self, mock_redis: AsyncMock, mock_clob: MagicMock, sample_metadata: MarketMetadata
) -> None:
"""Test getting markets by category."""
# Setup scan to return keys
@@ -449,7 +412,7 @@ class TestMarketMetadataSync:
cached_data = json.dumps(sample_metadata.to_dict())
mock_redis.get = AsyncMock(return_value=cached_data)
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
# Don't start to avoid initial sync complexity
sync._state = SyncState.IDLE
@@ -459,11 +422,9 @@ class TestMarketMetadataSync:
assert results[0].category == "crypto"
@pytest.mark.asyncio
async def test_cannot_start_twice(
self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
) -> None:
async def test_cannot_start_twice(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
"""Test that starting twice doesn't double-start."""
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
await sync.start()
await sync.start() # Should be a no-op
@@ -473,11 +434,9 @@ class TestMarketMetadataSync:
await sync.stop()
@pytest.mark.asyncio
async def test_stop_when_stopped(
self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock
) -> None:
async def test_stop_when_stopped(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
"""Test stopping when already stopped."""
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma)
sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
await sync.stop() # Should be a no-op
+14 -49
View File
@@ -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 -244
View File
@@ -327,10 +327,6 @@ class TestGetTransferLogs:
await funding_tracer._get_transfer_logs(
to_address=TEST_WALLET,
token_address=USDC_BRIDGED,
# Explicit numeric range so we stay inside one chunk and skip
# the "latest" → block_number resolution path.
from_block=1,
to_block=8_000,
)
mock_w3.eth.get_logs.assert_called_once()
@@ -338,15 +334,10 @@ class TestGetTransferLogs:
# Verify topics structure
assert len(call_args["topics"]) == 3
# The Transfer event topic must be 0x-prefixed; drpc rejects bare hex.
assert call_args["topics"][0] == "0x" + TRANSFER_EVENT_SIGNATURE.hex().removeprefix("0x")
assert call_args["topics"][0].startswith("0x")
assert call_args["topics"][0] == TRANSFER_EVENT_SIGNATURE.hex()
assert call_args["topics"][1] is None # from (any)
# to address should be padded to 32 bytes
assert call_args["topics"][2].endswith(TEST_WALLET.lower().replace("0x", ""))
# And the chunk bounds match what we asked for.
assert call_args["fromBlock"] == 1
assert call_args["toBlock"] == 8_000
@pytest.mark.asyncio
async def test_get_transfer_logs_respects_limit(
@@ -364,8 +355,6 @@ class TestGetTransferLogs:
to_address=TEST_WALLET,
token_address=USDC_BRIDGED,
limit=3,
from_block=1,
to_block=8_000,
)
assert len(result) == 3
@@ -385,242 +374,10 @@ class TestGetTransferLogs:
await funding_tracer._get_transfer_logs(
to_address=TEST_WALLET,
token_address=USDC_BRIDGED,
from_block=1,
to_block=8_000,
)
mock_fallback.eth.get_logs.assert_called_once()
@pytest.mark.asyncio
async def test_get_transfer_logs_chunks_large_ranges(
self,
funding_tracer: FundingTracer,
mock_polygon_client: MagicMock,
) -> None:
"""Ranges wider than chunk_size are split into multiple eth_getLogs calls.
This is the regression guard for the publicnode 10_000-block cap that
was rejecting every funding trace before chunking landed.
"""
mock_w3 = MagicMock()
mock_w3.eth.get_logs = AsyncMock(return_value=[])
mock_polygon_client._w3 = mock_w3
# 25_000 blocks at 9_000-per-chunk → 3 calls (9000 + 9000 + 7001).
await funding_tracer._get_transfer_logs(
to_address=TEST_WALLET,
token_address=USDC_BRIDGED,
from_block=1_000_000,
to_block=1_025_000,
)
assert mock_w3.eth.get_logs.call_count == 3
windows = [call[0][0] for call in mock_w3.eth.get_logs.call_args_list]
assert windows[0]["fromBlock"] == 1_000_000
assert windows[0]["toBlock"] == 1_008_999
assert windows[1]["fromBlock"] == 1_009_000
assert windows[1]["toBlock"] == 1_017_999
assert windows[2]["fromBlock"] == 1_018_000
assert windows[2]["toBlock"] == 1_025_000
# No window exceeds the chunk size — that's what RPC providers reject.
for win in windows:
assert win["toBlock"] - win["fromBlock"] + 1 <= 9_000
@pytest.mark.asyncio
async def test_get_transfer_logs_stops_when_limit_hit_mid_walk(
self,
funding_tracer: FundingTracer,
mock_polygon_client: MagicMock,
) -> None:
"""Walking should stop as soon as ``limit`` matches are gathered."""
mock_w3 = MagicMock()
# First chunk yields 5 logs, more than the limit, so subsequent chunks
# must not be queried.
mock_w3.eth.get_logs = AsyncMock(return_value=[MagicMock() for _ in range(5)])
mock_polygon_client._w3 = mock_w3
result = await funding_tracer._get_transfer_logs(
to_address=TEST_WALLET,
token_address=USDC_BRIDGED,
limit=2,
from_block=1_000_000,
to_block=1_025_000,
)
assert len(result) == 2
mock_w3.eth.get_logs.assert_called_once()
@pytest.mark.asyncio
async def test_get_transfer_logs_skips_failing_chunk(
self,
funding_tracer: FundingTracer,
mock_polygon_client: MagicMock,
) -> None:
"""A flaky chunk must not abort the whole trace — we move on."""
mock_w3 = MagicMock()
good_log = MagicMock()
responses: list[Any] = [
RuntimeError("RPC hiccup"),
[good_log],
]
async def fake_get_logs(_params: dict[str, Any]) -> list[Any]:
outcome = responses.pop(0)
if isinstance(outcome, BaseException):
raise outcome
return outcome
mock_w3.eth.get_logs = AsyncMock(side_effect=fake_get_logs)
mock_polygon_client._w3 = mock_w3
result = await funding_tracer._get_transfer_logs(
to_address=TEST_WALLET,
token_address=USDC_BRIDGED,
from_block=1_000_000,
to_block=1_018_000, # forces 3 chunks; we exercise chunks 1+2
)
# The error chunk is skipped; the second chunk contributes one log.
assert result == [dict(good_log)]
assert mock_w3.eth.get_logs.call_count >= 2
@pytest.mark.asyncio
async def test_get_transfer_logs_resolves_latest_via_block_number(
self,
funding_tracer: FundingTracer,
mock_polygon_client: MagicMock,
) -> None:
"""``to_block='latest'`` should resolve via ``eth.block_number``.
And ``from_block=0`` should not become a full-history scan it must
be clamped to ``latest - max_lookback_blocks``.
"""
async def _block_number_coro() -> int:
return 5_000
mock_eth = MagicMock()
mock_eth.get_logs = AsyncMock(return_value=[])
# Property-style awaitable: web3.py exposes block_number as a property
# returning a coroutine, so each access must yield a fresh awaitable.
type(mock_eth).block_number = property( # type: ignore[misc]
lambda _self: _block_number_coro()
)
mock_w3 = MagicMock()
mock_w3.eth = mock_eth
mock_polygon_client._w3 = mock_w3
await funding_tracer._get_transfer_logs(
to_address=TEST_WALLET,
token_address=USDC_BRIDGED,
)
# block_number=5000 < chunk_size, so it's one chunk that bottoms at 0.
mock_eth.get_logs.assert_called_once()
call_args = mock_eth.get_logs.call_args[0][0]
assert call_args["fromBlock"] == 0
assert call_args["toBlock"] == 5_000
@pytest.mark.asyncio
async def test_get_transfer_logs_breaks_on_pruned_history(
self,
funding_tracer: FundingTracer,
mock_polygon_client: MagicMock,
) -> None:
"""A pruned-history error must short-circuit the whole walk.
Public Polygon RPCs prune log history. Once we walk past the cutoff,
every subsequent chunk will raise the same error keep walking and
we just burn quota on guaranteed failures. The first such error must
end the walk and return whatever we already collected.
"""
mock_w3 = MagicMock()
good_log = MagicMock()
responses: list[Any] = [
[good_log],
RuntimeError(
"{'code': -32701, 'message': 'History has been pruned for "
"this block. To remove restrictions, order a dedicated full "
"node here: https://www.allnodes.com/pol/host'}"
),
# If the early-break logic is missing, this third chunk would
# also be requested. The test asserts it isn't.
[MagicMock()],
]
async def fake_get_logs(_params: dict[str, Any]) -> list[Any]:
outcome = responses.pop(0)
if isinstance(outcome, BaseException):
raise outcome
return outcome
mock_w3.eth.get_logs = AsyncMock(side_effect=fake_get_logs)
mock_polygon_client._w3 = mock_w3
# 3 chunks total. The pruned error fires on chunk #2; chunk #3 must
# never be issued.
result = await funding_tracer._get_transfer_logs(
to_address=TEST_WALLET,
token_address=USDC_BRIDGED,
from_block=1_000_000,
to_block=1_027_000,
)
assert result == [dict(good_log)]
assert mock_w3.eth.get_logs.call_count == 2
@pytest.mark.asyncio
async def test_get_transfer_logs_default_lookback_fits_pruned_horizon(
self,
) -> None:
"""Default ``max_lookback_blocks`` must stay inside what public RPCs serve.
publicnode prunes after ~100k blocks. If we default to 1.3M, every
funding trace blows through the archive horizon and produces nothing
but pruned-history warnings. Pin the default at <= 100k as a
regression guard.
"""
from polymarket_insider_tracker.profiler.funding import (
DEFAULT_MAX_LOOKBACK_BLOCKS,
)
assert DEFAULT_MAX_LOOKBACK_BLOCKS <= 100_000
@pytest.mark.asyncio
async def test_get_transfer_logs_topic_is_0x_prefixed(
self,
funding_tracer: FundingTracer,
mock_polygon_client: MagicMock,
) -> None:
"""The Transfer event topic passed to ``eth_getLogs`` must begin with ``0x``.
``HexBytes.hex()`` returns a bare hex string. publicnode tolerates
that, but stricter providers like drpc (our fallback) reject it with
``invalid argument 0: hex string without 0x prefix`` and every chunk
in the trace fails. This guards against regressing back to the
bare-hex form.
"""
mock_w3 = MagicMock()
mock_w3.eth.get_logs = AsyncMock(return_value=[])
mock_polygon_client._w3 = mock_w3
await funding_tracer._get_transfer_logs(
to_address=TEST_WALLET,
token_address=USDC_BRIDGED,
from_block=1,
to_block=8_000,
)
topics = mock_w3.eth.get_logs.call_args[0][0]["topics"]
assert topics[0].startswith("0x")
# And the topic also has to be 32 bytes (64 hex chars) as required by
# the JSON-RPC spec.
assert len(topics[0]) == 2 + 64
# The padded `to` topic was already 0x-prefixed; double-check that
# didn't regress either.
assert topics[2].startswith("0x")
class TestLogToFundingTransfer:
"""Tests for _log_to_funding_transfer method."""
-175
View File
@@ -1,175 +0,0 @@
"""Tests for RiskAssessment persistence inside Pipeline._score_and_alert.
Verifies:
1. Every signal-bearing assessment is written to risk_assessments, even
when ``should_alert`` is False (i.e. below the alert threshold).
2. A DB failure during persistence never blocks alert dispatching.
"""
from __future__ import annotations
from datetime import UTC, datetime
from decimal import Decimal
from unittest.mock import AsyncMock, MagicMock
import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from polymarket_insider_tracker.config import Settings
from polymarket_insider_tracker.detector.models import RiskAssessment
from polymarket_insider_tracker.detector.scorer import SignalBundle
from polymarket_insider_tracker.ingestor.models import TradeEvent
from polymarket_insider_tracker.pipeline import Pipeline
from polymarket_insider_tracker.storage.database import DatabaseManager
from polymarket_insider_tracker.storage.models import Base, RiskAssessmentModel
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def mock_settings():
"""Settings stub with the attributes Pipeline reaches for at runtime."""
detector = MagicMock()
detector.persist_assessments = True
detector.alert_threshold = 0.8
settings = MagicMock(spec=Settings)
settings.detector = detector
settings.dry_run = False
return settings
@pytest.fixture
async def async_engine():
engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield engine
await engine.dispose()
@pytest.fixture
async def db_manager(async_engine):
manager = DatabaseManager.__new__(DatabaseManager)
manager.database_url = "sqlite+aiosqlite:///:memory:"
manager.async_mode = True
manager._pool_size = 5
manager._max_overflow = 10
manager._echo = False
manager._sync_engine = None
manager._async_engine = async_engine
manager._sync_session_factory = None
manager._async_session_factory = async_sessionmaker(bind=async_engine, expire_on_commit=False)
return manager
@pytest.fixture
def sample_trade() -> TradeEvent:
return TradeEvent(
trade_id="0x" + "a" * 64,
wallet_address="0x" + "b" * 40,
market_id="0x" + "c" * 64,
asset_id="asset_xyz",
side="BUY",
price=Decimal("0.42"),
size=Decimal("1000"),
timestamp=datetime.now(UTC),
outcome="Yes",
outcome_index=0,
event_title="Test Event",
market_slug="test-market",
)
def _make_assessment(trade: TradeEvent, *, should_alert: bool, score: float) -> RiskAssessment:
return RiskAssessment(
trade_event=trade,
wallet_address=trade.wallet_address,
market_id=trade.market_id,
fresh_wallet_signal=None,
size_anomaly_signal=None,
signals_triggered=1,
weighted_score=score,
should_alert=should_alert,
)
def _build_pipeline(
mock_settings,
*,
db_manager=None,
assessment: RiskAssessment,
dispatcher: MagicMock | None = None,
) -> Pipeline:
"""Construct a Pipeline with the minimum collaborators wired in."""
pipeline = Pipeline(mock_settings)
pipeline._db_manager = db_manager
pipeline._risk_scorer = MagicMock()
pipeline._risk_scorer.assess = AsyncMock(return_value=assessment)
pipeline._alert_formatter = MagicMock()
pipeline._alert_formatter.format = MagicMock(return_value=MagicMock())
if dispatcher is None:
dispatcher = MagicMock()
dispatcher.dispatch = AsyncMock(
return_value=MagicMock(all_succeeded=True, success_count=1, failure_count=0)
)
pipeline._alert_dispatcher = dispatcher
pipeline._dry_run = False
return pipeline
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
class TestPersistAssessment:
@pytest.mark.asyncio
async def test_below_threshold_assessment_is_persisted(
self, mock_settings, db_manager, sample_trade, async_engine
):
"""Assessments with should_alert=False must still hit the DB; no dispatch."""
assessment = _make_assessment(sample_trade, should_alert=False, score=0.45)
pipeline = _build_pipeline(mock_settings, db_manager=db_manager, assessment=assessment)
await pipeline._score_and_alert(SignalBundle(trade_event=sample_trade))
# Row landed in risk_assessments
async with async_sessionmaker(bind=async_engine, expire_on_commit=False)() as session:
rows = (await session.execute(select(RiskAssessmentModel))).scalars().all()
assert len(rows) == 1
row = rows[0]
assert row.assessment_id == assessment.assessment_id
assert row.should_alert is False
assert float(row.weighted_score) == pytest.approx(0.45, abs=1e-3)
assert row.wallet_address == sample_trade.wallet_address.lower()
# No alert dispatched for sub-threshold assessments
pipeline._alert_dispatcher.dispatch.assert_not_called()
assert pipeline.stats.alerts_sent == 0
@pytest.mark.asyncio
async def test_persistence_failure_does_not_block_dispatch(self, mock_settings, sample_trade):
"""If repo.insert blows up, the alert pipeline still ships the alert."""
assessment = _make_assessment(sample_trade, should_alert=True, score=0.92)
# db_manager whose get_async_session raises -> _persist_assessment swallows it
broken_db = MagicMock()
broken_db.get_async_session = MagicMock(side_effect=RuntimeError("DB connection failed"))
pipeline = _build_pipeline(mock_settings, db_manager=broken_db, assessment=assessment)
await pipeline._score_and_alert(SignalBundle(trade_event=sample_trade))
# DB write was attempted and failed silently
broken_db.get_async_session.assert_called_once()
# Dispatcher still ran and the stats counter incremented
pipeline._alert_dispatcher.dispatch.assert_awaited_once()
assert pipeline.stats.alerts_sent == 1
-4
View File
@@ -44,9 +44,6 @@ def mock_settings():
telegram.bot_token = None
telegram.chat_id = None
detector = MagicMock()
detector.persist_assessments = False
settings = MagicMock(spec=Settings)
settings.redis = redis
settings.database = database
@@ -54,7 +51,6 @@ def mock_settings():
settings.polymarket = polymarket
settings.discord = discord
settings.telegram = telegram
settings.detector = detector
settings.dry_run = True
return settings
-334
View File
@@ -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