#!/usr/bin/env python3 """ Perez Turning Point = rho Minimum -- test_audit U4 Theory (Paper 3, Remark 16.2; Paper 5, Section U4): The aggregate complementarity parameter rho_bar follows a V-shape over Kondratiev-wave timescales: declining during the installation/frenzy phase as firms adopt tightly-coupled complementary architectures, reaching a minimum near the Perez turning point (financial crisis), then rising during deployment as standardization and modularization reverse the architectural lock-in. Since rho_bar(t) is not directly observable at technology-wave frequency (see Remark U4_identification in Paper 5), we use the capital intensity proxy from Series C: Δg_INFO(t) = g_INFO(t) - g_Total(t), the ICT sector's excess capital growth rate. High Δg_INFO signals massive parallel complementary investments (installation phase, low rho_bar); low Δg_INFO signals capital-efficiency normalisation (deployment phase, higher rho_bar). Lean theorem: perez_phases in CESProofs/Dynamics/EndogenousRho.lean Subtests: 1. Direction: install_mean(Δg_INFO) > deploy_mean(Δg_INFO) for the canonical ICT/Internet 2001 wave. Permutation test against random turning points. 2. Recovery: Δg_INFO recovers to >0 within 7 years of trough. 3. Peak-trough magnitude: The peak-to-trough swing in Δg_INFO should be economically significant (> 3 pp). Data: BEA Fixed Assets by industry (current-cost net stock, 1980-2023). Review Notes (2026-03-11): Complete rewrite. Previous test used NBER-CES labor share via sigma = 1/(1-s_L), which trivially reduces to rho = s_L (just the labor share). Manufacturing labor share declines monotonically 1958-2018, producing zero alignment with Perez dates. Fixed to use BEA capital intensity proxy (Series C from Paper 5), which correctly captures the supply-side investment cycle. Known limitations: - Only N=1 canonical Perez wave is testable (ICT/Internet 2001) - Capital data is nominal (understates real ICT growth due to Moore's Law deflation, which is conservative for direction test) - Finance/Digital 2007 is non-canonical and confounded by housing boom - The conclusive test requires BEA Benchmark I-O Use Tables or plant-level production function estimates Review Notes (2026-03-12): Theory match: PARTIAL — tests capital intensity V-shape (supply-side investment cycle), not rho_bar directly. Paper 3 Remark rem:credit-driven-rho revises the theory: rho_bar cycles are credit-driven (NFCI peaks lead rho_bar by 4yr), not Perez-driven. This test measures the dot-com boom/bust in ICT investment, which is real but the link to rho_bar is indirect. I/O audit: CORRECT — Δg_INFO is a supply-side quantity (input investment), not an output variable. Data-signal alignment: ADEQUATE — BEA capital stock is direct; proxy quality REASONABLE (capital intensity → complementarity is indirect). Code quality: CLEAN — no bugs found. Lean: perez_phases is schematic (True := trivial). Confidence: LOW — N=1 canonical wave; direction test p=0.0696 borderline; recovery and magnitude tests have no p-values. Related tests: test_rho_credit_vs_perez (NFCI leads rho_bar by 4yr, V-shapes align with NFCI peaks not Perez dates). """ 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.stats import log_result from common.plotting import setup_figure, setup_subplots, save_figure _cache = {} _ALL_SUBTESTS = [ 'perez_turning_point_direction', 'perez_turning_point_recovery', 'perez_turning_point_magnitude', ] def load_capital_stock(): """Load BEA capital stock by industry.""" data_dir = os.path.join(os.path.dirname(__file__), '..', '..', 'data') path = os.path.join(data_dir, 'compiled', 'bea_capital_stock_by_industry.csv') if not os.path.exists(path): raise FileNotFoundError("BEA capital stock by industry not found.") return pd.read_csv(path) # Canonical Perez turning points with testability PEREZ_WAVES = { 'ICT_Internet_2001': { 'turning_point': 2001, 'install_range': (1997, 2001), # 5-year install phase 'deploy_range': (2002, 2006), # 5-year deploy phase 'sector': 'Information', 'canonical': True, }, 'Finance_Digital_2007': { 'turning_point': 2007, 'install_range': (2003, 2007), 'deploy_range': (2008, 2012), 'sector': 'Information', 'canonical': False, # non-canonical; confounded by housing boom }, } def main(): results = [] try: cap = load_capital_stock() except FileNotFoundError: return [log_result( 'perez_turning_point', stat=np.nan, p=np.nan, verd='SKIPPED', description='BEA capital stock data not found.', )] if 'Information' not in cap.columns or 'Total_Private' not in cap.columns: return [log_result( 'perez_turning_point', stat=np.nan, p=np.nan, verd='SKIPPED', description='BEA capital stock missing Information or Total_Private columns.', )] # Compute excess capital growth rate cap = cap.sort_values('year').copy() cap['g_info'] = cap['Information'].pct_change() * 100 cap['g_total'] = cap['Total_Private'].pct_change() * 100 cap['delta_g_info'] = cap['g_info'] - cap['g_total'] cap = cap.dropna(subset=['delta_g_info']) if len(cap) < 15: return [log_result( 'perez_turning_point', stat=np.nan, p=np.nan, verd='SKIPPED', description=f'Insufficient capital stock data: {len(cap)} years.', )] _cache['cap'] = cap # ========================================================= # Subtest 1: Direction test for canonical ICT/Internet 2001 # ========================================================= wave = PEREZ_WAVES['ICT_Internet_2001'] tp = wave['turning_point'] i_lo, i_hi = wave['install_range'] d_lo, d_hi = wave['deploy_range'] install_data = cap[(cap['year'] >= i_lo) & (cap['year'] <= i_hi)] deploy_data = cap[(cap['year'] >= d_lo) & (cap['year'] <= d_hi)] if len(install_data) < 3 or len(deploy_data) < 3: results.append(log_result( 'perez_turning_point_direction', stat=np.nan, p=np.nan, verd='SKIPPED', description='Insufficient data in install/deploy windows.', )) else: install_mean = float(install_data['delta_g_info'].mean()) deploy_mean = float(deploy_data['delta_g_info'].mean()) obs_diff = install_mean - deploy_mean # Permutation test: random turning points rng = np.random.default_rng(42) n_perms = 5000 window_half = (i_hi - i_lo + d_hi - d_lo) // 2 + 1 valid_years = cap[(cap['year'] >= cap['year'].min() + window_half + 1) & (cap['year'] <= cap['year'].max() - window_half)]['year'].values perm_diffs = [] for _ in range(n_perms): rand_tp = rng.choice(valid_years) inst_len = i_hi - i_lo dep_len = d_hi - d_lo r_install = cap[(cap['year'] >= rand_tp - inst_len) & (cap['year'] <= rand_tp)] r_deploy = cap[(cap['year'] >= rand_tp + 1) & (cap['year'] <= rand_tp + 1 + dep_len)] if len(r_install) >= 3 and len(r_deploy) >= 3: perm_diffs.append( r_install['delta_g_info'].mean() - r_deploy['delta_g_info'].mean() ) perm_diffs = np.array(perm_diffs) # One-sided p: fraction of random turning points with diff >= observed p_dir = float(np.mean(perm_diffs >= obs_diff)) if len(perm_diffs) > 0 else np.nan direction_ok = obs_diff > 0 if direction_ok and p_dir < 0.10: v1 = 'CONSISTENT' elif direction_ok: v1 = 'AMBIGUOUS' else: v1 = 'INCONSISTENT' results.append(log_result( 'perez_turning_point_direction', stat=float(obs_diff), p=float(p_dir), verd=v1, details={ 'turning_point': tp, 'install_mean': round(install_mean, 2), 'deploy_mean': round(deploy_mean, 2), 'diff_pp': round(obs_diff, 2), 'install_range': list(wave['install_range']), 'deploy_range': list(wave['deploy_range']), 'perm_mean_diff': round(float(np.mean(perm_diffs)), 2) if len(perm_diffs) > 0 else None, 'perm_95th': round(float(np.percentile(perm_diffs, 95)), 2) if len(perm_diffs) > 0 else None, 'n_perms_valid': len(perm_diffs), }, description=( f'Tests Perez installation/deployment asymmetry via ICT capital ' f'intensity proxy. Permutation test (N=5000): observed diff in ' f'mean Δg_INFO (install minus deploy) = {obs_diff:+.2f} pp vs ' f'random turning points. Data: BEA Fixed Assets, Information ' f'sector excess growth rate g_INFO - g_Total, 1980-2023. ' f'Install {i_lo}-{i_hi}: {install_mean:+.2f} pp; deploy ' f'{d_lo}-{d_hi}: {deploy_mean:+.2f} pp. p = {p_dir:.4f}. ' f'CONSISTENT if diff > 0 and p < 0.10. N=1 canonical wave ' f'(ICT/Internet 2001); inherently underpowered.' ), )) _cache['direction'] = { 'obs_diff': obs_diff, 'perm_diffs': perm_diffs, 'install_mean': install_mean, 'deploy_mean': deploy_mean, 'p': p_dir, } # ========================================================= # Subtest 2: Recovery after trough # ========================================================= # Find trough in Δg_INFO within [tp, tp+7] post_tp = cap[(cap['year'] >= tp) & (cap['year'] <= tp + 7)] if len(post_tp) < 3: results.append(log_result( 'perez_turning_point_recovery', stat=np.nan, p=np.nan, verd='SKIPPED', description='Insufficient post-turning-point data.', )) else: trough_idx = post_tp['delta_g_info'].idxmin() trough_year = int(cap.loc[trough_idx, 'year']) trough_val = float(cap.loc[trough_idx, 'delta_g_info']) # Check recovery: does Δg_INFO return to > 0 within 7 years? recovery_window = cap[(cap['year'] > trough_year) & (cap['year'] <= trough_year + 7)] recovered = bool(any(recovery_window['delta_g_info'] > 0)) if recovered: first_recovery = recovery_window[recovery_window['delta_g_info'] > 0].iloc[0] recovery_year = int(first_recovery['year']) recovery_val = float(first_recovery['delta_g_info']) else: recovery_year = None recovery_val = None v2 = 'CONSISTENT' if recovered else 'INCONSISTENT' results.append(log_result( 'perez_turning_point_recovery', stat=1.0 if recovered else 0.0, p=np.nan, verd=v2, details={ 'trough_year': trough_year, 'trough_delta_g_info': round(trough_val, 2), 'recovered': recovered, 'recovery_year': recovery_year, 'recovery_delta_g_info': round(recovery_val, 2) if recovery_val else None, 'years_to_recovery': recovery_year - trough_year if recovery_year else None, }, description=( f'Tests deployment-phase recovery in ICT capital intensity. ' f'Binary check: does Δg_INFO (BEA Information sector excess ' f'capital growth) return to positive within 7 years of post-' f'turning-point trough? Trough at {trough_year} ' f'({trough_val:+.2f} pp). ' + (f'Recovered to {recovery_val:+.2f} pp in {recovery_year} ' f'({recovery_year - trough_year} yr). ' if recovered else 'No recovery within 7 years. ') + 'Theory: deployment phase restores modular architecture ' '(rising rho_bar), normalizing ICT capital growth.' ), )) _cache['recovery'] = { 'trough_year': trough_year, 'trough_val': trough_val, 'recovered': recovered, 'recovery_year': recovery_year, 'recovery_val': recovery_val, } # ========================================================= # Subtest 3: Peak-trough magnitude # ========================================================= # The peak-to-trough swing should be economically significant pre_crash = cap[(cap['year'] >= tp - 3) & (cap['year'] <= tp)] post_crash = cap[(cap['year'] >= tp + 1) & (cap['year'] <= tp + 4)] if len(pre_crash) > 0 and len(post_crash) > 0: peak_val = float(pre_crash['delta_g_info'].max()) trough_val_3 = float(post_crash['delta_g_info'].min()) peak_year = int(cap.loc[pre_crash['delta_g_info'].idxmax(), 'year']) trough_year_3 = int(cap.loc[post_crash['delta_g_info'].idxmin(), 'year']) swing = peak_val - trough_val_3 v3 = 'CONSISTENT' if swing > 3.0 else ( 'AMBIGUOUS' if swing > 1.5 else 'INCONSISTENT') results.append(log_result( 'perez_turning_point_magnitude', stat=float(swing), p=np.nan, verd=v3, details={ 'peak_year': peak_year, 'peak_delta_g_info': round(peak_val, 2), 'trough_year': trough_year_3, 'trough_delta_g_info': round(trough_val_3, 2), 'swing_pp': round(swing, 2), }, description=( f'Tests economic significance of ICT investment reversal. ' f'Peak Δg_INFO ({peak_year}) to trough ({trough_year_3}) swing ' f'= {swing:.2f} pp. Data: BEA Fixed Assets, Information sector ' f'excess capital growth. CONSISTENT if swing > 3 pp (threshold ' f'for economically significant architectural shift). ' f'AMBIGUOUS if 1.5-3 pp. N=1 canonical wave (ICT/Internet 2001).' ), )) _cache['magnitude'] = { 'peak_year': peak_year, 'peak_val': peak_val, 'trough_year': trough_year_3, 'trough_val': trough_val_3, 'swing': swing, } else: results.append(log_result( 'perez_turning_point_magnitude', stat=np.nan, p=np.nan, verd='SKIPPED', description='Insufficient data for peak-trough magnitude.', )) return results def figures(): """Generate figures for each subtest.""" if not _cache: main() paths = [] cap = _cache.get('cap') if cap is None: return paths tp = 2001 # --- Figure 1: Direction (permutation histogram) --- d = _cache.get('direction') if d is not None: fig, ax = setup_figure() ax.hist(d['perm_diffs'], bins=50, alpha=0.7, color='steelblue', edgecolor='white', label='Random turning points') ax.axvline(d['obs_diff'], color='red', linewidth=2, linestyle='--', label=f'ICT 2001: {d["obs_diff"]:+.2f} pp') ax.set_xlabel('Install mean − Deploy mean (Δg_INFO, pp)') ax.set_ylabel('Count') ax.set_title('Perez Direction Test: ICT/Internet 2001 vs Random TPs') ax.legend() ax.text(0.98, 0.95, f'p = {d["p"]:.4f}', transform=ax.transAxes, ha='right', va='top', fontsize=11) paths.extend(save_figure(fig, 'perez_turning_point_direction', subfolder='dynamics')) # --- Figure 2: Time series of Δg_INFO with phases --- fig, ax = setup_figure() ax.plot(cap['year'], cap['delta_g_info'], 'o-', color='steelblue', markersize=4, linewidth=1.5, label='Δg_INFO') ax.axhline(0, color='gray', linewidth=0.8, linestyle='-') ax.axvline(tp, color='red', linewidth=1.5, linestyle='--', label=f'Turning point ({tp})') ax.axvspan(1997, 2001, alpha=0.15, color='orange', label='Install phase') ax.axvspan(2002, 2006, alpha=0.15, color='green', label='Deploy phase') ax.set_xlabel('Year') ax.set_ylabel('Δg_INFO (pp)') ax.set_title('ICT Capital Excess Growth Rate: Installation vs Deployment') ax.legend(loc='lower left', fontsize=8) # Annotate peak and trough if available m = _cache.get('magnitude') if m: ax.annotate(f'Peak: {m["peak_val"]:+.1f}pp', xy=(m['peak_year'], m['peak_val']), xytext=(m['peak_year'] - 5, m['peak_val'] + 1), arrowprops=dict(arrowstyle='->', color='black'), fontsize=9) ax.annotate(f'Trough: {m["trough_val"]:+.1f}pp', xy=(m['trough_year'], m['trough_val']), xytext=(m['trough_year'] + 3, m['trough_val'] - 1), arrowprops=dict(arrowstyle='->', color='black'), fontsize=9) paths.extend(save_figure(fig, 'perez_turning_point_timeseries', subfolder='dynamics')) return paths if __name__ == '__main__': results = main() for r in results: print(f" {r['name']}: {r['verdict']} " f"(stat={r['statistic']}, p={r.get('p_value', 'n/a')})") if r.get('details'): for k, v in r['details'].items(): print(f" {k}: {v}")

swing pp	8.5000
peak year	2000
statistic	8.5002
trough year	2003
peak delta g info	3.8200
trough delta g info	-4.6800

Perez Turning Point Magnitude

Description

Figures

Statistics

Script

▸ Source Code

Reproduction Steps