broken stuff

app/api/background_drift.py

@@ -30,7 +30,7 @@ async def drift_loop(interval_seconds: int = 10):
                 continue
 
             prod_df = pd.read_csv(PROD_LOG_PATH)
-
+            
             # ---- Retention window (prevents infinite growth) ----
             if len(prod_df) > MAX_ROWS:
                 prod_df = prod_df.tail(MAX_ROWS)
@@ -50,10 +50,9 @@ async def drift_loop(interval_seconds: int = 10):
 
             reference_df = pd.read_csv(REFERENCE_PATH)
 
+            # ---- FIX: pass reference_df to run_drift_check ----
             _, drift_dict = run_drift_check(
-                prod_df[predictor.features],
-                reference_df[predictor.features],
-                model_version="v1",
+                prod_df[predictor.features], reference_df[predictor.features], model_version="v1"
             )
 
             dashboard_payload = {

app/api/routes.py

@@ -20,6 +20,25 @@ predictor = Predictor()
 # Production log file
 PROD_LOG = "data/production/predictions_log.csv"
 
+# ------------------------------------------------------------------
+# ENSURE production log exists at server startup (CRITICAL FIX)
+# ------------------------------------------------------------------
+os.makedirs(os.path.dirname(PROD_LOG), exist_ok=True)
+
+if not os.path.exists(PROD_LOG):
+    # Create empty production log with correct schema
+    base_cols = list(predictor.features)
+    extra_cols = [
+        "prediction",
+        "probability",
+        "risk_level",
+        "model_version",
+        "timestamp",
+    ]
+    empty_df = pd.DataFrame(columns=base_cols + extra_cols)
+    empty_df.to_csv(PROD_LOG, index=False)
+# ------------------------------------------------------------------
+
 
 @router.post("/predict")
 async def predict_file(background_tasks: BackgroundTasks, file: UploadFile = File(...)):
@@ -28,7 +47,10 @@ async def predict_file(background_tasks: BackgroundTasks, file: UploadFile = Fil
     # ---- STRICT MODE: schema enforcement ----
     missing = set(predictor.features) - set(df.columns)
     if missing:
-        raise HTTPException(status_code=400, detail=f"Invalid schema. Missing required columns: {sorted(missing)}")
+        raise HTTPException(
+            status_code=400,
+            detail=f"Invalid schema. Missing required columns: {sorted(missing)}",
+        )
 
     # ---- Model inference ----
     preds, probas = predictor.predict(df)
@@ -43,9 +65,12 @@ async def predict_file(background_tasks: BackgroundTasks, file: UploadFile = Fil
 
     # ---- Drift: immediate for frontend ----
     reference_df = pd.read_csv("models/v1/reference_data.csv")
-    _, drift_dict = run_drift_check(df[predictor.features], reference_df[predictor.features], "v1")
+    _, drift_dict = run_drift_check(
+        df[predictor.features],
+        reference_df[predictor.features],
+        "v1",
+    )
 
-    # Safe numeric drift values for chart
     drift_for_chart = []
     for col, score in drift_dict.items():
         try:
@@ -58,40 +83,43 @@ async def predict_file(background_tasks: BackgroundTasks, file: UploadFile = Fil
 
     # ---- Append predictions to production log ----
     df_log = df.copy()
+
+    # ---- FIX: Remove existing prediction/risk/probability/etc columns to avoid extra column issue ----
+    for col in ["prediction", "probability", "risk_level", "model_version", "timestamp"]:
+        if col in df_log.columns:
+            df_log = df_log.drop(columns=[col])
+
     df_log["prediction"] = preds
     df_log["probability"] = probas
-    df_log["risk_level"] = ["High" if p >= 0.75 else "Medium" if p >= 0.5 else "Low" for p in probas]
+    df_log["risk_level"] = [
+        "High" if p >= 0.75 else "Medium" if p >= 0.5 else "Low"
+        for p in 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)
+    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")
+    # ---- Dashboard JSON ----
     DASHBOARD_JSON = "reports/evidently/drift_report.json"
 
-    # After computing drift_for_chart
     dashboard_payload = {
         "n_rows": len(results),
         "results": results,
-        "drift": drift_for_chart
+        "drift": drift_for_chart,
     }
 
-    # Write JSON for dashboard frontend
     os.makedirs(os.path.dirname(DASHBOARD_JSON), exist_ok=True)
-    # atomic write to avoid read/write collision
-    import tempfile
     tmp_path = DASHBOARD_JSON + ".tmp"
     with open(tmp_path, "w") as f:
         json.dump(dashboard_payload, f, indent=2)
     os.replace(tmp_path, DASHBOARD_JSON)
 
-    
-    return JSONResponse({"n_rows": len(results), "results": results, "drift": drift_for_chart})
+    return JSONResponse({
+        "n_rows": len(results),
+        "results": results,
+        "drift": drift_for_chart,
+    })
 
 
 @router.get("/health")
@@ -106,17 +134,6 @@ def run_drift():
     return {"status": "drift_check_completed", "report_path": report_path}
 
 
-@router.get("/monitoring/run")
-def monitoring_run(background_tasks: BackgroundTasks, model_version: str = "v1"):
-    current_data = pd.read_csv("data/processed/current_data.csv")
-    reference_data = pd.read_csv("data/processed/credit_default_clean.csv")
-
-    background_tasks.add_task(run_drift_check, current_data[predictor.features], reference_data[predictor.features], model_version)
-    background_tasks.add_task(run_governance_checks, current_data, model_version=model_version)
-
-    return {"status": "monitoring triggered", "model_version": model_version}
-
-
 @router.get("/dashboard")
 def dashboard(request: Request):
     return templates.TemplateResponse("dashboard.html", {"request": request})

app/api/traffic_daemon.py

@@ -0,0 +1,52 @@
+# app/api/traffic_daemon.py
+import asyncio
+import pandas as pd
+import random
+import requests
+import os
+import time
+
+API_URL = "http://localhost:8000/predict"
+SOURCE_DATA = "data/processed/current_data.csv"
+
+MIN_SLEEP = 2
+MAX_SLEEP = 8
+MIN_BATCH = 1
+MAX_BATCH = 5
+STARTUP_DELAY = 10  # seconds – allow FastAPI to fully start
+
+
+async def traffic_loop():
+    """
+    Continuously generate inference traffic against /predict.
+    """
+    await asyncio.sleep(STARTUP_DELAY)
+
+    if not os.path.exists(SOURCE_DATA):
+        print("Traffic daemon: source data not found, disabled.")
+        return
+
+    df = pd.read_csv(SOURCE_DATA)
+
+    print("Traffic daemon started.")
+
+    while True:
+        try:
+            batch_size = random.randint(MIN_BATCH, MAX_BATCH)
+            sample = df.sample(batch_size)
+
+            csv_bytes = sample.to_csv(index=False).encode("utf-8")
+
+            response = requests.post(
+                API_URL,
+                files={"file": ("sample.csv", csv_bytes, "text/csv")},
+                timeout=10,
+            )
+
+            if response.status_code != 200:
+                print("Traffic daemon warning:", response.status_code)
+
+        except Exception as e:
+            print("Traffic daemon error:", e)
+
+        await asyncio.sleep(random.uniform(MIN_SLEEP, MAX_SLEEP))

app/main.py

@@ -1,4 +1,4 @@
-# app/main.py (no other changes)
+# app/main.py
 from fastapi import FastAPI
 from fastapi.staticfiles import StaticFiles
 import asyncio
@@ -8,19 +8,35 @@ from app.api.routes import router
 from app.api.dashboard_data import router as dashboard_data_router
 from app.core.logging import init_db
 from app.api.background_drift import drift_loop
+from app.api.traffic_daemon import traffic_loop
+
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
+    # ---- Startup ----
     init_db()
-    task = asyncio.create_task(drift_loop(interval_seconds=10))
+
+    # Start drift detection loop
+    drift_task = asyncio.create_task(drift_loop(interval_seconds=10))
+
+    # Start traffic daemon (delayed internally, HF-safe)
+    traffic_task = asyncio.create_task(traffic_loop())
+
     yield
-    task.cancel()
-    try:
-        await task
-    except asyncio.CancelledError:
-        pass
 
-app = FastAPI(title="ML Inference Service", lifespan=lifespan)
+    # ---- Shutdown ----
+    for task in (drift_task, traffic_task):
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+
+app = FastAPI(
+    title="ML Inference Service",
+    lifespan=lifespan,
+)
 
 app.mount("/static", StaticFiles(directory="app/static"), name="static")
 app.mount("/reports", StaticFiles(directory="reports"), name="reports")

scripts/simulate_inference.py

@@ -3,23 +3,66 @@ import pandas as pd
 import requests
 import random
 import time
+import sys
 
-df = pd.read_csv("data/processed/current_data.csv")
+API_URL = "http://localhost:8000/predict"
+SOURCE_DATA = "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")
+# Traffic behavior (tune freely)
+MIN_SLEEP = 2      # seconds
+MAX_SLEEP = 8      # seconds
+MIN_BATCH = 1
+MAX_BATCH = 5
 
-# POST to FastAPI predict endpoint
-response = requests.post(
-    "http://localhost:8000/predict",
-    files={"file": ("sample.csv", csv_bytes, "text/csv")}
-)
+print("Starting inference traffic daemon...")
+print(f"Target API: {API_URL}")
+print(f"Source data: {SOURCE_DATA}")
+print("Press Ctrl+C to stop.\n")
 
-print("Status:", response.status_code)
+# Load once (realistic: upstream feature store snapshot)
 try:
-    print("Response:", response.json())
-except Exception:
-    print("Server returned non-JSON response:")
-    print(response.text)
+    df = pd.read_csv(SOURCE_DATA)
+except Exception as e:
+    print("Failed to load source data:", e)
+    sys.exit(1)
 
+required_cols = set(df.columns)
+
+while True:
+    try:
+        # ---- Random batch size ----
+        batch_size = random.randint(MIN_BATCH, MAX_BATCH)
+        sample = df.sample(batch_size)
+
+        # ---- Serialize to CSV ----
+        csv_bytes = sample.to_csv(index=False).encode("utf-8")
+
+        # ---- Send request ----
+        response = requests.post(
+            API_URL,
+            files={"file": ("sample.csv", csv_bytes, "text/csv")},
+            timeout=10,
+        )
+
+        if response.status_code == 200:
+            payload = response.json()
+            print(
+                f"[OK] rows={payload['n_rows']} "
+                f"predictions_logged=True"
+            )
+        else:
+            print(
+                f"[WARN] HTTP {response.status_code} "
+                f"{response.text}"
+            )
+
+    except KeyboardInterrupt:
+        print("\nTraffic daemon stopped by user.")
+        break
+
+    except Exception as e:
+        print("[ERROR] Inference request failed:", e)
+
+    # ---- Sleep (non-uniform traffic) ----
+    sleep_time = random.uniform(MIN_SLEEP, MAX_SLEEP)
+    time.sleep(sleep_time)