docs: update license header in compactts/agent/meta_router.py to BSL-1.1

d78f825 verified about 19 hours ago

3.25 kB

"""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"

	"""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"