144 lines
6.7 KiB
Python
144 lines
6.7 KiB
Python
"""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) |