"""Trading execution service (Milestone 4/5 glue). Connects ready signals to broker execution with risk controls and an approval workflow: signal_ready -> risk checks (sizing, max concurrent, daily-loss circuit breaker) -> order_approval_requested -> [manual approval in the dashboard] or [auto-approve] -> order_approved -> broker submit -> order_submitted (+ filled) Exit orders close existing broker positions on demand. """ from __future__ import annotations import functools import json import logging import os import threading from dataclasses import dataclass from datetime import datetime, timedelta from runtime.flatten import cancel_protective_and_close from alpaca_paper.execution import ( AlpacaPaperExecutor, ExecutionRequest, ) from storage.event_schema import ( EventMode, OrderApprovalRequestedEvent, OrderApprovedEvent, OrderRejectedEvent, RiskRuleTriggeredEvent, ) from storage.event_store import EventStore from storage.projections import ( query_account_positions_snapshot, query_approval_queue, query_ready_signals_snapshot, ) from strategy.risk.position_sizing import ( PositionSizingConfig, calculate_position_size, ) from trading_mode import TradingModeSettings logger = logging.getLogger(__name__) def _locked(method): """Serialize a mutating service method behind self._lock (a re-entrant lock). The fast trigger thread (submit_breakout_now) and the main loop (tick, approvals) both mutate execution state (_armed, _requested_symbols) and emit events; without this they would race on the shared psycopg2 connection and the in-memory sets. RLock so a public method may call another (tick -> expire_stale_entries) without deadlocking. """ @functools.wraps(method) def wrapper(self, *args, **kwargs): with self._lock: return method(self, *args, **kwargs) return wrapper @dataclass class ExecutionSettings: enabled: bool = True auto_approve: bool = False max_orders_per_tick: int = 1 max_concurrent_positions: int = 3 risk_per_trade_pct: float = 0.01 default_equity: float = 100_000.0 max_daily_loss_pct: float = 0.03 # on a daily-loss breach, also flatten open positions + cancel unfilled entries flatten_on_breach: bool = True # --- entry mechanism (Ross-Cameron-style; all tunable) ----------------- # reward target as a multiple of risk (entry + reward_multiple * (entry-stop)) reward_multiple: float = 2.0 # how the entry order is placed: "limit" rests at the entry trigger so an # unfilled order is a real, cancellable state; "market" fills immediately entry_order_type: str = "limit" # cancel an unfilled entry after this many minutes of resting (the "back # out" time box; ~1 bar == 1 minute). 0 disables the timeout. Wall-clock # based, so the invalidation guard can run far more often than this. entry_timeout_bars: int = 2 # cancel an unfilled entry if price trades back below the entry trigger by # this fraction. NOT 0.0: at zero tolerance any sub-cent wobble below the # trigger cancels the entry one tick after it fires, so a breakout that # oscillates around the level churns and never holds (observed live). Give # it room — the bracket STOP (opening-range low) is the real protection and # risk is capped at 1%/trade by sizing. Negative disables price-break cancel. entry_invalidate_pct: float = 0.015 # live-trigger fast path (submit_breakout_now): how far above the trigger to # cap the marketable limit so a breakout FILLS on a runner instead of # resting forever at the trigger, while still bounding slippage. trigger_slippage_pct: float = 0.004 # don't cancel an entry within this many seconds of arming — a marketable # fill needs a moment to confirm at the broker; cancelling first strips its # bracket (the naked-stop failure mode). entry_grace_seconds: float = 5.0 # --- account-aware sizing (matters most on a small REAL account) ------- # one position's dollar value <= this fraction of equity, so a single trade # can't exceed buying power (critical at $300: 3 positions must fit, so ~1/3). max_position_pct: float = 0.33 # liquidity cap: shares <= this fraction of the symbol's day volume so at # SIZE you don't move the market. 0 = off (irrelevant at small size). liquidity_max_volume_pct: float = 0.0 # PORTFOLIO gross-notional cap: total $ across ALL open+pending positions must # stay under this fraction of equity. Without it, max_concurrent (6) x # max_position_pct (0.16) = ~96% gross in correlated small-cap gappers — one # regime move hits the whole book together. A later entry is shrunk to fit the # remaining budget, or blocked if none remains. 0 = off. max_gross_notional_pct: float = 0.60 # --- backout cooldown (anti-thrash) ------------------------------------ # after an entry backs out (timeout / price-break) the symbol is benched for # this many seconds before it can re-arm. Without it a breakout that won't # fill on a marketable limit re-arms every pass and churns (observed live: # NEOV armed+backed-out 68x in one session, bloating the order list). 0 = off. backout_cooldown_seconds: float = 180.0 # after this many backouts in a session, bench the symbol for the rest of the # day (a setup that has failed to fill repeatedly isn't going to start). 0 = no cap. max_backouts_per_symbol: int = 3 # once a position in a symbol closes at a LOSS, don't re-enter that name this # session. Stops the "throw good money after bad" re-entry into a name that # already failed (observed live: APWC stopped out -904, was bought back, lost # another -1012). Only blocks LOSING exits (see reentry_min_loss_pct) so a # quick scratch-then-re-enter winner isn't killed (GRAB +513 in the replay). reentry_block_after_exit: bool = True # only bench a closed name if it exited DOWN more than this fraction — a real # stop-out, not a scratch/win. 0 = bench on any close (the blunt original). reentry_min_loss_pct: float = 0.01 # --- anti-chase / halt protection (the "+100% then 3 halts down" failure) --- # Block a live entry whose price has already extended more than this fraction # ABOVE the trigger (opening-range high). A clean break crosses the level # smoothly (~0% extension and passes); a violent gap-THROUGH — a parabolic # spike, or a halt reopening past the level — lands far above it, and chasing # that buys the top tick that then reverses/halts down. 0 = off. # Default 0.15 calibrated on real fires: historical entries cluster <=9.8% # extension, with a clean gap to a disaster tail at 26-64% (NEOV x4 @27%, # ICCM @64%/26%, BIRD @30%). 15% sits in that dead zone — it blocks the # spike-top chases and passes every normal breakout. extension_max_pct: float = 0.15 # Day-extension ceiling: block when the entry is more than this fraction ABOVE # the SESSION OPEN — i.e. the stock is already parabolic on the day. This # catches what extension_max_pct can't: a name up huge intraday whose ORB-high # break is itself clean (small trigger-extension) but is a chase of a parabolic # (e.g. ATPC entered +1.7% above trigger yet +32% above the open, and lost). # Default 0.30 calibrated on real fires: normal entries are <=14% above the # open, the parabolic tail is 20-83%. 0 = off. Needs day_open from the trigger. day_extension_max_pct: float = 0.30 # Skip entries on a name that looks HALTED: in a trading halt you can't exit # and it gaps through your stop on resume. The caller passes a per-symbol # halt flag (a bar-gap heuristic over the liquid gappers we trade). This flag # only gates whether that signal is honoured. halt_guard_enabled: bool = True # treat a symbol as halted when, during RTH, its freshest minute bar is older # than this many minutes WHILE other names are still printing (LULD halts run # >= 5 min, so 3 catches a real halt without tripping on normal IEX lag). halt_max_bar_gap_min: float = 3.0 # --- catalyst dilution veto (Phase 2; ships OFF) ----------------------- # block an entry when the LLM catalyst advisory says the move is driven by a # CONFIRMED dilutive offering (a gap that fades as new shares hit). Requires # a catalyst_provider AND a conviction at/above the floor below. Default OFF # so the LLM stays advisory-only until its accuracy is trusted. catalyst_veto_enabled: bool = False catalyst_veto_min_conviction: float = 0.6 @classmethod def from_env(cls, env: dict[str, str] | None = None) -> "ExecutionSettings": values = dict(os.environ) if env is not None: values.update(env) def flag(key: str, default: str) -> bool: return values.get(key, default).strip().lower() in {"1", "true", "yes", "on"} order_type = values.get("TRADING_ENTRY_ORDER_TYPE", "limit").strip().lower() if order_type not in {"limit", "market"}: order_type = "limit" return cls( enabled=flag("TRADING_EXECUTION_ENABLED", "1"), auto_approve=flag("TRADING_AUTO_APPROVE", "0"), max_orders_per_tick=int(values.get("TRADING_MAX_ORDERS_PER_TICK", "1")), max_concurrent_positions=int( values.get("TRADING_MAX_CONCURRENT_POSITIONS", "3") ), risk_per_trade_pct=float(values.get("TRADING_RISK_PER_TRADE_PCT", "0.01")), default_equity=float(values.get("TRADING_DEFAULT_EQUITY", "100000")), max_daily_loss_pct=float(values.get("TRADING_MAX_DAILY_LOSS_PCT", "0.03")), flatten_on_breach=flag("TRADING_FLATTEN_ON_BREACH", "1"), reward_multiple=float(values.get("TRADING_REWARD_MULTIPLE", "2.0")), entry_order_type=order_type, entry_timeout_bars=int(values.get("TRADING_ENTRY_TIMEOUT_BARS", "2")), entry_invalidate_pct=float( values.get("TRADING_ENTRY_INVALIDATE_PCT", "0.015") ), trigger_slippage_pct=float( values.get("TRADING_TRIGGER_SLIP_PCT", "0.004") ), entry_grace_seconds=float(values.get("TRADING_ENTRY_GRACE_SECONDS", "5")), max_position_pct=float(values.get("TRADING_MAX_POSITION_PCT", "0.33")), liquidity_max_volume_pct=float( values.get("TRADING_LIQUIDITY_MAX_VOLUME_PCT", "0.0") ), max_gross_notional_pct=float( values.get("TRADING_MAX_GROSS_NOTIONAL_PCT", "0.60") ), backout_cooldown_seconds=float( values.get("TRADING_BACKOUT_COOLDOWN_SECONDS", "180") ), max_backouts_per_symbol=int( values.get("TRADING_MAX_BACKOUTS_PER_SYMBOL", "3") ), reentry_block_after_exit=flag("TRADING_BLOCK_REENTRY_AFTER_EXIT", "1"), reentry_min_loss_pct=float( values.get("TRADING_REENTRY_MIN_LOSS_PCT", "0.01") ), extension_max_pct=float( values.get("TRADING_EXTENSION_MAX_PCT", "0.15") ), day_extension_max_pct=float( values.get("TRADING_DAY_EXTENSION_MAX_PCT", "0.30") ), halt_guard_enabled=flag("TRADING_HALT_GUARD_ENABLED", "1"), halt_max_bar_gap_min=float( values.get("TRADING_HALT_MAX_BAR_GAP_MIN", "3") ), catalyst_veto_enabled=flag("NEWS_DILUTION_VETO_ENABLED", "0"), catalyst_veto_min_conviction=float( values.get("NEWS_DILUTION_VETO_CONVICTION", "0.6") ), ) class TradingExecutionService: """Drives the signal -> approval -> order pipeline each tick.""" def __init__( self, store: EventStore, executor: AlpacaPaperExecutor | None = None, settings: ExecutionSettings | None = None, trading_mode: TradingModeSettings | None = None, session_id: str | None = None, equity: float | None = None, price_provider=None, now_fn=None, catalyst_provider=None, ): self.store = store self.settings = settings or ExecutionSettings.from_env() # optional callable: symbol -> catalyst advisory dict (or None), from the # LLM enrichment cache. Used ONLY for the (gated) dilution veto. Injected # so the execution service never reads the DB itself. self.catalyst_provider = catalyst_provider self.trading_mode = trading_mode or TradingModeSettings.from_env() self.session_id = session_id self.equity = equity or self.settings.default_equity self.executor = executor or AlpacaPaperExecutor( store, session_id=session_id ) self.mode = EventMode.PAPER # optional callable: symbol -> latest price, used to invalidate unfilled # entries that break back below the entry trigger before filling self.price_provider = price_provider # injectable clock so the entry timeout is wall-clock based (and so the # invalidation check can run far more often than the timeout window # without changing the timeout's meaning); tests pass a fake clock self._now = now_fn or datetime.now # signals we've already requested approval for this session self._requested_symbols: set[str] = set() # armed (submitted, awaiting fill) entries we may need to cancel. # order_id -> {symbol, entry_price, broker_order_id, armed_at, checks} self._armed: dict[str, dict] = {} # backout cooldown (anti-thrash): symbol -> wall-clock time it may re-arm, # and how many times it has backed out this session. A symbol that won't # fill is benched instead of re-arming every pass. datetime.max == benched # for the rest of the session (hit max_backouts_per_symbol). self._cooldown_until: dict[str, datetime] = {} self._backout_counts: dict[str, int] = {} # re-entry block: symbols whose position has CLOSED this session (detected # by a name leaving the broker's held set) — benched from re-entry so we # don't buy back a name that just stopped out. _prev_held is the last # fresh broker snapshot used to spot departures. self._exited_today: set[str] = set() self._prev_held: set[str] = set() # last-seen unrealized return (fraction) per held symbol — used to decide, # when a name leaves the book, whether it exited at a real loss (bench) or # a scratch/win (allow re-entry). self._held_ret: dict[str, float] = {} # daily-loss circuit breaker: once tripped, no new entries this session # (a REAL loss is permanent for the session — a bounce can't re-open it) self._halted = False # transient data-halt: equity unreadable (network/DNS). Blocks new entries # WHILE unreadable but RECOVERS on the next good read — a DNS blip must not # end the trading day the way a real loss does. self._data_halt = False # session closed (end-of-day flatten): no new entries for the rest of the day self._session_closed = False # consecutive equity-read failures; after the limit we pause (recoverable) self._equity_read_failures = 0 self._equity_fail_limit = int( os.environ.get("TRADING_EQUITY_FAIL_LIMIT", "5") ) # serializes state mutations across the main loop and the trigger thread self._lock = threading.RLock() # -- pipeline ----------------------------------------------------------- def _open_position_count(self) -> int: snapshots = query_account_positions_snapshot( self.store, broker_name=self.executor.broker_name ) if not snapshots: return 0 return len(snapshots[-1].get("positions") or []) def _held_symbols(self) -> set[str]: snapshots = query_account_positions_snapshot( self.store, broker_name=self.executor.broker_name ) if not snapshots: return set() return { str(p.get("symbol")) for p in snapshots[-1].get("positions") or [] if p.get("symbol") } def _current_equity(self) -> float: """Size off the REAL broker equity, not the $100k default. On a $300 live account the default would size $1,000 positions and get rejected; this reads actual equity (the get_account read is cached, so it's cheap) and caches it on the service for fallback. """ client = getattr(self.executor, "client", None) if client is not None and hasattr(client, "get_account"): try: eq = float((client.get_account() or {}).get("equity") or 0.0) if eq > 0: self.equity = eq return eq except Exception: # noqa: BLE001 pass return self.equity def _remaining_gross_budget(self, equity: float) -> float: """$ of new notional still allowed under the portfolio gross-notional cap (cap*equity minus the market value of currently-open positions). Inf when the cap is disabled. Caps how big the NEXT entry can be so the whole book can't pile into ~100% gross of correlated small-caps.""" cap = float(self.settings.max_gross_notional_pct or 0.0) if cap <= 0: return float("inf") positions = self._broker_positions() or [] open_gross = 0.0 for p in positions: try: open_gross += abs(float(p.get("market_value") or 0.0)) except (TypeError, ValueError): pass return max(0.0, cap * equity - open_gross) def _broker_held_symbols(self) -> set[str] | None: """Symbols the broker reports as OPEN POSITIONS right now (truth). Used to guard entry cancellation: our own order_filled events lag the actual fill (they arrive via the 60s account sync), so the entry-timeout guard once cancelled an entry it thought was unfilled AFTER it had really filled — and cancelling a bracket parent cancels its stop/take-profit legs, leaving the position NAKED. This queries the broker directly so a filled name is never cancelled. None if positions can't be read (caller then falls back to the synced snapshot and stays conservative). """ client = getattr(self.executor, "client", None) if client is None or not hasattr(client, "get_positions"): return None try: positions = client.get_positions(fresh=True) or [] return {str(p.get("symbol")) for p in positions if p.get("symbol")} except Exception as exc: # noqa: BLE001 logger.warning("broker positions unreadable in entry guard: %s", exc) return None def _release_and_close(self, client, symbol: str) -> None: """Cancel a symbol's resting bracket legs, then market-close it. Thin wrapper over the shared helper (runtime/flatten.py) — the held-qty 403 fix lives in one place, shared with the live exit manager. Fetches the symbol's open orders itself (no pre-fetched snapshot at EOD).""" cancel_protective_and_close(client, symbol) def _flatten_all(self, reason: str) -> dict: """Cancel unfilled entries and market-close every open position. Used at EOD and when the daily-loss circuit breaker trips: stop the bleeding by cancelling resting entries and flattening the book. Best-effort — errors are collected, not raised, so one bad symbol can't block the rest. Resting bracket legs are cancelled before each close so the held quantity is freed (else close_position 403s).""" result: dict = {"cancelled_entries": 0, "closed_positions": [], "errors": []} # 1) cancel unfilled (armed) entries we are tracking for order_id in list(self._armed): armed = self._armed.pop(order_id) symbol = armed.get("symbol", "") try: self.executor.cancel_entry( order_id, armed.get("broker_order_id"), symbol, reason ) result["cancelled_entries"] += 1 except Exception as exc: # noqa: BLE001 result["errors"].append(f"cancel {symbol}: {exc}") self._requested_symbols.discard(symbol) # 2) market-close open positions client = getattr(self.executor, "client", None) for symbol in sorted(self._held_symbols()): if client is None or not hasattr(client, "close_position"): result["errors"].append(f"close {symbol}: no broker client to flatten") continue try: self._release_and_close(client, symbol) result["closed_positions"].append(symbol) # only on a confirmed close except Exception as exc: # noqa: BLE001 result["errors"].append(f"close {symbol}: {exc}") return result @_locked def _daily_loss_breach(self) -> bool: """True while new entries should be blocked. Two distinct halts: a REAL daily-loss breach (``_halted``) is permanent for the session so a bounce can't re-open the floodgates; an equity-read outage (``_data_halt``) blocks WHILE unreadable but recovers on the next good read — a transient DNS/network blip must not end the trading day. """ if self._halted or self._session_closed: return True max_loss = abs(float(self.settings.max_daily_loss_pct or 0.0)) if max_loss <= 0: return False client = getattr(self.executor, "client", None) if client is None: return False try: acct = client.get_account() equity_raw = acct.get("equity") if equity_raw is None: raise ValueError("account 'equity' field missing") equity = float(equity_raw) baseline = float(acct.get("last_equity") or equity) # good read -> clear the transient data-halt and resume if self._data_halt or self._equity_read_failures: if self._data_halt: self.store.emit( RiskRuleTriggeredEvent( timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id, message="Equity readable again — resuming new entries", rule_type="equity_recovered", rule_value=0.0, current_state={"equity": equity}, action_taken="resumed_new_entries", ) ) self._data_halt = False self._equity_read_failures = 0 except Exception as exc: # noqa: BLE001 self._equity_read_failures += 1 logger.warning( "equity read failed (%d consecutive): %s", self._equity_read_failures, exc, ) # PAUSE (recoverable): block new entries while equity is unreadable, # rather than silently disabling the most important safety control — # but DO NOT permanently halt the session for a transient outage. if self._equity_read_failures >= self._equity_fail_limit and not self._data_halt: self._data_halt = True self.store.emit( RiskRuleTriggeredEvent( timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id, message=( f"Equity unreadable {self._equity_read_failures}x — " "pausing new entries until it recovers" ), rule_type="equity_unreadable", rule_value=float(self._equity_read_failures), current_state={"failures": self._equity_read_failures}, action_taken="paused_new_entries", ) ) return self._data_halt if baseline <= 0: return False pnl_pct = (equity - baseline) / baseline if pnl_pct <= -max_loss: self._halted = True flat = ( self._flatten_all("daily_loss_circuit_breaker") if self.settings.flatten_on_breach else {"closed_positions": [], "cancelled_entries": 0} ) extra = "" action = "halted_new_entries" if self.settings.flatten_on_breach: n_closed = len(flat["closed_positions"]) n_cancel = flat["cancelled_entries"] flat_errs = flat.get("errors") or [] extra = ( f" — flattened {n_closed} position(s), " f"cancelled {n_cancel} unfilled entr{'y' if n_cancel == 1 else 'ies'}" ) if flat_errs: extra += f"; FLATTEN INCOMPLETE: {flat_errs}" action = "halted_flatten_incomplete" if flat_errs else "halted_and_flattened" self.store.emit( RiskRuleTriggeredEvent( timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id, message=( f"Daily-loss circuit breaker tripped: {pnl_pct:+.2%} " f"(limit -{max_loss:.0%}) — halting new entries{extra}" ), rule_type="daily_loss", rule_value=max_loss, current_state={ "equity": equity, "baseline_equity": baseline, "pnl_pct": round(pnl_pct, 4), "flatten": flat, }, action_taken=action, ) ) return self._halted @_locked def _dilution_vetoed(self, symbol: str) -> dict | None: """If the gated catalyst veto is on AND the LLM advisory flags a confirmed dilutive offering at/above the conviction floor, return the advisory (so the caller can emit a risk event + skip). Otherwise None. Never raises — a missing/odd advisory simply doesn't veto.""" if not (self.settings.catalyst_veto_enabled and self.catalyst_provider): return None try: adv = self.catalyst_provider(symbol) except Exception: # noqa: BLE001 — advisory is best-effort, never blocks on its own fault return None if not adv or not adv.get("is_dilutive"): return None try: conviction = float(adv.get("conviction") or 0.0) except (TypeError, ValueError): return None if conviction >= self.settings.catalyst_veto_min_conviction: return adv return None def _emit_dilution_veto(self, symbol: str, adv: dict) -> None: self.store.emit( RiskRuleTriggeredEvent( timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id, message=f"{symbol}: entry blocked — confirmed dilutive catalyst", rule_type="catalyst_dilution_veto", rule_value=float(adv.get("conviction") or 0.0), current_state={ "symbol": symbol, "catalyst_type": adv.get("catalyst_type"), "sentiment": adv.get("sentiment"), "rationale": adv.get("rationale"), }, action_taken="blocked_dilutive_offering", ) ) def request_approvals_for_ready_signals(self) -> list[str]: """Turn fresh ready signals into approval requests. Returns order ids.""" if not self.settings.enabled: return [] if self._daily_loss_breach(): return [] if self._open_position_count() >= self.settings.max_concurrent_positions: self.store.emit( RiskRuleTriggeredEvent( timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id, message="Max concurrent positions reached — no new entries", rule_type="max_concurrent_positions", rule_value=float(self.settings.max_concurrent_positions), current_state={"open_positions": self._open_position_count()}, action_taken="skipped_new_entries", ) ) return [] held = self._held_symbols() pending = {row["symbol"] for row in query_approval_queue(self.store)} signals = query_ready_signals_snapshot(self.store, session_id=self.session_id) created: list[str] = [] for signal in signals: if len(created) >= self.settings.max_orders_per_tick: break symbol = signal["symbol"] if (symbol in held or symbol in pending or symbol in self._requested_symbols or self._in_cooldown(symbol) # benched after recent backouts or symbol in self._exited_today): # already round-tripped today continue vetoed = self._dilution_vetoed(symbol) if vetoed is not None: self._emit_dilution_veto(symbol, vetoed) self._requested_symbols.add(symbol) # don't re-evaluate every tick continue entry = signal.get("entry_price") stop = signal.get("stop_loss_price") if not entry or not stop or stop >= entry: continue eq = self._current_equity() # cap this entry to BOTH the per-position limit AND the portfolio's # remaining gross-notional budget (so the book can't pile into ~100%). pos_cap = min(eq * self.settings.max_position_pct, self._remaining_gross_budget(eq)) sizing = calculate_position_size( float(entry), float(stop), equity=eq, config=PositionSizingConfig( risk_per_trade_pct=self.settings.risk_per_trade_pct, default_equity=self.settings.default_equity, ), max_position_value=pos_cap, ) if sizing.position_size <= 0: continue entry_f = float(entry) stop_f = float(stop) risk_per_share = entry_f - stop_f request = ExecutionRequest( symbol=symbol, side="buy", quantity=sizing.position_size, entry_price=entry_f, stop_loss_price=stop_f, take_profit_price=round( entry_f + self.settings.reward_multiple * risk_per_share, 2 ), order_type=self.settings.entry_order_type, ) approval_mode = "auto" if self.settings.auto_approve else "manual" self.store.emit( OrderApprovalRequestedEvent( timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id, message=( f"Approval requested ({approval_mode}): buy " f"{request.quantity} {symbol} @ ~{entry} stop {stop}" ), order_id=request.order_id, symbol=symbol, requested_by="trading_execution", approval_mode=approval_mode, execution_mode=self.trading_mode.execution_mode, execution_request=request.to_payload(), ) ) self._requested_symbols.add(symbol) created.append(request.order_id) logger.info("approval requested for %s (%s)", symbol, request.order_id) return created @_locked def approve_order( self, order_id: str, approved_by: str = "dashboard", notes: str | None = None ) -> dict: """Approve a pending order and execute it.""" entry = self._find_pending(order_id) if entry is None: return {"ok": False, "error": f"order {order_id} not pending"} request = ExecutionRequest.from_payload(entry["execution_request"]) self.store.emit( OrderApprovedEvent( timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id, message=f"Order {order_id} approved by {approved_by}", order_id=order_id, symbol=request.symbol, approved_by=approved_by, approval_notes=notes, ) ) result = self.executor.execute(request) # If the entry didn't fill immediately (a resting limit at the trigger), # arm it so expire_stale_entries() can back out on timeout or a break # below the entry. Market orders that fill on submit are never armed. # Only arm genuinely OPEN resting entries — never a filled, partial, or # rejected order (a rejected entry would otherwise occupy a position slot # forever and a partial would be backed out while real shares are held). if ( result.ok and result.status in {"new", "accepted", "pending_new", "held", "accepted_for_bidding"} and request.entry_price ): self._armed[order_id] = { "symbol": request.symbol, "entry_price": float(request.entry_price), "broker_order_id": result.broker_order_id, "armed_at": self._now(), "checks": 0, } return { "ok": result.ok, "order_id": order_id, "broker_order_id": result.broker_order_id, "status": result.status, "error": result.error, } @_locked def submit_breakout_now( self, symbol: str, trigger: float, stop: float, last_price: float | None = None, reason: str = "orb_live_break", cum_volume: float = 0.0, halted: bool = False, day_open: float | None = None, ) -> dict: """Fire a breakout entry immediately on a LIVE price cross. The disciplined-but-fast counterpart to the watcher->approval cadence: used by the armed-trigger loop when price crosses a pre-computed opening-range high. Runs the SAME risk gates as the slow path (daily-loss breaker, max-concurrent, per-symbol dedup, 1%-risk sizing), but submits a *marketable* limit (capped a hair above the trigger) so it fills on a runner instead of resting at the level. Returns a result dict; ``skipped`` explains a no-op (the symbol can simply try again next tick). """ if not self.settings.enabled: return {"ok": False, "skipped": "disabled"} if self._daily_loss_breach(): return {"ok": False, "skipped": "halted"} if self._open_position_count() >= self.settings.max_concurrent_positions: return {"ok": False, "skipped": "max_positions"} # per-symbol dedup shared with the slow path: never double-enter a name # that's already held, pending approval, or requested this session. held = self._held_symbols() pending = {row["symbol"] for row in query_approval_queue(self.store)} if symbol in held or symbol in pending or symbol in self._requested_symbols: return {"ok": False, "skipped": "already_active"} if self._in_cooldown(symbol): # benched after recent backouts (anti-thrash) return {"ok": False, "skipped": "cooldown"} if symbol in self._exited_today: # already round-tripped today — no re-entry return {"ok": False, "skipped": "reentry_blocked"} vetoed = self._dilution_vetoed(symbol) # gated catalyst dilution veto if vetoed is not None: self._emit_dilution_veto(symbol, vetoed) return {"ok": False, "skipped": "dilution_veto"} # Use last_price only when it's a real positive tick. `last_price or # trigger` would coerce a 0.0/garbage price to the trigger — silently # skipping the extension guard AND sizing on a phantom price. if last_price is not None and float(last_price) <= 0: return {"ok": False, "skipped": "bad_price"} entry_ref = float(last_price) if last_price is not None else float(trigger) stop_f = float(stop) trigger_f = float(trigger) if stop_f >= entry_ref or trigger_f <= 0: return {"ok": False, "skipped": "bad_geometry"} # Anti-chase ceiling: distance of the entry price ABOVE the breakout # trigger (ORB high). A clean break crosses the level smoothly (~0%) and # passes; a violent gap-THROUGH — an intraday parabolic, or a halt # reopening past the level — lands far above it, and chasing that buys # the top tick that reverses/halts down. (Overnight +100% gappers are # already excluded upstream by the trigger book's gap_max; this catches # the intraday chase / gap-through that gap_max can't see.) ext_max = self.settings.extension_max_pct if ext_max > 0 and entry_ref > trigger_f * (1.0 + ext_max): return {"ok": False, "skipped": "over_extended"} # Day-extension ceiling: the stock is already parabolic on the DAY (far # above the session open) — chasing that is the "+100% then halts down" # trap even when the trigger break itself looks clean. day_max = self.settings.day_extension_max_pct if (day_max > 0 and day_open is not None and float(day_open) > 0 and entry_ref > float(day_open) * (1.0 + day_max)): return {"ok": False, "skipped": "over_extended_day"} # Halt guard: you can't exit during a trading halt and it gaps through the # stop on resume — don't enter a name the caller flagged as halted. if halted and self.settings.halt_guard_enabled: return {"ok": False, "skipped": "halted_symbol"} eq = self._current_equity() liq = self.settings.liquidity_max_volume_pct max_shares = int(liq * cum_volume) if (liq > 0 and cum_volume > 0) else None # per-position cap AND remaining portfolio gross-notional budget pos_cap = min(eq * self.settings.max_position_pct, self._remaining_gross_budget(eq)) if pos_cap <= 0: return {"ok": False, "skipped": "gross_notional_cap"} sizing = calculate_position_size( entry_ref, stop_f, equity=eq, config=PositionSizingConfig( risk_per_trade_pct=self.settings.risk_per_trade_pct, default_equity=self.settings.default_equity, ), max_position_value=pos_cap, max_shares=max_shares, ) if sizing.position_size <= 0: return {"ok": False, "skipped": "zero_size"} # marketable limit: fill now (price is already at/above the trigger), but # cap how far above we'll chase so a gap-through doesn't pay any price. limit_price = round( max(entry_ref, trigger_f) * (1.0 + self.settings.trigger_slippage_pct), 2 ) risk_per_share = entry_ref - stop_f request = ExecutionRequest( symbol=symbol, side="buy", quantity=sizing.position_size, entry_price=limit_price, stop_loss_price=stop_f, take_profit_price=round( entry_ref + self.settings.reward_multiple * risk_per_share, 2 ), order_type="limit", ) # audit trail mirrors the slow path: requested(auto) -> approved -> submit self.store.emit( OrderApprovalRequestedEvent( timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id, message=( f"Live ORB break ({reason}): buy {request.quantity} {symbol} " f"@ ~{entry_ref:.2f} (trigger {trigger_f:.2f}, stop {stop_f:.2f})" ), order_id=request.order_id, symbol=symbol, requested_by="orb_trigger", approval_mode="auto", execution_mode=self.trading_mode.execution_mode, execution_request=request.to_payload(), ) ) self.store.emit( OrderApprovedEvent( timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id, message=f"Order {request.order_id} auto-approved (live ORB break)", order_id=request.order_id, symbol=symbol, approved_by="orb_trigger", approval_notes=reason, ) ) self._requested_symbols.add(symbol) result = self.executor.execute(request) if ( result.ok and result.status in {"new", "accepted", "pending_new", "held", "accepted_for_bidding"} and request.entry_price ): self._armed[request.order_id] = { "symbol": symbol, "entry_price": float(request.entry_price), "broker_order_id": result.broker_order_id, "armed_at": self._now(), "checks": 0, } logger.info( "live ORB break %s qty=%s -> ok=%s status=%s", symbol, request.quantity, result.ok, result.status, ) return { "ok": result.ok, "order_id": request.order_id, "broker_order_id": result.broker_order_id, "status": result.status, "error": result.error, "quantity": sizing.position_size, "entry": limit_price, "stop": stop_f, } @_locked def reject_order( self, order_id: str, rejected_by: str = "dashboard", reason: str = "manual" ) -> dict: entry = self._find_pending(order_id) if entry is None: return {"ok": False, "error": f"order {order_id} not pending"} self.store.emit( OrderRejectedEvent( timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id, message=f"Order {order_id} rejected by {rejected_by}: {reason}", order_id=order_id, symbol=entry.get("symbol") or "", rejected_by=rejected_by, rejection_reason=reason, ) ) # allow the symbol to re-signal later self._requested_symbols.discard(entry.get("symbol") or "") return {"ok": True, "order_id": order_id} @_locked def process_auto_approvals(self) -> list[dict]: """Auto-approve any pending requests marked approval_mode=auto.""" results = [] for entry in query_approval_queue(self.store): if entry.get("approval_mode") == "auto": results.append( self.approve_order(entry["order_id"], approved_by="auto") ) return results @_locked def submit_exit_order(self, symbol: str, reason: str = "manual_exit") -> dict: """Close an open broker position for symbol with a market sell.""" held = self._held_symbols() quantity = None snapshots = query_account_positions_snapshot( self.store, broker_name=self.executor.broker_name ) if snapshots: for p in snapshots[-1].get("positions") or []: if str(p.get("symbol")) == symbol: quantity = int(abs(float(p.get("quantity") or 0))) if symbol not in held or not quantity: return {"ok": False, "error": f"no open position for {symbol}"} # Prefer close_position: closes the FULL position regardless of long/short # sign and fractional qty. A plain 'sell' of abs(qty) would DOUBLE a short # position, and int-truncation would leave residual shares un-flat. client = getattr(self.executor, "client", None) if client is not None and hasattr(client, "close_position"): try: raw = client.close_position(symbol) or {} return { "ok": True, "order_id": str(raw.get("id") or f"close-{symbol}"), "broker_order_id": raw.get("id"), "status": raw.get("status") or "closing", "error": None, "reason": reason, } except Exception as exc: # noqa: BLE001 return {"ok": False, "error": str(exc), "reason": reason} request = ExecutionRequest(symbol=symbol, side="sell", quantity=quantity) result = self.executor.execute(request) return { "ok": result.ok, "order_id": request.order_id, "broker_order_id": result.broker_order_id, "status": result.status, "error": result.error, "reason": reason, } @_locked def close_session(self, reason: str = "eod_flatten") -> dict: """End-of-day: block new entries and flatten the book. A day-trading strategy shouldn't carry overnight gap risk, so at the configured time the loop calls this to cancel resting entries, market- close every open position (which also cancels its bracket legs), and halt new entries for the rest of the session. """ self._session_closed = True result = self._flatten_all(reason) # belt-and-suspenders: close any broker position the store snapshot missed client = getattr(self.executor, "client", None) if client is not None and hasattr(client, "get_positions"): try: for p in client.get_positions(): sym = p.get("symbol") if sym and sym not in result["closed_positions"]: try: self._release_and_close(client, sym) result["closed_positions"].append(sym) except Exception as exc: # noqa: BLE001 result["errors"].append(f"close {sym}: {exc}") except Exception as exc: # noqa: BLE001 result["errors"].append(f"positions: {exc}") self.store.emit( RiskRuleTriggeredEvent( timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id, message=(f"End-of-day flatten ({reason}): closed " f"{len(result['closed_positions'])} position(s), " f"cancelled {result['cancelled_entries']} entr" f"{'y' if result['cancelled_entries'] == 1 else 'ies'}" + (f"; ERRORS {result['errors']}" if result['errors'] else "")), rule_type="eod_flatten", rule_value=0.0, current_state=result, action_taken="closed_session", ) ) return result def _filled_order_ids(self) -> set[str]: ids: set[str] = set() for e in self.store.query_events(event_type="order_filled", limit=None): payload = json.loads(e.get("payload_json", "{}")) if payload.get("order_id"): ids.add(str(payload["order_id"])) return ids def _cancelled_order_ids(self) -> set[str]: ids: set[str] = set() for e in self.store.query_events(event_type="order_cancelled", limit=None): payload = json.loads(e.get("payload_json", "{}")) if payload.get("order_id"): ids.add(str(payload["order_id"])) return ids @_locked def _in_cooldown(self, symbol: str) -> bool: """True if symbol is benched after recent backouts (anti-thrash).""" until = self._cooldown_until.get(symbol) if until is None: return False if self._now() >= until: del self._cooldown_until[symbol] # cooldown elapsed return False return True def _register_backout(self, symbol: str) -> str: """Record a backout and bench the symbol; returns a short reason suffix.""" n = self._backout_counts.get(symbol, 0) + 1 self._backout_counts[symbol] = n cap = self.settings.max_backouts_per_symbol if cap and n >= cap: self._cooldown_until[symbol] = datetime.max # benched for the session return f"benched for session ({n} backouts)" secs = self.settings.backout_cooldown_seconds if secs > 0: self._cooldown_until[symbol] = self._now() + timedelta(seconds=secs) return f"cooldown {secs:.0f}s (backout {n})" return f"backout {n}" def _broker_positions(self) -> list[dict] | None: """Fresh broker positions (full dicts), or None if unreadable.""" client = getattr(self.executor, "client", None) if client is None or not hasattr(client, "get_positions"): return None try: return client.get_positions(fresh=True) or [] except Exception as exc: # noqa: BLE001 logger.warning("broker positions unreadable in re-entry guard: %s", exc) return None def _note_position_exits(self) -> None: """Detect names whose position closed since last pass and bench the ones that exited at a real LOSS from re-entry this session — re-buying a name that just stopped out is throwing good money after bad. A name that scratched or won is left alone (a quick re-entry can be the right call).""" if not self.settings.reentry_block_after_exit: return positions = self._broker_positions() if positions is None: # broker unreadable — don't trust a stale snapshot return held: set[str] = set() for p in positions: sym = p.get("symbol") if not sym: continue held.add(sym) try: self._held_ret[sym] = float(p.get("unrealized_plpc") or 0.0) except (TypeError, ValueError): pass thr = abs(float(self.settings.reentry_min_loss_pct or 0.0)) for sym in (self._prev_held - held): last_ret = self._held_ret.pop(sym, 0.0) if thr > 0 and last_ret >= -thr: continue # scratched or won — allow re-entry self._exited_today.add(sym) self.store.emit(RiskRuleTriggeredEvent( timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id, message=(f"{sym} closed at {last_ret*100:+.1f}% — benched from " f"re-entry today"), rule_type="reentry_block", rule_value=last_ret, current_state={"symbol": sym, "exit_return": last_ret}, action_taken="reentry_blocked", )) self._prev_held = held def expire_stale_entries(self) -> list[dict]: """Back out of unfilled entries that timed out or broke their trigger. This is the disciplined other half of auto-arming: once a setup is recognized we enter with conviction, but if the fill never comes (the move didn't follow through) we don't sit there forever — we cancel on a time box or when price falls back through the entry, freeing risk budget and the concurrent-position slot for the next setup. Safe to call as often as you like: the price-break check uses the live ``price_provider`` each call, while the timeout is wall-clock based, so running this every few seconds reacts fast without shortening the timeout window. """ # track positions that have CLOSED since last pass (so a stopped-out name # is benched from re-entry). Done BEFORE the early-return so it runs every # pass, not only when there are armed entries to expire. self._note_position_exits() if not self._armed: return [] filled = self._filled_order_ids() cancelled = self._cancelled_order_ids() # broker truth: never cancel an entry whose position is actually open # (cancelling its bracket parent would strip the protective stop/TP). # Fall back to the synced snapshot if the broker can't be read. broker_held = self._broker_held_symbols() if broker_held is None: broker_held = self._held_symbols() invalidate_pct = self.settings.entry_invalidate_pct timeout_bars = self.settings.entry_timeout_bars now = self._now() actions: list[dict] = [] for order_id in list(self._armed): armed = self._armed[order_id] symbol = armed["symbol"] # already resolved at the broker -> stop tracking if order_id in filled: self._armed.pop(order_id, None) continue if order_id in cancelled: self._armed.pop(order_id, None) self._requested_symbols.discard(symbol) continue # FILLED at the broker (position open) -> stop tracking, NEVER cancel # (this is the naked-stop fix: cancelling here killed the stop leg). if symbol in broker_held: self._armed.pop(order_id, None) continue # grace window: a just-armed entry may still be filling — don't strip # its bracket before the fill can confirm (belt to the fresh-positions # check above; together they close the naked-stop race) if (now - armed["armed_at"]).total_seconds() < self.settings.entry_grace_seconds: continue armed["checks"] += 1 reason = None # (a) price-break invalidation: traded back below the entry trigger. # Uses the live last-trade price, so this reacts tick-by-tick rather # than waiting for a bar to close. if invalidate_pct >= 0 and self.price_provider is not None: try: last = self.price_provider(symbol) except Exception: # noqa: BLE001 last = None if last is not None: threshold = armed["entry_price"] * (1.0 - invalidate_pct) if float(last) < threshold: reason = ( f"entry invalidated: {float(last):.4f} < trigger " f"{threshold:.4f}" ) # (b) time box: unfilled for too long (wall-clock minutes ~ bars) if reason is None and timeout_bars > 0: elapsed_min = (now - armed["armed_at"]).total_seconds() / 60.0 if elapsed_min >= timeout_bars: reason = f"entry timed out: unfilled {elapsed_min:.1f} min" if reason is not None: self.executor.cancel_entry( order_id, armed.get("broker_order_id"), symbol, reason ) bench = self._register_backout(symbol) # anti-thrash: cool the name self.store.emit( RiskRuleTriggeredEvent( timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id, message=f"Backed out of {symbol} entry — {reason} [{bench}]", rule_type="entry_backout", rule_value=float(timeout_bars), current_state={ "symbol": symbol, "order_id": order_id, "checks": armed["checks"], "backouts": self._backout_counts.get(symbol, 0), }, action_taken="cancelled_unfilled_entry", ) ) self._armed.pop(order_id, None) self._requested_symbols.discard(symbol) actions.append({"order_id": order_id, "symbol": symbol, "reason": reason, "cooldown": bench}) return actions @_locked def tick(self) -> dict: """One execution pass: expire stale entries, request approvals, auto-execute.""" backed_out = self.expire_stale_entries() requested = self.request_approvals_for_ready_signals() auto = self.process_auto_approvals() if self.settings.auto_approve else [] return { "approvals_requested": requested, "auto_executed": auto, "backed_out": backed_out, } # -- helpers -------------------------------------------------------------- def _find_pending(self, order_id: str) -> dict | None: for entry in query_approval_queue(self.store): if str(entry.get("order_id")) == str(order_id): return entry return None