Time-Series DB in Machine Learning

from influxdb_client_3 import InfluxDBClient3
import pandas as pd
import time

# Initialize client
client = InfluxDBClient3(
    host="https://us-east-1-1.aws.cloud2.influxdata.com",
    token="YOUR_API_TOKEN",
    org="ml-team",
    database="ml_metrics"
)

# Write training metrics using line protocol
def log_training_step(epoch, step, loss, accuracy, lr, model_name):
    """Log a single training step's metrics to InfluxDB."""
    record = (
        f"training_metrics,model={model_name},stage=train "
        f"loss={loss},accuracy={accuracy},learning_rate={lr},"
        f"epoch={epoch}i,step={step}i {int(time.time_ns())}"
    )
    client.write(record=record)

# Simulate logging during training
for epoch in range(10):
    for step in range(100):
        log_training_step(
            epoch=epoch, step=step,
            loss=2.5 / (epoch + 1) + 0.01 * step,
            accuracy=0.5 + 0.05 * epoch,
            lr=0.001 * (0.95 ** epoch),
            model_name="bert-classifier-v3"
        )

# Query: average loss per epoch for the last 24 hours
query = """
SELECT
    epoch,
    AVG(loss) AS avg_loss,
    AVG(accuracy) AS avg_accuracy,
    MIN(learning_rate) AS lr
FROM training_metrics
WHERE time > now() - INTERVAL '24 hours'
    AND model = 'bert-classifier-v3'
GROUP BY epoch
ORDER BY epoch
"""
df = client.query(query=query, language="sql", mode="pandas")
print(df)

-- Create a hypertable for inference metrics
CREATE TABLE inference_metrics (
    time        TIMESTAMPTZ NOT NULL,
    model_name  TEXT NOT NULL,
    endpoint    TEXT NOT NULL,
    latency_ms  DOUBLE PRECISION,
    confidence  DOUBLE PRECISION,
    prediction  TEXT,
    is_drift    BOOLEAN DEFAULT FALSE
);

SELECT create_hypertable('inference_metrics', by_range('time'));

-- Create an index on model_name for fast filtered queries
CREATE INDEX idx_model_time ON inference_metrics (model_name, time DESC);

-- Continuous aggregate: 5-minute rollups per model
CREATE MATERIALIZED VIEW inference_5min
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('5 minutes', time) AS bucket,
    model_name,
    COUNT(*) AS request_count,
    AVG(latency_ms) AS avg_latency,
    percentile_cont(0.99) WITHIN GROUP (ORDER BY latency_ms) AS p99_latency,
    AVG(confidence) AS avg_confidence,
    STDDEV(confidence) AS stddev_confidence,
    SUM(CASE WHEN is_drift THEN 1 ELSE 0 END)::FLOAT / COUNT(*) AS drift_rate
WITH NO DATA;

-- Add a refresh policy: refresh every 5 minutes, covering the last 30 minutes
SELECT add_continuous_aggregate_policy('inference_5min',
    start_offset => INTERVAL '30 minutes',
    end_offset => INTERVAL '5 minutes',
    schedule_interval => INTERVAL '5 minutes'
);

-- Retention policy: keep raw data for 7 days, aggregates for 1 year
SELECT add_retention_policy('inference_metrics', INTERVAL '7 days');
SELECT add_retention_policy('inference_5min', INTERVAL '365 days');

-- Query: detect latency spikes in the last hour
SELECT
    bucket,
    model_name,
    avg_latency,
    p99_latency,
    drift_rate
FROM inference_5min
WHERE bucket > NOW() - INTERVAL '1 hour'
    AND p99_latency > 200
ORDER BY p99_latency DESC;

from prometheus_client import (
    start_http_server, Histogram, Counter, Gauge, Summary
)
import time
import random

# Define ML-specific metrics
INFERENCE_LATENCY = Histogram(
    'ml_inference_latency_seconds',
    'Time spent processing inference request',
    ['model_name', 'model_version'],
    buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5]
)

PREDICTION_CONFIDENCE = Summary(
    'ml_prediction_confidence',
    'Confidence score of model predictions',
    ['model_name']
)

PREDICTION_COUNT = Counter(
    'ml_predictions_total',
    'Total number of predictions made',
    ['model_name', 'predicted_class']
)

MODEL_DRIFT_SCORE = Gauge(
    'ml_model_drift_score',
    'Current drift score (PSI or KL divergence)',
    ['model_name', 'feature_name']
)

GPU_MEMORY_USAGE = Gauge(
    'ml_gpu_memory_bytes',
    'GPU memory usage in bytes',
    ['gpu_id', 'model_name']
)

def simulate_inference():
    """Simulate an ML inference with metric instrumentation."""
    model = "fraud-detector-v2"
    version = "2.1.0"

    # Time the inference
    with INFERENCE_LATENCY.labels(
        model_name=model, model_version=version
    ).time():
        # Simulate model inference
        time.sleep(random.uniform(0.01, 0.15))
        confidence = random.gauss(0.85, 0.1)
        predicted_class = "fraud" if confidence > 0.7 else "legitimate"

    # Record prediction metrics
    PREDICTION_CONFIDENCE.labels(model_name=model).observe(confidence)
    PREDICTION_COUNT.labels(
        model_name=model, predicted_class=predicted_class
    ).inc()

    # Update drift score (computed externally, e.g., by Evidently)
    MODEL_DRIFT_SCORE.labels(
        model_name=model, feature_name="transaction_amount"
    ).set(random.uniform(0.01, 0.3))

    # GPU memory (from nvidia-smi or pynvml)
    GPU_MEMORY_USAGE.labels(
        gpu_id="0", model_name=model
    ).set(random.randint(2_000_000_000, 8_000_000_000))

if __name__ == '__main__':
    # Start Prometheus metrics server on port 8000
    start_http_server(8000)
    print("Metrics server started on :8000/metrics")
    while True:
        simulate_inference()
        time.sleep(0.1)  # ~10 inferences/sec

import requests
import json
from datetime import datetime, timedelta

VICTORIA_METRICS_URL = "http://victoriametrics:8428"

def import_metrics_bulk(metrics: list[dict]):
    """Bulk import metrics to VictoriaMetrics via JSON import API."""
    payload = ""
    for m in metrics:
        payload += json.dumps({
            "metric": {
                "__name__": m["name"],
                "model": m.get("model", "unknown"),
                "environment": m.get("env", "production")
            },
            "values": m["values"],
            "timestamps": m["timestamps"]
        }) + "\n"

    resp = requests.post(
        f"{VICTORIA_METRICS_URL}/api/v1/import",
        data=payload,
        headers={"Content-Type": "application/json"}
    )
    resp.raise_for_status()
    print(f"Imported {len(metrics)} series successfully")

def query_model_latency_trend(model_name: str, hours: int = 24):
    """Query P99 inference latency trend using PromQL."""
    query = (
        f'histogram_quantile(0.99, '
        f'rate(ml_inference_latency_seconds_bucket{{model_name="{model_name}"}}[5m]))'
    )
    end = datetime.utcnow()
    start = end - timedelta(hours=hours)

    resp = requests.get(
        f"{VICTORIA_METRICS_URL}/api/v1/query_range",
        params={
            "query": query,
            "start": start.isoformat() + "Z",
            "end": end.isoformat() + "Z",
            "step": "5m"
        }
    )
    data = resp.json()
    for result in data.get("data", {}).get("result", []):
        print(f"Labels: {result['metric']}")
        for ts, val in result["values"]:
            print(f"  {datetime.fromtimestamp(ts)}: {float(val)*1000:.1f}ms")

def detect_latency_anomaly(model_name: str, threshold_multiplier: float = 2.0):
    """Detect if current P99 latency exceeds 2x the 7-day average."""
    current_query = (
        f'histogram_quantile(0.99, '
        f'rate(ml_inference_latency_seconds_bucket{{model_name="{model_name}"}}[5m]))'
    )
    baseline_query = (
        f'avg_over_time('
        f'histogram_quantile(0.99, '
        f'rate(ml_inference_latency_seconds_bucket{{model_name="{model_name}"}}[5m]))[7d:1h])'
    )

    current = requests.get(
        f"{VICTORIA_METRICS_URL}/api/v1/query",
        params={"query": current_query}
    ).json()
    baseline = requests.get(
        f"{VICTORIA_METRICS_URL}/api/v1/query",
        params={"query": baseline_query}
    ).json()

    current_val = float(current["data"]["result"][0]["value"][1])
    baseline_val = float(baseline["data"]["result"][0]["value"][1])

    if current_val > baseline_val * threshold_multiplier:
        print(f"ANOMALY: P99 latency {current_val*1000:.1f}ms "
              f"exceeds {threshold_multiplier}x baseline {baseline_val*1000:.1f}ms")
        return True
    return False

# Usage
query_model_latency_trend("fraud-detector-v2", hours=6)
detect_latency_anomaly("fraud-detector-v2")

# Prometheus configuration for ML metrics scraping
global:
  scrape_interval: 15s
  evaluation_interval: 15s
  external_labels:
    cluster: ml-prod-ap-south-1
    environment: production

# Remote write to VictoriaMetrics for long-term storage
remote_write:
  - url: http://victoriametrics:8428/api/v1/write
    queue_config:
      max_samples_per_send: 10000
      batch_send_deadline: 5s
      max_shards: 30

scrape_configs:
  - job_name: ml-inference-service
    scrape_interval: 15s
    metrics_path: /metrics
    kubernetes_sd_configs:
      - role: pod
    relabel_configs:
      - source_labels: [__meta_kubernetes_pod_label_app]
        regex: ml-inference-.*
        action: keep
      - source_labels: [__meta_kubernetes_pod_label_model]
        target_label: model_name

  - job_name: ml-training-jobs
    scrape_interval: 30s
    kubernetes_sd_configs:
      - role: pod
    relabel_configs:
      - source_labels: [__meta_kubernetes_pod_label_app]
        regex: training-.*
        action: keep

# Recording rules for pre-computed ML aggregates
rule_files:
  - /etc/prometheus/ml_recording_rules.yml

# Alerting rules for model health
alerting:
  alertmanagers:
    - static_configs:
        - targets: ['alertmanager:9093']