feat: replace regime monitor with v2 methodology
This commit is contained in:
@@ -1,4 +1,4 @@
|
||||
"""Unit tests for the breadth indicator and the event-study measurement."""
|
||||
"""Tests for v2 correction events and warning alarm episodes."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -6,124 +6,67 @@ from datetime import date, timedelta
|
||||
|
||||
from app.services.breadth_service import _breadth_from_closes, compute_divergence_series
|
||||
from app.services.event_study_service import (
|
||||
_lead,
|
||||
_percentile,
|
||||
alarm_episodes,
|
||||
detect_events,
|
||||
event_centered,
|
||||
signal_centered,
|
||||
evaluate_alarms,
|
||||
)
|
||||
|
||||
|
||||
def _days(n: int, start: date = date(2021, 1, 1)) -> list[date]:
|
||||
return [start + timedelta(days=i) for i in range(n)]
|
||||
def _days(count: int, start: date = date(2021, 1, 1)) -> list[date]:
|
||||
return [start + timedelta(days=index) for index in range(count)]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Event detection
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_detect_events_single_drawdown():
|
||||
closes = [100.0] * 300 + [85.0] * 5 # 15% off the trailing high -> one event
|
||||
dates = _days(len(closes))
|
||||
events = detect_events(closes, dates, threshold_pct=15.0)
|
||||
assert len(events) == 1
|
||||
assert events[0]["index"] == 300
|
||||
def test_detect_events_uses_rising_edge_and_cooldown():
|
||||
closes = [100.0] * 300 + [85.0] * 5 + [100.0] * 50 + [85.0] * 5
|
||||
events = detect_events(closes, _days(len(closes)), threshold_pct=15.0, cooldown=40)
|
||||
assert [event["index"] for event in events] == [300, 355]
|
||||
|
||||
|
||||
def test_detect_events_dedup_without_recovery():
|
||||
closes = [100.0] * 300 + [85.0] * 5 + [80.0] * 5 # deepens but never recovers
|
||||
events = detect_events(closes, _days(len(closes)), threshold_pct=15.0)
|
||||
assert len(events) == 1
|
||||
def test_percentile_is_fixed_from_supplied_values():
|
||||
values = [float(value) for value in range(0, 101, 10)]
|
||||
assert _percentile(values, 50) == 50.0
|
||||
assert _percentile(values, 80) == 80.0
|
||||
assert _percentile([], 80) is None
|
||||
|
||||
|
||||
def test_detect_events_two_after_recovery():
|
||||
closes = [100.0] * 300 + [85.0] * 10 + [100.0] * 300 + [85.0] * 10
|
||||
events = detect_events(closes, _days(len(closes)), threshold_pct=15.0)
|
||||
assert len(events) == 2
|
||||
def test_alarm_requires_upward_crossing_and_reset():
|
||||
dates = _days(10)
|
||||
values = [10, 70, 80, 75, 20, 70, 80, 20, 20, 70]
|
||||
indicator = dict(zip(dates, values))
|
||||
assert alarm_episodes(indicator, dates, threshold=60) == [1, 5, 9]
|
||||
|
||||
|
||||
def test_detect_events_cooldown_suppresses_close_recross():
|
||||
# Dips below threshold then re-crosses only a few bars later.
|
||||
closes = [100.0] * 300 + [85.0] * 3 + [100.0] * 3 + [85.0] * 3
|
||||
dates = _days(len(closes))
|
||||
assert len(detect_events(closes, dates, threshold_pct=15.0, cooldown=40)) == 1
|
||||
assert len(detect_events(closes, dates, threshold_pct=15.0, cooldown=3)) == 2
|
||||
def test_holdout_start_does_not_invent_crossing_when_already_high():
|
||||
dates = _days(6)
|
||||
indicator = dict(zip(dates, [10, 70, 80, 80, 20, 70]))
|
||||
assert alarm_episodes(indicator, dates, threshold=60, start_index=3) == [5]
|
||||
|
||||
|
||||
def test_percentile_interpolation():
|
||||
vals = [float(v) for v in range(0, 101, 10)] # 0,10,...,100
|
||||
assert _percentile(vals, 50) == 50.0
|
||||
assert _percentile(vals, 80) == 80.0
|
||||
assert _percentile([], 50) is None
|
||||
def test_evaluate_alarms_counts_episodes_not_alarm_days():
|
||||
dates = _days(100)
|
||||
result = evaluate_alarms([10, 50, 80], [25, 70], dates, horizon=20)
|
||||
assert result["events_warned"] == 2
|
||||
assert result["events_missed"] == 0
|
||||
assert result["false_alarms"] == 1
|
||||
assert result["median_lead_days"] == 17.5
|
||||
|
||||
|
||||
def test_lead_earliest_crossing():
|
||||
dates = _days(200)
|
||||
t0 = 120
|
||||
indicator = {dates[i]: (70.0 if t0 - 30 <= i <= t0 else 10.0) for i in range(len(dates))}
|
||||
assert _lead(indicator, t0, dates, pre=60, threshold=60.0) == 30
|
||||
assert _lead(indicator, t0, dates, pre=60, threshold=80.0) is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Event-centered lead time
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_event_centered_lead_time():
|
||||
dates = _days(200)
|
||||
t0 = 120
|
||||
# Indicator goes hot 30 days before t0 and stays hot through t0.
|
||||
indicator = {dates[i]: (70.0 if t0 - 30 <= i <= t0 else 10.0) for i in range(len(dates))}
|
||||
res = event_centered(indicator, [t0], dates, pre=60, post=20, threshold=60.0)
|
||||
assert res["median_lead_days"] == 30
|
||||
assert res["events_with_signal"] == 1
|
||||
|
||||
|
||||
def test_breadth_divergence_leads_coincident():
|
||||
dates = _days(200)
|
||||
t0 = 120
|
||||
breadth_ind = {dates[i]: (70.0 if t0 - 30 <= i <= t0 else 10.0) for i in range(len(dates))}
|
||||
coincident = {dates[i]: (70.0 if t0 - 2 <= i <= t0 else 10.0) for i in range(len(dates))}
|
||||
bd = event_centered(breadth_ind, [t0], dates, threshold=60.0)
|
||||
cd = event_centered(coincident, [t0], dates, threshold=60.0)
|
||||
assert bd["median_lead_days"] > cd["median_lead_days"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Signal-centered precision / recall
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_signal_centered_base_rate_and_recall():
|
||||
dates = _days(200)
|
||||
t0 = 120
|
||||
indicator = {dates[i]: (70.0 if t0 - 30 <= i <= t0 else 10.0) for i in range(len(dates))}
|
||||
res = signal_centered(indicator, [t0], dates, horizon=20)
|
||||
assert 0.0 < res["base_rate"] < 1.0
|
||||
# An aligned indicator should catch some of the pre-event window at a mid threshold.
|
||||
row60 = next(r for r in res["rows"] if r["threshold"] == 60)
|
||||
assert row60["recall"] is not None and row60["recall"] > 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Breadth aggregation + divergence
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_breadth_from_closes_fraction_above_sma():
|
||||
dates = _days(5)
|
||||
def test_breadth_from_fixed_closes_and_pure_divergence():
|
||||
dates = _days(10)
|
||||
closes_by_symbol = {
|
||||
"A": list(zip(dates, [1.0, 2.0, 3.0, 4.0, 5.0])), # rising -> above its SMA
|
||||
"B": list(zip(dates, [5.0, 4.0, 3.0, 2.0, 1.0])), # falling -> below
|
||||
"C": list(zip(dates, [3.0, 3.0, 3.0, 3.0, 3.0])), # flat -> not strictly above
|
||||
"A": list(zip(dates, [1.0 + index for index in range(10)])),
|
||||
"B": list(zip(dates, [10.0 - index for index in range(10)])),
|
||||
"C": list(zip(dates, [5.0] * 10)),
|
||||
}
|
||||
breadth = _breadth_from_closes(closes_by_symbol, window=3, min_tickers=2)
|
||||
# At d2: SMA(3) over each -> only A is strictly above -> 1/3.
|
||||
assert breadth[dates[2]] == round(1 / 3 * 100, 2)
|
||||
|
||||
falling_breadth = {dates[index]: 80.0 - index * 3 for index in range(10)}
|
||||
rising_benchmark = list(zip(dates, [100.0 + index for index in range(10)]))
|
||||
divergence = compute_divergence_series(falling_breadth, rising_benchmark, lookback=3)
|
||||
assert divergence[dates[-1]] > 0
|
||||
|
||||
def test_divergence_high_when_price_up_breadth_down():
|
||||
dates = _days(10)
|
||||
breadth = {dates[i]: 80.0 - i * 3 for i in range(len(dates))} # falling breadth
|
||||
benchmark = list(zip(dates, [100.0 + i for i in range(len(dates))])) # rising price
|
||||
div = compute_divergence_series(breadth, benchmark, lookback=3)
|
||||
last = div[dates[-1]]
|
||||
assert last > 50.0 # fragile: price up while breadth deteriorates
|
||||
falling_benchmark = list(zip(dates, [100.0 - index for index in range(10)]))
|
||||
no_divergence = compute_divergence_series(falling_breadth, falling_benchmark, lookback=3)
|
||||
assert no_divergence[dates[-1]] == 0
|
||||
|
||||
Reference in New Issue
Block a user