tessl/pypi-alpaca-trade-api
Python client library for Alpaca's commission-free trading API with support for both REST and streaming data interfaces
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
cryptocurrency.mddocs/
Cryptocurrency Data
Specialized market data operations for cryptocurrency trading pairs including trades, quotes, bars, and orderbook data. Supports multiple crypto exchanges with location-based data access.
Capabilities
Crypto Historical Data
Access historical cryptocurrency market data with exchange-specific data sources.
def get_crypto_bars(
symbol: Union[str, List[str]],
timeframe: TimeFrame,
start: str,
end: str,
limit: int = None,
sort: Sort = None,
loc: str = "us"
) -> BarsV2:
"""Get historical crypto bars."""
def get_crypto_bars_iter(
symbol: Union[str, List[str]],
timeframe: TimeFrame,
start: str,
end: str,
limit: int = None,
sort: Sort = None,
loc: str = "us",
raw: bool = False
) -> Iterator[Union[BarV2, dict]]:
"""Get historical crypto bars as an iterator."""
def get_crypto_trades(
symbol: Union[str, List[str]],
start: str,
end: str,
limit: int = None,
sort: Sort = None,
loc: str = "us"
) -> TradesV2:
"""Get historical crypto trades."""
def get_crypto_trades_iter(
symbol: Union[str, List[str]],
start: str,
end: str,
limit: int = None,
sort: Sort = None,
loc: str = "us",
raw: bool = False
) -> Iterator[Union[TradeV2, dict]]:
"""Get historical crypto trades as an iterator."""
def get_crypto_quotes(
symbol: Union[str, List[str]],
start: str,
end: str,
limit: int = None,
sort: Sort = None,
loc: str = "us"
) -> QuotesV2:
"""Get historical crypto quotes."""
def get_crypto_quotes_iter(
symbol: Union[str, List[str]],
start: str,
end: str,
limit: int = None,
sort: Sort = None,
loc: str = "us",
raw: bool = False
) -> Iterator[Union[QuoteV2, dict]]:
"""Get historical crypto quotes as an iterator."""Usage Examples:
# Get Bitcoin daily bars
btc_bars = api.get_crypto_bars(
'BTC/USD',
TimeFrame.Day,
'2023-01-01',
'2023-01-31'
)
# Convert to DataFrame for analysis
df = btc_bars.df
print(f"BTC Price Range: ${df['low'].min():.2f} - ${df['high'].max():.2f}")
print(f"Average Daily Volume: {df['volume'].mean():.2f} BTC")
# Get Ethereum trades with location filter
eth_trades = api.get_crypto_trades(
'ETH/USD',
'2023-01-15T00:00:00',
'2023-01-15T23:59:59',
loc='us' # US exchanges only
)
# Analyze trade sizes
df = eth_trades.df
large_trades = df[df['size'] >= 10] # 10+ ETH trades
print(f"Large trades: {len(large_trades)} out of {len(df)} total trades")Latest Crypto Data
Get the most recent cryptocurrency market data across multiple exchanges.
def get_latest_crypto_bars(symbols: List[str], loc: str = None) -> LatestBarsV2:
"""Get latest crypto bars for multiple symbols."""
def get_latest_crypto_trades(symbols: List[str], loc: str = None) -> LatestTradesV2:
"""Get latest crypto trades for multiple symbols."""
def get_latest_crypto_quotes(symbols: List[str], loc: str = None) -> LatestQuotesV2:
"""Get latest crypto quotes for multiple symbols."""Usage Examples:
# Get latest data for major crypto pairs
crypto_symbols = ['BTC/USD', 'ETH/USD', 'LTC/USD', 'BCH/USD']
latest_bars = api.get_latest_crypto_bars(crypto_symbols)
for symbol, bar in latest_bars.items():
print(f"{symbol}: ${bar.close:,.2f} (24h Vol: {bar.volume:,.2f})")
# Get latest quotes for spread analysis
latest_quotes = api.get_latest_crypto_quotes(['BTC/USD', 'ETH/USD'])
for symbol, quote in latest_quotes.items():
spread = quote.ask_price - quote.bid_price
spread_bps = (spread / quote.bid_price) * 10000
print(f"{symbol} Spread: ${spread:.2f} ({spread_bps:.1f} bps)")Crypto Market Snapshots
Comprehensive market snapshots for cryptocurrency pairs with latest trade, quote, and bar data.
def get_crypto_snapshot(symbol: str, loc: str = "us") -> SnapshotV2:
"""Get crypto market snapshot for a symbol."""
def get_crypto_snapshots(symbols: List[str], loc: str = "us") -> SnapshotsV2:
"""Get crypto market snapshots for multiple symbols."""Usage Examples:
# Get comprehensive crypto market overview
crypto_pairs = ['BTC/USD', 'ETH/USD', 'ADA/USD', 'SOL/USD']
snapshots = api.get_crypto_snapshots(crypto_pairs)
for symbol, snapshot in snapshots.items():
current_price = snapshot.latest_trade.price
daily_open = snapshot.daily_bar.open
change_pct = ((current_price - daily_open) / daily_open) * 100
print(f"{symbol}:")
print(f" Price: ${current_price:,.2f} ({change_pct:+.2f}%)")
print(f" 24h High: ${snapshot.daily_bar.high:,.2f}")
print(f" 24h Low: ${snapshot.daily_bar.low:,.2f}")
print(f" 24h Volume: {snapshot.daily_bar.volume:,.2f}")Crypto Orderbook Data
Access level-2 orderbook data for cryptocurrency pairs showing bid/ask depth.
def get_latest_crypto_orderbook(symbol: str, loc: str = "us") -> OrderbookV2:
"""Get latest crypto orderbook for a symbol."""
def get_latest_crypto_orderbooks(symbols: List[str], loc: str = "us") -> OrderbooksV2:
"""Get latest crypto orderbooks for multiple symbols."""Usage Examples:
# Get Bitcoin orderbook depth
btc_orderbook = api.get_latest_crypto_orderbook('BTC/USD')
print("BTC/USD Orderbook:")
print("Bids:")
for bid in btc_orderbook.bids[:5]: # Top 5 bid levels
print(f" ${bid.price:,.2f} x {bid.size:.4f} BTC")
print("Asks:")
for ask in btc_orderbook.asks[:5]: # Top 5 ask levels
print(f" ${ask.price:,.2f} x {ask.size:.4f} BTC")
# Calculate orderbook imbalance
total_bid_size = sum(bid.size for bid in btc_orderbook.bids[:10])
total_ask_size = sum(ask.size for ask in btc_orderbook.asks[:10])
imbalance = (total_bid_size - total_ask_size) / (total_bid_size + total_ask_size)
print(f"Orderbook imbalance: {imbalance:.3f} ({'buy pressure' if imbalance > 0 else 'sell pressure'})")Crypto Streaming
Real-time streaming for cryptocurrency market data including trades, quotes, bars, and orderbook updates.
def subscribe_crypto_trades(handler: Callable, *symbols: str) -> None:
"""Subscribe to crypto trade streams."""
def subscribe_crypto_quotes(handler: Callable, *symbols: str) -> None:
"""Subscribe to crypto quote streams."""
def subscribe_crypto_bars(handler: Callable, *symbols: str) -> None:
"""Subscribe to crypto bar streams."""
def subscribe_crypto_updated_bars(handler: Callable, *symbols: str) -> None:
"""Subscribe to crypto updated bar streams."""
def subscribe_crypto_daily_bars(handler: Callable, *symbols: str) -> None:
"""Subscribe to crypto daily bar streams."""
def subscribe_crypto_orderbooks(handler: Callable, *symbols: str) -> None:
"""Subscribe to crypto orderbook streams."""
def on_crypto_trade(*symbols: str) -> Callable:
"""Decorator for crypto trade handlers."""
def on_crypto_quote(*symbols: str) -> Callable:
"""Decorator for crypto quote handlers."""
def on_crypto_bar(*symbols: str) -> Callable:
"""Decorator for crypto bar handlers."""
def on_crypto_updated_bar(*symbols: str) -> Callable:
"""Decorator for crypto updated bar handlers."""
def on_crypto_daily_bar(*symbols: str) -> Callable:
"""Decorator for crypto daily bar handlers."""
def on_crypto_orderbook(*symbols: str) -> Callable:
"""Decorator for crypto orderbook handlers."""Usage Examples:
# Subscribe to Bitcoin and Ethereum real-time data
@stream.on_crypto_trade('BTC/USD', 'ETH/USD')
def crypto_trade_handler(trade):
# Monitor large trades
if trade.symbol == 'BTC/USD' and trade.size >= 1.0:
print(f"Large BTC trade: {trade.size:.4f} BTC @ ${trade.price:,.2f}")
elif trade.symbol == 'ETH/USD' and trade.size >= 10.0:
print(f"Large ETH trade: {trade.size:.2f} ETH @ ${trade.price:,.2f}")
@stream.on_crypto_quote('BTC/USD')
def crypto_quote_handler(quote):
# Monitor tight spreads
spread_bps = ((quote.ask_price - quote.bid_price) / quote.bid_price) * 10000
if spread_bps < 5: # Very tight spread
print(f"Tight BTC spread: {spread_bps:.1f} bps")
@stream.on_crypto_orderbook('BTC/USD', 'ETH/USD')
def crypto_orderbook_handler(orderbook):
# Monitor orderbook depth
best_bid_size = orderbook.bids[0].size if orderbook.bids else 0
best_ask_size = orderbook.asks[0].size if orderbook.asks else 0
if best_bid_size > 5.0 or best_ask_size > 5.0: # Large size at best price
print(f"{orderbook.symbol} large size: Bid={best_bid_size:.2f}, Ask={best_ask_size:.2f}")
# Unsubscribe from crypto streams
stream.unsubscribe_crypto_trades('BTC/USD')
stream.unsubscribe_crypto_quotes('ETH/USD')
stream.unsubscribe_crypto_orderbooks('BTC/USD', 'ETH/USD')Crypto Exchange Support
The library supports multiple cryptocurrency exchanges through the loc parameter:
- US exchanges: Set
loc='us'for US-based crypto exchanges - Global exchanges: Leave
loc=Noneor omit for global data aggregation - Exchange-specific: Specify exchange codes for targeted data sources
Usage Example:
# Compare prices across different locations
us_btc_bars = api.get_crypto_bars('BTC/USD', TimeFrame.Hour, '2023-01-15', '2023-01-16', loc='us')
global_btc_bars = api.get_crypto_bars('BTC/USD', TimeFrame.Hour, '2023-01-15', '2023-01-16')
us_avg_price = us_btc_bars.df['close'].mean()
global_avg_price = global_btc_bars.df['close'].mean()
price_diff = abs(us_avg_price - global_avg_price)
print(f"US Average: ${us_avg_price:,.2f}")
print(f"Global Average: ${global_avg_price:,.2f}")
print(f"Price Difference: ${price_diff:.2f}")Types
class OrderbookV2:
@property
def symbol(self) -> str: ...
@property
def timestamp(self) -> pd.Timestamp: ...
@property
def exchange(self) -> str: ...
@property
def bids(self) -> List[BidOrAsk]: ...
@property
def asks(self) -> List[BidOrAsk]: ...
class BidOrAsk:
@property
def price(self) -> float: ...
@property
def size(self) -> float: ...
# Other types are the same as standard market data types:
# BarV2, TradeV2, QuoteV2, SnapshotV2, etc.Install with Tessl CLI
npx tessl i tessl/pypi-alpaca-trade-api