#!/usr/bin/env python3 """ Stablecoin Treasury Test -- Paper 7, Settlement Feedback ========================================================= Lean: CESProofs/Applications/SettlementFeedback.lean (safety_zone_shrinking, mp_effectiveness_decreasing) Tests whether stablecoin market cap flows move US Treasury yields. Two channels with opposite signs at different timescales: 1. ABSORPTION channel (weekly): stablecoin issuers buy T-bills as backing -> yields DOWN (negative beta). Operates only when yields > 0 (boundary condition: no absorption at zero lower bound). 2. TRIFFIN/supply channel (structural): growing stablecoin demand forces more Treasury issuance -> yields UP (positive beta). Sub-tests: A. Full-sample FD OLS: d_DGS3MO ~ d_ln_mcap + controls (baseline) B. Granger causality: lagged d_ln_mcap -> d_DGS3MO C. Post-hike FD: absorption in non-zero-yield regime (Mar 2022+) D. Post-hike FD + WALCL: absorption after controlling for QE/QT E. Subsample stability + Chow structural break test F. Maturity gradient: |beta| should decrease with maturity G. Triffin channel: quarterly d_DGS2 ~ d_ln_mcap (supply-side) Review Notes (2026-03-12) ------------------------- Theory match: PARTIAL -- Lean theorems prove structural capacity/effectiveness, not weekly absorption mechanism. Absorption requires positive yields. I/O audit: CORRECT -- mcap (independent) -> yield (dependent), correct signs. Data-signal alignment: ADEQUATE -- total mcap includes non-Treasury-backed (~75% signal). 319 weekly obs. Code quality: CLEAN -- Newey-West HAC SEs for autocorrelation; one-tailed p-values; Bonferroni for Granger; sign checks throughout. Confidence in verdict: LOW -- full-sample FD is pooling artifact (pre-hike wrong sign); post-hike signal is strong but conditional on yield regime. """ import warnings import numpy as np import pandas as pd from scipy import stats from common.config import COMPILED_DIR, RESULTS_DIR from common.data import load_compiled, load_fred, load_fred_multi from common.plotting import setup_figure, setup_subplots, save_figure from common.stats import verdict, log_result warnings.filterwarnings('ignore') # Constants SPLIT_DATE = '2022-03-16' # First Fed rate hike HAC_LAGS = 4 # Newey-West lag truncation for weekly data CONTROLS = ['d_VIXCLS', 'd_DFF'] MATURITIES = [ ('d_DGS1MO', 'DGS1MO', '1 month'), ('d_DGS3MO', 'DGS3MO', '3 month'), ('d_DGS1', 'DGS1', '1 year'), ('d_DGS2', 'DGS2', '2 year'), ] def load_stablecoin_data(): """Load stablecoin market cap data from compiled.""" try: return load_compiled('stablecoin_treasury_weekly') except FileNotFoundError: try: return load_compiled('stablecoin_mcap') except FileNotFoundError: return None def _fd_regression(data, y_col, x_controls, hac_lags=HAC_LAGS): """ Run first-difference OLS with Newey-West (HAC) standard errors. Returns dict with beta, p_two, p_one (for negative-beta theory), sign_ok, r2, n, model — or None if insufficient data. """ import statsmodels.api as sm cols = ['d_ln_mcap', y_col] + x_controls available = [c for c in cols if c in data.columns] if len(available) < len(cols): return None sub = data[cols].dropna() if len(sub) < 30: return None X = sm.add_constant(sub[['d_ln_mcap'] + x_controls]) y = sub[y_col] model = sm.OLS(y, X).fit(cov_type='HAC', cov_kwds={'maxlags': hac_lags}) b = model.params.get('d_ln_mcap', np.nan) p_two = model.pvalues.get('d_ln_mcap', np.nan) sign_ok = b < 0 p_one = p_two / 2 if sign_ok else 1 - p_two / 2 return { 'beta': float(b), 'p_two': float(p_two), 'p_one': float(p_one), 'sign_ok': sign_ok, 'r2': float(model.rsquared), 'n': len(sub), 'model': model, } _cache = {} _SKIP_DESC_FD = ( "First-difference regression testing the absorption channel: increases " "in stablecoin market cap mean issuers buy more T-bills as backing, " "pushing yields DOWN (negative beta). OLS with Newey-West HAC SEs on " "weekly data controlling for VIX and fed funds rate." ) def main(): results = [] stbl = load_stablecoin_data() if stbl is None: return [log_result('stablecoin_treasury_fd', np.nan, np.nan, 'SKIPPED', details={'reason': 'Stablecoin data not found'}, description=_SKIP_DESC_FD)] # Try to load FRED yield data try: yields = load_fred_multi(['DGS3MO', 'DGS1MO', 'DGS1', 'DGS2', 'VIXCLS', 'DFF'], start='2020-01-01') except FileNotFoundError: if 'DGS3MO' not in stbl.columns: return [log_result('stablecoin_treasury_fd', np.nan, np.nan, 'SKIPPED', details={'reason': 'FRED yield data not found'}, description=_SKIP_DESC_FD)] yields = None # Build weekly panel if 'ln_mcap' in stbl.columns and 'd_ln_mcap' in stbl.columns: weekly = stbl.copy() elif yields is not None: if 'date' in stbl.columns: stbl.index = pd.to_datetime(stbl['date']) stbl_w = stbl.resample('W-FRI').last() fred_w = yields.resample('W-FRI').last() weekly = stbl_w.join(fred_w, how='inner') if 'stablecoin_mcap' in weekly.columns: weekly['ln_mcap'] = np.log(weekly['stablecoin_mcap']) weekly['d_ln_mcap'] = weekly['ln_mcap'].diff() for col in ['DGS3MO', 'DGS1MO', 'DGS1', 'DGS2', 'VIXCLS', 'DFF']: if col in weekly.columns: weekly[f'd_{col}'] = weekly[col].diff() weekly = weekly.dropna() else: weekly = stbl.copy() # Ensure DatetimeIndex (data loads with RangeIndex) if 'date' in weekly.columns and not isinstance(weekly.index, pd.DatetimeIndex): weekly.index = pd.to_datetime(weekly['date']) fd_cols = ['d_ln_mcap', 'd_DGS3MO'] + CONTROLS # ── Test A: Full-sample FD (HAC SEs, baseline) ────────────────────── if all(c in weekly.columns for c in fd_cols): try: res_a = _fd_regression(weekly, 'd_DGS3MO', CONTROLS) if res_a: verd_a = verdict(res_a['p_one'], res_a['sign_ok']) results.append(log_result( 'stablecoin_treasury_fd', res_a['beta'], res_a['p_one'], verd_a, details={'n_obs': res_a['n'], 'r2': res_a['r2'], 'se_type': 'HAC(4)'}, description=( "Tests absorption channel (Paper 7, Lean: " "safety_zone_shrinking): stablecoin issuers buy T-bills " "as backing, pushing yields DOWN. First-difference OLS: " "d(DGS3MO) ~ b*d(ln mcap) + d(VIX) + d(DFF), " "Newey-West HAC(4) SEs, one-tailed (theory: b < 0). " "Data: weekly stablecoin total mcap and FRED Treasury " "yields, Jan 2020 - Feb 2026. Prediction: b < 0 " "(p < 0.05). NOTE: full-sample pools zero-yield and " "positive-yield regimes; see post-hike test for clean signal." ))) _cache['fd_full'] = res_a except Exception as e: results.append(log_result('stablecoin_treasury_fd', np.nan, np.nan, 'SKIPPED', details={'reason': str(e)}, description=_SKIP_DESC_FD)) else: results.append(log_result('stablecoin_treasury_fd', np.nan, np.nan, 'SKIPPED', details={'reason': 'Missing FD columns'}, description=_SKIP_DESC_FD)) # ── Test B: Granger causality ─────────────────────────────────────── gc_cols = ['d_ln_mcap', 'd_DGS3MO'] if all(c in weekly.columns for c in gc_cols): try: from statsmodels.tsa.stattools import grangercausalitytests from statsmodels.tsa.api import VAR gc_data = weekly[gc_cols].dropna() gc_array = gc_data[['d_DGS3MO', 'd_ln_mcap']].values gc_out = grangercausalitytests(gc_array, maxlag=4, verbose=False) min_p = min(gc_out[l][0]['ssr_ftest'][1] for l in range(1, 5)) best_lag = min(range(1, 5), key=lambda l: gc_out[l][0]['ssr_ftest'][1]) best_f = gc_out[best_lag][0]['ssr_ftest'][0] bonf_p = min(min_p * 4, 1.0) # Check sign via VAR try: var_fit = VAR(gc_data).fit(maxlags=best_lag) mcap_coefs = [var_fit.params.loc[f'L{i}.d_ln_mcap', 'd_DGS3MO'] for i in range(1, best_lag + 1)] coef_sum = sum(mcap_coefs) sign_ok = coef_sum < 0 except Exception: sign_ok = True coef_sum = np.nan mcap_coefs = [] verd_b = verdict(bonf_p, sign_ok) results.append(log_result( 'stablecoin_treasury_granger_yield', best_f, bonf_p, verd_b, details={ 'best_lag': best_lag, 'raw_min_p': float(min_p), 'bonferroni_p': float(bonf_p), 'lag_coef_sum': float(coef_sum) if np.isfinite(coef_sum) else None, 'lag_coefs': [float(c) for c in mcap_coefs] if mcap_coefs else None, }, description=( "Granger causality: do lagged stablecoin mcap changes " "predict Treasury yield changes? SSR F-test with lags " "1-4 on weekly first-differenced data, Bonferroni-corrected " "for 4 lags. VAR coefficient sign checked (theory predicts " "negative sum = absorption channel). Data: weekly " "d(ln mcap) and d(DGS3MO). Prediction: significant " "Granger causality with negative coefficient sum." ))) except Exception: results.append(log_result('stablecoin_treasury_granger_yield', np.nan, np.nan, 'SKIPPED', description="Granger causality test for settlement feedback.")) else: results.append(log_result('stablecoin_treasury_granger_yield', np.nan, np.nan, 'SKIPPED', description="Granger causality test for settlement feedback.")) # ── Test C: Post-hike FD (primary absorption signal) ──────────────── if all(c in weekly.columns for c in fd_cols): try: post = weekly[weekly.index >= SPLIT_DATE] res_c = _fd_regression(post, 'd_DGS3MO', CONTROLS) if res_c: verd_c = verdict(res_c['p_one'], res_c['sign_ok']) results.append(log_result( 'stablecoin_treasury_fd_posthike', res_c['beta'], res_c['p_one'], verd_c, details={'n_obs': res_c['n'], 'r2': res_c['r2'], 'se_type': 'HAC(4)', 'period': f'{SPLIT_DATE} onwards'}, description=( "Primary absorption channel test restricted to post-Fed-" "hiking regime (Mar 2022+) where yields > 0 and the " "absorption mechanism operates. FD OLS: d(DGS3MO) ~ " "b*d(ln mcap) + d(VIX) + d(DFF), Newey-West HAC(4) SEs, " "one-tailed. The full-sample result pools zero-yield " "(pre-2022) and positive-yield regimes, diluting the " "signal. Prediction: b < 0 (p < 0.05)." ))) _cache['fd_posthike'] = res_c except Exception: pass # ── Test D: Post-hike FD + WALCL (QE/QT control) ─────────────────── if all(c in weekly.columns for c in fd_cols): try: import statsmodels.api as sm walcl = load_fred('WALCL') if walcl is not None and not walcl.empty: walcl_w = walcl.resample('W-FRI').last() walcl_w.name = 'WALCL' wk_walcl = weekly.join(walcl_w, how='left') wk_walcl['ln_WALCL'] = np.log(wk_walcl['WALCL'].clip(lower=1)) wk_walcl['d_ln_WALCL'] = wk_walcl['ln_WALCL'].diff() # Post-hike only (absorption requires yields > 0) post_w = wk_walcl[wk_walcl.index >= SPLIT_DATE] walcl_controls = CONTROLS + ['d_ln_WALCL'] res_d = _fd_regression(post_w, 'd_DGS3MO', walcl_controls) if res_d: b_walcl = res_d['model'].params.get('d_ln_WALCL', np.nan) p_walcl = res_d['model'].pvalues.get('d_ln_WALCL', np.nan) verd_d = verdict(res_d['p_one'], res_d['sign_ok']) results.append(log_result( 'stablecoin_treasury_fd_walcl', res_d['beta'], res_d['p_one'], verd_d, details={ 'n_obs': res_d['n'], 'r2': res_d['r2'], 'beta_walcl': float(b_walcl), 'p_walcl': float(p_walcl), 'se_type': 'HAC(4)', 'period': f'{SPLIT_DATE} onwards', }, description=( "Absorption channel with QE/QT control (post-hike only): " "d(DGS3MO) ~ b*d(ln mcap) + d(VIX) + d(DFF) + d(ln WALCL). " "WALCL = Fed total assets (controls for balance sheet " "confound on Treasury yields). Newey-West HAC(4) SEs, " "one-tailed, Mar 2022+. Prediction: b < 0 after " "controlling for QE/QT." ))) _cache['walcl_model'] = { 'b_mcap': float(res_d['beta']), 'p_mcap': float(res_d['p_one']), 'b_walcl': float(b_walcl), 'p_walcl': float(p_walcl), 'r2': res_d['r2'], 'n': res_d['n'], } except Exception: pass # ── Test E: Subsample stability + Chow structural break ───────────── if all(c in weekly.columns for c in fd_cols): try: import statsmodels.api as sm sub_all = weekly[fd_cols].dropna() pre = sub_all[sub_all.index < SPLIT_DATE] post = sub_all[sub_all.index >= SPLIT_DATE] betas_sub = {} rss_parts = 0.0 for label, chunk in [('pre_hike', pre), ('post_hike', post)]: if len(chunk) < 20: continue X_s = sm.add_constant(chunk[['d_ln_mcap'] + CONTROLS]) y_s = chunk['d_DGS3MO'] m_s = sm.OLS(y_s, X_s).fit() b_s = m_s.params.get('d_ln_mcap', np.nan) p2_s = m_s.pvalues.get('d_ln_mcap', np.nan) betas_sub[label] = { 'beta': float(b_s), 'p_two': float(p2_s), 'p_one': float(p2_s / 2 if b_s < 0 else 1 - p2_s / 2), 'n': len(chunk), 'r2': float(m_s.rsquared), } rss_parts += m_s.ssr if len(betas_sub) == 2: # Chow F-test for structural break X_full = sm.add_constant(sub_all[['d_ln_mcap'] + CONTROLS]) y_full = sub_all['d_DGS3MO'] m_full = sm.OLS(y_full, X_full).fit() rss_pooled = m_full.ssr k = X_full.shape[1] # number of parameters n = len(sub_all) chow_f = ((rss_pooled - rss_parts) / k) / (rss_parts / (n - 2 * k)) chow_p = 1 - stats.f.cdf(chow_f, k, n - 2 * k) both_neg = all(v['beta'] < 0 for v in betas_sub.values()) if both_neg: sv = 'CONSISTENT' elif chow_p < 0.05: sv = 'INCONSISTENT' else: sv = 'AMBIGUOUS' results.append(log_result( 'stablecoin_treasury_subsample', chow_f, chow_p, sv, details={**betas_sub, 'chow_F': float(chow_f), 'chow_p': float(chow_p)}, description=( "Subsample stability with Chow structural break test at " "Mar 2022 (first Fed rate hike). Runs absorption FD " "separately pre-hike (Jan 2020 - Mar 2022) and post-hike " "(Mar 2022+). CONSISTENT if b < 0 in both subsamples. " "Chow F-test checks whether coefficients differ across " "regimes. Key finding: absorption requires positive yields " "-- pre-2022 (yields near 0) shows wrong sign." ))) _cache['subsample'] = betas_sub _cache['chow'] = {'F': float(chow_f), 'p': float(chow_p)} except Exception: pass # ── Test F: Maturity gradient (post-hike) ─────────────────────────── # Theory: stablecoin issuers primarily buy short-term T-bills, so # absorption should be strongest at shortest maturities. available_mats = [(dcol, col, label) for dcol, col, label in MATURITIES if dcol in weekly.columns] if (len(available_mats) >= 2 and all(c in weekly.columns for c in ['d_ln_mcap'] + CONTROLS)): try: post = weekly[weekly.index >= SPLIT_DATE] mat_betas = [] for dcol, col, label in available_mats: res_m = _fd_regression(post, dcol, CONTROLS) if res_m: mat_betas.append({ 'maturity': label, 'dcol': dcol, 'beta': res_m['beta'], 'p_one': res_m['p_one'], 'n': res_m['n'], 'r2': res_m['r2'], }) if len(mat_betas) >= 2: from scipy.stats import spearmanr betas_ordered = [m['beta'] for m in mat_betas] # Theory predicts TWO channels: absorption (short end, # beta < 0) and Triffin (long end, beta > 0). The maturity # gradient prediction is: (1) short-end absorption exists, # (2) effect weakens with maturity (beta increases). # At long maturities Triffin may dominate (beta > 0). short_betas = [m for m in mat_betas if m['maturity'] in ('1 month', '3 month')] short_neg = all(m['beta'] < 0 for m in short_betas) any_short_sig = any(m['p_one'] < 0.05 for m in short_betas) # Spearman: positive r = beta increases with maturity ranks = list(range(len(mat_betas))) r_sp, _ = spearmanr(ranks, betas_ordered) if short_neg and any_short_sig and r_sp > 0: sv_m = 'CONSISTENT' elif short_neg and any_short_sig: sv_m = 'AMBIGUOUS' else: sv_m = 'INCONSISTENT' results.append(log_result( 'stablecoin_treasury_maturity', r_sp, np.nan, sv_m, details={ 'maturity_betas': mat_betas, 'short_end_negative': short_neg, 'short_end_significant': any_short_sig, 'spearman_r': float(r_sp), 'period': f'{SPLIT_DATE} onwards', }, description=( "Maturity gradient test (post-hike): stablecoin issuers " "primarily buy short-term T-bills, so absorption (b < 0) " "should be strongest at short maturities and weaken with " "maturity. At long maturities the Triffin channel (b > 0) " "may dominate. Runs absorption FD on 4 yield maturities " "(1MO, 3MO, 1Y, 2Y). CONSISTENT if short-end b < 0 " "(significant) and Spearman rank-maturity correlation " "is positive (effect weakens). Newey-West HAC(4) SEs." ))) _cache['maturity'] = mat_betas except Exception: pass # ── Test G: Triffin channel (quarterly, structural) ───────────────── # Theory: growing stablecoin demand -> more Treasury issuance -> # higher long-term yields. Opposite sign from absorption. if 'stablecoin_mcap' in weekly.columns and 'DGS2' in weekly.columns: try: import statsmodels.api as sm qtr = weekly[['stablecoin_mcap', 'DGS2', 'DFF']].resample( 'QE').last().dropna() qtr['ln_mcap'] = np.log(qtr['stablecoin_mcap']) qtr['d_ln_mcap_q'] = qtr['ln_mcap'].diff() qtr['d_DGS2_q'] = qtr['DGS2'].diff() qtr['d_DFF_q'] = qtr['DFF'].diff() qtr = qtr.dropna() if len(qtr) >= 10: X_q = sm.add_constant(qtr[['d_ln_mcap_q', 'd_DFF_q']]) y_q = qtr['d_DGS2_q'] m_q = sm.OLS(y_q, X_q).fit( cov_type='HAC', cov_kwds={'maxlags': 2}) b_q = m_q.params.get('d_ln_mcap_q', np.nan) p2_q = m_q.pvalues.get('d_ln_mcap_q', np.nan) # Triffin channel predicts POSITIVE beta sign_ok_q = b_q > 0 p_q = p2_q / 2 if sign_ok_q else 1 - p2_q / 2 verd_q = verdict(p_q, sign_ok_q) results.append(log_result( 'stablecoin_treasury_triffin', b_q, p_q, verd_q, details={ 'n_quarters': len(qtr), 'r2': float(m_q.rsquared), 'beta_dff': float(m_q.params.get('d_DFF_q', np.nan)), 'se_type': 'HAC(2)', }, description=( "Tests the Triffin/supply channel (Paper 7): growing " "stablecoin demand forces more Treasury issuance, pushing " "long-term yields UP. Quarterly FD OLS: d(DGS2) ~ " "b*d(ln mcap) + d(DFF), Newey-West HAC(2) SEs, " "one-tailed. Data: quarterly stablecoin mcap and 2Y " "yield. Prediction: b > 0 (p < 0.05). " "N=24 quarters -- underpowered." ))) _cache['triffin'] = { 'qtr_dates': qtr.index.tolist(), 'd_ln_mcap_q': qtr['d_ln_mcap_q'].values, 'd_DGS2_q': qtr['d_DGS2_q'].values, 'beta': float(b_q), 'p': float(p_q), 'r2': float(m_q.rsquared), 'n': len(qtr), } except Exception: pass # ── Rolling window beta (diagnostic, for figure only) ─────────────── if all(c in weekly.columns for c in ['d_ln_mcap', 'd_DGS3MO'] + CONTROLS): try: import statsmodels.api as sm WINDOW = 52 # 1-year rolling window sub_r = weekly[['d_ln_mcap', 'd_DGS3MO'] + CONTROLS].dropna() if len(sub_r) >= WINDOW + 20: roll_dates, roll_betas, roll_lo, roll_hi = [], [], [], [] for i in range(WINDOW, len(sub_r)): chunk = sub_r.iloc[i - WINDOW:i] X_r = sm.add_constant(chunk[['d_ln_mcap'] + CONTROLS]) y_r = chunk['d_DGS3MO'] m_r = sm.OLS(y_r, X_r).fit( cov_type='HAC', cov_kwds={'maxlags': 4}) b_r = m_r.params.get('d_ln_mcap', np.nan) ci = m_r.conf_int().loc['d_ln_mcap'] roll_dates.append(sub_r.index[i]) roll_betas.append(float(b_r)) roll_lo.append(float(ci[0])) roll_hi.append(float(ci[1])) _cache['rolling'] = { 'dates': roll_dates, 'betas': roll_betas, 'lo': roll_lo, 'hi': roll_hi, } except Exception: pass _cache['results'] = results _cache['weekly'] = weekly return results def figures(): """Generate stablecoin-treasury figures.""" if not _cache: main() paths = [] weekly = _cache.get('weekly') if weekly is None or len(weekly) == 0: return paths from scipy import stats as sp_stats # ── Figure 1: Time series (mcap vs yield) ────────────────────────── has_mcap = 'stablecoin_mcap' in weekly.columns or 'ln_mcap' in weekly.columns has_yield = 'DGS3MO' in weekly.columns if has_mcap and has_yield: fig, ax1 = setup_figure(width=12, height=6) color_mcap = '#1f77b4' color_yield = '#d62728' mcap_col = ('stablecoin_mcap' if 'stablecoin_mcap' in weekly.columns else 'ln_mcap') ax1.plot(weekly.index, weekly[mcap_col], color=color_mcap, linewidth=1.2, alpha=0.85, label='Stablecoin Mcap') ax1.set_xlabel('Date') ax1.set_ylabel('Stablecoin Market Cap', color=color_mcap) ax1.tick_params(axis='y', labelcolor=color_mcap) # Mark regime break ax1.axvline(pd.Timestamp(SPLIT_DATE), color='gray', linewidth=1.5, linestyle='--', alpha=0.7) ax1.annotate('Fed hike', xy=(pd.Timestamp(SPLIT_DATE), ax1.get_ylim()[1]), fontsize=8, color='gray', ha='center', va='top') ax2 = ax1.twinx() ax2.plot(weekly.index, weekly['DGS3MO'], color=color_yield, linewidth=1.2, alpha=0.85, label='DGS3MO') ax2.set_ylabel('3-Month Treasury Yield (%)', color=color_yield) ax2.tick_params(axis='y', labelcolor=color_yield) lines1, labels1 = ax1.get_legend_handles_labels() lines2, labels2 = ax2.get_legend_handles_labels() ax1.legend(lines1 + lines2, labels1 + labels2, loc='upper left') fig.suptitle('Stablecoin Market Cap vs. Treasury Yield', fontsize=13) fig.tight_layout() paths += save_figure(fig, 'stablecoin_treasury_timeseries', subfolder='applications') # ── Figure 2: Regression scatter (full sample) ────────────────────── fd_cols = ['d_ln_mcap', 'd_DGS3MO'] if all(c in weekly.columns for c in fd_cols): sub = weekly[fd_cols].dropna() if len(sub) > 10: fig, ax = setup_figure(width=8, height=6) # Color-code pre vs post hike pre_mask = sub.index < SPLIT_DATE ax.scatter(sub.loc[pre_mask, 'd_ln_mcap'], sub.loc[pre_mask, 'd_DGS3MO'], alpha=0.3, s=12, color='steelblue', label='Pre-hike') ax.scatter(sub.loc[~pre_mask, 'd_ln_mcap'], sub.loc[~pre_mask, 'd_DGS3MO'], alpha=0.3, s=12, color='coral', label='Post-hike') slope, intercept, r_val, p_val, se = sp_stats.linregress( sub['d_ln_mcap'], sub['d_DGS3MO']) x_range = np.linspace(sub['d_ln_mcap'].min(), sub['d_ln_mcap'].max(), 100) ax.plot(x_range, intercept + slope * x_range, 'k-', linewidth=2, label=f'Full: b={slope:.3f} (p={p_val:.4f})') ax.axhline(0, color='gray', linewidth=0.5, linestyle='--') ax.axvline(0, color='gray', linewidth=0.5, linestyle='--') ax.set_xlabel('d ln(Stablecoin Mcap)') ax.set_ylabel('d DGS3MO (pp)') ax.set_title('Stablecoin Mcap Growth vs Treasury Yield Changes') ax.legend(fontsize=9) paths += save_figure(fig, 'stablecoin_treasury_regression', subfolder='applications') # ── Figure 3: Rolling window beta ─────────────────────────────────── rolling = _cache.get('rolling') if rolling and len(rolling['dates']) > 0: fig, ax = setup_figure(width=12, height=5) dates = rolling['dates'] betas = rolling['betas'] lo = rolling['lo'] hi = rolling['hi'] ax.plot(dates, betas, color='#1f77b4', linewidth=1.5, label='52-week rolling b') ax.fill_between(dates, lo, hi, alpha=0.2, color='#1f77b4', label='95% CI') ax.axhline(0, color='black', linewidth=1, linestyle='-') ax.axvline(pd.Timestamp(SPLIT_DATE), color='red', linewidth=1.5, linestyle='--', alpha=0.7) ax.annotate('Fed hike\nMar 2022', xy=(pd.Timestamp(SPLIT_DATE), 0), fontsize=9, color='red', ha='right', va='bottom', xytext=(-10, 5), textcoords='offset points') ax.set_xlabel('Date (end of 52-week window)') ax.set_ylabel('b (absorption coefficient)') ax.set_title('Rolling 52-Week Absorption Coefficient (HAC SEs)') ax.legend(fontsize=9) ax.grid(True, alpha=0.3) fig.tight_layout() paths += save_figure(fig, 'stablecoin_treasury_rolling', subfolder='applications') # ── Figure 4: Subsample bar chart ─────────────────────────────────── sub_data = _cache.get('subsample') chow_data = _cache.get('chow') if sub_data and len(sub_data) == 2: fig, ax = setup_figure(width=7, height=5) labels = ['Pre-Hike\n(Jan 2020 - Mar 2022)', 'Post-Hike\n(Mar 2022 - Feb 2026)'] betas = [sub_data['pre_hike']['beta'], sub_data['post_hike']['beta']] colors = ['steelblue', 'coral'] ax.bar(labels, betas, color=colors, edgecolor='black', linewidth=0.5) ax.axhline(0, color='gray', linewidth=1, linestyle='--') for i, (label, v) in enumerate(sub_data.items()): ax.annotate( f"b={v['beta']:.3f}\np={v['p_one']:.3f}\nn={v['n']}", xy=(i, v['beta']), xytext=(0, -15 if v['beta'] > 0 else 15), textcoords='offset points', ha='center', fontsize=9, bbox=dict(boxstyle='round', fc='wheat', alpha=0.5)) if chow_data: ax.annotate( f"Chow F={chow_data['F']:.2f}, p={chow_data['p']:.4f}", xy=(0.5, 0.95), xycoords='axes fraction', fontsize=9, ha='center', va='top', bbox=dict(boxstyle='round', fc='lightyellow', alpha=0.8)) ax.set_ylabel('b (d_ln_mcap coefficient)') ax.set_title('Absorption Channel: Pre vs Post Fed Hiking') ax.grid(True, alpha=0.3, axis='y') paths += save_figure(fig, 'stablecoin_treasury_subsample', subfolder='applications') # ── Figure 5: Maturity gradient ───────────────────────────────────── mat_data = _cache.get('maturity') if mat_data and len(mat_data) >= 2: fig, ax = setup_figure(width=8, height=5) mat_labels = [m['maturity'] for m in mat_data] mat_betas = [m['beta'] for m in mat_data] mat_ps = [m['p_one'] for m in mat_data] colors = ['#2ca02c' if p < 0.05 else '#ff7f0e' if p < 0.10 else '#d62728' for p in mat_ps] bars = ax.bar(mat_labels, mat_betas, color=colors, edgecolor='black', linewidth=0.5) ax.axhline(0, color='gray', linewidth=1, linestyle='--') for i, (b, p) in enumerate(zip(mat_betas, mat_ps)): sig = '***' if p < 0.01 else '**' if p < 0.05 else '*' if p < 0.10 else '' ax.annotate(f"b={b:.3f}{sig}\np={p:.3f}", xy=(i, b), xytext=(0, -12 if b > 0 else 12), textcoords='offset points', ha='center', fontsize=8, bbox=dict(boxstyle='round', fc='wheat', alpha=0.5)) ax.set_xlabel('Yield Maturity') ax.set_ylabel('b (absorption coefficient)') ax.set_title('Absorption by Maturity (Post-Hike, HAC SEs)') ax.annotate('Green=p<0.05, Orange=p<0.10, Red=n.s.', xy=(0.02, 0.02), xycoords='axes fraction', fontsize=8, color='gray') ax.grid(True, alpha=0.3, axis='y') fig.tight_layout() paths += save_figure(fig, 'stablecoin_treasury_maturity', subfolder='applications') # ── Figure 6: Triffin scatter ─────────────────────────────────────── triffin = _cache.get('triffin') if triffin is not None: fig, ax = setup_figure(width=8, height=6) ax.scatter(triffin['d_ln_mcap_q'], triffin['d_DGS2_q'], alpha=0.6, s=40, color='steelblue', edgecolor='black', linewidth=0.5) x_vals = triffin['d_ln_mcap_q'] if len(x_vals) > 2: sl, ic, _, _, _ = sp_stats.linregress( x_vals, triffin['d_DGS2_q']) x_line = np.linspace(x_vals.min(), x_vals.max(), 50) ax.plot(x_line, ic + sl * x_line, 'r-', linewidth=2, label=f"b={triffin['beta']:.3f} (p={triffin['p']:.3f})") ax.axhline(0, color='gray', linewidth=0.5, linestyle='--') ax.axvline(0, color='gray', linewidth=0.5, linestyle='--') ax.set_xlabel('d ln(Stablecoin Mcap) [quarterly]') ax.set_ylabel('d DGS2 (pp) [quarterly]') ax.set_title('Triffin Channel: Quarterly Stablecoin Growth vs 2Y Yield') ax.annotate(f"N={triffin['n']} quarters\nR2={triffin['r2']:.3f}", xy=(0.05, 0.95), xycoords='axes fraction', fontsize=10, va='top', bbox=dict(boxstyle='round', fc='wheat', alpha=0.5)) ax.legend(fontsize=9) ax.grid(True, alpha=0.3) paths += save_figure(fig, 'stablecoin_treasury_triffin', subfolder='applications') return paths if __name__ == '__main__': results = main() for r in results: print(f"{r['name']}: {r['verdict']} " f"(stat={r['statistic']}, p={r['p_value']})")

r2	0.1112
n obs	319
p value	0.0923
statistic	-0.3001

Stablecoin Treasury Fd

Description

Figures

Statistics

Script

▸ Source Code

Reproduction Steps