1"""Tests of the static blocks: Restrepo replication, KKT conditions, national accounting identity."""2import os3import sys45import numpy as np6import pytest78sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src"))910from agi_ge.calibration import calibrate # noqa: E40211from agi_ge.production import housing_rent, income_identity, static_period, value_added_by_sector # noqa: E40212from agi_ge.tasks import Technology, restrepo_benchmark, technology_from_params # noqa: E4021314NU = [0.3, 0.25, 0.25, 0.2, 0, 0]15GAMMA = [0, 0, 0, 0, 0.05, 0.05]16SKILL = [0, 0, 1, 1, 0, 1]17ALPHA = np.array([1.0, 2.0, 4.0, 8.0, 3.0, 6.0])18H = [0.6, 0.4]192021@pytest.fixture(scope="module")22def bench():23 return restrepo_benchmark(NU, GAMMA, SKILL, -1.0, ALPHA, H, [0.0, 1.0, 10.0, 100.0, 1e3, 1e4, 1e5])242526def test_kkt_conditions(bench):27 tech = Technology(nu=np.array(NU), gamma=np.array(GAMMA), skill=np.array(SKILL), rho_B=-1.0,28 nu_N=0, rho_N=0, nu_T=0, nu_E=0, rho_T=0)29 for s in bench:30 assert tech.check_kkt(s, H, s["Q_total"], ALPHA) < 1e-6313233def test_restrepo_P1_all_bottlenecks_automated(bench):34 s = bench[-1]35 assert np.all(s["Q"][:4] > 0) # compute used in every bottleneck task363738def test_restrepo_P2_P3_output_linear_in_compute(bench):39 s1, s2 = bench[-2], bench[-1]40 A = s2["A"]41 # Y = A (Q + sum CEU_tilde H): slope in Q equals A, intercept equals the compute-equivalent labor42 slope = (s2["Y"] - s1["Y"]) / (s2["Q_total"] - s1["Q_total"])43 assert abs(slope / A - 1) < 1e-644 intercept = s2["Y"] - A * s2["Q_total"]45 assert abs(intercept - np.sum(s2["w"] * np.array(H))) < 1e-6464748def test_restrepo_P4_wages_equal_A_times_CEU(bench):49 s = bench[-1]50 A = s["A"]51 # skill H: its best automated task has alpha = 8 = CEU(H) -> w_H = A * 852 assert abs(s["w"][1] / (A * 8.0) - 1) < 1e-653 # skill L: best automated task alpha = 2 (accessory task alpha=3 is not worth automating): discounted CEU54 assert abs(s["w"][0] / (A * 2.0) - 1) < 1e-655 assert s["w"][0] < A * 3.0565758def test_restrepo_P5_labor_share_vanishes(bench):59 ls = [s["labor_share"] for s in bench]60 assert ls[0] == pytest.approx(1.0)61 assert ls[-1] < 1e-362 assert bench[-1]["compute_share"] > 0.999636465def test_restrepo_P6_total_wages_higher_with_AGI(bench):66 pre = np.sum(bench[0]["w"] * np.array(H))67 post = np.sum(bench[-1]["w"] * np.array(H))68 assert post > pre697071@pytest.fixture(scope="module")72def calibrated():73 p, info = calibrate()74 return p, info, technology_from_params(p)757677def test_income_identity_holds(calibrated):78 p, info, tech = calibrated79 al = p.alpha_path()80 for t, KQ in [(0, 1.0), (20, 50.0), (60, 5e3)]:81 per = static_period(p, tech, p.KN0, KQ, p.KE0, p.KH0, al[t], p.Z0, p.eps_path()[t], gf=1.0)82 housing_rent(p, per, 0.6)83 assert abs(income_identity(p, per)) < 1e-6 * per["GDP"]84 va = value_added_by_sector(p, per)85 assert abs(va["total"] - per["GDP"]) < 1e-6 * per["GDP"]868788def test_income_identity_fails_when_accounting_is_broken(calibrated):89 """A deliberately inconsistent accounting (compute income counted twice) must be detected."""90 p, info, tech = calibrated91 per = static_period(p, tech, p.KN0, 1.0, p.KE0, p.KH0, p.alpha_path()[0], p.Z0, p.eps0, gf=1.0)92 housing_rent(p, per, 0.6)93 per_bad = dict(per)94 per_bad["GDP"] = per["GDP"] + per["A"] * per["Qtot"] # counting compute services as final output95 assert abs(income_identity(p, per_bad)) > 1e-3969798def test_calibration_shares(calibrated):99 p, info, tech = calibrated100 per = static_period(p, tech, p.KN0, p.KQ0, p.KE0, p.KH0, p.alpha_path()[0], p.Z0, p.eps0, gf=1.0)101 housing_rent(p, per, info["C"])102 assert per["wage_bill"] / per["GDP"] == pytest.approx(0.568303, abs=2e-3)103 assert per["PE"] * per["E"] / per["GDP"] == pytest.approx(0.04, abs=2e-3)104 assert per["util"] == 1.0105106107def test_energy_bottleneck_lowers_utilization(calibrated):108 p, info, tech = calibrated109 al = p.alpha_path()[60]110 per = static_period(p, tech, p.KN0, 1e7, p.KE0, p.KH0, al, p.Z0, p.eps_path()[60], gf=1.0)111 assert per["util"] < 1.0112 assert abs(per["A"] - p.eps_path()[60] * per["PE"]) < 1e-8 * per["A"] # idle capacity: A = energy cost113