Models API

creditriskengine.models.pd.scorecard

PD (Probability of Default) modeling framework.

Provides logistic regression scorecard, master scale construction, and PD calibration using the anchor point method.

References: - BCBS d350: Regulatory treatment of accounting provisions - EBA GL/2017/16: PD estimation, LGD estimation - Engelmann & Rauhmeier: The Basel II Risk Parameters (2nd ed.)

ScorecardBuilder

Bases: BaseEstimator, ClassifierMixin

Sklearn-compatible PD scorecard builder.

Implements fit/predict interface for logistic regression-based PD models. Follows sklearn estimator patterns per project spec.

Parameters:

Name	Type	Description	Default
base_score	float	Base scorecard score (default 600).	600.0
pdo	float	Points to double the odds (default 20).	20.0
base_odds	float	Odds at base score (default 50).	50.0

Source code in creditriskengine\models\pd\scorecard.py

class ScorecardBuilder(BaseEstimator, ClassifierMixin):  # type: ignore[misc]
    """Sklearn-compatible PD scorecard builder.

    Implements fit/predict interface for logistic regression-based PD models.
    Follows sklearn estimator patterns per project spec.

    Parameters:
        base_score: Base scorecard score (default 600).
        pdo: Points to double the odds (default 20).
        base_odds: Odds at base score (default 50).
    """

    def __init__(
        self,
        base_score: float = 600.0,
        pdo: float = 20.0,
        base_odds: float = 50.0,
    ) -> None:
        self.base_score = base_score
        self.pdo = pdo
        self.base_odds = base_odds
        # Fitted attributes (set in fit())
        self.intercept_: float | None = None
        self.coefficients_: np.ndarray | None = None
        self.master_scale_: list[dict[str, float]] | None = None
        self.is_fitted_: bool = False

    def fit(self, X: np.ndarray, y: np.ndarray) -> "ScorecardBuilder":  # noqa: N803
        """Fit logistic regression and build master scale.

        Uses sklearn LogisticRegression internally.
        """
        from sklearn.linear_model import LogisticRegression

        lr = LogisticRegression(penalty=None, solver="lbfgs", max_iter=1000)
        lr.fit(X, y)
        self.intercept_ = float(lr.intercept_[0])
        self.coefficients_ = lr.coef_[0].copy()
        self.is_fitted_ = True
        return self

    def predict_proba(self, X: np.ndarray) -> np.ndarray:  # noqa: N803
        """Predict PD (probability of default) for each observation."""
        assert self.is_fitted_, "Call fit() first"
        assert self.coefficients_ is not None and self.intercept_ is not None
        scores = logistic_score(self.coefficients_, X, self.intercept_)
        # Handle both 1D single-sample and 2D cases
        if scores.ndim == 0:
            scores = np.array([float(scores)])
        pds = score_to_pd(scores)
        return np.column_stack([1 - pds, pds])

    def predict(self, X: np.ndarray) -> np.ndarray:  # noqa: N803
        """Predict default (1) or non-default (0)."""
        proba = self.predict_proba(X)
        return (proba[:, 1] >= 0.5).astype(int)

    def score_points(self, X: np.ndarray) -> np.ndarray:  # noqa: N803
        """Convert to scorecard points."""
        assert self.is_fitted_, "Call fit() first"
        assert self.coefficients_ is not None and self.intercept_ is not None
        scores = logistic_score(self.coefficients_, X, self.intercept_)
        if scores.ndim == 0:
            scores = np.array([float(scores)])
        pds = score_to_pd(scores)
        return pd_to_score(pds, self.base_score, self.base_odds, self.pdo)

fit(X, y)

Fit logistic regression and build master scale.

Uses sklearn LogisticRegression internally.

Source code in creditriskengine\models\pd\scorecard.py

def fit(self, X: np.ndarray, y: np.ndarray) -> "ScorecardBuilder":  # noqa: N803
    """Fit logistic regression and build master scale.

    Uses sklearn LogisticRegression internally.
    """
    from sklearn.linear_model import LogisticRegression

    lr = LogisticRegression(penalty=None, solver="lbfgs", max_iter=1000)
    lr.fit(X, y)
    self.intercept_ = float(lr.intercept_[0])
    self.coefficients_ = lr.coef_[0].copy()
    self.is_fitted_ = True
    return self

predict_proba(X)

Predict PD (probability of default) for each observation.

Source code in creditriskengine\models\pd\scorecard.py

def predict_proba(self, X: np.ndarray) -> np.ndarray:  # noqa: N803
    """Predict PD (probability of default) for each observation."""
    assert self.is_fitted_, "Call fit() first"
    assert self.coefficients_ is not None and self.intercept_ is not None
    scores = logistic_score(self.coefficients_, X, self.intercept_)
    # Handle both 1D single-sample and 2D cases
    if scores.ndim == 0:
        scores = np.array([float(scores)])
    pds = score_to_pd(scores)
    return np.column_stack([1 - pds, pds])

predict(X)

Predict default (1) or non-default (0).

Source code in creditriskengine\models\pd\scorecard.py

def predict(self, X: np.ndarray) -> np.ndarray:  # noqa: N803
    """Predict default (1) or non-default (0)."""
    proba = self.predict_proba(X)
    return (proba[:, 1] >= 0.5).astype(int)

score_points(X)

Convert to scorecard points.

Source code in creditriskengine\models\pd\scorecard.py

def score_points(self, X: np.ndarray) -> np.ndarray:  # noqa: N803
    """Convert to scorecard points."""
    assert self.is_fitted_, "Call fit() first"
    assert self.coefficients_ is not None and self.intercept_ is not None
    scores = logistic_score(self.coefficients_, X, self.intercept_)
    if scores.ndim == 0:
        scores = np.array([float(scores)])
    pds = score_to_pd(scores)
    return pd_to_score(pds, self.base_score, self.base_odds, self.pdo)

logistic_score(coefficients, features, intercept=0.0)

Compute logistic regression log-odds scores.

Score = intercept + sum(coef_i * feature_i)

Parameters:

Name	Type	Description	Default
coefficients	ndarray	Model coefficients (length = n_features).	required
features	ndarray	Feature matrix (n_obs × n_features).	required
intercept	float	Model intercept.	0.0

Returns:

Type	Description
ndarray	Array of log-odds scores.

Source code in creditriskengine\models\pd\scorecard.py

def logistic_score(
    coefficients: np.ndarray,
    features: np.ndarray,
    intercept: float = 0.0,
) -> np.ndarray:
    """Compute logistic regression log-odds scores.

    Score = intercept + sum(coef_i * feature_i)

    Args:
        coefficients: Model coefficients (length = n_features).
        features: Feature matrix (n_obs × n_features).
        intercept: Model intercept.

    Returns:
        Array of log-odds scores.
    """
    features = np.asarray(features, dtype=np.float64)
    coefficients = np.asarray(coefficients, dtype=np.float64)
    return np.asarray(features @ coefficients + intercept)

score_to_pd(scores)

Convert log-odds scores to PD using the logistic function.

PD = 1 / (1 + exp(-score))

Parameters:

Name	Type	Description	Default
scores	ndarray	Log-odds scores.	required

Returns:

Type	Description
ndarray	Array of PDs in [0, 1].

Source code in creditriskengine\models\pd\scorecard.py

def score_to_pd(scores: np.ndarray) -> np.ndarray:
    """Convert log-odds scores to PD using the logistic function.

    PD = 1 / (1 + exp(-score))

    Args:
        scores: Log-odds scores.

    Returns:
        Array of PDs in [0, 1].
    """
    scores = np.asarray(scores, dtype=np.float64)
    return np.asarray(1.0 / (1.0 + np.exp(-scores)))

pd_to_score(pds, base_score=600.0, base_odds=50.0, pdo=20.0)

Convert PDs to scorecard points using the industry standard formula.

Score = base_score - pdo/ln(2) * ln(odds) where odds = (1-PD)/PD

Parameters:

Name	Type	Description	Default
pds	ndarray	Probability of default values.	required
base_score	float	Score at which odds = base_odds.	600.0
base_odds	float	Odds at the base score.	50.0
pdo	float	Points to double the odds.	20.0

Returns:

Type	Description
ndarray	Array of scorecard points.

Source code in creditriskengine\models\pd\scorecard.py

def pd_to_score(
    pds: np.ndarray,
    base_score: float = 600.0,
    base_odds: float = 50.0,
    pdo: float = 20.0,
) -> np.ndarray:
    """Convert PDs to scorecard points using the industry standard formula.

    Score = base_score - pdo/ln(2) * ln(odds)
    where odds = (1-PD)/PD

    Args:
        pds: Probability of default values.
        base_score: Score at which odds = base_odds.
        base_odds: Odds at the base score.
        pdo: Points to double the odds.

    Returns:
        Array of scorecard points.
    """
    pds = np.asarray(pds, dtype=np.float64)
    pds = np.clip(pds, 1e-10, 1.0 - 1e-10)
    odds = (1.0 - pds) / pds
    factor = pdo / math.log(2)
    offset = base_score - factor * math.log(base_odds)
    return np.asarray(offset + factor * np.log(odds))

scorecard_to_pd(scores, base_score=600.0, base_odds=50.0, pdo=20.0)

Convert scorecard points to PD (inverse of :func:pd_to_score).

Inverts the industry standard scorecard formula:

odds = exp((score - offset) / factor)
PD   = 1 / (1 + odds)

where factor = pdo / ln(2) and offset = base_score - factor * ln(base_odds).

Parameters:

Name	Type	Description	Default
scores	ndarray	Scorecard point values (e.g. 350, 500, 650, 800).	required
base_score	float	Score at which odds = base_odds.	600.0
base_odds	float	Odds at the base score.	50.0
pdo	float	Points to double the odds.	20.0

Returns:

Type	Description
ndarray	Array of PDs in (0, 1).

Source code in creditriskengine\models\pd\scorecard.py

def scorecard_to_pd(
    scores: np.ndarray,
    base_score: float = 600.0,
    base_odds: float = 50.0,
    pdo: float = 20.0,
) -> np.ndarray:
    """Convert scorecard points to PD (inverse of :func:`pd_to_score`).

    Inverts the industry standard scorecard formula:

        odds = exp((score - offset) / factor)
        PD   = 1 / (1 + odds)

    where ``factor = pdo / ln(2)`` and
    ``offset = base_score - factor * ln(base_odds)``.

    Args:
        scores: Scorecard point values (e.g. 350, 500, 650, 800).
        base_score: Score at which odds = base_odds.
        base_odds: Odds at the base score.
        pdo: Points to double the odds.

    Returns:
        Array of PDs in (0, 1).
    """
    scores = np.asarray(scores, dtype=np.float64)
    factor = pdo / math.log(2)
    offset = base_score - factor * math.log(base_odds)
    odds = np.exp((scores - offset) / factor)
    return np.asarray(1.0 / (1.0 + odds))

build_master_scale(grade_boundaries, grade_labels=None)

Build a master scale mapping PD ranges to rating grades.

Parameters:

Name	Type	Description	Default
grade_boundaries	list[float]	Sorted ascending PD boundary values. N boundaries produce N-1 grades. First boundary is the lower PD bound, last is the upper.	required
grade_labels	list[str] | None	Optional labels for each grade.	None

Returns:

Type	Description
list[dict[str, object]]	List of dicts with grade, pd_lower, pd_upper, pd_midpoint.

Source code in creditriskengine\models\pd\scorecard.py

def build_master_scale(
    grade_boundaries: list[float],
    grade_labels: list[str] | None = None,
) -> list[dict[str, object]]:
    """Build a master scale mapping PD ranges to rating grades.

    Args:
        grade_boundaries: Sorted ascending PD boundary values.
            N boundaries produce N-1 grades.
            First boundary is the lower PD bound, last is the upper.
        grade_labels: Optional labels for each grade.

    Returns:
        List of dicts with grade, pd_lower, pd_upper, pd_midpoint.
    """
    if len(grade_boundaries) < 2:
        raise ValueError("Need at least 2 boundaries to define 1 grade")

    n_grades = len(grade_boundaries) - 1
    if grade_labels is None:
        grade_labels = [f"Grade_{i + 1}" for i in range(n_grades)]

    if len(grade_labels) != n_grades:
        raise ValueError(
            f"Expected {n_grades} labels, got {len(grade_labels)}"
        )

    scale = []
    for i in range(n_grades):
        lo = grade_boundaries[i]
        hi = grade_boundaries[i + 1]
        mid = math.sqrt(lo * hi) if lo > 0 else hi / 2.0
        scale.append({
            "grade": grade_labels[i],
            "pd_lower": lo,
            "pd_upper": hi,
            "pd_midpoint": mid,
        })
    return scale

assign_rating_grade(pd, master_scale)

Assign a rating grade based on PD and master scale.

Parameters:

Name	Type	Description	Default
pd	float	Probability of default.	required
master_scale	list[dict[str, object]]	Master scale from build_master_scale().	required

Returns:

Type	Description
str	Grade label.

Source code in creditriskengine\models\pd\scorecard.py

def assign_rating_grade(
    pd: float,
    master_scale: list[dict[str, object]],
) -> str:
    """Assign a rating grade based on PD and master scale.

    Args:
        pd: Probability of default.
        master_scale: Master scale from build_master_scale().

    Returns:
        Grade label.
    """
    for grade in master_scale:
        if grade["pd_lower"] <= pd < grade["pd_upper"]:  # type: ignore[operator]
            return str(grade["grade"])
    # Default to last grade if PD >= last upper boundary
    return str(master_scale[-1]["grade"])

calibrate_pd_anchor_point(central_tendency, raw_pds)

Calibrate PDs using the anchor-point method.

Scales raw (model) PDs so their portfolio-weighted average matches the long-run central tendency.

Formula

calibrated_PD_i = raw_PD_i × (central_tendency / mean(raw_PDs))

Parameters:

Name	Type	Description	Default
central_tendency	float	Long-run average default rate.	required
raw_pds	ndarray	Raw/uncalibrated PD estimates.	required

Returns:

Type	Description
ndarray	Calibrated PD array, clipped to [PD_floor, 1.0].

Source code in creditriskengine\models\pd\scorecard.py

def calibrate_pd_anchor_point(
    central_tendency: float,
    raw_pds: np.ndarray,
) -> np.ndarray:
    """Calibrate PDs using the anchor-point method.

    Scales raw (model) PDs so their portfolio-weighted average
    matches the long-run central tendency.

    Formula:
        calibrated_PD_i = raw_PD_i × (central_tendency / mean(raw_PDs))

    Args:
        central_tendency: Long-run average default rate.
        raw_pds: Raw/uncalibrated PD estimates.

    Returns:
        Calibrated PD array, clipped to [PD_floor, 1.0].
    """
    raw_pds = np.asarray(raw_pds, dtype=np.float64)
    avg = float(np.mean(raw_pds))
    if avg < 1e-15:
        return np.full_like(raw_pds, central_tendency)
    scaling = central_tendency / avg
    calibrated = raw_pds * scaling
    # Basel III PD floor 3 bps per CRE32.13
    return np.clip(calibrated, 0.0003, 1.0)

calibrate_pd_bayesian(prior_pd, observed_defaults, n_observations, weight=0.5)

Bayesian PD calibration combining prior with observed data.

PD_calibrated = weight × prior_PD + (1-weight) × observed_DR

Parameters:

Name	Type	Description	Default
prior_pd	float	Prior (long-run) PD estimate.	required
observed_defaults	int	Number of observed defaults.	required
n_observations	int	Number of observations.	required
weight	float	Weight on prior (0 = pure data, 1 = pure prior).	0.5

Returns:

Type	Description
float	Calibrated PD.

Source code in creditriskengine\models\pd\scorecard.py

def calibrate_pd_bayesian(
    prior_pd: float,
    observed_defaults: int,
    n_observations: int,
    weight: float = 0.5,
) -> float:
    """Bayesian PD calibration combining prior with observed data.

    PD_calibrated = weight × prior_PD + (1-weight) × observed_DR

    Args:
        prior_pd: Prior (long-run) PD estimate.
        observed_defaults: Number of observed defaults.
        n_observations: Number of observations.
        weight: Weight on prior (0 = pure data, 1 = pure prior).

    Returns:
        Calibrated PD.
    """
    if n_observations <= 0:
        return prior_pd
    observed_dr = observed_defaults / n_observations
    return weight * prior_pd + (1.0 - weight) * observed_dr

vasicek_single_factor_pd(pd_ttc, rho, confidence=0.999)

Conditional PD under the Vasicek single-factor model.

PD_conditional = Phi( (Phi^-1(PD) + sqrt(rho)*Phi^-1(confidence)) / sqrt(1-rho) )

This is the stressed/downturn PD used in regulatory capital.

Parameters:

Name	Type	Description	Default
pd_ttc	float	Through-the-cycle PD.	required
rho	float	Asset correlation.	required
confidence	float	Confidence level (default 99.9%).	0.999

Returns:

Type	Description
float	Conditional (stressed) PD.

Source code in creditriskengine\models\pd\scorecard.py

def vasicek_single_factor_pd(
    pd_ttc: float,
    rho: float,
    confidence: float = 0.999,
) -> float:
    """Conditional PD under the Vasicek single-factor model.

    PD_conditional = Phi( (Phi^-1(PD) + sqrt(rho)*Phi^-1(confidence)) / sqrt(1-rho) )

    This is the stressed/downturn PD used in regulatory capital.

    Args:
        pd_ttc: Through-the-cycle PD.
        rho: Asset correlation.
        confidence: Confidence level (default 99.9%).

    Returns:
        Conditional (stressed) PD.
    """
    if pd_ttc <= 0:
        return 0.0
    if pd_ttc >= 1:
        return 1.0
    g_pd = norm.ppf(pd_ttc)
    g_conf = norm.ppf(confidence)
    numerator = g_pd + math.sqrt(rho) * g_conf
    denominator = math.sqrt(1.0 - rho)
    return float(norm.cdf(numerator / denominator))

creditriskengine.models.lgd.lgd_model

LGD (Loss Given Default) modeling framework.

Provides workout LGD estimation, downturn LGD adjustment, and LGD curve generation.

References: - EBA GL/2017/16: LGD estimation - BCBS d424: CRE36 (LGD under IRB) - BCBS CRE32.22-32.24: Supervisory LGD values

LGDModel

Bases: BaseEstimator, RegressorMixin

Sklearn-compatible LGD model.

Wraps workout LGD estimation with fit/predict interface.

Parameters:

Name	Type	Description	Default
method	str	LGD estimation method ('workout', 'downturn', 'regulatory').	'workout'
downturn_method	str	Downturn LGD method ('additive', 'haircut', 'regulatory_formula').	'regulatory_formula'
downturn_add_on	float	Additive stress for downturn LGD.	0.1
collateral_type	str	For regulatory LGD floor application.	'unsecured'

Source code in creditriskengine\models\lgd\lgd_model.py

class LGDModel(BaseEstimator, RegressorMixin):  # type: ignore[misc]
    """Sklearn-compatible LGD model.

    Wraps workout LGD estimation with fit/predict interface.

    Parameters:
        method: LGD estimation method ('workout', 'downturn', 'regulatory').
        downturn_method: Downturn LGD method ('additive', 'haircut', 'regulatory_formula').
        downturn_add_on: Additive stress for downturn LGD.
        collateral_type: For regulatory LGD floor application.
    """

    def __init__(
        self,
        method: str = "workout",
        downturn_method: str = "regulatory_formula",
        downturn_add_on: float = 0.10,
        collateral_type: str = "unsecured",
    ) -> None:
        self.method = method
        self.downturn_method = downturn_method
        self.downturn_add_on = downturn_add_on
        self.collateral_type = collateral_type
        self.mean_lgd_: float | None = None
        self.is_fitted_: bool = False

    def fit(self, X: np.ndarray, y: np.ndarray) -> "LGDModel":  # noqa: N803
        """Fit LGD model. X = features, y = realized LGDs.

        Stores mean LGD for baseline predictions.
        """
        y = np.asarray(y, dtype=np.float64)
        self.mean_lgd_ = float(np.mean(y))
        self.is_fitted_ = True
        return self

    def predict(self, X: np.ndarray) -> np.ndarray:  # noqa: N803
        """Predict LGD values. Returns mean LGD (baseline model).

        Override in subclasses for more sophisticated models.
        """
        assert self.is_fitted_, "Call fit() first"
        n = X.shape[0] if hasattr(X, "shape") else len(X)
        base = np.full(n, self.mean_lgd_)
        if self.method == "downturn":
            base = np.array([
                downturn_lgd(lgd, self.downturn_add_on, self.downturn_method)
                for lgd in base
            ])
        is_secured = self.collateral_type != "unsecured"
        base = np.array([
            apply_lgd_floor(lgd, is_secured, self.collateral_type)
            for lgd in base
        ])
        return base

fit(X, y)

Fit LGD model. X = features, y = realized LGDs.

Stores mean LGD for baseline predictions.

Source code in creditriskengine\models\lgd\lgd_model.py

def fit(self, X: np.ndarray, y: np.ndarray) -> "LGDModel":  # noqa: N803
    """Fit LGD model. X = features, y = realized LGDs.

    Stores mean LGD for baseline predictions.
    """
    y = np.asarray(y, dtype=np.float64)
    self.mean_lgd_ = float(np.mean(y))
    self.is_fitted_ = True
    return self

predict(X)

Predict LGD values. Returns mean LGD (baseline model).

Override in subclasses for more sophisticated models.

Source code in creditriskengine\models\lgd\lgd_model.py

def predict(self, X: np.ndarray) -> np.ndarray:  # noqa: N803
    """Predict LGD values. Returns mean LGD (baseline model).

    Override in subclasses for more sophisticated models.
    """
    assert self.is_fitted_, "Call fit() first"
    n = X.shape[0] if hasattr(X, "shape") else len(X)
    base = np.full(n, self.mean_lgd_)
    if self.method == "downturn":
        base = np.array([
            downturn_lgd(lgd, self.downturn_add_on, self.downturn_method)
            for lgd in base
        ])
    is_secured = self.collateral_type != "unsecured"
    base = np.array([
        apply_lgd_floor(lgd, is_secured, self.collateral_type)
        for lgd in base
    ])
    return base

workout_lgd(ead_at_default, total_recoveries, total_costs, discount_rate=0.0, time_to_recovery_years=0.0)

Calculate workout LGD from realized recovery data.

LGD = 1 - (Recoveries - Costs) / EAD, discounted to default date.

Parameters:

Name	Type	Description	Default
ead_at_default	float	Exposure at the time of default.	required
total_recoveries	float	Total gross recoveries.	required
total_costs	float	Total workout/collection costs.	required
discount_rate	float	Annual discount rate for time value.	0.0
time_to_recovery_years	float	Average time from default to recovery.	0.0

Returns:

Type	Description
float	Workout LGD in [0, 1].

Source code in creditriskengine\models\lgd\lgd_model.py

def workout_lgd(
    ead_at_default: float,
    total_recoveries: float,
    total_costs: float,
    discount_rate: float = 0.0,
    time_to_recovery_years: float = 0.0,
) -> float:
    """Calculate workout LGD from realized recovery data.

    LGD = 1 - (Recoveries - Costs) / EAD, discounted to default date.

    Args:
        ead_at_default: Exposure at the time of default.
        total_recoveries: Total gross recoveries.
        total_costs: Total workout/collection costs.
        discount_rate: Annual discount rate for time value.
        time_to_recovery_years: Average time from default to recovery.

    Returns:
        Workout LGD in [0, 1].
    """
    if ead_at_default <= 0:
        return 1.0
    net_recovery = total_recoveries - total_costs
    if discount_rate > 0 and time_to_recovery_years > 0:
        discount_factor = 1.0 / (1.0 + discount_rate) ** time_to_recovery_years
        net_recovery *= discount_factor
    lgd = 1.0 - net_recovery / ead_at_default
    return float(np.clip(lgd, 0.0, 1.0))

downturn_lgd(base_lgd, downturn_add_on=None, method='additive')

Estimate downturn LGD per EBA GL/2017/16 Section 6.3.

Functions:

Name	Description
- "additive"	LGD_dt = base_LGD + add_on
- "haircut"	LGD_dt = 1 - (1 - base_LGD) × (1 - add_on)
- "regulatory"	Use BCBS formula max(base_LGD, 0.10 + 0.40 × base_LGD)

Parameters:

Name	Type	Description	Default
base_lgd	float	Through-the-cycle / long-run average LGD.	required
downturn_add_on	float | None	Downturn add-on (for additive/haircut).	None
method	str	Estimation method.	'additive'

Returns:

Type	Description
float	Downturn LGD, clipped to [0, 1].

Source code in creditriskengine\models\lgd\lgd_model.py

def downturn_lgd(
    base_lgd: float,
    downturn_add_on: float | None = None,
    method: str = "additive",
) -> float:
    """Estimate downturn LGD per EBA GL/2017/16 Section 6.3.

    Methods:
        - "additive": LGD_dt = base_LGD + add_on
        - "haircut": LGD_dt = 1 - (1 - base_LGD) × (1 - add_on)
        - "regulatory": Use BCBS formula max(base_LGD, 0.10 + 0.40 × base_LGD)

    Args:
        base_lgd: Through-the-cycle / long-run average LGD.
        downturn_add_on: Downturn add-on (for additive/haircut).
        method: Estimation method.

    Returns:
        Downturn LGD, clipped to [0, 1].
    """
    if method == "regulatory":
        dt_lgd = max(base_lgd, 0.10 + 0.40 * base_lgd)
    elif method == "haircut":
        add_on = downturn_add_on if downturn_add_on is not None else 0.08
        dt_lgd = 1.0 - (1.0 - base_lgd) * (1.0 - add_on)
    else:  # additive
        add_on = downturn_add_on if downturn_add_on is not None else 0.08
        dt_lgd = base_lgd + add_on

    return float(np.clip(dt_lgd, 0.0, 1.0))

lgd_term_structure(base_lgd, n_periods, recovery_curve=None)

Generate LGD term structure over multiple periods.

If a recovery curve is provided, LGD(t) = 1 - cumulative_recovery(t). Otherwise returns flat LGD.

Parameters:

Name	Type	Description	Default
base_lgd	float	Base LGD estimate.	required
n_periods	int	Number of periods.	required
recovery_curve	ndarray | None	Optional cumulative recovery rates by period.	None

Returns:

Type	Description
ndarray	Array of LGD values by period.

Source code in creditriskengine\models\lgd\lgd_model.py

def lgd_term_structure(
    base_lgd: float,
    n_periods: int,
    recovery_curve: np.ndarray | None = None,
) -> np.ndarray:
    """Generate LGD term structure over multiple periods.

    If a recovery curve is provided, LGD(t) = 1 - cumulative_recovery(t).
    Otherwise returns flat LGD.

    Args:
        base_lgd: Base LGD estimate.
        n_periods: Number of periods.
        recovery_curve: Optional cumulative recovery rates by period.

    Returns:
        Array of LGD values by period.
    """
    if recovery_curve is not None:
        recovery_curve = np.asarray(recovery_curve, dtype=np.float64)
        lgds = 1.0 - recovery_curve[:n_periods]
        # Pad if recovery_curve shorter than n_periods
        if len(lgds) < n_periods:
            pad_value = lgds[-1] if len(lgds) > 0 else base_lgd
            lgds = np.concatenate([lgds, np.full(n_periods - len(lgds), pad_value)])
        return np.clip(lgds, 0.0, 1.0)
    return np.full(n_periods, base_lgd)

apply_lgd_floor(lgd, is_secured=False, collateral_type='unsecured')

Apply Basel III A-IRB LGD floors per CRE32.25.

Parameters:

Name	Type	Description	Default
lgd	float	Estimated LGD.	required
is_secured	bool	Whether exposure is secured.	False
collateral_type	str	Type of collateral.	'unsecured'

Returns:

Type	Description
float	Floored LGD.

Source code in creditriskengine\models\lgd\lgd_model.py

def apply_lgd_floor(
    lgd: float,
    is_secured: bool = False,
    collateral_type: str = "unsecured",
) -> float:
    """Apply Basel III A-IRB LGD floors per CRE32.25.

    Args:
        lgd: Estimated LGD.
        is_secured: Whether exposure is secured.
        collateral_type: Type of collateral.

    Returns:
        Floored LGD.
    """
    if not is_secured:
        floor = LGD_FLOOR_UNSECURED
    elif collateral_type == "financial":
        floor = LGD_FLOOR_SECURED_FINANCIAL
    elif collateral_type == "receivables":
        floor = LGD_FLOOR_SECURED_RECEIVABLES
    elif collateral_type in ("cre", "rre"):
        floor = LGD_FLOOR_SECURED_CRE_RRE
    else:
        floor = LGD_FLOOR_SECURED_OTHER
    return max(lgd, floor)

creditriskengine.models.ead.ead_model

EAD (Exposure at Default) modeling framework.

Provides CCF estimation, regulatory EAD calculation, and EAD term structure generation for off-balance sheet exposures.

References: - BCBS d424: CRE31 (EAD under IRB), CRE20 (EAD under SA) - CRE32.26-32.32: CCF parameters - EBA GL/2017/16: EAD estimation for A-IRB

EADModel

Bases: BaseEstimator, RegressorMixin

Sklearn-compatible EAD model.

Wraps EAD/CCF estimation with fit/predict interface.

Parameters:

Name	Type	Description	Default
ccf_method	str	CCF estimation method ('supervisory', 'estimated').	'supervisory'
facility_type	str	For supervisory CCF lookup.	'committed_other'

Source code in creditriskengine\models\ead\ead_model.py

class EADModel(BaseEstimator, RegressorMixin):  # type: ignore[misc]
    """Sklearn-compatible EAD model.

    Wraps EAD/CCF estimation with fit/predict interface.

    Parameters:
        ccf_method: CCF estimation method ('supervisory', 'estimated').
        facility_type: For supervisory CCF lookup.
    """

    def __init__(
        self,
        ccf_method: str = "supervisory",
        facility_type: str = "committed_other",
    ) -> None:
        self.ccf_method = ccf_method
        self.facility_type = facility_type
        self.mean_ccf_: float | None = None
        self.is_fitted_: bool = False

    def fit(self, X: np.ndarray, y: np.ndarray) -> "EADModel":  # noqa: N803
        """Fit EAD model. X = [drawn, undrawn], y = realized EAD."""
        y = np.asarray(y, dtype=np.float64)
        X = np.asarray(X, dtype=np.float64)  # noqa: N806
        # Estimate mean CCF from data
        if X.shape[1] >= 2:
            undrawn = X[:, 1]
            drawn = X[:, 0]
            mask = undrawn > 0
            if mask.any():
                ccfs = (y[mask] - drawn[mask]) / undrawn[mask]
                self.mean_ccf_ = float(np.clip(np.mean(ccfs), 0, 1))
            else:
                self.mean_ccf_ = 0.75  # Default
        else:
            self.mean_ccf_ = 0.75
        self.is_fitted_ = True
        return self

    def predict(self, X: np.ndarray) -> np.ndarray:  # noqa: N803
        """Predict EAD. X columns: [drawn, undrawn]."""
        assert self.is_fitted_, "Call fit() first"
        X = np.asarray(X, dtype=np.float64)  # noqa: N806
        if self.ccf_method == "supervisory":
            ccf = get_supervisory_ccf(self.facility_type)
        else:
            ccf = self.mean_ccf_ or 0.75
        return np.array([calculate_ead(row[0], row[1], ccf) for row in X])

fit(X, y)

Fit EAD model. X = [drawn, undrawn], y = realized EAD.

Source code in creditriskengine\models\ead\ead_model.py

def fit(self, X: np.ndarray, y: np.ndarray) -> "EADModel":  # noqa: N803
    """Fit EAD model. X = [drawn, undrawn], y = realized EAD."""
    y = np.asarray(y, dtype=np.float64)
    X = np.asarray(X, dtype=np.float64)  # noqa: N806
    # Estimate mean CCF from data
    if X.shape[1] >= 2:
        undrawn = X[:, 1]
        drawn = X[:, 0]
        mask = undrawn > 0
        if mask.any():
            ccfs = (y[mask] - drawn[mask]) / undrawn[mask]
            self.mean_ccf_ = float(np.clip(np.mean(ccfs), 0, 1))
        else:
            self.mean_ccf_ = 0.75  # Default
    else:
        self.mean_ccf_ = 0.75
    self.is_fitted_ = True
    return self

predict(X)

Predict EAD. X columns: [drawn, undrawn].

Source code in creditriskengine\models\ead\ead_model.py

def predict(self, X: np.ndarray) -> np.ndarray:  # noqa: N803
    """Predict EAD. X columns: [drawn, undrawn]."""
    assert self.is_fitted_, "Call fit() first"
    X = np.asarray(X, dtype=np.float64)  # noqa: N806
    if self.ccf_method == "supervisory":
        ccf = get_supervisory_ccf(self.facility_type)
    else:
        ccf = self.mean_ccf_ or 0.75
    return np.array([calculate_ead(row[0], row[1], ccf) for row in X])

calculate_ead(drawn_amount, undrawn_commitment, ccf)

Calculate Exposure at Default.

EAD = Drawn + CCF × Undrawn

Parameters:

Name	Type	Description	Default
drawn_amount	float	Current outstanding balance.	required
undrawn_commitment	float	Undrawn committed amount.	required
ccf	float	Credit Conversion Factor.	required

Returns:

Type	Description
float	EAD value.

Source code in creditriskengine\models\ead\ead_model.py

def calculate_ead(
    drawn_amount: float,
    undrawn_commitment: float,
    ccf: float,
) -> float:
    """Calculate Exposure at Default.

    EAD = Drawn + CCF × Undrawn

    Args:
        drawn_amount: Current outstanding balance.
        undrawn_commitment: Undrawn committed amount.
        ccf: Credit Conversion Factor.

    Returns:
        EAD value.
    """
    return drawn_amount + ccf * undrawn_commitment

estimate_ccf(ead_at_default, drawn_at_reference, limit)

Estimate realized CCF from default observation.

CCF = (EAD - Drawn_ref) / (Limit - Drawn_ref)

Parameters:

Name	Type	Description	Default
ead_at_default	float	Actual EAD observed at default.	required
drawn_at_reference	float	Drawn amount at reference date (12m before default).	required
limit	float	Total committed limit.	required

Returns:

Type	Description
float	Realized CCF, clipped to [0, 1].

Source code in creditriskengine\models\ead\ead_model.py

def estimate_ccf(
    ead_at_default: float,
    drawn_at_reference: float,
    limit: float,
) -> float:
    """Estimate realized CCF from default observation.

    CCF = (EAD - Drawn_ref) / (Limit - Drawn_ref)

    Args:
        ead_at_default: Actual EAD observed at default.
        drawn_at_reference: Drawn amount at reference date (12m before default).
        limit: Total committed limit.

    Returns:
        Realized CCF, clipped to [0, 1].
    """
    undrawn = limit - drawn_at_reference
    if undrawn <= 0:
        return 1.0
    ccf = (ead_at_default - drawn_at_reference) / undrawn
    return float(np.clip(ccf, 0.0, 1.0))

get_supervisory_ccf(facility_type)

Get the F-IRB supervisory CCF for a facility type.

Parameters:

Name	Type	Description	Default
facility_type	str	Facility type key (see SUPERVISORY_CCFS).	required

Returns:

Type	Description
float	Supervisory CCF value.

Raises:

Type	Description
ValueError	If facility type is unknown.

Source code in creditriskengine\models\ead\ead_model.py

def get_supervisory_ccf(facility_type: str) -> float:
    """Get the F-IRB supervisory CCF for a facility type.

    Args:
        facility_type: Facility type key (see SUPERVISORY_CCFS).

    Returns:
        Supervisory CCF value.

    Raises:
        ValueError: If facility type is unknown.
    """
    ccf = SUPERVISORY_CCFS.get(facility_type)
    if ccf is None:
        raise ValueError(
            f"Unknown facility type: {facility_type}. "
            f"Valid types: {list(SUPERVISORY_CCFS.keys())}"
        )
    return ccf

apply_ccf_floor(ccf, approach='airb')

Apply regulatory CCF floor.

A-IRB floor per CRE32.33: CCF >= 50% for revolving. F-IRB: supervisory values are fixed, no floor needed.

Parameters:

Name	Type	Description	Default
ccf	float	Estimated CCF.	required
approach	str	"airb" or "firb".	'airb'

Returns:

Type	Description
float	Floored CCF.

Source code in creditriskengine\models\ead\ead_model.py

def apply_ccf_floor(ccf: float, approach: str = "airb") -> float:
    """Apply regulatory CCF floor.

    A-IRB floor per CRE32.33: CCF >= 50% for revolving.
    F-IRB: supervisory values are fixed, no floor needed.

    Args:
        ccf: Estimated CCF.
        approach: "airb" or "firb".

    Returns:
        Floored CCF.
    """
    if approach == "airb":
        return max(ccf, CCF_FLOOR_AIRB)
    return ccf

ead_term_structure(drawn_amount, undrawn_commitment, ccf, n_periods, amortization_rate=0.0)

Generate EAD term structure with optional amortization.

Parameters:

Name	Type	Description	Default
drawn_amount	float	Current drawn amount.	required
undrawn_commitment	float	Undrawn commitment.	required
ccf	float	Credit conversion factor.	required
n_periods	int	Number of periods.	required
amortization_rate	float	Annual amortization rate for the drawn amount.	0.0

Returns:

Type	Description
ndarray	Array of EAD values by period.

Source code in creditriskengine\models\ead\ead_model.py

def ead_term_structure(
    drawn_amount: float,
    undrawn_commitment: float,
    ccf: float,
    n_periods: int,
    amortization_rate: float = 0.0,
) -> np.ndarray:
    """Generate EAD term structure with optional amortization.

    Args:
        drawn_amount: Current drawn amount.
        undrawn_commitment: Undrawn commitment.
        ccf: Credit conversion factor.
        n_periods: Number of periods.
        amortization_rate: Annual amortization rate for the drawn amount.

    Returns:
        Array of EAD values by period.
    """
    eads = np.empty(n_periods, dtype=np.float64)
    for t in range(n_periods):
        remaining_drawn = drawn_amount * (1.0 - amortization_rate) ** t
        ead = remaining_drawn + ccf * undrawn_commitment
        eads[t] = max(ead, 0.0)
    return eads

creditriskengine.models.concentration.concentration

Concentration risk analytics.

Single-name concentration, sector/geographic HHI, and the Granularity Adjustment (GA) per BCBS.

References: - BCBS d424: Pillar 2 concentration risk - Gordy (2003): A risk-factor model foundation for ratings-based capital rules - Gordy & Lütkebohmert (2013): Granularity adjustment for regulatory capital

single_name_concentration(eads)

Measure single-name concentration in a portfolio.

Returns HHI and top-N exposure shares.

Parameters:

Name	Type	Description	Default
eads	ndarray	EAD per obligor.	required

Returns:

Type	Description
dict[str, float]	Dict with hhi, top_1_share, top_5_share, top_10_share, n_obligors.

Source code in creditriskengine\models\concentration\concentration.py

def single_name_concentration(
    eads: np.ndarray,
) -> dict[str, float]:
    """Measure single-name concentration in a portfolio.

    Returns HHI and top-N exposure shares.

    Args:
        eads: EAD per obligor.

    Returns:
        Dict with hhi, top_1_share, top_5_share, top_10_share, n_obligors.
    """
    eads = np.asarray(eads, dtype=np.float64)
    total = float(np.sum(eads))
    if total <= 0:
        return {
            "hhi": 0.0,
            "top_1_share": 0.0,
            "top_5_share": 0.0,
            "top_10_share": 0.0,
            "n_obligors": 0,
        }

    shares = eads / total
    hhi = float(np.sum(shares ** 2))

    sorted_eads = np.sort(eads)[::-1]
    n = len(eads)

    top_1 = float(sorted_eads[0] / total) if n >= 1 else 0.0
    top_5 = float(np.sum(sorted_eads[:5]) / total) if n >= 5 else float(np.sum(sorted_eads) / total)
    top_10 = (
        float(np.sum(sorted_eads[:10]) / total) if n >= 10
        else float(np.sum(sorted_eads) / total)
    )

    return {
        "hhi": hhi,
        "top_1_share": top_1,
        "top_5_share": top_5,
        "top_10_share": top_10,
        "n_obligors": n,
    }

sector_concentration(eads, sector_labels)

Measure sector concentration via HHI.

Parameters:

Name	Type	Description	Default
eads	ndarray	EAD per exposure.	required
sector_labels	ndarray	Sector label per exposure.	required

Returns:

Type	Description
dict[str, float | dict[str, float]]	Dict with sector_hhi and sector_shares.

Source code in creditriskengine\models\concentration\concentration.py

def sector_concentration(
    eads: np.ndarray,
    sector_labels: np.ndarray,
) -> dict[str, float | dict[str, float]]:
    """Measure sector concentration via HHI.

    Args:
        eads: EAD per exposure.
        sector_labels: Sector label per exposure.

    Returns:
        Dict with sector_hhi and sector_shares.
    """
    eads = np.asarray(eads, dtype=np.float64)
    total = float(np.sum(eads))
    if total <= 0:
        return {"sector_hhi": 0.0, "sector_shares": {}}

    unique_sectors = np.unique(sector_labels)
    shares: dict[str, float] = {}
    for sector in unique_sectors:
        mask = sector_labels == sector
        shares[str(sector)] = float(np.sum(eads[mask]) / total)

    hhi = sum(s ** 2 for s in shares.values())
    return {"sector_hhi": hhi, "sector_shares": shares}

granularity_adjustment(eads, pds, lgds, rho)

Gordy (2003) Granularity Adjustment.

GA = (1/2) × C_3 × HHI_adj

Where C_3 captures the curvature of the loss distribution and HHI_adj is the EAD-weighted HHI of PD×LGD contributions.

Simplified single-factor version for Pillar 2 add-on estimation.

Parameters:

Name	Type	Description	Default
eads	ndarray	EAD per obligor.	required
pds	ndarray	PD per obligor.	required
lgds	ndarray	LGD per obligor.	required
rho	float	Common asset correlation.	required

Returns:

Type	Description
float	Granularity adjustment as a fraction of total EAD.

Source code in creditriskengine\models\concentration\concentration.py

def granularity_adjustment(
    eads: np.ndarray,
    pds: np.ndarray,
    lgds: np.ndarray,
    rho: float,
) -> float:
    """Gordy (2003) Granularity Adjustment.

    GA = (1/2) × C_3 × HHI_adj

    Where C_3 captures the curvature of the loss distribution
    and HHI_adj is the EAD-weighted HHI of PD×LGD contributions.

    Simplified single-factor version for Pillar 2 add-on estimation.

    Args:
        eads: EAD per obligor.
        pds: PD per obligor.
        lgds: LGD per obligor.
        rho: Common asset correlation.

    Returns:
        Granularity adjustment as a fraction of total EAD.
    """
    eads = np.asarray(eads, dtype=np.float64)
    pds = np.asarray(pds, dtype=np.float64)
    lgds = np.asarray(lgds, dtype=np.float64)

    total_ead = float(np.sum(eads))
    if total_ead <= 0:
        return 0.0

    # EAD-weighted expected loss contributions
    el_contributions = eads * pds * lgds
    total_el = float(np.sum(el_contributions))
    if total_el <= 0:
        return 0.0

    # HHI of loss contributions
    shares = el_contributions / total_el
    hhi = float(np.sum(shares ** 2))

    # Simplified GA: proportional to HHI and variance
    # Higher correlation → less idiosyncratic risk → smaller GA
    idiosyncratic_factor = (1.0 - rho)
    ga = 0.5 * hhi * idiosyncratic_factor * total_el / total_ead

    return ga