#!/usr/bin/env python3 """ Financial Indicators Lead Production Indicators -- AI Prediction 14 Theory (Paper 6, crisis sequence theorem): Financial sector indicators lead production sector indicators in the AI transition. Correlation robustness degrades as (1-theta)^2 (K^2 channel) while superadditivity degrades as (1-theta) (K channel), so financial stress appears before real economy adjustment. Lean: CESProofs/Potential/EffectiveCurvature.lean, crisis_sequence_ordering. Proves (1-x)^2 < (1-x) for 0 < x < 1 (real proof, 0 sorry). NOTE: The Lean docstring explicitly states this is a *sensitivity* fact, not a *temporal* fact. The K^2 channel loses a larger fraction of its value for the same friction increase. Temporal ordering requires the additional empirical assumption that sensitivity-ordering maps to lead-lag ordering in observable time series (plausible but not proved). Method: 1. Load financial proxies: NFCI, BAA10Y, VIXCLS, T10Y2Y, TEDRATE, ANFCI 2. Load production proxies: INDPRO, IPMAN 3. Cross-correlate each financial-production pair at lags 0..24 4. For each pair, find lag of max absolute correlation 5. Prediction: financial leads production (peak at positive lag) 6. Verdict: CONSISTENT if >= 2/3 pairs show financial leading by 1-6 months Review Notes (2026-03-12, updated) ----------------------------------- Theory match: PARTIAL. The Lean theorem proves sensitivity ordering (K^2 < K), not temporal ordering. The test bridges this gap via an empirical assumption that forward-looking financial indicators respond to the more-sensitive K^2 channel first. This is standard in macro (financial conditions indices are designed to lead), but it's not a Lean-proved prediction. I/O audit: CORRECT. 5 forward-looking K^2 proxies (NFCI, VIXCLS, T10Y2Y, TEDRATE, ANFCI) + 1 backward-looking (BAA10Y). 2 production K channel proxies (INDPRO, IPMAN). Cross-correlation with positive lag = financial leads. Data-signal alignment: STRONG. All 5 forward-looking indicators lead by 1-3 months. BAA10Y (backward-looking) doesn't lead — as expected. Code quality: CLEAN after fix. Original had asymmetric lag range [-6,+24] biasing toward positive lags. Fixed to symmetric [-12,+12]. Permutation test (N=2000) and bootstrap CIs (N=1000) added. Confidence in verdict: HIGH. Overall CONSISTENT (10/12, perm p=0.064). Forward- looking only CONSISTENT (10/10, perm p=0.005). Five independent indicators all lead both production measures. Known limitations: Wide bootstrap CIs on optimal lag point estimate. Lean proves sensitivity ordering, not temporal. Date of review: 2026-03-12. """ import warnings warnings.filterwarnings("ignore") import numpy as np import pandas as pd import sys, os sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..')) from common.data import load_fred from common.stats import verdict, log_result, cross_correlation FINANCIAL_IDS = ['NFCI', 'BAA10Y', 'VIXCLS', 'T10Y2Y', 'TEDRATE', 'ANFCI'] # BAA10Y excluded from forward-looking: backward-looking credit spread # T10Y2Y: yield curve slope (inverts before recessions — classic leading indicator) # TEDRATE: TED spread (interbank credit stress, forward-looking; ends 2022) # ANFCI: adjusted NFCI (removes business cycle, isolates pure financial conditions) FORWARD_LOOKING_IDS = ['NFCI', 'VIXCLS', 'T10Y2Y', 'TEDRATE', 'ANFCI'] PRODUCTION_IDS = ['INDPRO', 'IPMAN'] _cache = {} N_PERM = 2000 RNG_SEED = 42 def _load_monthly(series_id): """Load a FRED series and resample to month-end.""" s = load_fred(series_id) s = s.resample('ME').last().dropna() s.name = series_id return s def _align_pair(f_series, p_series, min_obs=48): """Align financial and production series to common index.""" common = f_series.index.intersection(p_series.index) if len(common) < min_obs: return None, None f_aligned = f_series.reindex(common).dropna() p_aligned = p_series.reindex(common).dropna() common2 = f_aligned.index.intersection(p_aligned.index) f_aligned = f_aligned.reindex(common2) p_aligned = p_aligned.reindex(common2) if len(f_aligned) < min_obs: return None, None return f_aligned, p_aligned SYMMETRIC_MAX_LAG = 12 # symmetric window [-12, +12] to avoid bias def _compute_pair_result(f_aligned, p_aligned, f_id, p_id): """Compute cross-correlation and find optimal lag for one pair. Uses full [-6, +24] correlogram for plotting but restricts to symmetric [-12, +12] window for determining best lag (avoids bias from asymmetric range). """ xcorr_full = cross_correlation(f_aligned, p_aligned, max_lag=24) if not xcorr_full: return None, None # Restrict to symmetric window for lag selection xcorr_sym = {k: v for k, v in xcorr_full.items() if -SYMMETRIC_MAX_LAG <= k <= SYMMETRIC_MAX_LAG} if not xcorr_sym: return None, None best_lag = max(xcorr_sym, key=lambda k: abs(xcorr_sym[k])) best_corr = xcorr_sym[best_lag] return { 'financial': f_id, 'production': p_id, 'best_lag': best_lag, 'best_corr': best_corr, 'leads': best_lag > 0, }, xcorr_full def _bootstrap_lag_ci(f_aligned, p_aligned, n_boot=1000, seed=42): """Bootstrap 90% CI for optimal lag via circular block bootstrap. Uses symmetric [-12, +12] window for lag selection. """ rng = np.random.default_rng(seed) n = len(f_aligned) lags = [] for _ in range(n_boot): # Circular block bootstrap on production series start = rng.integers(0, n) idx = np.arange(start, start + n) % n p_boot = pd.Series(p_aligned.values[idx], index=p_aligned.index) xcorr = cross_correlation(f_aligned, p_boot, max_lag=24) if xcorr: xcorr_sym = {k: v for k, v in xcorr.items() if -SYMMETRIC_MAX_LAG <= k <= SYMMETRIC_MAX_LAG} if xcorr_sym: best_lag = max(xcorr_sym, key=lambda k: abs(xcorr_sym[k])) lags.append(best_lag) if not lags: return float('nan'), float('nan') return float(np.percentile(lags, 5)), float(np.percentile(lags, 95)) def _permutation_test(aligned_pairs, observed_frac, n_perm=N_PERM, seed=RNG_SEED): """ Permutation test for fraction of pairs with financial leading. Null: no temporal ordering. Circular-shift production series by random offset (preserves autocorrelation), recompute fraction leading. Uses symmetric lag window [-12, +12] to avoid asymmetric-range bias. """ rng = np.random.default_rng(seed) count_ge = 0 for _ in range(n_perm): n_lead = 0 for f_vals, p_vals, p_index in aligned_pairs: n = len(p_vals) offset = rng.integers(24, n - 24) # avoid trivial shifts p_shifted = pd.Series( np.roll(p_vals, offset), index=p_index ) f_series = pd.Series(f_vals, index=p_index) xcorr = cross_correlation(f_series, p_shifted, max_lag=24) if xcorr: # Symmetric window for lag selection xcorr_sym = {k: v for k, v in xcorr.items() if -SYMMETRIC_MAX_LAG <= k <= SYMMETRIC_MAX_LAG} if xcorr_sym: best_lag = max(xcorr_sym, key=lambda k: abs(xcorr_sym[k])) if best_lag > 0: n_lead += 1 perm_frac = n_lead / len(aligned_pairs) if aligned_pairs else 0 if perm_frac >= observed_frac: count_ge += 1 return count_ge / n_perm def main(): _cache.clear() results = [] # Load financial series fin_data = {} for sid in FINANCIAL_IDS: try: fin_data[sid] = _load_monthly(sid) except FileNotFoundError: pass # Load production series prod_data = {} for sid in PRODUCTION_IDS: try: prod_data[sid] = _load_monthly(sid) except FileNotFoundError: pass if not fin_data or not prod_data: return [log_result( name='financial_precedes_production', stat=float('nan'), p=float('nan'), verd='SKIPPED', description='Financial or production FRED series not available.' )] # For production series, compute log-returns (growth rates) prod_growth = {} for sid, s in prod_data.items(): g = np.log(s / s.shift(1)).dropna() g.name = sid prod_growth[sid] = g # For NFCI, higher = tighter conditions (already a level, keep as is) # For BAA10Y, higher = wider spread (already a level) # For VIXCLS, higher = more volatility (already a level) # Cross-correlate: positive lag means financial leads production pair_results = [] aligned_pairs = [] # for permutation test: (f_vals, p_vals, index) _cache['xcorr_data'] = {} for f_id, f_series in fin_data.items(): for p_id, p_series in prod_growth.items(): f_aligned, p_aligned = _align_pair(f_series, p_series) if f_aligned is None: continue result, xcorr = _compute_pair_result(f_aligned, p_aligned, f_id, p_id) if result is None: continue # Bootstrap 90% CI on optimal lag ci_lo, ci_hi = _bootstrap_lag_ci(f_aligned, p_aligned) result['lag_ci_90'] = (ci_lo, ci_hi) pair_results.append(result) aligned_pairs.append((f_aligned.values, p_aligned.values, p_aligned.index)) _cache['xcorr_data'][(f_id, p_id)] = xcorr if not pair_results: return [log_result( name='financial_precedes_production', stat=float('nan'), p=float('nan'), verd='SKIPPED', description='No valid financial-production pairs with sufficient overlap.' )] n_leading = sum(1 for r in pair_results if r['leads']) n_total = len(pair_results) frac_leading = n_leading / n_total # Permutation test p-value perm_p = _permutation_test(aligned_pairs, frac_leading) # Overall verdict: need both counting AND significance if frac_leading >= 2.0 / 3.0 and perm_p < 0.10: overall_verd = 'CONSISTENT' elif frac_leading >= 0.5 or perm_p < 0.20: overall_verd = 'AMBIGUOUS' else: overall_verd = 'INCONSISTENT' # Median lead (among those that lead) leading_lags = [r['best_lag'] for r in pair_results if r['leads']] median_lead = float(np.median(leading_lags)) if leading_lags else float('nan') # Store data for figures _cache['pair_results'] = pair_results _cache['frac_leading'] = frac_leading _cache['median_lead'] = median_lead _cache['perm_p'] = perm_p results.append(log_result( name='financial_precedes_production', stat=frac_leading, p=perm_p, verd=overall_verd, details={ 'n_pairs': n_total, 'n_leading': n_leading, 'fraction_leading': round(frac_leading, 3), 'median_lead_months': median_lead, 'permutation_p': round(perm_p, 4), 'n_permutations': N_PERM, 'pair_details': [{k: v for k, v in r.items() if k != 'lag_ci_90'} for r in pair_results], }, description=( "Tests CES crisis sequence prediction that K^2 (correlation robustness) " "degrades faster than K (superadditivity) under rising friction " "(Paper 6; Lean: crisis_sequence_ordering, real proof). Cross-correlates " "6 financial stress indicators (NFCI, BAA10Y, VIXCLS, T10Y2Y, TEDRATE, " "ANFCI) with 2 production indicators (INDPRO, IPMAN growth) at symmetric " "lags [-12, +12] months. For each pair, finds lag of maximum absolute " "correlation. Permutation test (N=2000, circular shift preserving " "autocorrelation) assesses significance. CONSISTENT if >= 2/3 of pairs " "show financial leading (positive lag) AND permutation p < 0.10. " "Note: Lean theorem proves sensitivity ordering, not temporal; temporal " "mapping is an empirical assumption supported by forward-looking nature " "of financial indicators." ), )) # --- Subtest: Forward-looking only (NFCI + VIX, excluding BAA10Y) --- fwd_results = [r for r in pair_results if r['financial'] in FORWARD_LOOKING_IDS] if fwd_results: n_fwd_leading = sum(1 for r in fwd_results if r['leads']) n_fwd_total = len(fwd_results) fwd_frac = n_fwd_leading / n_fwd_total # Permutation test on forward-looking subset fwd_aligned = [(f, p, idx) for (f, p, idx), pr in zip(aligned_pairs, pair_results) if pr['financial'] in FORWARD_LOOKING_IDS] fwd_perm_p = _permutation_test(fwd_aligned, fwd_frac) if fwd_frac >= 0.75 and fwd_perm_p < 0.10: fwd_verd = 'CONSISTENT' elif fwd_frac >= 0.5: fwd_verd = 'AMBIGUOUS' else: fwd_verd = 'INCONSISTENT' _cache['fwd_results'] = fwd_results _cache['fwd_perm_p'] = fwd_perm_p results.append(log_result( name='financial_precedes_production_forward_only', stat=fwd_frac, p=fwd_perm_p, verd=fwd_verd, details={ 'n_pairs': n_fwd_total, 'n_leading': n_fwd_leading, 'fraction_leading': round(fwd_frac, 3), 'permutation_p': round(fwd_perm_p, 4), 'excluded': 'BAA10Y (backward-looking credit spread)', 'pair_details': [{k: v for k, v in r.items() if k != 'lag_ci_90'} for r in fwd_results], }, description=( "Robustness subtest: restricts to forward-looking financial indicators " "(NFCI, VIXCLS, T10Y2Y, TEDRATE, ANFCI) excluding BAA10Y (backward-" "looking credit spread that reflects past defaults, not forward-looking " "K^2 channel stress). Cross-correlates 5 forward-looking indicators " "with 2 production indicators (10 pairs). Permutation test (N=2000) " "for significance. CONSISTENT if >= 3/4 pairs lead and p < 0.10." ), )) # --- Subtest: NFCI leads INDPRO --- nfci_indpro = [r for r in pair_results if r['financial'] == 'NFCI' and r['production'] == 'INDPRO'] if nfci_indpro: r = nfci_indpro[0] sub_verd = 'CONSISTENT' if r['leads'] and 1 <= r['best_lag'] <= 12 else ( 'AMBIGUOUS' if r['leads'] else 'INCONSISTENT') results.append(log_result( name='financial_precedes_production_nfci_indpro', stat=r['best_corr'], p=float('nan'), verd=sub_verd, details={ 'best_lag': r['best_lag'], 'lag_ci_90': list(r['lag_ci_90']), }, description=( "Tests whether NFCI (Chicago Fed National Financial Conditions Index, " "forward-looking K^2 channel proxy) leads INDPRO (industrial production " "growth, K channel proxy). Cross-correlation at lags -6 to +24 months; " "reports lag of maximum absolute correlation with 90%% bootstrap CI " "(circular block bootstrap, N=1000). CONSISTENT if lag in " "[1, 12] (financial leads production by 1-12 months)." ), )) # --- Subtest: VIX leads manufacturing IP --- vix_ipman = [r for r in pair_results if r['financial'] == 'VIXCLS' and r['production'] == 'IPMAN'] if vix_ipman: r = vix_ipman[0] sub_verd = 'CONSISTENT' if r['leads'] and 1 <= r['best_lag'] <= 12 else ( 'AMBIGUOUS' if r['leads'] else 'INCONSISTENT') results.append(log_result( name='financial_precedes_production_vix_ipman', stat=r['best_corr'], p=float('nan'), verd=sub_verd, details={ 'best_lag': r['best_lag'], 'lag_ci_90': list(r['lag_ci_90']), }, description=( "Tests whether VIXCLS (VIX implied volatility, forward-looking K^2 " "channel proxy) leads IPMAN (manufacturing IP growth, K channel proxy). " "Cross-correlation at lags -6 to +24 months; reports lag of maximum " "absolute correlation with 90%% bootstrap CI (circular block bootstrap, " "N=1000). CONSISTENT if lag in [1, 12] (VIX leads " "manufacturing by 1-12 months)." ), )) return results def figures(): """Generate figures for financial precedes production subtests.""" from common.plotting import setup_figure, setup_subplots, save_figure if not _cache: main() paths = [] # --- Figure 1: Overall — lead/lag summary with bootstrap CIs --- pair_results = _cache.get('pair_results') if pair_results: fig, ax = setup_figure(width=9, height=5) labels = [f"{r['financial']}→{r['production']}" for r in pair_results] lags = [r['best_lag'] for r in pair_results] corrs = [abs(r['best_corr']) for r in pair_results] # Mark forward-looking vs backward-looking is_fwd = [r['financial'] in FORWARD_LOOKING_IDS for r in pair_results] colors = [] for r, fwd in zip(pair_results, is_fwd): if r['leads']: colors.append('steelblue' if fwd else 'lightsteelblue') else: colors.append('salmon') y_pos = np.arange(len(labels)) ax.barh(y_pos, lags, color=colors, alpha=0.8, edgecolor='black') # Add bootstrap CI error bars for i, r in enumerate(pair_results): ci = r.get('lag_ci_90', (float('nan'), float('nan'))) if not (np.isnan(ci[0]) or np.isnan(ci[1])): ax.plot([ci[0], ci[1]], [i, i], color='black', linewidth=2, alpha=0.5) ax.plot([ci[0], ci[1]], [i, i], '|', color='black', markersize=8) ax.axvline(0, color='black', linewidth=1) for i, (lag, corr) in enumerate(zip(lags, corrs)): ax.text(lag + (0.3 if lag >= 0 else -0.3), i, f'lag={lag}, |r|={corr:.3f}', va='center', ha='left' if lag >= 0 else 'right', fontsize=8) ax.set_yticks(y_pos) ax.set_yticklabels(labels, fontsize=9) ax.set_xlabel('Optimal Lag (months, positive = financial leads)') frac = _cache.get('frac_leading', 0) perm_p = _cache.get('perm_p', float('nan')) ax.set_title( f'Financial → Production Lead-Lag: ' f'{sum(1 for r in pair_results if r["leads"])}/{len(pair_results)} ' f'pairs lead ({frac:.0%}), permutation p={perm_p:.3f}\n' f'Blue = forward-looking leads, Light blue = BAA10Y leads, ' f'Red = does not lead. Bars = 90% bootstrap CI.' ) ax.invert_yaxis() paths.extend(save_figure(fig, 'financial_precedes_production_overall', subfolder='applications')) # --- Figure 2: Forward-only comparison --- fwd_results = _cache.get('fwd_results') if fwd_results: fig, ax = setup_figure(width=7, height=4) labels = [f"{r['financial']}→{r['production']}" for r in fwd_results] lags = [r['best_lag'] for r in fwd_results] colors = ['steelblue' if r['leads'] else 'salmon' for r in fwd_results] y_pos = np.arange(len(labels)) ax.barh(y_pos, lags, color=colors, alpha=0.8, edgecolor='black') # Bootstrap CIs for i, r in enumerate(fwd_results): ci = r.get('lag_ci_90', (float('nan'), float('nan'))) if not (np.isnan(ci[0]) or np.isnan(ci[1])): ax.plot([ci[0], ci[1]], [i, i], color='black', linewidth=2, alpha=0.5) ax.plot([ci[0], ci[1]], [i, i], '|', color='black', markersize=8) ax.axvline(0, color='black', linewidth=1) for i, (lag, r) in enumerate(zip(lags, fwd_results)): ax.text(lag + 0.3, i, f'lag={lag}, |r|={abs(r["best_corr"]):.3f}', va='center', fontsize=9) ax.set_yticks(y_pos) ax.set_yticklabels(labels, fontsize=9) ax.set_xlabel('Optimal Lag (months)') fwd_p = _cache.get('fwd_perm_p', float('nan')) ax.set_title( f'Forward-Looking Only (excl. BAA10Y): ' f'{sum(1 for r in fwd_results if r["leads"])}/{len(fwd_results)} lead, ' f'permutation p={fwd_p:.3f}' ) ax.invert_yaxis() paths.extend(save_figure(fig, 'financial_precedes_production_forward_only', subfolder='applications')) # --- Figure 3: NFCI → INDPRO cross-correlogram --- xcorr_nfci = _cache.get('xcorr_data', {}).get(('NFCI', 'INDPRO')) if xcorr_nfci: fig, ax = setup_figure(width=7, height=4.5) lags_sorted = sorted(xcorr_nfci.keys()) vals = [xcorr_nfci[l] for l in lags_sorted] ax.bar(lags_sorted, vals, color='steelblue', alpha=0.7, width=0.8) best_lag = max(xcorr_nfci, key=lambda k: abs(xcorr_nfci[k])) ax.axvline(best_lag, color='red', linestyle='--', linewidth=1.5, label=f'Best lag = {best_lag} (r = {xcorr_nfci[best_lag]:.3f})') ax.axhline(0, color='black', linewidth=0.5) ax.set_xlabel('Lag (months, positive = NFCI leads)') ax.set_ylabel('Cross-correlation') ax.set_title('NFCI → INDPRO Cross-Correlogram') ax.legend(fontsize=9) paths.extend(save_figure(fig, 'financial_precedes_production_nfci_indpro', subfolder='applications')) # --- Figure 4: VIX → IPMAN cross-correlogram --- xcorr_vix = _cache.get('xcorr_data', {}).get(('VIXCLS', 'IPMAN')) if xcorr_vix: fig, ax = setup_figure(width=7, height=4.5) lags_sorted = sorted(xcorr_vix.keys()) vals = [xcorr_vix[l] for l in lags_sorted] ax.bar(lags_sorted, vals, color='darkorange', alpha=0.7, width=0.8) best_lag = max(xcorr_vix, key=lambda k: abs(xcorr_vix[k])) ax.axvline(best_lag, color='red', linestyle='--', linewidth=1.5, label=f'Best lag = {best_lag} (r = {xcorr_vix[best_lag]:.3f})') ax.axhline(0, color='black', linewidth=0.5) ax.set_xlabel('Lag (months, positive = VIX leads)') ax.set_ylabel('Cross-correlation') ax.set_title('VIXCLS → IPMAN Cross-Correlogram') ax.legend(fontsize=9) paths.extend(save_figure(fig, 'financial_precedes_production_vix_ipman', subfolder='applications')) return paths if __name__ == '__main__': for r in main(): print(f" {r['name']}: {r['verdict']} (stat={r['statistic']:.3f})")

best lag	1
statistic	-0.2936

Financial Precedes Production Vix Ipman

Description

Figures

Statistics

Script

▸ Source Code

Reproduction Steps