""" Position Manager for backtesting engine. This module provides position management functionality including: - Opening and closing positions with proper P&L calculation - Tracking open positions and margin requirements - Stop-loss and take-profit management - Kelly criterion position sizing """ import math from dataclasses import dataclass, field from datetime import datetime from typing import Dict, List, Optional, Tuple, Any from loguru import logger from .trade import Trade, TradeDirection, TradeStatus @dataclass class PositionSizingConfig: """ Configuration for position sizing algorithms. Attributes: method: Sizing method ('fixed', 'kelly', 'fixed_risk', 'volatility_adjusted') max_position_pct: Maximum position size as percentage of capital kelly_fraction: Fraction of Kelly criterion to use (0.25-0.5 recommended) risk_per_trade_pct: Risk per trade as percentage of capital max_leverage: Maximum allowed leverage min_position_size: Minimum position size in units """ method: str = "kelly" max_position_pct: float = 0.02 kelly_fraction: float = 0.25 risk_per_trade_pct: float = 0.02 max_leverage: float = 10.0 min_position_size: float = 0.01 @dataclass class MarginRequirements: """ Margin requirements for a position. Attributes: initial_margin: Required margin to open position maintenance_margin: Minimum margin to maintain position margin_call_level: Level at which margin call is triggered """ initial_margin: float = 0.0 maintenance_margin: float = 0.0 margin_call_level: float = 0.0 @dataclass class PositionSummary: """ Summary of current position state. Attributes: total_positions: Number of open positions total_long: Number of long positions total_short: Number of short positions total_exposure: Total market exposure unrealized_pnl: Total unrealized P&L used_margin: Total margin in use available_margin: Available margin for new positions """ total_positions: int = 0 total_long: int = 0 total_short: int = 0 total_exposure: float = 0.0 unrealized_pnl: float = 0.0 used_margin: float = 0.0 available_margin: float = 0.0 class PositionManager: """ Manages trading positions for backtesting. This class handles all aspects of position management including: - Opening new positions with proper sizing - Closing positions with P&L calculation - Managing stop-loss and take-profit orders - Tracking margin requirements - Kelly criterion position sizing Attributes: capital: Current account capital initial_capital: Starting capital open_positions: List of currently open trades closed_positions: List of closed trades sizing_config: Position sizing configuration leverage: Account leverage margin_pct: Margin requirement percentage """ def __init__( self, initial_capital: float = 10000.0, sizing_config: Optional[PositionSizingConfig] = None, leverage: float = 1.0, margin_pct: float = 0.01, max_positions: int = 10 ): """ Initialize the position manager. Args: initial_capital: Starting capital amount sizing_config: Position sizing configuration leverage: Account leverage margin_pct: Margin requirement as percentage max_positions: Maximum concurrent positions allowed """ self.initial_capital = initial_capital self.capital = initial_capital self.sizing_config = sizing_config or PositionSizingConfig() self.leverage = leverage self.margin_pct = margin_pct self.max_positions = max_positions self.open_positions: List[Trade] = [] self.closed_positions: List[Trade] = [] self.trade_counter = 0 self.used_margin = 0.0 self.peak_capital = initial_capital self.max_drawdown = 0.0 self._win_count = 0 self._loss_count = 0 self._avg_win = 0.0 self._avg_loss = 0.0 logger.info( f"PositionManager initialized: capital=${initial_capital:,.2f}, " f"leverage={leverage}x, max_positions={max_positions}" ) def reset(self) -> None: """Reset position manager to initial state.""" self.capital = self.initial_capital self.open_positions = [] self.closed_positions = [] self.trade_counter = 0 self.used_margin = 0.0 self.peak_capital = self.initial_capital self.max_drawdown = 0.0 self._win_count = 0 self._loss_count = 0 self._avg_win = 0.0 self._avg_loss = 0.0 def can_open_position(self) -> bool: """ Check if a new position can be opened. Returns: True if new position is allowed """ if len(self.open_positions) >= self.max_positions: return False available = self.capital - self.used_margin if available <= 0: return False return True def calculate_position_size( self, entry_price: float, stop_loss: float, direction: TradeDirection, win_rate: Optional[float] = None, avg_win_loss_ratio: Optional[float] = None ) -> float: """ Calculate optimal position size based on configured method. Args: entry_price: Entry price for the trade stop_loss: Stop loss price direction: Trade direction win_rate: Historical win rate (for Kelly criterion) avg_win_loss_ratio: Average win/loss ratio (for Kelly criterion) Returns: Position size in units """ method = self.sizing_config.method if method == "fixed": return self._fixed_position_size(entry_price) elif method == "kelly": return self._kelly_position_size( entry_price, stop_loss, direction, win_rate, avg_win_loss_ratio ) elif method == "fixed_risk": return self._fixed_risk_position_size(entry_price, stop_loss, direction) elif method == "volatility_adjusted": return self._volatility_adjusted_size(entry_price, stop_loss, direction) else: logger.warning(f"Unknown sizing method: {method}, using fixed") return self._fixed_position_size(entry_price) def _fixed_position_size(self, entry_price: float) -> float: """ Calculate fixed percentage position size. Args: entry_price: Entry price Returns: Position size """ available_capital = self.capital - self.used_margin position_value = available_capital * self.sizing_config.max_position_pct size = position_value / entry_price return max(size, self.sizing_config.min_position_size) def _kelly_position_size( self, entry_price: float, stop_loss: float, direction: TradeDirection, win_rate: Optional[float] = None, avg_win_loss_ratio: Optional[float] = None ) -> float: """ Calculate position size using Kelly criterion. The Kelly criterion determines optimal bet sizing: f* = (p * b - q) / b where: - f* = fraction of capital to risk - p = probability of winning - q = probability of losing (1 - p) - b = win/loss ratio We use a fractional Kelly (kelly_fraction) to reduce volatility. Args: entry_price: Entry price stop_loss: Stop loss price direction: Trade direction win_rate: Historical win rate avg_win_loss_ratio: Average win/loss ratio Returns: Position size """ if win_rate is None: if self._win_count + self._loss_count > 10: win_rate = self._win_count / (self._win_count + self._loss_count) else: win_rate = 0.5 if avg_win_loss_ratio is None: if self._avg_loss > 0: avg_win_loss_ratio = self._avg_win / self._avg_loss else: avg_win_loss_ratio = 1.5 win_rate = max(0.1, min(0.9, win_rate)) avg_win_loss_ratio = max(0.5, min(5.0, avg_win_loss_ratio)) q = 1 - win_rate kelly_fraction = (win_rate * avg_win_loss_ratio - q) / avg_win_loss_ratio kelly_fraction = max(0, kelly_fraction) kelly_fraction *= self.sizing_config.kelly_fraction kelly_fraction = min(kelly_fraction, self.sizing_config.max_position_pct) if direction == TradeDirection.LONG: risk_per_unit = entry_price - stop_loss else: risk_per_unit = stop_loss - entry_price if risk_per_unit <= 0: logger.warning("Invalid stop loss, using fixed position size") return self._fixed_position_size(entry_price) available_capital = self.capital - self.used_margin risk_amount = available_capital * kelly_fraction size = risk_amount / risk_per_unit size = min(size, (available_capital * self.sizing_config.max_position_pct) / entry_price) size = max(size, self.sizing_config.min_position_size) return size def _fixed_risk_position_size( self, entry_price: float, stop_loss: float, direction: TradeDirection ) -> float: """ Calculate position size based on fixed risk percentage. Args: entry_price: Entry price stop_loss: Stop loss price direction: Trade direction Returns: Position size """ if direction == TradeDirection.LONG: risk_per_unit = entry_price - stop_loss else: risk_per_unit = stop_loss - entry_price if risk_per_unit <= 0: return self._fixed_position_size(entry_price) available_capital = self.capital - self.used_margin risk_amount = available_capital * self.sizing_config.risk_per_trade_pct size = risk_amount / risk_per_unit max_size = (available_capital * self.sizing_config.max_position_pct) / entry_price size = min(size, max_size) size = max(size, self.sizing_config.min_position_size) return size def _volatility_adjusted_size( self, entry_price: float, stop_loss: float, direction: TradeDirection ) -> float: """ Calculate volatility-adjusted position size. This method adjusts position size based on the distance to stop loss, which serves as a proxy for volatility. Args: entry_price: Entry price stop_loss: Stop loss price direction: Trade direction Returns: Position size """ if direction == TradeDirection.LONG: volatility = (entry_price - stop_loss) / entry_price else: volatility = (stop_loss - entry_price) / entry_price volatility = max(0.001, volatility) base_risk = self.sizing_config.risk_per_trade_pct adjusted_risk = base_risk * (0.01 / volatility) adjusted_risk = min(adjusted_risk, self.sizing_config.max_position_pct) available_capital = self.capital - self.used_margin position_value = available_capital * adjusted_risk size = position_value / entry_price return max(size, self.sizing_config.min_position_size) def open_position( self, symbol: str, direction: TradeDirection, entry_price: float, entry_time: datetime, size: Optional[float] = None, stop_loss: Optional[float] = None, take_profit: Optional[float] = None, commission: float = 0.0, slippage: float = 0.0, strategy_name: str = "default", timeframe: str = "1H", signal_confidence: float = 0.0, metadata: Optional[Dict[str, Any]] = None ) -> Optional[Trade]: """ Open a new trading position. Args: symbol: Trading symbol direction: Trade direction entry_price: Entry price entry_time: Entry timestamp size: Position size (calculated if None) stop_loss: Stop loss price take_profit: Take profit price commission: Commission cost slippage: Slippage cost strategy_name: Name of the strategy timeframe: Trading timeframe signal_confidence: Confidence of the entry signal metadata: Additional trade metadata Returns: Trade object if opened, None if unable to open """ if not self.can_open_position(): logger.warning("Cannot open position: max positions reached or insufficient margin") return None if size is None: if stop_loss is None: size = self._fixed_position_size(entry_price) else: size = self.calculate_position_size(entry_price, stop_loss, direction) entry_price_adjusted = entry_price if direction == TradeDirection.LONG: entry_price_adjusted += slippage else: entry_price_adjusted -= slippage self.trade_counter += 1 trade = Trade( trade_id=self.trade_counter, symbol=symbol, direction=direction, entry_price=entry_price_adjusted, entry_time=entry_time, size=size, stop_loss=stop_loss, take_profit=take_profit, commission=commission, slippage=slippage, strategy_name=strategy_name, timeframe=timeframe, signal_confidence=signal_confidence, metadata=metadata or {} ) position_value = entry_price_adjusted * size margin_required = position_value * self.margin_pct self.used_margin += margin_required self.capital -= commission self.open_positions.append(trade) logger.debug( f"Opened {direction.value} position: {symbol} @ {entry_price_adjusted:.5f}, " f"size={size:.4f}, SL={stop_loss}, TP={take_profit}" ) return trade def close_position( self, trade: Trade, exit_price: float, exit_time: datetime, status: TradeStatus = TradeStatus.CLOSED, commission: float = 0.0, slippage: float = 0.0 ) -> float: """ Close an open position. Args: trade: Trade to close exit_price: Exit price exit_time: Exit timestamp status: Exit status commission: Commission cost slippage: Slippage cost Returns: Realized P&L """ if trade not in self.open_positions: logger.warning(f"Trade {trade.trade_id} not found in open positions") return 0.0 exit_price_adjusted = exit_price if trade.is_long: exit_price_adjusted -= slippage else: exit_price_adjusted += slippage pnl = trade.close( exit_price=exit_price_adjusted, exit_time=exit_time, status=status, commission=commission, slippage=slippage ) position_value = trade.entry_price * trade.size margin_released = position_value * self.margin_pct self.used_margin -= margin_released self.capital += pnl self.capital -= commission if pnl > 0: self._win_count += 1 self._avg_win = ( (self._avg_win * (self._win_count - 1) + pnl) / self._win_count ) elif pnl < 0: self._loss_count += 1 self._avg_loss = ( (self._avg_loss * (self._loss_count - 1) + abs(pnl)) / self._loss_count ) if self.capital > self.peak_capital: self.peak_capital = self.capital else: drawdown = (self.peak_capital - self.capital) / self.peak_capital self.max_drawdown = max(self.max_drawdown, drawdown) self.open_positions.remove(trade) self.closed_positions.append(trade) logger.debug( f"Closed {trade.direction.value} position: {trade.symbol} @ {exit_price_adjusted:.5f}, " f"PnL={pnl:+.2f} ({status.value})" ) return pnl def close_position_by_id( self, trade_id: int, exit_price: float, exit_time: datetime, status: TradeStatus = TradeStatus.CLOSED, commission: float = 0.0, slippage: float = 0.0 ) -> Optional[float]: """ Close a position by trade ID. Args: trade_id: ID of the trade to close exit_price: Exit price exit_time: Exit timestamp status: Exit status commission: Commission cost slippage: Slippage cost Returns: Realized P&L or None if trade not found """ trade = self.get_position_by_id(trade_id) if trade is None: logger.warning(f"Trade {trade_id} not found") return None return self.close_position( trade, exit_price, exit_time, status, commission, slippage ) def close_all_positions( self, exit_price_func, exit_time: datetime, status: TradeStatus = TradeStatus.CLOSED, commission_func=None, slippage: float = 0.0 ) -> float: """ Close all open positions. Args: exit_price_func: Function that takes trade and returns exit price exit_time: Exit timestamp status: Exit status commission_func: Function that takes trade and returns commission slippage: Slippage cost Returns: Total realized P&L """ total_pnl = 0.0 for trade in self.open_positions[:]: exit_price = exit_price_func(trade) commission = commission_func(trade) if commission_func else 0.0 pnl = self.close_position( trade, exit_price, exit_time, status, commission, slippage ) total_pnl += pnl return total_pnl def update_positions( self, current_prices: Dict[str, float], current_time: datetime, commission_rate: float = 0.0, slippage: float = 0.0 ) -> List[Trade]: """ Update all open positions with current prices. Check for stop-loss and take-profit triggers. Args: current_prices: Dictionary of symbol -> current price current_time: Current timestamp commission_rate: Commission rate for closing slippage: Slippage for closing Returns: List of trades that were closed """ closed_trades = [] for trade in self.open_positions[:]: current_price = current_prices.get(trade.symbol) if current_price is None: continue if trade.should_stop_out(current_price): exit_price = trade.stop_loss commission = abs(exit_price * trade.size) * commission_rate self.close_position( trade, exit_price, current_time, TradeStatus.STOPPED_OUT, commission, slippage ) closed_trades.append(trade) elif trade.should_take_profit(current_price): exit_price = trade.take_profit commission = abs(exit_price * trade.size) * commission_rate self.close_position( trade, exit_price, current_time, TradeStatus.TAKE_PROFIT, commission, slippage ) closed_trades.append(trade) return closed_trades def update_position_with_ohlc( self, trade: Trade, high: float, low: float, close: float, current_time: datetime, commission_rate: float = 0.0, slippage: float = 0.0 ) -> Optional[Tuple[TradeStatus, float]]: """ Update a single position with OHLC bar data. Checks if SL or TP was hit within the bar. Args: trade: Trade to update high: High price of the bar low: Low price of the bar close: Close price of the bar current_time: Current timestamp commission_rate: Commission rate slippage: Slippage amount Returns: Tuple of (status, pnl) if closed, None if still open """ if trade not in self.open_positions: return None sl_hit = False tp_hit = False if trade.stop_loss is not None: if trade.is_long and low <= trade.stop_loss: sl_hit = True elif trade.is_short and high >= trade.stop_loss: sl_hit = True if trade.take_profit is not None: if trade.is_long and high >= trade.take_profit: tp_hit = True elif trade.is_short and low <= trade.take_profit: tp_hit = True if sl_hit and tp_hit: if trade.is_long: sl_hit = low <= trade.stop_loss else: sl_hit = high >= trade.stop_loss if sl_hit: tp_hit = False if sl_hit: exit_price = trade.stop_loss commission = abs(exit_price * trade.size) * commission_rate pnl = self.close_position( trade, exit_price, current_time, TradeStatus.STOPPED_OUT, commission, slippage ) return (TradeStatus.STOPPED_OUT, pnl) elif tp_hit: exit_price = trade.take_profit commission = abs(exit_price * trade.size) * commission_rate pnl = self.close_position( trade, exit_price, current_time, TradeStatus.TAKE_PROFIT, commission, slippage ) return (TradeStatus.TAKE_PROFIT, pnl) return None def get_position_by_id(self, trade_id: int) -> Optional[Trade]: """ Get an open position by trade ID. Args: trade_id: Trade ID to find Returns: Trade object or None """ for trade in self.open_positions: if trade.trade_id == trade_id: return trade return None def get_positions_by_symbol(self, symbol: str) -> List[Trade]: """ Get all open positions for a symbol. Args: symbol: Symbol to filter by Returns: List of matching trades """ return [t for t in self.open_positions if t.symbol == symbol] def get_unrealized_pnl(self, current_prices: Dict[str, float]) -> float: """ Calculate total unrealized P&L for all open positions. Args: current_prices: Dictionary of symbol -> current price Returns: Total unrealized P&L """ total = 0.0 for trade in self.open_positions: current_price = current_prices.get(trade.symbol) if current_price is not None: total += trade.calculate_unrealized_pnl(current_price) return total def get_total_exposure(self) -> float: """ Calculate total market exposure. Returns: Total exposure in account currency """ return sum(t.entry_price * t.size for t in self.open_positions) def get_position_summary(self, current_prices: Optional[Dict[str, float]] = None) -> PositionSummary: """ Get summary of current position state. Args: current_prices: Current prices for unrealized P&L calculation Returns: PositionSummary object """ long_count = sum(1 for t in self.open_positions if t.is_long) short_count = sum(1 for t in self.open_positions if t.is_short) unrealized = 0.0 if current_prices: unrealized = self.get_unrealized_pnl(current_prices) return PositionSummary( total_positions=len(self.open_positions), total_long=long_count, total_short=short_count, total_exposure=self.get_total_exposure(), unrealized_pnl=unrealized, used_margin=self.used_margin, available_margin=self.capital - self.used_margin ) def get_equity(self, current_prices: Optional[Dict[str, float]] = None) -> float: """ Calculate current account equity. Args: current_prices: Current prices for unrealized P&L Returns: Account equity """ equity = self.capital if current_prices: equity += self.get_unrealized_pnl(current_prices) return equity def get_win_rate(self) -> float: """ Calculate current win rate from closed trades. Returns: Win rate as decimal (0 to 1) """ total = self._win_count + self._loss_count if total == 0: return 0.0 return self._win_count / total def get_profit_factor(self) -> float: """ Calculate profit factor from closed trades. Returns: Profit factor (gross profit / gross loss) """ if self._loss_count == 0 or self._avg_loss == 0: return float('inf') if self._win_count > 0 else 0.0 gross_profit = self._win_count * self._avg_win gross_loss = self._loss_count * self._avg_loss return gross_profit / gross_loss if gross_loss > 0 else 0.0 def get_all_trades(self) -> List[Trade]: """ Get all trades (open and closed). Returns: List of all trades """ return self.open_positions + self.closed_positions def to_dict(self) -> Dict[str, Any]: """ Convert position manager state to dictionary. Returns: Dictionary representation """ return { "capital": self.capital, "initial_capital": self.initial_capital, "used_margin": self.used_margin, "peak_capital": self.peak_capital, "max_drawdown": self.max_drawdown, "open_positions_count": len(self.open_positions), "closed_positions_count": len(self.closed_positions), "total_trades": self.trade_counter, "win_count": self._win_count, "loss_count": self._loss_count, "avg_win": self._avg_win, "avg_loss": self._avg_loss, "win_rate": self.get_win_rate(), "profit_factor": self.get_profit_factor() }