ParallelLLC/algorithmic_trading

1"""Performance and risk metrics.2 3All ratios are computed from *net* per-bar returns and annualised with the4periodicity inferred from the index, so daily / hourly / minute series all get5comparable numbers.6"""7 8from __future__ import annotations9 10from typing import Dict11 12import numpy as np13import pandas as pd14 15__all__ = [16    "infer_periods_per_year",17    "sharpe_ratio",18    "sortino_ratio",19    "max_drawdown",20    "drawdown_series",21    "compute_metrics",22]23 24_SECONDS_PER_YEAR = 365.25 * 24 * 360025_TRADING_DAYS = 25226 27# A return stream with dispersion below this is constant to floating-point28# noise. Without an absolute floor, a flat series divides by ~1e-19 and reports29# a Sharpe of 1e16 -- the exact kind of nonsense number this project exists to30# catch, so it must not originate here.31_DEGENERATE_SD = 1e-1232 33 34def infer_periods_per_year(index: pd.Index) -> int:35    """Guess bars-per-year from an index, defaulting to daily trading bars."""36    if not isinstance(index, pd.DatetimeIndex) or len(index) < 3:37        return _TRADING_DAYS38    nanos = index.to_numpy(dtype="datetime64[ns]").astype("int64")39    deltas = np.diff(nanos) / 1e9  # seconds40    deltas = deltas[deltas > 0]41    if deltas.size == 0:42        return _TRADING_DAYS43    step = float(np.median(deltas))44    if step >= 20 * 3600:  # daily or slower -> use trading-day convention45        days = step / 86400.046        return max(1, int(round(_TRADING_DAYS / max(days / 1.4, 1.0))))47    # Intraday: assume a 6.5h session, 252 days a year.48    bars_per_session = (6.5 * 3600) / step49    return max(1, int(round(bars_per_session * _TRADING_DAYS)))50 51 52def _clean(returns: pd.Series) -> np.ndarray:53    arr = np.asarray(returns, dtype=float)54    return arr[np.isfinite(arr)]55 56 57def sharpe_ratio(returns: pd.Series, periods_per_year: int, rf: float = 0.0) -> float:58    """Annualised Sharpe. ``rf`` is an annual risk-free rate."""59    arr = _clean(returns)60    if arr.size < 2:61        return 0.062    excess = arr - rf / periods_per_year63    sd = excess.std(ddof=1)64    if not np.isfinite(sd) or sd < _DEGENERATE_SD:65        return 0.066    return float(excess.mean() / sd * np.sqrt(periods_per_year))67 68 69def sortino_ratio(returns: pd.Series, periods_per_year: int, rf: float = 0.0) -> float:70    arr = _clean(returns)71    if arr.size < 2:72        return 0.073    excess = arr - rf / periods_per_year74    downside = excess[excess < 0]75    if downside.size == 0:76        return float("inf") if excess.mean() > 0 else 0.077    dd = np.sqrt(np.mean(downside**2))78    if not np.isfinite(dd) or dd < _DEGENERATE_SD:79        return 0.080    return float(excess.mean() / dd * np.sqrt(periods_per_year))81 82 83def drawdown_series(equity: pd.Series) -> pd.Series:84    peak = equity.cummax()85    return equity / peak - 1.086 87 88def max_drawdown(equity: pd.Series) -> float:89    if equity.empty:90        return 0.091    return float(drawdown_series(equity).min())92 93 94def _time_under_water(equity: pd.Series, periods_per_year: int) -> float:95    """Longest stretch below a prior peak, in years."""96    if equity.empty:97        return 0.098    dd = drawdown_series(equity).to_numpy()99    longest = current = 0100    for value in dd:101        current = current + 1 if value < 0 else 0102        longest = max(longest, current)103    return longest / periods_per_year104 105 106def compute_metrics(107    returns: pd.Series,108    equity: pd.Series,109    position: pd.Series | None = None,110    periods_per_year: int | None = None,111    rf: float = 0.0,112) -> Dict[str, float]:113    """Full metric bundle for one equity curve."""114    ppy = periods_per_year or infer_periods_per_year(returns.index)115    arr = _clean(returns)116    n = arr.size117    if n == 0 or equity.empty:118        return {"periods_per_year": float(ppy)}119 120    years = n / ppy121    total_return = float(equity.iloc[-1] / equity.iloc[0] - 1.0)122    cagr = float((equity.iloc[-1] / equity.iloc[0]) ** (1.0 / years) - 1.0) if years > 0 else 0.0123    vol = float(arr.std(ddof=1) * np.sqrt(ppy))124    mdd = max_drawdown(equity)125    sr = sharpe_ratio(returns, ppy, rf)126 127    out: Dict[str, float] = {128        "total_return": total_return,129        "cagr": cagr,130        "ann_vol": vol,131        "sharpe": sr,132        "sortino": sortino_ratio(returns, ppy, rf),133        "calmar": float(cagr / abs(mdd)) if mdd < 0 else 0.0,134        "max_drawdown": mdd,135        "time_under_water_yrs": _time_under_water(equity, ppy),136        "hit_rate": float((arr > 0).mean()),137        "skew": float(pd.Series(arr).skew()) if n > 2 else 0.0,138        "kurtosis": float(pd.Series(arr).kurtosis()) if n > 3 else 0.0,139        "var_95": float(np.percentile(arr, 5)),140        "cvar_95": float(arr[arr <= np.percentile(arr, 5)].mean()) if n > 20 else 0.0,141        "best_bar": float(arr.max()),142        "worst_bar": float(arr.min()),143        "n_bars": float(n),144        "years": float(years),145        "periods_per_year": float(ppy),146    }147 148    if position is not None and not position.empty:149        pos = position.fillna(0.0)150        turnover = pos.diff().abs().fillna(pos.abs().iloc[0] if len(pos) else 0.0)151        out["exposure"] = float(pos.abs().mean())152        out["long_share"] = float((pos > 0).mean())153        out["short_share"] = float((pos < 0).mean())154        out["turnover_ann"] = float(turnover.sum() / years) if years > 0 else 0.0155        # A "trade" is any change in sign or size of exposure.156        out["n_trades"] = float((turnover > 1e-9).sum())157    return out158