ftdt-quant-lab/common/risk.py

"""
Risk analytics module.

Value at Risk, Conditional VaR, drawdown analysis, correlation,
and composite risk ratios. Builds on common/metrics.py for
Sharpe/Sortino/max_drawdown which are re-exported here.

All functions accept equity curves as either plain lists of floats
or lists of {t: timestamp, v: equity_value} dicts.
"""

import math
import numpy as np


# ═══════════════════════════════════════════════════════════
# Helpers
# ═══════════════════════════════════════════════════════════

def _to_values(curve):
    """Normalise an equity curve to a list of floats."""
    if not curve:
        return []
    if isinstance(curve[0], dict):
        return [p["v"] for p in curve]
    return [float(v) for v in curve]


def _daily_returns(curve):
    """Compute daily log returns from an equity curve."""
    vals = _to_values(curve)
    if len(vals) < 2:
        return []
    return [math.log(vals[i] / vals[i - 1]) for i in range(1, len(vals))]


# ═══════════════════════════════════════════════════════════
# Value at Risk & Conditional VaR
# ═══════════════════════════════════════════════════════════

def var_95(daily_returns):
    """
    95% historical Value at Risk.

    Returns a *positive* number representing the loss threshold
    (e.g. 0.02 means "we are 95% confident daily loss won't exceed 2%").
    """
    if len(daily_returns) < 5:
        return 0.0
    sorted_ret = sorted(daily_returns)
    idx = max(0, int(len(sorted_ret) * 0.05))
    var = sorted_ret[idx]
    return -min(var, 0.0)  # return positive loss magnitude


def cvar_95(daily_returns):
    """
    95% Conditional Value at Risk (Expected Shortfall).

    Average loss *beyond* the VaR threshold. Returns a positive number.
    """
    if len(daily_returns) < 5:
        return 0.0
    sorted_ret = sorted(daily_returns)
    cutoff = max(0, int(len(sorted_ret) * 0.05))
    tail = [r for r in sorted_ret[:cutoff + 1] if r < 0]
    if not tail:
        return 0.0
    return -np.mean(tail)


def var_95_from_equity(equity_curve):
    """Convenience: VaR computed directly from an equity curve."""
    return var_95(_daily_returns(equity_curve))


def cvar_95_from_equity(equity_curve):
    """Convenience: CVaR computed directly from an equity curve."""
    return cvar_95(_daily_returns(equity_curve))


# ═══════════════════════════════════════════════════════════
# Drawdown
# ═══════════════════════════════════════════════════════════

def max_drawdown(equity_curve):
    """
    Maximum drawdown from an equity curve (peak-to-trough).

    Returns a positive fraction (0.25 = 25% max DD).
    Accepts list of floats or list of {t, v} dicts.
    """
    vals = _to_values(equity_curve)
    if not vals:
        return 0.0
    peak = vals[0]
    worst = 0.0
    for v in vals:
        if v > peak:
            peak = v
        if peak > 0:
            dd = (peak - v) / peak
            worst = max(worst, dd)
    return worst


# ═══════════════════════════════════════════════════════════
# Ratios
# ═══════════════════════════════════════════════════════════

def sharpe(returns, rf=0.0, periods=365):
    """
    Annualised Sharpe ratio.

    `returns` should be a list of daily log returns (floats).
    """
    if len(returns) < 2:
        return 0.0
    excess = np.mean(returns) - rf
    std = np.std(returns, ddof=1)
    if std <= 0:
        return 0.0
    return (excess / std) * math.sqrt(periods)


def sortino(returns, rf=0.0, periods=365):
    """
    Annualised Sortino ratio (downside deviation only).
    """
    if len(returns) < 2:
        return 0.0
    excess = np.mean(returns) - rf
    downside = [r for r in returns if r < 0]
    d_std = np.std(downside, ddof=1) if downside else 0.0
    if d_std <= 0:
        return 0.0
    return (excess / d_std) * math.sqrt(periods)


def calmar_ratio(returns, max_dd):
    """
    Calmar ratio = annualised return / maximum drawdown.

    `returns` is a list of daily log returns.
    `max_dd` is a positive fraction (0.25 = 25% drawdown).
    """
    if len(returns) < 2 or max_dd <= 0:
        return 0.0
    ann_return = np.mean(returns) * 365
    return ann_return / max_dd


def sharpe_from_equity(equity_curve):
    """Sharpe ratio computed from an equity curve."""
    return sharpe(_daily_returns(equity_curve))


def sortino_from_equity(equity_curve):
    """Sortino ratio computed from an equity curve."""
    return sortino(_daily_returns(equity_curve))


def calmar_from_equity(equity_curve):
    """Calmar ratio computed from an equity curve."""
    dr = _daily_returns(equity_curve)
    dd = max_drawdown(equity_curve)
    return calmar_ratio(dr, dd)


# ═══════════════════════════════════════════════════════════
# Correlation
# ═══════════════════════════════════════════════════════════

def correlation_matrix(strategy_returns_dict):
    """
    Pearson correlation matrix between strategies.

    Args:
        strategy_returns_dict: {name: [daily_log_returns], ...}

    Returns:
        {name: {name: float, ...}, ...}
        or empty dict if fewer than 2 strategies.
    """
    names = list(strategy_returns_dict.keys())
    if len(names) < 2:
        return {}

    # Align lengths (truncate to shortest)
    min_len = min(len(strategy_returns_dict[n]) for n in names)
    if min_len < 2:
        return {}

    matrix = {}
    for n1 in names:
        r1 = strategy_returns_dict[n1][-min_len:]
        row = {}
        for n2 in names:
            r2 = strategy_returns_dict[n2][-min_len:]
            if n1 == n2:
                row[n2] = 1.0
            else:
                corr = np.corrcoef(r1, r2)[0, 1]
                row[n2] = float(corr) if not np.isnan(corr) else 0.0
        matrix[n1] = row
    return matrix


def correlation_from_equity(strategy_equity_dict):
    """
    Convenience: correlation matrix from {name: [{t,v},...]} equity curves.
    """
    returns_dict = {}
    for name, curve in strategy_equity_dict.items():
        dr = _daily_returns(curve)
        if len(dr) >= 2:
            returns_dict[name] = dr
    return correlation_matrix(returns_dict)


# ═══════════════════════════════════════════════════════════
# Composite risk summary
# ═══════════════════════════════════════════════════════════

def risk_summary(equity_history, strategy_equity=None):
    """
    One-shot: compute all risk metrics for a portfolio.

    Args:
        equity_history: portfolio-level equity curve [{t, v}, ...]
        strategy_equity: optional {name: [{t, v}, ...]}

    Returns dict with VaR, CVaR, MaxDD, Calmar, Sharpe, Sortino,
    and optionally correlation/cross-strategy metrics.
    """
    dr = _daily_returns(equity_history)
    dd = max_drawdown(equity_history)

    summary = {
        "var_95": round(var_95(dr), 6),
        "cvar_95": round(cvar_95(dr), 6),
        "max_drawdown": round(dd, 6),
        "calmar_ratio": round(calmar_ratio(dr, dd), 4),
        "sharpe": round(sharpe(dr), 4),
        "sortino": round(sortino(dr), 4),
        "num_observations": len(dr),
    }

    if strategy_equity and len(strategy_equity) >= 2:
        summary["correlation"] = correlation_from_equity(strategy_equity)

        # Per-strategy metrics
        per_strat = {}
        for name, curve in strategy_equity.items():
            sdr = _daily_returns(curve)
            sdd = max_drawdown(curve)
            per_strat[name] = {
                "sharpe": round(sharpe(sdr), 4),
                "sortino": round(sortino(sdr), 4),
                "max_drawdown": round(sdd, 6),
                "calmar_ratio": round(calmar_ratio(sdr, sdd), 4),
                "var_95": round(var_95(sdr), 6),
                "cvar_95": round(cvar_95(sdr), 6),
            }
        summary["per_strategy"] = per_strat

    return summary