backtest/portfolio.py

import numpy as np
from config import INITIAL_CAPITAL, TRADE_FEE


class Portfolio:
    def __init__(self, initial_capital: float = INITIAL_CAPITAL):
        self.initial_capital = initial_capital
        self.cash = initial_capital
        self.position = 0.0  # units of asset held
        self.trades = []
        self.equity_history = []  # [{date, value, price, action}]
        self.peak_value = initial_capital

    def apply_decision(self, decision: dict, price: float, date: str):
        action = decision.get("action", "HOLD")
        size = float(decision.get("size", 0.5))
        size = max(0.0, min(1.0, size))

        trade_executed = False
        trade_value = 0.0

        if action == "BUY" and self.cash > 0:
            spend = self.cash * size
            fee = spend * TRADE_FEE
            net_spend = spend - fee
            units = net_spend / price
            self.position += units
            self.cash -= spend
            trade_executed = True
            trade_value = spend
            self.trades.append({
                "date": date,
                "action": "BUY",
                "price": price,
                "units": units,
                "value": spend,
                "fee": fee,
            })

        elif action == "SELL" and self.position > 0:
            sell_units = self.position * size
            gross = sell_units * price
            fee = gross * TRADE_FEE
            net = gross - fee
            self.position -= sell_units
            self.cash += net
            trade_executed = True
            trade_value = gross
            self.trades.append({
                "date": date,
                "action": "SELL",
                "price": price,
                "units": sell_units,
                "value": gross,
                "fee": fee,
            })

        total_value = self.cash + self.position * price
        self.peak_value = max(self.peak_value, total_value)

        self.equity_history.append({
            "date": date,
            "value": round(total_value, 2),
            "cash": round(self.cash, 2),
            "position": self.position,
            "price": price,
            "action": action,
            "trade_executed": trade_executed,
            "trade_value": round(trade_value, 2),
        })

    def current_value(self, price: float) -> float:
        return self.cash + self.position * price

    def drawdown(self, price: float) -> float:
        current = self.current_value(price)
        if self.peak_value == 0:
            return 0.0
        return (self.peak_value - current) / self.peak_value

    def snapshot(self, price: float) -> dict:
        total = self.current_value(price)
        return {
            "cash": round(self.cash, 2),
            "position": self.position,
            "total_value": round(total, 2),
            "drawdown": round(self.drawdown(price), 4),
        }


def compute_metrics(equity_history: list, initial_capital: float, hodl_final: float) -> dict:
    if not equity_history:
        return {}

    values = [e["value"] for e in equity_history]
    dates = [e["date"] for e in equity_history]

    # Daily returns
    returns = []
    for i in range(1, len(values)):
        r = (values[i] - values[i - 1]) / values[i - 1] if values[i - 1] != 0 else 0
        returns.append(r)

    returns_arr = np.array(returns)
    final_value = values[-1]

    # Cumulative Return
    cumulative_return = (final_value - initial_capital) / initial_capital

    # Sharpe Ratio (annualized, risk-free = 0)
    if len(returns_arr) > 1 and returns_arr.std() > 0:
        sharpe = (returns_arr.mean() / returns_arr.std()) * np.sqrt(252)
    else:
        sharpe = 0.0

    # Sortino Ratio
    downside = returns_arr[returns_arr < 0]
    if len(downside) > 0 and downside.std() > 0:
        sortino = (returns_arr.mean() / downside.std()) * np.sqrt(252)
    else:
        sortino = 0.0

    # Max Drawdown
    peak = initial_capital
    max_dd = 0.0
    for v in values:
        if v > peak:
            peak = v
        dd = (peak - v) / peak if peak > 0 else 0
        max_dd = max(max_dd, dd)

    # Win Rate
    winning_trades = sum(1 for e in equity_history if e.get("trade_executed") and e.get("action") == "BUY")
    num_trades = sum(1 for e in equity_history if e.get("trade_executed"))
    win_rate = winning_trades / num_trades if num_trades > 0 else 0.0

    # vs HODL
    hodl_return = (hodl_final - initial_capital) / initial_capital
    alpha = cumulative_return - hodl_return

    return {
        "cumulative_return": round(cumulative_return, 4),
        "sharpe_ratio": round(sharpe, 4),
        "sortino_ratio": round(sortino, 4),
        "max_drawdown": round(max_dd, 4),
        "win_rate": round(win_rate, 4),
        "num_trades": num_trades,
        "final_value": round(final_value, 2),
        "hodl_return": round(hodl_return, 4),
        "alpha": round(alpha, 4),
        "start_date": dates[0] if dates else "",
        "end_date": dates[-1] if dates else "",
        "num_days": len(values),
    }