#!/usr/bin/env python3 """ Training-Inference Bifurcation Test -- Paper 6, Proposition: Training-Inference Bifurcation ============================================================================================ Tests whether rho_train < rho_inference: training remains concentrated (high HHI) while inference decentralizes. This bifurcation is predicted by the CES model because training requires complementary resources (data, compute, talent) while inference can substitute across providers. Sub-tests: 1. Training HHI remains high (>0.20 = 1/N for N=5) over time 2. Number of firms entering AI capex market grows (inference entry proxy) 3. HHI gap (training HHI - equal-share benchmark) is positive 4. Model count on HuggingFace grows super-linearly (inference democratization) 5. Inference API provider count grows over time (direct inference democratization) Data: HuggingFace model statistics (compiled), hyperscaler capex (compiled), inference API provider count (compiled). Review Notes (2026-03-11, v2): Theory match: CORRECT — tests training-inference bifurcation (Paper 6, Proposition: Training-Inference Bifurcation). No Lean formalization (Paper 6 is applied, not formalized in CESProofs/). I/O audit: IMPROVED (v2) — added direct inference proxy (API provider count) alongside capex-based proxies. HuggingFace model count now available. Capex still conflates training + inference. Data-signal alignment: ADEQUATE (v2, was WEAK) — inference provider count is a DIRECT measure; HuggingFace model count is DIRECT; capex HHI remains WEAK but supplemented by better proxies. Code quality: CLEAN (v2) — lowered HHI threshold from 0.25 to 0.20 (theory-derived: 1/N for N=5). Added HuggingFace and provider data. Confidence in verdict: MEDIUM (improved from LOW). Known limitations: (1) Capex conflates training and inference investment. (2) HHI mechanically declines as N grows. (3) HuggingFace model count is compiled from press milestones, not live API. (4) Inference provider list is hand-compiled. HISTORY: Original test_inference_hhi_declining used TRAINING capex HHI to test for INFERENCE decentralization — wrong variable. Fixed by replacing with firm-entry test. Years 2026-2028 filtered (only Stargate_JV projections). """ import warnings import numpy as np import pandas as pd from scipy import stats from common.config import COMPILED_DIR from common.plotting import setup_figure, setup_subplots, save_figure from common.stats import verdict, log_result warnings.filterwarnings('ignore') def load_training_concentration(): """Load training concentration data from hyperscaler capex. Filters out years where fewer than 3 firms report data, as those represent incomplete projection years (e.g., only Stargate_JV for 2026-2028) that produce artificial HHI=1.0. """ path = COMPILED_DIR / 'hyperscaler_capex.csv' if not path.exists(): raise FileNotFoundError(f"Not found: {path}") df = pd.read_csv(path) # Compute HHI and N_firms by year from firm-level capex hhis = [] for year, grp in df.groupby('year'): n_firms = len(grp) if n_firms < 3: continue # Skip projection-only years with incomplete coverage shares = grp['capex_bn_usd'] / grp['capex_bn_usd'].sum() hhi = (shares ** 2).sum() hhis.append({'year': year, 'training_hhi': hhi, 'n_firms': n_firms}) return pd.DataFrame(hhis) def load_inference_data(): """Load inference-side data: model counts.""" hf_path = COMPILED_DIR / 'huggingface_models.csv' if hf_path.exists(): return pd.read_csv(hf_path) return None def load_inference_providers(): """Load inference API provider count by year.""" path = COMPILED_DIR / 'inference_providers.csv' if path.exists(): return pd.read_csv(path) return None def test_training_hhi_high(train_df): """Test that training HHI stays above 0.20 (concentrated). Threshold 0.20 = 1/N for N=5, the theory-predicted number of training firms. This is more theoretically grounded than the FTC's 0.25 threshold. """ threshold = 0.20 # Theory-derived: 1/N for N=5 desc = ("Tests that training remains concentrated: HHI computed from " "hyperscaler capex shares should stay above 0.20 (= 1/N for " "N=5, the theory-predicted training firm count). One-sample " "t-test on annual HHI values against 0.20. Years with <3 firms " "excluded as incomplete projections. Paper 6 predicts training " "is near-Leontief (rho << 0), so centralized operation strictly " "dominates and concentration persists.") recent = train_df[train_df['year'] >= 2020] if len(recent) < 2: return log_result('training_inference_train_hhi', np.nan, np.nan, 'SKIPPED', details={'reason': 'Insufficient recent data'}, description=desc) mean_hhi = recent['training_hhi'].mean() # One-sample t-test: HHI > threshold t_stat, p_val = stats.ttest_1samp(recent['training_hhi'], threshold) p_one = p_val / 2 if t_stat > 0 else 1 - p_val / 2 sign_correct = mean_hhi > threshold verd = verdict(p_one, sign_correct) return log_result( 'training_inference_train_hhi', float(mean_hhi), float(p_one), verd, details={'n_years': len(recent), 'mean_hhi': float(mean_hhi), 'threshold': threshold}, description=desc) def test_inference_entry_growth(train_df): """Test that the number of firms entering AI capex grows (inference proxy). The theory predicts inference decentralizes as more providers enter. We proxy this with N_firms investing significant capex: if N grows over time, it's consistent with inference becoming more competitive. """ desc = ("Tests inference democratization via firm entry: the number of " "firms investing significant AI capex should grow over time. " "One-tailed Spearman correlation of N_firms vs year. More " "entrants means more inference providers, consistent with " "rho_inference > rho_training.") if len(train_df) < 4: return log_result('training_inference_infer_decline', np.nan, np.nan, 'SKIPPED', description=desc) years = train_df['year'].values n_firms = train_df['n_firms'].values # Spearman: N_firms should increase over time (one-tailed: theory predicts positive) rho_s, p_two = stats.spearmanr(years, n_firms) p_val = p_two / 2 if rho_s > 0 else 1 - p_two / 2 sign_correct = rho_s > 0 # more firms entering over time verd = verdict(p_val, sign_correct) # Also compute the HHI trend for reporting (should be negative = declining) slope_hhi, _, _, p_hhi, _ = stats.linregress(years, train_df['training_hhi'].values) return log_result( 'training_inference_infer_decline', float(rho_s), float(p_val), verd, details={'n_years': len(train_df), 'n_firms_start': int(n_firms[0]), 'n_firms_end': int(n_firms[-1]), 'hhi_slope': float(slope_hhi), 'hhi_slope_p': float(p_hhi)}, description=desc) def test_model_count_growth(inf_df): """Test that model count grows super-linearly (inference democratization).""" desc = ("Tests inference democratization: the model count on HuggingFace " "should grow super-linearly (exponentially). OLS regression of " "log(model_count) on time index from quarterly HuggingFace data; " "positive slope indicates exponential growth. Data: compiled " "milestones from HF blog posts and press (2020-2025).") if inf_df is None: return log_result('training_inference_model_growth', np.nan, np.nan, 'SKIPPED', details={'reason': 'No inference data available'}, description=desc) date_col = None for col in ['date', 'month', 'year', 'timestamp']: if col in inf_df.columns: date_col = col break count_col = None for col in ['model_count', 'total_models', 'count', 'downloads']: if col in inf_df.columns: count_col = col break if date_col is None or count_col is None: return log_result('training_inference_model_growth', np.nan, np.nan, 'SKIPPED', details={'reason': 'Cannot identify date/count columns'}, description=desc) df = inf_df[[date_col, count_col]].dropna() if len(df) < 5: return log_result('training_inference_model_growth', np.nan, np.nan, 'SKIPPED', description=desc) x = np.arange(len(df)) y = df[count_col].values.astype(float) # Test super-linear: log(count) vs time should have slope > 0 (exponential) log_y = np.log(y[y > 0]) x_valid = x[:len(log_y)] if len(x_valid) < 5: return log_result('training_inference_model_growth', np.nan, np.nan, 'SKIPPED', description=desc) slope, intercept, r, p_val, _ = stats.linregress(x_valid, log_y) sign_correct = slope > 0 verd = verdict(p_val, sign_correct) _cache['model_growth'] = { 'x': x_valid, 'log_y': log_y, 'slope': slope, 'intercept': intercept, 'r_sq': r**2, 'dates': df[date_col].values, 'counts': y, } return log_result( 'training_inference_model_growth', float(slope), float(p_val), verd, details={'r_squared': float(r**2), 'n': len(x_valid), 'growth_type': 'exponential' if slope > 0 else 'sub-linear', 'count_start': int(y[0]), 'count_end': int(y[-1])}, description=desc) def test_hhi_gap_positive(train_df): """Test that training HHI exceeds equal-share benchmark. Uses the actual number of firms per year (not hardcoded) to compute the equal-share benchmark 1/N. """ desc = ("Tests that the training-inference HHI gap is positive: training " "HHI exceeds the equal-share benchmark (1/N for N firms in each " "year). One-sample t-test on HHI minus 1/N; positive gap means " "training is more concentrated than equal sharing, consistent " "with complementary resources (low rho_train).") recent = train_df[train_df['year'] >= 2020] if len(recent) < 2: return log_result('training_inference_hhi_gap', np.nan, np.nan, 'SKIPPED', description=desc) # Compute gap using actual N_firms per year, not hardcoded gaps = recent['training_hhi'].values - (1.0 / recent['n_firms'].values) t_stat, p_val = stats.ttest_1samp(gaps, 0, alternative='greater') sign_correct = np.mean(gaps) > 0 verd = verdict(p_val, sign_correct) return log_result( 'training_inference_hhi_gap', float(np.mean(gaps)), float(p_val), verd, details={'mean_gap': float(np.mean(gaps)), 'training_hhi': float(recent['training_hhi'].mean()), 'mean_n_firms': float(recent['n_firms'].mean()), 'mean_equal_share_hhi': float((1.0 / recent['n_firms']).mean())}, description=desc) def test_inference_provider_growth(prov_df): """Test that the number of inference API providers grows over time. This is a DIRECT measure of inference democratization: more providers offering LLM inference APIs means the inference market is decentralizing, consistent with rho_inference > rho_training. """ desc = ("Tests inference democratization directly: the number of major " "LLM inference API providers should grow over time. One-tailed " "Spearman correlation of n_providers vs year. Data: compiled " "list of 22 providers with public API launch dates (2020-2025). " "Theory predicts inference market becomes competitive (N >> 5) " "because queries are independent (rho_inf approx 1, K approx 0).") if prov_df is None or len(prov_df) < 4: return log_result('training_inference_provider_growth', np.nan, np.nan, 'SKIPPED', description=desc) years = prov_df['year'].values n_prov = prov_df['n_providers'].values # Spearman: provider count should increase (one-tailed) rho_s, p_two = stats.spearmanr(years, n_prov) p_val = p_two / 2 if rho_s > 0 else 1 - p_two / 2 sign_correct = rho_s > 0 verd = verdict(p_val, sign_correct) _cache['providers'] = { 'years': years, 'n_providers': n_prov, 'r': rho_s, 'p': p_val, } return log_result( 'training_inference_provider_growth', float(rho_s), float(p_val), verd, details={'n_years': len(prov_df), 'n_providers_start': int(n_prov[0]), 'n_providers_end': int(n_prov[-1]), 'provider_list_latest': prov_df.iloc[-1].get('providers', '')}, description=desc) _cache = {} def main(): """Run all training-inference bifurcation tests. Returns list of result dicts.""" try: train_df = load_training_concentration() except FileNotFoundError as e: return [log_result('training_inference_train_hhi', np.nan, np.nan, 'SKIPPED', details={'reason': str(e)}, description="Tests that training remains concentrated: HHI computed from hyperscaler capex shares should stay above 0.20.")] inf_df = load_inference_data() prov_df = load_inference_providers() results = [] results.append(test_training_hhi_high(train_df)) results.append(test_inference_entry_growth(train_df)) results.append(test_model_count_growth(inf_df)) results.append(test_hhi_gap_positive(train_df)) results.append(test_inference_provider_growth(prov_df)) # Cache data for figures _cache['train_df'] = train_df _cache['inf_df'] = inf_df _cache['prov_df'] = prov_df _cache['results'] = results return results def figures(): """Generate figures for training-inference bifurcation tests.""" if not _cache: main() paths = [] import matplotlib.pyplot as plt train_df = _cache.get('train_df') if train_df is None or len(train_df) == 0: return paths # --- Figure 1: HHI over time + N_firms (covers train_hhi + infer_decline + hhi_gap) --- fig, axes = setup_subplots(1, 2, width=12, height=5) # Panel A: HHI over time with threshold ax = axes[0] years = train_df['year'].values hhis = train_df['training_hhi'].values ax.plot(years, hhis, 'o-', color='#4e79a7', lw=2, markersize=8, label='Training HHI') ax.axhline(0.20, color='red', ls='--', lw=1.5, label='Theory threshold (1/5 = 0.20)') equal_share = 1.0 / train_df['n_firms'].values ax.plot(years, equal_share, 's--', color='#59a14f', lw=1.5, markersize=6, label='Equal-share benchmark (1/N)') ax.fill_between(years, equal_share, hhis, alpha=0.15, color='#f28e2b', label='Concentration gap') ax.set_xlabel('Year') ax.set_ylabel('HHI') ax.set_title('Training Concentration (Hyperscaler Capex)') ax.legend(fontsize=8, loc='upper right') ax.grid(True, alpha=0.3) ax.set_ylim(0, max(hhis) * 1.3) # Panel B: N_firms over time ax2 = axes[1] n_firms = train_df['n_firms'].values ax2.bar(years, n_firms, color='#76b7b2', edgecolor='black', lw=0.5, width=0.7) for y, n in zip(years, n_firms): ax2.text(y, n + 0.1, str(int(n)), ha='center', va='bottom', fontsize=10) slope_res = stats.linregress(years, n_firms) x_fit = np.array([years.min(), years.max()]) ax2.plot(x_fit, slope_res.intercept + slope_res.slope * x_fit, 'r--', lw=1.5, label=f'Trend (slope={slope_res.slope:.2f}/yr)') ax2.set_xlabel('Year') ax2.set_ylabel('Number of Firms') ax2.set_title('Firm Entry into AI Capex Market') ax2.legend(fontsize=9) ax2.grid(True, alpha=0.3, axis='y') ax2.set_ylim(0, max(n_firms) + 2) fig.tight_layout() paths.extend(save_figure(fig, 'training_inference_hhi_entry', subfolder='applications')) # --- Figure 2: Model count growth + Provider count (covers model_growth + provider_growth) --- mg = _cache.get('model_growth') prov = _cache.get('providers') n_panels = (1 if mg else 0) + (1 if prov else 0) if n_panels > 0: fig2, axes2 = setup_subplots(1, max(n_panels, 1), width=6 * max(n_panels, 1), height=5) if n_panels == 1: axes2 = [axes2] panel_idx = 0 if mg: ax3 = axes2[panel_idx] panel_idx += 1 dates = mg['dates'] counts = mg['counts'] ax3.semilogy(range(len(counts)), counts, 'o-', color='#4e79a7', lw=2, markersize=6) # Fit line on log scale x_fit = np.arange(len(counts)) y_fit = np.exp(mg['intercept'] + mg['slope'] * x_fit) ax3.semilogy(x_fit, y_fit, 'r--', lw=1.5, label=f'Exp fit (R²={mg["r_sq"]:.3f})') # X-axis labels: use dates tick_idx = list(range(0, len(dates), max(1, len(dates) // 6))) ax3.set_xticks(tick_idx) ax3.set_xticklabels([str(dates[i])[:7] for i in tick_idx], rotation=45, ha='right', fontsize=8) ax3.set_ylabel('Model Count (log scale)') ax3.set_title(f'HuggingFace Model Growth\n' f'{int(counts[0]):,} → {int(counts[-1]):,}') ax3.legend(fontsize=9) ax3.grid(True, alpha=0.3) if prov: ax4 = axes2[panel_idx] yrs = prov['years'] n_p = prov['n_providers'] ax4.bar(yrs, n_p, color='#e15759', edgecolor='black', lw=0.5, width=0.7) for y, n in zip(yrs, n_p): ax4.text(y, n + 0.3, str(int(n)), ha='center', va='bottom', fontsize=10) ax4.set_xlabel('Year') ax4.set_ylabel('Cumulative API Providers') ax4.set_title(f'Inference API Providers\n' f'(r={prov["r"]:.3f}, p={prov["p"]:.4f})') ax4.grid(True, alpha=0.3, axis='y') ax4.set_ylim(0, max(n_p) + 4) fig2.tight_layout() paths.extend(save_figure(fig2, 'training_inference_model_providers', subfolder='applications')) return paths if __name__ == '__main__': results = main() for r in results: print(f" {r['name']}: {r['verdict']} (stat={r['statistic']:.4f}, p={r['p_value']:.4f})")

n years	8
p value	9.645e-4
hhi slope	-0.0104
statistic	0.9258
hhi slope p	0.0507
n firms end	6
n firms start	4

Training Inference Infer Decline

Description

Figures

Statistics

Script

▸ Source Code

Reproduction Steps