import math from dataclasses import asdict, dataclass from datetime import date, timedelta from statistics import pstdev from typing import Iterable @dataclass(frozen=True) class VolatilityResult: annualized_volatility: float observations: int return_observations: int start_date: str end_date: str years_covered: float requested_years: int annualization_periods: float used_partial_history: bool def to_dict(self) -> dict: return asdict(self) def calculate_annualized_volatility( values: Iterable[tuple[date, float]], *, requested_years: int = 5, ) -> VolatilityResult | None: """Calculate annualized volatility from the available history up to requested_years. The annualization factor is inferred from the average calendar gap between observations, so sparse fund NAV histories do not get treated like daily data. """ ordered = sorted((d, float(v)) for d, v in values if d and v and float(v) > 0) if len(ordered) < 3: return None latest_date = ordered[-1][0] cutoff = latest_date - timedelta(days=round(365.25 * requested_years)) window = [(d, v) for d, v in ordered if d >= cutoff] if len(window) < 3: window = ordered log_returns = [ math.log(current / previous) for (_, previous), (_, current) in zip(window, window[1:]) if previous > 0 and current > 0 ] if len(log_returns) < 2: return None start_date = window[0][0] end_date = window[-1][0] span_days = max((end_date - start_date).days, 1) average_gap_days = span_days / max(len(window) - 1, 1) annualization_periods = 365.25 / max(average_gap_days, 1 / 365.25) annualized = pstdev(log_returns) * math.sqrt(annualization_periods) return VolatilityResult( annualized_volatility=annualized, observations=len(window), return_observations=len(log_returns), start_date=start_date.isoformat(), end_date=end_date.isoformat(), years_covered=span_days / 365.25, requested_years=requested_years, annualization_periods=annualization_periods, used_partial_history=(span_days / 365.25) < requested_years * 0.95, )