Something went wrong. Try again.
Deep Learning Tripping Balls
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392#!/usr/bin/env python3# Permission to use, copy, modify, and/or distribute this software for# any purpose with or without fee is hereby granted.## THE SOFTWARE IS PROVIDED “AS IS” AND THE AUTHOR DISCLAIMS ALL# WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES# OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE# FOR ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY# DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN# AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING# OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
"""Reconstructed stress tests for dltb.detect_stabilization.
The design notes the tool was ported from described a synthetic stressbattery whose exact series/values were never saved, so these testsrebuild it from the explanations: shapes that must be located correctly(rise-plateau, spike-then-fall, two-step plateau, noisy transient), shapesthat must be flagged (monotone rise, drift running into the end of therun, slow in-band creep, sine/alternating limit cycles, random walks,too-short runs), and false-alarm checks (stationarycorrelated noise). Plus unit cases for the pandas-compatible rollingmedian, the label-aligned CSV reader, and the documented results on thereal feedback run (untracked/output_sd_turbo/feedback_s0.4_a0.8/distance_metrics.csv, skipped when absent -- untracked/ is user data).
Plain Python, no pytest, CPU-only. Run from anywhere:
uv run python tests/test_detect_stabilization.py
Exit code 0 = all green, 1 = at least one failure. All series are seeded,so a failure is reproducible.
Two findings from the reconstruction, recorded here as regression guards(the design notes' original series were never saved, so their exactcounts are not reproducible): - stationary AR(1) noise is false-alarm-free up to rho=0.85, but at rho=0.9 roughly a quarter of runs cross the 'wandering' boundary (lag-1 autocorr ~0.9 + variance ratio ~0.71 sit right on the thresholds); the rho=0.9 case below pins the measured rate. - 'residual_drift' never fires for monotone drift shapes (the tail MAD inflates with the drift and always outpaces the slope test); it is reachable only for drift mixed with oscillation, where Theil-Sen's robustness to oscillation lets the slope test see what the MAD band absorbs -- and there it always co-fires with level_shift/wandering/ periodic_oscillation, never alone. Drift that runs into the end of the run is caught by 'plateau_too_short' instead."""
from __future__ import annotations
import mathimport sysimport tempfilefrom pathlib import Path
import numpy as np
try: from dltb.detect_stabilization import (_rolling_median, analyze_csv, detect_stabilization, read_metrics_csv)except ImportError: # running without the project installed sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) from dltb.detect_stabilization import (_rolling_median, analyze_csv, detect_stabilization, read_metrics_csv)
REPO = Path(__file__).resolve().parents[1]REAL_CSV = (REPO / "untracked/output_sd_turbo/feedback_s0.4_a0.8" / "distance_metrics.csv")# Documented verdicts for the real file (design notes, "Results" table).REAL_EXPECTED = { "dreamsim_to_ref": ("frame_0006", 0.09246), "dreamsim_to_prev": ("frame_0013", 7.8e-05), "dreamsim_input_to_ref": ("frame_0005", 0.01057),}
TESTS: list = []
def test(fn): TESTS.append(fn) return fn
# ---------------------------------------------------------------------------# unit cases# ---------------------------------------------------------------------------
@testdef rolling_median_convention(): """Hand-computed centered medians, edges included (pandas convention).""" cases = [ ([1.0, 9.0, 2.0, 8.0, 3.0], 3, [5.0, 2.0, 8.0, 3.0, 5.5]), # even window: one extra element on the left (matches pandas 2.x) ([0.0, 1.0, 2.0, 3.0, 4.0], 4, [0.5, 1.0, 1.5, 2.5, 3.0]), ([5.0, 1.0, 3.0], 1, [5.0, 1.0, 3.0]), ] for y, w, want in cases: got = _rolling_median(np.asarray(y), w).tolist() if not np.allclose(got, want): return False, f"w={w}: {got} != {want}" # a single-sample spike is annihilated by a w=3 median (why the # backward scan ignores isolated spikes) y = np.concatenate([np.zeros(20), [100.0], np.zeros(20)]) if not np.allclose(_rolling_median(y, 3), 0.0): return False, "w=3 median did not remove a single-sample spike" return True, "odd/even windows, edges, spike kill all match"
@testdef reader_label_alignment(): """Non-numeric cells drop per column, labels stay aligned; missing file fails fast.""" with tempfile.TemporaryDirectory() as tmp: p = Path(tmp) / "m.csv" p.write_text("frame,a,b\n" "f0,1,10\n" "f1,n/a,11\n" "f2,3,12\n" "f3,4\n") # short row: b missing _, cols, data = read_metrics_csv(p) if cols != ["a", "b"]: return False, f"columns {cols}" la, va = data["a"] lb, vb = data["b"] if (la, va, lb, vb) != (["f0", "f2", "f3"], [1.0, 3.0, 4.0], ["f0", "f1", "f2"], [10.0, 11.0, 12.0]): return False, f"parsed {la}{va} {lb}{vb}" try: read_metrics_csv(Path(tmp) / "nope.csv") except SystemExit: return True, "drop + alignment + short row + missing-file exit OK" return False, "missing file did not SystemExit"
# ---------------------------------------------------------------------------# shapes that must be LOCATED correctly (stabilized, right t_stab/value)# ---------------------------------------------------------------------------
@testdef rise_then_plateau(): rng = np.random.default_rng(1) t = np.arange(600) y = 1.0 - np.exp(-t / 40.0) + rng.normal(0, 0.01, t.size) r = detect_stabilization(y) problems = [] if r["status"] != "stabilized": problems.append(f"status={r['status']} flags={r['flags']}") if not 50 < r["t_stab"] < 250: # 40*ln(20) ~ 120 problems.append(f"t_stab={r['t_stab']}") if abs(r["value"] - 1.0) > 0.02: problems.append(f"value={r['value']}") return not problems, f"t_stab={r['t_stab']} value={r['value']:.4f}"
@testdef spike_then_fall(): rng = np.random.default_rng(2) y = 0.5 + rng.normal(0, 0.02, 300) y[50:52] += 3.0 # 2-sample spike survives smoothing r = detect_stabilization(y) problems = [] if r["status"] != "stabilized": problems.append(f"status={r['status']} flags={r['flags']}") if not 40 <= r["t_stab"] <= 70: # just after the spike problems.append(f"t_stab={r['t_stab']}") if abs(r["value"] - 0.5) > 0.05: problems.append(f"value={r['value']}") return not problems, f"t_stab={r['t_stab']} value={r['value']:.4f}"
@testdef two_step_plateau(): rng = np.random.default_rng(3) y = np.where(np.arange(500) < 200, 0.0, 1.0) + rng.normal(0, 0.02, 500) r = detect_stabilization(y) problems = [] if r["status"] != "stabilized": problems.append(f"status={r['status']} flags={r['flags']}") if not 195 <= r["t_stab"] <= 215: # right after the jump problems.append(f"t_stab={r['t_stab']}") if abs(r["value"] - 1.0) > 0.02: # the SECOND plateau's level problems.append(f"value={r['value']}") return not problems, f"t_stab={r['t_stab']} value={r['value']:.4f}"
@testdef noisy_linear_transient(): rng = np.random.default_rng(4) y = np.clip(np.arange(400) / 100.0, 0.0, 1.0) + rng.normal(0, 0.03, 400) r = detect_stabilization(y) problems = [] if r["status"] != "stabilized": problems.append(f"status={r['status']} flags={r['flags']}") if not 60 < r["t_stab"] < 160: # ramp ends at t=100 problems.append(f"t_stab={r['t_stab']}") if abs(r["value"] - 1.0) > 0.03: problems.append(f"value={r['value']}") return not problems, f"t_stab={r['t_stab']} value={r['value']:.4f}"
@testdef constant_series(): r = detect_stabilization(np.full(100, 3.14)) problems = [] if r["status"] != "stabilized" or "constant_series" not in r["flags"]: problems.append(f"status={r['status']} flags={r['flags']}") if r["t_stab"] != 0 or abs(r["value"] - 3.14) > 1e-12 or r["stderr"] != 0.0: problems.append(f"t_stab={r['t_stab']} value={r['value']} " f"stderr={r['stderr']}") return not problems, "constant detected, flagged informational"
# ---------------------------------------------------------------------------# shapes that must be FLAGGED (needs_attention, right reason)# ---------------------------------------------------------------------------
@testdef monotone_rise_never_settles(): rng = np.random.default_rng(5) y = 0.01 * np.arange(400) + rng.normal(0, 0.001, 400) r = detect_stabilization(y) ok = (r["status"] == "needs_attention" and set(r["flags"]) - {"flat_from_start"}) return bool(ok), f"status={r['status']} flags={r['flags']}"
@testdef drift_into_the_end_flagged(): """Still drifting when the run ends: the tail band cannot certify a plateau, so the verdict must be needs_attention via plateau_too_short ("run ended before settling"), not a fabricated stabilization.""" rng = np.random.default_rng(6) y = rng.normal(0, 0.05, 400) t = np.arange(400) y[t >= 350] += 0.04 * (t[t >= 350] - 350) # ramp +2.0 over last 50 r = detect_stabilization(y) ok = (r["status"] == "needs_attention" and "plateau_too_short" in r["flags"]) return bool(ok), f"status={r['status']} flags={r['flags']}"
@testdef slow_creep_with_oscillation(): """Slow in-band creep under an oscillation: the only shape family where 'residual_drift' is reachable (see module docstring) -- and it co-fires with the other diagnostics. Fires for 20/20 noise seeds around these parameters, but the region is narrow: the same creep without the oscillation is caught by 'wandering' instead.""" t = np.arange(400, dtype=float) rng = np.random.default_rng(0) y = (0.0018 * t + 0.05 * np.sin(2 * np.pi * t / 115.6) + rng.normal(0, 0.001, 400)) r = detect_stabilization(y) ok = (r["status"] == "needs_attention" and "residual_drift" in r["flags"]) return bool(ok), f"status={r['status']} flags={r['flags']}"
@testdef sine_limit_cycle(): t = np.arange(420) y = 1.0 + 0.5 * np.sin(2 * np.pi * t / 60.0) r = detect_stabilization(y) ok = (r["status"] == "needs_attention" and "periodic_oscillation" in r["flags"]) return bool(ok), f"status={r['status']} flags={r['flags']}"
@testdef alternating_series(): t = np.arange(200) y = 1.0 + 0.3 * (-1.0) ** t r = detect_stabilization(y) ok = (r["status"] == "needs_attention" and "alternating_oscillation" in r["flags"]) return bool(ok), f"status={r['status']} flags={r['flags']}"
@testdef too_short_run(): rng = np.random.default_rng(7) r = detect_stabilization(rng.normal(0, 1, 15)) ok = (r["status"] == "needs_attention" and r["flags"] == ["too_few_samples"] and r["t_stab"] is None) return bool(ok), f"flags={r['flags']} t_stab={r['t_stab']}"
@testdef random_walks_are_flagged(): flagged = 0 for seed in range(100, 150): rng = np.random.default_rng(seed) y = np.cumsum(rng.normal(0, 0.01, 400)) if detect_stabilization(y)["status"] == "needs_attention": flagged += 1 # design notes: 47/50 caught; short walk segments can be genuinely # indistinguishable from stationarity -- allow a small miss rate ok = flagged >= 44 return ok, f"{flagged}/50 flagged (needs >= 44)"
@testdef stationary_ar1_no_false_alarm(): for rho in (0.8, 0.85): false = 0 for seed in range(200, 250): rng = np.random.default_rng(seed) eps = rng.normal(0, 1, 900) # 500 burn-in + 400 kept x = np.zeros(900) for i in range(1, 900): x[i] = rho * x[i - 1] + eps[i] if detect_stabilization(x[500:])["status"] != "stabilized": false += 1 if false: return False, f"rho={rho}: {false}/50 false alarms" return True, "0/50 false alarms at rho=0.8 and rho=0.85"
@testdef stationary_ar09_wandering_boundary(): """rho=0.9 stationary AR(1) sits on the wandering boundary (lag-1 autocorr ~0.9, variance ratio ~0.71 vs thresholds 0.90/0.75): pinned at the measured 12/50 rejections. If this number moves a lot, either the thresholds or the diagnostics changed -- revisit deliberately.""" false = 0 for seed in range(200, 250): rng = np.random.default_rng(seed) eps = rng.normal(0, 1, 900) x = np.zeros(900) for i in range(1, 900): x[i] = 0.9 * x[i - 1] + eps[i] if detect_stabilization(x[500:])["status"] != "stabilized": false += 1 return false <= 15, f"{false}/50 rejected at rho=0.9 (pinned ~12, cap 15)"
# ---------------------------------------------------------------------------# real-file regression# ---------------------------------------------------------------------------
@testdef real_file_regression(): if not REAL_CSV.is_file(): return True, "SKIP" rows = {r["column"]: r for r in analyze_csv(REAL_CSV)} problems = [] if set(rows) != set(REAL_EXPECTED): problems.append(f"columns {sorted(rows)}") for col, (label, value) in REAL_EXPECTED.items(): r = rows.get(col) if r is None: continue if r["status"] != "stabilized": problems.append(f"{col}: {r['status']} ({r['flags']})") if r["t_stab_label"] != label: problems.append(f"{col}: t_stab_label={r['t_stab_label']} " f"want {label}") if not math.isclose(r["stabilized_value"], value, rel_tol=0.01): problems.append(f"{col}: value={r['stabilized_value']} " f"want ~{value}") detail = "; ".join(problems) or "3 columns match the documented table" return not problems, detail
# ---------------------------------------------------------------------------
def main() -> int: passed = failed = skipped = 0 for fn in TESTS: try: ok, detail = fn() except Exception as exc: # report, keep running ok, detail = False, f"uncaught {type(exc).__name__}: {exc}" if detail == "SKIP": skipped += 1 print(f"SKIP {fn.__name__}") elif ok: passed += 1 print(f"PASS {fn.__name__} [{detail}]") else: failed += 1 print(f"FAIL {fn.__name__} [{detail}]") print(f"\n{passed} passed, {failed} failed, {skipped} skipped") return 1 if failed else 0
if __name__ == "__main__": sys.exit(main())