Kafka (predictions)    Kafka (ground truth / labels)
         \                     /
          \                   /
           Spark Structured Streaming
                    ↓
          Join predictions + labels by transaction_id
                    ↓
         Real-time metrics (accuracy, precision, recall, AUC, calibration)
                    ↓
   ───────────────────────────────────────┐
   │                                      │
Delta Lake (historical)      Prometheus + Grafana (live dashboards)
   │                                      │
   ↓                                      ↓
Auto-alerts (Slack/Email/PagerDuty)   Live Model Health Dashboard

# producer_predictions.py  →  runs forever
from kafka import KafkaProducer
import json, time, random, uuid
from datetime import datetime

producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda x: json.dumps(x).encode('utf-8')
)

while True:
    tx_id = str(uuid.uuid4())
    event = {
        "transaction_id": tx_id,
        "amount": round(random.uniform(5, 5000), 2),
        "predicted_fraud_score": random.random(),
        "predicted_fraud": int(random.random() > 0.97),   # ~3% fraud predictions
        "event_time": datetime.utcnow().isoformat() + "Z",
        "model_version": "rf_v2_2025_11"
    }
    producer.send("fraud_predictions", value=event)
    print("Pred:", event)

    # Ground truth arrives 30 seconds – 10 minutes later
    delay = random.randint(30, 600)
    time.sleep(delay)

    actual = 1 if random.random() > 0.99 else 0   # true fraud ~1%
    label_event = {
        "transaction_id": tx_id,
        "actual_fraud": actual,
        "label_time": datetime.utcnow().isoformat() + "Z"
    }
    producer.send("fraud_labels", value=label_event)
    print("Label:", label_event)

    time.sleep(0.1)

# performance_monitoring.py  →  run 24/7
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
import matplotlib.pyplot as plt

spark = SparkSession.builder \
    .appName("LiveModelPerformance") \
    .config("spark.sql.streaming.statefulOperator.checkCorrectness.enabled", "false") \
    .getOrCreate()

# 1. Read both streams
pred_schema = StructType([
    StructField("transaction_id", StringType()),
    StructField("predicted_fraud_score", DoubleType()),
    StructField("predicted_fraud", LongType()),
    StructField("event_time", TimestampType()),
    StructField("model_version", StringType())
])

label_schema = StructType([
    StructField("transaction_id", StringType()),
    StructField("actual_fraud", LongType()),
    StructField("label_time", TimestampType())
])

predictions = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "fraud_predictions") \
    .load() \
    .select(from_json(col("value").cast("string"), pred_schema).alias("p")) \
    .select("p.*") \
    .withWatermark("event_time", "12 hours")

labels = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "fraud_labels") \
    .load() \
    .select(from_json(col("value").cast("string"), label_schema).alias("l")) \
    .select("l.*") \
    .withWatermark("label_time", "12 hours")

# 2. Join predictions + labels with 6-hour tolerance
joined = predictions.alias("p") \
    .join(
        labels.alias("l"),
        expr("""
            p.transaction_id = l.transaction_id AND
            l.label_time BETWEEN p.event_time AND p.event_time + interval 6 hours
        """),
        "left"
    ) \
    .select(
        "p.*",
        "l.actual_fraud"
    )

# 3. Real-time metrics every 60 seconds
windowed_metrics = joined \
    .withWatermark("event_time", "2 hours") \
    .groupBy(window(col("event_time"), "60 seconds"), col("model_version")) \
    .agg(
        count("*").alias("total"),
        sum(when(col("actual_fraud").isNotNull(), 1).otherwise(0)).alias("labelled"),
        sum(when((col("predicted_fraud") == col("actual_fraud")) & col("actual_fraud").isNotNull(), 1).otherwise(0)).alias("correct"),
        sum(when((col("predicted_fraud") == 1) & col("actual_fraud").isNotNull(), 1).otherwise(0)).alias("tp_pred"),
        sum(when((col("actual_fraud") == 1) & col("predicted_fraud").isNotNull(), 1).otherwise(0)).alias("tp_actual")
    ) \
    .withColumn("accuracy", col("correct") / col("labelled")) \
    .withColumn("precision", col("tp_pred") / when(sum("predicted_fraud") > 0, sum("predicted_fraud")).otherwise(1)) \
    .withColumn("recall", col("tp_actual") / when(sum("actual_fraud") > 0, sum("actual_fraud")).otherwise(1)) \
    .withColumn("f1", 2 * col("precision") * col("recall") / (col("precision") + col("recall")))

# 4. Write to monitoring table + Prometheus
monitoring_query = windowed_metrics \
    .writeStream \
    .outputMode("update") \
    .format("delta") \
    .option("checkpointLocation", "/tmp/checkpoint_perf") \
    .table("model_performance_live")

# 5. Alert if accuracy drops >10% from baseline
def alert_on_degradation(batch_df, batch_id):
    pdf = batch_df.toPandas()
    for _, row in pdf.iterrows():
        if row["accuracy"] < 0.85:  # your threshold
            requests.post(
                "https://hooks.slack.com/services/YOUR/WEBHOOK",
                json={"text": f"*MODEL DEGRADED* Accuracy = {row['accuracy']:.1%} (window {row['window']})"}
            )

alert_query = windowed_metrics.writeStream.foreachBatch(alert_on_degradation).start()

monitoring_query.awaitTermination()

docker run -d -p 9090:9090 prom/prometheus
docker run -d -p 3000:3000 grafana/grafana