"""IFRS 9 SICR staging -- the RELATIVE deterioration test (notes section 2.2). METHODOLOGY (knowledge/sources/ifrs9_credit_risk_notes.md section 2) ---------------------------------------------------------------------- SICR is a *relative deterioration* test, not an absolute credit-quality test: compare the lifetime PD over the REMAINING life at the reporting date with the lifetime PD expected FOR THAT SAME PERIOD at initial recognition. A loan originated risky and still equally risky has NOT suffered SICR. For a loan observed at reporting quarter t (end-of-quarter convention): R = mat_time - t remaining contractual life, in quarters (floored at 1: a handful of rows sit at/past scheduled maturity while still on book -- they get a one-quarter window) lifetime_pd_now = cumulative default PD over quarters t+1 .. t+R, loan ages age_t+1 .. age_t+R, computed from the competing-risk hazard engine (engine/hazard.py pd_term_structure conventions) with covariates FROZEN at their CURRENT (time-t) values lifetime_pd_orig = cumulative default PD over the SAME quarters and the SAME ages, with covariates frozen at their ORIGINATION values (below) -- i.e. what the model would have projected for exactly this window at initial recognition Both PDs live on the identical age window and identical competing-risk survival algebra, so their ratio isolates covariate deterioration (collateral, macro, refinancing moneyness) from seasoning: the age baseline is common to numerator and denominator projections. TIMING CONVENTION (inherits engine/hazard.py's binding section): the reporting date is the END of quarter t; the row's own event flags for quarter t are known (a default row is Stage 3, not a projection). The first projected quarter is t+1 at age age_t + 1, matching pd_term_structure's "loan_age is the age in the FIRST projected quarter". All macro covariates enter as the panel's lagged columns; no lookahead is introduced anywhere in this module -- origination covariates reference only quarters <= orig_time (up to the flagged clamp below). ORIGINATION-COVARIATE RECONSTRUCTION (per-loan, from panel columns) ---------------------------------------------------------------------- updated_ltv -> LTV_orig_time (the contractual origination LTV. The panel's updated-LTV formula collapses to it at origination GIVEN balance_time = original balance; DCR's first observed balance deviates for ~22% of age-0 rows -- median |gap| 0.2pp, 1% tail large -- so the CONTRACTUAL LTV is the initial- recognition benchmark, not the first observed panel row) FICO / occupancy / property flags unchanged (origination snapshots) prepay_incentive -> Interest_Rate_orig_time - market_rate(orig_time) where market_rate(q) is the panel's per-quarter MEDIAN rate_time (rate_time varies mildly across loans within a quarter, so a median map is a documented approximation). If the origination note rate is missing-coded (orig_rate_missing == 1) the CURRENT note rate substitutes -- flagged per loan in orig_rate_proxy (mostly fixed-rate loans, so the current note rate is a good proxy for the origination one). uer_lag1 -> uer(orig_time - 1) uer_chg4_lag1 -> uer(orig_time - 1) - uer(orig_time - 5) gdp_lag1 -> gdp(orig_time - 1) hpi_growth_lag1 -> ln hpi(orig_time - 1) - ln hpi(orig_time - 2) from the panel-internal time -> macro map (macros are national, verified constant within quarter by the panel build). FLAGGED APPROXIMATION: for loans originated before the macro window (orig_time - 5 < first panel quarter; ~2-4% of snapshot rows -- left-truncated seasoned entrants) the referenced quarter is clamped to the earliest available one, i.e. the earliest observed macro proxies the pre-window macro. Such loans carry orig_macro_approx = True in the output so downstream governance can quarantine them. STAGE RULE (StagingConfig; defaults follow the notes' conventions) ---------------------------------------------------------------------- Stage 3 default rows (default_event == 1 at the reporting quarter). The panel truncates histories at the first terminal event, so Stage-3 share at a snapshot equals that quarter's default incidence by construction. Stage 2 quantitative SICR trigger, EBA-style doubling convention: lifetime_pd_now > ratio_threshold * lifetime_pd_orig (2.0x) AND an absolute add-on so tiny PDs cannot flip stages on noise: ann_pd_now - ann_pd_orig > abs_addon (0.5pp ANNUALISED) where ann_pd = 1 - (1 - lifetime_pd)^(4/R) is the per-annum default probability equivalent to the R-quarter lifetime PD (documented conversion: R quarters = R/4 years; the annualised basis makes one add-on threshold comparable across loans with different remaining lives). OR the 30-days-past-due backstop -- see LOUD note below. Stage 1 everything else. >>> BACKSTOP IS STRUCTURALLY INERT ON THIS DATASET <<< IFRS 9's rebuttable presumption of SICR at 30 DPD requires a delinquency ladder. The Deep Credit Risk panel's status is only performing / default / payoff -- NO 30/60/90 DPD states exist, so the backstop hook (config.backstop_30dpd, config.dpd_col) can never fire here. The hook is implemented and tested for the day a dpd column exists; its absence today is a DOCUMENTED SIMPLIFICATION, not an implementation choice: Stage-2 populations below are quantitative-trigger-only and would be strictly larger with a live 30-DPD backstop. PROBATION / CURE: transfer back to Stage 1 requires the trigger to have been off for cure_quarters consecutive quarters. Because the quantitative trigger is a pure function of the quarter's covariates (memoryless), the sticky state machine is EXACTLY equivalent to the stateless window rule "Stage 2 iff the trigger fired at any of t, t-1, .., t-(cure_quarters-1)", which is what assign_stages evaluates (it scores the trigger on the history rows present in the input frame). Missing history rows (loan entered at t, within-loan gap, or a single-quarter input frame) contribute no trigger -- with a pure snapshot frame probation degrades gracefully to the raw trigger, documented behaviour. PERFORMANCE STRATEGY (documented; exactness cross-checked) ---------------------------------------------------------------------- Naively each loan needs a pd_term_structure call over up to ~190 quarters, twice (now / origination scenarios): O(n_loans * horizon) GLM design builds. Because loan_age enters the hazard ONLY through the cr() spline basis and every other covariate is frozen over the projection, the cloglog linear predictor SEPARATES: eta_i(a) = s(a) + c_i, s(a) = spline-basis columns @ beta_spline c_i = the loan's frozen-covariate offset so the whole snapshot needs ONE design-matrix build per cause on the n snapshot rows (for the offsets c_i) plus ONE on the shared integer age grid (for s(a)) -- O(n_loans + n_ages) instead of O(n_loans * horizon). Hazards, competing-risk survival and cumulative PD are then pure numpy on an (n_loans x max_R) grid. This is an EXACT factorisation, not an approximation: crosscheck_term_structure() re-computes a fixed-seed sample of loans through pd_term_structure row-by-row and both the test suite and the analysis harness assert agreement to ~1e-10 (observed ~6e-16). The survival fan is chunked over loans (_CHUNK_LOANS = 8192), so peak memory is a few (8192 x max_R) float arrays (~13 MB each at R~200) regardless of book size; the full 49,974-loan panel benchmark runs both legs in under a second. DOCUMENTED SIMPLIFICATIONS (also surfaced in outputs/staging/staging_report.md) ---------------------------------------------------------------------- * Frozen-covariate projections on both legs (rung-1 tail assumption inherited from pd_term_structure): no macro path, no LTV amortisation. Scenario conditioning arrives with the satellite-model rung. * 30-DPD backstop inert (no delinquency ladder in DCR) -- see above. * No low-credit-risk (investment-grade) exemption: retail mortgage book, the exemption is a bond-book device and would only relabel loans the quantitative test already leaves in Stage 1. * No qualitative triggers (watchlist / forbearance flags do not exist in the dataset). * Origination macro clamp and median-market-rate approximations, flagged per loan (orig_macro_approx, orig_rate_proxy). * Loans at/past scheduled maturity while still alive (R <= 0; a handful of rows) get a floored one-quarter window rather than an exception. """ from __future__ import annotations from dataclasses import dataclass import numpy as np import pandas as pd from patsy import build_design_matrices from engine.hazard import ( REQUIRED_COLS, HazardModel, _prepare, pd_term_structure, ) #: same per-period exit-probability cap as pd_term_structure _EXIT_CAP = 0.999999 #: loans per numpy chunk in _cum_default_pd (memory bound; result-invariant) _CHUNK_LOANS = 8192 #: panel columns lifetime_pd_table needs beyond REQUIRED_COLS _PANEL_COLS = [ "id", "time", "orig_time", "mat_time", "loan_age", "LTV_orig_time", "Interest_Rate_orig_time", "interest_rate_time", "orig_rate_missing", "default_event", "payoff_event", ] @dataclass(frozen=True) class StagingConfig: """SICR staging thresholds (module docstring for the rule they drive). ratio_threshold : lifetime-PD ratio trigger; 2.0 = the doubling convention (EBA 2018 stress-test methodology). abs_addon : absolute add-on on the ANNUALISED lifetime PD, decimal (0.005 = 0.5pp p.a.); annualisation ann = 1 - (1-PD_life)^(4/R). cure_quarters : consecutive trigger-free quarters required before a Stage-2 loan returns to Stage 1 (probation). backstop_30dpd / dpd_col : the 30-DPD rebuttable presumption hook. INERT on the DCR panel -- no delinquency ladder exists (module docstring, loud note). Fires only if dpd_col is present. max_horizon : optional cap on the remaining-life window R (quarters). Both PD legs use the SAME window, so a cap is SICR-neutral in the ratio; None (default) = full contractual remaining life. """ ratio_threshold: float = 2.0 abs_addon: float = 0.005 cure_quarters: int = 2 backstop_30dpd: bool = True dpd_col: str = "dpd_time" max_horizon: int | None = None # --------------------------------------------------------------------------- # macro map + origination covariates # --------------------------------------------------------------------------- def build_macro_map(panel_df: pd.DataFrame) -> pd.DataFrame: """Per-quarter national macro map from the panel itself. uer/gdp/hpi are national series, constant across loans within a quarter (verified in the panel build), so first() is exact. rate_time varies mildly across loans within a quarter; the MEDIAN defines the market-rate map used for the origination prepayment incentive (documented approximation). Index: contiguous quarters min(time)..max(time). """ g = panel_df.groupby("time") out = pd.DataFrame({ "uer": g["uer_time"].first(), "gdp": g["gdp_time"].first(), "hpi": g["hpi_time"].first(), "rate_med": g["rate_time"].median(), }) full = np.arange(out.index.min(), out.index.max() + 1) out = out.reindex(full) if out.isna().any().any(): raise ValueError("macro map has gaps: quarters missing from panel_df") return out def origination_covariates(snapshot_df: pd.DataFrame, macro_map: pd.DataFrame) -> pd.DataFrame: """Rebuild the hazard covariates as they stood at initial recognition. Returns a frame aligned to snapshot_df with REQUIRED_COLS (loan_age kept at the CURRENT age -- the projection window is the same for both legs and the frozen-covariate offset is age-independent) plus the two approximation flags orig_macro_approx / orig_rate_proxy (module docstring). """ o = snapshot_df["orig_time"].to_numpy() tmin, tmax = int(macro_map.index.min()), int(macro_map.index.max()) def series(name: str, lag: int) -> np.ndarray: pos = np.clip(o - lag, tmin, tmax) - tmin return macro_map[name].to_numpy()[pos] orig_rate_missing = snapshot_df["orig_rate_missing"].to_numpy() == 1 note0 = np.where(orig_rate_missing, snapshot_df["interest_rate_time"].to_numpy(), snapshot_df["Interest_Rate_orig_time"].to_numpy()) out = pd.DataFrame({ "loan_age": snapshot_df["loan_age"].to_numpy(), "FICO_orig_time": snapshot_df["FICO_orig_time"].to_numpy(), "updated_ltv": snapshot_df["LTV_orig_time"].to_numpy(), "prepay_incentive": note0 - series("rate_med", 0), "investor_orig_time": snapshot_df["investor_orig_time"].to_numpy(), "REtype_CO_orig_time": snapshot_df["REtype_CO_orig_time"].to_numpy(), "REtype_PU_orig_time": snapshot_df["REtype_PU_orig_time"].to_numpy(), "REtype_SF_orig_time": snapshot_df["REtype_SF_orig_time"].to_numpy(), "uer_lag1": series("uer", 1), "uer_chg4_lag1": series("uer", 1) - series("uer", 5), "gdp_lag1": series("gdp", 1), "hpi_growth_lag1": np.log(series("hpi", 1)) - np.log(series("hpi", 2)), # deepest reference is orig_time - 5: clamped iff it precedes the map "orig_macro_approx": (o - 5) < tmin, "orig_rate_proxy": orig_rate_missing, }, index=snapshot_df.index) return out # --------------------------------------------------------------------------- # vectorised competing-risk lifetime PD (exact factorisation; module docstring) # --------------------------------------------------------------------------- def _design_matrix(model: HazardModel, frame: pd.DataFrame) -> np.ndarray: """Training design matrix for new rows; NaN-safe (raises, never drops).""" prep = _prepare(frame) bad = int(prep[REQUIRED_COLS].isna().any(axis=1).sum()) if bad: raise ValueError( f"{bad} scoring rows carry NaN in required covariates " "(lag-warm-up quarter? staging needs time >= 6)") di = model.result.model.data.design_info X = np.asarray(build_design_matrices([di], prep)[0], dtype=float) if X.shape[0] != len(prep): # patsy NAAction would silently drop rows raise RuntimeError("design matrix dropped rows; misaligned scoring") return X def _spline_cols(model: HazardModel) -> list[int]: names = model.result.model.data.design_info.column_names return [i for i, c in enumerate(names) if c.startswith("cr(loan_age")] def _age_curve(model: HazardModel, max_age: int) -> np.ndarray: """s(a) = spline-basis contribution to the linear predictor, a=0..max_age. Non-age covariates are dummies: their design columns do not touch the cr(loan_age) columns, so any finite values give the identical s(a). """ ages = np.arange(max_age + 1, dtype=float) grid = pd.DataFrame({c: np.ones(len(ages)) for c in REQUIRED_COLS if c != "loan_age"}) grid["loan_age"] = ages X = _design_matrix(model, grid) idx = _spline_cols(model) beta = model.result.params.to_numpy() return X[:, idx] @ beta[idx] def _cum_default_pd(models: dict[str, HazardModel], frame: pd.DataFrame, horizons: np.ndarray) -> np.ndarray: """Cumulative competing-risk default PD over per-loan horizons. frame: one row per loan with REQUIRED_COLS; loan_age = age at the reporting quarter (projection starts at loan_age + 1). horizons: per-loan R >= 1 in quarters. Reproduces pd_term_structure's algebra exactly (per-period exit = lambda_d + lambda_p capped at _EXIT_CAP, survival cumprod, marginal = S(t-1) * lambda_d) via the linear-predictor factorisation in the module docstring. """ ages0 = frame["loan_age"].to_numpy() if not np.allclose(ages0, np.round(ages0)): raise ValueError("loan_age must be integral quarters") ages0 = np.round(ages0).astype(int) R = np.asarray(horizons, dtype=int) if (R < 1).any(): raise ValueError("horizons must be >= 1 (floor R upstream)") # the age matrix below indexes s[] up to ages0[i] + max(R) for EVERY loan # (masked cells included -- np.where evaluates both branches), so the # shared curve must extend to ages0.max() + R.max(), not (ages0+R).max() max_age = int(ages0.max() + R.max()) offsets, curves = {}, {} for name in ("default", "prepay"): m = models[name] beta = m.result.params.to_numpy() idx = _spline_cols(m) X = _design_matrix(m, frame) offsets[name] = X @ beta - X[:, idx] @ beta[idx] # age-free, per loan curves[name] = _age_curve(m, max_age) # shared seasoning # chunk over loans: peak memory is a few (chunk x max_R) float arrays out = np.empty(len(R), dtype=float) for lo in range(0, len(R), _CHUNK_LOANS): sl = slice(lo, min(lo + _CHUNK_LOANS, len(R))) a0, r = ages0[sl], R[sl] k = np.arange(int(r.max())) ages = a0[:, None] + 1 + k[None, :] # age in projected quarter on = k[None, :] < r[:, None] # inside the loan's window lam = {} for name in ("default", "prepay"): eta = offsets[name][sl][:, None] + curves[name][ages] # cloglog inverse saturates: exp overflow at extrapolated ages # just means lambda -> 1 (then capped); silence the benign warning with np.errstate(over="ignore"): lam[name] = np.where(on, 1.0 - np.exp(-np.exp(eta)), 0.0) exit_p = np.clip(lam["default"] + lam["prepay"], 0.0, _EXIT_CAP) surv = np.cumprod(1.0 - exit_p, axis=1) surv_prev = np.hstack([np.ones((len(r), 1)), surv[:, :-1]]) out[sl] = (surv_prev * lam["default"]).sum(axis=1) return out # --------------------------------------------------------------------------- # lifetime-PD table, trigger, stages # --------------------------------------------------------------------------- def lifetime_pd_table(panel_df: pd.DataFrame, models: dict[str, HazardModel], at_time: int, macro_map: pd.DataFrame | None = None, config: StagingConfig | None = None) -> pd.DataFrame: """Per-loan lifetime PD now vs at initial recognition, same window. One row per loan-quarter row of panel_df at time == at_time, in id order: R (effective window, floored at 1 and optionally capped), lifetime and annualised PDs on both legs, pd_ratio, ann_addon, approximation flags, and the snapshot's event flags. panel_df should carry enough history for the macro map (pass rows with time <= at_time; a pure snapshot frame degrades the origination-macro clamp, flagged per loan). """ cfg = config or StagingConfig() snap = (panel_df.loc[panel_df["time"] == at_time] .sort_values("id").reset_index(drop=True)) if snap.empty: raise ValueError(f"no rows at time {at_time}") if snap["id"].duplicated().any(): raise ValueError(f"duplicate loan ids at time {at_time}") mm = macro_map if macro_map is not None else build_macro_map(panel_df) R = np.maximum((snap["mat_time"] - at_time).to_numpy(), 1) if cfg.max_horizon is not None: R = np.minimum(R, cfg.max_horizon) orig = origination_covariates(snap, mm) pd_now = _cum_default_pd(models, snap, R) pd_orig = _cum_default_pd(models, orig, R) years = R / 4.0 ann_now = 1.0 - (1.0 - pd_now) ** (1.0 / years) ann_orig = 1.0 - (1.0 - pd_orig) ** (1.0 / years) out = pd.DataFrame({ "id": snap["id"].to_numpy(), "time": at_time, "loan_age": snap["loan_age"].to_numpy(), "R": R, "lifetime_pd_now": pd_now, "lifetime_pd_orig": pd_orig, "ann_pd_now": ann_now, "ann_pd_orig": ann_orig, # ratio conventions at the boundary: x/0 -> inf (infinite relative # deterioration), 0/0 -> 1 (no deterioration; degenerate one-quarter # windows where both hazards underflow) "pd_ratio": np.where(pd_orig > 0.0, pd_now / np.where(pd_orig > 0.0, pd_orig, 1.0), np.where(pd_now > 0.0, np.inf, 1.0)), "ann_addon": ann_now - ann_orig, "orig_macro_approx": orig["orig_macro_approx"].to_numpy(), "orig_rate_proxy": orig["orig_rate_proxy"].to_numpy(), "default_event": snap["default_event"].to_numpy(), "payoff_event": snap["payoff_event"].to_numpy(), }) # carry the backstop input: cfg.dpd_col FIRST (it is what # _backstop_30dpd reads under this config), plus the default-named # column if it also exists so a cached table restaged under the default # config still finds its input for col in dict.fromkeys([cfg.dpd_col, StagingConfig().dpd_col]): if col in snap.columns: out[col] = snap[col].to_numpy() return out def quantitative_sicr(pd_table: pd.DataFrame, config: StagingConfig | None = None) -> np.ndarray: """Ratio-AND-add-on quantitative SICR trigger (bool per table row).""" cfg = config or StagingConfig() return ((pd_table["pd_ratio"].to_numpy() > cfg.ratio_threshold) & (pd_table["ann_addon"].to_numpy() > cfg.abs_addon)) def _backstop_30dpd(pd_table: pd.DataFrame, cfg: StagingConfig) -> np.ndarray: """30-DPD rebuttable-presumption hook. INERT on DCR: no dpd column exists (no delinquency ladder), so this returns all-False today -- documented loudly in the module docstring.""" if cfg.backstop_30dpd and cfg.dpd_col in pd_table.columns: return pd_table[cfg.dpd_col].to_numpy() >= 30 return np.zeros(len(pd_table), dtype=bool) def assign_stages(panel_df: pd.DataFrame, models: dict[str, HazardModel], config: StagingConfig | None = None, snapshot_time: int | None = None, pd_tables: dict[int, pd.DataFrame] | None = None, ) -> pd.DataFrame: """IFRS 9 stage per live loan at the reporting quarter. Parameters ---------- panel_df : loan-quarter rows; must contain the snapshot quarter and should contain history (time <= snapshot) so the origination macro map and the probation lookback have data. NEVER uses rows after snapshot_time. models : {'default': HazardModel, 'prepay': HazardModel} (engine.hazard). config : StagingConfig (defaults = doubling + 0.5pp p.a. add-on, 2-quarter probation, inert 30-DPD hook). snapshot_time : reporting quarter; default = max(time) in panel_df. pd_tables : optional {quarter: lifetime_pd_table(...)} cache so threshold sweeps (governance sensitivity) do not recompute the PDs. Returns the snapshot lifetime-PD table plus: stage 1 / 2 / 3 trigger_reason 'default' | 'backstop_30dpd' | 'sicr_quantitative' | 'sicr_probation' (history trigger only) | 'none' """ cfg = config or StagingConfig() t = int(snapshot_time if snapshot_time is not None else panel_df["time"].max()) hist = panel_df.loc[panel_df["time"] <= t] window = [t - k for k in range(max(1, cfg.cure_quarters))] tables: dict[int, pd.DataFrame] = dict(pd_tables or {}) need = [u for u in window if u not in tables and (u == t or (hist["time"] == u).any())] if need: mm = build_macro_map(hist) for u in need: tables[u] = lifetime_pd_table(hist, models, u, mm, cfg) tbl = tables[t].copy() trig_now = quantitative_sicr(tbl, cfg) backstop = _backstop_30dpd(tbl, cfg) trig_hist = np.zeros(len(tbl), dtype=bool) for u in window: if u == t or u not in tables: continue tu = tables[u] fired = set(tu.loc[quantitative_sicr(tu, cfg), "id"]) trig_hist |= tbl["id"].isin(fired).to_numpy() is_default = tbl["default_event"].to_numpy() == 1 sicr = trig_now | trig_hist | backstop stage = np.where(is_default, 3, np.where(sicr, 2, 1)) reason = np.select( [is_default, backstop, trig_now, trig_hist], ["default", "backstop_30dpd", "sicr_quantitative", "sicr_probation"], default="none") tbl["stage"] = stage tbl["trigger_reason"] = reason return tbl # --------------------------------------------------------------------------- # exactness cross-check of the factorised fast path # --------------------------------------------------------------------------- def crosscheck_term_structure(panel_df: pd.DataFrame, models: dict[str, HazardModel], at_time: int, n_sample: int = 25, seed: int = 0) -> float: """Max |fast-path PD - pd_term_structure PD| over a fixed-seed sample. Re-derives lifetime_pd_now and lifetime_pd_orig for n_sample loans by calling engine.hazard.pd_term_structure loan-by-loan (the slow reference path) and compares against lifetime_pd_table. Deterministic (fixed seed). Expected ~1e-12 (identical algebra; module docstring PERFORMANCE STRATEGY). """ hist = panel_df.loc[panel_df["time"] <= at_time] mm = build_macro_map(hist) tbl = lifetime_pd_table(hist, models, at_time, mm) snap = (hist.loc[hist["time"] == at_time] .sort_values("id").reset_index(drop=True)) orig = origination_covariates(snap, mm) rng = np.random.default_rng(seed) pick = rng.choice(len(tbl), size=min(n_sample, len(tbl)), replace=False) worst = 0.0 for i in pick: horizon = int(tbl.loc[i, "R"]) for leg, frame in (("lifetime_pd_now", snap), ("lifetime_pd_orig", orig)): prof = frame.iloc[[i]][REQUIRED_COLS].copy() prof["loan_age"] = prof["loan_age"] + 1.0 # first projected qtr ts = pd_term_structure(models, prof, horizon) ref = float(ts["cum_pd"].iloc[-1]) worst = max(worst, abs(ref - float(tbl.loc[i, leg]))) return worst