"""Core: Risk management and capital protection.""" from typing import Dict, Any from .order_manager import OrderManager from ..storage.models import Order from decimal import Decimal class RiskManager: def __init__(self, order_manager: OrderManager, settings: Dict[str, Any], logger: Any): self.order_manager = order_manager self.settings = settings.get("risk", {}) self.logger = logger self.max_daily_loss_percentage = Decimal(str(self.settings.get("max_daily_loss_percentage", 0.01))) self.default_position_size = self.settings.get("default_position_size", 10) self.stop_loss_percentage = Decimal(str(self.settings.get("stop_loss_percentage", 0.005))) self.take_profit_percentage = Decimal(str(self.settings.get("take_profit_percentage", 0.01))) self.risk_per_trade_percentage = Decimal(str(self.settings.get("risk_per_trade_percentage", 0.005))) self.daily_profit_loss_today = Decimal('0.0') # Track daily P&L for current day self.logger.info("RiskManager initialized with settings: %s", self.settings) self._reset_daily_metrics() def _reset_daily_metrics(self): """ Resets daily metrics like profit/loss. In a real application, this would be triggered at the start of a new trading day. """ self.daily_profit_loss_today = Decimal('0.0') self.logger.info("Daily risk metrics reset.") def validate_order(self, order_data: Dict[str, Any]) -> bool: """ Validates if an order can be placed based on risk rules and adjusts order parameters. """ if not self._monitor_daily_loss(): self.logger.warning("Order rejected due to exceeding maximum daily loss.") return False symbol = order_data.get("symbol") side = order_data.get("side") current_price = Decimal(str(order_data.get("current_price"))) # Create a dummy Order object for _calculate_stop_loss_take_profit # In a real system, the order would be more fully formed here or SL/TP calculation # would be refactored to take individual parameters. dummy_order = Order(symbol=symbol, side=side, quantity=0, price=current_price) # Only side and symbol are needed for SL/TP calcs sl_tp_prices = self._calculate_stop_loss_take_profit(dummy_order, current_price) stop_loss_price = sl_tp_prices["stop_loss"] take_profit_price = sl_tp_prices["take_profit"] quantity = self._calculate_position_size(symbol, current_price, stop_loss_price) if quantity == 0: self.logger.warning("Order rejected due to calculated position size being zero.") return False # Update order_data with calculated values order_data["quantity"] = quantity order_data["stop_loss_price"] = stop_loss_price order_data["take_profit_price"] = take_profit_price self.logger.info("Order validated and adjusted by RiskManager: %s", order_data) return True def _get_current_portfolio_value(self) -> Decimal: """ Placeholder to fetch the current total portfolio value. In a real system, this would interact with the portfolio manager or broker. """ # For now, return a dummy value return Decimal('100000.00') def _calculate_position_size(self, symbol: str, current_price: Decimal, stop_loss_price: Decimal) -> int: """ Calculates the appropriate position size based on risk settings, portfolio, and stop-loss price. """ if current_price <= Decimal('0.0') or stop_loss_price <= Decimal('0.0') or current_price == stop_loss_price: self.logger.warning( "Cannot calculate position size with invalid prices: current_price=%s, stop_loss_price=%s", current_price, stop_loss_price ) return 0 portfolio_value = self._get_current_portfolio_value() # Risk a small percentage of portfolio per trade risk_per_trade_capital = portfolio_value * self.settings.get("risk_per_trade_percentage", Decimal('0.005')) # e.g., 0.5% # Calculate the potential loss per share if stop-loss is hit loss_per_share = abs(current_price - stop_loss_price) if loss_per_share == Decimal('0.0'): self.logger.warning("Loss per share is zero, cannot calculate position size safely. Defaulting to 1.") return 1 # Calculate quantity quantity = int(risk_per_trade_capital / loss_per_share) if quantity == 0: quantity = 1 # Ensure at least 1 share is traded if valid self.logger.info( "Calculated position size for %s: %s (Current Price: %s, Stop Loss: %s, Portfolio Value: %s, Risk per Trade Capital: %s)", symbol, quantity, current_price, stop_loss_price, portfolio_value, risk_per_trade_capital ) return min(quantity, self.default_position_size) # Cap at default for now def _calculate_stop_loss_take_profit(self, order: Order, current_price: Decimal) -> Dict[str, Decimal]: """ Calculates stop loss and take profit prices for an order. """ stop_loss_price = Decimal('0.0') take_profit_price = Decimal('0.0') if order.side == "BUY": stop_loss_price = current_price * (Decimal('1.0') - self.stop_loss_percentage) take_profit_price = current_price * (Decimal('1.0') + self.take_profit_percentage) elif order.side == "SELL": stop_loss_price = current_price * (Decimal('1.0') + self.stop_loss_percentage) take_profit_price = current_price * (Decimal('1.0') - self.take_profit_percentage) self.logger.info( "Calculated SL/TP for %s at price %s: SL=%s, TP=%s", order.symbol, current_price, stop_loss_price, take_profit_price ) return {"stop_loss": stop_loss_price, "take_profit": take_profit_price} def _monitor_daily_loss(self) -> bool: """ Monitors the daily profit/loss and returns False if the max daily loss is exceeded. """ portfolio_value = self._get_current_portfolio_value() if portfolio_value > 0 and self.daily_profit_loss_today / portfolio_value < -self.max_daily_loss_percentage: self.logger.warning( "Maximum daily loss exceeded. Current P&L: %s, Max Loss: %s", self.daily_profit_loss_today, -self.max_daily_loss_percentage * portfolio_value ) return False return True def update_daily_profit_loss(self, pnl: Decimal): """ Updates the daily profit and loss. """ self.daily_profit_loss_today += pnl self.logger.info("Daily P&L updated: %s", self.daily_profit_loss_today)