File size: 7,907 Bytes
6a65e1e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 | """Mathematically Rigorous, Scale-Invariant Time-Series Feature Extractor.
Extracts principled statistical, spectral, temporal, and information-theoretic
signatures from raw time-series trajectories for leak-free autonomous agent dispatching.
Guarantees:
- Strict Scale Invariance: Multiplying y by any scalar c > 0 yields identical routing features.
- Strict Zero Leakage: No dataset names, metadata lookups, or benchmark splits.
- Pure Numerical Input: Operates exclusively on array y in R^T and runtime constraints (H, freq).
Attribution: Daksh Bhalala
License: Business Source License 1.1 (BSL-1.1)
"""
from typing import Dict, Any, Union
import numpy as np
import pandas as pd
from scipy import stats
from scipy.signal import periodogram
class TimeSeriesFeatureExtractor:
"""Scale-invariant, robust feature extractor for autonomous time-series profiling."""
FREQ_TO_SECONDS = {
"10S": 10.0,
"5T": 300.0,
"10T": 600.0,
"15T": 900.0,
"H": 3600.0,
"D": 86400.0,
"W": 604800.0,
"M": 2629746.0,
"Q": 7889238.0,
"A": 31556952.0,
}
FREQ_TO_SEASONALITY = {
"10S": 60,
"5T": 12,
"10T": 6,
"15T": 4,
"H": 24,
"D": 7,
"W": 52,
"M": 12,
"Q": 4,
"A": 1,
}
def __init__(self, freq: str = "H"):
self.freq = freq
self.period = self.FREQ_TO_SEASONALITY.get(freq, 24)
self.sampling_interval_seconds = self.FREQ_TO_SECONDS.get(freq, 3600.0)
def extract(self, series: Union[np.ndarray, list], horizon: int = 24) -> Dict[str, float]:
"""Extract a scale-invariant, comprehensive feature vector.
Parameters
----------
series : Union[np.ndarray, list]
Raw historical trajectory y in R^T.
horizon : int
Forecast horizon H requested by the user.
Returns
-------
Dict[str, float]
Scale-invariant features capturing dynamics, intermittency, shifts, and spectrum.
"""
y = np.asarray(series, dtype=np.float64)
if y.ndim > 1:
y = y.squeeze()
if y.ndim > 1:
# Multivariate: select primary variate
y = y[0] if y.shape[0] < y.shape[1] else y[:, 0]
# Handle missing values and infs cleanly
if np.isnan(y).any() or np.isinf(y).any():
s = pd.Series(y).ffill().bfill().to_numpy()
y = np.nan_to_num(s, nan=0.0).astype(np.float64)
n = len(y)
if n < 4:
return self._fallback_features(n, horizon)
# 1. Physical Support Characteristics
min_val = float(np.min(y))
max_val = float(np.max(y))
is_strictly_positive = 1.0 if min_val >= 0.0 else 0.0
val_range = max_val - min_val
# 2. Intermittency & Sparsity
# Ratio of exact or near-zeros (relative to overall scale)
threshold = 1e-5 if val_range < 1e-4 else 1e-4 * val_range
zero_ratio = float(np.mean(np.abs(y) <= threshold))
# 3. Scale-Normalized Moments
mean_val = float(np.mean(y))
std_val = float(np.std(y))
if std_val < 1e-8:
return self._constant_series_features(n, horizon, is_strictly_positive)
# Z-score normalization for scale invariance
z = (y - mean_val) / std_val
# Higher-order normalized moments
try:
kurt = float(stats.kurtosis(z))
skew = float(stats.skew(z))
except Exception:
kurt, skew = 0.0, 0.0
# 4. Non-Stationarity & Regime Shifts
# Evaluate divergence between historical past and recent window
half = n // 2
past_z = z[:half]
recent_z = z[half:]
mean_shift = abs(float(np.mean(recent_z)) - float(np.mean(past_z)))
std_shift = abs(float(np.std(recent_z)) - float(np.std(past_z)))
# 5. Temporal Autocorrelation & Stationarity
# First lag autocorrelation (persistence)
if n > 2:
try:
ac1 = float(np.corrcoef(z[1:], z[:-1])[0, 1])
if not np.isfinite(ac1):
ac1 = 0.0
except Exception:
ac1 = 0.0
else:
ac1 = 0.0
# Seasonal lag autocorrelation
S = self.period
seasonal_corr = 0.0
if n > 2 * S and S > 1:
try:
ac_s = float(np.corrcoef(z[S:], z[:-S])[0, 1])
if np.isfinite(ac_s):
seasonal_corr = max(0.0, ac_s)
except Exception:
seasonal_corr = 0.0
# 6. Spectral Energy & Seasonality Purity
# Computes power spectrum via periodogram
spectral_entropy = 0.5
dominant_freq_energy_ratio = 0.0
if n >= 16:
try:
_, pxx = periodogram(z)
pxx_sum = np.sum(pxx)
if pxx_sum > 1e-8:
pxx_norm = pxx / pxx_sum
# Shannon entropy of power spectrum (normalized to [0, 1])
pxx_pos = pxx_norm[pxx_norm > 1e-12]
h_spec = -np.sum(pxx_pos * np.log(pxx_pos))
spectral_entropy = float(h_spec / np.log(len(pxx)))
dominant_freq_energy_ratio = float(np.max(pxx_norm))
except Exception:
pass
# 7. ARCH Volatility Clustering (Autocorrelation of squared returns)
diff_z = np.diff(z)
vol_clustering = 0.0
if len(diff_z) > 4:
sq_diff = diff_z**2
sq_std = float(np.std(sq_diff))
if sq_std > 1e-6:
try:
c_arch = float(np.corrcoef(sq_diff[1:], sq_diff[:-1])[0, 1])
if np.isfinite(c_arch):
vol_clustering = max(0.0, c_arch)
except Exception:
pass
# 8. Horizon-to-Context Ratio
horizon_ratio = float(horizon / min(n, 512))
return {
"length": float(n),
"zero_ratio": zero_ratio,
"is_strictly_positive": is_strictly_positive,
"kurtosis": kurt,
"skewness": skew,
"mean_shift": mean_shift,
"std_shift": std_shift,
"autocorr_lag1": ac1,
"seasonal_corr": seasonal_corr,
"spectral_entropy": spectral_entropy,
"dominant_spectral_energy": dominant_freq_energy_ratio,
"volatility_clustering": vol_clustering,
"horizon_ratio": horizon_ratio,
"sampling_seconds_log": float(np.log10(max(self.sampling_interval_seconds, 1.0))),
}
def _fallback_features(self, n: int, horizon: int) -> Dict[str, float]:
return {
"length": float(n),
"zero_ratio": 0.0,
"is_strictly_positive": 1.0,
"kurtosis": 0.0,
"skewness": 0.0,
"mean_shift": 0.0,
"std_shift": 0.0,
"autocorr_lag1": 0.0,
"seasonal_corr": 0.0,
"spectral_entropy": 1.0,
"dominant_spectral_energy": 0.0,
"volatility_clustering": 0.0,
"horizon_ratio": float(horizon / max(n, 1)),
"sampling_seconds_log": float(np.log10(max(self.sampling_interval_seconds, 1.0))),
}
def _constant_series_features(self, n: int, horizon: int, is_pos: float) -> Dict[str, float]:
res = self._fallback_features(n, horizon)
res["is_strictly_positive"] = is_pos
res["autocorr_lag1"] = 1.0
res["spectral_entropy"] = 0.0
return res
|