ML Anomaly Detection

Truthound provides comprehensive ML-based anomaly detection, drift detection, and model monitoring capabilities.

---

Table of Contents

1. Overview
2. Anomaly Detection
3. Drift Detection
4. Model Monitoring
5. Rule Learning
6. Configuration Reference

---

Overview

The ML module (truthound.ml) provides:

• 6 Anomaly Detectors: Statistical and tree-based algorithms
• 4 Drift Detectors: Distribution, feature, concept, and multivariate drift
• Model Monitoring: Performance, drift, and quality metrics with alerting
• Rule Learning: Automatic validation rule generation from data profiles

Location: src/truthound/ml/

ml/
├── __init__.py              # Public API exports
├── base.py                  # Core abstractions (1,183 lines)
├── anomaly_models/          # Anomaly detection algorithms
├── drift_detection/         # Drift detection methods
├── monitoring/              # ML model monitoring
└── rule_learning/           # Automatic rule generation

---

Anomaly Detection

Supported Algorithms

Detector	Category	Algorithm	Use Case
ZScoreAnomalyDetector	Statistical	Z-score (std devs from mean)	Normally distributed data
IQRAnomalyDetector	Statistical	Interquartile range	Robust to mild outliers
MADAnomalyDetector	Statistical	Median Absolute Deviation	Extremely robust to outliers
IsolationForestDetector	Tree-based	Isolation Forest	Non-linear, high dimensions
EnsembleAnomalyDetector	Meta	Multiple strategy voting	Best coverage

Basic Usage

from truthound import ml
import polars as pl

# Load data
train_data = pl.scan_parquet("train.parquet")
test_data = pl.scan_parquet("test.parquet")

# Statistical anomaly detection
detector = ml.ZScoreAnomalyDetector(z_threshold=3.0)
detector.fit(train_data)
result = detector.predict(test_data)

print(f"Found {result.anomaly_count} anomalies ({result.anomaly_ratio:.1%})")

Statistical Detectors

Z-Score Detector

from truthound.ml.anomaly_models.statistical import (
    StatisticalAnomalyDetector,
    StatisticalConfig,
)

config = StatisticalConfig(
    z_threshold=3.0,           # Standard deviations from mean
    iqr_multiplier=1.5,        # IQR multiplier for bounds
    use_robust_stats=False,    # Use median/MAD instead of mean/std
    per_column=True,           # Compute per-column statistics
    columns=None,              # Specific columns (None = all numeric)
)

detector = StatisticalAnomalyDetector(config)
detector.fit(train_data)

# Get computed statistics
stats = detector.get_statistics()
# {'column1': {'mean': 100.5, 'std': 15.2, ...}, ...}

# Score data (0-1 normalized)
scores = detector.score(test_data)

Z-Score Computation:

z_score = |value - mean| / std
normalized_score = min(1.0, z_score / threshold)

IQR Computation:

bounds = [Q1 - k*IQR, Q3 + k*IQR]  where k=1.5 (default)
normalized_score = min(1.0, distance_from_bounds / IQR)

MAD Computation:

MAD = median(|Xi - median(X)|)
modified_z = 0.6745 * |value - median| / MAD
normalized_score = min(1.0, modified_z / threshold)

Isolation Forest Detector

Pure Python implementation of Isolation Forest algorithm.

from truthound.ml.anomaly_models.isolation_forest import (
    IsolationForestDetector,
    IsolationForestConfig,
)

config = IsolationForestConfig(
    n_estimators=100,          # Number of trees
    max_samples=256,           # Samples per tree
    max_depth=None,            # Tree depth (auto if None)
    bootstrap=True,            # Use bootstrap sampling
    columns=None,              # Specific columns
)

detector = IsolationForestDetector(config)
detector.fit(train_data)

# Get anomaly scores (0-1 where 1 = most anomalous)
scores = detector.score(test_data)

# Get feature importance
importance = detector.get_feature_importance()
# {'column1': 0.25, 'column2': 0.35, ...}

Score Computation:

score = 2^(-E(h(x)) / c(n))

Where:
- E(h(x)) = average path length to isolate sample
- c(n) = average path length for unsuccessful search
- c(n) = 2 * (ln(n-1) + 0.5772) - (2 * (n-1) / n)

Ensemble Detector

Combine multiple detectors with different strategies.

from truthound.ml.anomaly_models.ensemble import (
    EnsembleAnomalyDetector,
    EnsembleConfig,
    EnsembleStrategy,
)

config = EnsembleConfig(
    strategy=EnsembleStrategy.WEIGHTED_AVERAGE,
    weights=[0.3, 0.3, 0.4],   # Detector weights
    vote_threshold=0.5,         # For VOTE strategy
)

ensemble = EnsembleAnomalyDetector(config)

# Add detectors
ensemble.add_detector(ml.ZScoreAnomalyDetector(), weight=0.3)
ensemble.add_detector(ml.IQRAnomalyDetector(), weight=0.3)
ensemble.add_detector(ml.IsolationForestDetector(), weight=0.4)

ensemble.fit(train_data)
result = ensemble.predict(test_data)

Ensemble Strategies:

Strategy	Description
AVERAGE	Simple arithmetic mean of scores
WEIGHTED_AVERAGE	Weighted by detector reliability
MAX	Conservative - maximum score across detectors
MIN	Very conservative - minimum score
VOTE	Majority voting with threshold
UNANIMOUS	All detectors must agree

Result Types

from truthound.ml.base import AnomalyResult, AnomalyScore, AnomalyType

# AnomalyResult contains detection results
result = detector.predict(test_data)

print(f"Anomaly count: {result.anomaly_count}")
print(f"Anomaly ratio: {result.anomaly_ratio:.1%}")
print(f"Total points: {result.total_points}")
print(f"Threshold used: {result.threshold_used}")
print(f"Detection time: {result.detection_time_ms:.2f}ms")

# Get only anomalous points
anomalies = result.get_anomalies()

# Iterate through scores
for score in result:
    if score.is_anomaly:
        print(f"Index {score.index}: score={score.score:.3f}, type={score.anomaly_type}")
        print(f"  Contributing features: {score.contributing_features}")

Anomaly Types:

Type	Description
POINT	Single point anomaly
CONTEXTUAL	Anomaly in context (normal value, wrong time)
COLLECTIVE	Group of related anomalies
PATTERN	Pattern violation
TREND	Trend deviation
SEASONAL	Seasonal violation

---

Drift Detection

Quick Drift Detection with truthound.drift.compare()

For simple drift detection between two datasets, use compare() from truthound.drift with 14 available methods:

from truthound.drift import compare

# Auto-select best method based on column type
drift = compare("baseline.csv", "current.csv", method="auto")

# Statistical tests (numeric columns)
drift = compare("baseline.csv", "current.csv", method="ks")          # Kolmogorov-Smirnov
drift = compare("baseline.csv", "current.csv", method="psi")         # Population Stability Index
drift = compare("baseline.csv", "current.csv", method="cvm")         # Cramér-von Mises
drift = compare("baseline.csv", "current.csv", method="anderson")    # Anderson-Darling

# Divergence metrics
drift = compare("baseline.csv", "current.csv", method="kl")          # Kullback-Leibler divergence
drift = compare("baseline.csv", "current.csv", method="js")          # Jensen-Shannon divergence

# Distance metrics (any column type)
drift = compare("baseline.csv", "current.csv", method="wasserstein")   # Earth Mover's Distance
drift = compare("baseline.csv", "current.csv", method="hellinger")     # Hellinger distance (bounded)
drift = compare("baseline.csv", "current.csv", method="bhattacharyya") # Bhattacharyya distance
drift = compare("baseline.csv", "current.csv", method="tv")            # Total Variation distance

# Distance metrics (numeric only)
drift = compare("baseline.csv", "current.csv", method="energy")      # Energy distance
drift = compare("baseline.csv", "current.csv", method="mmd")         # Maximum Mean Discrepancy

# For categorical columns
drift = compare("baseline.csv", "current.csv", method="chi2")        # Chi-squared

# Check results
if drift.has_drift:
    for col in drift.columns:
        if col.result.drifted:
            print(f"{col.column}: {col.result.method} = {col.result.statistic:.4f}")

See Python API: compare() in truthound.drift for full documentation.

ML-Based Drift Detection

For advanced ML-based drift detection with model monitoring capabilities:

Detector	Type	Methods	Use Case
DistributionDriftDetector	Univariate	PSI, KS, Jensen-Shannon, Wasserstein	Input distribution changes
FeatureDriftDetector	Feature-level	Per-feature statistics	Individual feature monitoring
ConceptDriftDetector	Concept	DDM, ADWIN, Page-Hinkley	Model decision boundary shifts
MultivariateDriftDetector	Multivariate	PCA, Correlation, Mahalanobis	Joint distribution changes

Distribution Drift Detection

Detects changes in univariate distributions.

from truthound.ml.drift_detection.distribution import (
    DistributionDriftDetector,
    DistributionDriftConfig,
)

config = DistributionDriftConfig(
    method="psi",              # psi, ks, jensen_shannon, wasserstein
    n_bins=10,                 # Histogram bins
    min_samples=30,            # Minimum required samples
)

detector = DistributionDriftDetector(config)
detector.fit(reference_data)

result = detector.detect(reference_data, current_data)

print(f"Drift detected: {result.is_drifted}")
print(f"Drift score: {result.drift_score:.3f}")
print(f"Drift type: {result.drift_type}")  # gradual, sudden, none

# Per-column scores
for col, score in result.column_scores:
    if score > 0.1:
        print(f"  {col}: {score:.3f}")

# Get drifted columns
drifted = result.get_drifted_columns(threshold=0.1)

Distribution Methods:

Method	Description	Interpretation
psi	Population Stability Index	<0.1 stable, 0.1-0.25 small drift, >0.25 significant
ks	Kolmogorov-Smirnov test	p-value based significance
jensen_shannon	Symmetric KL divergence	0-1 range, 0 = identical
wasserstein	Earth-Mover distance	Minimum transport cost
kl	Kullback-Leibler divergence	Information loss measurement
cvm	Cramér-von Mises test	More sensitive to tails than KS
anderson	Anderson-Darling test	Most sensitive to tail differences
hellinger	Hellinger distance	Bounded [0,1], true metric
bhattacharyya	Bhattacharyya distance	Classification error bounds
tv	Total Variation distance	Max probability difference
energy	Energy distance	Location/scale sensitivity
mmd	Maximum Mean Discrepancy	High-dimensional kernel-based

Note: For quick drift checks, use compare() from truthound.drift instead. The ML module is better suited for continuous monitoring and model training workflows.

Feature Drift Detection

Monitors per-feature statistics for drift.

from truthound.ml.drift_detection.feature import (
    FeatureDriftDetector,
    FeatureDriftConfig,
)

config = FeatureDriftConfig(
    track_stats=["mean", "std", "min", "max", "null_ratio"],
    relative_threshold=True,   # Use relative vs absolute thresholds
    alert_on_new_values=True,  # Flag new categorical values
    categorical_threshold=0.2,
)

detector = FeatureDriftDetector(config)
detector.fit(reference_data)

result = detector.detect(reference_data, current_data)

# Access feature-level drift scores
for score in result.feature_scores:
    print(f"{score.feature}: {score.drift_score:.3f}")
    print(f"  Reference: {score.reference_stats}")
    print(f"  Current: {score.current_stats}")

Tracked Statistics:

• Numeric: mean, std, min, max, Q1, Q2, Q3, null_ratio, skewness, kurtosis
• Categorical: value distribution, new unique values, category shifts

Concept Drift Detection

Detects changes in feature-target relationships.

from truthound.ml.drift_detection.concept import (
    ConceptDriftDetector,
    ConceptDriftConfig,
)

config = ConceptDriftConfig(
    target_column="label",
    method="ddm",              # ddm, adwin, page_hinkley
    warning_threshold=2.0,     # Std devs for warning
    drift_threshold=3.0,       # Std devs for drift
    min_window=30,             # Minimum samples before check
    feature_columns=None,      # Specific features
)

detector = ConceptDriftDetector(config)
detector.fit(reference_data)

result = detector.detect(reference_data, current_data)

Concept Drift Methods:

Method	Description
ddm	Drift Detection Method - monitors error rate
adwin	Adaptive Windowing - auto-adjusts window size
page_hinkley	Cumulative sum - detects small gradual changes

Multivariate Drift Detection

Detects joint distribution changes.

from truthound.ml.drift_detection.multivariate import (
    MultivariateDriftDetector,
    MultivariateDriftConfig,
)

config = MultivariateDriftConfig(
    method="correlation",      # pca, correlation, mahalanobis
    n_components=None,         # For PCA
    correlation_threshold=0.3,
    columns=None,              # Specific columns
)

detector = MultivariateDriftDetector(config)
detector.fit(reference_data)

result = detector.detect(reference_data, current_data)

Multivariate Methods:

Method	Description
correlation	Frobenius norm of correlation matrix difference
pca	Changes in explained variance ratios
mahalanobis	Covariance-aware distance metric

---

Model Monitoring

ModelMonitor

Unified monitoring interface for ML models in production.

from truthound.ml.monitoring import (
    ModelMonitor,
    MonitorConfig,
)

config = MonitorConfig(
    batch_size=100,
    collect_interval_seconds=60,
    alert_evaluation_interval_seconds=30,
    retention_hours=24,
    enable_drift_detection=True,
    enable_quality_metrics=True,
)

monitor = ModelMonitor(config)

# Register models
monitor.register_model("fraud-detector", config)

# Start monitoring (background tasks)
await monitor.start()

# Record predictions
await monitor.record_prediction(
    model_id="fraud-detector",
    features={"amount": 150.0, "merchant_type": "online"},
    prediction=0.85,
    actual=1,                  # Optional: for quality metrics
    latency_ms=5.2,
)

# Get dashboard data (includes metrics history)
dashboard = await monitor.get_dashboard_data("fraud-detector")

# Access historical metrics from dashboard
metrics_history = dashboard.metrics_history
active_alerts = dashboard.active_alerts
health_score = dashboard.health_score

# Stop monitoring
await monitor.stop()

Metric Collectors

from truthound.ml.monitoring.collectors import (
    PerformanceCollector,
    DriftCollector,
    QualityCollector,
    CompositeCollector,
)

# Performance metrics (latency, throughput, error rate)
perf = PerformanceCollector(
    window_size=1000,
    percentiles=[0.5, 0.95, 0.99],
    throughput_window_seconds=60,
)

# Drift metrics
drift = DriftCollector()

# Quality metrics (accuracy, precision, recall, F1)
quality = QualityCollector()

# Combine collectors
composite = CompositeCollector([perf, drift, quality])

Alert Rules

from truthound.ml.monitoring.alerting import (
    ThresholdRule,
    AnomalyRule,
    TrendRule,
    AlertSeverity,
)

# Threshold-based alerting
latency_rule = ThresholdRule(
    name="high-latency",
    metric_name="latency_ms",
    threshold=100.0,
    comparison="gt",
    severity=AlertSeverity.WARNING,
    for_duration_seconds=60,
)

# Anomaly-based alerting
anomaly_rule = AnomalyRule(
    name="latency-anomaly",
    metric_name="latency_ms",
    window_size=100,
    std_threshold=3.0,
    severity=AlertSeverity.ERROR,
)

# Trend-based alerting
trend_rule = TrendRule(
    name="degrading-accuracy",
    metric_name="accuracy",
    direction="decreasing",
    slope_threshold=0.01,
    lookback_minutes=60,
    severity=AlertSeverity.CRITICAL,
)

# Add rules to monitor
monitor.add_rule(latency_rule)
monitor.add_rule(anomaly_rule)
monitor.add_rule(trend_rule)

Alert Handlers

from truthound.ml.monitoring.alerting import (
    SlackAlertHandler,
    PagerDutyAlertHandler,
    WebhookAlertHandler,
)

# Slack notifications
slack = SlackAlertHandler(
    webhook_url="https://hooks.slack.com/...",
    channel="#ml-alerts",
    mention_users=["@oncall"],
)

# PagerDuty incidents
pagerduty = PagerDutyAlertHandler(
    routing_key="...",
    severity_mapping={
        AlertSeverity.CRITICAL: "critical",
        AlertSeverity.ERROR: "error",
    },
)

# Custom webhook
webhook = WebhookAlertHandler(url="https://my-service/alerts")

---

Rule Learning

DataProfileRuleLearner

Automatically generate validation rules from data profiles.

from truthound.ml.rule_learning import (
    DataProfileRuleLearner,
    ProfileLearnerConfig,
)

config = ProfileLearnerConfig(
    strictness="medium",           # loose, medium, strict
    include_range_rules=True,
    include_uniqueness_rules=True,
    include_format_rules=True,
    include_null_rules=True,
    include_type_rules=True,
    null_threshold=0.01,           # Gen rule if < 1% nulls
    uniqueness_threshold=0.99,     # Gen rule if > 99% unique
)

learner = DataProfileRuleLearner(config)
result = learner.learn_rules(data)

print(f"Learned {result.total_rules} rules in {result.learning_time_ms:.0f}ms")

for rule in result.rules:
    print(f"{rule.name}: {rule.condition}")
    print(f"  Support: {rule.support:.1%}, Confidence: {rule.confidence:.1%}")
    print(f"  Validator: {rule.validator_config}")

# Convert to validation suite
suite_config = result.to_validation_suite()

Generated Rule Types:

Rule Type	Description
Range	Min/max bounds for numeric columns
Null	Completeness requirements
Uniqueness	Primary key detection
Type	Data type validation
Format	Regex patterns, length constraints

---

Configuration Reference

AnomalyConfig

from truthound.ml.base import AnomalyConfig

config = AnomalyConfig(
    # Core settings
    contamination=0.1,         # Expected outlier ratio
    sensitivity=0.5,           # Detection sensitivity
    min_samples=100,           # Minimum training samples
    window_size=None,          # Temporal window size
    score_threshold=None,      # Override auto threshold
    columns=None,              # Specific columns

    # Common ML settings
    sample_size=None,          # Max training samples
    random_seed=42,            # Reproducibility
    n_jobs=1,                  # Parallel jobs
    cache_predictions=True,    # Cache results
    verbose=0,                 # 0=silent, 1=progress, 2=debug
)

DriftConfig

from truthound.ml.base import DriftConfig

config = DriftConfig(
    reference_window=1000,     # Reference data size
    detection_window=100,      # Detection data size
    threshold=0.05,            # Drift threshold
    min_samples_per_window=30, # Minimum samples
    n_bins=10,                 # Histogram bins
    detect_gradual=True,       # Detect gradual drift
    detect_sudden=True,        # Detect sudden drift
)

RuleLearningConfig

from truthound.ml.base import RuleLearningConfig

config = RuleLearningConfig(
    min_support=0.1,           # Rule support
    min_confidence=0.8,        # Rule confidence
    max_rules=100,             # Max rules
    max_antecedent_length=3,   # Max conditions
    include_negations=False,   # Include negations
)

---

Model Registry

from truthound.ml.base import model_registry, register_model

# List available models
print(model_registry.list_all())

# Get model by name
detector_class = model_registry.get("IsolationForestDetector")

# Register custom model
@register_model("my_detector")
class MyDetector(AnomalyDetector):
    ...

---

Model Persistence

# Save trained model
detector.save("my_detector.json")

# Load into a new instance of the same algorithm (load returns None)
loaded = IsolationForestDetector()
loaded.load("my_detector.json")

# Use loaded model
result = loaded.predict(new_data)

Development sources write a versioned truthound-ml JSON envelope (version 2). It retains configuration and learned state, including isolation trees, ensemble children/weights, drift reference frames and learned rules. Rich values use data-only tags; no pickle or file-directed imports are supported. Loading a different algorithm, ambiguous JSON or incomplete legacy learned state raises ModelLoadError and does not mark the model ready. Complete legacy statistical states remain readable. Keep the original runtime/dependency versions when comparing numerical results; arbitrary cross-version compatibility is not promised. Custom ensemble children must be explicitly supplied by the application before loading. Drift model files contain the reference data and must be protected as sensitive data. These persistence repairs are not part of the published 3.2.0.

---

Performance Characteristics

Operation	Complexity	Notes
Statistical fit	O(n)	Single pass
Statistical score	O(m)	m = test rows
Isolation Forest fit	O(n log n * k)	k = n_estimators
Isolation Forest score	O(m * k * d)	d = tree depth
Distribution drift	O(n + m)	Both datasets
Feature drift	O(n + m) * cols	Per-column

---

Thread Safety

• All models use threading.RLock() for concurrent access
• Model state machine tracks lifecycle: UNTRAINED → TRAINING → TRAINED → READY
• Global tracker can be set/retrieved with thread-safe locks

---

See Also

• Drift Detection - Validator-based drift detection
• Model Monitoring - CI/CD integration
• Performance Tuning - Optimization strategies