Add automatic exports and visualization artifacts (decay/slotting/scenarios/methods)

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"

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<t<taper_T`: `Δr=+{config.SHORT_UP_MAX_BP}bp*(1-t/taper_T)`
+  - `t>=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<t<tl`: 在两端点间线性插值。
+
+## e) EVE与NII(1Y window)的计算公式与实现口径（DF、PV、窗口内accrual）
+- 折现因子：`DF(t)=exp(-r(t)*t)`。
+- EVE口径：`PV = Σ_m amount_m * DF(t_mid,m)`；`ΔEVE = PV_shocked - PV_base`。
+- NII(1Y)口径：
+  - 窗口内计息年化时间：`accrual_dt_m = max(min(1, t_end,m)-t_start,m, 0)`。
+  - `NII_1Y = Σ_m amount_m * r(t_mid,m) * accrual_dt_m`（线性近似）。
+  - `ΔNII_1Y = NII_1Y_shocked - NII_1Y_base`。
+"""
+    path.write_text(text, encoding="utf-8")

src/nmd/repricing_profile.py

@@ -5,11 +5,26 @@
 import pandas as pd
 
 
-def build_decay_profile(weights: pd.Series) -> 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"]]

src/run.py

@@ -11,14 +11,28 @@
 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
 from src.nmd.repricing_profile import build_decay_profile
 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,13 +85,16 @@ 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)
     scenario_results.to_csv(config.SCENARIO_RESULTS_CSV, index=False)
 
     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__":

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)