"""Tables and LaTeX number macros from solved scenarios. Outputs: results/tables/tab02_calibration.tex, tab03_baseline.tex, tab04_scenarios.tex, tab05_decomposition.tex, tab06_policy.tex, tab08_diagnostics.tex, numbers.tex ; results/data/scenario_summary.csv""" import glob import json import os import re import sys import numpy as np import pandas as pd sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) from agi_ge.analysis import campbell_shiller_decomposition, summary_by_year, valuation_ratios, welfare_cev # noqa: E402 from agi_ge.calibration import load_targets # noqa: E402 from agi_ge.solve import load_paths # noqa: E402 ROOT = os.path.join(os.path.dirname(__file__), "..") SC = os.path.join(ROOT, "results", "scenarios") TAB = os.path.join(ROOT, "results", "tables") os.makedirs(TAB, exist_ok=True) LAB = {"baseline": "Baseline", "no_ai": "AI frozen (everything at 2026)", "no_capability": "Capability frozen, hardware cheapens", "alpha_inf_x5": "Terminal compute requirements $\\times5$", "rapid_agi": "Rapid AGI (all tasks in 2032)", "fast_moore_cheap_energy": "Fast Moore ($g_{pQ}=0.34$), cheap energy", "fast_moore_energy_constrained": "Fast Moore, no efficiency gains", "moore_ends_2040": "Moore's law ends 2040 (floor 0.12)", "moore_floor_1e3": "Compute price floor $10^{-3}$", "moore_never_ends": "Moore's law never ends", "slow_alpha": "Slow capability ($g_\\alpha=0.05$)", "fast_alpha": "Fast capability ($g_\\alpha=0.20$)", "complements_leontief_like": "Strong complements ($\\sigma_N=\\sigma_T=0.4$)", "complements_cobb_douglas": "Cobb--Douglas ($\\sigma=1$)", "substitutes": "Substitutes ($\\sigma=1.5$)", "ai_monopoly": "AI monopoly (markup 1.5)", "citizen_fund_20": "Citizen fund (20\\% of equity)", "land_tax_50": "Land tax 50\\%", "compute_tax_30": "Compute tax 30\\%", "capital_tax_20": "Capital-income tax 20\\%", "foreign_ai_50": "50\\% foreign-owned compute", "low_risk_aversion": "$\\gamma=1$", "high_risk_aversion": "$\\gamma=4$", "scarce_land": "Scarce land (share 8\\%, $\\sigma_T=0.4$)", "severe_chip_depreciation": "Severe chip depreciation ($\\delta_Q=0.45$)", "reinstatement": "Expert task never automated"} _DIGITS = {"0": "Zero", "1": "One", "2": "Two", "3": "Three", "4": "Four", "5": "Five", "6": "Six", "7": "Seven", "8": "Eight", "9": "Nine"} def macro_name(s): """LaTeX-safe macro fragment: digits spelled out, non-letters removed.""" s = "".join(_DIGITS.get(ch, ch) for ch in str(s)) return re.sub(r"[^A-Za-z]", "", s) def fmt(x, d=2): if x is None or (isinstance(x, float) and not np.isfinite(x)): return "--" if abs(x) >= 1000: return f"{x:,.0f}".replace(",", "{,}") return f"{x:.{d}f}" def write_numbers(nums): with open(os.path.join(TAB, "numbers.tex"), "w") as f: f.write("% generated by scripts/50_tables.py — one macro per number used in the text\n") for k, v in nums.items(): f.write(f"\\newcommand{{\\{k}}}{{{v}}}\n") def table_calibration(params, info): tg = load_targets() rows = [] src = params.get("source", {}) def add(name, sym, val, status, source): rows.append((name, sym, val, status, source)) v = {k: t.value for k, t in tg.items()} st = {k: t.status for k, t in tg.items()} so = {k: t.source for k, t in tg.items()} add("Labor share of GDP, 2023", "", fmt(v["labor_share"], 3), st["labor_share"], "PWT via FRED") add("Nonresidential fixed assets / GDP (comparison)", "", fmt(v["KN_over_GDP"]), st["KN_over_GDP"], "BEA via FRED") add("Structures / GDP implied by income shares", "$K^N_0$", fmt(params["KN0"]), "calibrated", "residual of the income identity") add("Residential structures / GDP", "$K^H_0$", fmt(params["KH0"]), st["KH_over_GDP"], "BEA via FRED (derived)") add("Energy capital / GDP", "$K^E_0$", fmt(params["KE0"]), st["KE_over_GDP"], "assumption") add("Value of compute capital / GDP", "$p^Q_0K^Q_0$", fmt(v["VQ_over_GDP"], 3), st["VQ_over_GDP"], "IPE investment capitalized") add("Depreciation: structures, housing, compute, energy", "$\\delta_k$", f"{params['delta']['N']:.3f}, {params['delta']['H']:.3f}, {params['delta']['Q']:.2f}, {params['delta']['E']:.2f}", "data / assumed", "BEA via FRED; accounting lives") add("Adjustment costs", "$\\phi_k$", f"{params['phi']['N']:.0f}, {params['phi']['H']:.0f}, {params['phi']['Q']:.0f}, {params['phi']['E']:.0f}", "literature", "Cooper--Haltiwanger (2006) order") add("Real return in the frozen BGP", "$r^*$", fmt(v["r_star"], 3), st["r_star"], "between TIPS (1.96\\%) and returns on wealth") add("Trend growth per efficiency unit", "$g_Z$", fmt(v["g_Z"], 3), st["g_Z"], "real GDP growth 1990--2025 minus population") add("Curvature of utility", "$\\gamma$", fmt(v["gamma_u"], 1), "literature", "") add("Discount factor (implied)", "$\\beta$", fmt(params["beta"], 4), "calibrated", "$\\beta=(1+g_Z)^\\gamma/(1+r^*)$") add("Housing expenditure share (implied)", "$\\theta_H$", fmt(params["theta_H"], 3), "calibrated", "housing value added 12\\% of GDP") add("Land share of housing (implied)", "$\\nu_H$", fmt(params["nu_H"], 3), "calibrated", "housing Euler") add("Resource share in energy (implied)", "$\\eta$", fmt(params["eta"], 3), "calibrated", "energy Euler") add("Energy sector output / GDP", "", fmt(v["energy_share"], 3), st["energy_share"], "assumption (electricity 1.7\\% + other)") add("Data-center share of electricity, 2023", "", fmt(v["datacenter_energy_share"], 3), st["datacenter_energy_share"], "Shehabi et al. (2024)") add("Production land rent / GDP", "", fmt(v["land_rent_share"], 3), st["land_rent_share"], "assumption") add("Elasticities: tasks; tasks--structures; top nest", "$\\sigma_B,\\sigma_N,\\sigma_T$", f"{v['sigma_B']}, {v['sigma_N']}, {v['sigma_T']}", "literature", "Oberfield--Raval; Chirinko") add("Workers of skill $L$; skill premium", "$H_L$; $w_H/w_L$", f"{v['share_L_workers']}; {v['skill_premium']}", "assumed", "") add("CES task weights (implied)", "$\\nu_i$", ", ".join(f"{x:.3f}" for x in params["tasks"]["nu"]), "calibrated", "task wage shares") add("Compute per worker, task 1, 2026 (implied)", "$\\alpha_{1,0}$", fmt(info["alpha10"], 1), "calibrated", "compute Euler and task-1 tie") add("Implied compute share of task-1 input, 2026", "", src.get("compute_in_task1_implied", ""), "implied", "") add("Terminal compute per worker by task", "$\\alpha_i^\\infty$", ", ".join(f"{x:g}" for x in params["tasks"]["alpha_inf"]), "scenario", "normalization $\\alpha_1^\\infty=1$") add("2026 compute per worker by task (multiples of terminal)", "$\\alpha_{i,0}/\\alpha_i^\\infty$", ", ".join(f"{x:.0f}" for x in params["alpha0_mult"]), "calibrated / scenario", "") add("Capability progress; compute price decline; efficiency gains", "$g_\\alpha,g_{pQ},g_\\varepsilon$", f"{params['g_alpha']}, {params['g_pQ']}, {params['g_eps']}", "scenario", "Epoch: 0.34 for GPU FLOP/\\$") add("Floors (fraction of 2026): compute price; energy per unit", "$\\underline p/p_0,\\underline\\varepsilon/\\varepsilon_0$", f"{params['pQ_floor_ratio']:g}, {params['eps_floor_ratio']:g}", "scenario", "") with open(os.path.join(TAB, "tab02_calibration.tex"), "w") as f: f.write("\\begin{tabular}{@{}p{6.3cm}lp{3.2cm}lp{3.8cm}@{}}\n\\toprule\nObject & Symbol & Value & Status & Source\\\\\n\\midrule\n") for r in rows: f.write(" & ".join(str(x) for x in r) + "\\\\\n") f.write("\\bottomrule\n\\end{tabular}\n") def table_baseline(P, years=(2026, 2030, 2040, 2050, 2075, 2100)): S = summary_by_year(P, years) rows = [("GDP (2026 = 1)", "GDP", 2), ("GDP growth (\\%)", "GDP_growth", 1), ("Labor share", "labor_share", 3), ("Compute income share", "compute_income_share", 3), ("Structures share", "structures_share", 3), ("Land share (both)", "land_share", 3), ("Energy share", "energy_share", 3), ("Wage $L$ per worker (2026 = 1)", "wL", 3), ("Wage $H$ per worker (2026 = 1)", "wH", 3), ("Real return $r$ (\\%)", "r", 1), ("Compute capacity (2026 = 1)", "KQ", 1), ("Value of compute / GDP", "VQ_over_GDP", 3), ("Wealth / GDP", "W_over_GDP", 2), ("Land price--rent ratio", "land_PR", 1), ("House price--rent ratio", "house_PR", 1), ("P/E compute", "PE_Q", 1), ("P/E structures", "PE_N", 1), ("Tobin's $q$ compute", "q_Q", 2), ("Share of energy used by compute", "energy_to_compute", 3), ("Investment / GDP", "inv_share", 3), ("Workers' consumption / GDP", "CW_over_GDP", 3), ("Tasks using compute (of 4)", "tasks_automated", 0)] with open(os.path.join(TAB, "tab03_baseline.tex"), "w") as f: f.write("\\begin{tabular}{@{}l" + "r" * len(years) + "@{}}\n\\toprule\n & " + " & ".join(str(y) for y in years) + "\\\\\n\\midrule\n") for lab, k, d in rows: vals = [] for y in years: x = S[y][k] if k in ("GDP_growth", "r"): x = 100 * x vals.append(fmt(x, d)) f.write(lab + " & " + " & ".join(vals) + "\\\\\n") f.write("\\bottomrule\n\\end{tabular}\n") return S def table_scenarios(allP, names, fname, years=(2050, 2100)): cols = [("labor_share", "Labor share", 3), ("wL", "$w_L$ (2026=1)", 3), ("GDP", "GDP (2026=1)", 1), ("r", "$r$ (\\%)", 1), ("W_over_GDP", "$W$/GDP", 1), ("land_PR", "Land P/R", 1), ("VQ_over_GDP", "$V^Q$/GDP", 3), ("CW_over_GDP", "$C_W$/GDP", 3)] recs = [] with open(os.path.join(TAB, fname), "w") as f: f.write("\\begin{tabular}{@{}l" + "r" * (len(cols) * len(years)) + "@{}}\n\\toprule\n") f.write(" & " + " & ".join(f"\\multicolumn{{{len(cols)}}}{{c}}{{{y}}}" for y in years) + "\\\\\n") f.write("Scenario & " + " & ".join(c[1] for c in cols for _ in years) if False else "Scenario & " + " & ".join(c[1] for _ in years for c in cols) + "\\\\\n\\midrule\n") for n in names: if n not in allP: continue S = summary_by_year(allP[n], years) vals = [] for y in years: for k, _, d in cols: x = S[y][k] x = 100 * x if k == "r" else x vals.append(fmt(x, d)) recs.append(dict(scenario=n, year=y, stat=k, value=S[y][k])) f.write(LAB.get(n, n) + " & " + " & ".join(vals) + "\\\\\n") f.write("\\bottomrule\n\\end{tabular}\n") return recs def table_decomposition(allP): if "baseline" not in allP or "no_ai" not in allP: return {} P1, P0 = allP["baseline"], allP["no_ai"] out = {} with open(os.path.join(TAB, "tab05_decomposition.tex"), "w") as f: f.write("\\begin{tabular}{@{}lrrr@{}}\n\\toprule\nClaim & Cash-flow channel & Discount-rate channel & Total\\\\\n\\midrule\n") for c, lab in [("land", "Production land"), ("housing_land", "Residential land"), ("N", "Structures"), ("Q", "Compute capital"), ("H", "Housing structures"), ("house", "Houses (structure + land)")]: try: d = campbell_shiller_decomposition(P1, P0, claim=c) f.write(f"{lab} & {fmt(100*d['cash_flow'],1)} & {fmt(100*d['discount_rate'],1)} & {fmt(100*d['total'],1)}\\\\\n") out[c] = d except Exception as e: # noqa: BLE001 print("decomp failed", c, e) f.write("\\bottomrule\n\\end{tabular}\n") return out def table_policy(allP, params): names = ["baseline", "citizen_fund_20", "land_tax_50", "compute_tax_30", "capital_tax_20", "ai_monopoly", "foreign_ai_50"] if "baseline" not in allP: return {} P0 = allP["baseline"] beta, gamma = params["beta"], params["gamma_u"] out = {} with open(os.path.join(TAB, "tab06_policy.tex"), "w") as f: f.write("\\begin{tabular}{@{}lrrrrrrr@{}}\n\\toprule\nScenario & GDP 2050 & GDP 2100 & $w_L$ 2050 & $C_W$/GDP 2050 & $C_W$/GDP 2100 & CEV workers 2026--65 (\\%) & CEV capitalists 2026--65 (\\%)\\\\\n\\midrule\n") for n in names: if n not in allP: continue P = allP[n] S = summary_by_year(P, (2050, 2100)) cevW = welfare_cev(P["X_W"], P["P_idx"], P0["X_W"], P0["P_idx"], beta, gamma) cevK = welfare_cev(P["X_K"], P["P_idx"], P0["X_K"], P0["P_idx"], beta, gamma) out[n] = dict(cevW=cevW, cevK=cevK, S=S) f.write(f"{LAB.get(n, n)} & {fmt(S[2050]['GDP'],2)} & {fmt(S[2100]['GDP'],1)} & {fmt(S[2050]['wL'],3)} & {fmt(S[2050]['CW_over_GDP'],3)} & {fmt(S[2100]['CW_over_GDP'],4)} & {fmt(100*cevW,1)} & {fmt(100*cevK,1)}\\\\\n") f.write("\\bottomrule\n\\end{tabular}\n") return out def table_diagnostics(names): rows = [] for n in names: mf = os.path.join(SC, f"{n}_meta.json") pf = os.path.join(SC, f"{n}_paths.npz") if not (os.path.exists(mf) and os.path.exists(pf)): continue meta = json.load(open(mf)) P = load_paths(pf) res = np.abs(P["residuals"]).max() inc = np.abs(P["income_residual"] / P["GDP"]).max() hc = meta.get("horizon_check") or {} rows.append((LAB.get(n, n), meta["converged"], res, inc, hc.get("max_dev_r", np.nan), max(hc.get("max_log_dev_K", {np.nan: np.nan}).values()) if hc else np.nan, meta["seconds"])) with open(os.path.join(TAB, "tab08_diagnostics.tex"), "w") as f: f.write("\\begin{tabular}{@{}lcrrrrr@{}}\n\\toprule\nScenario & Converged & max $|$residual$|$ & max income-identity gap / GDP & horizon check: $\\Delta r$ & $\\Delta\\log K$ & seconds\\\\\n\\midrule\n") for r in rows: f.write(f"{r[0]} & {'yes' if r[1] else 'no'} & {r[2]:.1e} & {r[3]:.1e} & {r[4]:.1e} & {r[5]:.1e} & {r[6]:.0f}\\\\\n") f.write("\\bottomrule\n\\end{tabular}\n") return rows if __name__ == "__main__": allP = {} for fpath in sorted(glob.glob(os.path.join(SC, "*_paths.npz"))): allP[os.path.basename(fpath).replace("_paths.npz", "")] = load_paths(fpath) print("scenarios:", list(allP)) nums = {} if "baseline" in allP: params = json.load(open(os.path.join(SC, "baseline_params.json"))) info = json.load(open(os.path.join(SC, "baseline_meta.json")))["calibration_info"] table_calibration(params, info) S = table_baseline(allP["baseline"]) vr = valuation_ratios(allP["baseline"]) P = allP["baseline"] nums.update({ "WoverGDPbase": fmt(vr["W_over_GDP"][0], 1), "LandOverGDPbase": fmt((P["PT"][0] + P["PTH"][0]) / P["GDP"][0], 1), "PEstructuresbase": fmt(vr["PE_N"][0], 1), "PEcomputebase": fmt(vr["PE_Q"][0], 1), "rBase": fmt(100 * P["r"][1], 1), "AlphaOneBase": fmt(info["alpha10"], 1), }) for y in (2030, 2040, 2050, 2075, 2100): nums[f"LS{macro_name(str(y))}"] = fmt(S[y]["labor_share"], 3) nums[f"wL{macro_name(str(y))}"] = fmt(S[y]["wL"], 2) nums[f"wH{macro_name(str(y))}"] = fmt(S[y]["wH"], 2) nums[f"GDP{macro_name(str(y))}"] = fmt(S[y]["GDP"], 1) nums[f"WY{macro_name(str(y))}"] = fmt(S[y]["W_over_GDP"], 1) nums[f"rr{macro_name(str(y))}"] = fmt(100 * S[y]["r"], 1) nums[f"CompShare{macro_name(str(y))}"] = fmt(S[y]["compute_income_share"], 3) nums[f"LandShare{macro_name(str(y))}"] = fmt(S[y]["land_share"], 3) nums[f"LandPR{macro_name(str(y))}"] = fmt(S[y]["land_PR"], 1) nums[f"CWshare{macro_name(str(y))}"] = fmt(S[y]["CW_over_GDP"], 2) nums[f"HousePR{macro_name(str(y))}"] = fmt(S[y]["house_PR"], 1) # years of automation (first year each task uses compute for > 50% of its input) Li, Qi, al = P["Li"], P["Qi"], P["alpha"] cs = (Qi / al) / (Li + Qi / al + 1e-300) for i, nm in enumerate(["One", "Two", "Three", "Four"]): idx = np.where(cs[:, i] > 0.5)[0] nums[f"AutoYear{nm}"] = str(int(P["years"][idx[0]])) if len(idx) else "never" pk = min(range(len(P["labor_share"])), key=lambda t: P["labor_share"][t] if P["years"][t] <= 2100 else 9) gmax = int(np.argmax(np.diff(np.log(P["GDP"]))[:100])) nums["GDPgrowthPeak"] = fmt(100 * np.diff(np.log(P["GDP"]))[gmax], 1) nums["GDPgrowthPeakYear"] = str(int(P["years"][gmax])) nums["rPeak"] = fmt(100 * np.max(P["r"][1:101]), 1) nums["rPeakYear"] = str(int(P["years"][int(np.argmax(P["r"][1:101]))])) nums["WYmin"] = fmt(np.min(vr["W_over_GDP"][:100]), 1) nums["WYminYear"] = str(int(P["years"][int(np.argmin(vr["W_over_GDP"][:100]))])) nums["LandPRmin"] = fmt(np.min(vr["land_PR"][:100]), 1) recs = table_scenarios(allP, ["baseline", "no_ai", "no_capability", "rapid_agi", "slow_alpha", "fast_alpha", "alpha_inf_x5", "fast_moore_cheap_energy", "fast_moore_energy_constrained", "moore_ends_2040", "moore_floor_1e3", "moore_never_ends", "complements_leontief_like", "complements_cobb_douglas", "substitutes", "scarce_land", "severe_chip_depreciation", "reinstatement", "low_risk_aversion", "high_risk_aversion"], "tab04_scenarios.tex") pd.DataFrame(recs).to_csv(os.path.join(ROOT, "results", "data", "scenario_summary.csv"), index=False) for n, key in [("moore_ends_2040", "MooreEnds"), ("moore_floor_1e3", "MooreFloor"), ("alpha_inf_x5", "AlphaInfFive"), ("rapid_agi", "RapidAGI"), ("no_capability", "NoCapability")]: if n in allP: S = summary_by_year(allP[n], (2050, 2100)) for y in (2050, 2100): nums[f"wL{key}{macro_name(str(y))}"] = fmt(S[y]["wL"], 3) nums[f"LS{key}{macro_name(str(y))}"] = fmt(S[y]["labor_share"], 3) nums[f"GDP{key}{macro_name(str(y))}"] = fmt(S[y]["GDP"], 1) dec = table_decomposition(allP) for c, d in dec.items(): nums[f"Dec{macro_name(c)}CF"] = fmt(100 * d["cash_flow"], 0) nums[f"Dec{macro_name(c)}DR"] = fmt(100 * d["discount_rate"], 0) nums[f"Dec{macro_name(c)}Tot"] = fmt(100 * d["total"], 0) if "baseline" in allP: pol = table_policy(allP, params) for n, d in pol.items(): nums[f"CEVW{macro_name(n)}"] = fmt(100 * d["cevW"], 1) nums[f"CEVK{macro_name(n)}"] = fmt(100 * d["cevK"], 1) rows = table_diagnostics(list(LAB)) nums["NumScenarios"] = str(len(rows)) nums["MaxResidualAll"] = f"{max(r[2] for r in rows):.0e}" if rows else "--" nums["MaxIncomeGapAll"] = f"{max(r[3] for r in rows):.0e}" if rows else "--" # keep previously generated HA numbers if present ha = os.path.join(ROOT, "results", "data", "ha_numbers.json") if os.path.exists(ha): nums.update(json.load(open(ha))) write_numbers(nums) print("numbers:", len(nums), "tables written")