Upload compactts/agent/profiler.py with huggingface_hub

compactts/agent/profiler.py

@@ -0,0 +1,102 @@
+"""Series Profiler for CompactTS-Agent.
+
+Extracts statistical signatures, spectral characteristics, seasonality,
+trend strength, and volatility profiles from time-series trajectories.
+"""
+
+from typing import Dict, Any, Optional
+import numpy as np
+import pandas as pd
+from scipy import stats
+from gluonts.time_feature import get_seasonality
+
+
+class SeriesProfiler:
+    """Profiles univariate and multivariate time series for autonomous agent arbitration."""
+
+    def __init__(self, freq: str = "H"):
+        self.freq = freq
+        try:
+            self.base_seasonality = max(get_seasonality(freq), 1)
+        except Exception:
+            self.base_seasonality = 1
+
+    def profile(self, series: np.ndarray) -> Dict[str, Any]:
+        """Extract comprehensive statistical and structural characteristics.
+
+        Parameters
+        ----------
+        series : np.ndarray
+            1D array of historical observations.
+
+        Returns
+        -------
+        Dict[str, Any]
+            Dictionary of structural features:
+            - length: sequence length
+            - mean, std: empirical scale
+            - seasonal_period: detected dominant lag
+            - seasonal_strength: autocorrelation at seasonal lag
+            - trend_strength: relative trend variance ratio
+            - kurtosis: tail weight indicator
+            - has_missing: boolean indicating gaps or NaNs
+        """
+        y = np.asarray(series, dtype=np.float32)
+        if y.ndim > 1:
+            y = y.squeeze()
+
+        has_missing = bool(np.isnan(y).any() or np.isinf(y).any())
+        if has_missing:
+            s = pd.Series(y).ffill().bfill().to_numpy()
+            y = np.nan_to_num(s, nan=0.0).astype(np.float32)
+
+        n = len(y)
+        if n < 4:
+            return {
+                "length": n,
+                "mean": float(np.mean(y)) if n > 0 else 0.0,
+                "std": float(np.std(y)) if n > 0 else 1.0,
+                "seasonal_period": self.base_seasonality,
+                "seasonal_strength": 0.0,
+                "trend_strength": 0.0,
+                "kurtosis": 0.0,
+                "has_missing": has_missing,
+            }
+
+        mean_val = float(np.mean(y))
+        std_val = float(np.std(y))
+        if std_val < 1e-8:
+            std_val = 1.0
+
+        # Kurtosis for heavy-tail detection
+        try:
+            kurt = float(stats.kurtosis(y))
+        except Exception:
+            kurt = 0.0
+
+        # Seasonal autocorrelation
+        S = self.base_seasonality
+        s_corr = 0.0
+        if n > 2 * S:
+            try:
+                c = np.corrcoef(y[S:], y[:-S])[0, 1]
+                if np.isfinite(c):
+                    s_corr = max(0.0, float(c))
+            except Exception:
+                s_corr = 0.0
+
+        # Trend strength: ratio of diff variance to raw variance
+        diff_var = float(np.var(np.diff(y)))
+        raw_var = float(np.var(y))
+        trend_ratio = max(0.0, 1.0 - (diff_var / (raw_var + 1e-8)))
+
+        return {
+            "length": n,
+            "mean": mean_val,
+            "std": std_val,
+            "seasonal_period": S,
+            "seasonal_strength": s_corr,
+            "trend_strength": trend_ratio,
+            "kurtosis": kurt,
+            "has_missing": has_missing,
+        }