""" Trade Executor Executes live trades via Alpaca broker with position management. ENHANCED with: - Proper correlation-based position limits (using max_correlated_positions config) - Drawdown-scaled position sizing - Enforced stop losses with tighter defaults - Active position reduction on circuit breaker / drawdown stop - Per-sector position limits - HARD maximum SHORT position cap (broker-level enforcement) - Per-position stop loss defaults (ATR-based, always set) - Broker-position sync check (authoritative position counting) - Yesterday-close drawdown circuit breaker (faster response) """ from datetime import datetime from typing import Dict, List, Optional, Tuple from loguru import logger import numpy as np class TradingExecutor: """ Executes live trades based on RL agent decisions. Manages positions, stop losses, and take profits. """ def __init__(self, broker, store, safety, config: Dict): """ Args: broker: AlpacaBroker instance store: DataStore instance safety: SafetyManager instance config: Trading configuration dict """ self.broker = broker self.store = store self.safety = safety self.config = config self.commission_rate = config.get('commission_rate', 0.001) # Stop loss / take profit defaults (can be overridden by strategy) self.default_stop_loss_pct = config.get('stop_loss_pct', 1.0) / 100 self.default_take_profit_pct = config.get('take_profit_pct', 1.5) / 100 # Track recent prices for volatility calculation self.recent_returns: Dict[str, List[float]] = {} # Sector map for correlation self.sector_map = { 'SPY': 'broad_market', 'QQQ': 'tech', 'IWM': 'small_cap', 'DIA': 'blue_chip', 'AAPL': 'tech', 'MSFT': 'tech', 'GOOGL': 'tech', 'AMZN': 'tech', 'NVDA': 'tech', 'META': 'tech', 'TSLA': 'auto/energy', 'JPM': 'financials', 'BAC': 'financials', 'GS': 'financials', 'MS': 'financials', 'JNJ': 'healthcare', 'UNH': 'healthcare', 'WMT': 'consumer', 'PG': 'consumer', 'XLE': 'energy', 'CVX': 'energy', 'XOM': 'energy', 'LMT': 'defense', 'RTX': 'defense', 'NOC': 'defense', 'GD': 'defense', 'USO': 'commodities', 'GLD': 'metals', } def _get_sector(self, symbol: str) -> str: return self.sector_map.get(symbol, 'other') def _update_volatility(self, symbol: str, current_price: float): """Track price returns for volatility calculation""" if symbol not in self.recent_returns: self.recent_returns[symbol] = [] # Returns are tracked externally; just store prices # Volatility is computed by safety manager def _apply_drawdown_scaling(self, invest_amount: float, portfolio_value: float) -> float: """Scale down position size based on current drawdown""" scale = self.safety.get_position_scale(portfolio_value) vol_mult = self.safety.current_volatility_mult return invest_amount * scale * vol_mult def _check_correlation_limits(self, symbol: str, direction: str, open_positions: List[Dict]) -> Tuple[bool, str]: """ Check if adding this position would violate correlation limits. Returns: (allowed: bool, reason: str) """ max_corr = self.safety.max_correlated_positions max_same_dir = self.safety.max_same_direction # Count current positions in same direction same_dir_positions = [ p for p in open_positions if p.get('metadata', {}).get('direction') == direction ] # Check same-direction limit if len(same_dir_positions) >= max_same_dir: return False, (f"Max {max_same_dir} {direction} positions reached " f"(have {len(same_dir_positions)})") # Check sector correlation limit sector = self._get_sector(symbol) if sector != 'other' and sector != 'broad_market': same_sector_same_dir = [ p for p in same_dir_positions if self._get_sector(p.get('symbol', '')) == sector ] if len(same_sector_same_dir) >= max_corr: return False, (f"Max {max_corr} {direction} positions in " f"{sector} sector (have {len(same_sector_same_dir)})") # Also limit broad market ETF correlation (SPY, QQQ, IWM, DIA all correlate) if symbol in ('SPY', 'QQQ', 'IWM', 'DIA'): market_etfs = [p for p in same_dir_positions if p.get('symbol') in ('SPY', 'QQQ', 'IWM', 'DIA')] if len(market_etfs) >= 2: return False, f"Max 2 market ETFs as {direction} positions (have {len(market_etfs)})" return True, "OK" def _enforce_stop_loss_tightness(self, stop_loss: float, entry_price: float, direction: str) -> float: """ Ensure stop loss is tight enough - don't let wide stops blow up risk. If strategy provides too-wide stop, override with our default. """ if stop_loss is None: return None if direction == 'short': # Short stop: price goes UP to hit stop (bad) stop_distance_pct = (stop_loss - entry_price) / entry_price else: # Long stop: price goes DOWN to hit stop (bad) stop_distance_pct = (entry_price - stop_loss) / entry_price # If strategy stop is wider than 3x our default, use our default instead if stop_distance_pct > self.default_stop_loss_pct * 3: logger.warning(f"Stop loss {stop_distance_pct:.2%} too wide, " f"using default {self.default_stop_loss_pct:.2%}") if direction == 'short': return entry_price * (1 + self.default_stop_loss_pct) else: return entry_price * (1 - self.default_stop_loss_pct) return stop_loss def execute_signal(self, symbol: str, action: int, current_price: float, strategy_params: Dict = None, combined_equity: float = None) -> Optional[Dict]: """ Execute a trading action from the RL agent. Actions: 0: Hold 1: Buy 25% of available capital (go long) 2: Buy 50% of available capital (go long) 3: Close 50% of position (long or short) 4: Close 100% of position (long or short) 5: Short 25% of available capital 6: Short 50% of available capital Args: combined_equity: Optional total equity across all brokers (for multi-broker setups) If provided, used for safety checks instead of single-broker equity Returns: trade record dict, or None if no action taken """ if action == 0: # Hold return None if current_price <= 0: return None strategy_params = strategy_params or {} # Get portfolio state try: portfolio = self.broker.get_portfolio() equity = portfolio['equity'] cash = portfolio['cash'] except Exception as e: logger.error(f"Error getting portfolio: {e}") return None # Use combined equity for safety checks if provided (multi-broker setup) safety_equity = combined_equity if combined_equity is not None else equity # Update safety peak self.safety.update_peak_equity(safety_equity) # Get current open positions from STORE (known trades) open_positions = self.store.get_open_positions() open_for_symbol = [p for p in open_positions if p['symbol'] == symbol] # CRITICAL: Get ACTUAL broker positions (authoritative) for hard cap checks # Store only knows what WE opened — broker knows everything (including manual trades) try: broker_positions = self.broker.get_positions() except Exception as e: logger.warning(f"Could not get broker positions for hard cap check: {e}") broker_positions = [] # HARD SHORT CAP CHECK: If broker already has max_short_positions shorts, block more # This is the PRIMARY defense against the "13 shorts" scenario broker_short_count = sum( 1 for p in broker_positions if p.get('side') == 'sell_short' or p.get('qty', 0) < 0 ) if action in (5, 6) and broker_short_count >= self.safety.max_short_positions: logger.warning( f"HARD CAP: Broker has {broker_short_count} shorts " f"(max={self.safety.max_short_positions}). Short blocked." ) return None # BROKER POSITION SYNC WARNING store_short_count = sum( 1 for p in open_positions if p.get('metadata', {}).get('direction') == 'short' ) if broker_short_count != store_short_count: logger.warning( f"BROKER/STORE SYNC MISMATCH: Broker={broker_short_count} shorts, " f"Store={store_short_count} shorts. Using broker as authoritative." ) # BUY LONG actions if action in (1, 2): if open_for_symbol: logger.debug(f"Already have position in {symbol}, skipping buy") return None # CORRELATION CHECK: Don't pile into same direction (uses store positions) allowed, reason = self._check_correlation_limits(symbol, 'long', open_positions) if not allowed: logger.debug(f"Correlation blocked LONG {symbol}: {reason}") return None pct = 0.25 if action == 1 else 0.50 invest = cash * pct # Apply drawdown + volatility scaling invest = self._apply_drawdown_scaling(invest, safety_equity) max_position_pct = self.config.get('max_position_pct', 8) / 100 effective_max = max_position_pct * self.safety.get_position_scale(safety_equity) max_invest = safety_equity * effective_max if invest > max_invest: invest = max_invest if invest < self.config.get('min_trade_value', 50): logger.debug(f"Invest amount ${invest:.2f} below minimum") return None shares = int(invest / current_price) if shares < 1: # Try fractional shares = round(invest / current_price, 4) if shares * current_price < self.config.get('min_trade_value', 50): logger.debug(f"Share value ${shares * current_price:.2f} below minimum") return None allowed, reason = self.safety.validate_trade( symbol, 'buy', shares, current_price, safety_equity, open_positions, broker_positions=broker_positions ) if not allowed: logger.debug(f"Trade blocked by safety: {reason}") return None try: order = self.broker.place_market_order(symbol, shares, 'buy') logger.info(f"BUY {shares} {symbol} @ ~${current_price:.4f}") except Exception as e: logger.error(f"Error placing buy order: {e}") return None # Stop loss and take profit with tightness enforcement + defaults stop_loss = strategy_params.get('stop_loss') take_profit = strategy_params.get('take_profit') # CRITICAL: If GA didn't provide stop loss, use SAFETY MANAGER DEFAULTS # This ensures EVERY position has a stop loss — the #1 defense against blowup if stop_loss is None: stop_loss = self.safety.get_default_stop_loss(current_price, 'long') logger.info(f"Applying DEFAULT stop loss for LONG {symbol}: ${stop_loss:.4f} " f"({((current_price - stop_loss) / current_price) * 100:.2f}% from entry)") # Enforce tightness ceiling on the stop stop_loss = self._enforce_stop_loss_tightness(stop_loss, current_price, 'long') # CRITICAL: If GA didn't provide take profit, use SAFETY MANAGER DEFAULTS if take_profit is None or take_profit == current_price: take_profit = self.safety.get_default_take_profit(current_price, 'long') logger.info(f"Applying DEFAULT take profit for LONG {symbol}: ${take_profit:.4f}") if take_profit and take_profit != current_price: tp_distance = (take_profit - current_price) / current_price if tp_distance < self.default_take_profit_pct * 0.5: # TP too tight relative to default take_profit = current_price * (1 + self.default_take_profit_pct) trade = { 'symbol': symbol, 'side': 'buy', 'amount': shares, 'entry_price': current_price, 'entry_time': datetime.utcnow().isoformat(), 'strategy_id': strategy_params.get('strategy_id', 'auto'), 'stop_loss': stop_loss, 'take_profit': take_profit, 'order_id': str(order.get('id', '')), 'status': 'open', 'metadata': { 'action': action, 'invest_pct': pct, 'equity_at_entry': equity, 'direction': 'long', 'sector': self._get_sector(symbol), }, } trade_id = self.store.record_trade(trade) trade['id'] = trade_id return trade # CLOSE POSITION actions (long or short) elif action in (3, 4): if not open_for_symbol: return None position = open_for_symbol[0] total_shares = position['amount'] is_short = position.get('metadata', {}).get('direction') == 'short' if action == 3: close_shares = abs(total_shares) * 0.5 else: close_shares = abs(total_shares) close_shares = round(close_shares, 4) if close_shares * current_price < self.config.get('min_trade_value', 50): close_shares = abs(total_shares) close_side = 'buy' if is_short else 'sell' try: order = self.broker.place_market_order(symbol, close_shares, close_side) logger.info(f"CLOSE({close_side.upper()}) {close_shares} {symbol} @ ~${current_price:.4f}") except Exception as e: logger.error(f"Error placing close order: {e}") return None # Calculate P&L if is_short: pnl = (position['entry_price'] - current_price) * close_shares else: pnl = (current_price - position['entry_price']) * close_shares if close_shares >= abs(total_shares) * 0.99: self.store.close_position( position['id'], current_price, datetime.utcnow(), fees=close_shares * current_price * self.commission_rate ) self.safety.record_trade_result(pnl) return { 'symbol': symbol, 'side': close_side, 'amount': close_shares, 'exit_price': current_price, 'pnl': round(pnl, 2), 'pnl_pct': round(pnl / (position['entry_price'] * close_shares) * 100, 2), 'action': action, 'direction': 'short' if is_short else 'long', } else: self.safety.record_trade_result(pnl * (close_shares / abs(total_shares))) self.store.reduce_position( position['id'], close_shares, current_price, datetime.utcnow(), fees=close_shares * current_price * self.commission_rate ) return { 'symbol': symbol, 'side': close_side, 'amount': close_shares, 'exit_price': current_price, 'pnl': round(pnl, 2), 'action': action, 'direction': 'short' if is_short else 'long', 'reduced': True, } # SHORT actions elif action in (5, 6): if open_for_symbol: logger.debug(f"Already have position in {symbol}, skipping short") return None # CORRELATION CHECK allowed, reason = self._check_correlation_limits(symbol, 'short', open_positions) if not allowed: logger.debug(f"Correlation blocked SHORT {symbol}: {reason}") return None pct = 0.25 if action == 5 else 0.50 invest = cash * pct # Apply drawdown + volatility scaling invest = self._apply_drawdown_scaling(invest, safety_equity) max_position_pct = self.config.get('max_position_pct', 8) / 100 effective_max = max_position_pct * self.safety.get_position_scale(safety_equity) max_invest = safety_equity * effective_max if invest > max_invest: invest = max_invest if invest < self.config.get('min_trade_value', 50): logger.debug(f"Short invest amount ${invest:.2f} below minimum") return None shares = int(invest / current_price) if shares < 1: shares = round(invest / current_price, 4) if shares * current_price < self.config.get('min_trade_value', 50): logger.debug(f"Short share value ${shares * current_price:.2f} below minimum") return None allowed, reason = self.safety.validate_trade( symbol, 'sell', shares, current_price, safety_equity, open_positions, broker_positions=broker_positions ) if not allowed: logger.debug(f"Short blocked by safety: {reason}") return None try: order = self.broker.place_market_order(symbol, shares, 'sell') logger.info(f"SHORT {shares} {symbol} @ ~${current_price:.4f}") except Exception as e: logger.error(f"Error placing short order: {e}") return None # For shorts: stop_loss is ABOVE entry (price rises = bad), take_profit is BELOW stop_loss = strategy_params.get('short_stop_loss') or strategy_params.get('stop_loss') take_profit = strategy_params.get('short_take_profit') or strategy_params.get('take_profit') # CRITICAL: If GA didn't provide stop loss, use SAFETY MANAGER DEFAULTS if stop_loss is None: stop_loss = self.safety.get_default_stop_loss(current_price, 'short') logger.info(f"Applying DEFAULT stop loss for SHORT {symbol}: ${stop_loss:.4f} " f"({((stop_loss - current_price) / current_price) * 100:.2f}% from entry)") # Enforce tightness ceiling on the stop stop_loss = self._enforce_stop_loss_tightness(stop_loss, current_price, 'short') # CRITICAL: If GA didn't provide take profit, use SAFETY MANAGER DEFAULTS if take_profit is None: take_profit = self.safety.get_default_take_profit(current_price, 'short') logger.info(f"Applying DEFAULT take profit for SHORT {symbol}: ${take_profit:.4f}") trade = { 'symbol': symbol, 'side': 'sell', 'amount': shares, 'entry_price': current_price, 'entry_time': datetime.utcnow().isoformat(), 'strategy_id': strategy_params.get('strategy_id', 'auto'), 'stop_loss': stop_loss, 'take_profit': take_profit, 'order_id': str(order.get('id', '')), 'status': 'open', 'metadata': { 'action': action, 'invest_pct': pct, 'equity_at_entry': equity, 'direction': 'short', 'sector': self._get_sector(symbol), }, } trade_id = self.store.record_trade(trade) trade['id'] = trade_id return trade return None def check_exits(self, current_prices: Dict[str, float]) -> List[Dict]: """ Check all open positions for stop loss / take profit exits. Also checks for drawdown stop and forces partial closes if needed. Returns list of closed trades. """ closed = [] open_positions = self.store.get_open_positions() # Get combined portfolio for drawdown check try: portfolio = self.broker.get_portfolio() equity = portfolio['equity'] except Exception: equity = 0 # Update yesterday's equity reference at start of each day self.safety.update_yesterday_equity(equity) # Check if drawdown stop is triggered (peak-equity based, slow response) should_reduce_dd, dd_reason, reduce_pct = False, "", 0.0 if equity > 0 and open_positions: should_reduce_dd, dd_reason, reduce_pct = \ self.safety.check_drawdown_stop(equity, open_positions) # NEW: Check YESTERDAY-CLOSE drawdown (FASTER response, catches intraday blowups) should_reduce_yc, yc_reason, yc_reduce_pct = False, "", 0.0 if equity > 0 and open_positions: should_reduce_yc, yc_reason, yc_reduce_pct = \ self.safety.check_yesterday_close_drawdown(equity) # Use whichever is more urgent if should_reduce_yc and not should_reduce_dd: should_reduce_dd = True dd_reason = yc_reason reduce_pct = yc_reduce_pct logger.warning(f"YESTERDAY-CLOSE CIRCUIT BREAKER TRIGGERED: {yc_reason}") # CRITICAL: If we have 10+ positions all losing, force reduce regardless of drawdown # This catches the "13 shorts all losing" scenario before drawdown thresholds are hit if len(open_positions) >= 10: losing_count = 0 for pos in open_positions: sym = pos.get('symbol') px = current_prices.get(sym) if px and pos.get('entry_price'): is_short = pos.get('metadata', {}).get('direction') == 'short' if is_short and px > pos['entry_price']: losing_count += 1 elif not is_short and px < pos['entry_price']: losing_count += 1 if losing_count >= 8: logger.critical( f"CRISIS MODE: {losing_count}/{len(open_positions)} positions losing money! " f"Force-reducing 50% of all positions immediately." ) return self.force_reduce_all_positions(0.50, current_prices) for position in open_positions: symbol = position['symbol'] price = current_prices.get(symbol) if price is None: continue should_exit = False exit_reason = '' should_reduce = False reduce_amount = 0 is_short = position.get('metadata', {}).get('direction') == 'short' # Normal stop loss / take profit checks if is_short: if position.get('stop_loss') and price >= position['stop_loss']: should_exit = True exit_reason = 'stop_loss' elif position.get('take_profit') and price <= position['take_profit']: should_exit = True exit_reason = 'take_profit' else: if position.get('stop_loss') and price <= position['stop_loss']: should_exit = True exit_reason = 'stop_loss' elif position.get('take_profit') and price >= position['take_profit']: should_exit = True exit_reason = 'take_profit' # Drawdown stop: force partial close if not should_exit and should_reduce_dd: should_reduce = True reduce_amount = abs(position['amount']) * reduce_pct exit_reason = f'drawdown_stop' if should_exit: close_side = 'buy' if is_short else 'sell' try: self.broker.place_market_order( symbol, position['amount'], close_side ) logger.info(f"EXIT ({exit_reason}) {symbol} @ ${price:.4f} [{'SHORT' if is_short else 'LONG'}]") except Exception as e: logger.error(f"Error executing exit for {symbol}: {e}") continue if is_short: pnl = (position['entry_price'] - price) * position['amount'] else: pnl = (price - position['entry_price']) * position['amount'] self.store.close_position( position['id'], price, datetime.utcnow(), fees=position['amount'] * price * self.commission_rate ) self.safety.record_trade_result(pnl) closed.append({ 'symbol': symbol, 'side': close_side, 'amount': position['amount'], 'entry_price': position['entry_price'], 'exit_price': price, 'pnl': round(pnl, 2), 'pnl_pct': round(pnl / (position['entry_price'] * position['amount']) * 100, 2), 'exit_reason': exit_reason, 'direction': 'short' if is_short else 'long', }) elif should_reduce and reduce_amount > 0: # Partial close due to drawdown stop reduce_amount = round(reduce_amount, 4) if reduce_amount < 0.0001: continue close_side = 'buy' if is_short else 'sell' try: self.broker.place_market_order( symbol, reduce_amount, close_side ) logger.warning(f"DRAWDOWN REDUCE ({reduce_amount:.2f} of {position['amount']:.2f}) " f"{symbol} @ ${price:.4f}") except Exception as e: logger.error(f"Error reducing position for {symbol}: {e}") continue if is_short: pnl = (position['entry_price'] - price) * reduce_amount else: pnl = (price - position['entry_price']) * reduce_amount self.store.reduce_position( position['id'], reduce_amount, price, datetime.utcnow(), fees=reduce_amount * price * self.commission_rate ) self.safety.record_trade_result(pnl * (reduce_amount / abs(position['amount']))) closed.append({ 'symbol': symbol, 'side': close_side, 'amount': reduce_amount, 'entry_price': position['entry_price'], 'exit_price': price, 'pnl': round(pnl, 2), 'exit_reason': 'drawdown_reduce', 'direction': 'short' if is_short else 'long', }) return closed def force_reduce_all_positions(self, reduction_pct: float, current_prices: Dict[str, float]) -> List[Dict]: """ Force-reduce ALL positions by reduction_pct (e.g., close 50% of everything). Used when drawdown stop triggers. """ closed = [] open_positions = self.store.get_open_positions() for position in open_positions: symbol = position['symbol'] price = current_prices.get(symbol) if price is None: continue reduce_amount = round(abs(position['amount']) * reduction_pct, 4) if reduce_amount < 0.0001: continue is_short = position.get('metadata', {}).get('direction') == 'short' close_side = 'buy' if is_short else 'sell' try: self.broker.place_market_order(symbol, reduce_amount, close_side) logger.warning(f"FORCE REDUCE ({reduction_pct:.0%}) {reduce_amount:.2f} {symbol} @ ${price:.4f}") except Exception as e: logger.error(f"Error force-reducing {symbol}: {e}") continue if is_short: pnl = (position['entry_price'] - price) * reduce_amount else: pnl = (price - position['entry_price']) * reduce_amount self.store.reduce_position( position['id'], reduce_amount, price, datetime.utcnow(), fees=reduce_amount * price * self.commission_rate ) self.safety.record_trade_result(pnl * (reduce_amount / abs(position['amount']))) closed.append({ 'symbol': symbol, 'side': close_side, 'amount': reduce_amount, 'exit_price': price, 'pnl': round(pnl, 2), 'exit_reason': 'force_reduce_all', 'direction': 'short' if is_short else 'long', }) return closed def get_portfolio_state(self) -> Dict: """Get current portfolio state for RL agent""" try: portfolio = self.broker.get_portfolio() positions = self.broker.get_positions() open_db = self.store.get_open_positions() total_position_value = sum(p.get('market_value', 0) for p in positions) equity = portfolio['equity'] # Count directions num_long = sum(1 for p in open_db if p.get('metadata', {}).get('direction') == 'long') num_short = sum(1 for p in open_db if p.get('metadata', {}).get('direction') == 'short') return { 'equity': equity, 'cash': portfolio['cash'], 'position_ratio': total_position_value / equity if equity > 0 else 0, 'unrealized_pnl': sum(p.get('unrealized_pl', 0) for p in positions) / equity if equity > 0 else 0, 'time_in_position': 0, 'num_positions': len(positions), 'num_long': num_long, 'num_short': num_short, } except Exception as e: logger.error(f"Error getting portfolio state: {e}") return { 'equity': 0, 'cash': 0, 'position_ratio': 0, 'unrealized_pnl': 0, 'time_in_position': 0, 'num_positions': 0, 'num_long': 0, 'num_short': 0, }