paulpas Role: Build detection systems that identify unusual price movements, volume spikes, and structural breaks
Philosophy: Anomalies signal information - either noise, news events, or market inefficiencies. Prioritize low false positive rates and explainable detections over pure sensitivity.
Key Principles
- Multi-Modal Detection: Combine statistical, temporal, and contextual anomaly detection
- Adaptive Thresholds: Adjust sensitivity based on market volatility regime
- Explainability: Provide clear rationale for each anomaly detection
- False Positive Control: Use multiple validation layers to reduce noise
- Temporal Awareness: Consider cyclical patterns (intraday, weekly, monthly)
Implementation Guidelines
Structure
- Core logic:
anomaly/detectors.py- Individual detector implementations - Aggregator:
anomaly/ensemble.py- Combine multiple detectors - Alerts:
anomaly/alerts.py- Anomaly notification system - Config:
config/anomaly_config.yaml- Detection parameters
Patterns to Follow
- Use unsupervised methods for novel anomaly detection
- Include both point anomalies and collective anomalies
- Add volume and volatility context to price anomalies
- Implement real-time detection pipelines
Adherence Checklist
Before completing your task, verify:
- Multiple detection methods combined (statistical, ML, rule-based)
- Thresholds adapt to current market volatility
- Each detection includes explanatory features
- False positive rate controlled through validation
- Both price and volume anomalies detected
Code Examples
Statistical Anomaly Detection
import numpy as np
import pandas as pd
from scipy import stats
from typing import Tuple, List, Dict
from dataclasses import dataclass
@dataclass
class Anomaly Detection:
"""Represents a detected anomaly with metadata."""
timestamp: int
score: float
type: str # 'price', 'volume', 'spread'
magnitude: float
explanation: str
z_score: float = None
class StatisticalAnomalyDetector:
"""Statistical anomaly detection using Z-scores and rolling statistics."""
def __init__(self, window_size: int = 50, z_threshold: float = 3.0,
min_std: float = 0.001):
self.window_size = window_size
self.z_threshold = z_threshold
self.min_std = min_std
self.rolling_stats = {}
def fit(self, prices: np.ndarray, volumes: np.ndarray = None):
"""Calculate rolling statistics for anomaly detection."""
prices = np.asarray(prices)
# Calculate rolling mean and std for prices
self.rolling_stats['price'] = {
'mean': pd.Series(prices).rolling(self.window_size).mean().values,
'std': pd.Series(prices).rolling(self.window_size).std().fillna(self.min_std).values
}
if volumes is not None:
self.rolling_stats['volume'] = {
'mean': pd.Series(volumes).rolling(self.window_size).mean().values,
'std': pd.Series(volumes).rolling(self.window_size).std().fillna(1.0).values
}
def detect_price_anomalies(self, prices: np.ndarray) -> List[Anomaly]:
"""Detect price anomalies using Z-score method."""
prices = np.asarray(prices)
anomalies = []
stats = self.rolling_stats['price']
returns = np.diff(np.log(prices))
for i in range(self.window_size, len(prices)):
z_score = (returns[i-1] - stats['mean'][i]) / max(stats['std'][i], self.min_std)
if abs(z_score) > self.z_threshold:
direction = 'upward' if z_score > 0 else 'downward'
magnitude = abs(z_score) / self.z_threshold
anomalies.append(Anomaly(
timestamp=i,
score=abs(z_score),
type='price',
magnitude=magnitude,
explanation=f'{direction} price spike: {z_score:.2f} std devs from mean',
z_score=z_score
))
return anomalies
def detect_volume_anomalies(self, volumes: np.ndarray) -> List[Anomaly]:
"""Detect volume anomalies using rolling statistics."""
volumes = np.asarray(volumes)
anomalies = []
if 'volume' not in self.rolling_stats:
return anomalies
stats = self.rolling_stats['volume']
for i in range(self.window_size, len(volumes)):
z_score = (volumes[i] - stats['mean'][i]) / max(stats['std'][i], 1.0)
if z_score > self.z_threshold:
anomalies.append(Anomaly(
timestamp=i,
score=z_score,
type='volume',
magnitude=z_score / self.z_threshold,
explanation=f'Unusual volume: {volumes[i]:.0f} (z-score: {z_score:.2f})',
z_score=z_score
))
return anomalies
def fit_detect(self, prices: np.ndarray, volumes: np.ndarray = None) -> List[Anomaly]:
"""Fit detector and detect anomalies in one step."""
self.fit(prices, volumes)
price_anomalies = self.detect_price_anomalies(prices)
volume_anomalies = self.detect_volume_anomalies(volumes) if volumes is not None else []
return price_anomalies + volume_anomalies
Isolation Forest for Multivariate Anomalies
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import numpy as np
import pandas as pd
from typing import Tuple, List
class IsolationForestAnomalyDetector:
"""Isolation Forest for detecting multivariate anomalies in market data."""
def __init__(self, contamination: float = 0.05, n_estimators: int = 100,
max_samples: str = 'auto', random_state: int = 42):
self.contamination = contamination
self.n_estimators = n_estimators
self.max_samples = max_samples
self.random_state = random_state
self.model = IsolationForest(
contamination=contamination,
n_estimators=n_estimators,
max_samples=max_samples,
random_state=random_state
)
self.scaler = StandardScaler()
self.feature_names = []
def extract_features(self, prices: np.ndarray, volumes: np.ndarray = None,
returns_window: List[int] = [1, 5, 10, 20]) -> np.ndarray:
"""Extract features for anomaly detection."""
features = []
# Price returns
prices = np.asarray(prices)
log_prices = np.log(prices)
for window in returns_window:
if window < len(prices):
returns = np.diff(log_prices, n=window) / window
features.append(returns)
features = np.array(features).T
# Add volume if available
if volumes is not None:
volumes = np.asarray(volumes)
log_volumes = np.log(volumes + 1)
features = np.column_stack([features, log_volumes])
# Fill NaN values
features = np.nan_to_num(features, nan=0.0)
return features
def fit(self, prices: np.ndarray, volumes: np.ndarray = None):
"""Fit Isolation Forest on market data."""
features = self.extract_features(prices, volumes)
self.feature_names = [f'return_{w}d' for w in [1, 5, 10, 20]]
if volumes is not None:
self.feature_names.append('log_volume')
scaled_features = self.scaler.fit_transform(features)
self.model.fit(scaled_features)
def detect(self, prices: np.ndarray, volumes: np.ndarray = None) -> Tuple[np.ndarray, np.ndarray]:
"""Detect anomalies and return anomaly scores."""
features = self.extract_features(prices, volumes)
scaled_features = self.scaler.transform(features)
predictions = self.model.predict(scaled_features) # -1 for anomaly, 1 for normal
scores = -self.model.score_samples(scaled_features) # Higher = more anomalous
return predictions, scores
def fit_detect(self, prices: np.ndarray, volumes: np.ndarray = None) -> List[dict]:
"""Fit detector and return anomaly list."""
features = self.extract_features(prices, volumes)
# Store feature values for explanation
feature_names = [f'return_{w}d' for w in [1, 5, 10, 20]]
if volumes is not None:
feature_names.append('log_volume')
self.fit(prices, volumes)
predictions, scores = self.detect(prices, volumes)
anomalies = []
for i, (pred, score) in enumerate(zip(predictions, scores)):
if pred == -1: # Anomaly detected
# Get feature values for explanation
feature_values = features[i]
feature_contributions = {
name: float(val) for name, val in zip(feature_names, feature_values)
}
anomalies.append({
'timestamp': i,
'score': float(score),
'z_score': float(self._normalize_score(score)),
'type': 'multivariate',
'magnitude': float(score / self._calculate_threshold()),
'explanation': f"Multivariate anomaly detected using isolation forest",
'features': feature_contributions
})
return anomalies
def _normalize_score(self, score: float, mean: float = 0.0, std: float = 1.0) -> float:
"""Normalize anomaly score to Z-score."""
return (score - mean) / std if std > 0 else 0.0
def _calculate_threshold(self) -> float:
"""Calculate threshold based on contamination rate."""
return np.percentile([1.5, 2.0, 3.0], 95) # Empirical threshold
Ensemble Anomaly Detection
import numpy as np
from typing import List, Dict, Tuple
class EnsembleAnomalyDetector:
"""Ensemble of multiple anomaly detection methods."""
def __init__(self, detectors: Dict[str, object], weights: Dict[str, float] = None):
self.detectors = detectors
self.weights = weights or {name: 1.0 / len(detectors) for name in detectors}
# Normalize weights
total_weight = sum(self.weights.values())
self.weights = {k: v / total_weight for k, v in self.weights.items()}
def fit(self, prices: np.ndarray, volumes: np.ndarray = None):
"""Fit all detectors."""
for name, detector in self.detectors.items():
if hasattr(detector, 'fit'):
detector.fit(prices, volumes)
def fit_detect(self, prices: np.ndarray, volumes: np.ndarray = None) -> List[dict]:
"""Fit detectors and combine anomaly scores."""
# Fit all detectors
self.fit(prices, volumes)
# Get anomalies from each detector
all_scores = np.zeros(len(prices))
anomaly_counts = np.zeros(len(prices))
for name, detector in self.detectors.items():
weights = self.weights[name]
if hasattr(detector, 'fit_detect'):
anomalies = detector.fit_detect(prices, volumes)
elif hasattr(detector, 'detect'):
if name == 'statistical':
anomalies = detector.detect_price_anomalies(prices)
else:
anomalies = detector.detect(prices, volumes)
else:
continue
for anomaly in anomalies:
idx = anomaly.timestamp
all_scores[idx] += weights * anomaly.score
anomaly_counts[idx] += 1
# Normalize combined scores
combined_scores = all_scores / (anomaly_counts + 1e-6)
# Create anomaly list
anomalies = []
threshold = np.percentile(combined_scores, 95)
for i, score in enumerate(combined_scores):
if score > threshold:
# Find individual detector scores for this timestamp
detector_scores = {}
for name, detector in self.detectors.items():
if hasattr(detector, 'fit_detect'):
det_anomalies = detector.fit_detect(prices, volumes)
elif hasattr(detector, 'detect_price_anomalies'):
det_anomalies = detector.detect_price_anomalies(prices)
else:
continue
for det_anomaly in det_anomalies:
if det_anomaly.timestamp == i:
detector_scores[name] = det_anomaly.score
break
anomalies.append({
'timestamp': i,
'score': float(score),
'type': 'ensemble',
'magnitude': float(score / threshold),
'explanation': f'Ensemble anomaly: {len(detector_scores)} detectors flagged',
'detector_scores': detector_scores
})
return sorted(anomalies, key=lambda x: x['score'], reverse=True)
Constraints
MUST DO
- Validate input feature distributions against training data baselines; flag drift exceeding 2 standard deviations
- Implement model versioning with reproducibility tags — every prediction must be traceable to the exact model artifact and config
- Include confidence intervals or probability estimates alongside all point predictions, never return raw scores without context
- Log all model inputs, outputs, and metadata to enable post-hoc analysis of prediction failures
- Implement feature computation consistently between training and inference — use the same transformation pipeline for both
MUST NOT DO
- Do not train models on look-ahead biased features (e.g., using future prices or events in training data)
- Avoid deploying a new model version without shadow-testing against the current production model first
- Never retrain a model on a data window that includes regime changes without explicit regime-aware validation
- Do not use accuracy as the primary metric for imbalanced datasets — use precision/recall, F1, or AUC-ROC
- Avoid hardcoding feature names; load them from a schema or config file to prevent mismatches between training and inference
Live References
Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.