56 lines
2.0 KiB
Python
56 lines
2.0 KiB
Python
"""
|
|
expectancy/decay.py — Time-weighted sample decay.
|
|
|
|
2024 market structure ≠ 2026 market structure.
|
|
Recent samples get higher weight via exponential decay.
|
|
"""
|
|
|
|
from datetime import date as Date
|
|
from typing import Optional
|
|
import numpy as np
|
|
|
|
|
|
class TimeDecay:
|
|
"""Exponential time decay for sample weighting."""
|
|
|
|
def __init__(self, half_life_days: int = 180):
|
|
self.half_life = half_life_days
|
|
self._decay_rate = np.log(2) / half_life_days
|
|
|
|
def weight(self, sample_date: Date, reference_date: Optional[Date] = None) -> float:
|
|
"""
|
|
Compute decay weight for a sample.
|
|
weight = exp(-days_ago * decay_rate)
|
|
"""
|
|
if reference_date is None:
|
|
reference_date = Date.today()
|
|
days = (reference_date - sample_date).days
|
|
return np.exp(-days * self._decay_rate)
|
|
|
|
def weights(self, dates: list[Date], reference_date: Optional[Date] = None) -> np.ndarray:
|
|
"""Compute decay weights for a list of dates."""
|
|
return np.array([self.weight(d, reference_date) for d in dates])
|
|
|
|
def weighted_win_rate(self, wins: np.ndarray, weights: np.ndarray) -> float:
|
|
"""Weighted win rate: sum(wins * weights) / sum(weights)."""
|
|
total_weight = weights.sum()
|
|
if total_weight == 0:
|
|
return 0.0
|
|
return float((wins * weights).sum() / total_weight)
|
|
|
|
def weighted_mean(self, values: np.ndarray, weights: np.ndarray) -> float:
|
|
"""Weighted mean."""
|
|
total_weight = weights.sum()
|
|
if total_weight == 0:
|
|
return 0.0
|
|
return float((values * weights).sum() / total_weight)
|
|
|
|
def effective_samples(self, weights: np.ndarray) -> float:
|
|
"""Effective number of samples after decay weighting."""
|
|
return float(weights.sum())
|
|
|
|
@staticmethod
|
|
def weight_at_age(days_ago: int, half_life_days: int = 180) -> float:
|
|
"""Quick weight lookup for a given age in days."""
|
|
return np.exp(-days_ago * np.log(2) / half_life_days)
|