turning batch into mlops

README.md

@@ -11,7 +11,7 @@ license: mit
 
 # Under Construction
 
-Build a production-ready ML inference service with post-deployment drift detection, governance, and alerting, demonstrating real MLOps practices rather than offline modeling.
+Building a production-ready ML inference service with post-deployment drift detection, governance, and alerting, demonstrating real MLOps practices rather than offline modeling.
 
 py -3.9 -m venv .venv
 

app/api/routes.py

@@ -9,15 +9,17 @@ from app.monitoring.drift import run_drift_check
 from app.monitoring.governance import run_governance_checks
 
 import pandas as pd
-import numpy as np  # for numeric handling
+import numpy as np
+import os
 
 templates = Jinja2Templates(directory="app/templates")
-
 router = APIRouter()
 predictor = Predictor()
 
+# Production log file
+PROD_LOG = "data/production/predictions_log.csv"
+
 
-# CSV upload & prediction
 @router.post("/predict")
 async def predict_file(background_tasks: BackgroundTasks, file: UploadFile = File(...)):
     df = pd.read_csv(file.file)
@@ -38,13 +40,11 @@ async def predict_file(background_tasks: BackgroundTasks, file: UploadFile = Fil
             "risk_level": "High" if proba >= 0.75 else "Medium" if proba >= 0.5 else "Low"
         })
 
-    # ---- Drift: run once immediately to return chart data ----
+    # ---- Drift: immediate for frontend ----
     reference_df = pd.read_csv("models/v1/reference_data.csv")
-
-    # Correctly get numeric drift scores per column
     _, drift_dict = run_drift_check(df[predictor.features], reference_df[predictor.features], "v1")
 
-    # Ensure numeric drift values safe for frontend Plotly chart
+    # Safe numeric drift values for chart
     drift_for_chart = []
     for col, score in drift_dict.items():
         try:
@@ -55,19 +55,30 @@ async def predict_file(background_tasks: BackgroundTasks, file: UploadFile = Fil
             score_value = 0.0
         drift_for_chart.append({"column": col, "score": score_value})
 
-    # Schedule full drift in background as before
+    # ---- Append predictions to production log ----
+    df_log = df.copy()
+    df_log["prediction"] = preds
+    df_log["probability"] = probas
+    df_log["model_version"] = predictor.model_version
+    df_log["timestamp"] = pd.Timestamp.utcnow()
+
+    os.makedirs(os.path.dirname(PROD_LOG), exist_ok=True)
+    if not os.path.exists(PROD_LOG):
+        df_log.to_csv(PROD_LOG, index=False)
+    else:
+        df_log.to_csv(PROD_LOG, mode="a", header=False, index=False)
+
+    # ---- Background full drift check ----
     background_tasks.add_task(run_drift_check, df[predictor.features], reference_df[predictor.features], "v1")
 
     return JSONResponse({"n_rows": len(results), "results": results, "drift": drift_for_chart})
 
 
-# Health
 @router.get("/health")
 def health():
     return {"status": "ok"}
 
 
-# Manual drift run
 @router.get("/run-drift")
 def run_drift():
     current_df = load_production_data()
@@ -75,7 +86,6 @@ def run_drift():
     return {"status": "drift_check_completed", "report_path": report_path}
 
 
-# Monitoring pipeline
 @router.get("/monitoring/run")
 def monitoring_run(background_tasks: BackgroundTasks, model_version: str = "v1"):
     current_data = pd.read_csv("data/processed/current_data.csv")
@@ -87,7 +97,6 @@ def monitoring_run(background_tasks: BackgroundTasks, model_version: str = "v1")
     return {"status": "monitoring triggered", "model_version": model_version}
 
 
-# Dashboard
 @router.get("/dashboard")
 def dashboard(request: Request):
     return templates.TemplateResponse("dashboard.html", {"request": request})

app/inference/predictor.py

@@ -1,3 +1,5 @@
+# app/inference/predictor.py
+
 # model.predict wrapper
 import json
 import joblib
@@ -11,6 +13,8 @@ class Predictor:
         with open(FEATURES_PATH, "r") as f:
             self.features = json.load(f)
 
+        self.model_version = "v1"
+
     def predict(self, df):
         X = df[self.features]
         probas = self.model.predict_proba(X)[:, 1]

data/production/predictions_log.csv

@@ -0,0 +1 @@
+50000.0,22,0,0,45458.0,46450.0,2051.0,2200.0,1,0,0.20854088888292008,v1,2026-01-14 13:41:57.526400+00:00

scripts/simulate_inference.py

@@ -0,0 +1,25 @@
+# scripts/simulate_inference.py
+import pandas as pd
+import requests
+import random
+import time
+
+df = pd.read_csv("data/processed/current_data.csv")
+
+# Sample 1-5 rows randomly
+sample = df.sample(random.randint(1,5))
+csv_bytes = sample.to_csv(index=False).encode("utf-8")
+
+# POST to FastAPI predict endpoint
+response = requests.post(
+    "http://localhost:8000/predict",
+    files={"file": ("sample.csv", csv_bytes, "text/csv")}
+)
+
+print("Status:", response.status_code)
+try:
+    print("Response:", response.json())
+except Exception:
+    print("Server returned non-JSON response:")
+    print(response.text)
+