admin/OpenFin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: wavelets

Browse Source
This commit is contained in:
OpenSquared
2026-07-19 12:55:45 +02:00
parent e6fb404062
commit 4154848fd3
2 changed files with 50 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
backend/services/wavelet_engine.py
View File

@@ -4,12 +4,15 @@ c:\\DataS\\InstrumentSimulator\\backend\\app\\wavelet.py (project "Macro Causal
No project-specific dependencies — pure numpy/ssqueezepy, safe to call from any service.
"""
import logging
import warnings
import numpy as np
from ssqueezepy import cwt, icwt
from ssqueezepy.utils.cwt_utils import center_frequency
logger = logging.getLogger(__name__)
MIN_SCALES_PER_BAND = 4
# Fixed period (days) lower-bounds for bands 0..5, independent of the analysis
@@ -72,6 +75,7 @@ def band_decompose(
mask = (log_scales >= lo) & (log_scales <= hi if i == num_levels - 1 else log_scales < hi)
scales_band = scales[mask]
reconstruction_failed = False
if len(scales_band) >= MIN_SCALES_PER_BAND:
try:
recon = icwt(Wx[mask, :], wavelet=wavelet, scales=scales_band, x_len=n)
@@ -79,14 +83,26 @@ def band_decompose(
# 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.
# Treat as a negligible band instead of failing the whole request —
# but log it (previously silently swallowed) and flag it, since a flat
# zero band otherwise looks identical to "genuinely no energy here".
logger.warning(
"band_decompose: icwt failed for band %d (n=%d scales, wavelet=%s) — "
"falling back to a zero band", i, len(scales_band), wavelet, exc_info=True,
)
recon = np.zeros(n)
reconstruction_failed = True
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.
logger.warning(
"band_decompose: only %d scale(s) in band %d (< %d minimum) — falling back to a zero band",
len(scales_band), i, MIN_SCALES_PER_BAND,
)
recon = np.zeros(n)
reconstruction_failed = True
scale_lo, scale_hi = float(np.exp(lo)), float(np.exp(hi))
freq_at_scale_hi = center_frequency(wavelet, scale=scale_hi, N=n)
@@ -105,6 +121,7 @@ def band_decompose(
"period_low_days": period_low_days,
"period_high_days": period_high_days,
"series": [round(float(v), 6) for v in recon],
"reconstruction_failed": reconstruction_failed,
}
)
@@ -148,6 +165,7 @@ def windowed_band_decompose(
all_residual: list[float] = []
band_series: list[list[float]] = [[] for _ in range(num_levels)]
band_meta: list[dict | None] = [None] * num_levels
band_failed: list[bool] = [False] * num_levels
chunk_count = 0
start = 0
@@ -164,6 +182,10 @@ def windowed_band_decompose(
for band in chunk["bands"]:
band_series[band["index"]].extend(band["series"])
band_meta[band["index"]] = band
# OR across chunks: if any slice couldn't reconstruct this band, flag the
# whole concatenated series as unreliable rather than losing that signal
# when a later chunk happens to succeed and overwrites band_meta.
band_failed[band["index"]] = band_failed[band["index"]] or band.get("reconstruction_failed", False)
start = end
bands = []
@@ -176,6 +198,7 @@ def windowed_band_decompose(
"period_low_days": meta["period_low_days"],
"period_high_days": meta["period_high_days"],
"series": band_series[i],
"reconstruction_failed": band_failed[i],
}
)
@@ -216,6 +239,7 @@ def rolling_causal_bands(
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)]
band_failed = [False] * num_levels
last_tip: list[float] | None = None
recomputations = 0
@@ -229,6 +253,8 @@ def rolling_causal_bands(
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"]]
for i, band in enumerate(result["bands"]):
band_failed[i] = band_failed[i] or band.get("reconstruction_failed", False)
recomputations += 1
out_dates.append(dates[t])
out_original.append(values[t])
@@ -242,6 +268,7 @@ def rolling_causal_bands(
"period_low_days": None,
"period_high_days": None,
"series": band_series[i],
"reconstruction_failed": band_failed[i],
}
for i in range(num_levels)
]
@@ -324,6 +351,7 @@ def band_decompose_ssq(
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)
reconstruction_failed = False
if mask.any():
try:
Tx_band = np.zeros_like(Tx)
@@ -333,12 +361,20 @@ def band_decompose_ssq(
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.
# case fail the whole request — but log it and flag it (see
# band_decompose's identical fix for why this used to be silent).
logger.warning(
"band_decompose_ssq: issq_cwt failed for band %d (wavelet=%s) — "
"falling back to a zero band", i, wavelet, exc_info=True,
)
recon = np.zeros(n)
reconstruction_failed = True
energy = np.sum(np.abs(Tx[mask, :]) ** 2, axis=0)
else:
logger.warning("band_decompose_ssq: no frequencies fall in band %d — falling back to a zero band", i)
recon = np.zeros(n)
energy = np.zeros(n)
reconstruction_failed = True
period_low_days = round(float(lo_period), 1)
period_high_days = round(float(hi_period), 1)
@@ -350,6 +386,7 @@ def band_decompose_ssq(
"period_high_days": period_high_days,
"series": [round(float(v), 6) for v in recon],
"energy": [round(float(v), 8) for v in energy],
"reconstruction_failed": reconstruction_failed,
}
)
@@ -397,6 +434,7 @@ def rolling_causal_bands_ssq(
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)]
band_failed = [False] * num_levels
last_tip: list[float] | None = None
last_energy_tip: list[float] | None = None
last_ridge_tip: float | None = None
@@ -414,6 +452,8 @@ def rolling_causal_bands_ssq(
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"]]
for i, band in enumerate(result["bands"]):
band_failed[i] = band_failed[i] or band.get("reconstruction_failed", False)
recomputations += 1
out_dates.append(dates[t])
out_original.append(values[t])
@@ -430,6 +470,7 @@ def rolling_causal_bands_ssq(
"period_high_days": None,
"series": band_series[i],
"energy": energy_series[i],
"reconstruction_failed": band_failed[i],
}
for i in range(num_levels)
]