change.py

"""
change.py — 'Did something actually change?' engine for normal people.

Same Clutch + MagnitudeGate as the compute demo, pointed at a human question:
is this series just its usual wobble, or did something really shift, and when?

Loop (identical structure to the drift substrate):
  cheap  = extrapolate the cached linear model of your recent numbers
  costly = refit that model on the last `window` points
  error  = |prediction - today's number| / typical wobble
A gate trip == the model of "normal" broke == a real change. Trips close together
are merged into one EVENT with a plain-language before/after summary.

Honesty rules: warm-up trips are ignored, pure noise must yield "no change",
slow steady trends are reported as trends (they never break a linear model, and
saying otherwise would be lying).
"""

import re
import numpy as np
from clutch import Clutch, MagnitudeGate


# ------------------------------------------------------------------ input
def parse_numbers(text=None, file_obj=None):
    raw = ""
    if file_obj is not None:
        path = file_obj if isinstance(file_obj, str) else getattr(file_obj, "name", None)
        if path:
            with open(path, "r", errors="ignore") as f:
                raw = f.read()
    elif text:
        raw = text
    if not raw.strip():
        return None, "No numbers yet — paste some, or pick an example above."
    rows = []
    for line in raw.strip().splitlines():
        nums = re.findall(r"[-+]?\d*[\.,]?\d+(?:[eE][-+]?\d+)?", line.replace(",", "."))
        if nums:
            rows.append([float(x) for x in nums])
    if not rows:
        return None, "I couldn't find any numbers in that."
    ncol = max(len(r) for r in rows)
    if ncol == 1:
        y = np.array([r[0] for r in rows if len(r) == 1], float)
    else:
        y = np.array([r[-1] for r in rows if len(r) == ncol], float)
    y = y[np.isfinite(y)]
    if len(y) < 14:
        return None, f"Only {len(y)} values — I need at least 14 to tell change from noise."
    return y, None


# ------------------------------------------------------------------ core
def _wobble(y):
    """Typical day-to-day wobble: robust std (MAD) of first differences."""
    d = np.diff(y)
    mad = np.median(np.abs(d - np.median(d)))
    return float(1.4826 * mad + 1e-9)


class _Model:
    def __init__(self, y, window, scale):
        self.y, self.window, self.scale = y, window, scale
        self.t = 0
        self.a, self.b, self.origin = 0.0, float(y[0]), 0
        self.last_resid = 1.0

    def predict(self, t):
        return self.a * (t - self.origin) + self.b

    def cheap(self, _):
        return self.predict(self.t)

    def costly(self, _):
        lo = max(0, self.t - self.window)
        xs = np.arange(lo, self.t + 1)
        ys = self.y[lo:self.t + 1]
        if len(xs) >= 2:
            a, b = np.polyfit(xs - lo, ys, 1)
            self.a, self.b, self.origin = float(a), float(b), lo
        insample = float(np.mean(np.abs(np.polyval([self.a, self.b], xs - lo) - ys))) if len(xs) else 0.0
        return self.predict(self.t), (insample / self.scale) < 1.2

    def err(self, _):
        return self.last_resid


def detect(y, sensitivity=1.0, sigma_mode="iid"):
    """Run the clutch over y. Returns dict with trips, events, checks, window, scale."""
    n = len(y)
    window = int(np.clip(n // 10, 7, 30))
    scale = _wobble(y)                       # day-to-day wobble (for the human text)
    # iid: noise around a trend -> one-step noise is wobble/sqrt(2)
    # walk: random-walk-like (stock prices) -> the daily move IS the innovation
    sigma = scale / np.sqrt(2.0) if sigma_mode == "iid" else scale
    # sensitivity 0.5 (paranoid) .. 2.0 (relaxed): scales the trip threshold
    gate = MagnitudeGate(gain=2.0, leak=1.8, trip=8.0 * sensitivity)
    clutch = Clutch(gate)
    m = _Model(y, window, sigma)
    trips, checks = [], 0
    for t in range(n):
        m.t = t
        before = clutch.stats.expensive_calls
        pred, _mode = clutch.step(None, m.cheap, m.costly, m.err)
        if clutch.stats.expensive_calls > before:
            checks += 1
            if t > window:                       # ignore warm-up
                trips.append(t)
        m.last_resid = abs(pred - y[t]) / sigma

    # merge trips within `window` of each other into events
    events = []
    for t in trips:
        if events and t - events[-1][-1] <= window:
            events[-1].append(t)
        else:
            events.append([t])

    out_events = []
    for grp in events:
        at0 = grp[0]
        last = min(grp[-1], at0 + 3 * window)
        lo = max(0, at0 - 2 * window)
        hi = min(n, last + 1 + window)
        # refine: best single step position within the local window
        best_c, best_sse = None, np.inf
        for c in range(lo + 3, hi - 2):
            l, r = y[lo:c], y[c:hi]
            sse = ((l - l.mean()) ** 2).sum() + ((r - r.mean()) ** 2).sum()
            if sse < best_sse:
                best_sse, best_c = sse, c
        cp = best_c if best_c is not None else at0
        before_mean = float(np.mean(y[lo:cp]))
        after_mean = float(np.mean(y[cp:hi]))
        shift = after_mean - before_mean
        kind = "shift" if abs(shift) >= 2.0 * sigma else "blip"
        out_events.append(dict(at=cp, span=(grp[0], last), before=before_mean,
                               after=after_mean, shift=shift, kind=kind))

    # overall slow trend (fits the whole series; never trips the gate, honestly reported)
    xs = np.arange(n)
    slope = float(np.polyfit(xs, y, 1)[0])
    trend_total = slope * n
    trendy = abs(trend_total) > 3.0 * scale and not any(e["kind"] == "shift" for e in out_events)

    return dict(events=out_events, trips=trips, checks=checks, window=window,
                scale=scale, slope=slope, trend_total=trend_total, trendy=trendy, n=n)


# ------------------------------------------------------------------ language
def verdict_text(y, res, unit="", period="day"):
    u = f" {unit}" if unit else ""
    n, scale = res["n"], res["scale"]
    shifts = [e for e in res["events"] if e["kind"] == "shift"]
    blips = [e for e in res["events"] if e["kind"] == "blip"]
    lines = []

    if not shifts and not res["trendy"]:
        lines.append(f"## 😌 Just noise — nothing actually changed")
        lines.append(f"Across all **{n} {period}s**, your numbers stayed inside their normal "
                     f"wobble of about **±{scale:.2g}{u}** per {period}. "
                     f"Ups and downs smaller than that are not signal — reacting to them is "
                     f"reacting to dice rolls.")
        if blips:
            days = ", ".join(f"{period} {e['at']}" for e in blips)
            lines.append(f"There were brief odd readings around **{days}**, but the numbers "
                         f"came straight back — one-off blips, not a real change.")
    elif res["trendy"]:
        direction = "upward" if res["slope"] > 0 else "downward"
        lines.append(f"## 📈 No sudden change — but a steady {direction} drift")
        lines.append(f"Nothing jumped, but over the whole {n} {period}s your numbers drifted "
                     f"**{res['trend_total']:+.3g}{u}** in total (about {res['slope']:+.3g}{u} "
                     f"per {period}). Day-to-day comparisons will feel like noise (wobble "
                     f"±{scale:.2g}{u}); the drift only shows over weeks. That slow kind of "
                     f"change is exactly what people miss.")
    else:
        lines.append(f"## 🔔 Yes — something really changed")
        for e in shifts:
            direction = "up" if e["shift"] > 0 else "down"
            times = abs(e["shift"]) / scale
            lines.append(f"- Around **{period} {e['at']}**, your typical level moved "
                         f"**{direction} from {e['before']:.3g}{u} to {e['after']:.3g}{u}** "
                         f"({e['shift']:+.3g}{u} — about {times:.0f}× your normal {period}-to-"
                         f"{period} wobble). That is a real shift, not luck.")
        if blips:
            lines.append(f"- ({len(blips)} brief blip(s) also detected that reversed on their "
                         f"own — those you can ignore.)")

    saved = (1 - res["checks"] / n) * 100
    lines.append("")
    lines.append(f"**Your attention, saved:** instead of judging every single {period} "
                 f"({n} looks), checking on the **{res['checks']} {period}s flagged above** "
                 f"would have caught everything that mattered — **{saved:.0f}% fewer looks, "
                 f"zero missed changes** on this data.")
    lines.append("")
    lines.append(f"<small>How it works: a tiny model keeps predicting your next number from "
                 f"the recent trend; only when reality breaks the prediction harder than your "
                 f"normal wobble (±{scale:.2g}{u}) does it flag a change. This is a statistics "
                 f"tool, not medical or financial advice.</small>")
    return "\n".join(lines)


# ------------------------------------------------------------------ examples
def example_series(name, seed=3):
    rng = np.random.default_rng(seed)
    if name.startswith("Weight"):
        n = 90
        y = 84.0 + rng.normal(0, 0.45, n)
        y[52:] -= np.linspace(0, 0.11 * (n - 52), n - 52)   # diet bites ~day 52 (~0.8 kg/wk)
        return np.round(y, 1), "kg", "day"
    if name.startswith("Sleep"):
        n = 60
        y = 7.1 + rng.normal(0, 0.55, n)             # pure noise: nothing changed
        return np.round(y, 1), "h", "night"
    if name.startswith("Electricity"):
        n = 52
        y = 62 + rng.normal(0, 4.5, n)
        y[30:] += 21                                  # heater breaks / tariff jumps week 30
        return np.round(y, 1), "€", "week"
    # "Spending — slow creep"
    n = 80
    y = 31 + np.linspace(0, 13.0, n) + rng.normal(0, 2.2, n)  # lifestyle creep, no jump
    return np.round(y, 2), "€", "day"