Benchmarking AI Agent Financial Performance: Metrics That Matter

#!/usr/bin/env python3
"""
AI Agent Financial Performance Benchmarking Framework
Computes Sharpe, Sortino, Calmar, MaxDD, Recovery Factor, Win Rate.
Reference: Purple Flea Research doi.org/10.5281/zenodo.18808440
"""

from dataclasses import dataclass, field
from typing import List, Optional, Tuple, Dict
import math
import statistics
import json
from datetime import date


@dataclass
class PerformanceReport:
    agent_id: str
    period_days: int
    # Returns
    total_return: float
    cagr: float
    # Risk-adjusted
    sharpe_ratio: float
    sortino_ratio: float
    calmar_ratio: float
    # Drawdown
    max_drawdown: float
    max_dd_duration_days: int
    recovery_factor: float
    # Trade stats
    win_rate: float
    profit_factor: float
    avg_win: float
    avg_loss: float
    # Volatility
    annual_volatility: float
    downside_deviation: float
    # Grade
    overall_grade: str
    grade_breakdown: Dict[str, str] = field(default_factory=dict)


class AgentBenchmark:
    """
    Comprehensive performance benchmarking for AI financial agents.

    Usage:
        returns = [0.02, -0.01, 0.03, ...]  # daily returns as decimals
        trades = [(pnl, bool_win), ...]       # optional trade-level data
        bench = AgentBenchmark(returns, trades, risk_free_daily=0.00013)
        report = bench.compute(agent_id="my-agent")
    """

    TRADING_DAYS_PER_YEAR = 365  # crypto / 24/7 agents
    # Use 252 for traditional equity agents

    def __init__(
        self,
        daily_returns: List[float],
        trade_pnls: Optional[List[float]] = None,
        risk_free_daily: float = 0.00013  # ~5% annual / 365
    ):
        self.returns = daily_returns
        self.trade_pnls = trade_pnls or []
        self.rf_daily = risk_free_daily

    # ─── Core metric calculations ───────────────────────────────────────────

    def total_return(self) -> float:
        result = 1.0
        for r in self.returns:
            result *= (1 + r)
        return result - 1

    def cagr(self) -> float:
        if len(self.returns) == 0:
            return 0.0
        total = self.total_return()
        years = len(self.returns) / self.TRADING_DAYS_PER_YEAR
        if years == 0 or (1 + total) <= 0:
            return 0.0
        return (1 + total) ** (1 / years) - 1

    def annualized_volatility(self) -> float:
        if len(self.returns) < 2:
            return 0.0
        std = statistics.stdev(self.returns)
        return std * math.sqrt(self.TRADING_DAYS_PER_YEAR)

    def downside_deviation(self, threshold: float = 0.0) -> float:
        """Annualized downside deviation (for Sortino ratio)."""
        downside = [min(0, r - threshold) for r in self.returns]
        if not downside:
            return 0.0
        squared = [d ** 2 for d in downside]
        return math.sqrt(statistics.mean(squared)) * math.sqrt(self.TRADING_DAYS_PER_YEAR)

    def sharpe_ratio(self) -> float:
        excess_returns = [r - self.rf_daily for r in self.returns]
        if len(excess_returns) < 2:
            return 0.0
        mean_excess = statistics.mean(excess_returns)
        std_excess = statistics.stdev(excess_returns)
        if std_excess == 0:
            return float('inf') if mean_excess > 0 else 0.0
        return (mean_excess / std_excess) * math.sqrt(self.TRADING_DAYS_PER_YEAR)

    def sortino_ratio(self) -> float:
        ann_excess = (statistics.mean(self.returns) - self.rf_daily) * self.TRADING_DAYS_PER_YEAR
        dd = self.downside_deviation()
        if dd == 0:
            return float('inf') if ann_excess > 0 else 0.0
        return ann_excess / dd

    def drawdown_series(self) -> List[float]:
        """Returns the drawdown at each point (negative values)."""
        equity = [1.0]
        for r in self.returns:
            equity.append(equity[-1] * (1 + r))
        peak = equity[0]
        drawdowns = []
        for val in equity:
            if val > peak:
                peak = val
            drawdowns.append((val - peak) / peak if peak > 0 else 0.0)
        return drawdowns

    def max_drawdown(self) -> float:
        return min(self.drawdown_series()) if self.returns else 0.0

    def max_drawdown_duration(self) -> int:
        """Returns max consecutive days spent in drawdown (underwater period)."""
        dds = self.drawdown_series()
        max_dur, cur_dur = 0, 0
        for dd in dds:
            if dd < 0:
                cur_dur += 1
                max_dur = max(max_dur, cur_dur)
            else:
                cur_dur = 0
        return max_dur

    def calmar_ratio(self) -> float:
        mdd = abs(self.max_drawdown())
        if mdd == 0:
            return float('inf')
        return self.cagr() / mdd

    def recovery_factor(self) -> float:
        mdd = abs(self.max_drawdown())
        if mdd == 0:
            return float('inf')
        return self.total_return() / mdd

    def win_stats(self) -> Tuple[float, float, float, float]:
        """Returns (win_rate, profit_factor, avg_win, avg_loss)."""
        if not self.trade_pnls:
            # Infer from daily returns
            wins = [r for r in self.returns if r > 0]
            losses = [r for r in self.returns if r < 0]
        else:
            wins = [p for p in self.trade_pnls if p > 0]
            losses = [p for p in self.trade_pnls if p < 0]

        total = len(wins) + len(losses)
        win_rate = len(wins) / total if total > 0 else 0.0
        gross_wins = sum(wins) if wins else 0
        gross_losses = abs(sum(losses)) if losses else 0
        pf = gross_wins / gross_losses if gross_losses > 0 else float('inf')
        avg_win = statistics.mean(wins) if wins else 0.0
        avg_loss = statistics.mean(losses) if losses else 0.0
        return win_rate, pf, avg_win, avg_loss

    def _grade_metric(self, metric: str, value: float) -> str:
        thresholds = {
            "sharpe": [(2.0, "A"), (1.0, "B"), (0.5, "C"), (0.0, "D"), (float('-inf'), "F")],
            "sortino": [(3.0, "A"), (2.0, "B"), (1.0, "C"), (0.5, "D"), (float('-inf'), "F")],
            "calmar": [(3.0, "A"), (1.0, "B"), (0.5, "C"), (0.2, "D"), (float('-inf'), "F")],
            "max_dd": [(-0.05, "A"), (-0.15, "B"), (-0.30, "C"), (-0.50, "D"), (float('-inf'), "F")],
        }
        for threshold, grade in thresholds.get(metric, []):
            if value >= threshold:
                return grade
        return "F"

    def compute(self, agent_id: str = "agent") -> PerformanceReport:
        """Compute the full performance report."""
        sharpe = self.sharpe_ratio()
        sortino = self.sortino_ratio()
        mdd = self.max_drawdown()
        calmar = self.calmar_ratio()
        win_rate, pf, avg_win, avg_loss = self.win_stats()

        grade_breakdown = {
            "sharpe": self._grade_metric("sharpe", sharpe),
            "sortino": self._grade_metric("sortino", sortino),
            "calmar": self._grade_metric("calmar", calmar),
            "max_drawdown": self._grade_metric("max_dd", mdd),
        }
        grade_values = {"A": 4, "B": 3, "C": 2, "D": 1, "F": 0}
        avg_grade = statistics.mean([grade_values[g] for g in grade_breakdown.values()])
        overall = ["F", "D", "C", "B", "A"][round(avg_grade)]

        return PerformanceReport(
            agent_id=agent_id,
            period_days=len(self.returns),
            total_return=self.total_return(),
            cagr=self.cagr(),
            sharpe_ratio=sharpe,
            sortino_ratio=sortino,
            calmar_ratio=calmar,
            max_drawdown=mdd,
            max_dd_duration_days=self.max_drawdown_duration(),
            recovery_factor=self.recovery_factor(),
            win_rate=win_rate,
            profit_factor=pf,
            avg_win=avg_win,
            avg_loss=avg_loss,
            annual_volatility=self.annualized_volatility(),
            downside_deviation=self.downside_deviation(),
            overall_grade=overall,
            grade_breakdown=grade_breakdown,
        )

"""Compare multiple agents and produce a ranked benchmark table."""
import random
import math
from agent_benchmark import AgentBenchmark

def simulate_returns(
    n_days: int,
    daily_mean: float,
    daily_std: float,
    seed: int = 42
) -> list:
    """Generate synthetic return series for testing."""
    rng = random.Random(seed)
    return [rng.gauss(daily_mean, daily_std) for _ in range(n_days)]

# Define agent configurations (mean daily return, daily volatility)
agents = {
    "casino-agent-01": (0.0009, 0.025),
    "trading-agent-07": (0.0007, 0.018),
    "arb-agent-03": (0.0004, 0.006),
    "high-vol-agent": (0.0015, 0.065),
    "wallet-yield-01": (0.0003, 0.008),
}

results = []
for agent_id, (mu, sigma) in agents.items():
    returns = simulate_returns(180, mu, sigma, seed=hash(agent_id) % 10000)
    bench = AgentBenchmark(returns)
    report = bench.compute(agent_id)
    results.append(report)

# Sort by Sharpe ratio
results.sort(key=lambda r: r.sharpe_ratio, reverse=True)

print(f"{'Rank':<5} {'Agent':<20} {'Sharpe':>8} {'MaxDD':>8} {'Calmar':>8} {'Grade':>6}")
print("-" * 60)
for i, r in enumerate(results, 1):
    print(
        f"{i:<5} {r.agent_id:<20} "
        f"{r.sharpe_ratio:>8.2f} {r.max_drawdown:>7.1%} "
        f"{r.calmar_ratio:>8.2f} {r.overall_grade:>6}"
    )