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), }