utils/metrics.py

import pandas as pd
import numpy as np

def israelsen_sharpe(excess_return, vol):
    """
    Modified Sharpe Ratio by Craig Israelsen.
    Standard Sharpe penalizes higher volatility even when returns are negative.
    This adjustment correctly rewards lower volatility when returns are negative.
    """
    if excess_return == 0 or vol == 0:
        return 0.0
    exponent = excess_return / abs(excess_return)
    return excess_return / (vol ** exponent)

def portfolio_gross_metrics(weights: pd.Series, current_yields: pd.Series = None, durations: pd.Series = None) -> dict:
    """
    Calculates gross leverage, total long exposure, and total short exposure.
    Optionally computes portfolio-level weighted yield and duration if data is provided.
    
    Returns:
        dict: Containing 'gross_lev', 'long_e', 'short_e', 'port_yield', 'port_duration'
    """
    w = weights.drop(labels=['CASH'], errors='ignore')
    long_e = w[w > 0].sum()
    short_e = w[w < 0].sum()
    gross_lev = abs(long_e) + abs(short_e)
    
    port_yield = None
    port_duration = None
    
    if current_yields is not None:
        port_yield = sum(w.get(t, 0) * current_yields.get(t, 0) for t in w.index)
        
    if durations is not None:
        port_duration = sum(w.get(t, 0) * durations.get(t, 0) for t in w.index)
        
    return {
        "gross_lev": float(gross_lev),
        "long_e": float(long_e),
        "short_e": float(short_e),
        "port_yield": float(port_yield) if port_yield is not None else None,
        "port_duration": float(port_duration) if port_duration is not None else None
    }

def liquidity_score(weights, spread_map):
    """Calculates the weighted average bid-ask spread of the portfolio."""
    w = weights.drop(labels=['CASH'], errors='ignore')
    if w.abs().sum() == 0:
        return 0.0
    w_norm = w.abs() / w.abs().sum()
    score = sum(w_norm.get(t, 0) * spread_map.get(t, 0.0008) for t in w.index)
    return score

def annual_returns(daily_returns_series):
    """Aggregates daily returns into calendar year returns."""
    if daily_returns_series.empty:
        return {}
    s = daily_returns_series.copy()
    s.index = pd.to_datetime(s.index)
    ann = s.groupby(s.index.year).apply(lambda x: (1 + x).prod() - 1)
    return {int(year): float(v) for year, v in ann.items() if not pd.isna(v)}