App/analysis/volatility.py

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,
    )