jackyu66gitandClaude
71951019fb
Phase A-C complete: 4 core factors, regime detection, signal tracking, Bayesian expectancy. chanmacro/ (32 files, ~4000 lines): - models: 12 enums + 15 Pydantic v2 models (DateAwareModel, MarketStateVector, etc.) - fetchers: OHLCV + Breadth (from data_provider) + Derivatives (new endpoint) - scoring: Price Structure / Breadth (quantile buckets) / OI Matrix (5 discrete states) / Volatility Regime - regime_detector: 3-state (TREND/RANGE/PANIC), factor-locked (Price+Breadth+Vol), versioned, 2-day confirmation - expectancy: SignalTracker (record+outcomes), TimeDecay (half-life=180d), BayesianExpectancyEngine (Empirical Bayes, Leveled, SufficiencyGuard) - validation: FactorValidator (IC/ICIR/Hit Ratio), RegimeValidator (MI/KL/ANOVA), TransitionValidator (stability) - CLI: fetch|score|regime|track|backfill|expectancy|validate|serve - tests: 52 passing (models, scoring, regime, expectancy) data_provider: - /api/derivatives endpoint: funding rate, OI, OI change, basis - _derivatives storage: same persist pattern as K-line (merge→lock→snapshot→atomic write) - background refresh every 60s Co-Authored-By: Claude <noreply@anthropic.com>
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)
|