146 lines
5.7 KiB
Python
146 lines
5.7 KiB
Python
import json
|
|
import math
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from fxhnt.domain.edge_decay import PageHinkleyDecay, WindowedEdgeHealth
|
|
|
|
|
|
def test_stationary_positive_never_kills():
|
|
# No false-positive on a clean positive-drift signal at the production default lam.
|
|
# Noise 0.003 < the lam=0.20 cumulative-deviation budget; proves the detector does not
|
|
# kill on ordinary variance (production lam calibration is validated separately, by the
|
|
# forward track). Deterministic via fixed seed.
|
|
ph = PageHinkleyDecay(reenter_days=10) # default lam=0.20
|
|
rng = np.random.default_rng(0)
|
|
alive = [ph.update(0.003 + 0.003 * float(rng.standard_normal())) for _ in range(500)]
|
|
assert all(alive)
|
|
assert not ph.killed
|
|
|
|
|
|
def test_mean_flip_kills():
|
|
ph = PageHinkleyDecay(lam=0.05, reenter_days=10)
|
|
for _ in range(200):
|
|
ph.update(0.002)
|
|
killed_at = None
|
|
for i in range(200):
|
|
if not ph.update(-0.01):
|
|
killed_at = i
|
|
break
|
|
assert killed_at is not None
|
|
assert ph.killed
|
|
|
|
|
|
def test_rearm_after_hysteresis():
|
|
ph = PageHinkleyDecay(lam=0.01, reenter_days=5)
|
|
for _ in range(50):
|
|
ph.update(0.001)
|
|
for _ in range(50):
|
|
if not ph.update(-0.02):
|
|
break
|
|
assert ph.killed
|
|
states = [ph.update(0.0) for _ in range(5)]
|
|
assert states[-1] is True
|
|
assert not ph.killed
|
|
|
|
|
|
def test_to_from_dict_roundtrip():
|
|
ph = PageHinkleyDecay(lam=0.05)
|
|
for _ in range(20):
|
|
ph.update(0.001)
|
|
d = json.loads(json.dumps(ph.to_dict()))
|
|
ph2 = PageHinkleyDecay.from_dict(d)
|
|
assert ph2.to_dict() == ph.to_dict()
|
|
|
|
|
|
# --- WindowedEdgeHealth: scale-invariant, bounded, continuously-recovering edge health -----------
|
|
|
|
def _deterministic_series(n: int, mu: float, amp: float = 0.0, *, freq: float = 0.37) -> list[float]:
|
|
"""Fixed pseudo-series: constant drift `mu` + a deterministic sine oscillation (NO RNG / time).
|
|
Reproducible across runs and across scaling."""
|
|
return [mu + amp * math.sin(freq * i) for i in range(n)]
|
|
|
|
|
|
def _health_seq(detector: WindowedEdgeHealth, series: list[float]) -> list[float]:
|
|
out = []
|
|
for r in series:
|
|
detector.update(r)
|
|
out.append(detector.health())
|
|
return out
|
|
|
|
|
|
def test_windowed_health_scale_invariant():
|
|
# Property 1: t-stat is unitless -> feeding [r*1000] yields the SAME health sequence as the base.
|
|
base = _deterministic_series(300, mu=0.001, amp=0.003)
|
|
scaled = [r * 1000.0 for r in base]
|
|
seq_base = _health_seq(WindowedEdgeHealth(), base)
|
|
seq_scaled = _health_seq(WindowedEdgeHealth(), scaled)
|
|
assert len(seq_base) == len(seq_scaled)
|
|
for a, b in zip(seq_base, seq_scaled):
|
|
assert a == pytest.approx(b, rel=1e-9, abs=1e-12)
|
|
|
|
|
|
def test_windowed_health_bounded_on_stationary_positive():
|
|
# Property 2: a long STATIONARY positive-mean series stays HIGH (>= ~0.8 most of the time);
|
|
# it must NOT pin at 0 like the inception-cumulative PH eventually does.
|
|
series = _deterministic_series(1000, mu=0.0015, amp=0.003)
|
|
seq = _health_seq(WindowedEdgeHealth(), series)
|
|
settled = seq[100:] # past bootstrap / warmup
|
|
high = sum(1 for h in settled if h >= 0.8)
|
|
assert high / len(settled) >= 0.8
|
|
assert min(settled) > 0.0 # never pinned to floor
|
|
|
|
|
|
def test_windowed_health_detects_sustained_decay():
|
|
# Property 3: positive regime then a sustained negative-mean regime -> health -> theta_min
|
|
# within ~window days of the flip.
|
|
window = 63
|
|
pos = _deterministic_series(200, mu=0.002, amp=0.002)
|
|
neg = _deterministic_series(200, mu=-0.004, amp=0.002)
|
|
det = WindowedEdgeHealth(window=window)
|
|
_health_seq(det, pos)
|
|
assert det.health() > 0.8 # healthy at the flip
|
|
seq_after = _health_seq(det, neg)
|
|
assert seq_after[window] < 0.2 # decayed toward theta_min within `window` days
|
|
assert min(seq_after) == pytest.approx(0.0, abs=1e-9) # reaches the floor
|
|
|
|
|
|
def test_windowed_health_rearms_after_recovery():
|
|
# Property 4: after decay, recovering to positive returns climbs health back toward 1.0
|
|
# within ~window days (no consecutive-day requirement).
|
|
window = 63
|
|
det = WindowedEdgeHealth(window=window)
|
|
_health_seq(det, _deterministic_series(200, mu=0.002, amp=0.002)) # healthy
|
|
_health_seq(det, _deterministic_series(200, mu=-0.004, amp=0.002)) # decay
|
|
assert det.health() < 0.2 # decayed
|
|
seq_recover = _health_seq(det, _deterministic_series(200, mu=0.003, amp=0.002))
|
|
assert seq_recover[window] > 0.8 # re-armed within `window` days
|
|
assert max(seq_recover) == pytest.approx(1.0)
|
|
|
|
|
|
def test_windowed_health_bootstrap_is_one():
|
|
# Property 5: health == 1.0 for the first `min_obs` updates (don't cut a young edge).
|
|
min_obs = 10
|
|
det = WindowedEdgeHealth(min_obs=min_obs)
|
|
# Even a clearly-negative young series (with variance, so it has a real t-stat) stays at 1.0
|
|
# while fewer than min_obs observations have accumulated.
|
|
neg = _deterministic_series(min_obs + 1, mu=-0.05, amp=0.01)
|
|
for i in range(min_obs - 1):
|
|
det.update(neg[i])
|
|
assert det.health() == 1.0, f"bootstrap broke at obs {i + 1}"
|
|
# Reaching min_obs observations ends the bootstrap; a negative series is no longer 1.0.
|
|
det.update(neg[min_obs - 1])
|
|
assert det.n == min_obs
|
|
assert det.health() < 1.0
|
|
|
|
|
|
def test_windowed_health_to_from_dict_roundtrip():
|
|
det = WindowedEdgeHealth(window=40, t_full=0.6, t_dead=-1.5, theta_min=0.05, min_obs=8)
|
|
for r in _deterministic_series(50, mu=0.001, amp=0.002):
|
|
det.update(r)
|
|
d = json.loads(json.dumps(det.to_dict()))
|
|
det2 = WindowedEdgeHealth.from_dict(d)
|
|
assert det2.to_dict() == det.to_dict()
|
|
assert det2.health() == det.health()
|