working on the daemon

README.md

@@ -26,61 +26,64 @@ uvicorn app.main:app --reload
 # Repo Structure
 
 ml-inference-drift-service/
-│
-├── app/
-│   ├── main.py                  # FastAPI entrypoint
-│   ├── api/
-│   │   ├── routes.py             # /predict, /health, /dashboard
-│   │   └── schemas.py            # Pydantic input/output schemas
-│   │
-│   ├── core/
-│   │   ├── config.py             # env vars, paths, thresholds
-│   │   ├── logging.py            # SQLite + file logging
-│   │   └── model_registry.py     # model loading/versioning
-│   │
-│   ├── inference/
-│   │   ├── predictor.py          # model.predict wrapper
-│   │   └── preprocessing.py      # feature handling
-│   │
-│   ├── monitoring/
-│   │   ├── drift.py              # Evidently logic
-│   │   ├── metrics.py            # feature stats extraction
-│   │   └── alerts.py             # threshold evaluation
-│   │
-│   ├── db/
-│   │   ├── session.py            # SQLite connection
-│   │   └── models.py             # ORM-style tables (optional)
-│   │
-│   ├── templates/
-│   │   └── dashboard.html        # Evidently embed + metrics
-│   │
-│   └── static/
-│       └── styles.css
-│
-├── models/
-│   ├── v1/
-│   │   ├── model.pkl
-│   │   └── reference_data.csv
-│   └── v2/
-│       └── ...
-│
-├── scripts/
-│   ├── train.py                  # offline training
-│   ├── evaluate.py               # offline evaluation
-│   └── run_drift_check.py        # batch drift job
-│
-├── reports/
-│   └── evidently/
-│       └── drift_report.html
-│
-├── tests/
-│   ├── test_api.py
-│   ├── test_drift.py
-│   └── test_schemas.py
-│
-├── Dockerfile
-├── Dockerfile.hf                 # HF Spaces–compatible
-├── requirements.txt
-├── requirements-dev.txt
-├── README.md
-└── .env.example
+Dockerfile
+LICENSE
+README.md
+requirements-dev.txt
+requirements.txt
+app/
+    main.py
+    api/
+        background_drift.py
+        dashboard_data.py
+        routes.py
+        schemas.py
+        traffic_daemon.py
+    core/
+        config.py
+        logging.py
+        templates.py
+    inference/
+        predictor.py
+    monitoring/
+        data_loader.py
+        drift.py
+        governance.py
+    static/
+        styles.css
+    templates/
+        dashboard.html
+    utils/
+        alerts.py
+data/
+    processed/
+        credit_default_clean.csv
+        current_data.csv
+    production/
+        predictions_log.csv
+    raw/
+        credit_default.csv
+database/
+logs/
+models/
+    v1/
+        features.json
+        reference_data.csv
+    v2/
+reports/
+    evidently/
+        drift_report.html
+        drift_report.json
+scripts/
+    prepare_data.py
+    simulate_inference.py
+    train.py
+tests/
+    conftest.py
+    integration/
+        test_api.py
+        test_governance.json
+        test_governance.py
+        test_run_drift.py
+    unit/
+        test_schemas.py

app/api/background_drift.py

@@ -13,7 +13,7 @@ REFERENCE_PATH = "models/v1/reference_data.csv"
 PROD_LOG_PATH = "data/production/predictions_log.csv"
 DASHBOARD_JSON = "reports/evidently/drift_report.json"
 
-MAX_ROWS = 5000  # rolling window
+MAX_ROWS = 5000
 os.makedirs(os.path.dirname(DASHBOARD_JSON), exist_ok=True)
 
 async def drift_loop(interval_seconds: int = 10):
@@ -25,15 +25,12 @@ async def drift_loop(interval_seconds: int = 10):
 
             prod_df = pd.read_csv(PROD_LOG_PATH)
 
-            # Retention window
             if len(prod_df) > MAX_ROWS:
                 prod_df = prod_df.tail(MAX_ROWS)
                 prod_df.to_csv(PROD_LOG_PATH, index=False)
 
-            # Keep only rows with all required features
-            missing_features = set(predictor.features) - set(prod_df.columns)
-            if missing_features:
-                print(f"Skipping drift check, missing features: {missing_features}")
+            missing = set(predictor.features) - set(prod_df.columns)
+            if missing:
                 await asyncio.sleep(interval_seconds)
                 continue
 
@@ -44,26 +41,34 @@ async def drift_loop(interval_seconds: int = 10):
 
             reference_df = pd.read_csv(REFERENCE_PATH)
 
-            # ---- Run drift on features only ----
             _, drift_dict = run_drift_check(
                 prod_df[predictor.features],
                 reference_df[predictor.features],
                 model_version="v1"
             )
 
+            # ---- RECENT PREDICTIONS FIX ----
+            recent_results = []
+            if "prediction" in prod_df.columns:
+                recent_results = (
+                    prod_df[["prediction"]]
+                    .tail(10)
+                    .to_dict(orient="records")
+                )
+
             dashboard_payload = {
                 "n_rows": len(prod_df),
-                "results": [],
+                "results": recent_results,
                 "drift": [
                     {"column": col, "score": float(score)}
                     for col, score in drift_dict.items()
                 ],
             }
 
-            tmp_path = DASHBOARD_JSON + ".tmp"
-            with open(tmp_path, "w") as f:
+            tmp = DASHBOARD_JSON + ".tmp"
+            with open(tmp, "w") as f:
                 json.dump(dashboard_payload, f, indent=2)
-            os.replace(tmp_path, DASHBOARD_JSON)
+            os.replace(tmp, DASHBOARD_JSON)
 
         except Exception as e:
             print("Drift loop error:", e)

app/api/traffic_daemon.py

@@ -2,8 +2,8 @@
 import asyncio
 import pandas as pd
 import random
-import requests
 import os
+import httpx
 
 API_URL = "http://localhost:8000/predict"
 SOURCE_DATA = "data/processed/current_data.csv"
@@ -12,7 +12,7 @@ MIN_SLEEP = 2
 MAX_SLEEP = 8
 MIN_BATCH = 1
 MAX_BATCH = 5
-STARTUP_DELAY = 10  # allow server startup
+STARTUP_DELAY = 7
 
 async def traffic_loop():
     await asyncio.sleep(STARTUP_DELAY)
@@ -24,23 +24,22 @@ async def traffic_loop():
     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")
+    async with httpx.AsyncClient(timeout=60.0) as client:
+        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")
 
-            # ---- Increased timeout to avoid ReadTimeout ----
-            response = requests.post(
-                API_URL,
-                files={"file": ("sample.csv", csv_bytes, "text/csv")},
-                timeout=60,
-            )
+                resp = await client.post(
+                    API_URL,
+                    files={"file": ("sample.csv", csv_bytes, "text/csv")}
+                )
 
-            if response.status_code != 200:
-                print("Traffic daemon warning:", response.status_code)
+                if resp.status_code != 200:
+                    print("Traffic daemon warning:", resp.status_code)
 
-        except Exception as e:
-            print("Traffic daemon error:", e)
+            except Exception as e:
+                print("Traffic daemon error:", e)
 
-        await asyncio.sleep(random.uniform(MIN_SLEEP, MAX_SLEEP))
+            await asyncio.sleep(random.uniform(MIN_SLEEP, MAX_SLEEP))

app/monitoring/governance.py

@@ -84,8 +84,11 @@ class Governance:
         # Log and send alerts
         for alert in alerts:
             self.log_alert(alert, model_version)
+        try:
             send_email_alert(alert)
             send_slack_alert(alert)
+        except Exception:
+            pass
 
         return alerts
 

data/production/predictions_log.csv

@@ -1,9 +0,0 @@
-credit_limit,age,pay_delay_sep,pay_delay_aug,bill_amt_sep,bill_amt_aug,pay_amt_sep,pay_amt_aug,target,model_prediction,model_probability,model_risk_level,model_version,timestamp
-70000.0,48,0,0,20744.0,22093.0,2000.0,2000.0,0,0,0.2563046708563498,Low,v1,2026-01-14 19:33:08.201187+00:00
-390000.0,42,0,0,310075.0,184647.0,10021.0,4000.0,0,0,0.07238720996682343,Low,v1,2026-01-14 19:33:08.201187+00:00
-230000.0,50,-2,-2,2789.0,2942.0,2942.0,2520.0,0,0,0.06061841289885345,Low,v1,2026-01-14 19:33:08.201187+00:00
-40000.0,47,2,2,52358.0,54892.0,4000.0,0.0,1,1,0.608097803732095,Medium,v1,2026-01-14 19:35:59.632246+00:00
-140000.0,41,0,0,130138.0,132726.0,4756.0,4912.0,1,0,0.19300589506044843,Low,v1,2026-01-14 19:35:59.632246+00:00
-50000.0,26,-1,-1,5052.0,0.0,0.0,0.0,0,0,0.12125611010883464,Low,v1,2026-01-14 19:37:40.778834+00:00
-400000.0,42,-1,-1,44198.0,10132.0,10132.0,16932.0,0,0,0.06632404342712281,Low,v1,2026-01-14 19:39:17.748870+00:00
-280000.0,45,-2,-2,26573.0,17597.0,18388.0,22302.0,0,0,0.036231471756518835,Low,v1,2026-01-14 19:39:17.748870+00:00

reports/evidently/drift_report.json

@@ -1,5 +1,5 @@
 {
-  "n_rows": 8,
+  "n_rows": 20,
   "results": [],
   "drift": [
     {
@@ -8,35 +8,35 @@
     },
     {
       "column": "age",
-      "score": 0.8679104867707123
+      "score": 0.47511682619708956
     },
     {
       "column": "bill_amt_aug",
-      "score": 0.23085843753803348
+      "score": 0.18806444456731744
     },
     {
       "column": "bill_amt_sep",
-      "score": 0.37457217443848057
+      "score": 0.24552730613716447
     },
     {
       "column": "credit_limit",
-      "score": 0.36873847560130574
+      "score": 0.36053044124390277
     },
     {
       "column": "pay_amt_aug",
-      "score": 0.20876449446182804
+      "score": 0.17193500059085817
     },
     {
       "column": "pay_amt_sep",
-      "score": 0.27529913358402724
+      "score": 0.6302425864834968
     },
     {
       "column": "pay_delay_aug",
-      "score": 0.32956762578626103
+      "score": 0.2880354597906329
     },
     {
       "column": "pay_delay_sep",
-      "score": 0.4885655407858251
+      "score": 0.37250685167671616
     }
   ]
 }