Upload folder using huggingface_hub

d53dc44 verified about 2 months ago

6.42 kB

"""
ml/adaptive.py — Online / adaptive learning module.

Prevents concept drift by:
  1. Rolling retraining every N days with expanding+rolling window hybrid
  2. Recency-weighted sampling (newer data weighted higher)
  3. Performance-triggered emergency retraining
  4. Exponential decay weighting on training samples
"""
from __future__ import annotations

import logging
from dataclasses import dataclass, field

import numpy as np

logger = logging.getLogger("trading_system.ml.adaptive")


@dataclass
class RetrainDecision:
    """Encapsulates whether and why to retrain."""
    should_retrain: bool
    reason: str
    bars_since_last: int
    performance_trigger: bool = False
    staleness_trigger: bool = False


@dataclass
class AdaptiveConfig:
    """Configuration for adaptive learning."""
    retrain_interval_bars: int = 200      # retrain every N bars
    min_train_samples: int = 1000         # minimum samples to train on
    max_train_samples: int = 10_000       # cap to prevent slow training
    recency_halflife: float = 500.0       # half-life for exponential decay weights
    expanding_floor_pct: float = 0.50     # minimum fraction of total history to keep
    perf_lookback: int = 100              # bars to check recent performance
    perf_sharpe_floor: float = 0.0        # retrain if rolling Sharpe drops below
    perf_winrate_floor: float = 0.40      # retrain if win rate drops below
    gap_bars: int = 78                    # purge gap between train/test (1 trading day)


class AdaptiveLearner:
    """Manages when and how to retrain ML models adaptively.

    Uses an expanding+rolling hybrid window:
      - Always include at least `expanding_floor_pct` of all history
      - Within window, apply exponential recency weighting
      - Retrain triggered by staleness or performance decay
    """

    def __init__(self, config: AdaptiveConfig | None = None):
        self.config = config or AdaptiveConfig()
        self._last_retrain_idx: int = 0
        self._retrain_count: int = 0
        self._performance_history: list[float] = []
        self._recent_returns: list[float] = []

    def check_retrain(self, current_idx: int) -> RetrainDecision:
        """Decide if model should be retrained at current bar index."""
        bars_since = current_idx - self._last_retrain_idx
        staleness = bars_since >= self.config.retrain_interval_bars

        # Performance trigger
        perf_trigger = False
        if len(self._recent_returns) >= self.config.perf_lookback:
            recent = np.array(self._recent_returns[-self.config.perf_lookback:])
            if len(recent) > 0 and np.std(recent) > 0:
                rolling_sharpe = (np.mean(recent) / np.std(recent)) * np.sqrt(252)
                if rolling_sharpe < self.config.perf_sharpe_floor:
                    perf_trigger = True

            wins = np.sum(recent > 0)
            total = np.sum(recent != 0)
            if total > 20:
                win_rate = wins / total
                if win_rate < self.config.perf_winrate_floor:
                    perf_trigger = True

        should = staleness or perf_trigger

        reason = []
        if staleness:
            reason.append(f"stale ({bars_since} bars)")
        if perf_trigger:
            reason.append("performance decay")
        if not reason:
            reason.append("not due yet")

        return RetrainDecision(
            should_retrain=should,
            reason="; ".join(reason),
            bars_since_last=bars_since,
            performance_trigger=perf_trigger,
            staleness_trigger=staleness,
        )

    def get_train_window(
        self, total_bars: int, current_idx: int,
    ) -> tuple[int, int, np.ndarray]:
        """Compute training window and sample weights.

        Returns:
            (start_idx, end_idx, sample_weights)
            end_idx is exclusive, respects gap_bars before current_idx.
        """
        end_idx = current_idx - self.config.gap_bars
        if end_idx <= 0:
            end_idx = max(1, current_idx // 2)

        # Expanding floor: keep at least N% of all history
        min_start = 0
        floor_samples = int(end_idx * self.config.expanding_floor_pct)
        rolling_start = max(0, end_idx - self.config.max_train_samples)

        # Hybrid: max of expanding floor and rolling cap
        start_idx = min(rolling_start, end_idx - floor_samples)
        start_idx = max(start_idx, 0)

        # Ensure minimum samples
        n_samples = end_idx - start_idx
        if n_samples < self.config.min_train_samples:
            start_idx = max(0, end_idx - self.config.min_train_samples)

        # Compute recency weights (exponential decay)
        n = end_idx - start_idx
        if n <= 0:
            return start_idx, end_idx, np.array([])

        positions = np.arange(n, dtype=float)
        # Recency: most recent sample is at position n-1
        distance_from_end = (n - 1) - positions
        weights = np.exp(-distance_from_end * np.log(2) / self.config.recency_halflife)
        weights /= weights.sum()  # normalize

        return start_idx, end_idx, weights

    def record_retrain(self, idx: int):
        """Mark that a retrain occurred at this index."""
        self._last_retrain_idx = idx
        self._retrain_count += 1

    def record_trade_return(self, ret: float):
        """Record a trade return for performance monitoring."""
        self._recent_returns.append(ret)
        # Keep bounded
        if len(self._recent_returns) > 5000:
            self._recent_returns = self._recent_returns[-3000:]

    @property
    def retrain_count(self) -> int:
        return self._retrain_count


def apply_sample_weights(
    X: np.ndarray, y: np.ndarray, weights: np.ndarray,
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
    """Apply sample weights via resampling for models that don't support weights natively.

    Returns resampled (X, y, uniform_weights) with size = len(X).
    """
    n = len(X)
    if len(weights) != n or n == 0:
        return X, y, np.ones(n) / max(n, 1)

    # Resample proportional to weights
    rng = np.random.default_rng(42)
    indices = rng.choice(n, size=n, replace=True, p=weights)
    return X[indices], y[indices], np.ones(n) / n

	"""
	ml/adaptive.py — Online / adaptive learning module.

	Prevents concept drift by:
	1. Rolling retraining every N days with expanding+rolling window hybrid
	2. Recency-weighted sampling (newer data weighted higher)
	3. Performance-triggered emergency retraining
	4. Exponential decay weighting on training samples
	"""
	from __future__ import annotations

	import logging
	from dataclasses import dataclass, field

	import numpy as np

	logger = logging.getLogger("trading_system.ml.adaptive")


	@dataclass
	class RetrainDecision:
	"""Encapsulates whether and why to retrain."""
	should_retrain: bool
	reason: str
	bars_since_last: int
	performance_trigger: bool = False
	staleness_trigger: bool = False


	@dataclass
	class AdaptiveConfig:
	"""Configuration for adaptive learning."""
	retrain_interval_bars: int = 200 # retrain every N bars
	min_train_samples: int = 1000 # minimum samples to train on
	max_train_samples: int = 10_000 # cap to prevent slow training
	recency_halflife: float = 500.0 # half-life for exponential decay weights
	expanding_floor_pct: float = 0.50 # minimum fraction of total history to keep
	perf_lookback: int = 100 # bars to check recent performance
	perf_sharpe_floor: float = 0.0 # retrain if rolling Sharpe drops below
	perf_winrate_floor: float = 0.40 # retrain if win rate drops below
	gap_bars: int = 78 # purge gap between train/test (1 trading day)


	class AdaptiveLearner:
	"""Manages when and how to retrain ML models adaptively.

	Uses an expanding+rolling hybrid window:
	- Always include at least `expanding_floor_pct` of all history
	- Within window, apply exponential recency weighting
	- Retrain triggered by staleness or performance decay
	"""

	def __init__(self, config: AdaptiveConfig \| None = None):
	self.config = config or AdaptiveConfig()
	self._last_retrain_idx: int = 0
	self._retrain_count: int = 0
	self._performance_history: list[float] = []
	self._recent_returns: list[float] = []

	def check_retrain(self, current_idx: int) -> RetrainDecision:
	"""Decide if model should be retrained at current bar index."""
	bars_since = current_idx - self._last_retrain_idx
	staleness = bars_since >= self.config.retrain_interval_bars

	# Performance trigger
	perf_trigger = False
	if len(self._recent_returns) >= self.config.perf_lookback:
	recent = np.array(self._recent_returns[-self.config.perf_lookback:])
	if len(recent) > 0 and np.std(recent) > 0:
	rolling_sharpe = (np.mean(recent) / np.std(recent)) * np.sqrt(252)
	if rolling_sharpe < self.config.perf_sharpe_floor:
	perf_trigger = True

	wins = np.sum(recent > 0)
	total = np.sum(recent != 0)
	if total > 20:
	win_rate = wins / total
	if win_rate < self.config.perf_winrate_floor:
	perf_trigger = True

	should = staleness or perf_trigger

	reason = []
	if staleness:
	reason.append(f"stale ({bars_since} bars)")
	if perf_trigger:
	reason.append("performance decay")
	if not reason:
	reason.append("not due yet")

	return RetrainDecision(
	should_retrain=should,
	reason="; ".join(reason),
	bars_since_last=bars_since,
	performance_trigger=perf_trigger,
	staleness_trigger=staleness,
	)

	def get_train_window(
	self, total_bars: int, current_idx: int,
	) -> tuple[int, int, np.ndarray]:
	"""Compute training window and sample weights.

	Returns:
	(start_idx, end_idx, sample_weights)
	end_idx is exclusive, respects gap_bars before current_idx.
	"""
	end_idx = current_idx - self.config.gap_bars
	if end_idx <= 0:
	end_idx = max(1, current_idx // 2)

	# Expanding floor: keep at least N% of all history
	min_start = 0
	floor_samples = int(end_idx * self.config.expanding_floor_pct)
	rolling_start = max(0, end_idx - self.config.max_train_samples)

	# Hybrid: max of expanding floor and rolling cap
	start_idx = min(rolling_start, end_idx - floor_samples)
	start_idx = max(start_idx, 0)

	# Ensure minimum samples
	n_samples = end_idx - start_idx
	if n_samples < self.config.min_train_samples:
	start_idx = max(0, end_idx - self.config.min_train_samples)

	# Compute recency weights (exponential decay)
	n = end_idx - start_idx
	if n <= 0:
	return start_idx, end_idx, np.array([])

	positions = np.arange(n, dtype=float)
	# Recency: most recent sample is at position n-1
	distance_from_end = (n - 1) - positions
	weights = np.exp(-distance_from_end * np.log(2) / self.config.recency_halflife)
	weights /= weights.sum() # normalize

	return start_idx, end_idx, weights

	def record_retrain(self, idx: int):
	"""Mark that a retrain occurred at this index."""
	self._last_retrain_idx = idx
	self._retrain_count += 1

	def record_trade_return(self, ret: float):
	"""Record a trade return for performance monitoring."""
	self._recent_returns.append(ret)
	# Keep bounded
	if len(self._recent_returns) > 5000:
	self._recent_returns = self._recent_returns[-3000:]

	@property
	def retrain_count(self) -> int:
	return self._retrain_count


	def apply_sample_weights(
	X: np.ndarray, y: np.ndarray, weights: np.ndarray,
	) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
	"""Apply sample weights via resampling for models that don't support weights natively.

	Returns resampled (X, y, uniform_weights) with size = len(X).
	"""
	n = len(X)
	if len(weights) != n or n == 0:
	return X, y, np.ones(n) / max(n, 1)

	# Resample proportional to weights
	rng = np.random.default_rng(42)
	indices = rng.choice(n, size=n, replace=True, p=weights)
	return X[indices], y[indices], np.ones(n) / n