feat: wavelets

backend/routers/wavelet.py

@@ -110,6 +110,48 @@ def wavelet_rolling(
     return result
 
 
+@router.get("/reliability")
+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),
+    levels: int = Query(4, ge=2, le=6),
+    wavelet: str = Query("gmw"),
+    step: int = Query(1, ge=1),
+    method: str = Query("cwt", description="cwt (default) or ssq"),
+    smooth_days: int = Query(3, ge=1, le=10, description="lag used to smooth the slope before flagging a sign-change as a turning point"),
+    tolerance_days: int = Query(5, ge=0, le=15, description="a hindsight reversal within this many days of the causal one still counts as confirming it"),
+    confirm_horizon: int = Query(10, ge=1, le=30, description="how many extra days of real data the hindsight recomputation gets"),
+):
+    """For every reversal a live (causal, walk-forward) decomposition would have flagged,
+    checks whether redoing the decomposition `confirm_horizon` days later still shows the
+    same reversal — a per-band reliability score for the wavelet's turning-point signals."""
+    from services.wavelet_engine import wavelet_reliability
+
+    # Needs confirm_horizon extra real days beyond the requested causal output range, on
+    # top of the usual lookback padding, so the most recent testable turning points aren't
+    # silently dropped for lack of "future" data.
+    values, dates, out_days = _fetch_padded_history(symbol, period, lookback + confirm_horizon)
+    if len(values) < lookback + confirm_horizon + 32:
+        raise HTTPException(400, "Historique insuffisant pour un test de fiabilité (32 points minimum au-delà de la fenêtre + horizon).")
+
+    cutoff = (datetime.utcnow() - timedelta(days=out_days)).date().isoformat()
+    start_idx = next((i for i, d in enumerate(dates) if d[:10] >= cutoff), None)
+    if start_idx is None:
+        raise HTTPException(400, "Pas de donnees dans la plage de trading demandee.")
+
+    try:
+        result = wavelet_reliability(
+            values, dates,
+            start_idx=start_idx, lookback=lookback,
+            num_levels=levels, wavelet=wavelet, method=method, step=step,
+            smooth_days=smooth_days, tolerance_days=tolerance_days, confirm_horizon=confirm_horizon,
+        )
+    except ValueError as exc:
+        raise HTTPException(400, str(exc)) from exc
+    return result
+
+
 # ── Saved simulation/optimization runs ────────────────────────────────────────
 
 class SimulationCreate(BaseModel):

backend/services/wavelet_engine.py

@@ -445,3 +445,100 @@ def rolling_causal_bands_ssq(
         "method": "ssq",
         "ridge_period_days": ridge_series,
     }
+
+
+def _turning_points(series: list[float], smooth_days: int) -> list[tuple[int, int]]:
+    """Indices (and new direction, +1/-1) where the sign of the `smooth_days`-lag slope of
+    `series` flips. Using the raw day-to-day slope would flag noise as a "reversal"; lagging
+    over a few days smooths that out."""
+    turns: list[tuple[int, int]] = []
+    prev_sign = 0
+    for t in range(smooth_days, len(series)):
+        slope = series[t] - series[t - smooth_days]
+        sign = 1 if slope > 0 else (-1 if slope < 0 else 0)
+        if sign != 0:
+            if prev_sign != 0 and sign != prev_sign:
+                turns.append((t, sign))
+            prev_sign = sign
+    return turns
+
+
+def wavelet_reliability(
+    values: list[float],
+    dates: list[str],
+    start_idx: int,
+    lookback: int,
+    num_levels: int = 4,
+    wavelet: str = "gmw",
+    method: str = "cwt",
+    step: int = 1,
+    smooth_days: int = 3,
+    tolerance_days: int = 5,
+    confirm_horizon: int = 10,
+) -> dict:
+    """
+    For every turning point a CAUSAL (walk-forward, no look-ahead) decomposition flags —
+    i.e. what a live signal would have shown on that date, using only data available up to
+    then — checks whether redoing the decomposition `confirm_horizon` days later (with that
+    much more real data, so the date in question is no longer sitting at the unstable tip of
+    the window) still shows a same-direction reversal within `tolerance_days` of the original
+    date. The fraction confirmed is a per-band confidence score: CWT/SSQ reconstructions are
+    well known to be least reliable right at the edge of the available data — exactly where a
+    live "the band just turned" reading gets made.
+    """
+    rolling_decomposer = rolling_causal_bands_ssq if method == "ssq" else rolling_causal_bands
+    batch_decompose = band_decompose_ssq if method == "ssq" else band_decompose
+
+    causal = rolling_decomposer(
+        values, dates, start_idx=start_idx, lookback=lookback,
+        num_levels=num_levels, wavelet=wavelet, step=step,
+    )
+    causal_dates = causal["dates"]
+    date_to_global = {d: i for i, d in enumerate(dates)}
+
+    bands_out = []
+    for band in causal["bands"]:
+        turn_points = _turning_points(band["series"], smooth_days)
+
+        tested = []
+        for t, direction in turn_points:
+            d_date = causal_dates[t]
+            g_idx = date_to_global.get(d_date)
+            if g_idx is None or g_idx + confirm_horizon >= len(values):
+                continue  # not enough real future data yet to judge this one
+
+            end_idx = g_idx + confirm_horizon
+            window_start = end_idx - lookback + 1
+            if window_start < 0:
+                continue
+            window_values = values[window_start:end_idx + 1]
+            window_dates = dates[window_start:end_idx + 1]
+            try:
+                hindsight = batch_decompose(window_values, window_dates, num_levels, wavelet)
+            except ValueError:
+                continue
+            h_series = hindsight["bands"][band["index"]]["series"]
+
+            d_pos = g_idx - window_start
+            lo, hi = d_pos - tolerance_days, d_pos + tolerance_days
+            hindsight_turns = _turning_points(h_series, smooth_days)
+            confirmed = any(lo <= idx <= hi and new_dir == direction for idx, new_dir in hindsight_turns)
+            tested.append({"date": d_date, "confirmed": confirmed})
+
+        n_tested = len(tested)
+        n_confirmed = sum(1 for t in tested if t["confirmed"])
+        bands_out.append({
+            "label": band["label"],
+            "n_tested": n_tested,
+            "n_confirmed": n_confirmed,
+            "confidence_pct": round(n_confirmed / n_tested * 100, 1) if n_tested else None,
+            "turns": tested,
+        })
+
+    return {
+        "bands": bands_out,
+        "smooth_days": smooth_days,
+        "tolerance_days": tolerance_days,
+        "confirm_horizon": confirm_horizon,
+        "method": method,
+    }