alpaclaudia/bot/alpaclaudia/client.py

"""Thin Alpaca client wrappers + helpers for spot, chains, and account state."""
from __future__ import annotations

from dataclasses import dataclass
from datetime import date, timedelta
from typing import Any

from alpaca.data.historical.option import OptionHistoricalDataClient
from alpaca.data.historical.stock import StockHistoricalDataClient
from alpaca.data.requests import OptionChainRequest, StockLatestQuoteRequest
from alpaca.trading.client import TradingClient
from alpaca.trading.enums import AssetClass
from alpaca.trading.requests import GetOptionContractsRequest

from .config import AlpacaConfig


@dataclass
class Clients:
    trading: TradingClient
    stock_data: StockHistoricalDataClient
    option_data: OptionHistoricalDataClient


def build_clients(cfg: AlpacaConfig) -> Clients:
    if not cfg.key or not cfg.secret:
        raise RuntimeError(
            "Alpaca credentials missing. Set ALPACA_API_KEY / ALPACA_API_SECRET."
        )
    trading = TradingClient(cfg.key, cfg.secret, paper=cfg.is_paper)
    stock_data = StockHistoricalDataClient(cfg.key, cfg.secret)
    option_data = OptionHistoricalDataClient(cfg.key, cfg.secret)
    return Clients(trading=trading, stock_data=stock_data, option_data=option_data)


def get_latest_spot(clients: Clients, symbol: str) -> float | None:
    req = StockLatestQuoteRequest(symbol_or_symbols=symbol)
    resp = clients.stock_data.get_stock_latest_quote(req)
    q = resp.get(symbol) if isinstance(resp, dict) else getattr(resp, symbol, None)
    if q is None:
        return None
    # mid-price if both sides present, else fallback to either
    bid = getattr(q, "bid_price", None) or 0
    ask = getattr(q, "ask_price", None) or 0
    if bid and ask:
        return (bid + ask) / 2
    return bid or ask or None


def list_option_contracts(
    clients: Clients,
    underlying: str,
    *,
    option_type: str,  # "put" or "call"
    dte_min: int,
    dte_max: int,
) -> list[Any]:
    today = date.today()
    req = GetOptionContractsRequest(
        underlying_symbols=[underlying],
        type=option_type,
        expiration_date_gte=today + timedelta(days=dte_min),
        expiration_date_lte=today + timedelta(days=dte_max),
        status="active",
        limit=500,
    )
    resp = clients.trading.get_option_contracts(req)
    return list(getattr(resp, "option_contracts", resp) or [])


def get_option_snapshots(clients: Clients, underlying: str) -> dict[str, Any]:
    """Returns {contract_symbol: snapshot} with greeks/quote when available."""
    try:
        req = OptionChainRequest(underlying_symbol=underlying)
        chain = clients.option_data.get_option_chain(req)
        return dict(chain) if chain else {}
    except Exception:
        return {}


def get_account_snapshot(clients: Clients) -> dict[str, Any]:
    acct = clients.trading.get_account()
    return {
        "equity": float(getattr(acct, "equity", 0) or 0),
        "cash": float(getattr(acct, "cash", 0) or 0),
        "buying_power": float(getattr(acct, "buying_power", 0) or 0),
        "portfolio_value": float(getattr(acct, "portfolio_value", 0) or 0),
        "status": str(getattr(acct, "status", "")),
        "pattern_day_trader": bool(getattr(acct, "pattern_day_trader", False)),
    }


def list_positions(clients: Clients) -> list[dict[str, Any]]:
    pos = clients.trading.get_all_positions() or []
    out: list[dict[str, Any]] = []
    for p in pos:
        out.append(
            {
                "symbol": getattr(p, "symbol", ""),
                "asset_class": str(getattr(p, "asset_class", "")),
                "qty": float(getattr(p, "qty", 0) or 0),
                "avg_entry_price": float(getattr(p, "avg_entry_price", 0) or 0),
                "market_value": float(getattr(p, "market_value", 0) or 0),
                "unrealized_pl": float(getattr(p, "unrealized_pl", 0) or 0),
                "unrealized_plpc": float(getattr(p, "unrealized_plpc", 0) or 0),
                "current_price": float(getattr(p, "current_price", 0) or 0),
                "side": str(getattr(p, "side", "")),
                "is_option": str(getattr(p, "asset_class", "")).lower().endswith("option"),
            }
        )
    return out


def list_recent_orders(clients: Clients, *, limit: int = 100) -> list[dict[str, Any]]:
    from alpaca.trading.requests import GetOrdersRequest

    req = GetOrdersRequest(status="all", limit=limit, nested=True)
    orders = clients.trading.get_orders(filter=req) or []
    out: list[dict[str, Any]] = []
    for o in orders:
        out.append(
            {
                "id": str(getattr(o, "id", "")),
                "symbol": getattr(o, "symbol", ""),
                "side": str(getattr(o, "side", "")),
                "qty": float(getattr(o, "qty", 0) or 0),
                "filled_qty": float(getattr(o, "filled_qty", 0) or 0),
                "order_type": str(getattr(o, "order_type", "")),
                "status": str(getattr(o, "status", "")),
                "submitted_at": str(getattr(o, "submitted_at", "")),
                "filled_avg_price": (
                    float(getattr(o, "filled_avg_price", 0) or 0)
                    if getattr(o, "filled_avg_price", None) is not None
                    else None
                ),
                "asset_class": str(getattr(o, "asset_class", AssetClass.US_EQUITY)),
            }
        )
    return out