stonks-oracle/services/trading/performance_tracker.py

"""Performance tracker for the autonomous trading engine.

Pure computation module that computes portfolio-wide performance metrics
from closed trades and portfolio state data.
"""

from __future__ import annotations

import math

from services.trading.models import ClosedTrade, PerformanceMetrics


class PerformanceComputer:
    """Computes portfolio performance metrics from trade data.

    All methods are pure computations with no side effects or I/O.
    """

    def compute_metrics(
        self,
        closed_trades: list[ClosedTrade],
        portfolio_value: float,
        active_pool: float,
        reserve_pool: float,
        daily_pnl: float,
        unrealized_pnl: float,
        portfolio_heat: float,
        daily_returns: list[float],
    ) -> PerformanceMetrics:
        """Compute all performance metrics from trade data and portfolio state.

        Args:
            closed_trades: List of completed trades.
            portfolio_value: Current total portfolio value.
            active_pool: Current active pool value.
            reserve_pool: Current reserve pool balance.
            daily_pnl: Today's P&L.
            unrealized_pnl: Unrealized P&L across open positions.
            portfolio_heat: Current portfolio heat value.
            daily_returns: List of daily return percentages for Sharpe/drawdown.

        Returns:
            PerformanceMetrics with all computed fields.
        """
        wins = [t for t in closed_trades if t.pnl > 0]
        losses = [t for t in closed_trades if t.pnl <= 0]

        win_count = len(wins)
        loss_count = len(losses)
        total_trades = len(closed_trades)

        win_rate = win_count / total_trades if total_trades > 0 else 0.0

        avg_win = (
            sum(t.pnl for t in wins) / win_count if win_count > 0 else 0.0
        )
        avg_loss = (
            sum(t.pnl for t in losses) / loss_count if loss_count > 0 else 0.0
        )

        gross_profits = sum(t.pnl for t in wins)
        gross_losses = abs(sum(t.pnl for t in losses))

        if gross_losses > 0:
            profit_factor = gross_profits / gross_losses
        else:
            profit_factor = float("inf") if gross_profits > 0 else 0.0

        realized_pnl = sum(t.pnl for t in closed_trades)

        sharpe_ratio = self._compute_sharpe_ratio(daily_returns)
        max_drawdown = self._compute_max_drawdown(daily_returns)
        current_drawdown_pct = self._compute_current_drawdown(daily_returns)

        return PerformanceMetrics(
            total_portfolio_value=portfolio_value,
            active_pool=active_pool,
            reserve_pool=reserve_pool,
            unrealized_pnl=unrealized_pnl,
            realized_pnl=realized_pnl,
            daily_pnl=daily_pnl,
            win_count=win_count,
            loss_count=loss_count,
            win_rate=win_rate,
            avg_win=avg_win,
            avg_loss=avg_loss,
            profit_factor=profit_factor,
            sharpe_ratio=sharpe_ratio,
            max_drawdown=max_drawdown,
            current_drawdown_pct=current_drawdown_pct,
            portfolio_heat=portfolio_heat,
        )

    def compute_trade_metrics(self, trade: ClosedTrade) -> dict:
        """Compute per-trade metrics for a single closed trade.

        Args:
            trade: A completed trade.

        Returns:
            Dictionary with per-trade metrics.
        """
        return {
            "ticker": trade.ticker,
            "entry_price": trade.entry_price,
            "exit_price": trade.exit_price,
            "quantity": trade.quantity,
            "pnl": trade.pnl,
            "pnl_pct": trade.pnl_pct,
            "hold_duration": str(trade.hold_duration),
            "recommendation_id": trade.recommendation_id,
            "is_micro_trade": trade.is_micro_trade,
            "is_win": trade.pnl > 0,
        }

    def filter_by_micro_trade(
        self,
        trades: list[ClosedTrade],
        is_micro: bool,
    ) -> list[ClosedTrade]:
        """Filter trades by micro-trade flag.

        Args:
            trades: List of closed trades.
            is_micro: If True, return only micro-trades; if False, only standard.

        Returns:
            Filtered list of trades.
        """
        return [t for t in trades if t.is_micro_trade == is_micro]

    @staticmethod
    def _compute_sharpe_ratio(daily_returns: list[float]) -> float:
        """Compute annualized Sharpe ratio from daily returns.

        Formula: (mean_daily_return / std_daily_return) * sqrt(252)
        Returns 0.0 if fewer than 2 data points or std is 0.
        """
        if len(daily_returns) < 2:
            return 0.0

        n = len(daily_returns)
        mean_return = sum(daily_returns) / n
        variance = sum((r - mean_return) ** 2 for r in daily_returns) / (n - 1)
        std_return = math.sqrt(variance)

        if std_return < 1e-12:
            return 0.0

        return (mean_return / std_return) * math.sqrt(252)

    @staticmethod
    def _compute_max_drawdown(daily_returns: list[float]) -> float:
        """Compute maximum drawdown from daily returns.

        Tracks cumulative returns and finds the largest peak-to-trough decline.
        Returns 0.0 if no drawdown or insufficient data.
        """
        if not daily_returns:
            return 0.0

        cumulative = 1.0
        peak = 1.0
        max_dd = 0.0

        for r in daily_returns:
            cumulative *= (1.0 + r)
            if cumulative > peak:
                peak = cumulative
            drawdown = (peak - cumulative) / peak if peak > 0 else 0.0
            if drawdown > max_dd:
                max_dd = drawdown

        return max_dd

    @staticmethod
    def _compute_current_drawdown(daily_returns: list[float]) -> float:
        """Compute current drawdown percentage from daily returns.

        Returns the drawdown from the most recent peak to the current value.
        """
        if not daily_returns:
            return 0.0

        cumulative = 1.0
        peak = 1.0

        for r in daily_returns:
            cumulative *= (1.0 + r)
            if cumulative > peak:
                peak = cumulative

        if peak <= 0:
            return 0.0

        return (peak - cumulative) / peak