ai_bust_models.py
raw file ยท 588 lines
"""
When the Buildout Breaks: a quantitative model of an AI investment bust.
Five linked models + a Monte Carlo driver:
A. Revenue gap - end-customer AI revenue needed to earn a return on the installed AI capital stock
B. Depreciation - hyperscaler operating income under shorter GPU/server useful lives
C. Neocloud solvency - interest coverage and default thresholds of a leveraged GPU cloud
D. Contagion network - revenue + credit shock propagation across 12 sectors (DebtRank-style, with feedback)
E. Macro transmission - GDP, unemployment and equity losses from D's outputs
MC. Monte Carlo - 10,000 draws over uncertain parameters, chaining D -> E
All $ figures are USD billions unless noted. Real anchors (as of Sept 2026) are marked [REAL];
everything else is a modelling assumption marked [ASSUMED] and is varied in the Monte Carlo.
Run: python3 ai_bust_models.py -> writes results.json
"""
import json
import numpy as np
RNG = np.random.default_rng(20260930)
# ---------------------------------------------------------------------------
# Model A - Revenue gap
# ---------------------------------------------------------------------------
# Global AI capex by year [REAL for 2026 ~$1T per Goldman; 2024/2025 ASSUMED from hyperscaler history]
GLOBAL_AI_CAPEX = {2024: 400.0, 2025: 650.0, 2026: 1000.0}
SHARE_SHORT_LIVED = 0.60 # [ASSUMED] share of capex in GPUs/servers/network (short life); rest = buildings, power
BUILDING_LIFE = 20 # [ASSUMED] years
STACK_MARGIN = 0.50 # [ASSUMED] cash operating margin (before D&A) of the whole AI stack on end-customer revenue
END_REVENUE_2026 = 220.0 # [ASSUMED from REAL] ~$140B frontier-lab run rate + ~$80B other direct AI revenue
def annuity(rate, years):
"""Annual payment that repays 1 unit of capital over `years` at `rate` (capital charge incl. return)."""
return rate / (1.0 - (1.0 + rate) ** (-years))
def required_revenue(capital_stock, gpu_life, hurdle, short_share=SHARE_SHORT_LIVED, margin=STACK_MARGIN):
short = capital_stock * short_share
long_ = capital_stock * (1 - short_share)
charge = short * annuity(hurdle, gpu_life) + long_ * annuity(hurdle, BUILDING_LIFE)
return charge / margin, charge
def model_a():
stock = sum(GLOBAL_AI_CAPEX.values()) # installed through end-2026
grid = []
for life in [3, 4, 5, 6]:
for hurdle in [0.08, 0.10, 0.12]:
req, charge = required_revenue(stock, life, hurdle)
grid.append({"gpu_life": life, "hurdle": hurdle, "required_revenue": round(req, 1),
"capital_charge": round(charge, 1), "gap_multiple": round(req / END_REVENUE_2026, 2)})
# forward view: capital stock keeps growing with planned capex; how fast must revenue grow?
capex_plan = {2027: 1350.0, 2028: 1600.0} # [ASSUMED] consistent with Goldman 2.5%/2.8% of GDP path
path = []
k = stock
for yr in [2026, 2027, 2028]:
if yr in capex_plan:
k += capex_plan[yr]
req, _ = required_revenue(k, 5, 0.10)
path.append({"year": yr, "capital_stock": round(k), "required_revenue_5y_10pct": round(req)})
# compound growth needed from 2026 revenue to hit 2028 requirement
need_2028 = path[-1]["required_revenue_5y_10pct"]
cagr = (need_2028 / END_REVENUE_2026) ** 0.5 - 1
return {"capital_stock_2026": stock, "end_revenue_2026": END_REVENUE_2026, "grid": grid,
"forward_path": path, "required_cagr_2026_2028": round(cagr, 3)}
# ---------------------------------------------------------------------------
# Model B - Depreciation stress (Big 4 hyperscalers)
# ---------------------------------------------------------------------------
BIG4_CAPEX = {2022: 150.0, 2023: 150.0, 2024: 230.0, 2025: 410.0, 2026: 730.0, # [REAL approx] 2025-26 per guidance
2027: 850.0, 2028: 900.0} # [ASSUMED] base plan
SERVER_SHARE = 0.60 # [ASSUMED] share of capex in servers/network
BIG4_OPINC_PRE_DEP = {2026: 650.0, 2027: 790.0, 2028: 930.0}
# [ASSUMED] operating income before *server* depreciation; calibrated so reported (6y) 2026 op income ~ $520B
def server_depreciation(year, life):
total = 0.0
for y, capex in BIG4_CAPEX.items():
age = year - y
if 0 <= age < life:
# half-year convention in year of purchase
frac = 0.5 if age == 0 else (1.0 if age < life else 0.0)
total += capex * SERVER_SHARE / life * frac
elif age == life:
total += capex * SERVER_SHARE / life * 0.5
return total
def model_b():
rows = []
for year in [2026, 2027, 2028]:
base = server_depreciation(year, 6)
rep = BIG4_OPINC_PRE_DEP[year] - base
for life in [6, 5, 4, 3]:
dep = server_depreciation(year, life)
opinc = BIG4_OPINC_PRE_DEP[year] - dep
rows.append({"year": year, "life": life, "server_depreciation": round(dep, 1),
"operating_income": round(opinc, 1),
"change_vs_6y": round(opinc - rep, 1),
"pct_change_vs_6y": round((opinc - rep) / rep, 3)})
cum_understatement_3y = sum(server_depreciation(y, 3) - server_depreciation(y, 6) for y in [2026, 2027, 2028])
cum_understatement_4y = sum(server_depreciation(y, 4) - server_depreciation(y, 6) for y in [2026, 2027, 2028])
return {"rows": rows, "cum_understatement_2026_28_vs_3y": round(cum_understatement_3y, 1),
"cum_understatement_2026_28_vs_4y": round(cum_understatement_4y, 1)}
# ---------------------------------------------------------------------------
# Model C - Neocloud solvency (stylized on CoreWeave's disclosed 2026 figures)
# ---------------------------------------------------------------------------
NC = dict(
revenue=12.8, # [REAL] 2026 guidance $12.4-13.2B
cash_cost_share=0.44, # [REAL-derived] adjusted EBITDA margin ~56%
debt=35.1, # [REAL] total debt June 2026
interest=2.56, # [REAL] ~$640M/quarter annualized
principal_2027=6.2, # [REAL] principal due 2027
contracted_share=0.80, # [ASSUMED] share of revenue under take-or-pay contracts
gpu_book=36.4, # [REAL] property & equipment, Mar 2026
)
def neocloud(spot_decline, counterparty_loss, renew_haircut=0.0, p=NC):
"""Return EBITDA, interest coverage and loan-to-value after shocks.
spot_decline: fractional fall in spot/uncontracted GPU rental prices
counterparty_loss: share of contracted revenue lost to customer failure/renegotiation
renew_haircut: price cut on contracts that roll within the year (applied to 1/3 of contracted book)
"""
contracted = p["revenue"] * p["contracted_share"]
spot = p["revenue"] - contracted
contracted_after = contracted * (1 - counterparty_loss) * (1 - renew_haircut / 3)
spot_after = spot * (1 - spot_decline)
rev = contracted_after + spot_after
cash_costs = p["revenue"] * p["cash_cost_share"] * 0.85 + rev * p["cash_cost_share"] * 0.15 # 85% fixed
ebitda = rev - cash_costs
coverage = ebitda / p["interest"]
# collateral: used-GPU value tracks rental prices with ~1:1 elasticity plus 25% forced-sale discount
collateral = p["gpu_book"] * (1 - spot_decline) * 0.75
ltv = p["debt"] / collateral if collateral > 0 else float("inf")
dscr = ebitda / (p["interest"] + p["principal_2027"])
return {"revenue": rev, "ebitda": ebitda, "coverage": coverage, "dscr": dscr, "ltv": ltv}
def model_c():
curves = []
for cl in [0.0, 0.10, 0.25, 0.40]:
for sd in np.linspace(0, 0.8, 17):
r = neocloud(sd, cl)
curves.append({"counterparty_loss": cl, "spot_decline": round(float(sd), 2),
"coverage": round(r["coverage"], 2), "dscr": round(r["dscr"], 2),
"ltv": round(r["ltv"], 2)})
base = neocloud(0, 0)
# find breakpoints: coverage < 1.0 (can't pay interest) for each counterparty loss
breaks = {}
for cl in [0.0, 0.10, 0.25, 0.40]:
bp = None
for sd in np.linspace(0, 1, 201):
if neocloud(sd, cl)["coverage"] < 1.0:
bp = round(float(sd), 3)
break
breaks[str(cl)] = bp
return {"base": {k: round(v, 2) for k, v in base.items()}, "curves": curves, "interest_breakpoints": breaks}
# ---------------------------------------------------------------------------
# Model D - Contagion network
# ---------------------------------------------------------------------------
SECTORS = ["Enterprise demand", "Frontier labs", "Hyperscalers", "Neoclouds", "Chip designers",
"Memory & foundry", "DC developers", "Utilities", "AI startups", "Private credit",
"Banks", "Pensions & insurers"]
IDX = {s: i for i, s in enumerate(SECTORS)}
N = len(SECTORS)
# Annual spending flows payer -> payee ($B, 2027 base run rate) [ASSUMED, calibrated to 2026 disclosures]
FLOWS = [
("Enterprise demand", "Frontier labs", 260),
("Enterprise demand", "Hyperscalers", 150),
("Enterprise demand", "AI startups", 60),
("AI startups", "Frontier labs", 35),
("AI startups", "Hyperscalers", 15),
("Frontier labs", "Hyperscalers", 150),
("Frontier labs", "Neoclouds", 45),
("Frontier labs", "Chip designers", 40),
("Hyperscalers", "Neoclouds", 25),
("Hyperscalers", "Chip designers", 230),
("Hyperscalers", "DC developers", 70),
("Hyperscalers", "Utilities", 30),
("Neoclouds", "Chip designers", 45),
("Neoclouds", "DC developers", 20),
("Neoclouds", "Utilities", 8),
("Chip designers", "Memory & foundry", 140),
("DC developers", "Utilities", 10),
]
# Which outflows are "investment-like" (cut more than proportionally: accelerator effect)
CAPEX_FLOWS = {("Frontier labs", "Chip designers"), ("Hyperscalers", "Chip designers"),
("Neoclouds", "Chip designers"), ("Chip designers", "Memory & foundry")}
# Share of a flow locked in by take-or-pay contracts / leases: it can only be cut if the payer fails
# [REAL-anchored: CoreWeave $104B backlog is committed contracts; leases ASSUMED]
CONTRACTED = {("Frontier labs", "Neoclouds"): 0.8, ("Hyperscalers", "Neoclouds"): 0.8,
("Hyperscalers", "DC developers"): 0.9, ("Neoclouds", "DC developers"): 0.9}
# Exogenous revenue (not from other modelled sectors), $B - sovereign, non-AI business, etc. [ASSUMED]
EXOG_REV = {"Frontier labs": 0, "Hyperscalers": 900, "Neoclouds": 5, "Chip designers": 110,
"Memory & foundry": 260, "DC developers": 30, "Utilities": 450, "AI startups": 5}
# Share of revenue that is variable cost (profit falls by revenue loss x (1 - this)) [ASSUMED]
VAR_COST = {"Frontier labs": 0.45, "Hyperscalers": 0.35, "Neoclouds": 0.20, "Chip designers": 0.30,
"Memory & foundry": 0.40, "DC developers": 0.15, "Utilities": 0.60, "AI startups": 0.50}
# Loss-absorbing buffers ($B): equity/cash that can absorb losses before default [ASSUMED]
BUFFER = {"Enterprise demand": 1e9, "Frontier labs": 200, "Hyperscalers": 1400, "Neoclouds": 18,
"Chip designers": 320, "Memory & foundry": 260, "DC developers": 55, "Utilities": 120,
"AI startups": 40, "Private credit": 230, "Banks": 1500, "Pensions & insurers": 900}
# Credit claims lender -> borrower ($B) [ASSUMED; private credit ~$200B+ to AI per Quinn Emanuel, plus SPVs]
CLAIMS = [
("Private credit", "Neoclouds", 55),
("Private credit", "DC developers", 140),
("Private credit", "Frontier labs", 20),
("Private credit", "AI startups", 15),
("Private credit", "Hyperscalers", 110), # off-balance-sheet SPV/lease financings
("Banks", "Neoclouds", 30),
("Banks", "DC developers", 60),
("Banks", "Private credit", 90), # fund leverage / NAV loans
("Pensions & insurers", "Hyperscalers", 280),
("Pensions & insurers", "DC developers", 60), # ABS/CMBS
("Pensions & insurers", "Private credit", 180), # LP stakes (marked to fund NAV)
]
LP_STAKE_SHARE = 0.6 # [ASSUMED] share of private-credit fund losses borne by pensions/insurers as LPs
HORIZON = 2 # stress horizon in years (2027-2028); annual profit losses accumulate over it
# Debt maturing inside the horizon that must be refinanced [REAL-anchored: CoreWeave $6.2B due 2027; rest ASSUMED]
REFI = {"Neoclouds": 22, "DC developers": 45}
# Baseline cash burn over the horizon, funded by new equity when markets are open [REAL-anchored:
# OpenAI plan shows ~$278B negative FCF 2026-30 (~$55B/yr); other labs + startups ASSUMED]
BURN = {"Frontier labs": 150, "AI startups": 50}
def build():
F = np.zeros((N, N))
for a, b, v in FLOWS:
F[IDX[a], IDX[b]] = v
C = np.zeros((N, N))
for a, b, v in CLAIMS:
C[IDX[a], IDX[b]] = v
capex_mask = np.zeros((N, N), dtype=bool)
for a, b in CAPEX_FLOWS:
capex_mask[IDX[a], IDX[b]] = True
K = np.zeros((N, N))
for (a, b), v in CONTRACTED.items():
K[IDX[a], IDX[b]] = v
return F, C, capex_mask, K
F0, C0, CAPEX_MASK, KONTRACT = build()
def run_contagion(demand_shock, accel=1.8, distress_cut=0.6, lgd=0.55, fire_sale=0.35,
buffer_scale=1.0, lab_funding_freeze=0.0, refi_sens=1.0, rounds=12, record=False,
legal_on=False, legal_shape=2.0, legal_scale=0.2, legal_disc=0.08, legal_freeze_k=0.5,
legal_rng=None):
"""Propagate an end-demand shock through revenue, funding and credit channels.
demand_shock : fractional fall in enterprise/consumer AI spending vs plan
accel : capex accelerator - % cut in investment flows per % revenue shortfall
distress_cut : extra spending cut per unit of buffer depletion (financial-stress channel)
lgd : loss given default on credit claims
fire_sale : extra LGD on GPU-collateralised loans when chip demand collapses
lab_funding_freeze : share of labs'/startups' planned cash burn that can no longer be raised
refi_sens : how sharply refinancing markets close as borrowers and lenders get stressed
legal_on : stochastic legal friction. A borrower's technical default is followed by a Gamma(shape, scale)
delay (years) before its collateral clears; lenders only book the loss on resolution, and
claims in the queue are 'frozen capital'. [ASSUMED shape/scale; mean = shape*scale = 0.4 yr]
legal_disc : annual time-value discount; added to LGD in proportion to the delay (years)
legal_freeze_k : weight of frozen claims on lenders' buffers in the refinancing-stress term
"""
F = F0.copy()
buffer = np.array([BUFFER[s] for s in SECTORS]) * buffer_scale
base_rev = F0.sum(axis=0) + np.array([EXOG_REV.get(s, 0) for s in SECTORS])
loss = np.zeros(N)
defaulted = np.zeros(N, dtype=bool)
cut = np.zeros(N) # current spending cut fraction by payer (non-capex)
capex_cut = np.zeros(N) # current cut on capex-like outflows
cut[IDX["Enterprise demand"]] = demand_shock
history = []
credit_loss_paid = np.zeros((N, N))
yrs_per_round = HORIZON / rounds
resolve_round = np.full(N, -1.0) # round at which a defaulted borrower's collateral clears
delay_yrs = np.zeros(N)
frozen_peak = 0.0
frozen_now = 0.0
frozen_by_lender = np.zeros(N)
if legal_on and legal_rng is None:
legal_rng = np.random.default_rng(0)
for t in range(rounds):
# apply cuts to flows
F = F0.copy()
for i in range(N):
protect = np.zeros(N) if defaulted[i] else KONTRACT[i] # contracts bind unless the payer fails
nm = ~CAPEX_MASK[i]
F[i, nm] = F0[i, nm] * (1 - cut[i] * (1 - protect[nm]))
F[i, CAPEX_MASK[i]] = F0[i, CAPEX_MASK[i]] * (1 - capex_cut[i])
rev = F.sum(axis=0) + np.array([EXOG_REV.get(s, 0) for s in SECTORS])
rev_short = np.clip(base_rev - rev, 0, None)
profit_loss = np.array([rev_short[i] * (1 - VAR_COST.get(SECTORS[i], 0.3)) for i in range(N)]) * HORIZON
rev_short_frac = np.divide(rev_short, base_rev, out=np.zeros(N), where=base_rev > 0)
# funding channel 1: planned cash burn that new equity rounds no longer cover
funding_loss = np.zeros(N)
for s, burn in BURN.items():
i = IDX[s]
uncovered = min(1.0, lab_funding_freeze + 1.5 * rev_short_frac[i])
funding_loss[i] += burn * uncovered
# funding channel 2: maturing debt that cannot be rolled over must be repaid from buffers
lender_stress = min(1.0, 0.7 * min(loss[IDX["Private credit"]] / BUFFER["Private credit"], 1)
+ 0.3 * min(loss[IDX["Banks"]] / BUFFER["Banks"], 1))
if legal_on: # frozen claims are illiquid: lenders cannot recycle that capital into new lending
fz = (0.7 * min(frozen_by_lender[IDX["Private credit"]] / BUFFER["Private credit"], 1)
+ 0.3 * min(frozen_by_lender[IDX["Banks"]] / BUFFER["Banks"], 1))
lender_stress = min(1.0, lender_stress + legal_freeze_k * fz)
for s, due in REFI.items():
i = IDX[s]
# operating stress only (revenue shortfall + operating losses), so a failed rollover does not feed on itself
borrower_stress = min(1.0, 2.0 * rev_short_frac[i] + min(profit_loss[i] / buffer[i], 1))
fail = min(1.0, refi_sens * (0.6 * borrower_stress + 2.0 * lender_stress))
funding_loss[i] += due * fail
# credit channel: losses on claims against defaulted borrowers
chip_collapse = 1 - F[:, IDX["Chip designers"]].sum() / F0[:, IDX["Chip designers"]].sum()
new_credit = np.zeros(N)
frozen_by_lender = np.zeros(N)
for a, b, v in CLAIMS:
ia, ib = IDX[a], IDX[b]
in_queue = legal_on and defaulted[ib] and t < resolve_round[ib]
if in_queue:
frozen_by_lender[ia] += v # claim stuck in the legal queue: no loss booked, no cash back
continue
if defaulted[ib]:
l = lgd + (fire_sale * chip_collapse if b in ("Neoclouds", "DC developers") else 0)
if legal_on:
l += legal_disc * delay_yrs[ib] # time-value cost of the wait
owed = v * min(l, 0.95)
inc = owed - credit_loss_paid[ia, ib]
if inc > 0:
new_credit[ia] += inc
credit_loss_paid[ia, ib] = owed
else:
# mark-to-market of stressed but not defaulted borrowers (spread widening)
stress = min(1.0, loss[ib] / max(BUFFER[b] * buffer_scale, 1e-9))
mtm = v * 0.10 * stress
inc = mtm - credit_loss_paid[ia, ib]
if inc > 0:
new_credit[ia] += inc
credit_loss_paid[ia, ib] = mtm
# private-credit fund losses pass through to LPs
pc_total = credit_loss_paid[IDX["Private credit"]].sum()
lp_loss = pc_total * LP_STAKE_SHARE
total_credit = credit_loss_paid.sum(axis=1).copy()
total_credit[IDX["Pensions & insurers"]] = credit_loss_paid[IDX["Pensions & insurers"]].sum() + lp_loss
loss = profit_loss + total_credit + funding_loss
loss[IDX["Enterprise demand"]] = 0
depletion = np.clip(loss / buffer, 0, None)
newly = (depletion >= 1.0) & ~defaulted
defaulted |= depletion >= 1.0
if legal_on:
for i in np.where(newly)[0]:
d_y = float(legal_rng.gamma(legal_shape, legal_scale))
delay_yrs[i] = d_y
resolve_round[i] = t + 1 + d_y / yrs_per_round # booked on a later round
frozen_now = float(frozen_by_lender.sum())
frozen_peak = max(frozen_peak, frozen_now)
# behavioural response for next round
for i, s in enumerate(SECTORS):
if s == "Enterprise demand":
continue
if defaulted[i]:
cut[i] = max(cut[i], 1 - 0.4) # operations shrink to 40% (run-off)
capex_cut[i] = 1.0
else:
ai_share = F0[:, i].sum() / base_rev[i] if base_rev[i] > 0 else 0
# AI revenue shortfall drives AI capex; exogenous revenue cushions
ai_short = rev_short[i] / max(F0[:, i].sum(), 1e-9)
cut[i] = min(0.9, 0.5 * rev_short_frac[i] + distress_cut * 0.3 * min(depletion[i], 1))
capex_cut[i] = min(0.95, accel * ai_short * max(ai_share, 0.35) / 0.35 * 0.5
+ distress_cut * min(depletion[i], 1))
if record:
history.append({"round": t + 1,
"depletion": {s: round(float(min(depletion[i], 1.5)), 3) for i, s in enumerate(SECTORS)},
"defaults": [s for i, s in enumerate(SECTORS) if newly[i]],
"frozen": round(frozen_now, 1)})
ai_capex_base = sum(F0[i, j] for i in range(N) for j in range(N) if CAPEX_MASK[i, j] and SECTORS[i] != "Chip designers")
ai_capex_now = sum(F[i, j] for i in range(N) for j in range(N) if CAPEX_MASK[i, j] and SECTORS[i] != "Chip designers")
out = {
"loss": {s: float(loss[i]) for i, s in enumerate(SECTORS)},
"depletion": {s: float(depletion[i]) for i, s in enumerate(SECTORS)},
"defaulted": [s for i, s in enumerate(SECTORS) if defaulted[i]],
"capex_cut": float(1 - ai_capex_now / ai_capex_base),
"chip_revenue_fall": float(chip_collapse),
"credit_losses": float(credit_loss_paid.sum() + lp_loss * 0),
"total_loss": float(sum(loss[i] for i in range(N) if SECTORS[i] != "Pensions & insurers") + total_credit[IDX["Pensions & insurers"]]),
"frozen_peak": float(frozen_peak), "frozen_end": float(frozen_by_lender.sum()),
"mean_delay_wk": float(52 * delay_yrs[delay_yrs > 0].mean()) if (delay_yrs > 0).any() else 0.0,
}
if record:
out["history"] = history
return out
# ---------------------------------------------------------------------------
# Model E - Macro transmission
# ---------------------------------------------------------------------------
US_GDP = 31_000.0 # [ASSUMED ~REAL] 2026 nominal GDP, $B
AI_CAPEX_US_SHARE = 0.025 # [REAL, Goldman] planned 2027 US AI capex share of GDP
DOMESTIC_CONTENT = 0.55 # [ASSUMED] share of AI capex that is US value added (chips largely imported)
MULTIPLIER = 1.3 # [ASSUMED] investment multiplier
HH_EQUITY = 58_000.0 # [ASSUMED ~REAL] US household direct+indirect equity holdings, $B
MPC_WEALTH = 0.03 # [ASSUMED, literature 0.02-0.05] consumption per $ of stock wealth
AI_INDEX_WEIGHT = 0.42 # [ASSUMED] AI-linked share of S&P 500 market cap
OKUN = 0.5 # [standard] unemployment rise per point of output gap
BASE_UNEMP = 4.3 # [ASSUMED ~REAL]
def model_e(capex_cut, ai_equity_drawdown, credit_losses, spill=0.35):
"""capex_cut: fall in AI capex vs plan; ai_equity_drawdown: fall in AI-linked stocks;
spill: drawdown of non-AI stocks as a share of the AI drawdown."""
inv = capex_cut * AI_CAPEX_US_SHARE * DOMESTIC_CONTENT * MULTIPLIER # share of GDP
sp500 = AI_INDEX_WEIGHT * ai_equity_drawdown + (1 - AI_INDEX_WEIGHT) * ai_equity_drawdown * spill
wealth = HH_EQUITY * sp500 * MPC_WEALTH / US_GDP
credit = credit_losses * 2.0 * 0.05 / US_GDP # lost lending ~ 2x losses, 5% of which hits spending
gdp_hit = inv + wealth + credit
unemp = BASE_UNEMP + OKUN * gdp_hit * 100
return {"gdp_hit_pct": gdp_hit * 100, "inv_pct": inv * 100, "wealth_pct": wealth * 100,
"credit_pct": credit * 100, "sp500_drawdown": sp500, "unemployment_peak": unemp}
def equity_drawdown(d, valuation_reset, demand_shock):
"""AI-linked equity drawdown = 1 - (1 - multiple reset) x (1 - realised earnings hit) x (1 - expectations hit).
Realised earnings hit comes from the contagion model; expectations hit prices the lower growth path."""
earnings_hit = 0.5 * min(d["depletion"]["Hyperscalers"] * 3, 1) + 0.5 * min(d["chip_revenue_fall"], 1)
expectations_hit = 0.7 * demand_shock
return min(0.9, 1 - (1 - valuation_reset) * (1 - 0.8 * earnings_hit) * (1 - expectations_hit))
# ---------------------------------------------------------------------------
# Monte Carlo
# ---------------------------------------------------------------------------
def classify(capex_cut, sp, defaults):
if "Banks" in defaults or "Pensions & insurers" in defaults:
return "Systemic crisis"
if capex_cut >= 0.35 or sp >= 0.30:
return "Bust"
if capex_cut >= 0.15 or sp >= 0.15:
return "Correction"
return "Soft landing"
def monte_carlo(n=10_000, shock_sampler=None, contagion_kwargs=None):
"""shock_sampler: optional function () -> shortfall in [0, 1). Default = the illustrative 45/35/20 prior;
ai_bust_probability.py passes the shortfall distribution estimated in Part I of the paper."""
rows = []
for _ in range(n):
if shock_sampler is not None:
shock = float(shock_sampler())
else:
# Demand shortfall vs plan: mixture - most mass small, fat right tail (illustrative prior)
regime = RNG.random()
if regime < 0.45:
shock = RNG.uniform(0.0, 0.10) # growth roughly meets plan
elif regime < 0.80:
shock = RNG.uniform(0.10, 0.30) # ROI wall, partial stall
else:
shock = RNG.uniform(0.30, 0.55) # hard stall
p = dict(
demand_shock=shock,
accel=RNG.triangular(1.0, 1.8, 3.0),
distress_cut=RNG.uniform(0.3, 0.9),
lgd=RNG.uniform(0.35, 0.75),
fire_sale=RNG.uniform(0.1, 0.5),
buffer_scale=RNG.uniform(0.8, 1.2),
lab_funding_freeze=0.0, # set below
refi_sens=RNG.uniform(0.5, 1.5),
)
# funding markets close in proportion to the disappointment, plus independent sentiment noise
funding_noise = RNG.uniform(-0.15, 0.25)
p["lab_funding_freeze"] = float(np.clip(1.2 * shock + funding_noise, 0.0, 0.85))
ck = dict(contagion_kwargs or {})
if ck.get("legal_on"):
ck["legal_rng"] = RNG
d = run_contagion(**p, **ck)
vr = RNG.uniform(0.0, 0.30) # pure multiple compression, independent of fundamentals
ai_dd = equity_drawdown(d, vr, shock)
spill = RNG.uniform(0.2, 0.5)
m = model_e(d["capex_cut"], ai_dd, d["credit_losses"], spill)
rows.append({**{k: float(v) for k, v in p.items()}, "funding_noise": funding_noise, "valuation_reset": vr, "spill": spill,
"capex_cut": d["capex_cut"], "chip_fall": d["chip_revenue_fall"],
"neocloud_default": "Neoclouds" in d["defaulted"],
"lab_default": "Frontier labs" in d["defaulted"],
"dc_default": "DC developers" in d["defaulted"],
"pc_default": "Private credit" in d["defaulted"],
"defaults": d["defaulted"], "credit_losses": d["credit_losses"],
"frozen_peak": d["frozen_peak"], "frozen_end": d["frozen_end"],
"pension_loss": d["loss"]["Pensions & insurers"],
"bank_loss": d["loss"]["Banks"],
"ai_equity_dd": ai_dd, **m,
"outcome": classify(d["capex_cut"], m["sp500_drawdown"], d["defaulted"])})
return rows
def summarize(rows):
import collections
out = {}
cnt = collections.Counter(r["outcome"] for r in rows)
out["outcome_probs"] = {k: cnt.get(k, 0) / len(rows) for k in ["Soft landing", "Correction", "Bust", "Systemic crisis"]}
def pct(key, qs=(5, 25, 50, 75, 95), subset=None):
vals = np.array([r[key] for r in (subset or rows)])
return {f"p{q}": round(float(np.percentile(vals, q)), 4) for q in qs}
out["capex_cut"] = pct("capex_cut")
out["sp500_drawdown"] = pct("sp500_drawdown")
out["gdp_hit_pct"] = pct("gdp_hit_pct")
out["unemployment_peak"] = pct("unemployment_peak")
out["credit_losses"] = pct("credit_losses")
out["frozen_peak"] = pct("frozen_peak")
for flag in ["neocloud_default", "lab_default", "dc_default", "pc_default"]:
out["p_" + flag] = float(np.mean([r[flag] for r in rows]))
out["p_bank_default"] = float(np.mean(["Banks" in r["defaults"] for r in rows]))
out["p_pension_default"] = float(np.mean(["Pensions & insurers" in r["defaults"] for r in rows]))
busts = [r for r in rows if r["outcome"] in ("Bust", "Systemic crisis")]
if busts:
out["conditional_on_bust"] = {k: pct(k, subset=busts) for k in
["capex_cut", "sp500_drawdown", "gdp_hit_pct", "unemployment_peak", "credit_losses", "chip_fall"]}
out["conditional_on_bust"]["p_neocloud_default"] = float(np.mean([r["neocloud_default"] for r in busts]))
# sensitivity: Spearman rank correlation of inputs with S&P drawdown and GDP hit
from scipy.stats import spearmanr
inputs = ["demand_shock", "accel", "distress_cut", "lgd", "fire_sale", "buffer_scale", "funding_noise", "refi_sens", "valuation_reset", "spill"]
sens = {}
for tgt in ["sp500_drawdown", "gdp_hit_pct", "capex_cut"]:
y = [r[tgt] for r in rows]
sens[tgt] = {k: round(float(spearmanr([r[k] for r in rows], y).statistic), 3) for k in inputs}
out["sensitivity"] = sens
# dose-response: outcome probabilities by demand shock bucket
buckets = [(0, .1), (.1, .2), (.2, .3), (.3, .4), (.4, .55)]
dr = []
for lo, hi in buckets:
sub = [r for r in rows if lo <= r["demand_shock"] < hi]
if not sub:
continue
dr.append({"bucket": f"{int(lo*100)}-{int(hi*100)}%", "n": len(sub),
"p_bust_or_worse": round(float(np.mean([r["outcome"] in ("Bust", "Systemic crisis") for r in sub])), 3),
"p_neocloud_default": round(float(np.mean([r["neocloud_default"] for r in sub])), 3),
"median_sp500_dd": round(float(np.median([r["sp500_drawdown"] for r in sub])), 3),
"median_gdp_hit": round(float(np.median([r["gdp_hit_pct"] for r in sub])), 3),
"median_unemp": round(float(np.median([r["unemployment_peak"] for r in sub])), 2)})
out["dose_response"] = dr
return out
if __name__ == "__main__":
res = {"model_a": model_a(), "model_b": model_b(), "model_c": model_c()}
# representative chain reaction: a 30% demand shortfall with central parameters
rep = run_contagion(0.30, accel=1.8, distress_cut=0.6, lgd=0.55, fire_sale=0.35,
lab_funding_freeze=0.35, refi_sens=1.0, record=True)
res["model_d_representative"] = rep
rep_dd = equity_drawdown(rep, 0.15, 0.30)
res["model_e_representative"] = model_e(rep["capex_cut"], rep_dd, rep["credit_losses"])
res["model_e_representative"]["ai_equity_dd"] = rep_dd
rows = monte_carlo(10_000)
res["monte_carlo"] = summarize(rows)
with open("results.json", "w") as f:
json.dump(res, f, indent=1, default=float)
print(json.dumps({k: v for k, v in res.items() if k != "model_d_representative"}, indent=1, default=float)[:6000])
print("REP", json.dumps({k: rep[k] for k in ["defaulted", "capex_cut", "chip_revenue_fall", "credit_losses", "total_loss"]}, default=float))
print("REP depletion", {k: round(v, 2) for k, v in rep["depletion"].items()})
def cliff_sweep():
"""Deterministic sweep of the demand shock with central parameters (funding freeze tied to shock)."""
out = []
for s in np.round(np.arange(0.0, 0.5001, 0.025), 3):
r = run_contagion(float(s), lab_funding_freeze=float(min(0.85, 1.2 * s + 0.05)))
out.append({"shock": float(s), "credit_losses": round(r["credit_losses"], 1),
"capex_cut": round(r["capex_cut"], 3), "defaults": r["defaulted"],
"lab_depletion": round(r["depletion"]["Frontier labs"], 3),
"neocloud_depletion": round(r["depletion"]["Neoclouds"], 3)})
return out
if __name__ == "__main__":
res = json.load(open("results.json"))
res["cliff"] = cliff_sweep()
json.dump(res, open("results.json", "w"), indent=1, default=float)