feat: wavelets

backend/services/auto_cycle.py

@@ -1016,6 +1016,17 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
         except Exception as _ce:
             logger.warning(f"[Cycle] Régime clustering failed (non-blocking): {_ce}")
 
+        # ── Step 5.7: Wavelet signal scan (watchlist instruments) ────────────
+        _wavelet_results: list = []
+        try:
+            from services.wavelet_signals import compute_and_save_wavelet_signals
+            _wavelet_results = compute_and_save_wavelet_signals(run_id)
+            summary["wavelet_signals_count"] = len(_wavelet_results)
+            logger.info(f"[Cycle {run_id[:16]}] Wavelet signals: {len(_wavelet_results)} detected")
+        except Exception as _we:
+            logger.warning(f"[Cycle] Wavelet signal scan failed (non-blocking): {_we}")
+            summary["wavelet_signals_count"] = 0
+
         # ── Step 6: GPT-4o cycle commentary ──────────────────────────────────
         logger.info(f"[Cycle {run_id[:16]}] Step 6: generating commentary")
         commentary = _generate_cycle_commentary(
@@ -1046,6 +1057,7 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
             dominant_regime=dominant,
             commentary=commentary,
             status="completed",
+            wavelet_signals_count=summary.get("wavelet_signals_count", 0),
         )
         _current_status["last_run_at"] = datetime.utcnow().isoformat()
         logger.info(f"[Cycle {run_id[:16]}] Completed — {added_count} new patterns, {len(scored)} scored")
@@ -1116,6 +1128,7 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
                 portfolio_monitor=summary.get("portfolio_monitor"),
                 commentary=commentary,
                 options_assessment=_options_assessment,
+                wavelet_signals=_wavelet_results,
             )
             if _cycle_report:
                 from services.database import save_cycle_report
@@ -1352,6 +1365,7 @@ def _generate_cycle_report(
     portfolio_monitor: Optional[Dict],
     commentary: Optional[str],
     options_assessment: Optional[Dict] = None,
+    wavelet_signals: Optional[List[Dict]] = None,
 ) -> Optional[Dict]:
     """
     Build the full cycle report dict:
@@ -1599,6 +1613,8 @@ Réponds en JSON avec ce schéma EXACT:
         # Risk
         "portfolio_monitor": portfolio_monitor,
         "risk_alerts": portfolio_monitor.get("alerts") if portfolio_monitor else None,
+        # Wavelet signals detected this cycle (watchlist scan, see services.wavelet_signals)
+        "wavelet_signals": wavelet_signals or [],
         # Top scored patterns
         "top_scored": [
             {"name": s.get("geo_trigger"), "score": s.get("score"),

backend/services/database.py

@@ -110,12 +110,42 @@ def init_db():
             sort_order INTEGER DEFAULT 0,
             added_at TEXT DEFAULT (datetime('now'))
         )""",
+        # Wavelets — saved optimization/simulation runs (ported from InstrumentSimulator's
+        # WaveletOptimizationRun: form/results are free-form JSON blobs, not modeled relationally)
+        """CREATE TABLE IF NOT EXISTS wavelet_simulations (
+            id TEXT PRIMARY KEY,
+            name TEXT NOT NULL,
+            created_at TEXT DEFAULT (datetime('now')),
+            updated_at TEXT DEFAULT (datetime('now')),
+            form_json TEXT DEFAULT '{}',
+            results_json TEXT DEFAULT '[]',
+            excluded_instruments_json TEXT DEFAULT '[]'
+        )""",
+        # Wavelets — latest per-ticker signal detected during the auto-cycle watchlist scan
+        """CREATE TABLE IF NOT EXISTS wavelet_watchlist_signals (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            run_id TEXT,
+            ticker TEXT NOT NULL,
+            computed_at TEXT DEFAULT (datetime('now')),
+            band_label TEXT,
+            period_low_days REAL,
+            period_high_days REAL,
+            signal_kind TEXT,
+            direction TEXT,
+            price_at_signal REAL
+        )""",
+        "ALTER TABLE cycle_runs ADD COLUMN wavelet_signals_count INTEGER DEFAULT 0",
     ]:
         try:
             c.execute(_sql)
         except Exception:
             pass
 
+    try:
+        c.execute("CREATE INDEX IF NOT EXISTS idx_wws_ticker_date ON wavelet_watchlist_signals(ticker, computed_at DESC)")
+    except Exception:
+        pass
+
     # Specialist Reports — surprise index + text sentiment columns
     try:
         c.execute("ALTER TABLE specialist_reports ADD COLUMN consensus_estimate REAL")
@@ -2209,7 +2239,7 @@ def update_cycle_run(run_id: str, **fields) -> None:
     if not fields:
         return
     allowed = {"completed_at", "patterns_suggested", "patterns_added", "patterns_scored",
-               "geo_score", "dominant_regime", "commentary", "status"}
+               "geo_score", "dominant_regime", "commentary", "status", "wavelet_signals_count"}
     sets = ", ".join(f"{k}=?" for k in fields if k in allowed)
     vals = [v for k, v in fields.items() if k in allowed]
     if not sets:
@@ -3056,6 +3086,133 @@ def reorder_instruments_watchlist(tickers: List[str]) -> None:
     conn.close()
 
 
+# ── Wavelets — saved simulation/optimization runs ─────────────────────────────
+
+def save_wavelet_simulation(name: str, form: Dict, results: Optional[List[Dict]] = None,
+                             excluded_instruments: Optional[List[str]] = None) -> Dict:
+    import uuid
+    sim_id = uuid.uuid4().hex
+    now = datetime.utcnow().isoformat()
+    conn = get_conn()
+    conn.execute(
+        "INSERT INTO wavelet_simulations (id, name, created_at, updated_at, form_json, results_json, excluded_instruments_json) "
+        "VALUES (?, ?, ?, ?, ?, ?, ?)",
+        (sim_id, name, now, now, json.dumps(form or {}), json.dumps(results or []), json.dumps(excluded_instruments or [])),
+    )
+    conn.commit()
+    conn.close()
+    return get_wavelet_simulation(sim_id)
+
+
+def get_wavelet_simulations() -> List[Dict]:
+    conn = get_conn()
+    rows = conn.execute(
+        "SELECT id, name, created_at, updated_at, results_json FROM wavelet_simulations ORDER BY updated_at DESC"
+    ).fetchall()
+    conn.close()
+    out = []
+    for r in rows:
+        d = dict(r)
+        try:
+            result_count = len(json.loads(d.pop("results_json")) or [])
+        except Exception:
+            result_count = 0
+        d["result_count"] = result_count
+        out.append(d)
+    return out
+
+
+def get_wavelet_simulation(sim_id: str) -> Optional[Dict]:
+    conn = get_conn()
+    row = conn.execute("SELECT * FROM wavelet_simulations WHERE id=?", (sim_id,)).fetchone()
+    conn.close()
+    if not row:
+        return None
+    d = dict(row)
+    d["form"] = json.loads(d.pop("form_json") or "{}")
+    d["results"] = json.loads(d.pop("results_json") or "[]")
+    d["excluded_instruments"] = json.loads(d.pop("excluded_instruments_json") or "[]")
+    return d
+
+
+def update_wavelet_simulation(sim_id: str, name: Optional[str] = None, form: Optional[Dict] = None,
+                               results: Optional[List[Dict]] = None, append_results: Optional[List[Dict]] = None,
+                               excluded_instruments: Optional[List[str]] = None) -> Optional[Dict]:
+    current = get_wavelet_simulation(sim_id)
+    if not current:
+        return None
+    new_name = name if name is not None else current["name"]
+    new_form = form if form is not None else current["form"]
+    if results is not None:
+        new_results = results
+    elif append_results:
+        new_results = [*current["results"], *append_results]
+    else:
+        new_results = current["results"]
+    new_excluded = excluded_instruments if excluded_instruments is not None else current["excluded_instruments"]
+
+    conn = get_conn()
+    conn.execute(
+        "UPDATE wavelet_simulations SET name=?, form_json=?, results_json=?, excluded_instruments_json=?, updated_at=? WHERE id=?",
+        (new_name, json.dumps(new_form), json.dumps(new_results), json.dumps(new_excluded), datetime.utcnow().isoformat(), sim_id),
+    )
+    conn.commit()
+    conn.close()
+    return get_wavelet_simulation(sim_id)
+
+
+def delete_wavelet_simulation(sim_id: str) -> bool:
+    conn = get_conn()
+    conn.execute("DELETE FROM wavelet_simulations WHERE id=?", (sim_id,))
+    changed = conn.total_changes > 0
+    conn.commit()
+    conn.close()
+    return changed
+
+
+# ── Wavelets — automated watchlist signal scan (per cycle) ────────────────────
+
+def save_wavelet_signals(run_id: str, signals: List[Dict]) -> None:
+    if not signals:
+        return
+    conn = get_conn()
+    for s in signals:
+        conn.execute(
+            "INSERT INTO wavelet_watchlist_signals "
+            "(run_id, ticker, band_label, period_low_days, period_high_days, signal_kind, direction, price_at_signal) "
+            "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+            (run_id, s.get("ticker"), s.get("band_label"), s.get("period_low_days"), s.get("period_high_days"),
+             s.get("signal_kind"), s.get("direction"), s.get("price_at_signal")),
+        )
+    conn.commit()
+    conn.close()
+
+
+def get_latest_wavelet_signals() -> List[Dict]:
+    """Most recent signal per ticker (one row per ticker, its latest computed_at)."""
+    conn = get_conn()
+    rows = conn.execute(
+        """SELECT w.* FROM wavelet_watchlist_signals w
+           INNER JOIN (
+               SELECT ticker, MAX(computed_at) AS max_computed_at
+               FROM wavelet_watchlist_signals GROUP BY ticker
+           ) latest ON w.ticker = latest.ticker AND w.computed_at = latest.max_computed_at
+           ORDER BY w.computed_at DESC"""
+    ).fetchall()
+    conn.close()
+    return [dict(r) for r in rows]
+
+
+def get_wavelet_signals_history(ticker: str, days: int = 30) -> List[Dict]:
+    conn = get_conn()
+    rows = conn.execute(
+        "SELECT * FROM wavelet_watchlist_signals WHERE ticker=? AND computed_at >= datetime('now', ?) ORDER BY computed_at DESC",
+        (ticker.upper(), f"-{days} days"),
+    ).fetchall()
+    conn.close()
+    return [dict(r) for r in rows]
+
+
 # ── System Logs ───────────────────────────────────────────────────────────────
 
 def log_system_event(

backend/services/geo_analyzer.py

@@ -640,6 +640,13 @@ GEOPOLITICAL_RISK_WEIGHTS = {
 }
 
 
+# A single extreme-severity article (e.g. imminent war, market-moving Fed shock) must not get diluted
+# away just because other categories are quiet that day. This floor is driven by the single worst
+# category score and stays inert below ~0.7, then ramps up steeply — solved so a lone 0.90 floors the
+# final score at 50 (see MAX_CATEGORY_FLOOR_EXPONENT below).
+MAX_CATEGORY_FLOOR_EXPONENT = 6.58
+
+
 def compute_geo_risk_score(events: List[Dict[str, Any]]) -> Dict[str, Any]:
     """Compute a global geopolitical risk score 0-100 from recent events."""
     if not events:
@@ -658,7 +665,13 @@ def compute_geo_risk_score(events: List[Dict[str, Any]]) -> Dict[str, Any]:
         category_scores.get(cat, 0) * weight
         for cat, weight in GEOPOLITICAL_RISK_WEIGHTS.items()
     )
-    score = min(100, round(weighted * 100, 1))
+
+    # Non-linear floor: the single most severe category score alone can force the score up,
+    # even if it's the only hot category — breaks the "many quiet categories dilute one severe one" effect.
+    max_category_impact = max(category_scores.values(), default=0.0)
+    floor = max_category_impact ** MAX_CATEGORY_FLOOR_EXPONENT
+
+    score = min(100, round(max(weighted, floor) * 100, 1))
 
     if score < 25:
         level = "low"

backend/services/wavelet_engine.py

@@ -0,0 +1,447 @@
+"""
+Wavelet band decomposition (CWT + synchrosqueezed variant), ported near-verbatim from
+c:\\DataS\\InstrumentSimulator\\backend\\app\\wavelet.py (project "Macro Causal Lab").
+
+No project-specific dependencies — pure numpy/ssqueezepy, safe to call from any service.
+"""
+import warnings
+
+import numpy as np
+from ssqueezepy import cwt, icwt
+from ssqueezepy.utils.cwt_utils import center_frequency
+
+MIN_SCALES_PER_BAND = 4
+
+# Fixed period (days) lower-bounds for bands 0..5, independent of the analysis
+# window's length. Without this, dividing [scales.min(), scales.max()] into
+# num_levels *equal* log-spaced buckets means "band i" covers a totally
+# different real-world period range depending on how much history is fed in
+# (a longer window resolves longer scales, which shifts every intermediate
+# boundary upward) - so the same band index is not comparable across window
+# lengths. Anchoring to absolute day-periods keeps "band i" meaning the same
+# oscillation whether the window is 3 months or 3 years; only the last band's
+# upper bound stays open, since a longer window can genuinely resolve slower
+# cycles a short one cannot.
+CANONICAL_PERIOD_EDGES_DAYS = [0.3, 1.5, 4.0, 12.0, 35.0, 90.0]
+
+
+def _period_to_log_scale_fn(wavelet, scales, n, num_calib_points=16):
+    """Build an interpolation from log(period_days) to log(scale) by sampling
+    center_frequency at a handful of scales (not every scale - it's not
+    vectorized and can be slow over hundreds/thousands of scales).
+    """
+    log_scales_full = np.log(scales)
+    calib_log_scales = np.linspace(log_scales_full.min(), log_scales_full.max(), min(num_calib_points, len(scales)))
+    calib_periods = np.array([1 / center_frequency(wavelet, scale=float(np.exp(s)), N=n) for s in calib_log_scales])
+    calib_log_periods = np.log(calib_periods)
+
+    def period_to_log_scale(period_days):
+        return float(np.interp(np.log(period_days), calib_log_periods, calib_log_scales))
+
+    return period_to_log_scale
+
+
+def band_decompose(
+    values: list[float],
+    dates: list[str],
+    num_levels: int = 4,
+    wavelet: str = "gmw",
+) -> dict:
+    x = np.asarray(values, dtype=float)
+    n = len(x)
+    if n < 32:
+        raise ValueError("Serie trop courte pour une analyse ondelette (32 points minimum).")
+
+    x_mean = float(x.mean())
+    xc = x - x_mean
+
+    Wx, scales = cwt(xc, wavelet=wavelet)
+    log_scales = np.log(scales)
+
+    period_to_log_scale = _period_to_log_scale_fn(wavelet, scales, n)
+    period_edges = list(CANONICAL_PERIOD_EDGES_DAYS[:num_levels])
+    while len(period_edges) < num_levels:
+        period_edges.append(period_edges[-1] * 2.5)
+
+    # scales[0] = smallest scale = highest frequency = shortest period (impulses/noise);
+    # scales[-1] = largest scale = lowest frequency = longest period (slow trend).
+    bands = []
+    for i in range(num_levels):
+        lo = period_to_log_scale(period_edges[i])
+        hi = period_to_log_scale(period_edges[i + 1]) if i + 1 < len(period_edges) else log_scales.max()
+        mask = (log_scales >= lo) & (log_scales <= hi if i == num_levels - 1 else log_scales < hi)
+        scales_band = scales[mask]
+
+        if len(scales_band) >= MIN_SCALES_PER_BAND:
+            try:
+                recon = icwt(Wx[mask, :], wavelet=wavelet, scales=scales_band, x_len=n)
+            except Exception:
+                # ssqueezepy's internal scale-type inference (log vs linear spacing
+                # detection) can raise on some narrow slices depending on how many
+                # scales land in this bucket for this specific series length.
+                # Treat as a negligible band instead of failing the whole request.
+                recon = np.zeros(n)
+            scale_lo, scale_hi = float(scales_band.min()), float(scales_band.max())
+        else:
+            # Too few scales in this bucket for a stable reconstruction; report a
+            # flat zero band (its content ends up folded into the residual) so the
+            # band list always stays num_levels long.
+            recon = np.zeros(n)
+            scale_lo, scale_hi = float(np.exp(lo)), float(np.exp(hi))
+
+        freq_at_scale_hi = center_frequency(wavelet, scale=scale_hi, N=n)
+        freq_at_scale_lo = center_frequency(wavelet, scale=scale_lo, N=n)
+        period_low_days = round(1 / freq_at_scale_lo, 1) if freq_at_scale_lo else None
+        period_high_days = round(1 / freq_at_scale_hi, 1) if freq_at_scale_hi else None
+        label = (
+            f"{period_low_days}-{period_high_days}j"
+            if period_low_days is not None and period_high_days is not None
+            else f"bande {i + 1}"
+        )
+        bands.append(
+            {
+                "index": i,
+                "label": label,
+                "period_low_days": period_low_days,
+                "period_high_days": period_high_days,
+                "series": [round(float(v), 6) for v in recon],
+            }
+        )
+
+    reconstructed_total = x_mean + sum(np.array(band["series"]) for band in bands)
+    residual = x - reconstructed_total
+
+    return {
+        "dates": dates,
+        "original": [round(float(v), 6) for v in x],
+        "mean": round(x_mean, 6),
+        "bands": bands,
+        "residual": [round(float(v), 6) for v in residual],
+        "wavelet": wavelet,
+    }
+
+
+def windowed_band_decompose(
+    values: list[float],
+    dates: list[str],
+    window_size: int,
+    num_levels: int = 4,
+    wavelet: str = "gmw",
+) -> dict:
+    """Decompose a long series by running band_decompose independently on
+    consecutive slices of `window_size` points, then concatenating the
+    results. Each slice is analyzed on its own terms (own local mean, own
+    CWT scale range), so precision near the reconstructed curve stays high
+    even over a long history; the trade-off is a visible discontinuity at
+    each slice boundary, which is acceptable for this diagnostic view.
+    """
+    n = len(values)
+    window_size = max(32, min(window_size, n)) if window_size else n
+    if n <= window_size:
+        result = band_decompose(values, dates, num_levels, wavelet)
+        result["window_size"] = window_size
+        result["chunks"] = 1
+        return result
+
+    all_dates: list[str] = []
+    all_original: list[float] = []
+    all_residual: list[float] = []
+    band_series: list[list[float]] = [[] for _ in range(num_levels)]
+    band_meta: list[dict | None] = [None] * num_levels
+
+    chunk_count = 0
+    start = 0
+    while start < n:
+        end = min(start + window_size, n)
+        if 0 < n - end < 32:
+            # Don't leave a too-short trailing slice: fold the remainder in.
+            end = n
+        chunk = band_decompose(values[start:end], dates[start:end], num_levels, wavelet)
+        chunk_count += 1
+        all_dates.extend(chunk["dates"])
+        all_original.extend(chunk["original"])
+        all_residual.extend(chunk["residual"])
+        for band in chunk["bands"]:
+            band_series[band["index"]].extend(band["series"])
+            band_meta[band["index"]] = band
+        start = end
+
+    bands = []
+    for i in range(num_levels):
+        meta = band_meta[i] or {"label": f"bande {i + 1}", "period_low_days": None, "period_high_days": None}
+        bands.append(
+            {
+                "index": i,
+                "label": meta["label"],
+                "period_low_days": meta["period_low_days"],
+                "period_high_days": meta["period_high_days"],
+                "series": band_series[i],
+            }
+        )
+
+    return {
+        "dates": all_dates,
+        "original": all_original,
+        "mean": round(float(np.mean(all_original)), 6) if all_original else 0.0,
+        "bands": bands,
+        "residual": all_residual,
+        "wavelet": wavelet,
+        "window_size": window_size,
+        "chunks": chunk_count,
+    }
+
+
+def rolling_causal_bands(
+    values: list[float],
+    dates: list[str],
+    start_idx: int,
+    lookback: int,
+    num_levels: int = 4,
+    wavelet: str = "gmw",
+    step: int = 1,
+) -> dict:
+    """Walk-forward band decomposition with no look-ahead: the band value
+    reported for day `t` is computed from a CWT run only on the trailing
+    `lookback` points ending at `t` (never anything after `t`). This is
+    deliberately much slower than `band_decompose` (one CWT per day instead
+    of one for the whole range) - that cost is the point: a single
+    whole-range CWT lets every point "see" the entire future through the
+    transform's global/symmetric support, which makes any backtest built on
+    it meaningless (the peaks/troughs it finds were computed with hindsight).
+    `step` > 1 re-runs the CWT every `step` days and holds the last computed
+    value in between, trading fidelity for speed on long ranges.
+    """
+    n = len(values)
+    out_dates: list[str] = []
+    out_original: list[float] = []
+    band_series: list[list[float]] = [[] for _ in range(num_levels)]
+    band_labels = [f"bande {i + 1}" for i in range(num_levels)]
+    last_tip: list[float] | None = None
+    recomputations = 0
+
+    for t in range(start_idx, n):
+        window_start = t - lookback + 1
+        if window_start < 0:
+            continue  # not enough trailing history yet for a full lookback window
+        if last_tip is None or (t - start_idx) % step == 0:
+            window_values = values[window_start:t + 1]
+            window_dates = dates[window_start:t + 1]
+            result = band_decompose(window_values, window_dates, num_levels, wavelet)
+            last_tip = [band["series"][-1] for band in result["bands"]]
+            band_labels = [band["label"] for band in result["bands"]]
+            recomputations += 1
+        out_dates.append(dates[t])
+        out_original.append(values[t])
+        for i in range(num_levels):
+            band_series[i].append(last_tip[i])
+
+    bands = [
+        {
+            "index": i,
+            "label": band_labels[i],
+            "period_low_days": None,
+            "period_high_days": None,
+            "series": band_series[i],
+        }
+        for i in range(num_levels)
+    ]
+
+    return {
+        "dates": out_dates,
+        "original": out_original,
+        "bands": bands,
+        "wavelet": wavelet,
+        "lookback": lookback,
+        "step": step,
+        "recomputations": recomputations,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Synchrosqueezed variant (opt-in, method="ssq"). Everything above this line
+# is completely untouched by what follows: band_decompose, windowed_band_decompose
+# and rolling_causal_bands are the exact same functions used when method="cwt"
+# (the default), so any existing simulation re-run with its original config
+# hits the same code path and produces byte-identical results as before.
+#
+# Rationale: plain cwt/icwt smears a single true frequency's energy across
+# several neighboring scales (that's *why* band_decompose needs canonical period
+# edges and a minimum-scales guard - the boundaries are fuzzy). Synchrosqueezing
+# reassigns that smeared energy onto its estimated true instantaneous frequency
+# before splitting into bands, which should reduce cross-band leakage. It also
+# exposes two things plain cwt does not: per-band energy |Tx|^2 (how dominant
+# that cycle is *right now*, independent of direction) and a dominant-frequency
+# ridge (which cycle length is currently winning, tracked over time).
+#
+# Separate period edges from the cwt path's CANONICAL_PERIOD_EDGES_DAYS: with
+# fs=1.0 (one sample = one day, matching our daily data), ssq_cwt has a hard
+# Nyquist floor at exactly 2.0 days (max resolvable frequency = fs/2). The cwt
+# path's edges start at 0.3 days - below that floor, so reusing them here would
+# leave the fastest band permanently empty. This does mean "band 0" covers a
+# different absolute range under ssq than under cwt; that's an inherent
+# consequence of ssq's stricter frequency floor, not a rounding choice.
+# ---------------------------------------------------------------------------
+
+CANONICAL_PERIOD_EDGES_DAYS_SSQ = [2.0, 4.0, 10.0, 25.0, 60.0, 120.0]
+
+
+def band_decompose_ssq(
+    values: list[float],
+    dates: list[str],
+    num_levels: int = 4,
+    wavelet: str = "gmw",
+) -> dict:
+    from ssqueezepy import issq_cwt, ssq_cwt
+    from ssqueezepy.ridge_extraction import extract_ridges
+
+    x = np.asarray(values, dtype=float)
+    n = len(x)
+    if n < 32:
+        raise ValueError("Serie trop courte pour une analyse ondelette (32 points minimum).")
+
+    x_mean = float(x.mean())
+    xc = x - x_mean
+
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", RuntimeWarning)
+        # fs=1.0: one sample = one day, so ssq_freqs comes out directly in
+        # cycles/day and 1/ssq_freqs is a period in days with no extra
+        # calibration step (unlike band_decompose's scale->period conversion,
+        # which needs _period_to_log_scale_fn precisely because plain cwt's
+        # scale axis has no absolute unit by itself).
+        Tx, _Wx, ssq_freqs, _scales = ssq_cwt(xc, wavelet=wavelet, fs=1.0)
+
+    ssq_freqs = np.asarray(ssq_freqs)
+    periods = 1.0 / ssq_freqs
+
+    period_edges = list(CANONICAL_PERIOD_EDGES_DAYS_SSQ[:num_levels])
+    while len(period_edges) < num_levels:
+        period_edges.append(period_edges[-1] * 2.5)
+
+    bands = []
+    for i in range(num_levels):
+        lo_period = period_edges[i]
+        hi_period = period_edges[i + 1] if i + 1 < len(period_edges) else float(periods.max())
+        mask = (periods >= lo_period) & (periods <= hi_period if i == num_levels - 1 else periods < hi_period)
+
+        if mask.any():
+            try:
+                Tx_band = np.zeros_like(Tx)
+                Tx_band[mask, :] = Tx[mask, :]
+                with warnings.catch_warnings():
+                    warnings.simplefilter("ignore", RuntimeWarning)
+                    recon = np.real(issq_cwt(Tx_band, wavelet=wavelet))
+            except Exception:
+                # Defensive, mirroring band_decompose: never let one band's edge
+                # case fail the whole request.
+                recon = np.zeros(n)
+            energy = np.sum(np.abs(Tx[mask, :]) ** 2, axis=0)
+        else:
+            recon = np.zeros(n)
+            energy = np.zeros(n)
+
+        period_low_days = round(float(lo_period), 1)
+        period_high_days = round(float(hi_period), 1)
+        bands.append(
+            {
+                "index": i,
+                "label": f"{period_low_days}-{period_high_days}j",
+                "period_low_days": period_low_days,
+                "period_high_days": period_high_days,
+                "series": [round(float(v), 6) for v in recon],
+                "energy": [round(float(v), 8) for v in energy],
+            }
+        )
+
+    reconstructed_total = x_mean + sum(np.array(band["series"]) for band in bands)
+    residual = x - reconstructed_total
+
+    try:
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", RuntimeWarning)
+            ridge_idxs = extract_ridges(Tx, ssq_freqs, n_ridges=1, bw=4)
+        ridge_periods = 1.0 / ssq_freqs[ridge_idxs.ravel()]
+        ridge_period_days = [round(float(v), 3) for v in ridge_periods]
+    except Exception:
+        ridge_period_days = [None] * n
+
+    return {
+        "dates": dates,
+        "original": [round(float(v), 6) for v in x],
+        "mean": round(x_mean, 6),
+        "bands": bands,
+        "residual": [round(float(v), 6) for v in residual],
+        "wavelet": wavelet,
+        "ridge_period_days": ridge_period_days,
+    }
+
+
+def rolling_causal_bands_ssq(
+    values: list[float],
+    dates: list[str],
+    start_idx: int,
+    lookback: int,
+    num_levels: int = 4,
+    wavelet: str = "gmw",
+    step: int = 1,
+) -> dict:
+    """Synchrosqueezed counterpart to rolling_causal_bands - identical
+    walk-forward, no-look-ahead structure (same day-by-day trailing window,
+    same step logic), just calling band_decompose_ssq instead of
+    band_decompose. rolling_causal_bands itself is untouched.
+    """
+    n = len(values)
+    out_dates: list[str] = []
+    out_original: list[float] = []
+    band_series: list[list[float]] = [[] for _ in range(num_levels)]
+    energy_series: list[list[float]] = [[] for _ in range(num_levels)]
+    ridge_series: list[float | None] = []
+    band_labels = [f"bande {i + 1}" for i in range(num_levels)]
+    last_tip: list[float] | None = None
+    last_energy_tip: list[float] | None = None
+    last_ridge_tip: float | None = None
+    recomputations = 0
+
+    for t in range(start_idx, n):
+        window_start = t - lookback + 1
+        if window_start < 0:
+            continue
+        if last_tip is None or (t - start_idx) % step == 0:
+            window_values = values[window_start:t + 1]
+            window_dates = dates[window_start:t + 1]
+            result = band_decompose_ssq(window_values, window_dates, num_levels, wavelet)
+            last_tip = [band["series"][-1] for band in result["bands"]]
+            last_energy_tip = [band["energy"][-1] for band in result["bands"]]
+            last_ridge_tip = result["ridge_period_days"][-1]
+            band_labels = [band["label"] for band in result["bands"]]
+            recomputations += 1
+        out_dates.append(dates[t])
+        out_original.append(values[t])
+        for i in range(num_levels):
+            band_series[i].append(last_tip[i])
+            energy_series[i].append(last_energy_tip[i])
+        ridge_series.append(last_ridge_tip)
+
+    bands = [
+        {
+            "index": i,
+            "label": band_labels[i],
+            "period_low_days": None,
+            "period_high_days": None,
+            "series": band_series[i],
+            "energy": energy_series[i],
+        }
+        for i in range(num_levels)
+    ]
+
+    return {
+        "dates": out_dates,
+        "original": out_original,
+        "bands": bands,
+        "wavelet": wavelet,
+        "lookback": lookback,
+        "step": step,
+        "recomputations": recomputations,
+        "method": "ssq",
+        "ridge_period_days": ridge_series,
+    }

backend/services/wavelet_signals.py

@@ -0,0 +1,137 @@
+"""
+Automated wavelet signal detection for the watchlist — run once per cycle.
+
+Ported (Python subset) from the trigger-signal detectors in
+c:\\DataS\\InstrumentSimulator\\frontend\\src\\main.tsx (lines 180-280, TypeScript).
+Only `extremum` and `level_threshold` are ported here: they're self-contained
+(single curve, no secondary curve/config needed) and robust enough for an
+unattended scan. The richer configurable trigger set (trend_flatten,
+acceleration, band_cross, ridge_shift, energy_threshold) stays exclusive to the
+interactive Wavelets Simulation page (frontend/src/lib/waveletTrade.ts), where a
+user picks and tunes them explicitly.
+"""
+from typing import Dict, List, Optional
+
+
+def _build_extremum_signal(series: List[float], direction: str) -> List[bool]:
+    n = len(series)
+    raw = [False] * n
+    for i in range(1, n - 1):
+        prev_slope = series[i] - series[i - 1]
+        next_slope = series[i + 1] - series[i]
+        if direction == "up" and prev_slope > 0 and next_slope <= 0:
+            raw[i] = True  # peak
+        if direction == "down" and prev_slope < 0 and next_slope >= 0:
+            raw[i] = True  # trough
+    # raw[j] needs series[j+1] to confirm — shift by one day so the signal never
+    # requires tomorrow's data.
+    shifted = [False] * n
+    for i in range(1, n):
+        shifted[i] = raw[i - 1]
+    return shifted
+
+
+def _build_level_threshold_signal(series: List[float], direction: str, threshold_k: float) -> List[bool]:
+    n = len(series)
+    signal = [False] * n
+    s = 0.0
+    sq = 0.0
+    for t in range(n):
+        s += series[t]
+        sq += series[t] * series[t]
+        count = t + 1
+        if count < 20:
+            continue  # not enough history yet for a stable mean/std
+        mean = s / count
+        variance = max(0.0, sq / count - mean * mean)
+        std = variance ** 0.5
+        if direction == "up" and series[t] > mean + threshold_k * std:
+            signal[t] = True
+        if direction == "down" and series[t] < mean - threshold_k * std:
+            signal[t] = True
+    return signal
+
+
+def detect_extremum_signal(series: List[float]) -> Optional[str]:
+    """Returns 'up' (confirmed peak) or 'down' (confirmed trough) if the most
+    recent point is a signal, else None."""
+    if len(series) < 3:
+        return None
+    if _build_extremum_signal(series, "up")[-1]:
+        return "up"
+    if _build_extremum_signal(series, "down")[-1]:
+        return "down"
+    return None
+
+
+def detect_level_threshold_signal(series: List[float], threshold_k: float = 2.0) -> Optional[str]:
+    """Returns 'up' (overbought) or 'down' (oversold) if the most recent point
+    breaches a causal z-score threshold, else None."""
+    if len(series) < 20:
+        return None
+    if _build_level_threshold_signal(series, "up", threshold_k)[-1]:
+        return "up"
+    if _build_level_threshold_signal(series, "down", threshold_k)[-1]:
+        return "down"
+    return None
+
+
+def scan_watchlist_wavelet_signals(num_levels: int = 4, wavelet: str = "gmw", lookback: int = 120, method: str = "cwt") -> List[Dict]:
+    """Compute a causal (no-look-ahead) band decomposition for each watchlist
+    instrument and flag any band whose most recent point is a signal. Only the
+    trailing ~60 output points are computed (not the whole history) — this scan
+    only needs to know about *today*, unlike the interactive Simulation page's
+    full-range backtest."""
+    from services.database import get_instruments_watchlist
+    from services.data_fetcher import get_historical
+    from services.wavelet_engine import rolling_causal_bands, rolling_causal_bands_ssq
+
+    results: List[Dict] = []
+    decomposer = rolling_causal_bands_ssq if method == "ssq" else rolling_causal_bands
+
+    for item in get_instruments_watchlist():
+        ticker = item["ticker"]
+        try:
+            hist = get_historical(ticker, period="1y", interval="1d")
+            if len(hist) < lookback + 32:
+                continue
+            values = [h["close"] for h in hist]
+            dates = [h["date"] for h in hist]
+            start_idx = max(lookback, len(values) - 60)
+            decomposed = decomposer(
+                values, dates,
+                start_idx=start_idx, lookback=lookback,
+                num_levels=num_levels, wavelet=wavelet, step=1,
+            )
+            if not decomposed["dates"]:
+                continue
+            price_at_signal = decomposed["original"][-1]
+
+            for band in decomposed["bands"]:
+                series = band["series"]
+                direction = detect_extremum_signal(series)
+                kind = "extremum" if direction else None
+                if not direction:
+                    direction = detect_level_threshold_signal(series)
+                    kind = "level_threshold" if direction else None
+                if kind and direction:
+                    results.append({
+                        "ticker": ticker,
+                        "band_label": band["label"],
+                        "period_low_days": band.get("period_low_days"),
+                        "period_high_days": band.get("period_high_days"),
+                        "signal_kind": kind,
+                        "direction": direction,
+                        "price_at_signal": price_at_signal,
+                    })
+        except Exception:
+            continue  # one bad ticker must not abort the whole scan
+
+    return results
+
+
+def compute_and_save_wavelet_signals(run_id: str) -> List[Dict]:
+    from services.database import save_wavelet_signals
+    results = scan_watchlist_wavelet_signals()
+    save_wavelet_signals(run_id, results)
+    return results