"""Argus — production ML observability dashboard.
Pages:
Overview - rolling error metrics and system status
Drift Analysis - PSI, KS-test, and heatmap per feature
Feature Insights - importance comparison and drift ranking
Retraining Log - history of retraining decisions
Live Demo - interactive API playground
Run with:
python -m streamlit run dashboard/app.py
"""
from __future__ import annotations
import json
import os
import random
import time
from pathlib import Path
import numpy as np
import pandas as pd
import plotly.graph_objects as go
import requests
import streamlit as st
import streamlit.components.v1 as _components
st.set_page_config(
page_title="Argus",
page_icon="\U0001f695",
layout="wide",
initial_sidebar_state="auto",
)
# HF Spaces auto-refresh via JavaScript (with cache-busting query param)
# Adds timestamp to bypass HF Spaces CDN caching
st.markdown("""
""", unsafe_allow_html=True)
st.markdown("""
""", unsafe_allow_html=True)
API_URL = os.environ.get("API_URL", "http://localhost:8000")
PROJECT_ROOT = Path(__file__).resolve().parent.parent
LOG_PATHS = {
"performance": PROJECT_ROOT / "data" / "logs" / "performance.jsonl",
"drift": PROJECT_ROOT / "data" / "logs" / "drift_reports.jsonl",
"retrain": PROJECT_ROOT / "data" / "logs" / "retraining.jsonl",
"feedback": PROJECT_ROOT / "data" / "logs" / "feedback.jsonl",
"predictions": PROJECT_ROOT / "data" / "logs" / "predictions.jsonl",
"importances": PROJECT_ROOT / "data" / "logs" / "feature_importances.json",
}
ACCENT = "#4f8ef7"
ACCENT2 = "#a78bfa"
OK = "#22d3a0"
WARN = "#fbbf24"
ERR = "#f87171"
PURPLE = "#7c3aed"
TEAL = "#06b6d4"
SURFACE = "#151f32"
BORDER = "#2d3f5a"
FEATURE_LABELS = {
"trip_distance": "Trip Distance",
"passenger_count": "Passenger Count",
"pickup_hour": "Pickup Hour",
"pickup_dow": "Day of Week",
"pickup_month": "Month",
"pickup_is_weekend": "Is Weekend",
"rate_code_id": "Rate Code",
"payment_type": "Payment Type",
"pu_location_zone": "Pickup Zone",
"do_location_zone": "Dropoff Zone",
"vendor_id": "Vendor",
}
PLOTLY_BASE = dict(
template="plotly_dark",
paper_bgcolor="rgba(0,0,0,0)",
plot_bgcolor="rgba(0,0,0,0)",
margin=dict(l=8, r=8, t=12, b=8),
font=dict(color="#e2eaf5", size=11),
)
def load_jsonl(path: Path, limit: int = 2000) -> pd.DataFrame:
"""Read the last `limit` lines of a JSONL log file into a DataFrame.
Note: Cache disabled to ensure synthetic data always shows on HF Spaces.
"""
if not path.exists():
return pd.DataFrame()
lines = path.read_text(encoding="utf-8").splitlines()[-limit:]
records = []
for line in lines:
try:
records.append(json.loads(line))
except json.JSONDecodeError:
pass
return pd.DataFrame(records) if records else pd.DataFrame()
@st.cache_data(ttl=15)
def load_importances(path: Path) -> pd.DataFrame:
"""Load feature importance JSON into a DataFrame."""
if not path.exists():
return pd.DataFrame()
try:
return pd.read_json(path)
except ValueError:
return pd.DataFrame()
@st.cache_data(ttl=5)
def api_health() -> dict:
"""Fetch the /health endpoint; return empty dict if unreachable."""
try:
http_resp = requests.get(f"{API_URL}/health", timeout=3)
return http_resp.json()
except requests.RequestException:
return {}
def _get_synthetic_data() -> pd.DataFrame:
"""Generate synthetic data: grows to 500 points, then resets.
Creates a continuous cycle for demos.
"""
baseline = 2.5
current_time = int(time.time())
# Use a 500-minute cycle (one point per minute, auto-reset after 500)
cycle_time = current_time // 60 # Minutes since epoch
local_minute = cycle_time % 500 # Position in 500-point cycle
# Generate points from 0 to current position in cycle
rmse_list = []
mae_list = []
r2_list = []
for local_idx in range(0, local_minute + 1):
random.seed(cycle_time - local_minute + local_idx)
np.random.seed(cycle_time - local_minute + local_idx)
# Linear upward drift (0 to 500)
drift = 1.0 + (local_idx / 250.0) * 0.6 # Increases 60% over 500 points
# Occasional spike (5% chance)
spike = 1.0 if random.random() > 0.05 else random.uniform(1.3, 1.6)
# Minimal noise
noise = random.gauss(1.0, 0.02)
rmse = baseline * drift * spike * noise
rmse = np.clip(rmse, 2.0, 8.0)
mae = rmse * 0.75
r2 = 0.80 - (local_idx / 500.0) * 0.5 # Decreases from 0.80 to 0.30
rmse_list.append(rmse)
mae_list.append(mae)
r2_list.append(r2)
return pd.DataFrame({
"rmse": rmse_list,
"mae": mae_list,
"r2": r2_list,
})
def _get_synthetic_metrics() -> dict:
"""Get current synthetic metrics as dict (latest point from synthetic data)."""
df = _get_synthetic_data()
if df.empty or len(df) == 0:
return {"rmse": 2.5, "mae": 1.75, "r2": 0.75, "baseline_rmse": 2.5, "n_samples": 0}
baseline = 2.5
return {
"rmse": float(df["rmse"].iloc[-1]),
"mae": float(df["mae"].iloc[-1]),
"r2": float(df["r2"].iloc[-1]),
"baseline_rmse": baseline,
"n_samples": len(df),
}
@st.cache_data(ttl=10)
def _api_metrics_cached() -> dict:
"""Cached API call (only for when API is online)."""
try:
http_resp = requests.get(f"{API_URL}/monitor/metrics", timeout=3)
return http_resp.json()
except requests.RequestException:
return None
def api_metrics() -> dict:
"""Get metrics: try API first, fall back to synthetic data if offline."""
result = _api_metrics_cached()
if result is not None:
return result
# API offline: use synthetic data (updates every rerun, no cache)
return _get_synthetic_metrics()
def _pct_color(pct: float | None) -> str:
if pct is None:
return ACCENT
return ERR if pct > 15 else (WARN if pct > 5 else OK)
def _plotly_layout(**kwargs) -> dict:
base = dict(PLOTLY_BASE)
base.update(kwargs)
return base
with st.sidebar:
st.markdown(
'\U0001f695 Argus
',
unsafe_allow_html=True,
)
st.markdown(
''
'NYC Taxi \u00b7 ML Observability Platform
',
unsafe_allow_html=True,
)
st.markdown(
''
'End-to-end MLOps platform: serves predictions via REST API, '
'monitors live data for distribution shift, and triggers '
'automated retraining when quality degrades.'
'
',
unsafe_allow_html=True,
)
st.markdown(
''
+ "".join(
f'{t}'
for t in ["FastAPI", "scikit-learn", "MLflow", "Streamlit", "Plotly", "Docker"]
)
+ "
",
unsafe_allow_html=True,
)
st.markdown("---")
health = api_health()
api_online = health.get("status") == "ok"
if api_online:
st.success("API Online")
else:
st.warning("API Offline — showing logged data")
if api_online:
st.markdown(
f''
f'
SYSTEM INFO
'
f'
'
f'Model'
f''
f'{health.get("model_version", "N/A")}
'
f'
'
f'Uptime'
f''
f'{health.get("uptime_seconds", 0):.0f}s
'
f'
'
f'Pending'
f''
f'{health.get("pending_feedback_count", 0)}
'
f'
'
f'Matched'
f''
f'{health.get("matched_feedback_count", 0)}
'
f'
NAVIGATION
',
unsafe_allow_html=True,
)
page = st.radio(
"Navigation",
["Overview", "Drift Analysis", "Feature Insights", "Retraining Log", "Live Demo"],
label_visibility="collapsed",
)
st.markdown("---")
auto_refresh = st.checkbox("Auto-refresh (15s)", value=True)
if st.button("Refresh Now", use_container_width=True):
st.cache_data.clear()
st.rerun()
st.markdown("---")
st.markdown(
f'API: '
f'{API_URL}
',
unsafe_allow_html=True,
)
# ── Overview ──────────────────────────────────────────────────────────────────
if page == "Overview":
st.markdown('\U0001f695 Argus
', unsafe_allow_html=True)
st.markdown(
'Production ML observability '
' · Drift detection '
' · Automated retraining
',
unsafe_allow_html=True,
)
st.markdown(
''
'
'
'
THE PROBLEM
'
'
'
'ML models degrade in production when real-world data distributions shift '
'away from training data. Manual monitoring does not scale.'
'
'
'
'
'
'
'
WHAT ARGUS DOES
'
'
'
'Serves predictions via FastAPI, continuously measures feature drift '
'and accuracy, then automatically retrains and promotes a new model '
'when degradation is confirmed.'
'
'
'
'
'
'
'
DOMAIN
'
'
'
'NYC taxi trip duration prediction. GradientBoostingRegressor trained '
'on TLC data. Simulated temporal drift shifts pickup patterns over time.'
'
'
'
'
'
'
f'{label}'
f'{value}'
f'
',
unsafe_allow_html=True,
)
n_triggered = int(retrain_df["should_retrain"].sum()) \
if not retrain_df.empty and "should_retrain" in retrain_df.columns \
else len(retrain_df)
st.markdown(
f''
f'
Retraining Events
'
f'
{n_triggered}
'
f'
triggered / '
f'{len(retrain_df)} evaluations
'
f'
Recent RMSE trend
',
unsafe_allow_html=True,
)
st.plotly_chart(spark_fig, width='stretch')
# ── Drift Analysis ────────────────────────────────────────────────────────────
elif page == "Drift Analysis":
st.markdown('Drift Analysis
', unsafe_allow_html=True)
st.markdown(
'Feature-level distribution monitoring '
'— PSI and Kolmogorov-Smirnov
',
unsafe_allow_html=True,
)
st.markdown(
''
'
'
'
PSI — POPULATION STABILITY INDEX
'
'
'
'Compares the distribution of each feature between training data and live '
'production data. '
'PSI < 0.10 = stable, '
'0.10–0.20 = moderate shift, '
'PSI > 0.20 = significant drift.'
'
'
'
'
'
'
'
KS TEST — KOLMOGOROV-SMIRNOV
'
'
'
'A non-parametric statistical test that checks whether two samples come '
'from the same distribution. A p-value below 0.05 rejects the null hypothesis '
'— the feature distribution has changed significantly.'
'
'
'
'
'
'
f'
Drift Detected
'
f'
'
f'Features with significant drift: '
f'{", ".join(FEATURE_LABELS.get(f, f) for f in drifted)}'
f'
Feature Insights
', unsafe_allow_html=True)
st.markdown(
'Drift ranking, root-cause scores, '
'and model feature importance
',
unsafe_allow_html=True,
)
st.markdown(
''
'
ROOT-CAUSE ANALYSIS
'
'
'
'Not all drifted features are equally dangerous. A feature that has shifted '
'but contributes little to the model\'s predictions is low risk. '
'The drift radar and '
'ranking table cross-reference '
'PSI drift scores with model feature importance, so the highest-risk '
'root causes are immediately visible.'
'
'
'
Retraining Log
', unsafe_allow_html=True)
st.markdown(
'Automated retraining decisions — '
'drift detection, performance gating, and champion-challenger promotion
',
unsafe_allow_html=True,
)
st.markdown("---")
# ── Design explanation ────────────────────────────────────────────────────
st.markdown(
''
'
RETRAINING POLICY
'
'
'
'Retraining requires two independent signals '
'to fire simultaneously — this is intentional conservative design to avoid '
'unnecessary churn in production.'
'
'
'
'
'
'
'
'
'GATE 1 — FEATURE DRIFT
'
'
'
'PSI > 0.20 or KS p < 0.05 on any feature. '
'Detects distribution shift in the incoming data pipeline.
'
'
'
'
'
'
'
'GATE 2 — PERFORMANCE DEGRADATION
'
'
'
'Rolling RMSE exceeds baseline by more than 15%. '
'Confirms the model is actually harmed, not just seeing new data.
'
'
'
'
'
'
'
'GATE 3 — SAMPLE BUDGET
'
'
'
'At least 1,000 new labeled samples required. '
'Prevents retraining on insufficient data that would destabilize the model.
'
'
'
'
'
'
'
'Current status: feature drift is active but performance has not degraded '
'(RMSE is below baseline), so retraining is correctly suppressed. '
'The model is handling the shifted distribution without accuracy loss.
'
'
" +
"
".join(str(r) for r in reasons) +
"
"
) if reasons else ""
blocking_html = (
""
"Blocked: " +
" • ".join(str(b) for b in blocking) +
"
"
) if blocking else ""
gate_badges = (
f''
f''
f'FEATURE DRIFT: {gate1_txt}'
f''
f'PERFORMANCE: {gate2_txt}'
f''
f'SAMPLES: {gate3_txt}'
f'
'
)
st.markdown(
f''
f'
'
f'{label}'
f'{ts}'
f'
'
f'{reasons_html}'
f'{gate_badges}'
f'{blocking_html}'
f'
Live Demo
', unsafe_allow_html=True)
st.markdown(
'Interact with the prediction API in real time
',
unsafe_allow_html=True,
)
st.markdown(
''
'
'
'
1
'
'
PREDICT
'
'
'
'Set trip parameters and call /predict. '
'The FastAPI service runs the GradientBoosting model and returns a '
'trip duration estimate with a unique request ID.'
'
'
'
'
'
'
'
2
'
'
SUBMIT GROUND TRUTH
'
'
'
'Paste the request ID and enter the actual trip duration. '
'This simulates the delayed feedback loop — real ground truth '
'arrives minutes or hours after prediction.'
'
'
'
'
'
'
'
3
'
'
MONITOR
'
'
'
'Each submitted feedback updates the rolling accuracy window. '
'Run a drift check to see if the live data distribution '
'has shifted from the training baseline.'
'
'
'
'
'
'
f'Is Weekend: '
f'{"Yes" if is_weekend else "No"}
',
unsafe_allow_html=True,
)
if st.button("Predict Duration", use_container_width=True, type="primary"):
payload = {
"passenger_count": passenger_count,
"trip_distance": trip_distance,
"pickup_hour": pickup_hour,
"pickup_dow": pickup_dow,
"pickup_month": pickup_month,
"pickup_is_weekend": is_weekend,
"rate_code_id": 1,
"payment_type": 1,
"pu_location_zone": 10,
"do_location_zone": 25,
"vendor_id": 1,
}
try:
resp = requests.post(f"{API_URL}/predict", json=payload, timeout=5)
result = resp.json()
if resp.status_code == 200:
dur = result.get("predicted_duration_min", "?")
st.success(f"Predicted duration: **{dur} minutes**")
st.json(result)
st.session_state["last_request_id"] = result.get("request_id", "")
else:
st.error(f"API error {resp.status_code}: {result}")
except requests.RequestException as exc:
st.error(f"Could not reach API: {exc}")
st.markdown("---")
st.markdown('',
unsafe_allow_html=True)
if st.button("Run Drift Check", use_container_width=True):
try:
resp = requests.get(f"{API_URL}/monitor/drift", timeout=10)
result = resp.json()
if result.get("drift_detected"):
st.error("Drift Detected — check Drift Analysis page for details.")
else:
st.success("No significant drift in current window.")
st.json(result)
except requests.RequestException as exc:
st.error(f"Could not reach API: {exc}")
with right:
st.markdown('',
unsafe_allow_html=True)
st.markdown(
''
'Simulate delayed feedback — ground truth arriving after the trip ends.
',
unsafe_allow_html=True,
)
req_id = st.text_input(
"Request ID",
value=st.session_state.get("last_request_id", ""),
placeholder="paste request_id from prediction response",
)
actual = st.number_input(
"Actual Duration (min)", min_value=1.0, max_value=120.0,
value=15.0, step=0.5,
)
if st.button("Submit Feedback", use_container_width=True):
if not req_id:
st.warning("Paste a request_id first.")
else:
try:
resp = requests.post(
f"{API_URL}/predict/feedback",
json={"request_id": req_id, "actual_duration_min": actual},
timeout=5,
)
result = resp.json()
if result.get("matched"):
st.success(result.get("message", "Feedback accepted."))
else:
st.warning(result.get("message", "Request ID not found."))
except requests.RequestException as exc:
st.error(f"Could not reach API: {exc}")
st.markdown("---")
st.markdown('',
unsafe_allow_html=True)
st.code(json.dumps({
"drift_detected": True,
"root_cause": ["trip_distance", "pickup_hour"],
"performance_drop": "18.3%",
"action": "retraining_triggered",
"drifted_features": ["trip_distance", "pickup_hour"],
"rca_details": [
{
"feature": "trip_distance",
"psi": 0.312,
"importance": 0.421,
"rca_score": 0.444,
},
{
"feature": "pickup_hour",
"psi": 0.241,
"importance": 0.187,
"rca_score": 0.286,
},
],
}, indent=2), language="json")
st.markdown("---")
st.markdown('',
unsafe_allow_html=True)
st.code(f"# Health check\ncurl {API_URL}/health\n\n"
f"# Performance metrics\ncurl {API_URL}/monitor/metrics\n\n"
f"# Drift report\ncurl {API_URL}/monitor/drift\n\n"
f"# Manual retrain trigger\ncurl -X POST {API_URL}/monitor/retrain",
language="bash")
# ── Scroll-to-top on page change (runs after all content is rendered) ─────────
# A monotonically increasing counter combined with the page name guarantees a
# unique HTML string on every navigation — even when revisiting a page —
# so Streamlit always creates a fresh iframe and re-executes the scroll script.
if st.session_state.get("_scroll_page") != page:
_nav_count = st.session_state.get("_scroll_count", 0) + 1
st.session_state["_scroll_page"] = page
st.session_state["_scroll_count"] = _nav_count
_components.html(
f"""""",
height=1,
scrolling=False,
)
# ── Auto-refresh (must run AFTER all page content is rendered) ────────────────
if auto_refresh:
time.sleep(15)
st.rerun()