watchdogv2 1.0.0

Ultra-lightweight structured metrics tracking and anomaly detection — Z-score, percentile, rate-of-change, alerts, SQLite persistence. Zero dependencies.

watchdog-lite

Ultra-lightweight metrics tracking and anomaly detection for Python services. Z-score, percentile, and rate-of-change detection. Structured alerts, SQLite persistence, and a live stats table. No Datadog, no Grafana, no infra. Zero dependencies.

---

The Problem

You have a Python service. Something goes wrong. You find out from a user. Setting up Datadog or Prometheus takes days. You just want:

"Tell me when my API latency spikes or error rate jumps — right inside my code."

---

Installation

pip install watchdog-lite

No dependencies. Requires Python 3.8+.

---

Quick Start

from watchdog_lite import Watchdog

wd = Watchdog(db_path="metrics.db")

@wd.track("payment_api", latency_threshold=2.0, error_rate_threshold=0.05)
def call_payment_api(amount):
    return stripe.charge(amount)

@wd.on_alert
def handle_alert(alert):
    slack.send(f"🚨 {alert.metric}: {alert.message}")

wd.print_stats()

---

Usage

Track any function automatically

@wd.track(
    name="payment_api",
    latency_threshold=2.0,       # alert if p95 > 2s
    error_rate_threshold=0.05,   # alert if >5% errors
    method="both",               # use both zscore + percentile detection
)
def call_payment_api(amount):
    return stripe.charge(amount)

Measure any code block

with wd.measure("db_query"):
    results = db.execute(query)

with wd.measure("render"):
    html = template.render(context)

Record custom metrics

wd.record("queue_depth", queue.size())
wd.record("cache_hit_rate", cache.hit_rate())
wd.record("model_latency", inference_time, is_error=failed)

Anomaly Detection Methods

# Z-score: statistical — alerts when value is N std deviations from mean
wd.add_rule("latency", method="zscore", threshold=3.0)

# Percentile: threshold-based — alerts when p95 exceeds a value
wd.add_rule("latency", method="percentile", threshold=2.0, percentile=95)

# Rate of change: alerts when metric changes >X% suddenly
wd.add_rule("queue_depth", method="rate_of_change", threshold=50.0)

# Error rate: alerts when errors exceed a fraction
wd.add_rule("payment_api", method="error_rate", threshold=0.05)

Alerts

from watchdog_lite import AlertLevel

# Decorator style
@wd.on_alert
def handle(alert):
    print(f"[{alert.level}] {alert.metric}: {alert.message}")
    print(f"  value={alert.value:.4f}, threshold={alert.threshold}")
    print(f"  triggered at {alert.triggered_at}")

# Or register directly
wd.on_alert(lambda a: send_pagerduty(a) if a.level == AlertLevel.CRITICAL else None)

# Set alert severity per rule
wd.add_rule("latency", method="zscore", threshold=3.0, level=AlertLevel.WARNING)
wd.add_rule("error_rate", method="error_rate", threshold=0.1, level=AlertLevel.CRITICAL)

Live Stats Table

wd.print_stats()

Metric                         Count     Mean      p50      p95      p99    Err%  Status
──────────────────────────────────────────────────────────────────────────────────────────
  payment_api                    842   0.3421   0.3100   0.8200   1.2100   0.2%  ✅
  db_query                       421   0.0821   0.0700   0.2100   0.4500   0.0%  ✅
  render                        1683   0.0121   0.0100   0.0300   0.0500   0.0%  ✅
  queue_depth                     12  42.0000  41.0000  68.0000  72.0000   0.0%  ⚠️

Manual Detection

# Check right now without waiting for a rule to fire
alert = wd.detect("latency", method="zscore", threshold=3.0)
if alert:
    print(f"Anomaly: {alert.message}")

SQLite Persistence

wd = Watchdog(db_path="metrics.db", flush_interval=30.0)

# Metrics are automatically flushed every 30 seconds
# Manual flush:
wd.flush()

# Query historical data
history = wd.history_from_db("payment_api", limit=100)
for snapshot in history:
    print(snapshot["snapped_at"], snapshot["p95"])

Export Snapshot

wd.export("metrics.json")
# Writes: {metrics: {...}, alerts: [...], exported_at: "..."}

---

Anomaly Detection Reference

Method	What it detects	When to use
zscore	Statistical outliers (N std deviations from mean)	Latency spikes, throughput anomalies
percentile	Absolute threshold violations (p95 > X)	SLA enforcement
rate_of_change	Sudden % change in a metric	Queue depth spikes, traffic surges
error_rate	Error fraction above threshold	API reliability

---

API Reference

Watchdog

Watchdog(
    window_size=100,      # Rolling window size
    db_path=None,         # SQLite file for persistence
    flush_interval=30.0,  # Seconds between DB flushes
)

Method	Description
track(name, latency_threshold, ...)	Decorator to auto-track a function
measure(metric)	Context manager for a code block
record(metric, value, is_error)	Record a raw value
add_rule(metric, method, threshold, ...)	Add anomaly detection rule
detect(metric, method, threshold)	Manual one-time detection
on_alert(func)	Register alert callback
stats(metric=None)	Get stats dict
alerts(metric=None)	Get triggered alerts
history_from_db(metric, limit)	Load history from SQLite
export(path)	Export snapshot to JSON
flush()	Manual SQLite flush
print_stats()	Print formatted table

---

Running Tests

pip install pytest
pytest tests/ -v

---

License

MIT © prabhay759