from pathlib import Path
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from IPython.display import display
from cycler import cycler
from quantfinlab.common.cache import cache_key
from quantfinlab.credit import events, features, models, structural, curves, pricing, portfolio, structured, market
from quantfinlab.dataio import read_equity_history, read_sec_filings, read_credit_facts
from quantfinlab.dataio.credit import read_credit_market, read_structured_credit
from quantfinlab.dataio.rates import read_credit_curves
from quantfinlab.dataio.fundamentals import read_statement_values
from quantfinlab.fundamentals import (classify_duration_facts, monthly_statement_values,
safe_ratio, ohlson_score, zmijewski_score)
from quantfinlab.fixed_income.discounting import discount_from_zero
from quantfinlab.ml.classifiers import probability_scores
from quantfinlab.numerics.copulas import gaussian_copula, student_t_copula
from quantfinlab.portfolio.covariance import oas_covariance, ledoit_wolf_covariance
from quantfinlab.plotting import credit as credit_plots
palette = ["#069AF3", "#FE420F", "#00008B", "#008080", "#CC79A7", "#9614fa",
"#DC143C", "#7BC8F6", "#0072B2", "#04D8B2", "#800080", "#FF8072"]
plt.rcParams["axes.prop_cycle"] = cycler(color=palette)
plt.rcParams.update({"figure.figsize": (6, 3), "figure.dpi": 200, "savefig.dpi": 300,
"axes.grid": True, "grid.alpha": .20, "axes.spines.top": False,
"axes.spines.right": False, "axes.titlesize": 12, "axes.labelsize": 12,
"xtick.labelsize": 9, "ytick.labelsize": 9, "legend.fontsize": 7})
data = Path("../data")
start = pd.Timestamp("2012-01-01")
oos = pd.Timestamp("2019-01-01")
horizons = (3, 6, 12, 24)
R = 0.40
key = cache_key([data / "sec_credit.parquet"],
{"calculation": "credit-library-1", "start": start, "oos": oos,
"horizons": horizons, "seed": 22, "concepts": features.credit_concepts})
cache = Path("../workspace/library_repeat/cache") / key
cache.mkdir(parents=True, exist_ok=True)
filings = read_sec_filings(data / "sec_credit.parquet")
history = events.risk_set(filings, start=start)
last = filings["accepted_at"].max().to_period("M").to_timestamp("M")
dates = pd.date_range(start + pd.offsets.MonthEnd(0), last, freq="ME")
months = events.issuer_months(history, dates, statements=True)[["decision_date", "cik"]]
concepts = [concept for section in features.credit_concepts.values()
for alternatives in section.values() for concept in alternatives]
statements = []
for first in range(0, len(history), 1000):
ciks = history.iloc[first:first + 1000]["cik"]
path = cache / f"statements_{first}.parquet"
if path.exists():
values = read_statement_values(path, instant_fields=features.credit_concepts["instant"])
else:
facts = read_credit_facts(data / "sec_credit.parquet", ciks=ciks, concepts=concepts)
facts = classify_duration_facts(facts, concepts=features.credit_concepts)
values = monthly_statement_values(
facts, months[months["cik"].isin(ciks)], availability_col="accepted_at")
keep = [name for name in values if name in {"cik", "decision_date", "filed_date", "latest_quarter_end",
"latest_duration_end", "latest_balance_end", "latest_period_end"}
or name in features.credit_concepts["instant"] or name.endswith(("_ttm", "_ttm_yoy", "_ttm_method"))]
values = values[keep].copy()
del facts
values.to_parquet(path, index=False)
statements.append(values)
accounting = pd.concat(statements, ignore_index=True)
del statements, values
credit = events.statement_sample(accounting, history, dates)
credit = features.credit_ratios(credit)
credit = pd.concat([credit, features.ratio_changes(credit)], axis=1)
counts = events.filing_counts(filings, credit)
credit = pd.concat([credit, counts.drop(columns=["decision_date", "cik"])], axis=1)
credit["distress_index"] = features.distress_index(credit)
credit = pd.concat([credit, features.credit_controls(credit["decision_date"], credit["sec_sic"])], axis=1)
credit = pd.concat([credit, events.event_labels(credit)], axis=1)
credit["origin"] = (credit["decision_date"].dt.year - start.year) * 12 + credit["decision_date"].dt.month - 1
sample = credit[credit["total_assets"].notna()
& credit[features.accounting_features].notna().sum(axis=1).ge(5)].copy()
sample["state"] = events.distress_states(sample)
cpi = pd.read_csv(data / "us_macro_factors.csv", parse_dates=["date"]).set_index("date")["cpi_all_items"]
NI = sample["net_income_ttm"]
NI0 = sample.groupby("cik")["net_income_ttm"].shift(12)
TA = sample["total_assets"]
sample["ohlson_o"] = ohlson_score(
size=TA.div(1e6).div(sample["decision_date"].map(cpi) / 100).where(TA.gt(0)),
liabilities_assets=safe_ratio(sample["total_liabilities"], TA),
working_capital_assets=safe_ratio(sample["wc"], TA),
current_liabilities_assets=safe_ratio(sample["current_liabilities"], sample["current_assets"]),
net_income_assets=safe_ratio(NI, TA), funds_liabilities=safe_ratio(sample["cfo_ttm"], sample["total_liabilities"]),
negative_equity=sample["total_liabilities"].gt(TA).astype(float).where(sample[["total_liabilities", "total_assets"]].notna().all(axis=1)),
two_losses=(NI.lt(0) & NI0.lt(0)).astype(float).where(NI.notna() & NI0.notna()),
income_change=safe_ratio(NI - NI0, NI.abs() + NI0.abs()))
sample["zmijewski_x"] = zmijewski_score(safe_ratio(NI, TA), safe_ratio(sample["total_liabilities"], TA), sample["current_ratio"])
refits = pd.date_range(oos, last, freq="YS")
forecast = sample[sample["decision_date"].ge(oos)][
["decision_date", "cik", *[name for h in horizons for name in [f"event_{h}m", f"observed_event_{h}m"]]]].copy()
fits = {}
for method, names in [("logit", features.linear_features), ("spline", features.model_features),
("lightgbm", features.model_features)]:
path = cache / f"hazard_{method}.parquet"
if path.exists():
predictions = pd.read_parquet(path)
else:
predictions, choices = models.default_forecasts(
sample, "event_1m", names, method=method, refit_dates=refits, n_jobs=4)
predictions.to_parquet(path, index=False)
forecast = forecast.merge(predictions[["decision_date", "cik", "p"]].rename(
columns={"p": f"h_{method}"}), on=["decision_date", "cik"], how="left")
for method, h in [("lightgbm", 12), ("logit", 12), *[("spline", h) for h in horizons],
*[("lightgbm", h) for h in [3, 6, 24]]]:
path = cache / f"direct_{method}_{h}.parquet"
choice_path = cache / "direct_choices.parquet"
if path.exists():
predictions = pd.read_parquet(path)
else:
choices = pd.read_parquet(choice_path) if method == "lightgbm" and h != 12 else None
predictions, decisions = models.default_forecasts(
sample, f"event_{h}m", features.model_features, method=method, horizon=h,
spacing=h, refit_dates=refits, choices=choices, n_jobs=4)
predictions.to_parquet(path, index=False)
if method == "lightgbm" and h == 12:
decisions.to_parquet(choice_path, index=False)
forecast = forecast.merge(predictions[["decision_date", "cik", "p"]].rename(
columns={"p": f"pd{h}_{method}"}), on=["decision_date", "cik"], how="left")
for h in horizons:
forecast[f"pd{h}_frozen"] = 1 - (1 - forecast["h_lightgbm"]) ** h
forecast[f"pd{h}"] = forecast[[f"pd{h}_spline", f"pd{h}_lightgbm", f"pd{h}_frozen"]].mean(axis=1)
for suffix in ["spline", "lightgbm", ""]:
columns = [f"pd{h}_{suffix}" if suffix else f"pd{h}" for h in horizons]
forecast[columns] = np.maximum.accumulate(forecast[columns].to_numpy(), axis=1)
forecast["pd12_l2"] = 1 - (1 - forecast["h_logit"]) ** 12
forecast["pd12_gam"] = 1 - (1 - forecast["h_spline"]) ** 12
naive = {}
for date in refits:
train = models.matured(sample, date, 1, "event_1m")
naive[date.year] = 1 - (1 - train["event_1m"].mean()) ** 12
forecast["pd12_naive"] = forecast["decision_date"].dt.year.map(naive)
forecast = forecast.merge(sample[["decision_date", "cik", "ohlson_o", "zmijewski_x", "debt_assets"]], on=["decision_date", "cik"])
scores = []
comparisons = {"Expanding event rate": "pd12_naive", "L2 hazard": "pd12_l2", "Spline hazard": "pd12_gam", "LightGBM hazard": "pd12_frozen",
"Direct logit": "pd12_logit", "Direct spline": "pd12_spline", "Direct LightGBM": "pd12_lightgbm",
"Ensemble": "pd12", "Ohlson proxy": "ohlson_o", "Zmijewski": "zmijewski_x"}
for name, column in comparisons.items():
known = forecast[forecast["decision_date"].le(last - pd.offsets.MonthEnd(12)) & forecast["observed_event_12m"]]
scores.append({"model": name, "horizon": 12, **probability_scores(
known["event_12m"], known[column], ranking_only=name in ["Ohlson proxy", "Zmijewski"])})
for h in [3, 6, 24]:
known = forecast[forecast["decision_date"].le(last - pd.offsets.MonthEnd(h)) & forecast[f"observed_event_{h}m"]]
scores.append({"model": "Ensemble", "horizon": h, **probability_scores(known[f"event_{h}m"], known[f"pd{h}"])})
scores = pd.DataFrame(scores).set_index(["model", "horizon"])
transitions = []
choices = pd.read_parquet(cache / "direct_choices.parquet")
for state, target, spacing in [("healthy", "broad_12m", 12), ("distressed", "event_12m", 3)]:
population = sample[sample["state"].eq(state)]
paths = []
for method in ["logit", "lightgbm"]:
path = cache / f"{state}_{method}.parquet"
if path.exists():
predictions = pd.read_parquet(path)
else:
predictions, _ = models.default_forecasts(
population, target, features.model_features, method=method, horizon=12,
spacing=spacing, refit_dates=refits, choices=choices if method == "lightgbm" else None, n_jobs=4)
predictions.to_parquet(path, index=False)
paths.append(predictions.set_index(["decision_date", "cik"])["p"])
prediction = pd.concat(paths, axis=1).mean(axis=1).rename("p").reset_index()
known = population.merge(prediction, on=["decision_date", "cik"])
known = known[known["decision_date"].le(last - pd.offsets.MonthEnd(12)) & known[f"observed_{target}"]]
transitions.append({"model": f"{state} transition", **probability_scores(known[target], known["p"])})
landmark = sample[sample["decision_date"].eq("2016-12-31")].copy()
status = landmark["strict_date"].gt("2016-12-31") & landmark["strict_date"].le(landmark["last_observed"])
landmark = pd.concat([landmark[status], landmark[~status].sample(min(2000, (~status).sum()), random_state=22)]).sort_values("cik")
status = landmark["strict_date"].gt("2016-12-31") & landmark["strict_date"].le(landmark["last_observed"])
duration = (landmark["last_observed"].where(~status, landmark["strict_date"]) - pd.Timestamp("2016-12-31")).dt.days.div(30.4375).clip(lower=1)
cox_names = ["liabilities_assets", "debt_assets", "interest_coverage", "cfo_debt", "cash_assets", "current_ratio",
"fcf_assets", "roa", "operating_margin", "sales_assets", "accruals", "distress_index"]
cox = models.fit_cox(landmark[cox_names], duration, status)
transitions.append({"model": "Landmark Cox PH (in-sample)", "Harrell C": models.harrell_c(duration, status, cox["risk"])})
train = models.matured(sample, pd.Timestamp("2016-12-31"), 1, "event_1m")
hazard = models.fit_logit(train[features.linear_features], train["event_1m"])
transitions.append({"model": "Pre-landmark L2 hazard", "Harrell C": models.harrell_c(
duration, status, models.predict_default(hazard, landmark))})
equity = read_equity_history(data / "sp500_market_data.parquet", start="2011-01-01",
columns=["date", "ticker", "adj_close", "close", "volume", "is_sp500_member",
"snapshot_date", "industry", "market_cap"])
members = events.market_members(equity["market"], filings)
returns = equity["returns"].loc[:, pd.Index(members["ticker"].unique()).intersection(equity["returns"].columns)]
volatility = structural.equity_volatility(returns)
rates = read_credit_curves(data / "treasury_credit_curves.parquet")
rates["decision_date"] = rates["date"].dt.to_period("M").dt.to_timestamp("M")
spot = rates[rates["rate_type"].eq("spot") & rates["observation_type"].eq("end_of_month")]
tnc = spot[spot["curve_family"].eq("TNC")].pivot(index="decision_date", columns="maturity_years", values="rate")
hqm = spot[spot["curve_family"].eq("HQM")].pivot(index="decision_date", columns="maturity_years", values="rate")
spread = hqm - tnc
merton = sample.merge(members[["decision_date", "cik", "ticker", "market_cap"]].rename(
columns={"ticker": "market_ticker", "market_cap": "E"}), on=["decision_date", "cik"])
merton = merton.merge(volatility[["decision_date", "ticker", "sigma_E"]],
left_on=["decision_date", "market_ticker"], right_on=["decision_date", "ticker"], suffixes=("", "_price"))
merton["r"] = merton["decision_date"].map(tnc[1.0])
merton["D"] = merton["total_liabilities"]
merton = merton[merton["E"].gt(0) & merton["D"].gt(0) & merton["sigma_E"].gt(0) & merton["r"].notna()].reset_index(drop=True)
solution = structural.merton_assets(merton["E"], merton["sigma_E"], merton["D"], merton["r"])
merton["merton_score"] = -solution["d2"]
merton["merton_score_sq"] = merton["merton_score"] ** 2
merton_predictions, _ = models.default_forecasts(
merton[merton["state"].eq("healthy")], "broad_12m", ["merton_score", "merton_score_sq"],
horizon=12, spacing=12, refit_dates=refits)
merton_common = merton_predictions.rename(columns={"p": "pd_merton"}).merge(
forecast[["decision_date", "cik", "pd12"]], on=["decision_date", "cik"])
known = merton_common[merton_common["observed_broad_12m"] & merton_common["decision_date"].le(last - pd.offsets.MonthEnd(12))]
transitions.append({"model": "Merton (broad distress)", **probability_scores(known["broad_12m"], known["pd_merton"])})
transitions.append({"model": "Accounting rank on broad distress", **probability_scores(known["broad_12m"], known["pd12"])})
transitions = pd.DataFrame(transitions).set_index("model")
risk = forecast.merge(sample[["decision_date", "cik", "debt", "entity_name", "sec_industry"]], on=["decision_date", "cik"])
risk = risk.merge(members[["decision_date", "cik", "ticker", "market_cap"]].assign(is_member=True),
on=["decision_date", "cik"], how="left")
risk["is_member"] = risk["is_member"].eq(True)
risk = risk[risk["debt"].gt(0)].copy()
risk["loss"] = 100 * (1 - R) * risk["pd12_l2"]
losses = market.aggregate_credit_loss(risk, "loss")
member_losses = market.aggregate_credit_loss(risk[risk["is_member"]], "loss")
fed = read_credit_market(data / "fed_credit.parquet")
fed["decision_date"] = fed["date"].dt.to_period("M").dt.to_timestamp("M")
premium = fed.set_index("decision_date")[["gz_spread", "ebp"]].join(losses, how="inner")
premium["default_component"] = premium["gz_spread"] - premium["ebp"]
premium["mapped_loss"] = market.loss_spread_history(premium, ["loss"], "default_component")
premium["public_excess_credit_premium"] = market.excess_credit_premium(premium["gz_spread"], premium["mapped_loss"])
premium_validation = market.premium_validation(premium, "public_excess_credit_premium")
finra = read_credit_market(data / "finra_credit_market.parquet")
breadth = market.credit_breadth(finra)
sentiment = market.customer_imbalance(finra)
activity = breadth.merge(sentiment[["date", "security_category", "customer_imbalance"]], on=["date", "security_category"], how="outer")
activity["decision_date"] = activity["date"].dt.to_period("M").dt.to_timestamp("M")
activity = activity[activity["security_category"].isin(["investment grade", "high yield"])]
activity["grade"] = activity["security_category"].map({"investment grade": "ig", "high yield": "hy"})
activity = activity.groupby(["decision_date", "grade"])[["advance_decline", "high_low", "customer_imbalance",
"total_volume", "total_trades"]].mean().unstack()
activity.columns = [f"{grade}_{name}" for name, grade in activity.columns]
cmdi = read_credit_market(data / "nyfed_cmdi.parquet").set_index("date")[["market_cmdi", "ig_cmdi", "hy_cmdi"]].resample("ME").last()
market_results = premium.join(cmdi, how="inner").join(activity, how="inner")
market_results["premium_change"] = market_results["public_excess_credit_premium"].diff()
market_results["cmdi_change"] = market_results["market_cmdi"].diff()
finra_names = [f"{grade}_{name}" for name in ["advance_decline", "high_low", "customer_imbalance"] for grade in ["ig", "hy"]]
finra_relationship = market_results[["premium_change", "cmdi_change", *finra_names]].corr().loc[
finra_names, ["premium_change", "cmdi_change"]]
cmdi_comparison = premium.join(cmdi, how="inner")
cmdi_names = ["public_excess_credit_premium", "ebp", "market_cmdi", "ig_cmdi", "hy_cmdi"]
cmdi_relationship = pd.concat({"Level": cmdi_comparison[cmdi_names].corr(),
"Change": cmdi_comparison[cmdi_names].diff().corr()})[["public_excess_credit_premium", "ebp"]]
T = np.array([1., 3., 5., 7., 10.])
market_hazards = []
for date in spread.index.intersection(tnc.index):
s = spread.loc[date, T].to_numpy()
r = tnc.loc[date, T].to_numpy()
if np.isfinite(np.r_[s, r]).all() and np.all(s > 0):
fit = curves.bootstrap_hazards(s, T, discount_from_zero(T, r), R=R,
quote_kind="yield_spread_proxy", on_failure="boundary")
market_hazards.append({"date": date, "pd5": curves.default_probability(5, T, fit["hazards"])[0],
"repricing_bp": np.max(np.abs(fit["error_bp"]))})
market_hazards = pd.DataFrame(market_hazards).set_index("date")
as_of = max(set(tnc.index) & set(premium.index[premium["default_component"].gt(0)])
& set(risk.loc[risk["is_member"], "decision_date"]))
issuers = risk[risk["decision_date"].eq(as_of) & risk["is_member"]].dropna(
subset=[f"pd{h}" for h in horizons]).sort_values("debt").drop_duplicates("cik", keep="last").copy()
knots = np.array([.25, .5, 1., 2., 5., 7., 10.])
lambda_p = curves.hazards_from_pd(issuers[[f"pd{h}" for h in horizons]], knots[:4])
tail = -np.log1p(-issuers["pd24"].to_numpy()) / 2
lambda_p = np.column_stack([lambda_p, tail, tail, tail])
discount = discount_from_zero(knots, np.interp(knots, tnc.columns, tnc.loc[as_of]))
calibration = pricing.fit_hazard_multiplier(lambda_p, knots, discount,
premium.loc[as_of, "default_component"] / 100, issuers["debt"], R=R)
lambda_q = calibration["lambda_q"]
issuers["pd5_physical"] = curves.default_probability(5, knots, lambda_p).ravel()
issuers["pd5_risk_neutral"] = curves.default_probability(5, knots, lambda_q).ravel()
issuers["relative_risk"] = tail / np.average(tail, weights=issuers["debt"])
cds = []
for T in [1, 3, 5, 7, 10]:
s = pricing.cds_spread(lambda_q, knots, discount, T, R=R)
pv01 = pricing.risky_pv01(lambda_q, knots, discount, T, R=R, notional=10_000_000)
cds.append(pd.DataFrame({"cik": issuers["cik"].to_numpy(), "ticker": issuers["ticker"].to_numpy(),
"maturity": T, "spread_bp": 1e4 * s, "risky_PV01": pv01,
"JTD": 10_000_000 * (1 - R),
"R30_spread_bp": 1e4 * pricing.cds_spread(lambda_q, knots, discount, T, R=.3),
"R50_spread_bp": 1e4 * pricing.cds_spread(lambda_q, knots, discount, T, R=.5),
"hazard150_spread_bp": 1e4 * pricing.cds_spread(1.5 * lambda_q, knots, discount, T, R=R)}))
cds = pd.concat(cds, ignore_index=True)
cs01 = {}
for cik, quotes in cds.groupby("cik"):
quotes = quotes.sort_values("maturity")
cs01[cik] = pricing.cds_cs01(quotes["spread_bp"].to_numpy() / 1e4, quotes["maturity"], discount,
5, coupon=quotes.loc[quotes["maturity"].eq(5), "spread_bp"].iloc[0] / 1e4,
R=R, notional=10_000_000)
issuers["CS01_5y"] = issuers["cik"].map(cs01)
issuers["decile"] = pd.qcut(issuers["relative_risk"].rank(method="first"), 10, labels=range(1, 11))
issuers["cds5_bp"] = issuers["cik"].map(cds[cds["maturity"].eq(5)].set_index("cik")["spread_bp"])
deciles = issuers.groupby("decile", observed=True)[["pd5_physical", "pd5_risk_neutral", "relative_risk", "cds5_bp", "CS01_5y"]].median()
return_counts = returns.loc[as_of - pd.DateOffset(years=5):as_of].notna().sum()
issuers["return_count"] = issuers["ticker"].map(return_counts)
book = issuers[issuers["return_count"].ge(756)].sort_values(["market_cap", "ticker"], ascending=[False, True]).head(125).copy()
book["exposure"] = 1_000_000.
covariance = oas_covariance(returns.loc[as_of - pd.DateOffset(years=5):as_of, book["ticker"]], return_df=True)
covariance_lw = ledoit_wolf_covariance(returns.loc[as_of - pd.DateOffset(years=5):as_of, book["ticker"]], return_df=True)
rho = covariance / np.outer(np.sqrt(np.diag(covariance)), np.sqrt(np.diag(covariance)))
book = book.set_index("ticker").loc[rho.index].reset_index()
p = book["pd5_physical"].to_numpy()
E = book["exposure"].to_numpy()
draws = {"Independent": np.random.default_rng(22).random((100_000, len(book))),
"Gaussian": gaussian_copula(rho, 60_000, seed=37),
"Student-t (5)": student_t_copula(rho, 60_000, nu=5, seed=37)}
loss_draws = {name: portfolio.portfolio_losses(U, p, E, recovery=R) for name, U in draws.items()}
rho_lw = covariance_lw.loc[rho.index, rho.index]
rho_lw = rho_lw / np.outer(np.sqrt(np.diag(rho_lw)), np.sqrt(np.diag(rho_lw)))
loss_draws["Gaussian, Ledoit-Wolf"] = portfolio.portfolio_losses(gaussian_copula(rho_lw, 60_000, seed=37), p, E, recovery=R)
loss_summary = pd.DataFrame({name: portfolio.loss_summary(loss, notional=E.sum()) for name, loss in loss_draws.items()}).T
concentration = portfolio.credit_concentration(E, p, book["sec_industry"], recovery=R)
tranches = [("Equity", 0, .03), ("Junior mezzanine", .03, .07), ("Senior mezzanine", .07, .15),
("Senior", .15, .30), ("Super senior", .30, 1)]
book_lambdas = pd.DataFrame(lambda_p, index=issuers["cik"]).loc[book["cik"]].to_numpy()
tranche_results = []
for year in range(1, 6):
probabilities = curves.default_probability(year, knots, book_lambdas).ravel()
losses = portfolio.portfolio_losses(draws["Student-t (5)"], probabilities, E, recovery=R) / E.sum()
tranche_results.append(structured.tranche_summary(losses, tranches).assign(year=year).reset_index())
tranche_results = pd.concat(tranche_results, ignore_index=True)
stress = []
dependence = {}
for i, (name, family, nu, multiplier) in enumerate([
("Gaussian, 0.75 rho", "gaussian", 5, .75), ("Gaussian, rho", "gaussian", 5, 1.),
("t(7), rho", "t", 7, 1.), ("t(4), 1.25 rho", "t", 4, 1.25)]):
correlation = np.eye(len(rho)) + multiplier * (rho.to_numpy() - np.eye(len(rho)))
dependence[name] = (gaussian_copula(correlation, 20_000, seed=220 + i) if family == "gaussian"
else student_t_copula(correlation, 20_000, nu=nu, seed=220 + i))
for scenario, multiplier in {"Half hazard": .5, "Base": 1., "Double hazard": 2., "Triple hazard": 3.,
"2008 spread ratio": spread.loc["2008-11-30", 5] / spread.loc[as_of, 5],
"2020 spread ratio": spread.loc["2020-03-31", 5] / spread.loc[as_of, 5]}.items():
for recovery in [.2, .4, .6]:
for name, U in dependence.items():
losses = portfolio.portfolio_losses(U, 1 - (1 - p) ** multiplier, E, recovery=recovery)
stress.append({"scenario": scenario, "recovery": recovery, "dependence": name,
**portfolio.loss_summary(losses, notional=E.sum()).to_dict(),
"equity_EL": structured.tranche_loss(losses / E.sum(), 0, .03).mean(),
"mezzanine_EL": structured.tranche_loss(losses / E.sum(), .03, .07).mean(),
"senior_EL": structured.tranche_loss(losses / E.sum(), .15, .30).mean()})
stress = pd.DataFrame(stress)
reports = read_structured_credit(data / "finra_structured_pricing.parquet", sheet="CBO-CDO-CLO", start="2024-12-09")
reports = structured.structured_categories(reports)
prices = reports[reports["metric"].eq("AVERAGE PRICE") & reports["value"].gt(0)].groupby(
["report_date", "category", "vintage"], as_index=False)["value"].mean()
prices["date"] = prices["report_date"].dt.to_period("M").dt.to_timestamp("M")
price_history = prices.groupby(["date", "category"])["value"].median().unstack()
vintages = prices.groupby(["category", "vintage"])["value"].agg(["count", "mean", "min", "last"])
trading = read_structured_credit(data / "finra_structured_activity.parquet")
trading = structured.trading_activity(trading[trading["row_label"].eq("CBO/CDO/CLO")])
errata = read_credit_market(data / "finra_structured_errata.parquet")
assert errata[errata["sub_asset_class"].str.contains("CBO/CDO/CLO", case=False, na=False)
& errata["trade_date"].ge("2024-12-09")].empty
direct_train = models.matured(sample, last + pd.offsets.MonthEnd(1), 12, "event_12m")
direct_train = direct_train[direct_train["origin"].mod(12).eq(0)]
choice = choices.sort_values("cutoff").iloc[-1]
tree = models.fit_lightgbm(direct_train[features.model_features], direct_train["event_12m"],
spec=models.boosting_specs[int(choice["spec"])], trees=int(choice["trees"]), n_jobs=4)
explain = direct_train.sample(min(5000, len(direct_train)), random_state=22)
shap = models.default_contributions(tree, explain).drop(columns="base_value")
drivers = shap.abs().mean().rename("mean absolute SHAP").nlargest(10).to_frame()
case_key = forecast[forecast["decision_date"].eq(forecast["decision_date"].max())].nlargest(1, "pd12")[["decision_date", "cik"]]
case = sample.merge(case_key, on=["decision_date", "cik"])
drivers["case SHAP"] = models.default_contributions(tree, case).iloc[0].drop("base_value")
train = models.matured(sample, last + pd.offsets.MonthEnd(1), 1, "event_1m")
linear_fit = models.fit_logit(train[features.linear_features], train["event_1m"])
drivers["L2 coefficient"] = pd.Series(linear_fit.model.coef_[0], index=linear_fit.columns)
spline_fit = models.fit_spline(direct_train[features.model_features], direct_train["event_12m"])
cfo = np.linspace(direct_train["cfo_debt"].quantile(.02), direct_train["cfo_debt"].quantile(.98), 120)
spline_points = pd.concat([direct_train[features.model_features].median().to_frame().T] * len(cfo), ignore_index=True)
spline_points["cfo_debt"] = cfo
spline_response = pd.DataFrame({"cfo_debt": cfo, "pd12": models.predict_default(spline_fit, spline_points)})
market_summary = finra_relationship[["premium_change"]].rename(columns={"premium_change": "Value"})
market_summary["Meaning"] = "Correlation with monthly premium change"
market_summary.loc["Fed EBP", ["Value", "Meaning"]] = [premium_validation["correlation"], "Correlation with premium level"]
market_summary.loc["Premium RMSE", ["Value", "Meaning"]] = [premium_validation["RMSE"], "Percentage points versus Fed EBP"]
market_summary.loc["EBP regression slope", ["Value", "Meaning"]] = [premium_validation["slope"], "EBP regressed on public premium"]
market_summary.loc["Risk-price multiplier", ["Value", "Meaning"]] = [calibration["kappa"], "kappa; dimensionless"]
market_summary.loc["Spread calibration error", ["Value", "Meaning"]] = [calibration["error_bp"], "Basis points"]
market_summary.loc["Portfolio notional", ["Value", "Meaning"]] = [E.sum() / 1e6, "USD millions"]
portfolio_table = pd.concat({"Portfolio": loss_summary,
"Tranche (5Y)": tranche_results[tranche_results["year"].eq(5)].set_index("tranche")[["expected_loss", "impairment"]],
"Stress, R=40%": stress[stress["recovery"].eq(.4) & stress["dependence"].eq("t(4), 1.25 rho")].set_index("scenario").drop(columns=["recovery", "dependence"])})
portfolio_table = portfolio_table[["expected_loss", "unexpected_loss", "var_99%", "es_99%", "impairment"]]
portfolio_table.columns = ["EL", "Loss volatility", "99% VaR", "99% ES", "Impairment probability"]
drivers = drivers.rename(columns={"mean absolute SHAP": "Mean |SHAP| (log odds)",
"case SHAP": f"CIK {int(case['cik'].iloc[0])} SHAP",
"L2 coefficient": "Monthly L2 coefficient"})
deciles = deciles.rename(columns={"pd5_physical": "Physical PD (5Y)", "pd5_risk_neutral": "Risk-neutral PD (5Y)",
"relative_risk": "Relative hazard", "cds5_bp": "CDS (bp)",
"CS01_5y": "5Y CS01 ($/bp; $10m)"})
display(scores.style.format(precision=4, na_rep="—"),
transitions.style.format(precision=4, na_rep="—"),
drivers.style.format(precision=4, na_rep="—"),
market_summary.style.format({"Value": "{:,.4f}"}),
deciles.style.format({"Physical PD (5Y)": "{:.2%}", "Risk-neutral PD (5Y)": "{:.2%}",
"Relative hazard": "{:.3f}", "CDS (bp)": "{:.1f}", "5Y CS01 ($/bp; $10m)": "{:,.1f}"}),
portfolio_table.style.format("{:.2%}", na_rep="—"))
known = forecast[forecast["decision_date"].le(last - pd.offsets.MonthEnd(12)) & forecast["observed_event_12m"]]
latest = merton_common[merton_common["decision_date"].eq(merton_common["decision_date"].max())]
representatives = []
for decile in [1, 3, 5, 7, 10]:
group = issuers[issuers["decile"].eq(decile)]
representatives.append(group.loc[(group["relative_risk"] - group["relative_risk"].median()).abs().idxmin(), "ticker"])
fig, axes = plt.subplots(4, 2, figsize=(16, 19), constrained_layout=True)
credit_plots.plot_default_capture(known, "event_12m", {"Ensemble": "pd12", "LightGBM": "pd12_lightgbm", "L2 hazard": "pd12_l2"}, ax=axes[0, 0])
credit_plots.plot_pd_calibration(known, "event_12m", {"Ensemble": "pd12", "Frozen hazard": "pd12_frozen"}, ax=axes[0, 1])
credit_plots.plot_structural_ranks(latest, ax=axes[1, 0])
credit_plots.plot_credit_premium(premium, ax=axes[1, 1])
credit_plots.plot_cds_curves(cds, names=representatives, ax=axes[2, 0])
credit_plots.plot_credit_losses(loss_draws, notional=E.sum(), ax=axes[2, 1])
credit_plots.plot_tranche_losses(tranche_results, ax=axes[3, 0])
credit_plots.plot_structured_prices(price_history, ax=axes[3, 1])
fig.suptitle(f"U.S. corporate credit · library repeat · {as_of:%B %Y}", fontsize=16)
plt.show()