import datetime
from functools import partial
from logging import Logger
from typing import Iterator, Union
from lime_trader.api.authenticated_api_client import AuthenticatedApiClient
from lime_trader.models.accounts import AccountTransaction
from lime_trader.models.market import (Quote, QuoteHistory, Period, Security, SecuritiesPage, Trade, TradesPage,
CurrentSchedule, OptionSeries, OptionChain)
from lime_trader.models.page import Page, PageRequest
from lime_trader.constants.urls import (MARKET_DATA_GET_CURRENT_QUOTE, MARKET_DATA_GET_QUOTES,
MARKET_DATA_GET_TRADING_SCHEDULE,
MARKET_DATA_LOOKUP_SECURITIES, MARKET_DATA_GET_TIME_AND_SALES,
MARKET_DATA_GET_QUOTES_HISTORY, MARKET_DATA_GET_OPTION_CHAIN,
MARKET_DATA_GET_OPTION_SERIES)
from lime_trader.utils.pagination import iterate_pages
[docs]
class MarketDataClient:
"""
Contains methods related to market data
"""
def __init__(self, api_client: AuthenticatedApiClient, logger: Logger):
"""
Args:
api_client: API client that will be used to execute all requests
logger: Logger used to submit client log messages
"""
self._api_client = api_client
self._logger = logger
[docs]
def get_current_quote(self, symbol: str) -> Quote:
"""
Retrieves current quote for the specified symbol
Args:
symbol: Security symbol
Returns:
Quote for the symbol
"""
self._logger.info(f"Getting current quote for symbol {symbol}")
return self._api_client.get(MARKET_DATA_GET_CURRENT_QUOTE, path_params={},
params={"symbol": symbol}, response_schema=Quote)
[docs]
def get_current_quotes(self, symbols: list[str]) -> list[Quote]:
"""
Retrieves current quotes for all specified symbols
Args:
symbols: List of security symbols
Returns:
List of quotes for specified symbols
"""
self._logger.info(f"Getting current quotes for symbols {symbols}")
return self._api_client.post(MARKET_DATA_GET_QUOTES, path_params={}, json=symbols, response_schema=list[Quote])
[docs]
def get_quotes_history(self, symbol: str, period: Period, from_date: datetime.datetime,
to_date: datetime.datetime) -> list[QuoteHistory]:
"""
Returns candle structures aggregated by specified period
Args:
symbol: The security symbol, stocks in Nasdaq CMS convention. Options in OCC
period: Aggregation period.
from_date: Period start
to_date: Period end
Returns:
Quote history between specified period
"""
self._logger.info(f"Getting quotes history for symbol={symbol}, period={period}, from_date={from_date},"
f"to_date={to_date}")
return self._api_client.get(MARKET_DATA_GET_QUOTES_HISTORY, path_params={},
params={"symbol": symbol,
"period": period.value,
"from": from_date,
"to": to_date
}, response_schema=list[QuoteHistory])
[docs]
def get_trading_schedule(self) -> CurrentSchedule:
"""
Returns trading session info depending on current date and time:
Returns:
Current trading schedule
"""
self._logger.info(f"Getting current trading schedule")
return self._api_client.get(MARKET_DATA_GET_TRADING_SCHEDULE, path_params={}, params={},
response_schema=CurrentSchedule)
[docs]
def lookup_securities(self, query: str, page: PageRequest) -> Page[Security]:
"""
Searches the securities reference by the specified criteria. The criteria can be a symbol,
part of a symbol or part of a company name.
Args:
query: Search criteria
page: Pagination parameters
Returns:
Page of security symbols which match criteria
"""
self._logger.info(f"Looking up securities with query={query}, page={page}")
response = self._api_client.get(MARKET_DATA_LOOKUP_SECURITIES, path_params={},
params={"query": query,
"limit": page.size,
"skip": page.get_offset()},
response_schema=SecuritiesPage)
return Page(data=response.securities, number=page.page, size=page.size, total_elements=response.count)
[docs]
def time_and_sales(self, symbol: str, from_date: datetime.datetime,
to_date: datetime.datetime, page: PageRequest) -> Page[Trade]:
"""
Retrieves the time and sales history, ordered by descending timestamp. Query parameters are:
Args:
symbol: The security symbol, stocks in Nasdaq CMS convention. Options are not supported
from_date: Period start
to_date: Period end
page: Pagination parameters
Returns:
Page of historical trades
"""
self._logger.info(f"Getting time and sales history for symbol={symbol}, from_date={from_date},"
f"to_date={to_date}, page={page}")
response = self._api_client.get(MARKET_DATA_GET_TIME_AND_SALES, path_params={},
params={"symbol": symbol,
"from": from_date,
"to": to_date,
"limit": page.size,
"skip": page.get_offset()},
response_schema=TradesPage)
return Page(data=response.trades, number=page.page, size=page.size, total_elements=response.count)
[docs]
def iterate_time_and_sales(self, symbol: str, from_date: datetime.datetime,
to_date: datetime.datetime, start_page: PageRequest,
) -> Iterator[Page[AccountTransaction]]:
"""
Returns iterator for the time and sales history, ordered by descending timestamp
Args:
symbol: The security symbol, stocks in Nasdaq CMS convention. Options are not supported
from_date: Period start
to_date: Period end
start_page: Start page. Iterator starts from this page
Returns:
Iterator of pages of historical trades
"""
for page in iterate_pages(start_page=start_page,
func=partial(self.time_and_sales, symbol=symbol,
from_date=from_date, to_date=to_date)):
yield page
[docs]
def get_option_series(self, symbol: str) -> list[OptionSeries]:
"""
Returns an array of option series by the specified underlying security's symbol.
Contains at least one element which is the default series.
Option Series is a specific set of calls or puts on the same underlying security,
with the same strike price and expiration date.
For the most part there is just one series name that matches the symbol but in some cases related to
corporate actions on the underlying security additional series are issued.
Args:
symbol: The security symbol
Returns:
Array of option series by the specified underlying security's symbol
"""
return self._api_client.get(MARKET_DATA_GET_OPTION_SERIES, path_params={"symbol": symbol}, params={},
response_schema=list[OptionSeries])
[docs]
def get_option_chain(self, symbol: str, expiration: datetime.date,
series: Union[str, None] = None) -> OptionChain:
"""
Returns option contracts for specified symbol, expiration date and series.
Args:
symbol: The security symbol.
expiration: Contract expiration date
series: By default the series is the same as the security symbol
Returns:
Option contracts for specified symbol, expiration date and series
"""
return self._api_client.get(MARKET_DATA_GET_OPTION_CHAIN, path_params={"symbol": symbol},
params={"expiration": expiration, "series": series},
response_schema=OptionChain)