"""Stop-loss and take-profit management for the autonomous trading engine. Computes initial stop/take-profit levels from ATR and risk tier parameters, re-evaluates levels when volatility or market conditions change, detects price crossings that should trigger exits, and tightens stops under high-heat or high-severity-event conditions. All public methods are synchronous (pure computation, no DB access). Persistence is handled by the caller (engine.py). """ from __future__ import annotations from datetime import datetime from services.trading.models import ( OpenPosition, RiskTierConfig, StopLevels, StopTrigger, ) class StopLossManager: """Compute and maintain dynamic stop-loss / take-profit levels.""" # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def compute_initial_levels( self, entry_price: float, atr: float, risk_tier: RiskTierConfig, is_micro_trade: bool = False, ) -> StopLevels: """Compute initial stop-loss and take-profit for a new position. For standard trades the risk tier's ATR multiplier and reward/risk ratio are used. Micro-trades use a tighter 1.0x ATR multiplier and 1.5x stop distance for the take-profit target. """ if is_micro_trade: atr_multiplier = 1.0 reward_risk_ratio = 1.5 else: atr_multiplier = risk_tier.stop_loss_atr_multiplier reward_risk_ratio = risk_tier.reward_risk_ratio stop_distance = atr * atr_multiplier stop_loss_price = entry_price - stop_distance if is_micro_trade: take_profit_price = entry_price + (stop_distance * reward_risk_ratio) else: take_profit_price = entry_price + (stop_distance * reward_risk_ratio) return StopLevels( stop_loss_price=stop_loss_price, take_profit_price=take_profit_price, trailing_stop_active=False, atr_value=atr, atr_multiplier=atr_multiplier, reward_risk_ratio=reward_risk_ratio, last_updated=datetime.utcnow(), ) def re_evaluate_levels( self, position: OpenPosition, current_price: float, atr: float, risk_tier: RiskTierConfig, last_levels: StopLevels, high_severity_event: bool = False, earnings_within_3_days: bool = False, portfolio_heat_pct: float = 0.0, max_portfolio_heat: float = 0.20, ) -> StopLevels | None: """Re-evaluate stop/take-profit levels for an open position. Returns updated ``StopLevels`` when a material change is needed, or ``None`` when the existing levels are still appropriate. Material change triggers: * ATR shift > 10 % * Trailing stop activation (price moved > 50 % of TP distance) * Earnings proximity tightening (0.7x ATR multiplier) * High-severity macro event tightening (0.5x normal multiplier) * Proactive heat tightening (heat > 80 % of max) """ entry_price = position.entry_price # --- Determine effective ATR multiplier ----------------------- base_multiplier = risk_tier.stop_loss_atr_multiplier if high_severity_event: effective_multiplier = base_multiplier * 0.5 elif earnings_within_3_days: effective_multiplier = base_multiplier * 0.7 else: effective_multiplier = base_multiplier # --- Trailing stop check -------------------------------------- trailing_stop_active = last_levels.trailing_stop_active tp_distance = last_levels.take_profit_price - entry_price favorable_move = current_price - entry_price if tp_distance > 0 and favorable_move > 0.5 * tp_distance: trailing_stop_active = True # --- Compute candidate stop-loss ------------------------------ candidate_stop = entry_price - (atr * effective_multiplier) # If trailing stop is active, floor the stop at entry (breakeven) if trailing_stop_active: candidate_stop = max(candidate_stop, entry_price) # --- Proactive heat tightening -------------------------------- # When portfolio heat exceeds 80% of max, tighten further if max_portfolio_heat > 0 and portfolio_heat_pct > 0.8 * max_portfolio_heat: heat_tightening_factor = 0.7 tightened_stop = entry_price - ( atr * effective_multiplier * heat_tightening_factor ) if trailing_stop_active: tightened_stop = max(tightened_stop, entry_price) candidate_stop = max(candidate_stop, tightened_stop) # --- Decide whether the change is material -------------------- atr_change_pct = ( abs(atr - last_levels.atr_value) / last_levels.atr_value if last_levels.atr_value > 0 else 0.0 ) trailing_changed = trailing_stop_active != last_levels.trailing_stop_active multiplier_changed = effective_multiplier != last_levels.atr_multiplier if not trailing_changed and not multiplier_changed and atr_change_pct < 0.10: return None # no material change # --- Compute new take-profit ---------------------------------- stop_distance = atr * effective_multiplier reward_risk_ratio = risk_tier.reward_risk_ratio candidate_tp = entry_price + (stop_distance * reward_risk_ratio) return StopLevels( stop_loss_price=candidate_stop, take_profit_price=candidate_tp, trailing_stop_active=trailing_stop_active, atr_value=atr, atr_multiplier=effective_multiplier, reward_risk_ratio=reward_risk_ratio, last_updated=datetime.utcnow(), ) def check_price_crossings( self, positions: list[OpenPosition], prices: dict[str, float], stop_levels: dict[str, StopLevels], ) -> list[StopTrigger]: """Return triggers for positions whose price has crossed a level. A ``StopTrigger`` is emitted when: * current price <= stop_loss_price → ``"stop_loss"`` * current price >= take_profit_price → ``"take_profit"`` """ triggers: list[StopTrigger] = [] for pos in positions: current_price = prices.get(pos.ticker) if current_price is None: continue levels = stop_levels.get(pos.ticker) if levels is None: continue if current_price <= levels.stop_loss_price: triggers.append( StopTrigger( ticker=pos.ticker, trigger_type="stop_loss", current_price=current_price, trigger_price=levels.stop_loss_price, ) ) elif current_price >= levels.take_profit_price: triggers.append( StopTrigger( ticker=pos.ticker, trigger_type="take_profit", current_price=current_price, trigger_price=levels.take_profit_price, ) ) return triggers def tighten_for_heat( self, positions: list[OpenPosition], stop_levels: dict[str, StopLevels], portfolio_heat: float, max_heat: float, active_pool: float, ) -> dict[str, StopLevels]: """Tighten stops on lowest-confidence positions when heat is high. When ``portfolio_heat > 0.8 * max_heat``, the lowest-confidence positions get their stops tightened first (moved closer to current price) to reduce overall portfolio heat. Returns a *new* dict containing only the tickers whose levels were actually changed. """ if max_heat <= 0 or portfolio_heat <= 0.8 * max_heat: return {} # Sort positions by confidence ascending (lowest first) sorted_positions = sorted(positions, key=lambda p: p.signal_confidence) updated: dict[str, StopLevels] = {} for pos in sorted_positions: levels = stop_levels.get(pos.ticker) if levels is None: continue # Tighten by reducing the stop distance by 30 % heat_factor = 0.7 new_stop_distance = levels.atr_value * levels.atr_multiplier * heat_factor new_stop = pos.entry_price - new_stop_distance # Never move stop further away from current price new_stop = max(new_stop, levels.stop_loss_price) # If trailing stop is active, floor at entry if levels.trailing_stop_active: new_stop = max(new_stop, pos.entry_price) if new_stop != levels.stop_loss_price: updated[pos.ticker] = StopLevels( stop_loss_price=new_stop, take_profit_price=levels.take_profit_price, trailing_stop_active=levels.trailing_stop_active, atr_value=levels.atr_value, atr_multiplier=levels.atr_multiplier, reward_risk_ratio=levels.reward_risk_ratio, last_updated=datetime.utcnow(), ) return updated