Spaces:
Running
Running
File size: 5,259 Bytes
b2cc431 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 | import os
import logging
from typing import Dict, Any, List, Tuple
logger = logging.getLogger("SentryTheta.RiskEngine")
class RiskEngine:
def __init__(self):
# Risk settings from env or defaults
self.max_position_size_pct = float(os.getenv("MAX_POSITION_SIZE_PERCENT", "5.0"))
self.max_daily_drawdown_pct = float(os.getenv("MAX_DAILY_DRAWDOWN_PERCENT", "3.0"))
self.stop_loss_pct = float(os.getenv("STOP_LOSS_PERCENT", "15.0"))
self.take_profit_pct = float(os.getenv("TAKE_PROFIT_PERCENT", "30.0"))
logger.info(
f"Risk Engine initialized: MaxPos={self.max_position_size_pct}%, "
f"MaxDrawdown={self.max_daily_drawdown_pct}%, "
f"SL={self.stop_loss_pct}%, TP={self.take_profit_pct}%"
)
def evaluate_trade(
self,
symbol: str,
qty: int,
premium_per_contract: float,
side: str,
account_info: Dict[str, Any]
) -> Tuple[bool, str]:
"""
Evaluate if a proposed trade satisfies risk parameters.
Returns:
(is_approved: bool, reason: str)
"""
equity = account_info.get("equity", 100000.0)
cash = account_info.get("cash", 100000.0)
today_pl = account_info.get("today_pl", 0.0)
# Calculate cost/credit of trade
# Option contract represents 100 shares
trade_value = premium_per_contract * qty * 100
# 1. Daily Drawdown Check
# Reject new trades if today's drawdown exceeds limit
drawdown_limit = equity * (self.max_daily_drawdown_pct / 100.0)
if today_pl < 0 and abs(today_pl) >= drawdown_limit:
reason = f"Daily drawdown limit of {self.max_daily_drawdown_pct}% exceeded (Current today's P&L: ${today_pl:.2f}). Trading blocked."
logger.warning(reason)
return False, reason
# 2. Position Size Limit Check
# A single position should not exceed MAX_POSITION_SIZE_PERCENT of total equity
max_position_size = equity * (self.max_position_size_pct / 100.0)
if side.lower() == "buy":
# For buying options, the risk is the premium paid
if trade_value > max_position_size:
reason = (
f"Position size exceeds limit. Proposed cost: ${trade_value:.2f}, "
f"Max allowed ({self.max_position_size_pct}% of equity): ${max_position_size:.2f}."
)
logger.warning(reason)
return False, reason
# Cash availability check
if trade_value > cash:
reason = f"Insufficient cash to execute trade. Cost: ${trade_value:.2f}, Cash: ${cash:.2f}."
logger.warning(reason)
return False, reason
else: # Selling option (premium collection / write)
# In short options, margin requirement or collateral must be evaluated.
# For simplicity, we limit short premium trades to the same position size
# based on collateral value or margin impact.
if trade_value > max_position_size:
reason = (
f"Premium size exceeds limit. Proposed margin value: ${trade_value:.2f}, "
f"Max allowed ({self.max_position_size_pct}% of equity): ${max_position_size:.2f}."
)
logger.warning(reason)
return False, reason
return True, "Trade satisfies all Sentry Risk limits."
def check_portfolio_risk(self, active_positions: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
Scan active positions and return positions that breach Stop Loss or Take Profit targets.
Returns a list of trade suggestions to close positions.
"""
exit_signals = []
for pos in active_positions:
unrealized_plpc = pos.get("unrealized_plpc", 0.0) * 100 # convert to %
symbol = pos.get("symbol")
qty = pos.get("qty", 0)
# Stop Loss (SL) check (Negative P&L)
if unrealized_plpc <= -self.stop_loss_pct:
logger.warning(f"Stop Loss breached for {symbol}: {unrealized_plpc:.2f}% (Limit: -{self.stop_loss_pct}%)")
exit_signals.append({
"symbol": symbol,
"qty": abs(qty),
"side": "sell" if qty > 0 else "buy",
"reason": f"Stop Loss target (-{self.stop_loss_pct}%) hit at {unrealized_plpc:.2f}%."
})
# Take Profit (TP) check (Positive P&L)
elif unrealized_plpc >= self.take_profit_pct:
logger.info(f"Take Profit hit for {symbol}: {unrealized_plpc:.2f}% (Limit: +{self.take_profit_pct}%)")
exit_signals.append({
"symbol": symbol,
"qty": abs(qty),
"side": "sell" if qty > 0 else "buy",
"reason": f"Take Profit target (+{self.take_profit_pct}%) hit at {unrealized_plpc:.2f}%."
})
return exit_signals
|