diff --git a/alpaca_paper/sync.py b/alpaca_paper/sync.py index b00b4e2..71272e3 100644 --- a/alpaca_paper/sync.py +++ b/alpaca_paper/sync.py @@ -8,6 +8,7 @@ from __future__ import annotations import logging from datetime import datetime +from zoneinfo import ZoneInfo import json @@ -31,6 +32,7 @@ _EXIT_REASON = { logger = logging.getLogger(__name__) BROKER_NAME = "alpaca_paper" +_ET = ZoneInfo("America/New_York") class AlpacaPaperSync: @@ -56,6 +58,10 @@ class AlpacaPaperSync: # until the first sync, when it's seeded from the last PERSISTED snapshot # so a close that happened while the loop was down is still journaled. self._prev_positions: list[dict] | None = None + # exit order ids already journaled as position_closed (position_id), so the + # orders-based pass below never double-counts an exit the snapshot diff + # (or an earlier run) already recorded. None until seeded from the store. + self._journaled_exits: set[str] | None = None def _emit_health(self, reason: str) -> None: """Make a broker sync failure VISIBLE (else stale snapshots look current).""" @@ -230,6 +236,92 @@ class AlpacaPaperSync: reason = _EXIT_REASON.get(str(f.get("type") or ""), "closed") return px, reason, stop_level, f.get("broker_order_id") + def _load_journaled_exit_ids(self) -> set[str]: + """position_ids of the closes already in the store (an exit's broker order + id, see reconcile_closed) so a restart does not re-journal them.""" + try: + rows = self.store.con.execute( + "SELECT payload_json FROM events WHERE event_type = 'position_closed' " + "ORDER BY timestamp DESC LIMIT 2000").fetchall() + return {str(json.loads(r[0] or "{}").get("position_id") or "") for r in rows} - {""} + except Exception: # noqa: BLE001 + return set() + + @staticmethod + def _same_session(submitted_at, today) -> bool: + """True if an order's submitted_at (ISO, UTC) falls on ``today`` (ET), or is + unparseable (Tradier's list is already scoped to the current session).""" + if not submitted_at: + return True + try: + ts = datetime.fromisoformat(str(submitted_at).replace("Z", "+00:00")) + return ts.astimezone(_ET).date() == today + except ValueError: + return True + + def reconcile_unseen_round_trips(self, prev_positions, new_positions, orders, + today=None) -> int: + """Journal round trips that opened AND closed between two position polls. + + 2026-10-02 TNON: bought at 09:38:37, the bracket stop filled at 09:39:26, + and no positions snapshot ever held it, so the snapshot diff in + reconcile_closed had nothing to diff and the trade vanished from the + journal (the 80-trade scorecard dropped a -1R loser). The broker's order + list still shows both fills, so for any symbol that is open in NEITHER + snapshot but has a filled buy and a filled sell submitted this session, + emit a position_closed per unjournaled exit fill: entry = VWAP of the buy + fills, exit = that sell's fill price, stop = the session's stop order. + Long round trips only (the bot is long-only). Returns the number emitted.""" + if self._journaled_exits is None: + self._journaled_exits = self._load_journaled_exit_ids() + today = today or datetime.now(_ET).date() + open_syms = {p.get("symbol") for p in (prev_positions or []) + (new_positions or []) + if abs(float(p.get("quantity") or 0)) > 0} + by_sym: dict[str, list[dict]] = {} + for o in orders or []: + sym = o.get("symbol") + if (not sym or sym in open_syms or str(o.get("status") or "").lower() != "filled" + or not o.get("type") or o.get("filled_avg_price") in (None, "") + or float(o.get("filled_quantity") or 0) <= 0 + or not self._same_session(o.get("submitted_at"), today)): + continue + by_sym.setdefault(sym, []).append(o) + emitted = 0 + for sym, fills in by_sym.items(): + buys = [o for o in fills if str(o.get("side") or "").lower() == "buy"] + sells = [o for o in fills if str(o.get("side") or "").lower() == "sell" + and str(o.get("broker_order_id") or "") not in self._journaled_exits] + if not buys or not sells: + continue + bought = sum(float(o["filled_quantity"]) for o in buys) + entry = sum(float(o["filled_quantity"]) * float(o["filled_avg_price"]) for o in buys) / bought + stop_level = None + for o in fills: + if str(o.get("type") or "") in ("stop", "stop_limit") and o.get("stop_price") not in (None, ""): + try: + stop_level = float(o["stop_price"]) + except (TypeError, ValueError): + pass + for f in sorted(sells, key=lambda o: str(o.get("submitted_at") or "")): + qty = min(float(f["filled_quantity"]), bought) + exit_price = float(f["filled_avg_price"]) + reason = _EXIT_REASON.get(str(f.get("type") or ""), "closed") + realized = (exit_price - entry) * qty + oid = str(f.get("broker_order_id") or f"{sym}-{datetime.now().isoformat()}") + self.store.emit(PositionClosedEvent( + timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id, + message=(f"{sym} closed @ {exit_price:.4f} ({reason}) pnl={realized:+.2f} " + f"[round trip between polls]"), + position_id=oid, symbol=sym, exit_price=round(exit_price, 4), + exit_reason=reason, realized_pnl=round(realized, 2), + entry_price=round(entry, 4), + stop_loss_price=round(stop_level, 4) if stop_level else None, + side="buy", quantity=qty, + )) + self._journaled_exits.add(oid) + emitted += 1 + return emitted + def reconcile_closed(self, prev_positions, new_positions, orders) -> int: """Emit a position_closed event for every symbol that was open in the previous snapshot and is gone (or flat) in the new one. This is the ONLY @@ -264,6 +356,10 @@ class AlpacaPaperSync: stop_loss_price=round(stop_level, 4) if stop_level else None, side=("sell" if is_short else "buy"), quantity=qty, )) + if oid: + if self._journaled_exits is None: + self._journaled_exits = self._load_journaled_exit_ids() + self._journaled_exits.add(str(oid)) emitted += 1 return emitted @@ -281,5 +377,9 @@ class AlpacaPaperSync: self.reconcile_closed(self._prev_positions, positions, orders or []) except Exception as exc: # noqa: BLE001 — never let journaling break sync logger.warning("position-close reconcile failed: %s", exc) + try: + self.reconcile_unseen_round_trips(self._prev_positions, positions, orders or []) + except Exception as exc: # noqa: BLE001 + logger.warning("round-trip reconcile failed: %s", exc) self._prev_positions = positions return {"account": account, "positions": positions, "orders": orders} diff --git a/tests/unit/alpaca_paper/test_execution_and_sync.py b/tests/unit/alpaca_paper/test_execution_and_sync.py index 281f5cc..f06d4d6 100644 --- a/tests/unit/alpaca_paper/test_execution_and_sync.py +++ b/tests/unit/alpaca_paper/test_execution_and_sync.py @@ -346,3 +346,72 @@ def test_broker_rejection_not_reconciled(store): assert not result.ok assert client.lookup_calls == 0 assert store.query_events(event_type="order_cancelled") + + +def _today_iso(hhmm): + """An ISO timestamp on today's ET session date, so the same-session filter admits it.""" + from datetime import datetime + from zoneinfo import ZoneInfo + d = datetime.now(ZoneInfo("America/New_York")).date() + return f"{d.isoformat()}T{hhmm}:00-04:00" + + +def test_round_trip_between_polls_is_journaled_once(store): + # TNON 2026-10-02: bought 7 @4.62 and the bracket stop filled 49 s later — no positions + # snapshot ever held it, so the snapshot diff alone journaled nothing. + c = ProgClient() + sync = AlpacaPaperSync(store, client=c, session_id="t") + c.positions = [] + sync.sync_all() # baseline: flat, no orders + c.orders = [ + {"id": "p-1", "symbol": "TNON", "side": "buy", "qty": "7", "filled_qty": "0", "type": None, + "status": "filled", "stop_price": None, "filled_avg_price": None, "submitted_at": _today_iso("09:38")}, + {"id": "b-1", "symbol": "TNON", "side": "buy", "qty": "7", "filled_qty": "7", "type": "limit", + "status": "filled", "stop_price": None, "filled_avg_price": "4.62", "submitted_at": _today_iso("09:38")}, + {"id": "tp-1", "symbol": "TNON", "side": "sell", "qty": "7", "filled_qty": "0", "type": "limit", + "status": "canceled", "stop_price": None, "filled_avg_price": None, "submitted_at": _today_iso("09:38")}, + {"id": "s-1", "symbol": "TNON", "side": "sell", "qty": "7", "filled_qty": "7", "type": "stop", + "status": "filled", "stop_price": "4.30", "filled_avg_price": "4.30", "submitted_at": _today_iso("09:38")}, + ] + sync.sync_all() # still flat: the whole trade happened between polls + rows = _closed(store) + assert len(rows) == 1 + p = rows[0] + assert (p["symbol"], p["entry_price"], p["exit_price"], p["exit_reason"]) == ("TNON", 4.62, 4.30, "stop_loss") + assert p["quantity"] == 7 and p["realized_pnl"] == -2.24 and p["stop_loss_price"] == 4.30 + assert p["position_id"] == "s-1" + sync.sync_all() # the next poll must not journal it again + assert len(_closed(store)) == 1 + # ...and neither must a fresh sync object after a restart (seeded from the store) + AlpacaPaperSync(store, client=c, session_id="t2").sync_all() + assert len(_closed(store)) == 1 + + +def test_round_trip_pass_leaves_snapshot_diff_closes_alone(store): + # the normal path: GOOD opens in one poll and is gone in the next; exactly one close, + # even though the orders list then shows a completed buy+sell for a flat symbol + c = ProgClient() + sync = AlpacaPaperSync(store, client=c, session_id="t") + c.positions = [_long("GOOD", 13.90, 14.10)] + c.orders = [_order("buy-1", "GOOD", "buy", "market", "filled", 13.90, at=_today_iso("09:31"))] + sync.sync_all() + c.positions = [] + c.orders.append(_order("stop-1", "GOOD", "sell", "stop", "filled", 13.45, stop=13.45, at=_today_iso("09:40"))) + sync.sync_all(); sync.sync_all() + assert [p["position_id"] for p in _closed(store)] == ["stop-1"] + + +def test_round_trip_pass_ignores_other_sessions_and_open_names(store): + c = ProgClient() + sync = AlpacaPaperSync(store, client=c, session_id="t") + c.positions = [_long("HOLD", 5.0, 5.2)] + c.orders = [ + # yesterday's completed trade must not be re-journaled today + _order("ob", "OLD", "buy", "limit", "filled", 2.0, at="2026-06-22T13:31:00Z"), + _order("os", "OLD", "sell", "stop", "filled", 1.9, stop=1.9, at="2026-06-22T13:35:00Z"), + # a name that is still open: its partial sell is the diff path's business + _order("hb", "HOLD", "buy", "limit", "filled", 5.0, at=_today_iso("09:31")), + _order("hs", "HOLD", "sell", "limit", "filled", 5.3, at=_today_iso("09:50")), + ] + sync.sync_all(); sync.sync_all() + assert _closed(store) == []