commit 4ed61a2427824a255f5eb6ce89cec457396f4efc Author: LincPay Date: Tue Aug 25 13:43:40 2026 +0530 feat: the UdyamLinc ML layer — scoring, explanation and quantile forecast Brings the Python service under version control for the first time. It has been running and serving the demo since 23 August; until now it existed only on one developer's disk. Contents: training/build_dataset.py point-in-time join over the feature store, with leakage assertions. dscr_p10 and dscr_p50 are excluded deliberately — both are derived from the forecast, which is derived from the instalment, so the model would read its own answer. training/train_stress.py LightGBM vs XGBoost vs two single-feature baselines, expanding-window validation over as-of dates, publishes to the model_metric table training/train_forecast.py three quantile heads (P10/P50/P90), horizon as a feature rather than one model per horizon serving/explain.py SHAP TreeExplainer folded into the five risk families the console already speaks. The family map mirrors RuleBasedScoringStrategy in the API repo; where they disagree the rule engine wins. serving/app.py /score, /score/batch, /forecast, /health, /model The trained models, their metrics and the training dataset are committed alongside the code. Without them the published AUC of 0.760 is a claim nobody else can check; at ~2.2 MB together they are worth the space. Measured, on 5,523 rows / 263 enterprises / 242 positives / 100% synthetic labels: LightGBM AUC-ROC 0.7599, and the emi_to_inflow baseline 0.7685 on the same folds. The baseline wins. That is recorded rather than tuned away — on generated labels, beating it would mostly prove the model had learned the generator. The quantile forecast beats the deterministic strategy on all three quantiles (pinball loss, 12-33% better) but flags every borrower as breaching, so the deployment runs udyam.forecast.engine=DETERMINISTIC and this model is held back pending calibration against real repayment outcomes. Co-Authored-By: Claude Opus 5 (1M context) diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b61d9fa --- /dev/null +++ b/.gitignore @@ -0,0 +1,26 @@ +# Python +__pycache__/ +*.py[cod] +*.egg-info/ +.pytest_cache/ +.ipynb_checkpoints/ + +# Virtualenvs. The project's venv lives at ../.mlvenv, outside this tree, but a +# developer who creates one in place should not commit it. +.venv/ +venv/ +.mlvenv/ + +# Secrets — the database URL and any credential override live in the +# environment, never in the repo. +.env +.env.* + +# Deliberately NOT ignored: +# models/*.pkl the trained stress and forecast models +# models/*metrics*.json the measured figures the deck quotes +# data/training.parquet the dataset those figures were computed from +# data/manifest.json row counts, label share, exclusions +# Together these are what make the published AUC reproducible by someone else. +# They total ~2.2 MB; if the dataset ever outgrows git, move it to a store and +# leave the manifest behind rather than dropping both. diff --git a/data/manifest.json b/data/manifest.json new file mode 100644 index 0000000..ae9251f --- /dev/null +++ b/data/manifest.json @@ -0,0 +1,55 @@ +{ + "rows": 5523, + "enterprises": 263, + "as_of_dates": 21, + "date_min": "2024-08-31", + "date_max": "2026-04-30", + "positives": 242, + "positive_rate": 0.0438, + "features": [ + "feed_cost_change", + "output_price_change", + "margin_squeeze", + "price_z_score", + "rainfall_anomaly", + "heat_anomaly", + "cash_flow_change", + "emi_delay", + "inventory_days", + "upi_inflow_change", + "feed_staleness_days", + "days_since_last_entry", + "emi_to_inflow" + ], + "excluded": { + "dscr_p10": "reads forecast_point with no as-of filter \u2014 present leaks into past", + "dscr_p50": "same" + }, + "horizon_months": 3, + "dpd_threshold": 30, + "label_source_split": { + "SEEDED": 273, + "SEEDED_HISTORY": 6039 + }, + "synthetic_label_share": 1.0, + "feature_coverage": { + "feed_cost_change": 84.1, + "output_price_change": 97.7, + "margin_squeeze": 83.4, + "price_z_score": 90.3, + "rainfall_anomaly": 100.0, + "heat_anomaly": 100.0, + "cash_flow_change": 100.0, + "emi_delay": 100.0, + "inventory_days": 100.0, + "upi_inflow_change": 0.0, + "feed_staleness_days": 100.0, + "days_since_last_entry": 100.0, + "emi_to_inflow": 100.0 + }, + "caveats": [ + "Feature rows are reconstructed by fact date, not knowledge date \u2014 the archive was ingested in one pass, so a past row uses a value a live run of that date would not yet have had.", + "Labels are generated DPD history (SEEDED_HISTORY). Metrics computed on this dataset measure the generator as much as the borrowers, and must be reported as computed on synthetic history.", + "dscr_p10 and dscr_p50 are excluded: both read the current forecast at every as-of date." + ] +} \ No newline at end of file diff --git a/data/training.parquet b/data/training.parquet new file mode 100644 index 0000000..e94a631 Binary files /dev/null and b/data/training.parquet differ diff --git a/models/forecast_metrics.json b/models/forecast_metrics.json new file mode 100644 index 0000000..98b271d --- /dev/null +++ b/models/forecast_metrics.json @@ -0,0 +1,61 @@ +{ + "rows": 30711, + "enterprises": 263, + "as_of_months": 22, + "horizons": 6, + "validation": "expanding-window time-series split over as-of months", + "feature_order": [ + "feed_cost_change", + "output_price_change", + "margin_squeeze", + "price_z_score", + "rainfall_anomaly", + "heat_anomaly", + "cash_flow_change", + "emi_delay", + "inventory_days", + "feed_staleness_days", + "days_since_last_entry", + "emi_to_inflow", + "feed_cost_change__missing", + "output_price_change__missing", + "margin_squeeze__missing", + "price_z_score__missing", + "rainfall_anomaly__missing", + "heat_anomaly__missing", + "cash_flow_change__missing", + "emi_delay__missing", + "inventory_days__missing", + "feed_staleness_days__missing", + "days_since_last_entry__missing", + "emi_to_inflow__missing", + "trail_mean", + "trail_sd", + "trail_min", + "trail_inflow", + "trail_months", + "emi", + "horizon", + "sector_id", + "district_id" + ], + "quantiles": { + "p10": { + "alpha": 0.1, + "pinball_model": 3150.78, + "pinball_deterministic": 4182.0 + }, + "p50": { + "alpha": 0.5, + "pinball_model": 6194.72, + "pinball_deterministic": 7061.24 + }, + "p90": { + "alpha": 0.9, + "pinball_model": 2770.14, + "pinball_deterministic": 4112.53 + } + }, + "verdict": "3/3 quantiles beat the deterministic strategy on held-out months", + "trainedAt": "2026-08-24T11:07:50.265843" +} \ No newline at end of file diff --git a/models/forecast_model.pkl b/models/forecast_model.pkl new file mode 100644 index 0000000..017a32a Binary files /dev/null and b/models/forecast_model.pkl differ diff --git a/models/metrics.json b/models/metrics.json new file mode 100644 index 0000000..f897d7d --- /dev/null +++ b/models/metrics.json @@ -0,0 +1,192 @@ +{ + "dataset": { + "rows": 5523, + "positives": 242, + "positive_rate": 0.0438, + "as_of_dates": 21, + "synthetic_label_share": 1.0 + }, + "validation": "expanding-window time-series split over as-of dates", + "feature_order": [ + "feed_cost_change", + "output_price_change", + "margin_squeeze", + "price_z_score", + "rainfall_anomaly", + "heat_anomaly", + "cash_flow_change", + "emi_delay", + "inventory_days", + "upi_inflow_change", + "feed_staleness_days", + "days_since_last_entry", + "emi_to_inflow", + "feed_cost_change__missing", + "output_price_change__missing", + "margin_squeeze__missing", + "price_z_score__missing", + "rainfall_anomaly__missing", + "heat_anomaly__missing", + "cash_flow_change__missing", + "emi_delay__missing", + "inventory_days__missing", + "upi_inflow_change__missing", + "feed_staleness_days__missing", + "days_since_last_entry__missing", + "emi_to_inflow__missing", + "dpd_at_t", + "outstanding_at_t", + "emi_at_t", + "sector_id", + "district_id" + ], + "algorithms": { + "lightgbm": { + "folds": [ + { + "n": 789, + "positives": 38, + "auc_roc": 0.6814422874763473, + "pr_auc": 0.12090605567917119, + "brier": 0.0690301962610968, + "red_flagged": 23, + "precision_red": 0.17391304347826086, + "recall_red": 0.10526315789473684, + "recall_flagged": 0.3684210526315789, + "train_to": "2025-03-31", + "test_from": "2025-04-30" + }, + { + "n": 789, + "positives": 31, + "auc_roc": 0.7962805345135756, + "pr_auc": 0.17203736204143655, + "brier": 0.08691410549130126, + "red_flagged": 54, + "precision_red": 0.14814814814814814, + "recall_red": 0.25806451612903225, + "recall_flagged": 0.6129032258064516, + "train_to": "2025-06-30", + "test_from": "2025-07-31" + }, + { + "n": 789, + "positives": 39, + "auc_roc": 0.8423931623931624, + "pr_auc": 0.22591607572996394, + "brier": 0.11095460084002903, + "red_flagged": 90, + "precision_red": 0.2222222222222222, + "recall_red": 0.5128205128205128, + "recall_flagged": 0.8205128205128205, + "train_to": "2025-09-30", + "test_from": "2025-10-31" + }, + { + "n": 1052, + "positives": 51, + "auc_roc": 0.7193982488100136, + "pr_auc": 0.20698077930278122, + "brier": 0.08188506640426521, + "red_flagged": 59, + "precision_red": 0.22033898305084745, + "recall_red": 0.2549019607843137, + "recall_flagged": 0.43137254901960786, + "train_to": "2025-12-31", + "test_from": "2026-01-31" + } + ], + "pooled": { + "auc_roc": 0.7599, + "pr_auc": 0.1815, + "brier": 0.0872, + "precision_red": 0.1912, + "recall_red": 0.2828, + "recall_flagged": 0.5583 + } + }, + "xgboost": { + "folds": [ + { + "n": 789, + "positives": 38, + "auc_roc": 0.6397084588969094, + "pr_auc": 0.10510222243195182, + "brier": 0.055753882974386215, + "red_flagged": 13, + "precision_red": 0.15384615384615385, + "recall_red": 0.05263157894736842, + "recall_flagged": 0.15789473684210525, + "train_to": "2025-03-31", + "test_from": "2025-04-30" + }, + { + "n": 789, + "positives": 31, + "auc_roc": 0.7543195165546004, + "pr_auc": 0.14310881976856776, + "brier": 0.05676882714033127, + "red_flagged": 24, + "precision_red": 0.20833333333333334, + "recall_red": 0.16129032258064516, + "recall_flagged": 0.3870967741935484, + "train_to": "2025-06-30", + "test_from": "2025-07-31" + }, + { + "n": 789, + "positives": 39, + "auc_roc": 0.8195897435897437, + "pr_auc": 0.17177662573810526, + "brier": 0.0742882490158081, + "red_flagged": 41, + "precision_red": 0.1951219512195122, + "recall_red": 0.20512820512820512, + "recall_flagged": 0.5641025641025641, + "train_to": "2025-09-30", + "test_from": "2025-10-31" + }, + { + "n": 1052, + "positives": 51, + "auc_roc": 0.7052947052947053, + "pr_auc": 0.1285152676436765, + "brier": 0.06149039790034294, + "red_flagged": 32, + "precision_red": 0.125, + "recall_red": 0.0784313725490196, + "recall_flagged": 0.27450980392156865, + "train_to": "2025-12-31", + "test_from": "2026-01-31" + } + ], + "pooled": { + "auc_roc": 0.7297, + "pr_auc": 0.1371, + "brier": 0.0621, + "precision_red": 0.1706, + "recall_red": 0.1244, + "recall_flagged": 0.3459 + } + } + }, + "caveats": [ + "Feature rows are reconstructed by fact date, not knowledge date \u2014 the archive was ingested in one pass, so a past row uses a value a live run of that date would not yet have had.", + "Labels are generated DPD history (SEEDED_HISTORY). Metrics computed on this dataset measure the generator as much as the borrowers, and must be reported as computed on synthetic history.", + "dscr_p10 and dscr_p50 are excluded: both read the current forecast at every as-of date." + ], + "baselines": { + "repayment_history_only": { + "auc_roc": 0.6181, + "pr_auc": 0.0869, + "precision_red": 0.11 + }, + "obligation_ratio_only": { + "auc_roc": 0.7685, + "pr_auc": 0.1828, + "precision_red": 0.1001 + } + }, + "winner": "lightgbm", + "base_rate": 0.0438 +} \ No newline at end of file diff --git a/models/stress_model.pkl b/models/stress_model.pkl new file mode 100644 index 0000000..d999190 Binary files /dev/null and b/models/stress_model.pkl differ diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..94676a3 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,26 @@ +# UdyamLinc ML service — pinned to the versions the published metrics were +# produced with (AUC 0.760 on 5,523 rows, run 289, 2026-08-24). LightGBM and +# SHAP in particular are pinned rather than floated: a different LightGBM minor +# can change tree construction, and stress_model.pkl is a pickle of a fitted +# LightGBM object, so an unpinned load is not guaranteed to work at all. +# +# py -m venv .mlvenv +# .mlvenv/Scripts/python -m pip install -r requirements.txt +# .mlvenv/Scripts/python -m uvicorn serving.app:app --port 8000 + +# model +lightgbm==4.7.0 +xgboost==3.4.1 +scikit-learn==1.9.0 +shap==0.52.0 + +# data +pandas==3.0.5 +numpy==2.5.2 +pyarrow==25.0.1 +psycopg[binary]==3.3.4 + +# service +fastapi==0.141.1 +uvicorn==0.52.4 +pydantic==2.13.4 diff --git a/serving/app.py b/serving/app.py new file mode 100644 index 0000000..22c8bcb --- /dev/null +++ b/serving/app.py @@ -0,0 +1,289 @@ +""" +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 + ) diff --git a/serving/explain.py b/serving/explain.py new file mode 100644 index 0000000..2dcf2e1 --- /dev/null +++ b/serving/explain.py @@ -0,0 +1,111 @@ +""" +SHAP contributions, folded into the five families the console already speaks. + +The console, the reason sentences and the Hindi PWA are all written against +``liquidity | behavioural | trendVolatility | obligationLoad | externalShock``. +A model that explained itself in thirteen feature names would need every one of +those surfaces rewritten, and would hand a field officer a vocabulary they have +never used. So the SHAP values are aggregated to the families instead, and the +feature that contributed most inside each family is named alongside it. + +The family map mirrors ``RuleBasedScoringStrategy`` exactly. Where the two +disagree, the rule engine is right and this file is wrong — an officer must not +see the same enterprise explained two different ways depending on which engine +happened to score it. +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import shap + +LIQUIDITY = "liquidity" +BEHAVIOURAL = "behavioural" +TREND_VOLATILITY = "trendVolatility" +OBLIGATION_LOAD = "obligationLoad" +EXTERNAL_SHOCK = "externalShock" + +FAMILIES = [LIQUIDITY, BEHAVIOURAL, TREND_VOLATILITY, OBLIGATION_LOAD, EXTERNAL_SHOCK] + +FAMILY_OF: dict[str, str] = { + # liquidity — is there cash in the business + "cash_flow_change": LIQUIDITY, + "inventory_days": LIQUIDITY, + "upi_inflow_change": LIQUIDITY, + # behavioural — how the borrower has been acting + "emi_delay": BEHAVIOURAL, + "days_since_last_entry": BEHAVIOURAL, + "dpd_at_t": BEHAVIOURAL, + # trend / volatility — how unstable the market around them is + "price_z_score": TREND_VOLATILITY, + # obligation load — how heavy the instalment is against what comes in + "emi_to_inflow": OBLIGATION_LOAD, + "emi_at_t": OBLIGATION_LOAD, + "outstanding_at_t": OBLIGATION_LOAD, + # external shock — what the weather and the mandi are doing to them + "margin_squeeze": EXTERNAL_SHOCK, + "feed_cost_change": EXTERNAL_SHOCK, + "output_price_change": EXTERNAL_SHOCK, + "rainfall_anomaly": EXTERNAL_SHOCK, + "heat_anomaly": EXTERNAL_SHOCK, +} + + +class Explainer: + """Wraps a TreeExplainer and does the family arithmetic.""" + + def __init__(self, model, features: list[str]): + self.features = features + self._shap = shap.TreeExplainer(model) + + def explain(self, row: pd.DataFrame) -> tuple[dict[str, int], list[dict]]: + """ + :return: (sub-scores 0–100 per family, drivers sorted by contribution) + + Only contributions that push the score **up** are reported as drivers. + A negative SHAP value means "this is one of the reasons the score is not + worse", which is true and useless to an officer deciding whether to visit. + """ + values = self._shap.shap_values(row[self.features]) + if isinstance(values, list): # older binary API returns one array per class + values = values[1] + contributions = np.asarray(values).reshape(-1) + + by_family: dict[str, float] = {f: 0.0 for f in FAMILIES} + best_feature: dict[str, tuple[str, float]] = {} + + for name, value in zip(self.features, contributions): + base = name.removesuffix("__missing") + family = FAMILY_OF.get(base) + if family is None: + continue + by_family[family] += float(value) + held = best_feature.get(family) + if held is None or abs(value) > abs(held[1]): + best_feature[family] = (base, float(value)) + + positive = {f: max(0.0, v) for f, v in by_family.items()} + total = sum(positive.values()) + + # No family pushed the score up — a genuinely calm enterprise. Spread the + # weight evenly rather than emitting five zeros, which the console would + # render as an empty explanation for a score that still exists. + if total <= 0: + sub_scores = {f: 20 for f in FAMILIES} + else: + sub_scores = {f: int(round(100 * v / total)) for f, v in positive.items()} + + drivers = [] + for family in FAMILIES: + feature, raw = best_feature.get(family, (None, 0.0)) + drivers.append({ + "family": family, + "weight": sub_scores[family], + "featureCode": feature, + "value": float(row[feature].iloc[0]) if feature in row and pd.notna(row[feature].iloc[0]) else None, + "onPrior": bool(feature is None or (feature in row and pd.isna(row[feature].iloc[0]))), + "shap": round(raw, 5), + }) + drivers.sort(key=lambda d: d["weight"], reverse=True) + return sub_scores, drivers diff --git a/training/build_dataset.py b/training/build_dataset.py new file mode 100644 index 0000000..e387c4e --- /dev/null +++ b/training/build_dataset.py @@ -0,0 +1,288 @@ +""" +Assemble the training set from the feature store, point-in-time correct. + +Everything downstream is worthless if this file is wrong, and nothing downstream +can detect it — a leaked column produces excellent metrics and a useless model. +So this module does three things and states all of them in the manifest it +writes: what it selected, what it deliberately excluded, and how much of the +label is real. + +Shape of a row +-------------- +One row per (enterprise, as-of month end): + + features at T -> did DPD exceed 30 by T + 3 months + +The features come from ``feature_value``, which carries both ``valid_for_date`` +(the date the fact describes) and ``observed_at`` (when we learned it). The label +comes from ``loan_record`` three months later. The two never touch the same +table, so the label cannot leak through the feature join. + +What is deliberately excluded +----------------------------- +``dscr_p10`` and ``dscr_p50`` are dropped. Both are computed by +``FeatureEngineeringService.dscr()``, which reads ``forecast_point`` with **no +as-of filter** — so a backfilled row for March 2025 carries a forecast produced +in August 2026. That is the present leaking into the past. They are the two most +predictive-looking columns in the store and they would flatter every metric. + +A caveat the manifest records +----------------------------- +The backfilled rows are reconstructed by *fact date*, not by *knowledge date*: a +mandi price for March was ingested in August, so a March row uses a number a live +March run would not yet have had. Not label leakage, but not a live recording +either, and anything trained on it should say so. +""" + +from __future__ import annotations + +import argparse +import json +from dataclasses import dataclass, asdict +from pathlib import Path + +import pandas as pd +import psycopg + +# The catalogue, minus the two forecast-derived columns. Order is fixed and +# written to the manifest: a model served its columns in a different order fails +# silently, which is the worst way for a model to fail. +FEATURES = [ + "feed_cost_change", + "output_price_change", + "margin_squeeze", + "price_z_score", + "rainfall_anomaly", + "heat_anomaly", + "cash_flow_change", + "emi_delay", + "inventory_days", + "upi_inflow_change", + "feed_staleness_days", + "days_since_last_entry", + "emi_to_inflow", +] + +EXCLUDED = { + "dscr_p10": "reads forecast_point with no as-of filter — present leaks into past", + "dscr_p50": "same", +} + +# Months between the feature date and the outcome we are predicting. +HORIZON_MONTHS = 3 +DPD_THRESHOLD = 30 + +DEFAULT_DSN = "postgresql://postgres:root@localhost:5432/UdyamLinc_DEMO" + + +# --------------------------------------------------------------------------- # + +FEATURE_SQL = """ +select v.enterprise_id, + v.valid_for_date as as_of_date, + v.feature_code, + v.value, + v.missing_reason, + v.staleness_days +from feature_value v +where v.valid_for_date between %(start)s and %(end)s + and v.feature_code = any(%(codes)s) +""" + +# The label, and the enterprise attributes that are legitimately known at T. +# `dpd_at_t` is the borrower's state on the feature date, not the outcome — it is +# knowable then and is the single strongest honest predictor available. +LABEL_SQL = """ +with lab as ( + select enterprise_id, as_of_date, max(dpd) as dpd, + sum(outstanding_amount) as outstanding, sum(instalment_amount) as emi_amount + from loan_record + group by 1, 2 +) +select f.enterprise_id, + f.as_of_date, + e.sector_id, + e.district_id, + now_.dpd as dpd_at_t, + now_.outstanding as outstanding_at_t, + now_.emi_amount as emi_at_t, + fut.dpd as dpd_at_horizon +from (select distinct enterprise_id, valid_for_date as as_of_date + from feature_value + where valid_for_date between %(start)s and %(end)s) f +join enterprise e on e.enterprise_id = f.enterprise_id +left join lab now_ on now_.enterprise_id = f.enterprise_id + and now_.as_of_date = f.as_of_date +join lab fut on fut.enterprise_id = f.enterprise_id + and fut.as_of_date = ( + date_trunc('month', f.as_of_date + make_interval(months => %(horizon)s)) + + interval '1 month - 1 day')::date +""" + +# Provenance of the label itself, reported rather than assumed. +PROVENANCE_SQL = """ +select coalesce(source, 'UNKNOWN') as source, count(*) as n +from loan_record group by 1 +""" + + +@dataclass +class Manifest: + rows: int + enterprises: int + as_of_dates: int + date_min: str + date_max: str + positives: int + positive_rate: float + features: list[str] + excluded: dict[str, str] + horizon_months: int + dpd_threshold: int + label_source_split: dict[str, int] + synthetic_label_share: float + feature_coverage: dict[str, float] + caveats: list[str] + + +# --------------------------------------------------------------------------- # + + +def build(dsn: str, start: str, end: str) -> tuple[pd.DataFrame, Manifest]: + with psycopg.connect(dsn) as conn: + params = {"start": start, "end": end, "codes": FEATURES, "horizon": HORIZON_MONTHS} + long_features = pd.read_sql(FEATURE_SQL, conn, params=params) + spine = pd.read_sql(LABEL_SQL, conn, params=params) + provenance = pd.read_sql(PROVENANCE_SQL, conn) + + if spine.empty: + raise SystemExit( + "No rows: the feature store and the loan book do not overlap at this horizon.\n" + "Run POST /admin/features/backfill first." + ) + + # Wide feature matrix, plus a null indicator per column. `missing_reason` is + # itself signal — NO_CONSENT is a different fact from NO_SIGNAL — but the tree + # only needs to know the value was absent, and the reason is preserved in the + # store for a human to read. + wide = long_features.pivot_table( + index=["enterprise_id", "as_of_date"], + columns="feature_code", + values="value", + aggfunc="first", + ).reset_index() + + for code in FEATURES: + if code not in wide.columns: + wide[code] = pd.NA + wide[f"{code}__missing"] = wide[code].isna().astype("int8") + + df = spine.merge(wide, on=["enterprise_id", "as_of_date"], how="inner") + + df["label"] = (df["dpd_at_horizon"] > DPD_THRESHOLD).astype("int8") + df = df.drop(columns=["dpd_at_horizon"]) + + # Sector and district stay ordinal ints rather than pandas categoricals. + # A one-row request at serving time cannot reproduce a category dtype with the + # training levels, and LightGBM rejects the mismatch rather than warning. With + # five sectors and fourteen districts an ordinal split costs nothing. + + for col in FEATURES + ["dpd_at_t", "outstanding_at_t", "emi_at_t"]: + df[col] = pd.to_numeric(df[col], errors="coerce") + + df = df.sort_values(["as_of_date", "enterprise_id"]).reset_index(drop=True) + + _assert_no_leakage(df, dsn, start, end) + + source_split = dict(zip(provenance["source"], provenance["n"].astype(int))) + synthetic = sum(n for s, n in source_split.items() if s != "REAL") + total_labels = sum(source_split.values()) or 1 + + coverage = {c: round(float(df[c].notna().mean() * 100), 1) for c in FEATURES} + + manifest = Manifest( + rows=len(df), + enterprises=int(df["enterprise_id"].nunique()), + as_of_dates=int(df["as_of_date"].nunique()), + date_min=str(df["as_of_date"].min()), + date_max=str(df["as_of_date"].max()), + positives=int(df["label"].sum()), + positive_rate=round(float(df["label"].mean()), 4), + features=FEATURES, + excluded=EXCLUDED, + horizon_months=HORIZON_MONTHS, + dpd_threshold=DPD_THRESHOLD, + label_source_split=source_split, + synthetic_label_share=round(synthetic / total_labels, 4), + feature_coverage=coverage, + caveats=[ + "Feature rows are reconstructed by fact date, not knowledge date — the " + "archive was ingested in one pass, so a past row uses a value a live run " + "of that date would not yet have had.", + "Labels are generated DPD history (SEEDED_HISTORY). Metrics computed on " + "this dataset measure the generator as much as the borrowers, and must be " + "reported as computed on synthetic history.", + "dscr_p10 and dscr_p50 are excluded: both read the current forecast at " + "every as-of date.", + ], + ) + return df, manifest + + +def _assert_no_leakage(df: pd.DataFrame, dsn: str, start: str, end: str) -> None: + """ + Two checks, because the failure is silent. + + The first is structural: no feature row may describe a date after the as-of + date it was filed under. The second is the excluded set — if a forecast-derived + column ever reappears in the matrix, the build must stop rather than train on it. + """ + leaked = [c for c in EXCLUDED if c in df.columns] + if leaked: + raise SystemExit(f"Excluded columns present in the matrix: {leaked}") + + with psycopg.connect(dsn) as conn, conn.cursor() as cur: + cur.execute( + """ + select count(*) from feature_value + where valid_for_date between %s and %s + and staleness_days is not null and staleness_days < 0 + """, + (start, end), + ) + future = cur.fetchone()[0] + if future: + raise SystemExit(f"{future} feature rows carry negative staleness — data from after their as-of date") + + +def main() -> None: + ap = argparse.ArgumentParser(description="Build the UdyamLinc training set") + ap.add_argument("--dsn", default=DEFAULT_DSN) + ap.add_argument("--start", default="2024-08-01") + ap.add_argument("--end", default="2026-06-30") + ap.add_argument("--out", default="data") + args = ap.parse_args() + + df, manifest = build(args.dsn, args.start, args.end) + + out = Path(args.out) + out.mkdir(parents=True, exist_ok=True) + df.to_parquet(out / "training.parquet", index=False) + (out / "manifest.json").write_text(json.dumps(asdict(manifest), indent=2), encoding="utf-8") + + print(f"rows {manifest.rows:,}") + print(f"enterprises {manifest.enterprises}") + print(f"as-of dates {manifest.as_of_dates} ({manifest.date_min} .. {manifest.date_max})") + print(f"positives {manifest.positives} ({manifest.positive_rate:.1%})") + print(f"synthetic labels {manifest.synthetic_label_share:.0%}") + print(f"written {out / 'training.parquet'}") + + if manifest.positives < 50: + raise SystemExit( + f"\nRefusing to call this trainable: {manifest.positives} positive cases.\n" + "A classifier fitted here would learn the majority class." + ) + + +if __name__ == "__main__": + main() diff --git a/training/train_forecast.py b/training/train_forecast.py new file mode 100644 index 0000000..90fa8a6 --- /dev/null +++ b/training/train_forecast.py @@ -0,0 +1,294 @@ +""" +Quantile heads for the cash-flow forecast. + +The problem statement is "AI-Driven **Cash Flow Prediction** & Risk Flagging". +The risk head is a model; the forecast, until now, was +``DeterministicForecastStrategy`` — a median of monthly ledger position with a +capped drift and a band widening as sqrt(h). Honest, explainable, and not AI. +This trains the replacement. + +Shape +----- +One row per (enterprise, as-of month end, horizon 1..6): + + features known at T -> net ledger position in month T + h + +Three models, at alpha = 0.10 / 0.50 / 0.90. **Horizon is a feature, not a +separate model per step**: six models per quantile would be eighteen artifacts +that cannot share the signal they all depend on, and the h=6 model would train on +a sixth of the rows. + +Two rules carried over from the deterministic strategy +----------------------------------------------------- +1. Quantile regression does not guarantee ordering — P10 can be predicted above + P50 on an individual row. The three predictions are sorted per row before + they leave here. A crossed band is worse than no band. +2. **Missing inputs widen the band; they never lower its centre.** A pessimistic + P50 built on ignorance manufactures breaches, which is how an early-warning + system loses an officer's trust in one season. + +What the metrics mean +--------------------- +Pinball loss, per quantile, against the deterministic strategy reproduced on the +same held-out months. Lower is better. This is the comparison that decides +whether the model earns the switch — and, as with the stress head, it is measured +against a generated ledger, so it measures the generator too. +""" + +from __future__ import annotations + +import argparse +import json +import pickle +from datetime import datetime +from pathlib import Path + +import lightgbm as lgb +import numpy as np +import pandas as pd +import psycopg + +DSN = "postgresql://postgres:root@localhost:5432/UdyamLinc_DEMO" + +QUANTILES = {"p10": 0.10, "p50": 0.50, "p90": 0.90} +HORIZONS = 6 + +# Features known at T. The point-in-time discipline is the same as the stress +# head's: `valid_for_date <= T`. Forecast-derived features are excluded for the +# same reason they are there — dscr_* reads the current forecast at every date. +FEATURE_CODES = [ + "feed_cost_change", + "output_price_change", + "margin_squeeze", + "price_z_score", + "rainfall_anomaly", + "heat_anomaly", + "cash_flow_change", + "emi_delay", + "inventory_days", + "feed_staleness_days", + "days_since_last_entry", + "emi_to_inflow", +] + +# --------------------------------------------------------------------------- # + +# Monthly net position, and the trailing statistics a forecaster would have had. +# Everything is computed from months strictly before the as-of month, so a row +# never sees the month it is filed under, let alone the one it predicts. +SPINE_SQL = """ +with monthly as ( + select enterprise_id, + date_trunc('month', entry_date)::date as mo, + sum(case when kind = 'IN' then amount when kind = 'OUT' then -amount else 0 end) as net, + sum(case when kind = 'IN' then amount else 0 end) as inflow + from ledger_entry + group by 1, 2 +), +stats as ( + select m.enterprise_id, + m.mo as as_of_mo, + avg(h.net) as trail_mean, + stddev_pop(h.net) as trail_sd, + min(h.net) as trail_min, + avg(h.inflow) as trail_inflow, + count(*) as trail_months + from monthly m + join monthly h + on h.enterprise_id = m.enterprise_id + and h.mo < m.mo + and h.mo >= m.mo - interval '6 month' + group by 1, 2 +), +obligation as ( + select enterprise_id, as_of_date, sum(instalment_amount) as emi + from loan_record group by 1, 2 +) +select s.enterprise_id, + s.as_of_mo, + s.trail_mean, s.trail_sd, s.trail_min, s.trail_inflow, s.trail_months, + e.sector_id, e.district_id, + o.emi, + t.mo as target_mo, + t.net as target_net, + (extract(year from t.mo) * 12 + extract(month from t.mo)) + - (extract(year from s.as_of_mo) * 12 + extract(month from s.as_of_mo)) as horizon +from stats s +join enterprise e on e.enterprise_id = s.enterprise_id +left join obligation o + on o.enterprise_id = s.enterprise_id + and o.as_of_date = (s.as_of_mo + interval '1 month - 1 day')::date +join monthly t + on t.enterprise_id = s.enterprise_id + and t.mo > s.as_of_mo + and t.mo <= s.as_of_mo + interval '6 month' +where s.trail_months >= 3 +""" + +FEATURES_SQL = """ +select enterprise_id, valid_for_date, feature_code, value +from feature_value +where feature_code = any(%(codes)s) +""" + + +def build(dsn: str) -> pd.DataFrame: + with psycopg.connect(dsn) as conn: + spine = pd.read_sql(SPINE_SQL, conn) + feats = pd.read_sql(FEATURES_SQL, conn, params={"codes": FEATURE_CODES}) + + if spine.empty: + raise SystemExit("No rows. Has the ledger backfill run?") + + # Features are filed at month end; the spine is keyed on month start. + feats["as_of_mo"] = pd.to_datetime(feats["valid_for_date"]).values.astype("datetime64[M]") + wide = feats.pivot_table( + index=["enterprise_id", "as_of_mo"], columns="feature_code", values="value", aggfunc="first" + ).reset_index() + + spine["as_of_mo"] = pd.to_datetime(spine["as_of_mo"]).values.astype("datetime64[M]") + df = spine.merge(wide, on=["enterprise_id", "as_of_mo"], how="left") + + for code in FEATURE_CODES: + if code not in df.columns: + df[code] = np.nan + df[f"{code}__missing"] = df[code].isna().astype("int8") + + numeric = FEATURE_CODES + [ + "trail_mean", "trail_sd", "trail_min", "trail_inflow", "trail_months", + "emi", "horizon", "target_net", "sector_id", "district_id", + ] + for col in numeric: + df[col] = pd.to_numeric(df[col], errors="coerce") + + df = df.dropna(subset=["target_net", "horizon"]) + return df.sort_values(["as_of_mo", "enterprise_id", "horizon"]).reset_index(drop=True) + + +def feature_columns(df: pd.DataFrame) -> list[str]: + cols = FEATURE_CODES + [f"{c}__missing" for c in FEATURE_CODES] + cols += ["trail_mean", "trail_sd", "trail_min", "trail_inflow", "trail_months", + "emi", "horizon", "sector_id", "district_id"] + return [c for c in cols if c in df.columns] + + +def pinball(y: np.ndarray, pred: np.ndarray, alpha: float) -> float: + """The loss quantile regression is actually optimising. Lower is better.""" + delta = y - pred + return float(np.mean(np.maximum(alpha * delta, (alpha - 1) * delta))) + + +def deterministic_baseline(frame: pd.DataFrame, alpha: float) -> np.ndarray: + """ + The strategy currently in production, reproduced so the comparison is fair. + + Centre is the trailing mean; the band widens with sqrt(horizon) at the same + 18% base spread the Java implementation uses. If the model cannot beat this, + the switch is not earned and the config value stays on DETERMINISTIC. + """ + centre = frame["trail_mean"].to_numpy() + spread = 0.18 * np.sqrt(frame["horizon"].to_numpy()) + if alpha < 0.5: + return centre * (1 - spread) + if alpha > 0.5: + return centre * (1 + spread) + return centre + + +def expanding_folds(months: np.ndarray, n_folds: int = 3, min_train: int = 8): + uniq = np.array(sorted(pd.unique(months))) + step = max(1, (len(uniq) - min_train) // max(1, n_folds)) + for k in range(n_folds): + cut = min_train + k * step + if cut >= len(uniq): + break + test = uniq[cut : cut + step] if k < n_folds - 1 else uniq[cut:] + yield uniq[:cut], test + + +def run(out_dir: Path, dsn: str = DSN) -> dict: + df = build(dsn) + cols = feature_columns(df) + months = df["as_of_mo"].to_numpy() + + results: dict = { + "rows": int(len(df)), + "enterprises": int(df["enterprise_id"].nunique()), + "as_of_months": int(df["as_of_mo"].nunique()), + "horizons": int(df["horizon"].max()), + "validation": "expanding-window time-series split over as-of months", + "feature_order": cols, + "quantiles": {}, + } + print(f"rows {len(df):,} · enterprises {results['enterprises']} · months {results['as_of_months']}") + + for name, alpha in QUANTILES.items(): + model_losses, base_losses = [], [] + for train_months, test_months in expanding_folds(months): + tr = df[df["as_of_mo"].isin(train_months)] + te = df[df["as_of_mo"].isin(test_months)] + if te.empty or tr.empty: + continue + model = lgb.LGBMRegressor( + objective="quantile", alpha=alpha, + n_estimators=400, learning_rate=0.05, num_leaves=15, + min_child_samples=30, subsample=0.85, subsample_freq=1, + colsample_bytree=0.85, verbose=-1, random_state=42, + ) + model.fit(tr[cols], tr["target_net"]) + pred = model.predict(te[cols]) + y = te["target_net"].to_numpy() + model_losses.append(pinball(y, pred, alpha)) + base_losses.append(pinball(y, deterministic_baseline(te, alpha), alpha)) + + results["quantiles"][name] = { + "alpha": alpha, + "pinball_model": round(float(np.mean(model_losses)), 2), + "pinball_deterministic": round(float(np.mean(base_losses)), 2), + } + q = results["quantiles"][name] + better = q["pinball_model"] < q["pinball_deterministic"] + print(f"{name} pinball model {q['pinball_model']:>10,.0f} " + f"deterministic {q['pinball_deterministic']:>10,.0f} " + f"{'model wins' if better else 'DETERMINISTIC WINS'}") + + # Refit each quantile on everything for serving. + final = {} + for name, alpha in QUANTILES.items(): + model = lgb.LGBMRegressor( + objective="quantile", alpha=alpha, + n_estimators=400, learning_rate=0.05, num_leaves=15, + min_child_samples=30, subsample=0.85, subsample_freq=1, + colsample_bytree=0.85, verbose=-1, random_state=42, + ) + model.fit(df[cols], df["target_net"]) + final[name] = model + + wins = sum(1 for q in results["quantiles"].values() + if q["pinball_model"] < q["pinball_deterministic"]) + results["verdict"] = ( + f"{wins}/3 quantiles beat the deterministic strategy on held-out months" + ) + print(f"\n{results['verdict']}") + + out_dir.mkdir(parents=True, exist_ok=True) + with open(out_dir / "forecast_model.pkl", "wb") as fh: + pickle.dump({"models": final, "features": cols, "horizons": HORIZONS, + "version": "lightgbm-forecast-v1"}, fh) + (out_dir / "forecast_metrics.json").write_text( + json.dumps(results | {"trainedAt": datetime.now().isoformat()}, indent=2), encoding="utf-8" + ) + print(f"written: {out_dir / 'forecast_model.pkl'}") + return results + + +def main() -> None: + ap = argparse.ArgumentParser(description="Train the UdyamLinc quantile forecast heads") + ap.add_argument("--out", default="models") + ap.add_argument("--dsn", default=DSN) + args = ap.parse_args() + run(Path(args.out), args.dsn) + + +if __name__ == "__main__": + main() diff --git a/training/train_stress.py b/training/train_stress.py new file mode 100644 index 0000000..871d70e --- /dev/null +++ b/training/train_stress.py @@ -0,0 +1,317 @@ +""" +Train the stress head, and measure it honestly. + +Two algorithms are fitted on identical folds — LightGBM and XGBoost — so the +choice between them is a recorded number rather than a preference. Validation is +an expanding-window time-series split: fold *k* trains on every as-of date before +a cut and tests on the dates after it, which is the only split that means +anything for a model whose job is to see forward. + +Nothing here is calibrated against real defaults. The labels are generated DPD +history, and every metric this writes is stamped with that fact. + +Outputs +------- +``models/stress_.pkl`` the fitted pipeline plus its feature order +``models/metrics.json`` fold-by-fold and pooled metrics +""" + +from __future__ import annotations + +import argparse +import json +import pickle +from datetime import datetime +from pathlib import Path + +import numpy as np +import pandas as pd +from sklearn.metrics import ( + average_precision_score, + brier_score_loss, + precision_score, + recall_score, + roc_auc_score, +) + +import lightgbm as lgb +import xgboost as xgb + +# Mirrors udyam.risk.red-threshold / watch-threshold in application.properties. +# The console, the apex suppression rules and the Hindi PWA all speak these three +# bands; the model maps into them rather than inventing a fourth. +RED_THRESHOLD = 65 +WATCH_THRESHOLD = 35 + + +def load(data_dir: Path) -> tuple[pd.DataFrame, dict]: + df = pd.read_parquet(data_dir / "training.parquet") + manifest = json.loads((data_dir / "manifest.json").read_text(encoding="utf-8")) + return df, manifest + + +def feature_columns(df: pd.DataFrame, manifest: dict) -> list[str]: + """ + Feature order is fixed here and written into the artifact. A model served its + columns in a different order does not error — it returns confident nonsense. + """ + cols = list(manifest["features"]) + cols += [f"{c}__missing" for c in manifest["features"]] + cols += ["dpd_at_t", "outstanding_at_t", "emi_at_t", "sector_id", "district_id"] + return [c for c in cols if c in df.columns] + + +def expanding_folds(dates: np.ndarray, n_folds: int = 4, min_train: int = 8): + """ + Expanding-window folds over the distinct as-of dates. + + Never a random split. A random split over a panel puts March and June of the + same enterprise on opposite sides of the fence, and the model learns that + enterprise rather than the pattern — which reports as excellent accuracy and + reproduces as nothing. + """ + uniq = np.array(sorted(pd.unique(dates))) + if len(uniq) < min_train + n_folds: + n_folds = max(1, len(uniq) - min_train) + step = max(1, (len(uniq) - min_train) // max(1, n_folds)) + for k in range(n_folds): + cut = min_train + k * step + if cut >= len(uniq): + break + yield uniq[:cut], uniq[cut : cut + step] if k < n_folds - 1 else uniq[cut:] + + +def fit_lightgbm(x: pd.DataFrame, y: np.ndarray, pos_weight: float): + model = lgb.LGBMClassifier( + objective="binary", + n_estimators=250, + learning_rate=0.04, + num_leaves=7, + min_child_samples=45, + subsample=0.85, + subsample_freq=1, + colsample_bytree=0.85, + reg_lambda=5.0, + scale_pos_weight=pos_weight, + verbose=-1, + random_state=42, + ) + model.fit(x, y) + return model + + +def fit_xgboost(x: pd.DataFrame, y: np.ndarray, pos_weight: float): + model = xgb.XGBClassifier( + objective="binary:logistic", + n_estimators=400, + learning_rate=0.04, + max_depth=4, + min_child_weight=5, + subsample=0.85, + colsample_bytree=0.85, + reg_lambda=1.0, + scale_pos_weight=pos_weight, + tree_method="hist", + enable_categorical=True, + eval_metric="aucpr", + random_state=42, + ) + model.fit(x, y) + return model + + +FITTERS = {"lightgbm": fit_lightgbm, "xgboost": fit_xgboost} + + +def score_to_band(score: int) -> str: + if score >= RED_THRESHOLD: + return "RED" + if score >= WATCH_THRESHOLD: + return "WATCH" + return "STANDARD" + + +def evaluate_fold(y_true: np.ndarray, proba: np.ndarray) -> dict: + """ + Precision at the Red band is the number that matters operationally: the cost + of a false positive there is one field visit, and an officer who is sent on + three wasted visits stops believing the fourth alert. + """ + scores = np.clip(np.round(proba * 100), 0, 100) + red = scores >= RED_THRESHOLD + flagged = red | (scores >= WATCH_THRESHOLD) + + out = { + "n": int(len(y_true)), + "positives": int(y_true.sum()), + "auc_roc": float(roc_auc_score(y_true, proba)) if y_true.sum() else None, + "pr_auc": float(average_precision_score(y_true, proba)) if y_true.sum() else None, + "brier": float(brier_score_loss(y_true, proba)), + "red_flagged": int(red.sum()), + "precision_red": float(precision_score(y_true, red, zero_division=0)), + "recall_red": float(recall_score(y_true, red, zero_division=0)), + "recall_flagged": float(recall_score(y_true, flagged, zero_division=0)), + } + return out + + +def run(data_dir: Path, out_dir: Path) -> dict: + df, manifest = load(data_dir) + cols = feature_columns(df, manifest) + dates = df["as_of_date"].values + + results: dict = { + "dataset": { + "rows": manifest["rows"], + "positives": manifest["positives"], + "positive_rate": manifest["positive_rate"], + "as_of_dates": manifest["as_of_dates"], + "synthetic_label_share": manifest["synthetic_label_share"], + }, + "validation": "expanding-window time-series split over as-of dates", + "feature_order": cols, + "algorithms": {}, + "caveats": manifest["caveats"], + } + + for name, fitter in FITTERS.items(): + folds = [] + for train_dates, test_dates in expanding_folds(dates): + tr = df[df["as_of_date"].isin(train_dates)] + te = df[df["as_of_date"].isin(test_dates)] + if te.empty or tr["label"].sum() == 0: + continue + + pos_weight = float((tr["label"] == 0).sum() / max(1, (tr["label"] == 1).sum())) + model = fitter(tr[cols], tr["label"].values, pos_weight) + proba = model.predict_proba(te[cols])[:, 1] + + fold = evaluate_fold(te["label"].values, proba) + fold["train_to"] = str(max(train_dates)) + fold["test_from"] = str(min(test_dates)) + folds.append(fold) + + pooled = { + k: round(float(np.mean([f[k] for f in folds if f[k] is not None])), 4) + for k in ("auc_roc", "pr_auc", "brier", "precision_red", "recall_red", "recall_flagged") + if any(f[k] is not None for f in folds) + } + results["algorithms"][name] = {"folds": folds, "pooled": pooled} + print(f"{name:10s} AUC {pooled.get('auc_roc', float('nan')):.3f} " + f"PR-AUC {pooled.get('pr_auc', float('nan')):.3f} " + f"precision@RED {pooled.get('precision_red', float('nan')):.3f}") + + # Baselines, on the identical folds. + # + # Without these a metric is unreadable: AUC 0.74 is either a real result or a + # restatement of "a borrower who is late today is late in three months", + # and only the comparison tells you which. The Reports page charts the model + # against these rather than against an invented bureau score. + results["baselines"] = {} + for label, column in (("repayment_history_only", "dpd_at_t"), + ("obligation_ratio_only", "emi_to_inflow")): + folds = [] + for train_dates, test_dates in expanding_folds(dates): + te = df[df["as_of_date"].isin(test_dates)] + if te.empty or te["label"].sum() == 0: + continue + raw = te[column].fillna(te[column].median()).values.astype(float) + # Rank-scale to [0,1] so it is comparable to a probability. + proba = (pd.Series(raw).rank(pct=True)).values + folds.append(evaluate_fold(te["label"].values, proba)) + if folds: + results["baselines"][label] = { + k: round(float(np.mean([f[k] for f in folds if f[k] is not None])), 4) + for k in ("auc_roc", "pr_auc", "precision_red") + if any(f[k] is not None for f in folds) + } + print(f"{label:24s} AUC {results['baselines'][label].get('auc_roc', float('nan')):.3f} " + f"PR-AUC {results['baselines'][label].get('pr_auc', float('nan')):.3f}") + + # The winner is chosen by PR-AUC, not AUC-ROC. At a 4% positive rate ROC is + # flattered by the majority class; precision-recall is the honest read. + winner = max(results["algorithms"], key=lambda a: results["algorithms"][a]["pooled"].get("pr_auc", 0)) + results["winner"] = winner + results["base_rate"] = round(float(df["label"].mean()), 4) + print(f"\nwinner: {winner} (by PR-AUC)") + + # Refit the winner on everything for serving. + pos_weight = float((df["label"] == 0).sum() / max(1, (df["label"] == 1).sum())) + final = FITTERS[winner](df[cols], df["label"].values, pos_weight) + + out_dir.mkdir(parents=True, exist_ok=True) + with open(out_dir / "stress_model.pkl", "wb") as fh: + pickle.dump({"model": final, "algorithm": winner, "features": cols, + "red_threshold": RED_THRESHOLD, "watch_threshold": WATCH_THRESHOLD}, fh) + (out_dir / "metrics.json").write_text(json.dumps(results, indent=2), encoding="utf-8") + print(f"written: {out_dir / 'stress_model.pkl'}") + return results + + +DSN = "postgresql://postgres:root@localhost:5432/UdyamLinc_DEMO" + +METRICS = ("auc_roc", "pr_auc", "precision_red", "recall_red", "brier") + + +def publish(results: dict, dsn: str = DSN) -> int: + """ + Write the backtest to `model_metric`, where the console reads it. + + The Reports page used to render 0.87 from a constant nobody computed. These + rows are the replacement, and each one carries the share of synthetic labels + it was measured against — so the accuracy cannot be quoted without the + caveat, because they are the same record. + + Baselines are written as rows too. A model AUC means nothing on its own; the + single-feature arms are what say whether it is a result or a restatement. + """ + import psycopg + + stamp = datetime.now() + ds = results["dataset"] + version = f"{results['winner']}-stress-v1" + rows = [] + + for arm, payload in results["algorithms"].items(): + for metric, value in payload["pooled"].items(): + if metric in METRICS: + rows.append((version, arm, False, metric, value)) + + for arm, payload in results.get("baselines", {}).items(): + for metric, value in payload.items(): + if metric in METRICS: + rows.append((version, arm, True, metric, value)) + + with psycopg.connect(dsn) as conn, conn.cursor() as cur: + for version_, arm, is_baseline, metric, value in rows: + cur.execute( + """ + insert into model_metric + (model_version, arm, baseline, metric, value, dataset_rows, + dataset_positives, synthetic_label_share, validation, computed_at) + values (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) + """, + (version_, arm, is_baseline, metric, round(float(value), 5), + ds["rows"], ds["positives"], ds["synthetic_label_share"], + results["validation"][:120], stamp), + ) + conn.commit() + + print(f"published {len(rows)} metric rows at {stamp:%Y-%m-%d %H:%M}") + return len(rows) + + +def main() -> None: + ap = argparse.ArgumentParser(description="Train the UdyamLinc stress head") + ap.add_argument("--data", default="data") + ap.add_argument("--out", default="models") + ap.add_argument("--no-publish", action="store_true", + help="train and evaluate without writing to model_metric") + args = ap.parse_args() + results = run(Path(args.data), Path(args.out)) + if not args.no_publish: + publish(results) + + +if __name__ == "__main__": + main()