Download src/transformations/wavelets/stats.py from ResearchEngineering/financial_news_bot: direct link, hf CLI and curl.
- Browser
- Download file 4.2 kB
-
https://huggingface.co/spaces/ResearchEngineering/financial_news_bot/resolve/main/src/transformations/wavelets/stats.py
- Command line
-
hf download hf://spaces/ResearchEngineering/financial_news_bot/src/transformations/wavelets/stats.py
-
curl -L -o stats.py https://huggingface.co/spaces/ResearchEngineering/financial_news_bot/resolve/main/src/transformations/wavelets/stats.py
4.2 kB
| """Performance statistics for wavelet strategy backtests. | |
| Stats dataclass mirrors what an allocator would ask for: | |
| CAGR, Sharpe, Sortino, Max DD, Calmar, hit rate, turnover, correlation. | |
| """ | |
| from __future__ import annotations | |
| from dataclasses import dataclass, field | |
| from typing import Optional | |
| import numpy as np | |
| import pandas as pd | |
| class Stats: | |
| name: str | |
| years: float | |
| total_return: float | |
| cagr: float | |
| volatility: float | |
| sharpe: float | |
| sortino: float | |
| max_drawdown: float | |
| calmar: float | |
| hit_rate: float | |
| annual_turnover: float | |
| time_in_market: float | |
| avg_gross_leverage: float | |
| correlation_to_spy: Optional[float] = None | |
| extra: dict = field(default_factory=dict) | |
| def as_dict(self) -> dict: | |
| return { | |
| "Name": self.name, | |
| "Years": f"{self.years:.1f}", | |
| "Total Return": f"{self.total_return:.1%}", | |
| "CAGR": f"{self.cagr:.2%}", | |
| "Volatility": f"{self.volatility:.2%}", | |
| "Sharpe": f"{self.sharpe:.3f}", | |
| "Sortino": f"{self.sortino:.3f}", | |
| "Max Drawdown": f"{self.max_drawdown:.2%}", | |
| "Calmar": f"{self.calmar:.3f}", | |
| "Hit Rate": f"{self.hit_rate:.1%}", | |
| "Annual Turnover": f"{self.annual_turnover:.1f}×", | |
| "Time in Market": f"{self.time_in_market:.1%}", | |
| "Avg Gross Leverage": f"{self.avg_gross_leverage:.2f}×", | |
| "Corr to SPY": f"{self.correlation_to_spy:.2f}" if self.correlation_to_spy is not None else "—", | |
| } | |
| def perf( | |
| returns: pd.Series, | |
| positions: pd.Series, | |
| name: str = "Strategy", | |
| benchmark: Optional[pd.Series] = None, | |
| ann_factor: int = 252, | |
| ) -> Stats: | |
| """Compute full performance statistics from a daily returns series. | |
| Args: | |
| returns: daily P&L series (after costs), aligned with calendar | |
| positions: daily position series (gross leverage proxy) | |
| name: strategy label | |
| benchmark: optional benchmark return series for correlation | |
| ann_factor: trading days per year (default 252) | |
| Returns: | |
| Stats dataclass with all metrics. | |
| """ | |
| rets = returns.dropna() | |
| if len(rets) == 0: | |
| return Stats( | |
| name=name, years=0, total_return=0, cagr=0, volatility=0, | |
| sharpe=0, sortino=0, max_drawdown=0, calmar=0, | |
| hit_rate=0, annual_turnover=0, time_in_market=0, avg_gross_leverage=0, | |
| ) | |
| years = len(rets) / ann_factor | |
| total_return = (1 + rets).prod() - 1 | |
| cagr = (1 + total_return) ** (1 / years) - 1 if years > 0 else 0.0 | |
| vol = rets.std() * np.sqrt(ann_factor) | |
| sharpe = (rets.mean() * ann_factor) / vol if vol > 0 else 0.0 | |
| downside = rets[rets < 0] | |
| downside_vol = downside.std() * np.sqrt(ann_factor) if len(downside) > 0 else 1e-9 | |
| sortino = (rets.mean() * ann_factor) / downside_vol if downside_vol > 0 else 0.0 | |
| cumulative = (1 + rets).cumprod() | |
| rolling_max = cumulative.cummax() | |
| drawdowns = cumulative / rolling_max - 1 | |
| max_dd = float(drawdowns.min()) | |
| calmar = cagr / abs(max_dd) if max_dd != 0 else 0.0 | |
| hit_rate = float((rets > 0).mean()) | |
| pos_aligned = positions.reindex(rets.index).fillna(0) | |
| turnover = pos_aligned.diff().abs().sum() / years if years > 0 else 0.0 | |
| time_in_market = float((pos_aligned.abs() > 0.01).mean()) | |
| avg_gross_leverage = float(pos_aligned.abs().mean()) | |
| corr = None | |
| if benchmark is not None: | |
| bm = benchmark.reindex(rets.index).dropna() | |
| common = rets.reindex(bm.index).dropna() | |
| bm_aligned = bm.reindex(common.index).dropna() | |
| common = common.reindex(bm_aligned.index) | |
| if len(common) > 30: | |
| corr = float(common.corr(bm_aligned)) | |
| return Stats( | |
| name=name, | |
| years=years, | |
| total_return=total_return, | |
| cagr=cagr, | |
| volatility=vol, | |
| sharpe=sharpe, | |
| sortino=sortino, | |
| max_drawdown=max_dd, | |
| calmar=calmar, | |
| hit_rate=hit_rate, | |
| annual_turnover=turnover, | |
| time_in_market=time_in_market, | |
| avg_gross_leverage=avg_gross_leverage, | |
| correlation_to_spy=corr, | |
| ) | |