Monitoring & Alerts API¶

from qldata.monitoring import DataQualityMonitor, AlertManager

# Setup monitoring
monitor = DataQualityMonitor(stale_threshold_seconds=10)
alerts = AlertManager()

# Configure alerts
alerts.on_stale_data(lambda: print("Data is stale!"))
alerts.on_high_latency(lambda ms: print(f"High latency: {ms}ms"))

from qldata.monitoring import DataQualityMonitor
from datetime import datetime, timezone

# Create monitor
monitor = DataQualityMonitor(stale_threshold_seconds=10)

# Record incoming messages
def on_data(df):
    if not df.empty:
        monitor.record_message(df.index[-1])
        # Process data...

# Get current metrics
metrics = monitor.get_metrics()
print(f"Throughput: {metrics['throughput']:.1f}/s")
print(f"P95 Latency: {metrics['latency_p95']:.1f}ms")
print(f"Is Stale: {metrics['is_stale']}")

status = monitor.get_health_status()
# Returns: "healthy", "degraded", or "unhealthy"

from qldata.monitoring import AlertManager

alerts = AlertManager()

# Register callbacks
alerts.on_high_latency(lambda ms: print(f"High latency: {ms}ms"))
alerts.on_stale_data(lambda: print("Data is stale!"))
alerts.on_low_throughput(lambda tps: print(f"Low throughput: {tps}/s"))
alerts.on_connection_lost(lambda: print("Connection lost"))
alerts.on_reconnected(lambda n: print(f"Reconnected after {n} attempts"))

from qldata.monitoring import AlertManager

alerts = AlertManager()

# Check latency with thresholds
alert = alerts.check_latency(
    latency_ms=1200,
    warning_threshold=500,   # Warning above 500ms
    error_threshold=1000     # Error above 1000ms
)

if alert:
    print(f"Alert: {alert.severity} - {alert.message}")

# Get recent alerts
recent = alerts.get_recent_alerts(limit=10)

for alert in recent:
    print(f"[{alert.severity}] {alert.timestamp}: {alert.message}")

import qldata as qd
from qldata.monitoring import DataQualityMonitor, AlertManager
import time

# Setup
monitor = DataQualityMonitor(stale_threshold_seconds=5)
alerts = AlertManager()

def on_high_latency(ms):
    print(f"⚠️ High latency: {ms:.1f}ms")

def on_stale():
    print("🚨 Data is STALE!")

alerts.on_high_latency(on_high_latency)
alerts.on_stale_data(on_stale)

def handle_data(df):
    if df.empty:
        return

    # Record message
    monitor.record_message(df.index[-1])

    # Check thresholds
    metrics = monitor.get_metrics()
    if metrics["latency_p95"]:
        alerts.check_latency(metrics["latency_p95"], warning_threshold=200)
    alerts.check_stale_data(monitor.is_stale())

# Start stream
stream = qd.stream(["BTCUSDT"], source="binance") \
    .resolution("tick") \
    .on_data(handle_data) \
    .get(start=True)

# Monitor health
while True:
    metrics = monitor.get_metrics()
    print(f"Health: {monitor.get_health_status()} | TPS: {metrics['throughput']:.1f}")
    time.sleep(10)

from qldata.monitoring import AlertManager
import requests

PAGERDUTY_KEY = "your-integration-key"

def send_pagerduty(severity: str, message: str):
    """Send alert to PagerDuty."""
    requests.post(
        "https://events.pagerduty.com/v2/enqueue",
        json={
            "routing_key": PAGERDUTY_KEY,
            "event_action": "trigger",
            "payload": {
                "summary": message,
                "severity": severity,
                "source": "qldata-monitor"
            }
        }
    )

alerts = AlertManager()
alerts.on_stale_data(lambda: send_pagerduty("critical", "Market data is stale"))
alerts.on_high_latency(lambda ms: send_pagerduty("warning", f"Latency: {ms}ms"))

from qldata.monitoring import AlertManager
import requests

SLACK_WEBHOOK = "https://hooks.slack.com/services/..."

def send_slack(message: str, emoji: str = "⚠️"):
    """Send message to Slack."""
    requests.post(
        SLACK_WEBHOOK,
        json={"text": f"{emoji} {message}"}
    )

alerts = AlertManager()
alerts.on_stale_data(lambda: send_slack("Data feed is stale!", "🚨"))
alerts.on_connection_lost(lambda: send_slack("Connection lost", "❌"))
alerts.on_reconnected(lambda n: send_slack(f"Reconnected after {n} attempts", "✅"))

from qldata.monitoring import DataQualityMonitor
from prometheus_client import Gauge, start_http_server

# Prometheus metrics
latency_p95 = Gauge("qldata_latency_p95_ms", "P95 latency in milliseconds")
throughput = Gauge("qldata_throughput_tps", "Throughput in messages/second")
is_stale = Gauge("qldata_is_stale", "1 if data is stale, 0 otherwise")

monitor = DataQualityMonitor()

def update_prometheus():
    """Update Prometheus metrics."""
    metrics = monitor.get_metrics()
    latency_p95.set(metrics["latency_p95"] or 0)
    throughput.set(metrics["throughput"])
    is_stale.set(1 if metrics["is_stale"] else 0)

# Start Prometheus server
start_http_server(8000)

# Update in background
import threading
import time

def metrics_loop():
    while True:
        update_prometheus()
        time.sleep(5)

threading.Thread(target=metrics_loop, daemon=True).start()

import logging
from qldata.monitoring import DataQualityMonitor, AlertManager

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger("qldata.monitor")

alerts = AlertManager()

alerts.on_high_latency(lambda ms: logger.warning(f"High latency: {ms:.1f}ms"))
alerts.on_stale_data(lambda: logger.error("Data is stale!"))
alerts.on_connection_lost(lambda: logger.error("Connection lost"))
alerts.on_reconnected(lambda n: logger.info(f"Reconnected after {n} attempts"))

# Periodic health logging
monitor = DataQualityMonitor()

def log_health():
    metrics = monitor.get_metrics()
    logger.info(
        f"Health: {monitor.get_health_status()} | "
        f"TPS: {metrics['throughput']:.1f} | "
        f"P95: {metrics['latency_p95']:.1f}ms"
    )

# Crypto markets: tighter thresholds
monitor = DataQualityMonitor(stale_threshold_seconds=5)
alerts.check_latency(ms, warning_threshold=100, error_threshold=500)

# Lower-frequency data: relaxed thresholds
monitor = DataQualityMonitor(stale_threshold_seconds=60)
alerts.check_latency(ms, warning_threshold=1000, error_threshold=5000)

# Debounce alerts
from datetime import datetime, timedelta

last_alert = {}

def debounced_alert(alert_type: str, callback):
    now = datetime.now()
    if alert_type not in last_alert or now - last_alert[alert_type] > timedelta(minutes=5):
        last_alert[alert_type] = now
        callback()

def on_stale(symbol: str, last_time: datetime):
    send_alert(f"Symbol {symbol} stale since {last_time}")

monitors = {
    "BTCUSDT": DataQualityMonitor(),
    "ETHUSDT": DataQualityMonitor(),
}

def handle_data(df):
    symbol = df["symbol"].iloc[0]
    monitors[symbol].record_message(df.index[-1])