138 lines
5.5 KiB
Python
138 lines
5.5 KiB
Python
"""
|
|
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
|