1import json
  2import random
  3from pathlib import Path
  4
  5import numpy as np
  6from sklearn.metrics import mutual_info_score
  7
  8
  9def set_seed(seed: int) -> None:
 10    random.seed(seed)
 11    np.random.seed(seed)
 12
 13
 14def entropy(x: np.ndarray) -> float:
 15    _, counts = np.unique(x, return_counts=True)
 16    p = counts.astype(float) / len(x)
 17    return float(-(p * np.log(p + 1e-15)).sum())
 18
 19
 20def mi(x: np.ndarray, y: np.ndarray) -> float:
 21    return float(mutual_info_score(x, y))
 22
 23
 24def quantize(x: np.ndarray, delta: float) -> np.ndarray:
 25    return np.floor(x / delta + 0.5).astype(np.int64)
 26
 27
 28def sample(seed: int, n: int = 120_000, noise: float = 0.35):
 29    set_seed(seed)
 30    theta = np.random.normal(size=n)
 31    g = theta + np.random.normal(scale=noise, size=n)
 32    # A fixed-magnitude update is useful on the quadratic iff it moves toward 0.
 33    useful_action = (theta * g > 0).astype(np.int64)
 34    return theta, g, useful_action
 35
 36
 37def channel_sweep(seed=7):
 38    theta, g, utility = sample(seed)
 39    deltas = [2.0, 1.0, 0.5, 0.25, 0.125, 0.0625, 0.03125]
 40    out = []
 41    for delta in deltas:
 42        z = quantize(g, delta)
 43        acquired = entropy(z)
 44        useful = mi(z, utility)
 45        unavailable = acquired - useful
 46        # Controller: sign observation, with zero-bin rejection.
 47        eta = 0.5
 48        update = -eta * np.sign(z)
 49        theta_new = theta + update
 50        progress = float(np.mean(0.5 * theta**2 - 0.5 * theta_new**2))
 51        action_correct = float(np.mean((z != 0) & (np.sign(z) == np.sign(theta))))
 52        out.append({"delta": delta, "I_acq": acquired, "I_use": useful,
 53                    "I_un": unavailable, "progress": progress,
 54                    "action_correct": action_correct,
 55                    "zero_fraction": float(np.mean(z == 0))})
 56    return out
 57
 58
 59def learning_sweep(seed=7):
 60    # Compare raw high-resolution SGD with the actionable finite-resolution
 61    # controller on L(theta)=theta^2/2, averaged over fixed initial points.
 62    set_seed(seed)
 63    initials = np.random.normal(size=2000)
 64    rows = []
 65    for delta in [None, 2.0, 1.0, 0.5, 0.25, 0.125]:
 66        theta = initials.copy()
 67        for _ in range(25):
 68            g = theta + np.random.normal(scale=0.35, size=len(theta))
 69            if delta is None:
 70                z = g
 71            else:
 72                z = quantize(g, delta) * delta
 73            theta -= 0.35 * np.sign(z) * np.minimum(np.abs(z), 2.0)
 74        rows.append({"method": "raw" if delta is None else f"quantized_{delta}",
 75                     "final_loss": float(np.mean(0.5 * theta**2))})
 76    return rows
 77
 78
 79def predictions(rows):
 80    # Prediction 1: acquired information grows approximately linearly in
 81    # -log(delta), with slope equal to the effective one-dimensional channel.
 82    x = np.array([-np.log(r["delta"]) for r in rows[-5:]])
 83    y = np.array([r["I_acq"] for r in rows[-5:]])
 84    slope, intercept = np.polyfit(x, y, 1)
 85    r2 = float(np.corrcoef(x, y)[0, 1] ** 2)
 86    # Prediction 2: useful information saturates while acquired information
 87    # keeps rising; compare last two increments.
 88    du_acq = rows[-1]["I_acq"] - rows[-2]["I_acq"]
 89    du_use = rows[-1]["I_use"] - rows[-2]["I_use"]
 90    # Prediction 3: progress saturates at fine resolution.
 91    dp = rows[-1]["progress"] - rows[-2]["progress"]
 92    return {"acq_log_slope": float(slope), "acq_log_r2": r2,
 93            "fine_delta_I_acq": float(du_acq),
 94            "fine_delta_I_use": float(du_use),
 95            "fine_delta_progress": float(dp),
 96            "predicted_acq_slope": 1.0,
 97            "predicted_useful_trend": "plateau relative to I_acq",
 98            "predicted_progress_trend": "plateau relative to I_acq"}
 99
100
101def main():
102    channel = channel_sweep()
103    result = {"channel": channel, "learning": learning_sweep(),
104              "predictions": predictions(channel)}
105    Path("toy_results.json").write_text(json.dumps(result, indent=2))
106    print(json.dumps(result, indent=2))
107
108
109if __name__ == "__main__":
110    main()