1import itertools, json, random
  2from pathlib import Path
  3import numpy as np
  4import torch
  5import torch.nn as nn
  6import torch.nn.functional as F
  7
  8SEED = 2028
  9np.random.seed(SEED)
 10random.seed(SEED)
 11torch.manual_seed(SEED)
 12
 13
 14def subsets(n, B):
 15    return [s for k in range(B + 1) for s in itertools.combinations(range(n), k)]
 16
 17
 18def exact_regret(c, pred, d, subset):
 19    x = np.zeros(len(c), dtype=float)
 20    x[list(subset)] = 1.0
 21    ct, cp = c + d * x, pred + d * x
 22    yt, yp = int(np.argmin(ct)), int(np.argmin(cp))
 23    return float(ct[yp] - ct[yt]), int(yp != yt)
 24
 25
 26def mechanism_check():
 27    # True costs are c=[2,1], prediction is [3,.5]. Both choose edge 1 nominally.
 28    # Interdicting edge 1 by delay z: true switch is z=1; predicted switch z=2.5.
 29    c = np.array([2.0, 1.0])
 30    pred = np.array([3.0, 0.5])
 31    zs = np.linspace(0, 3, 61)
 32    observed = []
 33    for z in zs:
 34        r, flip = exact_regret(c, pred, np.array([0., z]), (1,))
 35        observed.append((float(z), r, flip))
 36    positive = [z for z, r, _ in observed if r > 1e-9]
 37    # In the open interval (1, 2.5), regret is z-1, then vanishes.
 38    onset = min(positive)
 39    end = max(z for z, r, _ in observed if r > 1e-9)
 40    linear = [(z, r) for z, r, _ in observed if 1.05 <= z <= 2.45]
 41    slope = np.polyfit([z for z, _ in linear], [r for _, r in linear], 1)[0]
 42    max_abs_linear_error = max(abs(r - (z - 1.0)) for z, r in linear)
 43
 44    # Scaling prediction: multiplying all delays and costs by s scales regret by s,
 45    # while the switching thresholds scale by s.
 46    scale_rows = []
 47    for s in [0.5, 1., 2., 3.]:
 48        # use a fixed relative interdiction z=1.5*s, inside the disagreement band
 49        r, _ = exact_regret(s*c, s*pred, np.array([0., 1.5*s]), (1,))
 50        scale_rows.append((s, r, 0.5*s))
 51    scale_error = max(abs(r - expected) for s, r, expected in scale_rows)
 52
 53    # Zero-delay prediction: no adversarial effect at d=0.
 54    zero_regret = max(exact_regret(c, pred, np.zeros(2), x)[0] for x in subsets(2, 1))
 55    return {
 56        "predictions": {
 57            "onset_boundary": "regret starts when delay exceeds true margin 1.0",
 58            "linear_scaling": "regret = delay - 1 in the disagreement interval (slope 1)",
 59            "scale_law": "scaling costs and delays by s scales regret by s and thresholds by s",
 60            "zero_perturbation": "d=0 has zero extra regret for a nominally equivalent predictor"
 61        },
 62        "observed": {
 63            "onset_grid": onset,
 64            "last_positive_grid": end,
 65            "fitted_regret_vs_delay_slope": float(slope),
 66            "max_linear_absolute_error": float(max_abs_linear_error),
 67            "scale_rows_s_regret_expected": scale_rows,
 68            "max_scale_absolute_error": float(scale_error),
 69            "zero_delay_worst_regret": float(zero_regret)
 70        },
 71        "theory": {"onset": 1.0, "end": 2.5, "slope": 1.0}
 72    }
 73
 74
 75class Predictor(nn.Module):
 76    def __init__(self):
 77        super().__init__()
 78        self.net = nn.Sequential(nn.Linear(1, 12), nn.Tanh(), nn.Linear(12, 2))
 79    def forward(self, w):
 80        return F.softplus(self.net(w))
 81
 82
 83def train(kind, steps=900, B=1):
 84    torch.manual_seed(SEED + {"mse": 11, "nominal": 12, "adversarial": 13}[kind])
 85    model = Predictor()
 86    opt = torch.optim.Adam(model.parameters(), lr=0.025)
 87    # Training distribution includes varying margins and delays are known scenarios.
 88    w = torch.linspace(0.35, 2.0, 96).reshape(-1, 1)
 89    c = torch.cat([2*w, w], dim=1)
 90    d = torch.cat([torch.zeros_like(w), 1.5*w], dim=1)
 91    scen = subsets(2, B)
 92    for step in range(steps):
 93        pred = model(w)
 94        if kind == "mse":
 95            loss = ((pred-c)**2).mean()
 96        else:
 97            losses = []
 98            # Differentiable decision loss: CE on soft shortest-path probabilities.
 99            for sub in scen:
100                x = torch.zeros_like(c)
101                if sub:
102                    x[:, list(sub)] = 1.
103                ct, cp = c+d*x, pred+d*x
104                true_choice = torch.argmin(ct, dim=1)
105                losses.append(F.cross_entropy(-cp / 0.12, true_choice, reduction="mean"))
106            nominal = losses[0]
107            if kind == "nominal":
108                loss = nominal + 0.002*((pred-c)**2).mean()
109            else:
110                loss = torch.stack(losses).max() + 0.002*((pred-c)**2).mean()
111        opt.zero_grad(); loss.backward(); opt.step()
112    return model
113
114
115def evaluate(model, B=1):
116    w = np.linspace(.35, 2.0, 120)
117    c = np.stack([2*w, w], axis=1)
118    # Delays cover weak to strong perturbations, and are proportional to w.
119    d = np.stack([np.zeros_like(w), 1.5*w], axis=1)
120    with torch.no_grad():
121        pred = model(torch.tensor(w[:, None], dtype=torch.float32)).numpy()
122    nominal = []; worst = []; flips = []; mse = []
123    for i in range(len(w)):
124        nominal.append(exact_regret(c[i], pred[i], d[i], ())[0])
125        vals = [exact_regret(c[i], pred[i], d[i], s) for s in subsets(2, B)]
126        worst.append(max(v[0] for v in vals)); flips.append(max(v[1] for v in vals)); mse.append(np.mean((pred[i]-c[i])**2))
127    return {"nominal_regret_mean": float(np.mean(nominal)), "worst_regret_mean": float(np.mean(worst)), "worst_regret_max": float(np.max(worst)), "adversarial_flip_rate": float(np.mean(flips)), "cost_mse": float(np.mean(mse))}
128
129
130def main():
131    checks = mechanism_check()
132    results = {}
133    for kind in ["mse", "nominal", "adversarial"]:
134        model = train(kind)
135        results[kind] = evaluate(model)
136    out = {"seed": SEED, "checks": checks, "results": results, "note": "Exact hard paths are used for evaluation; training uses differentiable cross-entropy over the exact two-route oracle."}
137    Path("results.json").write_text(json.dumps(out, indent=2))
138    print(json.dumps(out, indent=2))
139
140if __name__ == "__main__":
141    main()