Spaces:
Runtime error
Runtime error
File size: 2,093 Bytes
af879c2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 | """MuleGuard scoring API.
Run: .venv/bin/uvicorn src.api.main:app --port 8000
"""
from __future__ import annotations
from datetime import datetime, timezone
from typing import Any
import pandas as pd
from fastapi import FastAPI
from pydantic import BaseModel, Field
from src import config
from src.models.scoring import explain_row, load_artifacts, narrative, score_frame
app = FastAPI(title="MuleGuard — Mule Account Risk Scoring", version="1.0")
class ScoreRequest(BaseModel):
account_id: str = Field(..., description="Account identifier")
features: dict[str, Any] = Field(..., description="Raw feature map (F1..F3923); missing allowed")
class ScoreResponse(BaseModel):
account_id: str
risk_score: float
risk_tier: str
decision: str
probability: float
reason_codes: list[dict]
narrative: str
timestamp: str
@app.on_event("startup")
def _warmup() -> None:
load_artifacts() # load model + explainers once
@app.get("/health")
def health() -> dict:
art = load_artifacts()
return {"status": "ok", "n_features": len(art.features), "threshold": art.threshold}
@app.get("/model-info")
def model_info() -> dict:
art = load_artifacts()
m = art.metadata
return {
"model": m["model"],
"n_features": m["n_features"],
"leakage_excluded": m["leakage_excluded"],
"cv_metrics": m["cv_metrics"],
"test_metrics": m["test_metrics"],
"threshold": art.threshold,
}
@app.post("/score", response_model=ScoreResponse)
def score(req: ScoreRequest) -> ScoreResponse:
row = pd.DataFrame([req.features])
scored = score_frame(row).iloc[0]
reasons = explain_row(row)
return ScoreResponse(
account_id=req.account_id,
risk_score=float(scored["risk_score"]),
risk_tier=str(scored["risk_tier"]),
decision=str(scored["decision"]),
probability=float(scored["probability"]),
reason_codes=reasons,
narrative=narrative(float(scored["risk_score"]), reasons),
timestamp=datetime.now(timezone.utc).isoformat(),
)
|