Yield Curve Explorer:

A fixed-income research toolkit that bootstraps zero-coupon term structures from US Treasury Constant Maturity (CMT) par yields, compares four interpolation methods, fits Nelson-Siegel and Nelson-Siegel-Svensson parametric models, prices fixed-rate bonds, and stress-tests portfolios under six named rate shock scenarios. Treasury data is pulled from the FRED API and cached in a local DuckDB database; the full curve history from any start date through today is available in under a minute. The six-tab Streamlit dashboard covers bootstrapped curves, forward rates, discount factors, an interpolation comparison, shock scenario analysis, bond analytics (price, YTM, duration, DV01, convexity), a key-rate DV01 ladder, a 25-year yield history chart, and a 3-D yield surface. Built on Python 3.11+ using pandas, numpy, scipy, plotly, streamlit, duckdb, and fredapi; all computation is pure-Python with no external pricing library.

%==========%

I. Streamlit Dashboard (dashboard/app.py):

The dashboard exposes the full module through a browser UI. The sidebar controls the FRED API key, data fetch date range, curve as-of date, interpolation method, shock magnitude, and bond parameters. Six tabs present the output: Zero Curve shows the bootstrapped curve, instantaneous forward rates, and discount factors alongside the bootstrap table; Interpolation renders side-by-side zero and forward curves for all four methods; Shock Scenarios applies user-selected rate scenarios at a configurable magnitude; Bond Analytics shows the full risk report and key-rate DV01 ladder; History plots the yield history and 10Y−2Y spread over time; 3-D Surface renders the yield surface as a Plotly 3-D mesh over the selected date range.

cd assets/projects/yield_curve
python -m venv .venv && .venv\Scripts\Activate.ps1   # Windows
pip install -r requirements.txt
pip install -e .

# Add your free FRED API key to .env
cp .env.example .env
# Edit .env and set FRED_API_KEY=<your key>

streamlit run dashboard/app.py
  

The dashboard below runs entirely in the browser via stlite (Streamlit on WebAssembly — no server required). It uses synthetic US Treasury par yields in four preset shapes (Inverted, Normal, Flat, Humped); all bootstrapping, interpolation, shock, and bond-analytics logic runs client-side. First load downloads Pyodide and may take 20–40 seconds; subsequent loads are cached.

%==========%

II. Project Layout:

yield-curve/
├── requirements.txt
├── .env.example                            # FRED_API_KEY template
├── src/yieldcurve/
│   ├── data/
│   │   ├── fetcher.py                      # FRED API client — pulls DGS* CMT series via fredapi
│   │   └── storage.py                      # DuckDB cache: date × tenor → yield
│   ├── curve/
│   │   ├── bootstrap.py                    # Iterative bootstrap: par yields → zero rates + DFs
│   │   ├── interpolation.py                # Linear, cubic spline, log-linear, cubic log-DF
│   │   └── nelson_siegel.py                # Nelson-Siegel + NSS parametric fit (global optimiser)
│   ├── risk/
│   │   ├── instruments.py                  # FixedRateBond: price, YTM, duration, DV01, convexity
│   │   └── scenarios.py                    # Parallel, steepener, flattener, twist; key-rate DV01
│   └── viz/
│       └── plots.py                        # Reusable Plotly figures (zero curve, forward, surface, …)
├── dashboard/
│   └── app.py                              # Streamlit six-tab dashboard
└── notebooks/
    └── 01_bootstrap_walkthrough.ipynb      # Step-by-step walkthrough with derivations
  

%==========%

III. Data — FRED API & DuckDB Cache (data/fetcher.py, data/storage.py):

Treasury Constant Maturity par yields are fetched from the Federal Reserve Economic Data (FRED) API using the fredapi package. Eleven series are pulled in parallel: DGS1MO, DGS3MO, DGS6MO, DGS1, DGS2, DGS3, DGS5, DGS7, DGS10, DGS20, DGS30. The raw FRED series returns percentages; the fetcher converts them to decimals and drops days with any missing tenor. The result is a wide DataFrame (date × tenor-label) written to a DuckDB file at data/yields.duckdb. The storage.py module provides save_yields(), load_yields(), and cached_date_range() helpers; all DuckDB writes use explicit column lists to avoid column-order fragility.

# fetcher.py
_FRED_SERIES = {
    "1M": "DGS1MO", "3M": "DGS3MO", "6M": "DGS6MO",
    "1Y": "DGS1",   "2Y": "DGS2",   "3Y": "DGS3",
    "5Y": "DGS5",   "7Y": "DGS7",   "10Y": "DGS10",
    "20Y": "DGS20", "30Y": "DGS30",
}

def fetch_treasury_yields(
    start: str | date, end: str | date, api_key: str | None = None
) -> pd.DataFrame:
    """Pull all 11 CMT par-yield series from FRED; return wide DataFrame in decimal."""
    fred = Fred(api_key=api_key or os.environ["FRED_API_KEY"])
    frames = {}
    for label, series_id in _FRED_SERIES.items():
        s = fred.get_series(series_id, observation_start=start, observation_end=end)
        frames[label] = s / 100.0
    df = pd.DataFrame(frames).dropna()
    df.index = pd.to_datetime(df.index)
    return df

def yields_for_date(df: pd.DataFrame, as_of: str | date) -> pd.Series:
    """Return the par yield row closest to but not after as_of."""
    ts = pd.Timestamp(as_of)
    mask = df.index <= ts
    if not mask.any():
        raise ValueError(f"No data on or before {as_of}")
    return df.loc[mask].iloc[-1]
  

# storage.py — DuckDB with explicit column lists
import duckdb, pandas as pd
from pathlib import Path

_DB = Path(__file__).parents[3] / "data" / "yields.duckdb"
_COLS = ["date", "1M", "3M", "6M", "1Y", "2Y", "3Y", "5Y", "7Y", "10Y", "20Y", "30Y"]

def save_yields(df: pd.DataFrame) -> None:
    df = df.reset_index().rename(columns={"index": "date"})
    df["date"] = pd.to_datetime(df["date"]).dt.date
    con = duckdb.connect(str(_DB))
    con.execute("CREATE TABLE IF NOT EXISTS yields (" +
                ", ".join(f'"{c}" DOUBLE' if c != "date" else '"date" DATE' for c in _COLS) + ")")
    con.execute("DELETE FROM yields WHERE date IN (SELECT date FROM df)")
    con.execute(f"INSERT INTO yields ({', '.join(_COLS)}) SELECT {', '.join(_COLS)} FROM df")
    con.close()

def load_yields() -> pd.DataFrame:
    if not _DB.exists():
        return pd.DataFrame()
    con = duckdb.connect(str(_DB), read_only=True)
    df = con.execute(f"SELECT {', '.join(_COLS)} FROM yields ORDER BY date").df()
    con.close()
    df["date"] = pd.to_datetime(df["date"]).dt.date
    return df.set_index("date")
  

%==========%

IV. Bootstrap Methodology (curve/bootstrap.py):

US Treasury CMT par yields are bootstrapped iteratively into zero-coupon discount factors. The procedure follows two rules by instrument type:

T-bills (\(T \le 6\text{M}\)): already zero-coupon discount instruments priced at par with simple interest: $$DF(T) = \frac{1}{1 + c \cdot T}$$

Notes and Bonds (\(T \ge 1\text{Y}\)): semiannual coupon bonds priced at par (\(P = 100\)). Given that coupon dates \(d_1, \ldots, d_{n-1}\) already have known discount factors (log-linearly interpolated for any coupon date that falls between bootstrapped nodes), the unknown \(DF(T)\) is solved from the par-bond pricing equation: $$100 = \frac{c}{2} \cdot 100 \sum_{i=1}^{n-1} DF(d_i) + 100\left(1 + \frac{c}{2}\right) DF(T)$$ $$\Rightarrow\quad DF(T) = \frac{100 - \tfrac{c}{2} \cdot 100 \sum_{i < n} DF(d_i)}{100\left(1 + c/2\right)}$$

The annually compounded zero rate is recovered as \(z(T) = DF(T)^{-1/T} - 1\). Log-linear interpolation inside the bootstrap uses:

# bootstrap.py — core bootstrap loop
@classmethod
def from_par_yields(cls, par_yields: dict[float, float],
                    curve_date: pd.Timestamp | None = None) -> "BootstrappedCurve":
    """par_yields: {maturity_years: par_yield_decimal}"""
    sorted_items = sorted(par_yields.items())
    tenors, zero_rates, discount_factors = [], [], []
    df_cache: dict[float, float] = {}

    for T, c in sorted_items:
        if T <= 6 / 12:                        # T-bill: simple interest discount
            df = 1.0 / (1.0 + c * T)
        else:                                   # Note/Bond: semiannual coupon, at par
            coupon_dates = np.round(np.arange(0.5, T + 1e-9, 0.5), 8)
            pv_known = sum(
                c / 2 * 100 * _interp_df(d, df_cache)
                for d in coupon_dates[:-1]
            )
            df = (100.0 - pv_known) / (100.0 * (1.0 + c / 2.0))

        df_cache[T] = df
        z = df ** (-1.0 / T) - 1.0             # annually compounded zero rate
        tenors.append(T); zero_rates.append(z); discount_factors.append(df)

    ...

def _interp_df(t: float, df_cache: dict[float, float]) -> float:
    """Log-linear interpolation of DF at t from already-bootstrapped nodes."""
    if t in df_cache:
        return df_cache[t]
    known = sorted(df_cache.keys())
    t_lo = max(k for k in known if k <= t)
    t_hi = min(k for k in known if k > t)
    alpha = (t - t_lo) / (t_hi - t_lo)
    log_df = (1 - alpha) * np.log(df_cache[t_lo]) + alpha * np.log(df_cache[t_hi])
    return float(np.exp(log_df))
  

%==========%

V. Interpolation Methods (curve/interpolation.py):

InterpolatedCurve wraps a BootstrappedCurve and fits one of four smooth interpolants to the bootstrapped nodes. The choice of interpolation method has a direct impact on the forward curve: methods that interpolate on zero rates can produce oscillating forwards at the interpolation knots, while methods that work on log-discount-factors enforce the no-arbitrage constraint more naturally.

# interpolation.py
class InterpolationMethod(str, Enum):
    LINEAR_ZERO   = "linear_zero"
    CUBIC_ZERO    = "cubic_zero"
    LOG_LINEAR_DF = "log_linear_df"
    CUBIC_LOG_DF  = "cubic_log_df"   # recommended

class InterpolatedCurve:
    def _build_interpolant(self) -> None:
        t, z, log_df = self.base.tenors, self.base.zero_rates, np.log(self.base.discount_factors)
        m = self.method
        if m == InterpolationMethod.LINEAR_ZERO:
            self._interp = interp1d(t, z, kind="linear", fill_value="extrapolate")
            self._query = lambda ts: self._interp(ts);  self._mode = "zero"
        elif m == InterpolationMethod.CUBIC_ZERO:
            cs = CubicSpline(t, z, bc_type="not-a-knot", extrapolate=True)
            self._query = lambda ts: cs(ts);            self._mode = "zero"
        elif m == InterpolationMethod.LOG_LINEAR_DF:
            self._interp = interp1d(t, log_df, kind="linear", fill_value="extrapolate")
            self._query = lambda ts: np.exp(self._interp(ts)); self._mode = "df"
        elif m == InterpolationMethod.CUBIC_LOG_DF:
            cs = CubicSpline(t, log_df, bc_type="not-a-knot", extrapolate=True)
            self._query = lambda ts: np.exp(cs(ts));    self._mode = "df"

    def zero_rate(self, t: float | np.ndarray) -> float | np.ndarray:
        t_arr = np.atleast_1d(np.asarray(t, dtype=float))
        if self._mode == "zero":
            result = self._query(t_arr)
        else:
            df     = self._query(t_arr)
            result = df ** (-1.0 / t_arr) - 1.0
        return float(result[0]) if np.isscalar(t) else result
  

%==========%

VI. Nelson-Siegel & NSS Parametric Fit (curve/nelson_siegel.py):

The Nelson-Siegel (NS) model expresses the zero rate at maturity \(t\) as a three-factor parametric form: $$r(t) = \beta_0 + (\beta_1 + \beta_2)\frac{\lambda}{t}(1 - e^{-t/\lambda}) - \beta_2 e^{-t/\lambda}$$ The three factors have clear economic interpretations: \(\beta_0\) is the long-run level (\(r(\infty) = \beta_0\)); \(\beta_1\) drives the short end (\(r(0) - r(\infty) = \beta_1\)); \(\beta_2\) creates the curvature or hump; \(\lambda\) controls the decay speed and the maturity at which the hump peaks.

The Nelson-Siegel-Svensson (NSS) extension adds a second curvature term with a separate decay \(\lambda_2\), allowing a second hump in the forward curve: $$r(t) = \beta_0 + \beta_1 \frac{\lambda_1}{t}(1-e^{-t/\lambda_1}) + \beta_2\!\left[\frac{\lambda_1}{t}(1-e^{-t/\lambda_1}) - e^{-t/\lambda_1}\right] + \beta_3\!\left[\frac{\lambda_2}{t}(1-e^{-t/\lambda_2}) - e^{-t/\lambda_2}\right]$$ Both models are fitted by minimising weighted sum-of-squared errors between the parametric curve and the bootstrapped zero rates. A global search via scipy.optimize.differential_evolution is followed by a Nelder-Mead polish. RMSE in basis points is reported as a fit-quality diagnostic.

# nelson_siegel.py
def _ns_rate(t, b0, b1, b2, lam):
    t      = np.where(t < 1e-8, 1e-8, t)
    decay  = np.exp(-t / lam)
    loading = (lam / t) * (1.0 - decay)
    return b0 + (b1 + b2) * loading - b2 * decay

def _ns_forward(t, b0, b1, b2, lam):
    """Instantaneous forward rate: r(t) + t·dr/dt for Nelson-Siegel."""
    t     = np.where(t < 1e-8, 1e-8, t)
    decay = np.exp(-t / lam)
    return b0 + b1 * decay + b2 * (t / lam) * decay

class NelsonSiegelCurve:
    @classmethod
    def fit(cls, tenors, zero_rates, weights=None) -> "NelsonSiegelCurve":
        """Global DE + Nelder-Mead polish over [β₀, β₁, β₂, λ]."""
        def objective(p):
            b0, b1, b2, lam = p
            if lam <= 0:
                return 1e10
            return float(np.sum(w * (_ns_rate(tenors, b0, b1, b2, lam) - zero_rates) ** 2))
        bounds = [(0.0, 0.20), (-0.15, 0.15), (-0.15, 0.15), (0.1, 5.0)]
        de     = differential_evolution(objective, bounds, seed=42, maxiter=500, tol=1e-10)
        result = minimize(objective, de.x, method="Nelder-Mead",
                          options={"maxiter": 10_000, "xatol": 1e-10, "fatol": 1e-12})
        b0, b1, b2, lam = result.x
        rmse = float(np.sqrt(np.mean((_ns_rate(tenors, b0, b1, b2, lam) - zero_rates)**2))) * 10_000
        return cls(NelsonSiegelParams(b0, b1, b2, lam, rmse))
  

%==========%

VII. Bond Analytics (risk/instruments.py):

FixedRateBond represents a generic fixed-rate bond with semiannual coupons and par value of 100. Pricing discounts cash flows using the zero curve rather than a flat yield, so the price correctly incorporates the full term structure. The yield-to-maturity is solved numerically via Brent’s method as the flat yield that reproduces the curve price. Duration and convexity follow standard semiannual-coupon conventions:

# instruments.py
@dataclass
class FixedRateBond:
    coupon_rate: float   # decimal annual coupon (e.g. 0.04)
    maturity:    float   # years to maturity
    par:         float = 100.0

    def price(self, curve) -> float:
        """Discount cash flows from the term structure (not a flat yield)."""
        times, flows = self._cash_flows()
        dfs = np.array([float(curve.discount_factor(t)) for t in times])
        return float(np.sum(flows * dfs))

    def yield_to_maturity(self, curve) -> float:
        """Flat yield (semiannual) that reprices the bond — solved via brentq."""
        px = self.price(curve)
        times, flows = self._cash_flows()
        def pv_diff(y):
            return float(np.sum(flows * (1.0 + y/2.0) ** (-times * 2))) - px
        return brentq(pv_diff, -0.5, 5.0, xtol=1e-10)

    def risk_report(self, curve) -> dict[str, float]:
        px  = self.price(curve)
        ytm = self.yield_to_maturity(curve)
        mac = self.macaulay_duration(curve)
        md  = mac / (1.0 + ytm / 2.0)
        return {"price": px, "ytm_pct": ytm*100, "macaulay_duration": mac,
                "modified_duration": md, "dv01": md*px*0.0001,
                "convexity": self.convexity(curve)}
  

bond_portfolio_dv01_ladder() computes the key-rate DV01 at each standard tenor bucket by applying a +1 bp single-key-rate shift (via ShockEngine.key_rate_shift) and measuring the resulting price change for each bond in the portfolio. The DV01 contributions are summed to produce a bucketed sensitivity profile, the standard tool for hedge ratio construction in fixed-income desks.

%==========%

VIII. Rate Shock Scenarios (risk/scenarios.py):

ShockedCurve wraps a base curve and adds a tenor-dependent additive shift to the zero rate at query time. ShockEngine constructs the shift vector from named scenario profiles anchored at the 2Y (short end) and 10Y (long end) tenors, blended linearly across the curve. All shocked curves expose the same interface as the base curve (zero_rate, discount_factor, forward_rate), so any downstream analytics (bond pricing, DV01) work unchanged.

# scenarios.py
_SHOCK_PROFILES: dict[ShockType, tuple[float, float]] = {
    ShockType.PARALLEL_UP:    (100,   100),
    ShockType.PARALLEL_DOWN:  (-100, -100),
    ShockType.BEAR_STEEPENER: (150,   50),
    ShockType.BULL_STEEPENER: (-150, -50),
    ShockType.BEAR_FLATTENER: (50,   150),
    ShockType.BULL_FLATTENER: (-50, -150),
    ShockType.TWIST:          (75,   -75),
}

class ShockEngine:
    @staticmethod
    def apply(base, shock_type: ShockType, magnitude_bps: float = 100.0) -> ShockedCurve:
        short_bps, long_bps = _SHOCK_PROFILES[shock_type]
        scale        = magnitude_bps / 100.0
        short_shift  = short_bps * scale / 10_000
        long_shift   = long_bps  * scale / 10_000
        # Anchor at 3M (short) and 30Y (long); blend linearly via 2Y and 10Y anchors
        t_nodes = np.array([0.25, 2.0, 10.0, 30.0])
        s_nodes = np.array([short_shift, short_shift, long_shift, long_shift])
        return ShockedCurve(base, dict(zip(t_nodes, s_nodes)))

    @staticmethod
    def key_rate_shift(base, tenor_label: str, shift_bps: float = 1.0) -> ShockedCurve:
        """Bump one key-rate tenor by shift_bps; taper linearly to adjacent nodes."""
        target = _LABEL_TO_YEARS[tenor_label]
        idx    = ALL_TENORS.index(target)
        lo     = ALL_TENORS[idx - 1] if idx > 0 else 0.0
        hi     = ALL_TENORS[idx + 1] if idx < len(ALL_TENORS) - 1 else ALL_TENORS[-1] + 5
        shift_map = {lo: 0.0, target: shift_bps / 10_000, hi: 0.0, 0.0: 0.0, ALL_TENORS[-1]: 0.0}
        return ShockedCurve(base, shift_map)
  

%==========%

IX. Visualization (viz/plots.py):

Eight reusable Plotly figure functions are provided by the viz module and embedded directly in the Streamlit dashboard. All figures use the plotly_white template for a clean presentation.

# plots.py — 3-D surface
def plot_3d_surface(df: pd.DataFrame) -> go.Figure:
    """
    df: wide DataFrame with date index and tenor-label columns (decimal yields).
    Returns a Plotly Surface figure.
    """
    tenor_order = ["1M","3M","6M","1Y","2Y","3Y","5Y","7Y","10Y","20Y","30Y"]
    tenor_years = [1/12, 3/12, 6/12, 1, 2, 3, 5, 7, 10, 20, 30]
    cols = [c for c in tenor_order if c in df.columns]
    z    = df[cols].dropna().values * 100           # percent
    x    = np.array(tenor_years[:len(cols)])        # tenor axis
    y    = np.arange(len(z))                        # date axis (ordinal)
    date_labels = df[cols].dropna().index.strftime("%Y-%m-%d").tolist()

    fig = go.Figure(go.Surface(
        x=x, y=y, z=z,
        colorscale="RdYlGn_r",
        colorbar=dict(title="Yield (%)", len=0.5),
    ))
    fig.update_layout(
        scene=dict(
            xaxis=dict(title="Tenor (yrs)", tickvals=x, ticktext=cols),
            yaxis=dict(title="Date", tickvals=list(range(0, len(y), max(1, len(y)//10))),
                       ticktext=date_labels[::max(1, len(date_labels)//10)]),
            zaxis=dict(title="Yield (%)"),
        ),
        title="US Treasury Yield Surface",
        height=600,
        margin=dict(l=0, r=0, t=40, b=0),
    )
    return fig
  

%==========%

X. Setup & API Usage:

# ── Bootstrap a curve from code ────────────────────────────────────────────────
from yieldcurve.data.fetcher import fetch_treasury_yields, yields_for_date
from yieldcurve.data.storage import load_yields, save_yields
from yieldcurve.curve.bootstrap import BootstrappedCurve
from yieldcurve.curve.interpolation import InterpolatedCurve, InterpolationMethod

# Fetch and cache 25 years of Treasury data (requires FRED_API_KEY in .env)
df = fetch_treasury_yields("2000-01-01", "2024-12-31")
save_yields(df)

# Bootstrap the curve for a specific date
par = yields_for_date(load_yields(), "2024-01-02")
boot  = BootstrappedCurve.from_series(par)
curve = InterpolatedCurve(boot, InterpolationMethod.CUBIC_LOG_DF)

print(f"10Y zero rate:   {curve.zero_rate(10)*100:.4f}%")
print(f"10Y discount DF: {curve.discount_factor(10):.6f}")
print(f"5Y→10Y fwd rate: {curve.forward_rate(5, 10)*100:.4f}%")
  

# ── Bond analytics ─────────────────────────────────────────────────────────────
from yieldcurve.risk.instruments import FixedRateBond, bond_portfolio_dv01_ladder
from yieldcurve.risk.scenarios import ShockEngine, ShockType

bond = FixedRateBond(coupon_rate=0.04, maturity=10)
report = bond.risk_report(curve)
# → {'price': ..., 'ytm_pct': ..., 'macaulay_duration': ...,
#    'modified_duration': ..., 'dv01': ..., 'convexity': ...}

# ── Rate shocks ────────────────────────────────────────────────────────────────
parallel_up  = ShockEngine.apply(curve, ShockType.PARALLEL_UP, 100)
steepener    = ShockEngine.apply(curve, ShockType.BEAR_STEEPENER, 100)
all_shocks   = ShockEngine.all_scenarios(curve, 100)

# Repricing under the bear-steepener scenario
bond_shocked = FixedRateBond(coupon_rate=0.04, maturity=10).price(steepener)

# ── Key-rate DV01 ladder ───────────────────────────────────────────────────────
portfolio = [FixedRateBond(0.04, 5), FixedRateBond(0.025, 10), FixedRateBond(0.035, 30)]
ladder = bond_portfolio_dv01_ladder(portfolio, curve)
# → {'1Y': ..., '2Y': ..., '3Y': ..., '5Y': ..., '7Y': ..., '10Y': ..., '20Y': ..., '30Y': ...}
  

%==========%

XI. Configuration & Data Sources:

Team:

Theodosios Dimitrasopoulos, personal project.

Tools & methods:

Python 3.11, pandas, NumPy, SciPy (cubic splines, Brent’s method, differential evolution), DuckDB (local OLAP yield cache), fredapi (FRED HTTP client), Plotly (interactive figures, 3-D surface), Streamlit (dashboard). Fixed-income methodology: iterative bootstrap of zero-coupon discount factors from CMT par yields; log-linear DF interpolation during bootstrap; four interpolation methods for the fitted curve (linear zero, cubic zero, log-linear DF, cubic log-DF); Nelson-Siegel and Nelson-Siegel-Svensson global parametric fit; bond pricing, YTM (Brent), Macaulay and modified duration, DV01, convexity; duration + convexity Taylor approximation; seven named rate shock scenarios with blended tenor-dependent shifts; key-rate DV01 ladder via single-tenor bumps.