gavindiaz/polymarket-insider-tracker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(detector): add position size anomaly detection (#15)

Browse Source 
Implement SizeAnomalyDetector for identifying trades with unusually
large position sizes relative to market liquidity.

Features:
- Volume impact analysis (trade size / 24h volume)
- Order book impact analysis (trade size / book depth)
- Niche market detection using category heuristics
- Confidence scoring with configurable thresholds
- Batch analysis for processing multiple trades

The detector gracefully handles missing volume/book data by falling
back to category-based heuristics for identifying niche markets
where large trades are more significant.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Patrick Selamy
2026-01-04 15:58:27 -05:00
co-authored by Claude Opus 4.5
parent 89bf80e3d8
commit a0bec521b6
4 changed files with 1264 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/polymarket_insider_tracker/detector/__init__.py
+8 -2
View File
@@ -1,6 +1,12 @@
"""Anomaly detection layer - Suspicious activity identification."""
from polymarket_insider_tracker.detector.fresh_wallet import FreshWalletDetector
from polymarket_insider_tracker.detector.models import FreshWalletSignal
from polymarket_insider_tracker.detector.models import FreshWalletSignal, SizeAnomalySignal
from polymarket_insider_tracker.detector.size_anomaly import SizeAnomalyDetector
__all__ = ["FreshWalletDetector", "FreshWalletSignal"]
__all__ = [
"FreshWalletDetector",
"FreshWalletSignal",
"SizeAnomalyDetector",
"SizeAnomalySignal",
]
src/polymarket_insider_tracker/detector/models.py
+73 -1
View File
@@ -4,7 +4,7 @@ from dataclasses import dataclass, field
from datetime import UTC, datetime
from decimal import Decimal
from polymarket_insider_tracker.ingestor.models import TradeEvent
from polymarket_insider_tracker.ingestor.models import MarketMetadata, TradeEvent
from polymarket_insider_tracker.profiler.models import WalletProfile
@@ -71,3 +71,75 @@ class FreshWalletSignal:
"factors": self.factors,
"timestamp": self.timestamp.isoformat(),
}
@dataclass(frozen=True)
class SizeAnomalySignal:
"""Signal emitted when a trade has unusually large position size.
This signal is generated when a trade's size significantly impacts
the market volume or order book depth, indicating potential informed
trading activity.
Attributes:
trade_event: The original trade event that triggered this signal.
market_metadata: Metadata about the market being traded.
volume_impact: Trade size as fraction of 24h volume (0.0 if unknown).
book_impact: Trade size as fraction of order book depth (0.0 if unknown).
is_niche_market: Whether the market is considered niche/low-volume.
confidence: Overall confidence score (0.0 to 1.0).
factors: Individual factor scores contributing to confidence.
timestamp: When this signal was generated.
"""
trade_event: TradeEvent
market_metadata: MarketMetadata
volume_impact: float
book_impact: float
is_niche_market: bool
confidence: float
factors: dict[str, float]
timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
@property
def wallet_address(self) -> str:
"""Return the wallet address from the trade event."""
return self.trade_event.wallet_address
@property
def market_id(self) -> str:
"""Return the market ID from the trade event."""
return self.trade_event.market_id
@property
def trade_size_usdc(self) -> Decimal:
"""Return the trade size in USDC (notional value)."""
return self.trade_event.notional_value
@property
def is_high_confidence(self) -> bool:
"""Return True if confidence exceeds 0.7."""
return self.confidence >= 0.7
@property
def is_very_high_confidence(self) -> bool:
"""Return True if confidence exceeds 0.85."""
return self.confidence >= 0.85
def to_dict(self) -> dict[str, object]:
"""Serialize to dictionary for Redis stream publishing."""
return {
"wallet_address": self.wallet_address,
"market_id": self.market_id,
"trade_id": self.trade_event.trade_id,
"trade_size": str(self.trade_size_usdc),
"trade_side": self.trade_event.side,
"trade_price": str(self.trade_event.price),
"market_category": self.market_metadata.category,
"volume_impact": self.volume_impact,
"book_impact": self.book_impact,
"is_niche_market": self.is_niche_market,
"confidence": self.confidence,
"factors": self.factors,
"timestamp": self.timestamp.isoformat(),
}
src/polymarket_insider_tracker/detector/size_anomaly.py
+354
View File
@@ -0,0 +1,354 @@
"""Position size anomaly detection algorithm.
This module provides the SizeAnomalyDetector class that identifies trades
with unusually large position sizes relative to market liquidity.
"""
import logging
from decimal import Decimal
from polymarket_insider_tracker.detector.models import SizeAnomalySignal
from polymarket_insider_tracker.ingestor.metadata_sync import MarketMetadataSync
from polymarket_insider_tracker.ingestor.models import MarketMetadata, TradeEvent
logger = logging.getLogger(__name__)
# Default configuration
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 market categories - markets in these categories with low specificity
# are more likely to have insider information value
NICHE_PRONE_CATEGORIES = frozenset({"science", "tech", "finance", "other"})
class SizeAnomalyDetector:
"""Detector for unusually large trade sizes.
This detector analyzes trade events for size anomalies by comparing
the trade size against market liquidity metrics:
- Volume impact: trade size / 24h volume
- Book impact: trade size / order book depth
When volume data is unavailable, the detector uses category-based
heuristics to identify niche markets where large trades are more
significant.
Confidence scoring:
- Volume impact > threshold: base score from impact ratio
- Book impact > threshold: additional score from impact ratio
- Niche market multiplier: 1.5x for low-volume markets
Example:
```python
sync = MarketMetadataSync(redis, clob_client)
detector = SizeAnomalyDetector(sync)
# Analyze a trade
signal = await detector.analyze(trade_event)
if signal is not None:
print(f"Size anomaly detected! Confidence: {signal.confidence}")
```
"""
def __init__(
self,
metadata_sync: MarketMetadataSync,
*,
volume_threshold: float = DEFAULT_VOLUME_THRESHOLD,
book_threshold: float = DEFAULT_BOOK_THRESHOLD,
niche_volume_threshold: Decimal = DEFAULT_NICHE_VOLUME_THRESHOLD,
) -> None:
"""Initialize the size anomaly detector.
Args:
metadata_sync: MarketMetadataSync for fetching market metadata.
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).
"""
self._metadata_sync = metadata_sync
self._volume_threshold = volume_threshold
self._book_threshold = book_threshold
self._niche_volume_threshold = niche_volume_threshold
async def analyze(
self,
trade: TradeEvent,
*,
daily_volume: Decimal | None = None,
book_depth: Decimal | None = None,
) -> SizeAnomalySignal | None:
"""Analyze a trade event for size anomalies.
This method:
1. Fetches market metadata
2. Calculates volume and book impact (if data available)
3. Determines if market is niche
4. Calculates confidence score
Args:
trade: TradeEvent to analyze.
daily_volume: Optional 24h volume in USDC. If provided, enables
volume impact calculation.
book_depth: Optional order book depth in USDC. If provided,
enables book impact calculation.
Returns:
SizeAnomalySignal if the trade triggers anomaly detection,
None otherwise.
"""
# Get market metadata
try:
metadata = await self._metadata_sync.get_market(trade.market_id)
if metadata is None:
logger.warning(
"No metadata found for market %s, creating minimal metadata",
trade.market_id,
)
metadata = self._create_minimal_metadata(trade)
except Exception as e:
logger.warning(
"Failed to get metadata for market %s: %s",
trade.market_id,
e,
)
metadata = self._create_minimal_metadata(trade)
trade_size = trade.notional_value
# Calculate impacts
volume_impact = self._calculate_volume_impact(trade_size, daily_volume)
book_impact = self._calculate_book_impact(trade_size, book_depth)
# Determine if niche market
is_niche = self._is_niche_market(metadata, daily_volume)
# Check if any threshold exceeded
exceeds_volume = volume_impact > self._volume_threshold
exceeds_book = book_impact > self._book_threshold
if not exceeds_volume and not exceeds_book and not is_niche:
logger.debug(
"Trade %s does not exceed thresholds: volume=%.4f, book=%.4f",
trade.trade_id,
volume_impact,
book_impact,
)
return None
# Calculate confidence score
confidence, factors = self.calculate_confidence(
volume_impact=volume_impact,
book_impact=book_impact,
is_niche=is_niche,
)
# Only emit signal if confidence is meaningful
if confidence < 0.1:
return None
logger.info(
"Size anomaly signal: market=%s, size=%s, volume_impact=%.4f, "
"book_impact=%.4f, niche=%s, confidence=%.2f",
trade.market_id[:10] + "...",
trade_size,
volume_impact,
book_impact,
is_niche,
confidence,
)
return SizeAnomalySignal(
trade_event=trade,
market_metadata=metadata,
volume_impact=volume_impact,
book_impact=book_impact,
is_niche_market=is_niche,
confidence=confidence,
factors=factors,
)
def _create_minimal_metadata(self, trade: TradeEvent) -> MarketMetadata:
"""Create minimal metadata from trade event."""
from polymarket_insider_tracker.ingestor.models import Token
return MarketMetadata(
condition_id=trade.market_id,
question=trade.event_title or "Unknown Market",
description="",
tokens=(
Token(
token_id=trade.asset_id,
outcome=trade.outcome,
price=trade.price,
),
),
category="other",
)
def _calculate_volume_impact(
self,
trade_size: Decimal,
daily_volume: Decimal | None,
) -> float:
"""Calculate trade size as fraction of daily volume.
Args:
trade_size: Trade notional value in USDC.
daily_volume: 24h trading volume in USDC.
Returns:
Volume impact ratio, or 0.0 if volume unknown.
"""
if daily_volume is None or daily_volume <= 0:
return 0.0
return float(trade_size / daily_volume)
def _calculate_book_impact(
self,
trade_size: Decimal,
book_depth: Decimal | None,
) -> float:
"""Calculate trade size as fraction of order book depth.
Args:
trade_size: Trade notional value in USDC.
book_depth: Visible order book depth in USDC.
Returns:
Book impact ratio, or 0.0 if depth unknown.
"""
if book_depth is None or book_depth <= 0:
return 0.0
return float(trade_size / book_depth)
def _is_niche_market(
self,
metadata: MarketMetadata,
daily_volume: Decimal | None,
) -> bool:
"""Determine if market is considered niche.
A market is niche if:
- Volume is below threshold ($50k), OR
- Category is prone to insider info AND volume is unknown
Args:
metadata: Market metadata with category.
daily_volume: Optional 24h volume.
Returns:
True if market is considered niche.
"""
# If volume known and below threshold, it's niche
if daily_volume is not None and daily_volume < self._niche_volume_threshold:
return True
# If volume unknown, use category heuristics
return daily_volume is None and metadata.category in NICHE_PRONE_CATEGORIES
def calculate_confidence(
self,
*,
volume_impact: float,
book_impact: float,
is_niche: bool,
) -> tuple[float, dict[str, float]]:
"""Calculate confidence score based on impact metrics.
Confidence scoring:
- Volume impact: min(impact/threshold, 3) / 3 * 0.5
- Book impact: min(impact/threshold, 3) / 3 * 0.3
- Niche multiplier: 1.5x final score
Final confidence clamped to [0.0, 1.0].
Args:
volume_impact: Trade size / daily volume ratio.
book_impact: Trade size / book depth ratio.
is_niche: Whether market is niche.
Returns:
Tuple of (confidence_score, factors_dict).
"""
factors: dict[str, float] = {}
confidence = 0.0
# Volume impact component
if volume_impact > self._volume_threshold:
ratio = min(volume_impact / self._volume_threshold, 3.0)
volume_score = ratio / 3.0 * 0.5
factors["volume_impact"] = volume_score
confidence += volume_score
# Book impact component
if book_impact > self._book_threshold:
ratio = min(book_impact / self._book_threshold, 3.0)
book_score = ratio / 3.0 * 0.3
factors["book_impact"] = book_score
confidence += book_score
# Niche market multiplier
if is_niche and confidence > 0:
factors["niche_multiplier"] = 1.5
confidence *= 1.5
# If niche but no other signals, give small base confidence
if is_niche and confidence == 0:
factors["niche_base"] = 0.2
confidence = 0.2
# Clamp to valid range
confidence = max(0.0, min(1.0, confidence))
return confidence, factors
async def analyze_batch(
self,
trades: list[TradeEvent],
*,
volume_data: dict[str, Decimal] | None = None,
book_data: dict[str, Decimal] | None = None,
) -> list[SizeAnomalySignal]:
"""Analyze multiple trades for size anomalies.
Processes trades in parallel for efficiency.
Args:
trades: List of trades to analyze.
volume_data: Optional dict mapping market_id to 24h volume.
book_data: Optional dict mapping market_id to book depth.
Returns:
List of SizeAnomalySignal for trades with anomalies.
"""
import asyncio
volume_data = volume_data or {}
book_data = book_data or {}
tasks = [
self.analyze(
trade,
daily_volume=volume_data.get(trade.market_id),
book_depth=book_data.get(trade.market_id),
)
for trade in trades
]
results = await asyncio.gather(*tasks, return_exceptions=True)
signals: list[SizeAnomalySignal] = []
for trade, result in zip(trades, results, strict=True):
if isinstance(result, BaseException):
logger.warning(
"Failed to analyze trade %s: %s",
trade.trade_id,
result,
)
continue
if result is not None:
signals.append(result)
return signals