momentum-system/runtime/exit_manager.py

"""Live exit manager — actively manages OPEN positions per strategy/exits.

The static bracket (fixed stop + fixed target) is replaced/augmented here with
the SAME managed-exit rules the backtest swept (strategy/exits.manage_live):
move the stop to breakeven, trail it, scale out, or exit on a first red candle.

Critically, the stop is RATCHETED via a broker order REPLACE (PATCH), never a
cancel — cancelling a bracket leg strips the protection (the naked-stop bug).
Runs on the main loop (not the fast trigger thread): stops update off closed
bars, so 10-15s cadence is plenty, and it keeps slow broker calls off the hot
path.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass

import pandas as pd

from runtime.flatten import cancel_protective_and_close
from storage.event_schema import EventMode, RiskRuleTriggeredEvent
from strategy.exits import ExitConfig, TRAIL_NONE, manage_live

logger = logging.getLogger(__name__)


@dataclass
class _Tracked:
    entry_price: float
    init_stop: float      # original opening-range-low stop (for the R math)
    entry_ts: object      # timestamp we first saw the position (~entry)
    last_stop: float      # last stop we set at the broker
    scaled: bool = False


def _is_active(cfg: ExitConfig) -> bool:
    return bool(cfg and (cfg.breakeven_at_r or cfg.breakeven_at_pct
                         or cfg.trail_mode != TRAIL_NONE
                         or cfg.scale_out_r or cfg.first_red_exit
                         or cfg.profit_lock_tiers))


class LiveExitManager:
    def __init__(self, client, store, bars_fn, cfg: ExitConfig | None = None,
                 session_id=None, mode: EventMode = EventMode.PAPER):
        self.client = client
        self.store = store
        self.bars_fn = bars_fn   # (symbol) -> today's RTH minute bars DataFrame
        self.cfg = cfg or ExitConfig.from_env()
        self.session_id = session_id
        self.mode = mode
        self._tracked: dict[str, _Tracked] = {}

    # -- broker helpers ----------------------------------------------------

    # a bracket stop leg sits in these states once the entry has filled; it is a
    # child of the (now FILLED) entry, so a status="open" query misses it.
    _ACTIVE = {"held", "new", "accepted", "pending_new",
               "accepted_for_bidding", "partially_filled"}

    def _all_orders(self, symbols=None):
        # scope to the held symbols — Alpaca's `limit` keeps only the most-recent
        # orders, so on a busy day an unscoped query drops older positions' stop
        # legs and we'd misread a protected position as naked.
        try:
            return self.client.get_orders(status="all", limit=500, nested=True,
                                          symbols=list(symbols) if symbols else None)
        except Exception:  # noqa: BLE001
            return None

    def _stop_leg(self, symbol: str, orders):
        """(order_id, stop_price) of the live sell-stop protecting symbol.

        Uses pre-fetched ``orders`` (status="all" — the protective leg is a
        'held' child of the FILLED entry, which status="open" wouldn't surface).
        """
        if orders is None:
            return (None, None)
        for o in orders:
            for c in [o, *(o.get("legs") or [])]:
                sym = c.get("symbol") or o.get("symbol")
                if (sym == symbol and c.get("type") in ("stop", "stop_limit")
                        and c.get("side") == "sell"
                        and c.get("status") in self._ACTIVE):
                    sp = c.get("stop_price")
                    return (c.get("id"), float(sp) if sp else None)
        return (None, None)

    def _flatten(self, symbol: str, orders) -> None:
        """Market-close a position, cancelling its resting protective legs first.

        Delegates to the shared helper so the held-qty 403 fix lives in exactly
        one place (see runtime/flatten.py). ``orders`` is the pass's pre-fetched
        nested snapshot, reused to avoid a redundant fetch."""
        cancel_protective_and_close(self.client, symbol, orders=orders)

    def _entry_time(self, symbol: str, orders):
        """UTC-naive fill time of the most recent FILLED buy entry for symbol."""
        if orders is None:
            return None
        best = None
        for o in orders:
            if (o.get("symbol") == symbol and o.get("side") == "buy"
                    and float(o.get("filled_qty") or 0) > 0):
                ft = o.get("filled_at") or o.get("submitted_at")
                if not ft:
                    continue
                try:
                    ts = pd.Timestamp(ft)
                    if ts.tzinfo is not None:
                        ts = ts.tz_convert("UTC").tz_localize(None)
                except Exception:  # noqa: BLE001
                    continue
                if best is None or ts > best:
                    best = ts
        return best

    def _emit(self, symbol: str, message: str, rule: str, state: dict) -> None:
        from datetime import datetime
        self.store.emit(RiskRuleTriggeredEvent(
            timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id,
            message=message, rule_type=rule, rule_value=0.0,
            current_state={"symbol": symbol, **state}, action_taken=rule,
        ))

    # -- main pass ---------------------------------------------------------

    def manage(self) -> list[str]:
        if not _is_active(self.cfg):
            return []  # static bracket — broker OCO handles it, nothing to do
        try:
            positions = self.client.get_positions()
        except Exception:  # noqa: BLE001
            return []
        held = {p.get("symbol"): p for p in positions if p.get("symbol")}
        for sym in list(self._tracked):  # stop tracking closed names
            if sym not in held:
                self._tracked.pop(sym, None)
        if not held:
            return []
        # fetch the orders snapshot ONCE per pass (status="all" is the heavy call;
        # reuse it for every position's stop-leg + entry-time lookups), scoped to
        # the held symbols so a busy day's order volume can't truncate the legs.
        orders = self._all_orders(held.keys())

        actions: list[str] = []
        for sym, p in held.items():
            try:
                bars = self.bars_fn(sym)
            except Exception:  # noqa: BLE001
                continue
            if bars is None or len(bars) == 0:
                continue
            entry = float(p.get("avg_entry_price") or 0.0)
            if entry <= 0:
                continue
            if sym not in self._tracked:
                _leg, cur_stop = self._stop_leg(sym, orders)
                if cur_stop is None or cur_stop >= entry:
                    continue  # no usable protective stop yet (or above entry)
                # trail from the ACTUAL entry fill time (so high-water/R reflect
                # the move since entry, not since we first noticed the position)
                entry_ts = self._entry_time(sym, orders) or _last_ts(bars)
                self._tracked[sym] = _Tracked(entry, cur_stop, entry_ts, cur_stop)
            tr = self._tracked[sym]

            since = _bars_since(bars, tr.entry_ts)
            if len(since) == 0:
                since = bars
            d = manage_live(tr.entry_price, tr.init_stop, since, self.cfg, tr.scaled)

            # (1) first-red: exit the whole position now
            if d.exit_now:
                try:
                    self._flatten(sym, orders)
                    self._emit(sym, f"first-red exit {sym} @~mkt", "exit_first_red",
                               {"reason": d.reason})
                    actions.append(f"{sym} first-red exit")
                    self._tracked.pop(sym, None)
                    continue
                except Exception as exc:  # noqa: BLE001
                    logger.warning("first-red close failed %s: %s", sym, exc)

            # (2) scale-out a fraction into strength
            if d.scale_out_frac > 0 and not tr.scaled:
                qty = int(abs(float(p.get("qty") or 0)))
                sell = int(qty * d.scale_out_frac)
                if sell >= 1:
                    try:
                        self.client.close_position(sym, qty=sell)
                        tr.scaled = True
                        self._emit(sym, f"scaled out {sell} {sym}", "exit_scale_out",
                                   {"qty": sell})
                        actions.append(f"{sym} scaled {sell}")
                    except Exception as exc:  # noqa: BLE001
                        logger.warning("scale-out failed %s: %s", sym, exc)

            # (3) ratchet the protective stop UP (breakeven / trail) via replace
            leg_id, cur_stop = self._stop_leg(sym, orders)
            if cur_stop is not None and d.desired_stop > cur_stop + 0.01:
                last_px = float(p.get("current_price") or 0.0)
                if last_px > 0 and d.desired_stop >= last_px:
                    # the trail is already BREACHED (price fell back through the
                    # prior swing low) — a sell-stop above market would be
                    # rejected, so exit at market now (this is the trail firing).
                    try:
                        self._flatten(sym, orders)
                        self._emit(sym, f"trail breached, exit {sym} @~{last_px:.2f}",
                                   "exit_trail", {"trail": d.desired_stop, "price": last_px})
                        actions.append(f"{sym} trail-exit @{last_px:.2f}")
                        self._tracked.pop(sym, None)
                    except Exception as exc:  # noqa: BLE001
                        logger.warning("trail exit failed %s: %s", sym, exc)
                elif leg_id:
                    try:
                        self.client.replace_order(leg_id, stop_price=round(d.desired_stop, 2))
                        tr.last_stop = d.desired_stop
                        self._emit(sym, f"stop {sym} {cur_stop:.2f} -> {d.desired_stop:.2f}",
                                   "stop_moved", {"from": cur_stop, "to": d.desired_stop})
                        actions.append(f"{sym} stop->{d.desired_stop:.2f}")
                    except Exception as exc:  # noqa: BLE001
                        logger.warning("stop move failed %s: %s", sym, exc)
        return actions


def _last_ts(bars: pd.DataFrame):
    try:
        return pd.Timestamp(bars["timestamp"].iloc[-1])
    except Exception:  # noqa: BLE001
        return None


def _bars_since(bars: pd.DataFrame, ts) -> pd.DataFrame:
    if ts is None or "timestamp" not in bars.columns:
        return bars
    try:
        return bars[pd.to_datetime(bars["timestamp"]) >= pd.Timestamp(ts)].reset_index(drop=True)
    except Exception:  # noqa: BLE001
        return bars