Initial commit

README.md

@@ -0,0 +1,82 @@
+# Staticplay CurioDynamics
+
+Curiosity-driven symbolic physics discovery from raw trajectories (position, velocity, time). The agent observes only state and learns a sparse symbolic model that recovers hidden dynamics such as gravity, drag, and wind.
+
+## What It Does
+- Generates a hidden physics world (projectile motion with drag + wind).
+- Observes only `position`, `velocity`, and `time`.
+- Learns a symbolic acceleration model via sparse regression.
+- Recovers interpretable formulas like:
+  - `ax = wind − k * vx * |vx|`
+  - `ay = −g − k * vy * |vy|`
+
+## Quick Start
+Requires Python 3.10+.
+
+```powershell
+python .\physics_world.py
+```
+
+## How It Works
+1. Simulates a hidden world with gravity + drag + wind.
+2. Collects state transitions only (no equations given).
+3. Fits a sparse symbolic model over a small feature library.
+4. Interprets the recovered constants as wind and gravity.
+
+## Latest Test Results
+Hidden world: quadratic drag + wind
+
+```
+g_true = 9.81
+drag_true = 0.12
+wind_true = 0.4
+
+model_ax: ax = 0.4 − 0.12 * vx|vx|
+model_ay: ay = −9.81 − 0.12 * vy|vy|
+wind_est = 0.4
+g_est = 9.81
+```
+
+## Install
+No external dependencies beyond standard library.
+
+If you want a virtual environment:
+```powershell
+python -m venv .venv
+.\.venv\Scripts\activate
+python .\physics_world.py
+```
+
+## Files
+- `physics_world.py`: hidden physics world + curiosity agent + symbolic regression
+
+## How To Run Different Scenarios
+Edit `run_stress_test(...)` in `physics_world.py`:
+- `wind`
+- `drag`
+- `drag_power`
+- `episodes`
+- `steps`
+
+Example:
+```python
+result = run_stress_test(
+    episodes=50,
+    steps=200,
+    g=9.81,
+    drag=0.12,
+    wind=0.4,
+    drag_power=2.0,
+    dt=0.1
+)
+print(result)
+```
+
+## Notes
+This is intentionally minimal and deterministic to highlight discovery mechanics. The feature library is constrained to keep the model interpretable.
+
+## Acknowledgements
+- teliov/symcat-to-synthea
+
+## Links
+- https://staticplay.co.uk

physics_world.py

@@ -0,0 +1,314 @@
+import math
+import random
+from dataclasses import dataclass
+
+
+@dataclass
+class State:
+    x: float
+    y: float
+    vx: float
+    vy: float
+
+
+class ProjectileWorld:
+    def __init__(self, g=9.81, drag=0.12, wind=0.4, dt=0.1, drag_power=2.0):
+        self.g = g
+        self.drag = drag
+        self.wind = wind
+        self.dt = dt
+        self.drag_power = drag_power
+
+    def step(self, s: State) -> State:
+        # Hidden physics: quadratic drag + wind
+        ax = -self.drag * s.vx * abs(s.vx) + self.wind
+        ay = -self.g - self.drag * s.vy * abs(s.vy)
+
+        vx = s.vx + ax * self.dt
+        vy = s.vy + ay * self.dt
+        x = s.x + vx * self.dt
+        y = s.y + vy * self.dt
+        return State(x=x, y=y, vx=vx, vy=vy)
+
+
+class CuriosityAgent:
+    def __init__(self, dt=0.1):
+        self.dt = dt
+        # learned symbolic model coefficients for ax and ay
+        self.model_ax = {}
+        self.model_ay = {}
+        # Invented concepts
+        self.invented = {}
+        self.surprise_window = []
+        self.err_vx_window = []
+        self.window_size = 20
+        self.surprise_threshold = 0.06
+        self.stable_threshold = 0.5
+        self.drag_power_window = []
+        # regression buffers per episode
+        self.samples = []
+        self.feature_means = {"vx_abs_vx": 0.0, "vy_abs_vy": 0.0}
+
+    def predict(self, s: State) -> State:
+        # If no model yet, predict no acceleration
+        if not self.model_ax and not self.model_ay:
+            ax = 0.0
+            ay = 0.0
+        else:
+            ax = self._eval_model(self.model_ax, s)
+            ay = self._eval_model(self.model_ay, s)
+        vx = s.vx + ax * self.dt
+        vy = s.vy + ay * self.dt
+        x = s.x + vx * self.dt
+        y = s.y + vy * self.dt
+        return State(x=x, y=y, vx=vx, vy=vy)
+
+    def update(self, s: State, s_next: State):
+        pred = self.predict(s)
+        # prediction error on velocity (scaled)
+        err_vx = s_next.vx - pred.vx
+        err_vy = s_next.vy - pred.vy
+        surprise = math.sqrt(err_vx * err_vx + err_vy * err_vy)
+
+        # keep surprise window
+        self.surprise_window.append(surprise)
+        if len(self.surprise_window) > self.window_size:
+            self.surprise_window.pop(0)
+        self.err_vx_window.append(err_vx)
+        if len(self.err_vx_window) > self.window_size:
+            self.err_vx_window.pop(0)
+
+        # store samples for regression (ax, ay from finite differences)
+        ax = (s_next.vx - s.vx) / self.dt
+        ay = (s_next.vy - s.vy) / self.dt
+        self.samples.append((s.vx, s.vy, ax, ay))
+
+        self._maybe_invent(surprise)
+
+        return surprise
+
+    def _maybe_invent(self, surprise):
+        if len(self.surprise_window) < self.window_size:
+            return
+        high = sum(1 for s in self.surprise_window if s > self.surprise_threshold)
+        ratio = high / self.window_size
+        if ratio >= self.stable_threshold and "drag" not in self.invented:
+            self.invented["drag"] = {
+                "confidence": round(ratio, 3),
+                "evidence_window": list(self.surprise_window),
+            }
+        # wind discovery: persistent bias in vx error
+        if "model_update" not in self.invented and ratio >= self.stable_threshold:
+            self.invented["model_update"] = {"confidence": round(ratio, 3)}
+
+    def fit_params(self):
+        if len(self.samples) < 20:
+            self.samples.clear()
+            return
+
+        # Symbolic regression via sparse linear model on feature library
+        features_ax_linear = []
+        features_ax_quad = []
+        features_ay_linear = []
+        features_ay_quad = []
+        targets_ax = []
+        targets_ay = []
+        mean_vx_abs_vx = sum((vx * abs(vx)) for vx, _, _, _ in self.samples) / len(self.samples)
+        mean_vy_abs_vy = sum((vy * abs(vy)) for _, vy, _, _ in self.samples) / len(self.samples)
+        self.feature_means["vx_abs_vx"] = mean_vx_abs_vx
+        self.feature_means["vy_abs_vy"] = mean_vy_abs_vy
+
+        for vx, vy, ax, ay in self.samples:
+            features_ax_linear.append({"1": 1.0, "vx": vx})
+            features_ax_quad.append({"1": 1.0, "vx_abs_vx": (vx * abs(vx)) - mean_vx_abs_vx})
+            features_ay_linear.append({"1": 1.0, "vy": vy})
+            features_ay_quad.append({"1": 1.0, "vy_abs_vy": (vy * abs(vy)) - mean_vy_abs_vy})
+            targets_ax.append(ax)
+            targets_ay.append(ay)
+
+        coeff_ax_lin, mse_ax_lin = self._fit_sparse(features_ax_linear, targets_ax, return_mse=True, center=True)
+        coeff_ax_quad, mse_ax_quad = self._fit_sparse(features_ax_quad, targets_ax, return_mse=True, center=True)
+        coeff_ay_lin, mse_ay_lin = self._fit_sparse(features_ay_linear, targets_ay, return_mse=True, center=True)
+        coeff_ay_quad, mse_ay_quad = self._fit_sparse(features_ay_quad, targets_ay, return_mse=True, center=True)
+
+        coeff_ax = coeff_ax_quad if mse_ax_quad < mse_ax_lin else coeff_ax_lin
+        coeff_ay = coeff_ay_quad if mse_ay_quad < mse_ay_lin else coeff_ay_lin
+        self.model_ax = coeff_ax
+        self.model_ay = coeff_ay
+
+        if self.model_ax or self.model_ay:
+            self.invented.setdefault(
+                "symbolic_model",
+                {"terms_ax": list(self.model_ax.keys()), "terms_ay": list(self.model_ay.keys())},
+            )
+
+        self.samples.clear()
+
+    def _fit_sparse(self, feature_rows, targets, return_mse=False, center=False):
+        # Sequential Thresholded Least Squares (SINDy-style)
+        keys = list(feature_rows[0].keys())
+        n = len(feature_rows)
+
+        # Build design matrix
+        X = [[row[k] for k in keys] for row in feature_rows]
+        y = targets[:]
+        y_mean = 0.0
+        if center:
+            y_mean = sum(y) / len(y)
+            y = [v - y_mean for v in y]
+
+        # Normalize columns (except constant)
+        means = [0.0] * len(keys)
+        stds = [1.0] * len(keys)
+        for j, k in enumerate(keys):
+            if k == "1":
+                means[j] = 0.0
+                stds[j] = 1.0
+                continue
+            col = [X[i][j] for i in range(n)]
+            m = sum(col) / n
+            v = sum((c - m) ** 2 for c in col) / n
+            s = math.sqrt(v) if v > 1e-12 else 1.0
+            means[j] = m
+            stds[j] = s
+            for i in range(n):
+                X[i][j] = (X[i][j] - m) / s
+
+        active = set(range(len(keys)))
+        coeff = [0.0] * len(keys)
+
+        def solve_least_squares(active_idx):
+            # Normal equations on active set
+            a_idx = sorted(active_idx)
+            m = len(a_idx)
+            if m == 0:
+                return [0.0] * len(keys)
+            xtx = [[0.0 for _ in range(m)] for _ in range(m)]
+            xty = [0.0 for _ in range(m)]
+            for i in range(n):
+                row = [X[i][j] for j in a_idx]
+                for r in range(m):
+                    xty[r] += row[r] * y[i]
+                    for c in range(m):
+                        xtx[r][c] += row[r] * row[c]
+            # Gauss-Seidel
+            beta = [0.0] * m
+            for _ in range(30):
+                for r in range(m):
+                    denom = xtx[r][r] if abs(xtx[r][r]) > 1e-8 else 1e-8
+                    num = xty[r]
+                    for c in range(m):
+                        if c == r:
+                            continue
+                        num -= xtx[r][c] * beta[c]
+                    beta[r] = num / denom
+            full = [0.0] * len(keys)
+            for r, j in enumerate(a_idx):
+                full[j] = beta[r]
+            return full
+
+        # Iterative thresholding
+        for _ in range(6):
+            coeff = solve_least_squares(active)
+            # Unnormalize coefficients
+            coeff_unnorm = coeff[:]
+            for j, k in enumerate(keys):
+                if k == "1":
+                    continue
+                coeff_unnorm[j] = coeff[j] / stds[j]
+            # Threshold
+            new_active = set(i for i, v in enumerate(coeff_unnorm) if abs(v) >= 0.02)
+            new_active.add(keys.index("1"))
+            if new_active == active:
+                coeff = coeff_unnorm
+                break
+            active = new_active
+            coeff = coeff_unnorm
+
+        pruned = {k: round(v, 3) for k, v in zip(keys, coeff) if abs(v) >= 0.02}
+        if center:
+            pruned["1"] = round(pruned.get("1", 0.0) + y_mean, 3)
+
+        if not return_mse:
+            return pruned
+
+        # compute mse
+        mse = 0.0
+        for row, y in zip(feature_rows, targets):
+            y_hat = sum(pruned.get(k, 0.0) * row[k] for k in row)
+            mse += (y - y_hat) ** 2
+        mse /= len(feature_rows)
+        return pruned, mse
+
+    def _eval_model(self, model, s: State):
+        features = {
+            "1": 1.0,
+            "vx": s.vx,
+            "vy": s.vy,
+            "vx_abs_vx": (s.vx * abs(s.vx)) - self.feature_means["vx_abs_vx"],
+            "vy_abs_vy": (s.vy * abs(s.vy)) - self.feature_means["vy_abs_vy"],
+        }
+        return sum(model.get(k, 0.0) * features[k] for k in model)
+
+
+def run_stress_test(
+    episodes=50,
+    steps=200,
+    g=9.81,
+    drag=0.12,
+    wind=0.4,
+    dt=0.1,
+    drag_power=2.0,
+    seed=123
+):
+    random.seed(seed)
+    world = ProjectileWorld(g=g, drag=drag, wind=wind, dt=dt, drag_power=drag_power)
+    agent = CuriosityAgent(dt=dt)
+
+    surprises = []
+    for _ in range(episodes):
+        # random launch
+        speed = random.uniform(8, 20)
+        angle = random.uniform(20, 70) * math.pi / 180.0
+        s = State(
+            x=0.0,
+            y=0.0,
+            vx=speed * math.cos(angle),
+            vy=speed * math.sin(angle),
+        )
+        for _ in range(steps):
+            s_next = world.step(s)
+            surprise = agent.update(s, s_next)
+            surprises.append(surprise)
+            s = s_next
+            if s.y < 0.0:
+                break
+        agent.fit_params()
+
+    # Interpret constants as wind and gravity when quadratic terms exist
+    wind_est = None
+    g_est = None
+    if "vx_abs_vx" in agent.model_ax and "1" in agent.model_ax:
+        wind_est = agent.model_ax["1"] - agent.model_ax["vx_abs_vx"] * agent.feature_means["vx_abs_vx"]
+    if "vy_abs_vy" in agent.model_ay and "1" in agent.model_ay:
+        g_est = -(agent.model_ay["1"] - agent.model_ay["vy_abs_vy"] * agent.feature_means["vy_abs_vy"])
+
+    return {
+        "g_true": g,
+        "drag_true": drag,
+        "wind_true": wind,
+        "drag_power_true": drag_power,
+        "model_ax": agent.model_ax,
+        "model_ay": agent.model_ay,
+        "wind_est": round(wind_est, 3) if wind_est is not None else None,
+        "g_est": round(g_est, 3) if g_est is not None else None,
+        "invented": agent.invented,
+        "avg_surprise": round(sum(surprises) / len(surprises), 3),
+        "max_surprise": round(max(surprises), 3),
+        "samples": len(surprises),
+    }
+
+
+if __name__ == "__main__":
+    result = run_stress_test()
+    print(result)