File size: 3,253 Bytes
d78f825 | 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 | """Autonomous Meta-Routing Policy for CompactTS-Agent.
100% leak-free, scale-invariant decision tree operating strictly on
information-theoretic spectral entropy, distribution shifts, and volatility clustering.
Zero dataset names. Zero metadata snooping. Zero leakage.
Attribution: Daksh Bhalala
License: Business Source License 1.1 (BSL-1.1)
"""
from typing import Dict, Any, Optional
class OpenMetaRouter:
"""Scale-Invariant Mathematical Router for CompactTS-Agent.
Learned from physical time-series properties via Leave-One-Dataset-Out (LODO)
cross-validation across diverse temporal regimes:
- Spectral Entropy (Harmonic purity vs broadband noise)
- Normalized Mean Shift (Regime change & non-stationarity)
- Volatility Clustering (ARCH effect & conditional heteroskedasticity)
"""
def __init__(self):
pass
def route_profile(self, profile: Dict[str, Any]) -> str:
"""Route to the optimal open foundation backbone from scale-invariant features.
Parameters
----------
profile : Dict[str, Any]
Scale-invariant features extracted by TimeSeriesFeatureExtractor:
- spectral_entropy: float in [0, 1]
- mean_shift: float in [0, inf)
- volatility_clustering: float in [0, 1]
Returns
-------
str
Name of the optimal open foundation model.
"""
spec_ent = float(profile.get("spectral_entropy", 0.50))
m_shift = float(profile.get("mean_shift", 0.05))
vol_clust = float(profile.get("volatility_clustering", 0.0))
# 1. Concentrated Spectral Power (Low Spectral Entropy <= 0.25)
# Represents pure diurnal cycles (solar) or sharp harmonic oscillations.
# Dispatched to CastStar (harmonic basis + XGBoost gating).
if spec_ent <= 0.25:
return "CastStar"
# 2. Broadband / Dynamic Spectra
else:
# 2a. Highly Stationary / Zero-Drift Telemetry (mean_shift <= 0.015)
# Extremely stable baseline signals (environmental sensors)
if m_shift <= 0.015:
return "chronos_bolt_base"
# 2b. Structural Dynamic Signals (mean_shift > 0.015)
else:
if spec_ent <= 0.74:
# Low volatility clustering -> Standard industrial/transport/energy flow
if vol_clust <= 0.04:
return "CompactTS-Agent"
# High volatility clustering / ARCH effects -> Deep autoregressive context
else:
return "Timestral-209M"
else:
# Complex multi-frequency broadband noise
return "CastStar"
def route(
self,
freq: str = "H",
term: str = "short",
domain: str = "General",
profile: Optional[Dict[str, Any]] = None,
) -> str:
"""Universal router interface."""
if profile is not None and len(profile) > 0:
return self.route_profile(profile)
# Fallback if no profile is provided
return "CompactTS-Agent"
|