detectkit 0.4.1

Metric monitoring with automatic anomaly detection

detectkit

Metric monitoring with automatic anomaly detection.

detectkit is a Python library for data analysts and engineers to monitor time-series metrics with automatic anomaly detection and alerting. dbt-like project structure and CLI.

Features

• Pure numpy arrays — no pandas dependency in core logic
• Statistical detectors — Z-Score, MAD, IQR, Manual Bounds
• Multi-channel alerting — Mattermost, Slack, Telegram, Email, Webhook
• @mentions — tag users/groups in alerts, each channel formats natively
• Alert lifecycle — consecutive anomalies, cooldown, recovery notifications
• Database agnostic — ClickHouse, PostgreSQL, MySQL
• Idempotent — resume from interruptions, no duplicate processing
• CLI — dtk init, dtk run --select, tag-based selectors

Installation

pip install detectkit

With database drivers:

pip install detectkit[clickhouse]   # ClickHouse
pip install detectkit[all-db]       # All databases

Quick Start

CLI (Recommended)

# Create project
dtk init my_monitoring
cd my_monitoring

# Configure database in profiles.yml, then:
dtk run --select cpu_usage
dtk run --select tag:critical
dtk run --select cpu_usage --steps load,detect
dtk run --select cpu_usage --from 2024-01-01

Metric Configuration

# metrics/api_errors.yml
name: api_error_rate
interval: "5min"

query: |
  SELECT
    toStartOfInterval(timestamp, INTERVAL 5 MINUTE) AS timestamp,
    countIf(status_code >= 500) / count() * 100 AS value
  FROM http_requests
  WHERE timestamp >= %(from_date)s AND timestamp < %(to_date)s
  GROUP BY timestamp ORDER BY timestamp

detectors:
  - type: mad
    params:
      threshold: 3.0
      window_size: 2016    # 7 days

alerting:
  enabled: true
  channels: [mattermost_ops]
  consecutive_anomalies: 3
  direction: "up"
  mentions: [oncall_engineer, here]
  alert_cooldown: "30min"
  notify_on_recovery: true
  suppress_until: "2026-04-11 18:00:00"  # Suppress alerts until this UTC time

Python API

import numpy as np
from detectkit.detectors.statistical import ZScoreDetector

detector = ZScoreDetector(threshold=3.0, window_size=100)
results = detector.detect({
    'timestamp': np.array([...], dtype='datetime64[ms]'),
    'value': np.array([1.0, 2.0, 1.5, 10.0, 1.8]),
})

for r in results:
    if r.is_anomaly:
        print(f"Anomaly at {r.timestamp}: {r.value}")

Documentation

• Getting Started — 5-minute quickstart
• Configuration Guide — all config options
• Detectors Guide — choosing the right detector
• Alerting Guide — channels, mentions, cooldown, recovery
• CLI Reference — command-line documentation
• Examples — real-world monitoring scenarios
• Changelog — version history

Requirements

• Python 3.10+
• numpy >= 1.24.0
• pydantic >= 2.0.0
• click >= 8.0
• PyYAML >= 6.0
• Jinja2 >= 3.0

License

MIT License — see LICENSE for details.