Rating Agency Methodologies¶

import numpy as np
import pandas as pd

# Complete factor tables -- every notch must be present
MOODYS_FACTORS = {
    "Aaa": 1, "Aa1": 10, "Aa2": 10, "Aa3": 20,
    "A1": 70, "A2": 120, "A3": 180,
    "Baa1": 260, "Baa2": 260, "Baa3": 610,
    "Ba1": 940, "Ba2": 1350, "Ba3": 1766,
    "B1": 2220, "B2": 2720, "B3": 3490,
    "Caa1": 4770, "Caa2": 6500, "Caa3": 8070,
    "Ca": 10000, "C": 10000,
}

SP_FACTORS = {
    "AAA": 0.52, "AA+": 8, "AA": 8, "AA-": 15,
    "A+": 35, "A": 65, "A-": 120,
    "BBB+": 195, "BBB": 195, "BBB-": 437,
    "BB+": 776, "BB": 1106, "BB-": 1543,
    "B+": 2040, "B": 2556, "B-": 3214,
    "CCC+": 4108, "CCC": 5520, "CCC-": 7362,
    "CC": 10000, "C": 10000, "D": 10000,
}

FACTOR_TABLES = {
    "moodys": MOODYS_FACTORS,
    "sp": SP_FACTORS,
}

def calculate_warf(df: pd.DataFrame, rating_source: str = "moodys") -> float:
    """
    Calculate Weighted Average Rating Factor using vectorized operations.

    Args:
        df: DataFrame with 'par_value' and rating columns.
        rating_source: 'moodys' or 'sp' -- determines factor table and column.

    Returns:
        WARF as a float. Returns NaN if no valid ratings exist.
    """
    factors = FACTOR_TABLES[rating_source]
    rating_col = "rating_moodys" if rating_source == "moodys" else "rating_sp"

    # Map ratings to factors -- unmapped ratings become NaN
    df["_factor"] = df[rating_col].map(factors)

    # Warn on unmapped ratings (data quality issue)
    unmapped = df[df["_factor"].isna() & df[rating_col].notna()]
    if len(unmapped) > 0:
        unmapped_ratings = unmapped[rating_col].unique().tolist()
        raise ValueError(
            f"Unmapped {rating_source} ratings: {unmapped_ratings}. "
            f"Update the factor table or fix the input data."
        )

    # Vectorized WARF: weighted average of factors by par value
    valid = df.dropna(subset=["_factor", "par_value"])
    if valid["par_value"].sum() == 0:
        return float("nan")

    warf = (valid["_factor"] * valid["par_value"]).sum() / valid["par_value"].sum()

    # Clean up temporary column
    df.drop(columns=["_factor"], inplace=True)

    return float(warf)

# Industry concentration depends on classification system
def calculate_industry_concentration(
    df: pd.DataFrame,
    classification: str = "moodys_33"
) -> pd.DataFrame:
    """
    Calculate industry concentration using the specified classification.

    Args:
        df: DataFrame with 'par_value' and industry columns.
        classification: 'moodys_33' or 'gics' -- determines grouping column.

    Returns:
        DataFrame with industry name, par exposure, and concentration percentage.
    """
    industry_col = (
        "industry_moodys" if classification == "moodys_33"
        else "industry_gics_sector"
    )

    total_par = df["par_value"].sum()

    concentration = (
        df.groupby(industry_col)["par_value"]
        .sum()
        .reset_index()
        .rename(columns={"par_value": "par_exposure"})
    )
    concentration["concentration_pct"] = (
        concentration["par_exposure"] / total_par * 100
    )

    return concentration.sort_values("concentration_pct", ascending=False)

# Bidirectional cross-mapping
MOODYS_TO_SP = {
    "Aaa": "AAA", "Aa1": "AA+", "Aa2": "AA", "Aa3": "AA-",
    "A1": "A+", "A2": "A", "A3": "A-",
    "Baa1": "BBB+", "Baa2": "BBB", "Baa3": "BBB-",
    "Ba1": "BB+", "Ba2": "BB", "Ba3": "BB-",
    "B1": "B+", "B2": "B", "B3": "B-",
    "Caa1": "CCC+", "Caa2": "CCC", "Caa3": "CCC-",
    "Ca": "CC", "C": "C",
}

SP_TO_MOODYS = {v: k for k, v in MOODYS_TO_SP.items()}

# Numeric scale for ordinal comparison (lower = better quality)
NUMERIC_SCALE = {
    "Aaa": 1, "AAA": 1,
    "Aa1": 2, "AA+": 2,
    "Aa2": 3, "AA": 3,
    "Aa3": 4, "AA-": 4,
    "A1": 5, "A+": 5,
    "A2": 6, "A": 6,
    "A3": 7, "A-": 7,
    "Baa1": 8, "BBB+": 8,
    "Baa2": 9, "BBB": 9,
    "Baa3": 10, "BBB-": 10,
    "Ba1": 11, "BB+": 11,
    "Ba2": 12, "BB": 12,
    "Ba3": 13, "BB-": 13,
    "B1": 14, "B+": 14,
    "B2": 15, "B": 15,
    "B3": 16, "B-": 16,
    "Caa1": 17, "CCC+": 17,
    "Caa2": 18, "CCC": 18,
    "Caa3": 19, "CCC-": 19,
    "Ca": 20, "CC": 20,
    "C": 21, "D": 21,
}

def cross_map_rating(
    rating: str,
    from_agency: str,
    to_agency: str
) -> str | None:
    """
    Convert a rating from one agency's scale to another.
    Returns None if the rating is not recognized.
    """
    if from_agency == "moodys" and to_agency == "sp":
        return MOODYS_TO_SP.get(rating)
    elif from_agency == "sp" and to_agency == "moodys":
        return SP_TO_MOODYS.get(rating)
    return None

# In the compliance test suite configuration
{
    "rating": {
        "warf_rating_source": "moodys",  # or "sp"
        "warf_maximum": 2850,
        "factor_table_version": "2024-01",  # Factor tables can be updated
    }
}