1"""Nelson–Siegel loadings and daily factor estimation."""2from __future__ import annotations3import numpy as np4import pandas as pd5from scipy.optimize import minimize_scalar67LAMBDA_DL = 0.7308 # per year: curvature loading maximised at 30 months (Diebold–Li 2006, 0.0609 per month × 12)8CMT = {"DGS1MO": 1/12, "DGS3MO": 0.25, "DGS6MO": 0.5, "DGS1": 1.0, "DGS2": 2.0, "DGS3": 3.0, "DGS5": 5.0, "DGS7": 7.0,9 "DGS10": 10.0, "DGS20": 20.0, "DGS30": 30.0}1011def loadings(tau: np.ndarray, lam: float = LAMBDA_DL) -> np.ndarray:12 tau = np.asarray(tau, float)13 x = lam * tau14 l2 = (1 - np.exp(-x)) / x15 l3 = l2 - np.exp(-x)16 return np.column_stack([np.ones_like(tau), l2, l3])1718def fit_day(y: np.ndarray, tau: np.ndarray, lam: float = LAMBDA_DL):19 X = loadings(tau, lam)20 b, res, *_ = np.linalg.lstsq(X, y, rcond=None)21 fit = X @ b22 return b, y - fit2324def fit_panel(yields: pd.DataFrame, lam: float = LAMBDA_DL, min_obs: int = 8) -> tuple[pd.DataFrame, pd.DataFrame]:25 """yields: DataFrame indexed by date, columns = FRED mnemonics (percent). Returns (factors, residuals)."""26 taus = np.array([CMT[c] for c in yields.columns])27 F, R = [], []28 for d, row in yields.iterrows():29 m = row.notna().values30 if m.sum() < min_obs:31 F.append([np.nan] * 3); R.append([np.nan] * len(taus)); continue32 b, r = fit_day(row.values[m].astype(float), taus[m], lam)33 F.append(b); rr = np.full(len(taus), np.nan); rr[m] = r; R.append(rr)34 F = pd.DataFrame(F, index=yields.index, columns=["level", "slope", "curvature"])35 R = pd.DataFrame(R, index=yields.index, columns=yields.columns)36 return F, R3738def estimate_lambda(yields: pd.DataFrame) -> float:39 taus = np.array([CMT[c] for c in yields.columns])40 Y = yields.dropna()41 def sse(lam):42 X = loadings(taus, lam)43 B = np.linalg.lstsq(X, Y.values.T, rcond=None)[0]44 return float(((Y.values.T - X @ B) ** 2).sum())45 return minimize_scalar(sse, bounds=(0.05, 3.0), method="bounded").x4647def read_fred(dirpath) -> pd.DataFrame:48 import pathlib49 frames = []50 for mn in CMT:51 f = pathlib.Path(dirpath) / f"{mn}.csv"52 s = pd.read_csv(f, na_values=["."]).rename(columns={"observation_date": "date"})53 s["date"] = pd.to_datetime(s["date"]); frames.append(s.set_index("date")[mn])54 return pd.concat(frames, axis=1).sort_index()55