momentum-system/alpaca_paper/sync.py

"""Sync the Alpaca paper account into the event store.

Emits account_summary_updated / account_positions_updated /
account_orders_updated events that the dashboard projections read.
"""

from __future__ import annotations

import logging
from datetime import datetime

from alpaca_paper.client import AlpacaPaperClient
from storage.event_schema import (
    AccountOrdersUpdatedEvent,
    AccountPositionsUpdatedEvent,
    AccountSummaryUpdatedEvent,
    BrokerHealthChangedEvent,
    EventMode,
)
from storage.event_store import EventStore

logger = logging.getLogger(__name__)

BROKER_NAME = "alpaca_paper"


class AlpacaPaperSync:
    def __init__(
        self,
        store: EventStore,
        client: AlpacaPaperClient | None = None,
        session_id: str | None = None,
        mode: EventMode = EventMode.PAPER,
    ):
        self.store = store
        self.client = client or AlpacaPaperClient()
        self.session_id = session_id
        self.mode = mode

    def _emit_health(self, reason: str) -> None:
        """Make a broker sync failure VISIBLE (else stale snapshots look current)."""
        self.store.emit(
            BrokerHealthChangedEvent(
                timestamp=datetime.now(),
                mode=self.mode,
                correlation_id=self.session_id,
                message=f"alpaca_paper sync degraded: {reason}",
                broker_name=BROKER_NAME,
                previous_health="healthy",
                new_health="degraded",
                health_reason=reason,
            )
        )

    def sync_account(self) -> dict | None:
        try:
            account = self.client.get_account()
        except Exception as exc:
            logger.warning("alpaca account sync failed: %s", exc)
            self._emit_health(f"account: {exc}")
            return None
        account_id = str(account.get("account_number") or account.get("id") or "paper")
        self.store.emit(
            AccountSummaryUpdatedEvent(
                timestamp=datetime.now(),
                mode=self.mode,
                correlation_id=self.session_id,
                message=f"Alpaca paper equity {account.get('equity')}",
                broker_name=BROKER_NAME,
                account_id=account_id,
                account_desc="Alpaca Paper",
                total_equity=float(account.get("equity") or 0),
                cash_balance=float(account.get("cash") or 0),
                buying_power=float(account.get("buying_power") or 0),
                net_liquidating_value=float(account.get("equity") or 0),
                last_equity=float(account.get("last_equity") or 0),
            )
        )
        return account

    def sync_positions(self) -> list[dict] | None:
        try:
            raw = self.client.get_positions()
        except Exception as exc:
            logger.warning("alpaca positions sync failed: %s", exc)
            self._emit_health(f"positions: {exc}")
            return None
        positions = [
            {
                "symbol": p.get("symbol"),
                "quantity": float(p.get("qty") or 0),
                "avg_entry_price": float(p.get("avg_entry_price") or 0),
                "current_price": float(p.get("current_price") or 0),
                "market_value": float(p.get("market_value") or 0),
                "unrealized_pnl": float(p.get("unrealized_pl") or 0),
                "unrealized_pnl_pct": float(p.get("unrealized_plpc") or 0),
                "side": p.get("side"),
            }
            for p in raw
        ]
        self.store.emit(
            AccountPositionsUpdatedEvent(
                timestamp=datetime.now(),
                mode=self.mode,
                correlation_id=self.session_id,
                message=f"Alpaca paper positions: {len(positions)}",
                broker_name=BROKER_NAME,
                account_id="paper",
                positions=positions,
            )
        )
        return positions

    def sync_orders(self, status: str = "all", limit: int = 500) -> list[dict] | None:
        try:
            raw = self.client.get_orders(status=status, limit=limit, nested=True)
        except Exception as exc:
            logger.warning("alpaca orders sync failed: %s", exc)
            self._emit_health(f"orders: {exc}")
            return None

        def _flat(o: dict, parent: dict | None = None) -> dict:
            return {
                "broker_order_id": o.get("id"),
                "client_order_id": o.get("client_order_id"),
                # bracket child legs (stop/take-profit) don't repeat the symbol —
                # inherit it from the parent so an exit fill isn't dropped.
                "symbol": o.get("symbol") or (parent or {}).get("symbol"),
                "side": o.get("side"),
                "quantity": float(o.get("qty") or 0),
                "filled_quantity": float(o.get("filled_qty") or 0),
                "type": o.get("type"),
                "status": o.get("status"),
                "limit_price": o.get("limit_price"),
                # capture the stop trigger too — needed for the dashboard R bar /
                # spark stop-line / any stop-level read (only limit_price was kept).
                "stop_price": o.get("stop_price"),
                "filled_avg_price": o.get("filled_avg_price"),
                "submitted_at": o.get("submitted_at"),
            }

        # Flatten parent orders AND their child legs. A position exited by its
        # bracket STOP or TAKE-PROFIT leg has that exit fill as a LEG, not a
        # top-level order — recording only parents made every stop/TP exit
        # invisible (positions showed falsely "open", realized P&L understated).
        orders = []
        for o in raw:
            orders.append(_flat(o))
            for leg in (o.get("legs") or []):
                orders.append(_flat(leg, parent=o))
        self.store.emit(
            AccountOrdersUpdatedEvent(
                timestamp=datetime.now(),
                mode=self.mode,
                correlation_id=self.session_id,
                message=f"Alpaca paper orders: {len(orders)}",
                broker_name=BROKER_NAME,
                account_id="paper",
                orders=orders,
            )
        )
        return orders

    def sync_all(self) -> dict:
        return {
            "account": self.sync_account(),
            "positions": self.sync_positions(),
            "orders": self.sync_orders(),
        }