Spaces:
Runtime error
Runtime error
File size: 5,536 Bytes
d5b7ee9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 | """Base adapter interface — all exchange adapters must implement this."""
from __future__ import annotations
import logging
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any
import pandas as pd
logger = logging.getLogger(__name__)
@dataclass
class Position:
"""Unified position object across all exchanges."""
symbol: str
qty: float
avg_entry_price: float
current_price: float
unrealized_pl: float
unrealized_plpc: float
market_value: float
side: str = "long"
@dataclass
class AccountInfo:
"""Unified account info across all exchanges."""
equity: float
cash: float
buying_power: float
portfolio_value: float
@dataclass
class OrderResult:
"""Unified order result across all exchanges."""
order_id: str
symbol: str
action: str # BUY or SELL
qty: int
status: str # filled, rejected, pending, etc.
filled_price: float | None = None
@dataclass
class MarketClock:
"""Market hours info."""
is_open: bool
next_open: str
next_close: str
class TradingAdapter(ABC):
"""Abstract base class for all trading platform adapters.
Implement this class to add support for new exchanges (Binance, Kraken, etc.).
Each adapter handles:
- Account info retrieval
- Position management
- Order execution
- Market data (OHLCV, quotes)
- Market clock
"""
@property
@abstractmethod
def adapter_id(self) -> str:
"""Unique identifier for this adapter (e.g., 'alpaca', 'binance', 'kraken')."""
...
@property
@abstractmethod
def supports_paper_trading(self) -> bool:
"""Whether this adapter supports paper/demo trading."""
...
@property
@abstractmethod
def is_demo_mode(self) -> bool:
"""True if running in demo/mock mode (no real API connection)."""
...
# ── Account & Positions ───────────────────────────────────────────────────
@abstractmethod
def get_account(self) -> AccountInfo:
"""Get account balance and buying power."""
...
@abstractmethod
def get_positions(self) -> list[Position]:
"""Get all open positions."""
...
# ── Orders ────────────────────────────────────────────────────────────────
@abstractmethod
def submit_market_order(self, symbol: str, qty: int, side: str) -> OrderResult:
"""Submit a market order.
Args:
symbol: Trading symbol (e.g., 'AAPL', 'BTC/USD').
qty: Number of shares/units.
side: 'BUY' or 'SELL'.
Returns:
OrderResult with status and fill details.
"""
...
@abstractmethod
def close_position(self, symbol: str) -> OrderResult | None:
"""Close an existing position at market price.
Returns None if no position exists for the symbol.
"""
...
# ── Market Data ───────────────────────────────────────────────────────────
@abstractmethod
def fetch_ohlcv(self, symbol: str, days: int = 90) -> pd.DataFrame:
"""Fetch historical OHLCV bars.
Returns DataFrame with columns: Open, High, Low, Close, Volume.
Index should be datetime.
"""
...
@abstractmethod
def get_latest_quote(self, symbol: str) -> float | None:
"""Get latest trade price for a symbol."""
...
def get_latest_quotes_batch(self, symbols: list[str]) -> dict[str, float]:
"""Get latest prices for multiple symbols (batch optimized).
Override if the exchange supports batch requests.
Default implementation calls get_latest_quote for each symbol.
"""
prices: dict[str, float] = {}
for sym in symbols:
price = self.get_latest_quote(sym)
if price is not None:
prices[sym] = price
return prices
# ── Market Info ───────────────────────────────────────────────────────────
@abstractmethod
def get_market_clock(self) -> MarketClock:
"""Get market open/closed status and next open/close times."""
...
# ── News (optional) ───────────────────────────────────────────────────────
def fetch_news(self, symbol: str, max_articles: int = 50,
days_ago: int = 0) -> list[tuple[str, float]]:
"""Fetch news headlines with timestamps.
Returns list of (headline, unix_timestamp) tuples.
Override if the exchange provides news data.
Default returns empty list.
"""
return []
# ── Utilities ─────────────────────────────────────────────────────────────
def __repr__(self) -> str:
return f"<{self.__class__.__name__} adapter_id={self.adapter_id} demo={self.is_demo_mode}>"
|