""" 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 / frontend/src/lib/waveletTrade.ts. All 7 trigger kinds from the interactive Wavelets Simulation page are now available here: extremum, level_threshold, trend_flatten, acceleration, band_cross, ridge_shift (ssq only), energy_threshold (ssq only). Parameters (engine + per-signal enable/thresholds) come from the "Technical Desk" (services.database.get_ai_desk_by_type("technical"), config.signals.wavelet_*) so they're editable from the existing AI Desks config UI — no hardcoded defaults here beyond a safe fallback when the desk/key is absent. The instrument scope stays get_instruments_watchlist() (the desk's own `instruments` list is NOT used, to avoid reintroducing a second overlapping instrument-list source). Every (ticker, band) gets a row every cycle now — signal or not — so the AI chat context always has fresh slope/energy/ridge state, not just firing events (see ai_chat_context.py:_block_wavelet_signals). """ import json from typing import Dict, List, Optional def _compute_slope(series: List[float]) -> List[float]: n = len(series) slope = [0.0] * n for i in range(1, n): slope[i] = series[i] - series[i - 1] if n > 1: slope[0] = slope[1] return slope def _compute_acceleration(slope: List[float]) -> List[float]: n = len(slope) accel = [0.0] * n for i in range(1, n): accel[i] = slope[i] - slope[i - 1] if n > 1: accel[0] = accel[1] return accel def _avg_slope_range(slope: List[float], frm: int, to: int) -> Optional[float]: if frm < 0 or to > len(slope) - 1 or to <= frm: return None return sum(slope[frm + 1:to + 1]) / (to - frm) 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 _build_trend_flatten_signal(series: List[float], direction: str, trend_days: int, flatten_days: int, trend_threshold_k: float, flatten_threshold_k: float) -> List[bool]: n = len(series) slope = _compute_slope(series) signal = [False] * n s = 0.0 sq = 0.0 for t in range(1, n): s += slope[t] sq += slope[t] * slope[t] count = t if t < trend_days + flatten_days or count < 20: continue mean = s / count variance = max(0.0, sq / count - mean * mean) std = variance ** 0.5 trend_thresh = trend_threshold_k * std flatten_thresh = flatten_threshold_k * std trend = _avg_slope_range(slope, t - flatten_days - trend_days, t - flatten_days) flat = _avg_slope_range(slope, t - flatten_days, t) if trend is None or flat is None: continue if direction == "up" and trend > trend_thresh and abs(flat) <= flatten_thresh: signal[t] = True if direction == "down" and trend < -trend_thresh and abs(flat) <= flatten_thresh: signal[t] = True return signal def _build_acceleration_signal(series: List[float], direction: str, days: int, threshold_k: float) -> List[bool]: n = len(series) slope = _compute_slope(series) accel = _compute_acceleration(slope) signal = [False] * n s = 0.0 sq = 0.0 for t in range(2, n): s += accel[t] sq += accel[t] * accel[t] count = t - 1 if t < days or count < 20: continue mean = s / count variance = max(0.0, sq / count - mean * mean) std = variance ** 0.5 thresh = threshold_k * std if direction == "up": if slope[t] <= 0: continue ok = True for d in range(days): idx = t - d if idx < 0 or not (accel[idx] < -thresh): ok = False break signal[t] = ok else: if slope[t] >= 0: continue ok = True for d in range(days): idx = t - d if idx < 0 or not (accel[idx] > thresh): ok = False break signal[t] = ok return signal def _build_band_cross_signal(primary: List[float], secondary: List[float], direction: str) -> List[bool]: n = min(len(primary), len(secondary)) signal = [False] * n for t in range(1, n): prev_diff = primary[t - 1] - secondary[t - 1] curr_diff = primary[t] - secondary[t] if direction == "down" and prev_diff >= 0 and curr_diff < 0: signal[t] = True if direction == "up" and prev_diff <= 0 and curr_diff > 0: 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 detect_trend_flatten_signal(series: List[float], trend_days: int = 10, flatten_days: int = 5, trend_threshold_k: float = 1.0, flatten_threshold_k: float = 0.3) -> Optional[str]: if len(series) < trend_days + flatten_days + 20: return None if _build_trend_flatten_signal(series, "up", trend_days, flatten_days, trend_threshold_k, flatten_threshold_k)[-1]: return "up" if _build_trend_flatten_signal(series, "down", trend_days, flatten_days, trend_threshold_k, flatten_threshold_k)[-1]: return "down" return None def detect_acceleration_signal(series: List[float], accel_days: int = 3, accel_threshold_k: float = 1.5) -> Optional[str]: if len(series) < accel_days + 20: return None if _build_acceleration_signal(series, "up", accel_days, accel_threshold_k)[-1]: return "up" if _build_acceleration_signal(series, "down", accel_days, accel_threshold_k)[-1]: return "down" return None def detect_band_cross_signal(primary: List[float], secondary: List[float]) -> Optional[str]: if len(primary) < 2 or len(secondary) < 2: return None if _build_band_cross_signal(primary, secondary, "up")[-1]: return "up" if _build_band_cross_signal(primary, secondary, "down")[-1]: return "down" return None def _technical_desk_wavelet_config() -> Dict: from services.database import get_ai_desk_by_type desk = get_ai_desk_by_type("technical") or {} return (desk.get("config") or {}).get("signals") or {} def scan_watchlist_wavelet_signals() -> List[Dict]: """Compute a causal (no-look-ahead) band decomposition for each watchlist instrument. Every (ticker, band) gets a row every cycle — current slope/ value/energy state always, plus signal_kind/direction/params_json when one of the enabled trigger kinds fires on the most recent point (first match wins, evaluated extremum -> level_threshold -> trend_flatten -> acceleration -> band_cross -> energy_threshold). ridge_shift is evaluated once per ticker (not per band — the ridge is a single track for the whole decomposition) and stored as an extra band_label="ridge" row.""" 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 sig_cfg = _technical_desk_wavelet_config() engine_cfg = sig_cfg.get("wavelet_engine") or {} if not engine_cfg.get("enabled", True): return [] num_levels = int(engine_cfg.get("num_levels", 4)) wavelet = engine_cfg.get("wavelet", "gmw") method = engine_cfg.get("method", "cwt") lookback = int(engine_cfg.get("lookback_days", 120)) extremum_cfg = sig_cfg.get("wavelet_extremum") or {"enabled": True} level_cfg = sig_cfg.get("wavelet_level_threshold") or {"enabled": True, "threshold_k": 2.0} trend_cfg = sig_cfg.get("wavelet_trend_flatten") or {"enabled": False} accel_cfg = sig_cfg.get("wavelet_acceleration") or {"enabled": False} cross_cfg = sig_cfg.get("wavelet_band_cross") or {"enabled": False} ridge_cfg = sig_cfg.get("wavelet_ridge_shift") or {"enabled": False} energy_cfg = sig_cfg.get("wavelet_energy_threshold") or {"enabled": False} 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] bands = decomposed["bands"] for i, band in enumerate(bands): series = band["series"] if not series: continue slope = _compute_slope(series) energy = band.get("energy") kind: Optional[str] = None direction: Optional[str] = None params: Optional[Dict] = None if extremum_cfg.get("enabled", True): direction = detect_extremum_signal(series) kind = "extremum" if direction else None if not direction and level_cfg.get("enabled", True): threshold_k = level_cfg.get("threshold_k", 2.0) direction = detect_level_threshold_signal(series, threshold_k) if direction: kind, params = "level_threshold", {"threshold_k": threshold_k} if not direction and trend_cfg.get("enabled"): direction = detect_trend_flatten_signal( series, trend_cfg.get("trend_days", 10), trend_cfg.get("flatten_days", 5), trend_cfg.get("trend_threshold_k", 1.0), trend_cfg.get("flatten_threshold_k", 0.3), ) if direction: kind = "trend_flatten" params = {k: trend_cfg.get(k) for k in ("trend_days", "flatten_days", "trend_threshold_k", "flatten_threshold_k")} if not direction and accel_cfg.get("enabled"): accel_days = accel_cfg.get("accel_days", 3) accel_threshold_k = accel_cfg.get("accel_threshold_k", 1.5) direction = detect_acceleration_signal(series, accel_days, accel_threshold_k) if direction: kind, params = "acceleration", {"accel_days": accel_days, "accel_threshold_k": accel_threshold_k} if not direction and cross_cfg.get("enabled"): sec_idx = int(cross_cfg.get("secondary_band", 1)) if 0 <= sec_idx < len(bands) and sec_idx != i: direction = detect_band_cross_signal(series, bands[sec_idx]["series"]) if direction: kind, params = "band_cross", {"secondary_band": sec_idx} if not direction and energy_cfg.get("enabled") and energy: threshold_k = energy_cfg.get("threshold_k", 2.0) direction = detect_level_threshold_signal(energy, threshold_k) if direction: kind, params = "energy_threshold", {"threshold_k": threshold_k} 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, "slope": slope[-1], "value": series[-1], "energy": energy[-1] if energy else None, "ridge_period_days": None, "params_json": json.dumps(params) if params else None, }) # Ridge — one row per ticker (ssq only), not per band if method == "ssq" and decomposed.get("ridge_period_days"): ridge_series = [v for v in decomposed["ridge_period_days"] if v is not None] if ridge_series: ridge_kind = None ridge_direction = None ridge_params = None if ridge_cfg.get("enabled"): threshold_k = ridge_cfg.get("threshold_k", 2.0) ridge_direction = detect_level_threshold_signal(ridge_series, threshold_k) if ridge_direction: ridge_kind, ridge_params = "ridge_shift", {"threshold_k": threshold_k} results.append({ "ticker": ticker, "band_label": "ridge", "period_low_days": None, "period_high_days": None, "signal_kind": ridge_kind, "direction": ridge_direction, "price_at_signal": price_at_signal, "slope": None, "value": None, "energy": None, "ridge_period_days": ridge_series[-1], "params_json": json.dumps(ridge_params) if ridge_params else None, }) 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