"""Public-safe return, risk, and cash-benchmark calculations. This module is intentionally standard-library-only so the reviewed methodology bundle can reproduce disclosed calculations without importing the private production repository. Private-repository tests compare the core return/risk calculations against ``backtest.metrics.MetricsCalculator`` to prevent silent methodological drift. The functions here calculate metrics from supplied public inputs. They do not contain strategy logic, broker integration, account data, or execution code. """ from __future__ import annotations from dataclasses import asdict, dataclass import math from typing import Iterable @dataclass(frozen=True) class CoreMetrics: """The public subset of core performance metrics used by the example.""" total_return: float cagr: float volatility: float downside_volatility: float sharpe_ratio: float sharpe_ratio_rf_zero: float sortino_ratio: float max_drawdown: float max_drawdown_duration_days: int avg_drawdown: float positive_periods_pct: float best_period: float worst_period: float def to_dict(self) -> dict[str, float | int]: return asdict(self) @dataclass(frozen=True) class TreasuryBillCashBenchmark: """Simple cash return implied by a Treasury bill purchase price.""" purchase_price_per_100: float maturity_value_per_100: float term_days: int holding_period_return: float simple_annualized_rate: float simple_annualized_rate_pct: float def to_dict(self) -> dict[str, float | int]: return asdict(self) def _equity_curve(returns: list[float]) -> list[float]: wealth = 1.0 curve = [wealth] for period_return in returns: wealth *= 1.0 + period_return curve.append(wealth) return curve def _cagr(total_return: float, years: float) -> float: if years <= 0: return 0.0 return (1.0 + total_return) ** (1.0 / years) - 1.0 def _volatility(returns: list[float], *, annualize: bool = True) -> float: if len(returns) < 2: return 0.0 mean = sum(returns) / len(returns) variance = sum((value - mean) ** 2 for value in returns) / (len(returns) - 1) result = math.sqrt(variance) return result * math.sqrt(12) if annualize else result def _downside_volatility( returns: list[float], *, threshold: float = 0.0, annualize: bool = True, ) -> float: downside = [min(value - threshold, 0.0) for value in returns] if len(downside) < 2: return 0.0 variance = sum(value**2 for value in downside) / len(downside) result = math.sqrt(variance) return result * math.sqrt(12) if annualize else result def _sharpe_ratio(returns: list[float], risk_free_rate: float) -> float: if len(returns) < 2: return 0.0 monthly_rf = (1.0 + risk_free_rate) ** (1.0 / 12.0) - 1.0 excess = [value - monthly_rf for value in returns] mean_excess = sum(excess) / len(excess) volatility = _volatility(excess, annualize=False) if volatility == 0: return 0.0 return (mean_excess * 12.0) / (volatility * math.sqrt(12)) def _sortino_ratio(returns: list[float], risk_free_rate: float) -> float: if len(returns) < 2: return 0.0 monthly_rf = (1.0 + risk_free_rate) ** (1.0 / 12.0) - 1.0 excess = [value - monthly_rf for value in returns] mean_excess = sum(excess) / len(excess) downside = _downside_volatility(excess, threshold=0.0, annualize=False) if downside == 0: return 0.0 return (mean_excess * 12.0) / (downside * math.sqrt(12)) def _max_drawdown(equity_curve: list[float]) -> tuple[float, int]: if len(equity_curve) < 2: return 0.0, 0 peak = equity_curve[0] max_drawdown = 0.0 drawdown_start = 0 max_duration_periods = 0 current_drawdown_start = 0 for index, value in enumerate(equity_curve): if value > peak: peak = value current_drawdown_start = index drawdown = (peak - value) / peak if peak > 0 else 0.0 if drawdown > max_drawdown: max_drawdown = drawdown drawdown_start = current_drawdown_start max_duration_periods = index - drawdown_start return max_drawdown, max_duration_periods def _average_drawdown(equity_curve: list[float]) -> float: if len(equity_curve) < 2: return 0.0 peak = equity_curve[0] drawdowns: list[float] = [] for value in equity_curve: if value > peak: peak = value drawdowns.append((peak - value) / peak if peak > 0 else 0.0) return sum(drawdowns) / len(drawdowns) if drawdowns else 0.0 def compute_core_metrics( returns: Iterable[float], *, years: float, risk_free_rate: float = 0.0, ) -> CoreMetrics: """Compute the disclosed monthly core metrics from supplied period returns.""" values = list(returns) equity_curve = _equity_curve(values) total_return = equity_curve[-1] - 1.0 if equity_curve else 0.0 max_drawdown, max_duration_periods = _max_drawdown(equity_curve) return CoreMetrics( total_return=total_return, cagr=_cagr(total_return, years), volatility=_volatility(values), downside_volatility=_downside_volatility(values), sharpe_ratio=_sharpe_ratio(values, risk_free_rate), sharpe_ratio_rf_zero=_sharpe_ratio(values, 0.0), sortino_ratio=_sortino_ratio(values, risk_free_rate), max_drawdown=max_drawdown, max_drawdown_duration_days=max_duration_periods * 21, avg_drawdown=_average_drawdown(equity_curve), positive_periods_pct=( sum(1 for value in values if value > 0) / len(values) if values else 0.0 ), best_period=max(values) if values else 0.0, worst_period=min(values) if values else 0.0, ) def compute_treasury_bill_cash_benchmark( *, purchase_price_per_100: float, term_days: int, maturity_value_per_100: float = 100.0, days_per_year: int = 365, ) -> TreasuryBillCashBenchmark: """Compute a simple Treasury-bill cash return from published price inputs. This reproduces the equivalent simple annualized investment-rate convention used in the selected public Treasury auction example: ``((maturity / purchase_price) - 1) * (days_per_year / term_days)`` """ if purchase_price_per_100 <= 0: raise ValueError("purchase_price_per_100 must be positive") if maturity_value_per_100 <= 0: raise ValueError("maturity_value_per_100 must be positive") if term_days <= 0: raise ValueError("term_days must be positive") if days_per_year <= 0: raise ValueError("days_per_year must be positive") holding_period_return = maturity_value_per_100 / purchase_price_per_100 - 1.0 simple_annualized_rate = holding_period_return * days_per_year / term_days return TreasuryBillCashBenchmark( purchase_price_per_100=purchase_price_per_100, maturity_value_per_100=maturity_value_per_100, term_days=term_days, holding_period_return=holding_period_return, simple_annualized_rate=simple_annualized_rate, simple_annualized_rate_pct=simple_annualized_rate * 100.0, )