feat: wavelets

This commit is contained in:
OpenSquared
2026-07-19 16:03:09 +02:00
parent b900d77aa0
commit 9079d4e9c5
3 changed files with 27 additions and 8 deletions

View File

@@ -73,7 +73,7 @@ def wavelet_analyze(
def wavelet_rolling(
symbol: str = Query("SPY"),
period: str = Query("1y", description="how much of the causal output range to return"),
lookback: int = Query(130, ge=32),
lookback: int = Query(260, ge=32),
levels: int = Query(4, ge=2, le=6),
wavelet: str = Query("gmw"),
step: int = Query(1, ge=1),
@@ -114,7 +114,7 @@ def wavelet_rolling(
def wavelet_reliability_endpoint(
symbol: str = Query("SPY"),
period: str = Query("1y", description="how much of the causal output range to scan for turning points"),
lookback: int = Query(130, ge=32),
lookback: int = Query(260, ge=32),
levels: int = Query(4, ge=2, le=6),
wavelet: str = Query("gmw"),
step: int = Query(1, ge=1),

View File

@@ -610,11 +610,20 @@ def wavelet_reliability(
# last index already IS "original date + horizon"; no separate bound needed.
d_pos = g_idx - window_start
hindsight_turns = _turning_points(h_series, smooth_days)
confirmed = any(idx >= d_pos and new_dir == direction for idx, new_dir in hindsight_turns)
tested.append({"date": d_date, "confirmed": confirmed})
matches = [idx for idx, new_dir in hindsight_turns if idx >= d_pos and new_dir == direction]
confirmed = bool(matches)
# Actual delay = how many days after the original date the confirming reversal
# showed up (earliest match) — compared against avg_cycle_days below to check
# whether "one cycle" is actually a well-calibrated horizon, or systematically
# too long/short relative to how quickly a real reversal actually confirms.
actual_delay = (min(matches) - d_pos) if confirmed else None
tested.append({"date": d_date, "confirmed": confirmed, "actual_delay_days": actual_delay})
n_tested = len(tested)
n_confirmed = sum(1 for t in tested if t["confirmed"])
delays = [t["actual_delay_days"] for t in tested if t["actual_delay_days"] is not None]
avg_actual_delay = round(sum(delays) / len(delays), 1) if delays else None
calibration_gap = round(sum(abs(avg_cycle_days - d) for d in delays) / len(delays), 1) if delays else None
bands_out.append({
"label": band["label"],
"n_tested": n_tested,
@@ -622,6 +631,8 @@ def wavelet_reliability(
"confidence_pct": round(n_confirmed / n_tested * 100, 1) if n_tested else None,
"avg_cycle_days": round(avg_cycle_days, 1),
"confirm_horizon": confirm_horizon,
"avg_actual_delay_days": avg_actual_delay,
"calibration_gap_days": calibration_gap,
"turns": tested,
})