diff --git a/src/config.py b/src/config.py index 760429a..94c72f3 100644 --- a/src/config.py +++ b/src/config.py @@ -52,3 +52,4 @@ PLOT_DECAY = OUTPUT_DIR / "plot_decay.png" PLOT_SLOTTING = OUTPUT_DIR / "plot_slotting.png" PLOT_SCENARIOS = OUTPUT_DIR / "plot_scenarios.png" +METHOD_SUMMARY_MD = OUTPUT_DIR / "method_summary.md" diff --git a/src/irrbb/report.py b/src/irrbb/report.py index de0eee4..f8659ae 100644 --- a/src/irrbb/report.py +++ b/src/irrbb/report.py @@ -7,6 +7,9 @@ import pandas as pd +from src import config +from src.nmd.decay_model import DecayParams + def summarize_worst_case(results: pd.DataFrame) -> dict[str, object]: """Compute most negative delta EVE and delta NII scenarios.""" @@ -24,3 +27,47 @@ def write_worst_case(path: Path, payload: dict[str, object]) -> None: """Persist worst-case json artifact.""" with path.open("w", encoding="utf-8") as f: json.dump(payload, f, indent=2) + + +def write_method_summary(path: Path, params: DecayParams) -> None: + """Write concise methods/assumptions summary for report or PPT.""" + text = f"""# Methods + assumptions summary + +## a) 数据字段与校验方式（balance identity diagnostic） +- NMD输入关键字段：`Date`, `Balance`, `Inflow`, `Outflow`（其余特征在预处理阶段派生）。 +- 曲线输入关键字段：`tenor_years`, `zero_rate`。 +- 余额恒等式诊断：对每期检验 `Balance_t ≈ Balance_(t-1) + Inflow_t - Outflow_t`，并输出最大绝对误差与超容差条数（容差：`{config.BALANCE_TOLERANCE}`）。 + +## b) decay模型形式（S(t)、w_bucket）与5Y cap处理 +- 衰减模型：二元指数混合生存函数 + - `S(t) = Σ_k w_k exp(-λ_k t)`, `k=1,2`, `Σ_k w_k = 1`。 +- 月度分桶权重：`w_bucket(m) = S((m-1)/12) - S(m/12)`，并做非负截断后归一化。 +- 5Y cap：行为期限上限取 `MAX_BEHAV_YEARS={config.MAX_BEHAV_YEARS}`（即 `MAX_BEHAV_MONTHS={config.MAX_BEHAV_MONTHS}`），仅在0~5年内分配权重。 +- 本次运行拟合参数：`weights={params.weights.tolist()}`, `lambdas_per_year={params.lambdas_per_year.tolist()}`。 + +## c) slotting/repricing bucket定义（用什么bucket） +- Repricing buckets 使用月度桶：`M01...M60`。 +- 每桶边界：`t_start=(m-1)/12`, `t_end=m/12`, `t_mid=(t_start+t_end)/2`（单位：年）。 +- 余额分配：`amount_m = B0 * w_bucket(m)`（总额守恒到 `B0`）。 + +## d) 4个shock情景的明确分布函数（含参数：taper_T, ts, tl等） +- `parallel_up`: `Δr(t)=+{config.PARALLEL_UP_BP}bp`。 +- `parallel_down`: `Δr(t)={config.PARALLEL_DOWN_BP}bp`。 +- `short_up_taper`（`taper_T={config.TAPER_T_YEARS}`）： + - `t<=0`: `Δr=+{config.SHORT_UP_MAX_BP}bp` + - `0=taper_T`: `Δr=0` +- `flattener`（`ts={config.TS_YEARS}`, `tl={config.TL_YEARS}`）： + - `t<=ts`: `Δr=+{config.FLATTENER_SHORT_BP}bp` + - `t>=tl`: `Δr={config.FLATTENER_LONG_BP}bp` + - `ts pd.DataFrame: - """Build decay profile table as required by README.""" - df = pd.DataFrame({"weight": weights.values}) - df["month"] = range(1, len(df) + 1) - df["tenor_years"] = df["month"] / 12.0 - df["bucket_label"] = df["month"].map(lambda x: f"M{x:02d}") - df["cum_survival"] = 1.0 - df["weight"].cumsum() - return df[["tenor_years", "bucket_label", "weight", "cum_survival"]] +def _monthly_bucket_grid(max_months: int) -> pd.DataFrame: + rows = [] + for m in range(1, max_months + 1): + t_start = (m - 1) / 12.0 + t_end = m / 12.0 + rows.append( + { + "bucket_label": f"M{m:02d}", + "t_start_years": t_start, + "t_end_years": t_end, + "t_mid_years": 0.5 * (t_start + t_end), + } + ) + return pd.DataFrame(rows) + + +def build_decay_profile(weights: pd.Series, balance: float) -> pd.DataFrame: + """Build required decay profile output table.""" + grid = _monthly_bucket_grid(max_months=len(weights)) + grid["weight"] = weights.to_numpy(dtype=float) + grid["survival"] = (1.0 - grid["weight"].cumsum()).clip(lower=0.0) + grid["amount"] = balance * grid["weight"] + return grid[["bucket_label", "t_start_years", "t_end_years", "t_mid_years", "weight", "survival", "amount"]] diff --git a/src/run.py b/src/run.py index 9e82ae7..332c6cc 100644 --- a/src/run.py +++ b/src/run.py @@ -11,7 +11,7 @@ from src.io.validators import balance_identity_diagnostics, print_validation_summary from src.irrbb.eve import delta_eve from src.irrbb.nii import delta_nii_1y -from src.irrbb.report import summarize_worst_case, write_worst_case +from src.irrbb.report import summarize_worst_case, write_method_summary, write_worst_case from src.nmd.basel_slotting import slot_balance_to_buckets from src.nmd.decay_model import fit_or_fallback, monthly_decay_weights from src.nmd.preprocess import add_behavioral_features, estimate_monthly_hazard @@ -19,6 +19,20 @@ from src.utils.plotting import plot_decay, plot_scenarios, plot_slotting +def _print_scenario_summary_table(scenario_results: pd.DataFrame) -> None: + eve_worst = scenario_results["delta_eve"].idxmin() + nii_worst = scenario_results["delta_nii_1y"].idxmin() + + table = scenario_results.copy() + markers = ["-"] * len(table) + markers[eve_worst] = "EVE" + markers[nii_worst] = "NII" if markers[nii_worst] == "-" else markers[nii_worst] + "/NII" + table["worst_case"] = markers + + print("\nScenario summary (ΔEVE / ΔNII_1Y):") + print(table.to_string(index=False, float_format=lambda x: f"{x:,.2f}")) + + def main() -> None: config.OUTPUT_DIR.mkdir(parents=True, exist_ok=True) @@ -36,8 +50,9 @@ def main() -> None: if used_fallback: print("WARNING: using fallback decay parameters.") - decay_profile = build_decay_profile(pd.Series(weights)) b0 = float(nmd[nmd["Date"] <= pd.Timestamp(config.CALC_DATE)]["Balance"].iloc[-1]) + decay_profile = build_decay_profile(pd.Series(weights), balance=b0) + slotting = slot_balance_to_buckets(balance=b0, weights=pd.Series(weights), alpha_core=config.ALPHA_CORE) slotting_total_err = abs(slotting["amount"].sum() - b0) if slotting_total_err > 1e-6: @@ -70,6 +85,8 @@ def main() -> None: if abs(decay_profile["weight"].sum() - 1.0) > 1e-6: raise ValueError("Decay weights do not sum to 1 within tolerance 1e-6.") + if len(scenario_results) != 4: + raise ValueError(f"Expected exactly 4 shock scenarios, got {len(scenario_results)}") decay_profile.to_csv(config.DECAY_PROFILE_CSV, index=False) slotting.to_csv(config.SLOTTING_TABLE_CSV, index=False) @@ -77,6 +94,7 @@ def main() -> None: worst_case = summarize_worst_case(scenario_results) write_worst_case(config.WORST_CASE_JSON, worst_case) + write_method_summary(config.METHOD_SUMMARY_MD, params=params) plot_decay(decay_profile, config.PLOT_DECAY) plot_slotting(slotting, config.PLOT_SLOTTING) @@ -89,6 +107,7 @@ def main() -> None: f"EVE={worst_case['worst_delta_eve']:.2f} ({worst_case['worst_eve_scenario']}), " f"NII={worst_case['worst_delta_nii_1y']:.2f} ({worst_case['worst_nii_scenario']})" ) + _print_scenario_summary_table(scenario_results) if __name__ == "__main__": diff --git a/src/utils/plotting.py b/src/utils/plotting.py index 7a0d13e..ffb5caa 100644 --- a/src/utils/plotting.py +++ b/src/utils/plotting.py @@ -5,42 +5,64 @@ from pathlib import Path import matplotlib.pyplot as plt +import numpy as np import pandas as pd def plot_decay(decay: pd.DataFrame, path: Path) -> None: - fig, ax = plt.subplots(figsize=(8, 4)) - ax.plot(decay["tenor_years"], decay["weight"], label="weight") - ax.set_title("Decay Profile") - ax.set_xlabel("Tenor (years)") - ax.set_ylabel("Weight") - ax.grid(True, alpha=0.3) + fig, ax1 = plt.subplots(figsize=(9, 4.5)) + + ax1.plot(decay["t_mid_years"], decay["survival"], color="tab:blue", lw=2.0, label="Survival S(t)") + ax1.set_xlabel("Time (years)") + ax1.set_ylabel("Survival S(t)") + ax1.set_ylim(bottom=0.0) + ax1.grid(True, alpha=0.25) + + ax2 = ax1.twinx() + bar_width = np.minimum(0.07, np.diff(decay["t_mid_years"]).min(initial=1 / 12.0) * 0.8) + ax2.bar(decay["t_mid_years"], decay["weight"], width=bar_width, alpha=0.35, color="tab:orange", label="Bucket weight") + ax2.set_ylabel("Weight") + ax2.set_ylim(bottom=0.0) + + ax1.set_title("Decay profile: survival curve and bucket weights") + + h1, l1 = ax1.get_legend_handles_labels() + h2, l2 = ax2.get_legend_handles_labels() + ax1.legend(h1 + h2, l1 + l2, loc="upper right") + fig.tight_layout() fig.savefig(path, dpi=150) plt.close(fig) def plot_slotting(slotting: pd.DataFrame, path: Path) -> None: - fig, ax = plt.subplots(figsize=(10, 4)) + fig, ax = plt.subplots(figsize=(11, 4.5)) ax.bar(slotting["bucket_label"], slotting["amount"], width=0.8) - ax.set_title("Slotting Amount by Bucket") - ax.set_xlabel("Bucket") + ax.set_title("Cashflow slotting table (Basel buckets)") + ax.set_xlabel("Bucket label") ax.set_ylabel("Amount") ax.tick_params(axis="x", labelrotation=90) + ax.grid(True, axis="y", alpha=0.25) fig.tight_layout() fig.savefig(path, dpi=150) plt.close(fig) def plot_scenarios(results: pd.DataFrame, path: Path) -> None: - fig, axes = plt.subplots(1, 2, figsize=(10, 4)) - axes[0].bar(results["scenario"], results["delta_eve"]) - axes[0].set_title("Delta EVE") - axes[0].tick_params(axis="x", labelrotation=25) - - axes[1].bar(results["scenario"], results["delta_nii_1y"]) - axes[1].set_title("Delta NII (1Y)") - axes[1].tick_params(axis="x", labelrotation=25) + fig, ax = plt.subplots(figsize=(10, 4.5)) + x = np.arange(len(results)) + w = 0.38 + + ax.bar(x - w / 2, results["delta_eve"], width=w, label="ΔEVE", color="tab:blue") + ax.bar(x + w / 2, results["delta_nii_1y"], width=w, label="ΔNII (1Y)", color="tab:green") + + ax.set_xticks(x) + ax.set_xticklabels(results["scenario"], rotation=20) + ax.set_title("Scenario comparison: ΔEVE and ΔNII (1Y)") + ax.set_xlabel("Shock scenario") + ax.set_ylabel("Amount") + ax.grid(True, axis="y", alpha=0.25) + ax.legend() fig.tight_layout() fig.savefig(path, dpi=150)