CtrlK
BlogDocsLog inGet started
Tessl Logo

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

Overview
Eval results
Files

market-data.mddocs/

Market Data Access

Historical and real-time market data for stocks and cryptocurrencies including bars, trades, quotes, snapshots, and news. Supports multiple data feeds and provides pandas DataFrame integration for analysis workflows.

Capabilities

Historical Bars

Retrieve historical price bars (OHLCV data) with customizable time frames and data feeds.

def get_bars(
    symbol: Union[str, List[str]],
    timeframe: TimeFrame,
    start: str,
    end: str,
    adjustment: str = "raw",
    limit: int = None,
    feed: str = None,
    asof: str = None,
    sort: Sort = None
) -> BarsV2:
    """Get historical bars for a symbol."""

def get_bars_iter(
    symbol: Union[str, List[str]],
    timeframe: TimeFrame,
    start: str,
    end: str,
    adjustment: str = "raw",
    limit: int = None,
    feed: str = None,
    asof: str = None,
    sort: Sort = None,
    raw: bool = False
) -> Iterator[Union[BarV2, dict]]:
    """Get historical bars as an iterator."""

Usage Examples:

from alpaca_trade_api import TimeFrame

# Get daily bars for the last month
bars = api.get_bars(
    'AAPL', 
    TimeFrame.Day,
    '2023-01-01',
    '2023-01-31'
)

# Convert to pandas DataFrame
df = bars.df
print(f"Retrieved {len(df)} bars")
print(df.head())

# Get minute bars with iterator for large datasets
for bar in api.get_bars_iter('TSLA', TimeFrame.Minute, '2023-01-01', '2023-01-02'):
    print(f"{bar.timestamp}: O={bar.open}, H={bar.high}, L={bar.low}, C={bar.close}, V={bar.volume}")

Historical Trades

Access individual trade execution data with detailed trade information.

def get_trades(
    symbol: Union[str, List[str]],
    start: str,
    end: str,
    limit: int = None,
    feed: str = None,
    asof: str = None,
    sort: Sort = None
) -> TradesV2:
    """Get historical trades for a symbol."""

def get_trades_iter(
    symbol: Union[str, List[str]],
    start: str,
    end: str,
    limit: int = None,
    feed: str = None,
    asof: str = None,
    sort: Sort = None,
    raw: bool = False
) -> Iterator[Union[TradeV2, dict]]:
    """Get historical trades as an iterator."""

Usage Examples:

# Get trades for a specific day
trades = api.get_trades('AAPL', '2023-01-15', '2023-01-16')
df = trades.df
print(f"Total trades: {len(df)}")
print(f"Volume weighted average price: {(df['price'] * df['size']).sum() / df['size'].sum()}")

# Process trades with iterator
total_volume = 0
for trade in api.get_trades_iter('TSLA', '2023-01-15T09:30:00', '2023-01-15T16:00:00'):
    total_volume += trade.size
    if trade.price > 200:
        print(f"Large trade: {trade.size} shares at ${trade.price}")

Historical Quotes

Retrieve bid/ask quote data for market analysis and spread calculations.

def get_quotes(
    symbol: Union[str, List[str]],
    start: str,
    end: str,
    limit: int = None,
    feed: str = None,
    asof: str = None,
    sort: Sort = None
) -> QuotesV2:
    """Get historical quotes for a symbol."""

def get_quotes_iter(
    symbol: Union[str, List[str]],
    start: str,
    end: str,
    limit: int = None,
    feed: str = None,
    asof: str = None,
    sort: Sort = None,
    raw: bool = False
) -> Iterator[Union[QuoteV2, dict]]:
    """Get historical quotes as an iterator."""

Usage Examples:

# Get quotes and analyze bid-ask spreads
quotes = api.get_quotes('AAPL', '2023-01-15T09:30:00', '2023-01-15T16:00:00')
df = quotes.df
df['spread'] = df['ask_price'] - df['bid_price']
df['spread_bps'] = (df['spread'] / df['bid_price']) * 10000
print(f"Average spread: {df['spread_bps'].mean():.2f} basis points")

Latest Market Data

Get the most recent market data for real-time analysis and decision making.

def get_latest_bar(symbol: str, feed: str = None) -> BarV2:
    """Get the latest bar for a symbol."""

def get_latest_bars(symbols: List[str], feed: str = None) -> LatestBarsV2:
    """Get latest bars for multiple symbols."""

def get_latest_trade(symbol: str, feed: str = None) -> TradeV2:
    """Get the latest trade for a symbol."""  

def get_latest_trades(symbols: List[str], feed: str = None) -> LatestTradesV2:
    """Get latest trades for multiple symbols."""

def get_latest_quote(symbol: str, feed: str = None) -> QuoteV2:
    """Get the latest quote for a symbol."""

def get_latest_quotes(symbols: List[str], feed: str = None) -> LatestQuotesV2:
    """Get latest quotes for multiple symbols."""

Usage Examples:

# Get latest data for multiple symbols
symbols = ['AAPL', 'TSLA', 'GOOGL', 'MSFT']
latest_bars = api.get_latest_bars(symbols)
for symbol, bar in latest_bars.items():
    print(f"{symbol}: ${bar.close} (Volume: {bar.volume:,})")

# Get current quote data
quote = api.get_latest_quote('AAPL')
print(f"AAPL Bid: ${quote.bid_price} x {quote.bid_size}")
print(f"AAPL Ask: ${quote.ask_price} x {quote.ask_size}")
print(f"Spread: ${quote.ask_price - quote.bid_price:.2f}")

Market Snapshots

Comprehensive market snapshots containing latest trade, quote, and bar data in a single request.

def get_snapshot(symbol: str, feed: str = None) -> SnapshotV2:
    """Get a market snapshot for a symbol."""

def get_snapshots(symbols: List[str], feed: str = None) -> SnapshotsV2:
    """Get market snapshots for multiple symbols."""

Usage Examples:

# Get comprehensive market data
snapshot = api.get_snapshot('AAPL')
print(f"Latest Trade: ${snapshot.latest_trade.price} at {snapshot.latest_trade.timestamp}")
print(f"Latest Quote: ${snapshot.latest_quote.bid_price} x ${snapshot.latest_quote.ask_price}")
print(f"Daily Bar: O=${snapshot.daily_bar.open} H=${snapshot.daily_bar.high} L=${snapshot.daily_bar.low} C=${snapshot.daily_bar.close}")

# Get snapshots for portfolio monitoring
portfolio_symbols = ['AAPL', 'TSLA', 'GOOGL']
snapshots = api.get_snapshots(portfolio_symbols)
for symbol, snapshot in snapshots.items():
    day_change = snapshot.daily_bar.close - snapshot.prev_daily_bar.close
    day_change_pct = (day_change / snapshot.prev_daily_bar.close) * 100
    print(f"{symbol}: ${snapshot.daily_bar.close} ({day_change_pct:+.2f}%)")

News Data

Access financial news articles with symbol filtering and content options.

def get_news(
    symbol: Union[str, List[str]] = None,
    start: str = None,
    end: str = None,
    limit: int = 10,
    sort: Sort = Sort.Desc,
    include_content: bool = False,
    exclude_contentless: bool = False
) -> NewsListV2:
    """Get news articles."""

def get_news_iter(
    symbol: Union[str, List[str]] = None,
    start: str = None,
    end: str = None,
    limit: int = 10,
    sort: Sort = Sort.Desc,
    include_content: bool = False,
    exclude_contentless: bool = False,
    raw: bool = False
) -> Iterator[Union[NewsV2, dict]]:
    """Get news articles as an iterator."""

Usage Examples:

# Get recent news for a symbol
news = api.get_news('AAPL', limit=10)
for article in news:
    print(f"{article.created_at}: {article.headline}")
    print(f"Summary: {article.summary}")
    print(f"URL: {article.url}")
    print("---")

# Get market-wide news
general_news = api.get_news(limit=20, sort='desc')
for article in general_news:
    print(f"{article.headline} (Symbols: {', '.join(article.symbols)})")

Market Calendar and Clock

Access market schedule information and current market status.

def get_clock() -> Clock:
    """Get current market clock information."""

def get_calendar(start: str = None, end: str = None) -> List[Calendar]:
    """Get market calendar information."""

Usage Examples:

# Check market status
clock = api.get_clock()
print(f"Market is {'open' if clock.is_open else 'closed'}")
print(f"Next open: {clock.next_open}")
print(f"Next close: {clock.next_close}")

# Get market schedule
calendar = api.get_calendar('2023-01-01', '2023-01-31')
for day in calendar:
    print(f"{day.date}: Open={day.open}, Close={day.close}")

Types

class BarV2:
    @property
    def symbol(self) -> str: ...
    @property
    def timestamp(self) -> pd.Timestamp: ...
    @property
    def open(self) -> float: ...
    @property
    def high(self) -> float: ...
    @property
    def low(self) -> float: ...
    @property
    def close(self) -> float: ...
    @property
    def volume(self) -> int: ...
    @property
    def trade_count(self) -> int: ...
    @property
    def vwap(self) -> float: ...

class BarsV2:
    @property
    def df(self) -> pd.DataFrame: ...
    def __iter__(self): ...
    def __len__(self) -> int: ...

class TradesV2:
    @property
    def df(self) -> pd.DataFrame: ...
    def __iter__(self): ...
    def __len__(self) -> int: ...

class QuotesV2:
    @property
    def df(self) -> pd.DataFrame: ...
    def __iter__(self): ...
    def __len__(self) -> int: ...

class NewsListV2:
    def __iter__(self): ...
    def __len__(self) -> int: ...

class LatestBarsV2:
    def items(self): ...
    def __getitem__(self, symbol: str) -> BarV2: ...

class LatestTradesV2:
    def items(self): ...
    def __getitem__(self, symbol: str) -> TradeV2: ...

class LatestQuotesV2:
    def items(self): ...
    def __getitem__(self, symbol: str) -> QuoteV2: ...

class SnapshotsV2:
    def items(self): ...
    def __getitem__(self, symbol: str) -> SnapshotV2: ...

class TradeV2:
    @property
    def symbol(self) -> str: ...
    @property
    def timestamp(self) -> pd.Timestamp: ...
    @property
    def price(self) -> float: ...
    @property
    def size(self) -> int: ...
    @property
    def exchange(self) -> str: ...
    @property
    def conditions(self) -> List[str]: ...

class QuoteV2:
    @property
    def symbol(self) -> str: ...
    @property
    def timestamp(self) -> pd.Timestamp: ...
    @property
    def bid_price(self) -> float: ...
    @property
    def bid_size(self) -> int: ...
    @property
    def ask_price(self) -> float: ...
    @property
    def ask_size(self) -> int: ...
    @property
    def bid_exchange(self) -> str: ...
    @property
    def ask_exchange(self) -> str: ...

class SnapshotV2:
    @property
    def symbol(self) -> str: ...
    @property
    def latest_trade(self) -> TradeV2: ...
    @property
    def latest_quote(self) -> QuoteV2: ...
    @property
    def minute_bar(self) -> BarV2: ...
    @property
    def daily_bar(self) -> BarV2: ...
    @property
    def prev_daily_bar(self) -> BarV2: ...

class NewsV2:
    @property
    def id(self) -> str: ...
    @property
    def headline(self) -> str: ...
    @property
    def summary(self) -> str: ...
    @property
    def content(self) -> str: ...
    @property
    def created_at(self) -> pd.Timestamp: ...
    @property
    def updated_at(self) -> pd.Timestamp: ...
    @property
    def author(self) -> str: ...
    @property
    def symbols(self) -> List[str]: ...
    @property
    def url(self) -> str: ...

class Clock:
    @property
    def timestamp(self) -> pd.Timestamp: ...
    @property
    def is_open(self) -> bool: ...
    @property
    def next_open(self) -> pd.Timestamp: ...
    @property
    def next_close(self) -> pd.Timestamp: ...

class Calendar:
    @property
    def date(self) -> str: ...
    @property
    def open(self) -> str: ...
    @property
    def close(self) -> str: ...

Install with Tessl CLI

npx tessl i tessl/pypi-alpaca-trade-api

docs

async-operations.md

cryptocurrency.md

index.md

market-data.md

streaming.md

trading-operations.md

tile.json