Loan Pricing and Compliance Impact¶

import numpy as np
import pandas as pd


def calculate_oc_numerator(
    df: pd.DataFrame,
    discount_threshold: float = 80.0,
) -> pd.Series:
    """
    Calculate the OC-adjusted value for each loan.

    Loans purchased below the discount threshold are valued at purchase price.
    Defaulted loans are valued at zero.
    All other loans are valued at par.

    Args:
        df: DataFrame with 'par_value', 'purchase_price_pct', and 'status' columns.
        discount_threshold: Percentage below which purchase price is used (default 80).

    Returns:
        Series of OC-adjusted values per loan.
    """
    return np.where(
        df["status"] == "defaulted",
        0.0,
        np.where(
            df["purchase_price_pct"] < discount_threshold,
            df["par_value"] * df["purchase_price_pct"] / 100,
            df["par_value"],
        ),
    )


def calculate_oc_ratio(
    df: pd.DataFrame,
    tranche_balance: float,
    discount_threshold: float = 80.0,
) -> float:
    """
    Calculate the OC ratio for a given tranche.

    Args:
        df: Portfolio DataFrame.
        tranche_balance: Outstanding principal of the tranche being tested.
        discount_threshold: Discount obligation cutoff percentage.

    Returns:
        OC ratio as a percentage (e.g., 125.0 means 125%).
    """
    df["_oc_value"] = calculate_oc_numerator(df, discount_threshold)
    collateral_balance = df["_oc_value"].sum()
    df.drop(columns=["_oc_value"], inplace=True)

    if tranche_balance == 0:
        return float("inf")

    return (collateral_balance / tranche_balance) * 100

import numpy as np
import pandas as pd


def adjusted_collateral_balance(
    df: pd.DataFrame,
    ccc_threshold: float = 0.075,
    discount_threshold: float = 80.0,
) -> float:
    """
    Calculate the adjusted collateral balance after CCC haircuts.

    This is the primary OC numerator calculation that accounts for:
    1. Defaulted loan exclusion
    2. Discount obligation adjustments
    3. CCC excess haircuts (lowest-priced first)

    Args:
        df: Portfolio DataFrame with columns:
            - par_value: Face value of each loan
            - purchase_price_pct: Original purchase price as % of par
            - market_price_pct: Current market price as % of par
            - status: Loan status ('performing', 'defaulted', etc.)
            - is_ccc: Boolean flag for CCC/Caa1 or worse rated
        ccc_threshold: Maximum CCC exposure as fraction of total par (default 7.5%)
        discount_threshold: Cutoff for discount obligation treatment (default 80%)

    Returns:
        Adjusted collateral balance as a float.
    """
    # Step 1: Exclude defaulted loans
    active = df[df["status"] != "defaulted"].copy()

    # Step 2: Calculate base OC value (handle discount obligations)
    active["oc_value"] = np.where(
        active["purchase_price_pct"] < discount_threshold,
        active["par_value"] * active["purchase_price_pct"] / 100,
        active["par_value"],
    )

    # Step 3: CCC haircut calculation
    total_par = active["par_value"].sum()
    ccc_mask = active["is_ccc"]
    ccc_par = active.loc[ccc_mask, "par_value"].sum()

    if ccc_par / total_par > ccc_threshold:
        excess = ccc_par - (total_par * ccc_threshold)

        # Sort CCC assets by market price ascending (worst first)
        ccc_assets = active[ccc_mask].sort_values("market_price_pct")

        haircut = 0.0
        remaining_excess = excess
        for _, loan in ccc_assets.iterrows():
            if remaining_excess <= 0:
                break
            affected_par = min(loan["par_value"], remaining_excess)
            # Haircut = difference between par and market value
            haircut += affected_par * max(0, 1 - loan["market_price_pct"] / 100)
            remaining_excess -= affected_par

        return active["oc_value"].sum() - haircut

    return active["oc_value"].sum()

def get_loan_value(
    par_value: float,
    purchase_price_pct: float,
    market_price_pct: float,
    status: str,
    valuation_method: str = "par",
    discount_threshold: float = 80.0,
    recovery_rate: float = 0.0,
) -> float:
    """
    Return the appropriate loan value based on valuation method and status.

    This function centralizes the valuation logic so that different compliance
    tests can request the correct value without reimplementing the rules.
    """
    if status == "defaulted":
        if valuation_method == "recovery":
            return par_value * recovery_rate / 100
        return 0.0

    if valuation_method == "par":
        if purchase_price_pct < discount_threshold:
            return par_value * purchase_price_pct / 100
        return par_value

    if valuation_method == "market":
        return par_value * market_price_pct / 100

    if valuation_method == "purchase":
        return par_value * purchase_price_pct / 100

    if valuation_method == "amortized_cost":
        # Simplified: linear accretion from purchase price to par
        return par_value * purchase_price_pct / 100

    raise ValueError(f"Unknown valuation method: {valuation_method}")

def validate_pricing_data(df: pd.DataFrame) -> list[str]:
    """
    Return a list of data quality warnings for pricing fields.
    Empty list means all checks passed.
    """
    warnings = []

    # Purchase price should be between 0 and 200
    invalid_purchase = df[
        (df["purchase_price_pct"] < 0) | (df["purchase_price_pct"] > 200)
    ]
    if len(invalid_purchase) > 0:
        warnings.append(
            f"{len(invalid_purchase)} loans have purchase_price_pct "
            f"outside valid range [0, 200]"
        )

    # Market price should exist for CCC loans (needed for haircut)
    ccc_no_market = df[df["is_ccc"] & df["market_price_pct"].isna()]
    if len(ccc_no_market) > 0:
        warnings.append(
            f"{len(ccc_no_market)} CCC-rated loans are missing market_price_pct. "
            f"CCC haircut calculation will be inaccurate."
        )

    # Defaulted loans should have status set
    likely_defaulted = df[
        (df["market_price_pct"] < 30) & (df["status"] != "defaulted")
    ]
    if len(likely_defaulted) > 0:
        warnings.append(
            f"{len(likely_defaulted)} loans trade below 30 but are not marked "
            f"as defaulted. Verify status field."
        )

    return warnings