354 lines
13 KiB
Python
354 lines
13 KiB
Python
"""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.
|
||
|
||
Also provides v3 regime-aware stop loss and take profit (Requirements 16.1–16.5).
|
||
|
||
All public methods are synchronous (pure computation, no DB access).
|
||
Persistence is handled by the caller (engine.py).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass
|
||
from datetime import datetime, timezone
|
||
|
||
from services.trading.models import (
|
||
OpenPosition,
|
||
RiskTierConfig,
|
||
StopLevels,
|
||
StopTrigger,
|
||
)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# v3 Regime-Aware Stop Loss and Take Profit (Requirements 16.1–16.5)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class V3StopLevels:
|
||
"""v3 stop-loss and take-profit levels computed from regime-aware volatility."""
|
||
|
||
stop_loss: float
|
||
take_profit: float
|
||
stop_distance_pct: float
|
||
reward_ratio: float
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class TrailingStopResult:
|
||
"""Result of trailing stop computation."""
|
||
|
||
trailing_stop: float
|
||
activated: bool
|
||
|
||
|
||
def compute_v3_stops(
|
||
entry_price: float,
|
||
atr_pct: float,
|
||
regime_atr_mult: float,
|
||
sigma_h: float,
|
||
reward_ratio: float,
|
||
) -> V3StopLevels:
|
||
"""Compute regime-aware stop loss and take profit.
|
||
|
||
stop_distance_pct = max(ATR_pct × regime_ATR_mult, sigma_h × 1.25, 0.005)
|
||
stop_loss = entry_price × (1 - stop_distance_pct)
|
||
take_profit = entry_price × (1 + b × stop_distance_pct)
|
||
|
||
Requirements: 16.1, 16.2, 16.3
|
||
"""
|
||
# Requirement 16.1: stop distance from regime-aware volatility
|
||
z_stop = 1.25
|
||
min_stop_pct = 0.005
|
||
stop_distance_pct = max(atr_pct * regime_atr_mult, sigma_h * z_stop, min_stop_pct)
|
||
|
||
# Requirement 16.2: stop loss for long position
|
||
stop_loss = entry_price * (1.0 - stop_distance_pct)
|
||
|
||
# Requirement 16.3: take profit using dynamic reward ratio
|
||
take_profit = entry_price * (1.0 + reward_ratio * stop_distance_pct)
|
||
|
||
return V3StopLevels(
|
||
stop_loss=stop_loss,
|
||
take_profit=take_profit,
|
||
stop_distance_pct=stop_distance_pct,
|
||
reward_ratio=reward_ratio,
|
||
)
|
||
|
||
|
||
def compute_trailing_stop(
|
||
existing_stop: float,
|
||
current_price: float,
|
||
entry_price: float,
|
||
take_profit: float,
|
||
atr_pct: float,
|
||
trailing_atr_mult: float,
|
||
sigma_h: float,
|
||
) -> TrailingStopResult:
|
||
"""Compute trailing stop level.
|
||
|
||
Activated when unrealized_gain >= 0.50 × TP distance.
|
||
trailing_stop = max(existing_stop, current_price × (1 - trailing_distance_pct))
|
||
trailing_distance_pct = max(ATR_pct × trailing_ATR_mult, sigma_h × 0.75)
|
||
|
||
Trailing stop is monotonically non-decreasing.
|
||
|
||
Requirements: 16.4, 16.5
|
||
"""
|
||
# Requirement 16.4: activation check
|
||
take_profit_distance = take_profit - entry_price
|
||
unrealized_gain = current_price - entry_price
|
||
|
||
# Activation threshold: gain >= 50% of TP distance
|
||
if take_profit_distance <= 0 or unrealized_gain < 0.50 * take_profit_distance:
|
||
# Not activated — return existing stop unchanged
|
||
return TrailingStopResult(trailing_stop=existing_stop, activated=False)
|
||
|
||
# Requirement 16.5: compute trailing stop
|
||
trailing_distance_pct = max(atr_pct * trailing_atr_mult, sigma_h * 0.75)
|
||
candidate_stop = current_price * (1.0 - trailing_distance_pct)
|
||
|
||
# Monotonically non-decreasing: never lower than existing stop
|
||
trailing_stop = max(existing_stop, candidate_stop)
|
||
|
||
return TrailingStopResult(trailing_stop=trailing_stop, activated=True)
|
||
|
||
|
||
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.now(tz=timezone.utc),
|
||
)
|
||
|
||
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.now(tz=timezone.utc),
|
||
)
|
||
|
||
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.now(tz=timezone.utc),
|
||
)
|
||
|
||
return updated
|