444 lines
20 KiB
Python
444 lines
20 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 / 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
|
|
import logging
|
|
from typing import Dict, List, Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Watchlist ticker -> a real yfinance symbol, for the fallback fetch when this instrument
|
|
# has no saxo_quote_symbol link (or its Saxo fetch fails) — mirrors
|
|
# instrument_service.py's _WAVELET_UNDERLYING_ALIASES (reverse direction: that one maps a
|
|
# yfinance/futures id back to the Watchlist ticker for cache lookups, this one maps the
|
|
# Watchlist ticker forward to a fetchable yfinance symbol).
|
|
_FRIENDLY_TO_YFINANCE = {
|
|
"GOLD": "GC=F", "SILVER": "SI=F", "COPPER": "HG=F", "PLATINUM": "PL=F",
|
|
"CRUDE": "CL=F", "BRENT": "BZ=F", "NATGAS": "NG=F",
|
|
"WHEAT": "ZW=F", "CORN": "ZC=F", "SOYBEANS": "ZS=F",
|
|
"SP500": "^GSPC", "NASDAQ": "^NDX", "DOW": "^DJI", "RUSSELL2000": "^RUT",
|
|
}
|
|
|
|
|
|
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 _fetch_close_series(ticker: str, saxo_symbol: Optional[str]):
|
|
"""Saxo-first when this watchlist instrument has a saxo_quote_symbol link, yfinance
|
|
otherwise or as a silent fallback on any Saxo failure — same pattern as
|
|
routers/wavelet.py's _fetch_history, duplicated locally rather than importing across a
|
|
router boundary. Needed because several Watchlist instruments (BRENT, COPPER...) have no
|
|
real yfinance ticker at all — get_historical(ticker, ...) always failed for them, which
|
|
silently dropped them out of the per-cycle scan entirely (one bad ticker just gets
|
|
skipped, see the try/except around the caller) — that's why the Wavelets Signal card
|
|
only ever showed the yfinance-recognized subset of the Watchlist."""
|
|
if saxo_symbol:
|
|
try:
|
|
from services.database import get_saxo_catalog_by_symbol
|
|
from services.saxo_client import get_price_history
|
|
entry = get_saxo_catalog_by_symbol(saxo_symbol)
|
|
asset_type = entry["asset_type"] if entry else "FxSpot"
|
|
bars = get_price_history(saxo_symbol, asset_type, days=400)
|
|
return [b["close"] for b in bars], [b["date"] for b in bars]
|
|
except Exception as e:
|
|
logger.warning(f"[wavelet_signals] Saxo fetch failed for '{saxo_symbol}', falling back to yfinance: {e}")
|
|
|
|
from services.data_fetcher import get_historical
|
|
yf_ticker = ticker.upper()
|
|
if yf_ticker in _FRIENDLY_TO_YFINANCE:
|
|
# GOLD/CRUDE/BRENT/SP500... aren't real yfinance symbols either — without this, any
|
|
# Saxo-linked commodity/index whose Saxo fetch fails falls through to a yfinance call
|
|
# that's guaranteed to return nothing, permanently keeping it out of the wavelet
|
|
# cache no matter how many refreshes run.
|
|
yf_ticker = _FRIENDLY_TO_YFINANCE[yf_ticker]
|
|
elif len(yf_ticker) == 6 and yf_ticker.isalpha():
|
|
# Bare 6-letter FX pairs (EURUSD, GBPUSD...) are a common Watchlist ticker
|
|
# convention here but not a real yfinance symbol (needs the "=X" suffix).
|
|
yf_ticker += "=X"
|
|
hist = get_historical(yf_ticker, period="1y", interval="1d")
|
|
return [h["close"] for h in hist], [h["date"] for h in hist]
|
|
|
|
|
|
def scan_watchlist_wavelet_signals(run_id: Optional[str] = None) -> 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.
|
|
|
|
Also caches the full (untruncated) decomposition per ticker in
|
|
wavelet_decomposition_cache — Instrument Analysis's Wavelet tab reads that instead of
|
|
running its own live decomposition on open, so it always agrees with the Watchlist
|
|
Signal card and never needs a click just to show the current state."""
|
|
from services.database import get_instruments_watchlist, save_wavelet_decomposition_cache
|
|
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:
|
|
values, dates = _fetch_close_series(ticker, item.get("saxo_quote_symbol"))
|
|
if len(values) < lookback + 32:
|
|
logger.warning(
|
|
f"[wavelet_signals] Skipping '{ticker}': only {len(values)} price points "
|
|
f"fetched, need >= {lookback + 32} (lookback={lookback}) — no cache written this pass."
|
|
)
|
|
continue
|
|
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"]:
|
|
logger.warning(f"[wavelet_signals] Skipping '{ticker}': decomposition produced no output dates — no cache written this pass.")
|
|
continue
|
|
save_wavelet_decomposition_cache(ticker, run_id, method, wavelet, num_levels, lookback, decomposed)
|
|
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 as e:
|
|
logger.warning(f"[wavelet_signals] Skipping '{ticker}': {e}") # one bad ticker must not abort the whole scan
|
|
continue
|
|
|
|
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(run_id)
|
|
save_wavelet_signals(run_id, results)
|
|
return results
|