""" position_manager.py Owns the full lifecycle of a single position from entry to exit, and the collection of all currently open positions. Talks to alpaca_client.py for order execution (skipped entirely in simulation mode) and risk_manager.py for stop math. State shape per position (matches project's required fields): { "symbol", "entry_price", "entry_time", "initial_stop", "current_stop", "highest_price", "current_price", "trailing_distance", "shares", "current_pl", "mfe" (max favorable excursion), "exit_price", "exit_reason", "status", # "open" | "closed" | "closing" "fade_grace_active", "grace_extensions_used" # [FEATURE 2026-08-18] "exit_pending", "exit_order_id", "exit_reason_pending" # [BUGFIX 2026-08-18] } [BUGFIX 2026-08-18] Order-race / "wash trade" fix --------------------------------------------------- Previously, exit_position() called client.close_position(symbol) and, on ANY exception (including Alpaca's "potential wash trade detected... opposite side market/stop order exists" rejection), simply logged an ERROR and returned without changing local state. Because status stayed "open", the next 5-second poll cycle (monitor.py's poll_interval_seconds_ intraday) saw the position as still open, re-evaluated the trailing stop, and called exit_position() again -- submitting ANOTHER close order against a broker that already had the first one resting and unfilled. Alpaca's wash-trade guard then rejected the new one too, and the cycle repeated every poll until the original order finally filled (seen live as 3+ minutes of repeated rejections on a single symbol, during which the position had no working stop protection at all). The fix makes exit_position() idempotent per symbol: - status gains a third value, "closing", set the moment we have a close order (submitted by us, or a pre-existing one we recovered from a wash-trade rejection) resting at the broker. - While "closing", exit_position() never submits a new order -- it polls the existing order via alpaca_client.get_order() and only finalizes the position once that order is actually FILLED, using the broker-confirmed fill price (not the price bars happened to show when the exit signal first fired). - If close_position() itself raises a wash-trade rejection, we parse the existing_order_id Alpaca already hands us in that error (it was previously discarded) and adopt it as the order we're tracking, instead of treating the rejection as a failure. """ from datetime import datetime, timedelta, timezone from config_loader import get_config from logger_setup import get_logger from alpaca_client import get_client, parse_wash_trade_error, parse_position_not_found_error from alpaca.trading.enums import OrderSide import risk_manager import intraday_health import data_store import market_time from stream_features import compute_features log = get_logger("position_manager") class PositionManager: def __init__(self, simulation: bool = True): self.simulation = simulation self.client = get_client() if not simulation else None self.positions: dict = data_store.load_positions() self._prune_stale_closed_positions() self._last_reconcile = None def _prune_stale_closed_positions(self): """ [BUGFIX 2026-08-31] positions.json is never rotated -- it's loaded wholesale from disk with no date filtering, so "closed" positions from any prior day pile up indefinitely (confirmed live: entries going back to 2026-08-17 still present). Because has_closed_position_today() and reentry_cooldown_remaining_minutes() both just check status == "closed" on this same self.positions dict with no date check of their own, a symbol that closed once, ever, looks identical to one that closed minutes ago -- permanently forcing the score-based re-entry gate's full-confirmation path (and, if same_symbol_reentry_cooldown_minutes is ever re-enabled, an indefinite phantom cooldown). Same underlying gap already found in state/intraday_health.json (see monitor.py's health-state pruning). Only "closed" entries are pruned -- "open"/"closing" positions are always kept regardless of date, since one from a prior day is most likely a real position recovered after a crash/restart that still needs active management, not stale data. """ today = market_time.now_et().date() stale = [symbol for symbol, p in self.positions.items() if p.get("status") == "closed" and self._exit_et_date(p) != today] for symbol in stale: del self.positions[symbol] if stale: log.info(f"[STARTUP] Pruned {len(stale)} closed position(s) from a prior day: {stale}") data_store.save_positions(self.positions) @staticmethod def _exit_et_date(p: dict): exit_time = p.get("exit_time") if not exit_time: return None return market_time.et_date(datetime.fromisoformat(exit_time)) # ------------------------------------------------------------------ def open_positions_count(self) -> int: # "closing" positions still occupy a slot -- shares are still on # the books until the close order actually fills. return len([p for p in self.positions.values() if p["status"] in ("open", "closing")]) def has_available_slot(self) -> bool: max_positions = get_config()["trading"]["max_positions"] return self.open_positions_count() < max_positions def is_symbol_open(self, symbol: str) -> bool: p = self.positions.get(symbol) return p is not None and p["status"] in ("open", "closing") def has_closed_position_today(self, symbol: str) -> bool: """ [FEATURE 2026-08-27] True if this symbol has a closed position already on record this session. Originally fed entry_engine.py's score-based re-entry gate (require confirmation_score==100 on a re-entry rather than the normal floor) -- that module was removed in the 2026-09-11 cleanup along with prediction_pipeline.py's entry_score.py, and fast_entry_gate.py (the current live decision path) has NO equivalent re-entry protection yet. Kept as-is, currently unused live -- flagged as a real gap to close, not dead code to delete, since the incident this originally guarded against (MARA re-entering the same second as a losing exit, CLSK re-entering 8 minutes after a winning exit) is a risk for ANY decision engine, not specific to the one that got removed. """ p = self.positions.get(symbol) return p is not None and p["status"] == "closed" def reentry_cooldown_remaining_minutes(self, symbol: str): """ [FEATURE 2026-08-25, superseded 2026-08-27] Minutes remaining in the same-symbol re-entry cooldown, or None if the symbol has no prior closed position on record (never traded, or the record was already overwritten by a newer entry -- in which case that newer entry's own timing governs, not this one). 0 or negative means the cooldown has elapsed. Retained for anyone who wants to combine a time-based wait WITH the newer score-based gate (has_closed_position_today() above), but same_symbol_reentry_cooldown_minutes defaults to 0 (disabled) as of 2026-08-27 -- the score-based gate is now the primary re-entry control. Exposed separately from enter_position()'s hard check so monitor.py's shortlist can also filter on this before spending a full evaluate_entry() call on a symbol that would just get rejected anyway. """ p = self.positions.get(symbol) if p is None or p["status"] != "closed" or not p.get("exit_time"): return None cooldown_minutes = get_config()["trading"].get("same_symbol_reentry_cooldown_minutes", 0) if cooldown_minutes <= 0: return None exit_time = datetime.fromisoformat(p["exit_time"]) elapsed_minutes = (datetime.now(timezone.utc) - exit_time).total_seconds() / 60.0 return cooldown_minutes - elapsed_minutes # ------------------------------------------------------------------ def enter_position(self, symbol: str, entry_price: float, account_equity: float, bars: list = None, reason: str = "") -> dict: allow_multi = get_config()["trading"]["allow_multiple_entries_same_symbol"] if self.is_symbol_open(symbol) and not allow_multi: log.warning(f"[ENTRY] Skipped {symbol}: already have an open position") return None # [FEATURE 2026-08-25] Same-symbol re-entry cooldown. 2026-08-24 # session review flagged two related patterns: MARA re-entered # immediately after a LOSING exit (chasing the same failed # setup twice), and CLSK re-entered immediately after a WINNING # exit (false confidence that the next entry into the same name # is also good, when it's often just chasing). One timer covers # both, since it's symbol-based, not outcome-based -- deliberately # NOT the price-based pullback gate tested earlier in this # project, which backfired on real data due to penny-level # noise; this is purely time-based and immune to that problem. # Uses the prior CLOSED position record still sitting in # self.positions[symbol] -- read here, BEFORE it gets # overwritten by the new position dict below, so no separate # persistence mechanism is needed. cooldown_minutes = get_config()["trading"].get("same_symbol_reentry_cooldown_minutes", 0) if cooldown_minutes > 0: remaining = self.reentry_cooldown_remaining_minutes(symbol) if remaining is not None and remaining > 0: log.info(f"[ENTRY] Skipped {symbol}: {remaining:.1f}min remaining in " f"{cooldown_minutes}min same-symbol re-entry cooldown") return None if not self.has_available_slot(): log.info(f"[ENTRY] Skipped {symbol}: no available position slots") return None initial_stop = risk_manager.compute_initial_stop(entry_price, bars) shares = risk_manager.compute_position_size(account_equity, entry_price, initial_stop) if shares <= 0: log.warning(f"[ENTRY] Skipped {symbol}: computed size is 0 shares " f"(entry={entry_price}, stop={initial_stop})") return None order_id = None if not self.simulation: try: order = self.client.submit_market_order(symbol, shares, "buy") order_id = getattr(order, "id", None) except Exception as e: log.error(f"[ENTRY] Order submission failed for {symbol}: {e}") return None position = { "symbol": symbol, "entry_price": entry_price, "entry_time": datetime.now(timezone.utc).isoformat(), "initial_stop": initial_stop, "current_stop": initial_stop, "highest_price": entry_price, "current_price": entry_price, "trailing_distance": round(entry_price - initial_stop, 4), "shares": shares, "current_pl": 0.0, "mfe": 0.0, "exit_price": None, "exit_reason": None, "status": "open", "order_id": order_id, "entry_reason": reason, # [FEATURE 2026-08-18] fade-confirmation grace state -- see # update_position() and config.json's stop.fade_confirmation "fade_grace_active": 0.0, "grace_extensions_used": 0, # [BUGFIX 2026-08-18] see exit_position()/_poll_pending_exit() "exit_pending": False, "exit_order_id": None, "exit_reason_pending": None, # [BUGFIX 2026-08-24] see exit_position()'s "position not # found" branch -- backoff timestamp so a close attempt # rejected because the ENTRY hasn't settled yet doesn't get # blindly resubmitted every single poll cycle. "exit_retry_after": None, } self.positions[symbol] = position self._save() log.info(f"[ENTRY] {symbol} @ ${entry_price:.2f} shares={shares} reason='{reason}'") log.info(f"[STOP] {symbol} initial stop=${initial_stop:.2f}") return position # ------------------------------------------------------------------ def update_position(self, symbol: str, current_price: float, bars: list = None, bars_sub: list = None, quote=None): """ Called every poll cycle for each open position. Updates highest_price / current_stop per the strict "only move up" trailing rule (risk_manager.py -- UNCHANGED by the logic below), and returns an exit signal if either: - the momentum-fade check below fires (independent of the trailing stop -- can exit a position that's still above its stop), or - the stop was hit and the exit wasn't forgiven by fade-confirmation (see _check_fade_confirmed_exit). bars_sub/quote are optional (default None) purely so existing callers/tests that only pass bars don't break; the momentum-fade check just no-ops without them. """ p = self.positions.get(symbol) if p is None or p["status"] != "open": return None p["current_price"] = current_price if current_price > p["highest_price"]: p["highest_price"] = current_price new_stop = risk_manager.update_trailing_stop(p["highest_price"], p["current_stop"], bars) if new_stop > p["current_stop"]: p["current_stop"] = new_stop log.info(f"[TRAIL] {symbol} high=${p['highest_price']:.2f} stop=${new_stop:.2f}") p["current_pl"] = round((current_price - p["entry_price"]) * p["shares"], 2) mfe = round((p["highest_price"] - p["entry_price"]) * p["shares"], 2) p["mfe"] = max(p["mfe"], mfe) momentum_exit = self._check_momentum_fade_exit(symbol, bars, bars_sub, quote) if momentum_exit: self._save() return momentum_exit exit_signal = self._check_fade_confirmed_exit(symbol, p, current_price, bars) self._save() return exit_signal def _check_momentum_fade_exit(self, symbol: str, bars: list, bars_sub: list, quote) -> str: """ [FEATURE 2026-09-11] Proactive exit, independent of the trailing stop entirely -- added after review of 2026-09-11's session found CHPT/EQX/STLA/USDE all gave back most or all of their peak MFE before the trailing stop (or its fade_confirmation grace, since disabled) ever caught them, because intraday_health's 8-minute- bar regression is far too slow to react to a reversal within the first 2-3 minutes of a trade (see that conversation for the worked CHPT example: 8-bar slope was still +0.41% -- "positive" -- the same minute price had already reversed and touched the stop). This check reuses stream_features.compute_features() -- the same fast, sub-30s-bucket pipeline fast_entry_gate.py already runs for entries -- instead of intraday_health, so it reacts on the same timescale entries do. Fires the moment BOTH: - price has stopped climbing: velocity_sub (live price vs. the close of the last completed sub-bucket, %/sec) has dropped to or below velocity_stall_threshold_pct_per_sec. - net buying pressure has flipped negative: pressure.score (blended uptick/downtick ratio + velocity + volume acceleration, -100..+100) is below pressure_threshold. No grace, no widening, no confirm-reads counter -- once both are true on real data, exit immediately. This deliberately can fire while price is still above the trailing stop; a stalled, seller-dominated tape is reason enough to exit on its own, and if it turns out to be a false alarm and the symbol re-firms up away from resistance, same_symbol_reentry_cooldown_minutes=0 already allows getting straight back in for another trail. The one guard against a false read: trade_flow.tick_count_sub can be tiny right after a new sub-bucket starts, making its uptick/downtick ratio (and therefore pressure.score) unreliable. min_ticks_sub holds off the check on that poll (not a multi-poll delay -- just "not enough prints yet to trust this bucket") until the current bucket has actually seen enough trades. """ cfg = get_config()["stop"].get("momentum_fade_exit", {}) if not cfg.get("enabled", False): return None if not bars or not bars_sub or not quote: return None features = compute_features(symbol, bars, bars_sub, quote) if features.insufficient_data: return None tick_count = features.trade_flow.get("tick_count_sub") if not tick_count or tick_count < cfg.get("min_ticks_sub", 3): return None velocity_sub = features.velocity.get("velocity_sub") pressure_score = features.pressure.get("score") if velocity_sub is None or pressure_score is None: return None stalled = velocity_sub <= cfg.get("velocity_stall_threshold_pct_per_sec", 0.0) selling = pressure_score < cfg.get("pressure_threshold", -10) fired = stalled and selling # [FEATURE 2026-09-11] see data_store.append_momentum_fade_snapshot's # docstring -- logged on every reading (fired or not) specifically # so these thresholds can be validated/tuned against real data, # since 2026-09-11's own trades turned out to be unreplayable # against this check for lack of exactly this history. data_store.append_momentum_fade_snapshot({ "symbol": symbol, "price": features.price.get("last"), "velocity_sub": velocity_sub, "pressure_score": pressure_score, "tick_count_sub": tick_count, "stalled": stalled, "selling": selling, "fired": fired, }) if fired: log.info(f"[MOMENTUM] {symbol} stopped climbing (velocity_sub=" f"{velocity_sub:.4f}%/s) with net selling pressure " f"(score={pressure_score:.1f}, ticks={tick_count}); exiting immediately") return "MOMENTUM_FADE" return None def _check_fade_confirmed_exit(self, symbol: str, p: dict, current_price: float, bars: list) -> str: """ [FEATURE 2026-08-18] The real trailing stop (p["current_stop"]) is computed entirely by risk_manager.py above, completely untouched by anything in this method -- it still only ever moves up, exactly as before and as covered by risk_manager's own tests. What changes: touching that stop no longer exits immediately by default. Instead: 1. If price is still above the real stop -> nothing to do; reset any active grace (a genuinely new dip should always get a fresh grace budget, not one partially used by an earlier, unrelated pullback). 2. If price is below the real stop but still within a previously-granted grace window -> hold, no new checks. 3. If price has breached even the grace-widened floor (or this is the first touch, no grace active yet): - Not confirmed fading (intraday_health says this doesn't look like real deterioration) AND grace budget remains -> widen the effective floor by stop.fade_confirmation.grace_distance and hold. - Confirmed fading, OR grace budget exhausted, OR fade-confirmation is disabled, OR no bar data available to evaluate health -> exit ("fail toward exiting", not toward holding, when we can't confirm one way or the other -- this protects capital over giving every dip the benefit of the doubt indefinitely). """ cfg = get_config()["stop"].get("fade_confirmation", {}) current_stop = p["current_stop"] if current_price > current_stop: if p.get("fade_grace_active", 0.0) > 0: log.info(f"[TRAIL] {symbol} recovered above stop ${current_stop:.2f}; " f"fade-confirmation grace reset") p["fade_grace_active"] = 0.0 p["grace_extensions_used"] = 0 return None effective_floor = current_stop - p.get("fade_grace_active", 0.0) if current_price > effective_floor: # Below the real stop but still inside previously-granted # grace room -- hold without re-checking health every poll. return None max_grace = cfg.get("max_grace_extensions", 1) grace_used = p.get("grace_extensions_used", 0) if not cfg.get("enabled", False): log.info(f"[TRAIL] {symbol} stop touched @ ${current_price:.2f} " f"(fade-confirmation disabled); exiting") return "TRAILING_STOP" if not bars: log.info(f"[TRAIL] {symbol} stop touched @ ${current_price:.2f} " f"(no bar data to confirm fade); exiting") return "TRAILING_STOP" if grace_used >= max_grace: log.info(f"[TRAIL] {symbol} stop touched @ ${current_price:.2f} " f"(grace extensions exhausted {grace_used}/{max_grace}); exiting") return "TRAILING_STOP" avg_vol_baseline = get_config()["universe"]["min_avg_daily_volume"] reading = intraday_health.compute_health(symbol, bars, avg_vol_baseline) if intraday_health.is_confirmed_fading(reading): log.info(f"[TRAIL] {symbol} stop touched @ ${current_price:.2f} -- health confirms " f"fading (price_slope={reading.price_slope_class}, " f"vwap_slope={reading.vwap_slope_class}); exiting") return "TRAILING_STOP" grace_distance = cfg.get("grace_distance", 0.05) p["fade_grace_active"] = p.get("fade_grace_active", 0.0) + grace_distance p["grace_extensions_used"] = grace_used + 1 new_floor = current_stop - p["fade_grace_active"] log.info(f"[TRAIL] {symbol} stop touched @ ${current_price:.2f} but health not confirmed " f"fading (price_slope={reading.price_slope_class}, " f"vwap_slope={reading.vwap_slope_class}) -- granting ${grace_distance:.2f} grace " f"({p['grace_extensions_used']}/{max_grace}), effective floor now ${new_floor:.2f}") return None # ------------------------------------------------------------------ def exit_position(self, symbol: str, exit_price: float, reason: str) -> dict: """ Idempotent per symbol. [BUGFIX 2026-08-18, extended 2026-08-24] Safe to call every poll cycle for a symbol that's already exiting -- it will never submit a second close order while one is in flight. First call for an "open" position submits the close (or adopts an already-resting one if Alpaca rejects it as a wash trade) and moves status to "closing". Every subsequent call for a "closing" position just polls that order and finalizes once it's actually filled. [BUGFIX 2026-08-24] A second, distinct rejection -- Alpaca's "position not found" (code 40410000, NOT the wash-trade code the 2026-08-18 fix covers) -- was still causing the exact same blind-retry-every-cycle symptom the 08-18 fix was written to eliminate. Real logs from that day (CRML/FSM/EXK/XPON) show 4-8 consecutive "position not found" rejections, 20-30 seconds apart, over 4-6 minutes, with the position completely unprotected the whole time. Root cause: this error fires when the ENTRY order itself hasn't settled at the broker yet, even though local state already marked the position "open" at submission time (market buys can occasionally take longer to settle than one 5-second poll interval, especially during the order-flow crunch right at market open). See exit_retry_after below. """ p = self.positions.get(symbol) if p is None or p["status"] not in ("open", "closing"): return None if self.simulation: return self._finalize_exit(p, exit_price, reason) if p["status"] == "closing": return self._poll_pending_exit(p) # [BUGFIX 2026-08-24] Respect any backoff window set by a prior # "position not found" rejection below -- skip this cycle # entirely rather than hammering the same request every 5s. retry_after = p.get("exit_retry_after") if retry_after is not None: now = datetime.now(timezone.utc) retry_after_dt = retry_after if isinstance(retry_after, datetime) else \ datetime.fromisoformat(retry_after) if now < retry_after_dt: return None # status == "open": first exit attempt for this position. try: order = self.client.close_position(symbol) order_id = getattr(order, "id", None) except Exception as e: existing_order_id = parse_wash_trade_error(e) if existing_order_id is not None and self._is_resting_sell_order(symbol, existing_order_id): # Alpaca is telling us a close order for this symbol # already exists (e.g. left over from a prior process, # or a previous attempt whose success response we # missed) -- track it instead of discarding the error # and looping forever. log.warning(f"[EXIT] {symbol}: close order already resting at broker " f"(order_id={existing_order_id}); adopting instead of resubmitting") order_id = existing_order_id elif existing_order_id is not None: # [BUGFIX 2026-09-02] existing_order_id from a wash-trade # rejection is NOT guaranteed to be a resting SELL order -- # it can be the entry's own still-unsettled BUY order (the # wash-trade guard fires on ANY opposite-side conflict, and # from Alpaca's perspective an unfilled buy is "opposite # side" to the sell we just tried to submit). Confirmed # live: OWL and NVAX on 2026-09-02 both had this branch # blindly adopt their own entry order_id as exit_order_id; # _poll_pending_exit() later saw that BUY fill and # _finalize_exit() recorded it as the exit -- the position # was marked "closed" locally with a fabricated exit price # while the real shares stayed open, unmanaged (no trailing # stop, no health checks), at the broker for the rest of # the session (RECONCILE flagged both as orphaned broker # positions from ~09:33 through market close). Refusing to # adopt here and backing off instead means the next cycle # retries close_position() cleanly once the entry has # actually settled. log.warning(f"[EXIT] {symbol}: wash-trade rejection named order " f"{existing_order_id} but it is not a resting SELL order " f"for {symbol} (most likely the still-unsettled entry " f"order) -- refusing to adopt it as the exit to avoid a " f"phantom close; backing off instead") self._handle_position_not_found(symbol, p, e) return None elif parse_position_not_found_error(e): self._handle_position_not_found(symbol, p, e) return None else: log.error(f"[EXIT] Failed to close broker position for {symbol}: {e}") # Genuinely unknown failure -- leave status "open" so the # next poll cycle retries the close from scratch. This is # the one case where an immediate retry is still correct: # no order exists yet at the broker, so there's nothing # to race against, and it's not the entry-settlement # race the backoff above is specifically for. return None p["status"] = "closing" p["exit_pending"] = True p["exit_order_id"] = order_id p["exit_reason_pending"] = reason p["exit_retry_after"] = None self._save() log.info(f"[EXIT] {symbol}: close order submitted (order_id={order_id}), awaiting fill") return self._poll_pending_exit(p) def _is_resting_sell_order(self, symbol: str, order_id: str) -> bool: """ [BUGFIX 2026-09-02] Verifies an order_id surfaced by a wash-trade rejection is actually a resting SELL order for this symbol before exit_position() trusts it as "our close order" -- see the long comment at its call site for the phantom-exit failure this prevents. Fails closed (returns False) on any lookup problem, so an unverifiable order is never adopted. """ order = self.client.get_order(order_id) if order is None: return False return (getattr(order, "symbol", None) == symbol and getattr(order, "side", None) == OrderSide.SELL) def _handle_position_not_found(self, symbol: str, p: dict, exc: Exception): """ [BUGFIX 2026-08-24] Distinguishes the benign, expected race (entry order hasn't settled yet -- log at INFO, back off, don't hammer retries) from a genuinely unusual case (entry settled but broker still has no position -- log at ERROR same as before, since we don't have evidence for what that would mean or how to safely auto-recover from it). """ cfg = get_config().get("stop", {}).get("exit_retry", {}) backoff_seconds = cfg.get("position_not_found_backoff_seconds", 15) entry_order_id = p.get("order_id") entry_settled = self._entry_order_settled(entry_order_id) if entry_order_id else None if entry_settled is False: p["exit_retry_after"] = ( datetime.now(timezone.utc) + timedelta(seconds=backoff_seconds) ).isoformat() self._save() log.info(f"[EXIT] {symbol}: close rejected as 'position not found', but entry " f"order {entry_order_id} hasn't settled at the broker yet -- this is " f"expected, backing off {backoff_seconds}s instead of resubmitting " f"immediately") return # entry_settled is True, or we couldn't determine it (no # order_id on record, or the lookup itself failed) -- fall back # to the pre-2026-08-24 behavior: log as an error and retry next # cycle with no backoff, since we don't have positive evidence # this is the settlement race the backoff is meant for. log.error(f"[EXIT] Failed to close broker position for {symbol}: {exc}") def _entry_order_settled(self, order_id: str): """Returns True if the given order is FILLED, False if it's still open/pending, None if the lookup itself failed (treated as 'unknown' by the caller, not as 'settled').""" order = self.client.get_order(order_id) if order is None: return None status = str(getattr(order, "status", "")).lower() if "filled" in status and "partially" not in status: return True return False def _poll_pending_exit(self, p: dict) -> dict: symbol = p["symbol"] order_id = p.get("exit_order_id") if not order_id: # No order to poll (shouldn't happen, but fail safe rather # than getting stuck in "closing" forever). log.error(f"[EXIT] {symbol} is 'closing' with no tracked order_id; " f"resetting to 'open' so the next cycle resubmits a close") p["status"] = "open" p["exit_pending"] = False self._save() return None order = self.client.get_order(order_id) if order is None: # Broker lookup failed this cycle (network blip etc) -- do # NOT resubmit a close; just try polling again next cycle. return None status = str(getattr(order, "status", "")).lower() if "filled" in status and "partially" not in status: fill_price = getattr(order, "filled_avg_price", None) exit_price = float(fill_price) if fill_price is not None else p["current_price"] return self._finalize_exit(p, exit_price, p.get("exit_reason_pending", "TRAILING_STOP")) if status in ("canceled", "expired", "rejected"): # The close order died without filling -- clear pending state # so the next poll cycle submits a fresh close instead of # polling a dead order forever. log.warning(f"[EXIT] {symbol}: tracked close order {order_id} ended in " f"status={status} without filling; will resubmit next cycle") p["status"] = "open" p["exit_pending"] = False p["exit_order_id"] = None self._save() return None # Still open/pending/partially filled -- keep waiting, no new # order submitted, no ERROR logged (this is expected, not a # failure). return None def _finalize_exit(self, p: dict, exit_price: float, reason: str) -> dict: symbol = p["symbol"] p["exit_price"] = exit_price p["exit_reason"] = reason p["exit_time"] = datetime.now(timezone.utc).isoformat() p["status"] = "closed" p["exit_pending"] = False p["current_pl"] = round((exit_price - p["entry_price"]) * p["shares"], 2) self._save() data_store.append_trade_record(p) log.info(f"[EXIT] {symbol} @ ${exit_price:.2f}") log.info(f"[EXIT_REASON] {reason}") pl_sign = "+" if p["current_pl"] >= 0 else "" log.info(f"[P/L] {pl_sign}${p['current_pl']:.2f}") return p # ------------------------------------------------------------------ def liquidate_all(self, reason="END_OF_DAY"): for symbol, p in list(self.positions.items()): if p["status"] in ("open", "closing"): self.exit_position(symbol, p["current_price"], reason) def poll_pending_exits(self): """Call once per poll cycle before anything else touches open positions -- advances any "closing" position toward finalization without re-running trailing-stop/health logic on it.""" for symbol in self.get_closing_symbols(): self.exit_position(symbol, self.positions[symbol]["current_price"], self.positions[symbol].get("exit_reason_pending", "TRAILING_STOP")) def get_open_symbols(self) -> list: """Symbols still eligible for trailing-stop/health evaluation. Excludes "closing" positions on purpose -- once an exit is in flight there's nothing to trail, and re-running update_position() on it would just be wasted work (the position is leaving).""" return [s for s, p in self.positions.items() if p["status"] == "open"] def get_closing_symbols(self) -> list: return [s for s, p in self.positions.items() if p["status"] == "closing"] def _save(self): data_store.save_positions(self.positions) # ------------------------------------------------------------------ def reconcile_with_broker(self): """ Cross-checks local position state against Alpaca's actual open positions. Logs (but does not silently auto-liquidate) any mismatch, since prior systems have shown that automatically adopting/liquidating unexplained broker-side quantities can itself cause losses — a human should review true orphans. [BUGFIX 2026-09-02] Throttled to schedule.reconcile_interval_seconds (default 30s), matching the pattern _update_intraday_health() and _run_intraday_full_rescan() already use. Previously this fired an Alpaca API call on every single 5s main-loop cycle with no gate at all, which is the documented cause of 2026-09-01's 84 RECONCILE warnings and 2026-09-02's 4,514 (42% of that day's entire log volume) -- almost all of them the same expected one-cycle settlement lag right after an entry, not real orphans. That noise is exactly what buried the one day a real orphan (OWL/NVAX, see _is_resting_sell_order()) sat unreconciled for 6+ hours. """ if self.simulation: return interval = get_config()["schedule"].get("reconcile_interval_seconds", 30) now = datetime.now(timezone.utc) if self._last_reconcile is not None and \ (now - self._last_reconcile).total_seconds() < interval: return self._last_reconcile = now try: broker_positions = {p.symbol: p for p in self.client.get_open_positions()} except Exception as e: log.error(f"[RECONCILE] Failed to fetch broker positions: {e}") return local_open = set(self.get_open_symbols()) broker_open = set(broker_positions.keys()) missing_locally = broker_open - local_open missing_at_broker = local_open - broker_open if missing_locally: log.warning(f"[RECONCILE] Broker has positions not tracked locally: " f"{sorted(missing_locally)}. Review manually.") if missing_at_broker: log.warning(f"[RECONCILE] Local ledger shows open positions the broker " f"does not have: {sorted(missing_at_broker)}. Review manually.")