udyamlinc-ml/serving/app.py

"""
The inference service Spring Boot calls.

One network boundary, and it is deliberately narrow: the request carries the
feature map and nothing else about the borrower. No name, no phone, no account
reference — those do not exist in the platform's schema, and this service must
never become the reason someone adds them.

The response mirrors ``ScoreOutcome`` field for field, so the Java adapter is a
mapping rather than a translation.
"""

from __future__ import annotations

import pickle
from pathlib import Path

import pandas as pd
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

from explain import Explainer, FAMILIES

MODEL_DIR = Path(__file__).resolve().parent.parent / "models"
MODEL_PATH = MODEL_DIR / "stress_model.pkl"

app = FastAPI(title="UdyamLinc ML", version="1.0.0")

_bundle: dict | None = None
_explainer: Explainer | None = None


def bundle() -> dict:
    global _bundle, _explainer
    if _bundle is None:
        with open(MODEL_PATH, "rb") as fh:
            _bundle = pickle.load(fh)
        _explainer = Explainer(_bundle["model"], _bundle["features"])
    return _bundle


# --------------------------------------------------------------------------- #


class ScoreRequest(BaseModel):
    enterpriseId: str
    sectorId: int | None = None
    districtId: int | None = None
    # Nulls are meaningful and must survive the wire: the engine has to be able to
    # see what it did not get, exactly as ScoringStrategy requires of a Java engine.
    features: dict[str, float | None] = Field(default_factory=dict)
    dpdAtT: float | None = None
    outstandingAtT: float | None = None
    emiAtT: float | None = None


class Driver(BaseModel):
    family: str
    weight: int
    featureCode: str | None
    value: float | None
    onPrior: bool


class ScoreResponse(BaseModel):
    enterpriseId: str
    score: int
    band: str
    confidence: int
    probability: float
    subScores: dict[str, int]
    drivers: list[Driver]
    familiesOnPrior: int
    engine: str = "MODEL"
    modelVersion: str


class BatchRequest(BaseModel):
    items: list[ScoreRequest]


def _frame(req: ScoreRequest, features: list[str]) -> pd.DataFrame:
    row: dict[str, float | None] = {}
    for name in features:
        if name.endswith("__missing"):
            base = name.removesuffix("__missing")
            row[name] = 1 if req.features.get(base) is None else 0
        elif name == "dpd_at_t":
            row[name] = req.dpdAtT
        elif name == "outstanding_at_t":
            row[name] = req.outstandingAtT
        elif name == "emi_at_t":
            row[name] = req.emiAtT
        elif name == "sector_id":
            row[name] = req.sectorId
        elif name == "district_id":
            row[name] = req.districtId
        else:
            row[name] = req.features.get(name)
    # Everything float64: a dict of mixed None and float builds an object column,
    # and LightGBM rejects object dtypes rather than coercing. None becomes NaN,
    # which is exactly how the model was trained to read a missing feature.
    return pd.DataFrame([row], columns=features).astype("float64")


def _score_one(req: ScoreRequest) -> ScoreResponse:
    b = bundle()
    frame = _frame(req, b["features"])
    probability = float(b["model"].predict_proba(frame)[:, 1][0])
    score = int(round(probability * 100))

    band = ("RED" if score >= b["red_threshold"]
            else "WATCH" if score >= b["watch_threshold"]
            else "STANDARD")

    sub_scores, drivers = _explainer.explain(frame)  # type: ignore[union-attr]
    on_prior = sum(1 for d in drivers if d["onPrior"])

    # Confidence falls with the share of the catalogue that came back null. A
    # prediction built on holes must report as one — the rule engine already does
    # this through familiesOnPrior, and the two must not disagree.
    supplied = sum(1 for v in req.features.values() if v is not None)
    coverage = supplied / max(1, len(req.features))
    confidence = int(round(max(20.0, min(95.0, 95 * coverage))))

    return ScoreResponse(
        enterpriseId=req.enterpriseId,
        score=score,
        band=band,
        confidence=confidence,
        probability=round(probability, 5),
        subScores=sub_scores,
        drivers=[Driver(**{k: d[k] for k in ("family", "weight", "featureCode", "value", "onPrior")})
                 for d in drivers],
        familiesOnPrior=on_prior,
        modelVersion=f"{b['algorithm']}-stress-v1",
    )


# --------------------------------------------------------------------------- #


@app.get("/health")
def health() -> dict:
    return {"status": "UP", "model": MODEL_PATH.exists()}


@app.get("/model")
def model_info() -> dict:
    b = bundle()
    return {
        "algorithm": b["algorithm"],
        "modelVersion": f"{b['algorithm']}-stress-v1",
        "features": b["features"],
        "families": FAMILIES,
        "redThreshold": b["red_threshold"],
        "watchThreshold": b["watch_threshold"],
        "trainedOn": "synthetic DPD history — metrics are provisional",
    }


@app.post("/score", response_model=ScoreResponse)
def score(req: ScoreRequest) -> ScoreResponse:
    return _score_one(req)


@app.post("/score/batch", response_model=list[ScoreResponse])
def score_batch(req: BatchRequest) -> list[ScoreResponse]:
    """One call for a nightly run over the whole book, rather than 263."""
    return [_score_one(item) for item in req.items]


# --------------------------------------------------------------------------- #
# Forecast — quantile heads                                                    #
# --------------------------------------------------------------------------- #

FORECAST_PATH = MODEL_DIR / "forecast_model.pkl"

_forecast: dict | None = None


def forecast_bundle() -> dict | None:
    """None when no forecast model has been trained — the caller falls back."""
    global _forecast
    if _forecast is None and FORECAST_PATH.exists():
        with open(FORECAST_PATH, "rb") as fh:
            _forecast = pickle.load(fh)
    return _forecast


class ForecastRequest(BaseModel):
    enterpriseId: str
    sectorId: int | None = None
    districtId: int | None = None
    features: dict[str, float | None] = Field(default_factory=dict)
    # Trailing ledger statistics, computed by the caller. The service stays
    # stateless: it holds no database credentials and cannot read a borrower's
    # ledger even if asked to. Java already reads the ledger for the
    # deterministic path, so this is the same query it was already making.
    trailMean: float | None = None
    trailSd: float | None = None
    trailMin: float | None = None
    trailInflow: float | None = None
    trailMonths: float | None = None
    emi: float | None = None
    horizons: int = 6


class ForecastPoint(BaseModel):
    horizon: int
    p10: float
    p50: float
    p90: float


class ForecastResponse(BaseModel):
    enterpriseId: str
    engine: str = "MODEL"
    modelVersion: str
    points: list[ForecastPoint]


@app.post("/forecast", response_model=ForecastResponse)
def forecast(req: ForecastRequest) -> ForecastResponse:
    b = forecast_bundle()
    if b is None:
        raise HTTPException(status_code=503, detail="No forecast model trained")

    # Three months of ledger is the floor the deterministic strategy uses too. An
    # enterprise with less gets no forecast rather than a fabricated one.
    if (req.trailMonths or 0) < 3:
        return ForecastResponse(
            enterpriseId=req.enterpriseId, modelVersion=b["version"], points=[]
        )

    rows = []
    for h in range(1, min(req.horizons, b["horizons"]) + 1):
        row: dict[str, float | None] = {}
        for name in b["features"]:
            if name.endswith("__missing"):
                row[name] = 1 if req.features.get(name.removesuffix("__missing")) is None else 0
            elif name == "horizon":
                row[name] = h
            elif name == "trail_mean":
                row[name] = req.trailMean
            elif name == "trail_sd":
                row[name] = req.trailSd
            elif name == "trail_min":
                row[name] = req.trailMin
            elif name == "trail_inflow":
                row[name] = req.trailInflow
            elif name == "trail_months":
                row[name] = req.trailMonths
            elif name == "emi":
                row[name] = req.emi
            elif name == "sector_id":
                row[name] = req.sectorId
            elif name == "district_id":
                row[name] = req.districtId
            else:
                row[name] = req.features.get(name)
        rows.append(row)

    frame = pd.DataFrame(rows, columns=b["features"]).astype("float64")

    preds = {q: b["models"][q].predict(frame) for q in ("p10", "p50", "p90")}

    points = []
    for i, h in enumerate(range(1, len(rows) + 1)):
        # Quantile regression does not guarantee ordering — sort per row. A
        # crossed band is worse than no band, and it is the kind of thing that
        # renders as a chart nobody can read rather than as an error.
        lo, mid, hi = sorted(float(preds[q][i]) for q in ("p10", "p50", "p90"))

        # Missing inputs widen the band and never lower its centre. The rule the
        # deterministic strategy already honoured: a pessimistic P50 built on
        # ignorance manufactures breaches out of not knowing.
        absent = sum(1 for v in req.features.values() if v is None)
        if absent:
            widen = 1 + 0.06 * absent
            half_lo = (mid - lo) * widen
            half_hi = (hi - mid) * widen
            lo, hi = mid - half_lo, mid + half_hi

        points.append(ForecastPoint(horizon=h, p10=round(lo, 2), p50=round(mid, 2), p90=round(hi, 2)))

    return ForecastResponse(
        enterpriseId=req.enterpriseId, modelVersion=b["version"], points=points
    )