assembly-economics/simulations/gadm02/gadm03.py

"""GADM-0.3 (power channels): domination P split into six channel sub-stocks.

Implements models/GADM-0.3.md § 2 (ASSEMBLY-WP-0005-T03). Everything except P
is GADM-0.2 unchanged (gadm02.py is reused). P becomes an auxiliary:

    P = 1 - prod_k (1 - P_k) ** (1/6)     (equal-weight geometric aggregation)

The plain product 1 - prod(1 - P_k) was tried first and rejected: with six
channels it compounds moderate values (six channels at 0.5 give P = 0.98) and
drove P to 1 in almost every scenario. The weighted form keeps the property
that one fully dominant channel yields P = 1, without compounding.

Each channel follows
    dP_k = (d_k + r_k * P_k**2) * (1 - P_k) - c_k * C_k * P_k
with driver d_k, superlinear self-reinforcement r_k * P_k**2 (increasing
returns to control, which allows threshold behaviour, hypothesis P1) and a
counter C_k scaled by c_k. The c_k are calibrated so that every channel starts
in equilibrium in the baseline (see calibrate()).

Wealth separated from power (ASSEMBLY-WP-0005-T04): a wealth-concentration
stock W grows with privately captured returns and shrinks through decay and a
wealth tax. It converts into the market and political channels at rate
kappa = k_w * (1 - Q*S): checks reduce conversion. Membrane and lineage control
do not depend on wealth. Set wealth_on = 0 to reproduce the T03 model.
"""

from __future__ import annotations

import math

import gadm02 as g2

MODEL_VERSION = "GADM-0.3-power"
TIME_UNIT = g2.TIME_UNIT

CHANNELS = ("Pm", "Pi", "Pd", "Pmem", "Plin", "Ppol")
CHANNEL_NAMES = {
    "Pm": "market", "Pi": "infrastructure", "Pd": "discovery/attention",
    "Pmem": "membrane", "Plin": "lineage", "Ppol": "political/institutional",
}
STOCKS = tuple(k for k in g2.STOCKS if k != "P") + CHANNELS + ("W",)

INIT = {k: v for k, v in g2.INIT.items() if k != "P"}
INIT.update({"Pm": 0.50, "Pi": 0.45, "Pd": 0.50, "Pmem": 0.05, "Plin": 0.05, "Ppol": 0.50})
INIT["W"] = 0.50  # wealth concentration

PARAMS = dict(g2.PARAMS)
PARAMS.update({
    # drivers
    "pi": 0.06,      # infrastructure control grows with capacity K where standards are closed
    "ppol": 0.10,    # economic control converts into political influence
    # self-reinforcement (increasing returns to control)
    "r_m": 0.30, "r_i": 0.30, "r_d": 0.40, "r_mem": 0.30, "r_lin": 0.30, "r_pol": 0.30,
    "lw": 2.0,       # learning-with-use amplification of lineage self-reinforcement (R7)
    # wealth separated from power (T04)
    "wealth_on": 1.0,  # 0 reproduces the T03 power-channel model
    "a_w": 0.10,     # accumulation of privately captured returns
    "d_w": 0.03,     # dissipation (inheritance splitting, failure, competition)
    "t_w": 0.10,     # wealth-tax effectiveness
    "wtax": 0.20,    # wealth-tax lever (0..1)
    "k_w": 0.04,     # wealth → market/political control conversion
    # attention extraction recalibrated against time-use evidence (T05):
    # GADM-0.2's 0.12 let baseline attention headroom rise to 0.65; 0.275 keeps
    # it near 0.45 (engagement media already take ~half of leisure time).
    "aext": 0.275,
    # counter coefficients: calibrated, see calibrate()
    "c_m": 0.0794, "c_i": 0.3117, "c_d": 0.2815, "c_mem": 0.1354, "c_lin": 0.1372, "c_pol": 0.4011,
})


def total_p(s: dict) -> float:
    return 1 - math.prod((1 - s[k]) ** (1 / len(CHANNELS)) for k in CHANNELS)


def _channel_terms(s: dict, p: dict, a: dict) -> dict:
    """Return {channel: (driver, r_eff, counter)}."""
    Ko, Q, S, X, N = s["Ko"], s["Q"], s["S"], s["X"], s["N"]
    B, vsel, K = a["B"], a["Vsel"], a["K"]
    conv = p["wealth_on"] * p["k_w"] * (1 - Q * S) * s["W"]  # wealth bought into control
    econ_ch = ("Pm", "Pi", "Pd", "Pmem", "Plin")
    econ = 1 - math.prod((1 - s[k]) ** (1 / len(econ_ch)) for k in econ_ch)
    return {
        "Pm": (p["p1"] * vsel * (1 - B) + conv, p["r_m"], B + N),
        "Pi": (p["pi"] * K * (1 - p["std"]), p["r_i"], X * p["std"] + 0.1),
        "Pd": (p["p2"] * p["disc"] * (0.5 + 0.5 * Ko), p["r_d"], (1 - p["disc"]) * Q + 0.1),
        "Pmem": (p["p3"] * Ko * (1 - p["m"]), p["r_mem"], p["m"] + 0.1),
        "Plin": (p["p4"] * Ko * (1 - p["L"]), p["r_lin"] * (1 + p["lw"] * Ko * Q), p["L"] + 0.1),
        "Ppol": (p["ppol"] * econ * (1 - B) + conv, p["r_pol"], Q * S),
    }


COEF = {"Pm": "c_m", "Pi": "c_i", "Pd": "c_d", "Pmem": "c_mem", "Plin": "c_lin", "Ppol": "c_pol"}


def derivatives(s: dict, p: dict) -> tuple[dict, dict]:
    s2 = {k: v for k, v in s.items() if k not in CHANNELS}
    s2["P"] = total_p(s)
    d, a = g2.derivatives(s2, p)
    d.pop("P")
    for k, (drv, r, ctr) in _channel_terms(s, p, a).items():
        d[k] = (drv + r * s[k] ** 2) * (1 - s[k]) - p[COEF[k]] * ctr * s[k]
    W = s["W"]
    d["W"] = p["a_w"] * (1 - a["B"]) * a["Ystar"] * (1 - W) - (p["d_w"] + p["t_w"] * p["wtax"]) * W
    return d, a


def channel_roots(s: dict, p: dict, k: str, n: int = 10000) -> list[float]:
    """Equilibria of channel k with every other stock frozen at state s."""
    s2 = {x: v for x, v in s.items() if x not in CHANNELS}
    s2["P"] = total_p(s)
    drv, r, ctr = _channel_terms(s, p, g2.auxiliaries(s2, p))[k]
    c = p[COEF[k]]
    f = lambda x: (drv + r * x * x) * (1 - x) - c * ctr * x  # noqa: E731
    xs = [i / n for i in range(1, n)]
    return sorted({round(b, 3) for a, b in zip(xs, xs[1:]) if f(a) * f(b) <= 0})


def channel_slope(s: dict, p: dict, k: str, h: float = 1e-5) -> float:
    """d(dP_k)/dP_k at state s; negative means the channel is locally stable."""
    up, dn = dict(s), dict(s)
    up[k] += h
    dn[k] -= h
    return (derivatives(up, p)[0][k] - derivatives(dn, p)[0][k]) / (2 * h)


def calibrate(params: dict | None = None, init: dict | None = None) -> dict:
    """Counter coefficients that put every channel in equilibrium at t = 0."""
    p = {**PARAMS, **(params or {})}
    s = {**INIT, **(init or {})}
    s2 = {k: v for k, v in s.items() if k not in CHANNELS}
    s2["P"] = total_p(s)
    a = g2.auxiliaries(s2, p)
    out = {}
    for k, (drv, r, ctr) in _channel_terms(s, p, a).items():
        out[COEF[k]] = round((drv + r * s[k] ** 2) * (1 - s[k]) / (ctr * s[k]), 4)
    return out


def t03_params() -> dict:
    """Parameter overrides that reproduce the T03 power-channel model (no wealth)."""
    return {"wealth_on": 0.0, "aext": g2.PARAMS["aext"], **calibrate({"wealth_on": 0.0})}


def t04_params() -> dict:
    """Parameter overrides that reproduce the T04 stage (wealth, old attention)."""
    return {"aext": g2.PARAMS["aext"]}


def simulate(params: dict | None = None, init: dict | None = None,
             T: float = 100.0, dt: float = 0.1, record_every: int = 10) -> list[dict]:
    p = {**PARAMS, **(params or {})}
    s = {**INIT, **(init or {})}
    unknown = set(p) - set(PARAMS)
    if unknown:
        raise KeyError(f"unknown parameters: {sorted(unknown)}")
    rows = []
    steps = int(round(T / dt))
    for i in range(steps + 1):
        d, a = derivatives(s, p)
        if i % record_every == 0:
            row = {"t": round(i * dt, 6), **s, "P": total_p(s),
                   **{k: a[k] for k in ("Ystar", "B", "V", "Vsel", "sigma")}}
            vs = {**row}
            row["VFAC"] = g2.vfac(vs)
            row["VFAC01"] = g2.vfac(vs, with_attention=False)
            rows.append(row)
        if i < steps:
            s = {k: min(1.0, max(0.0, s[k] + dt * d[k])) for k in STOCKS}
    return rows