import logging import os from typing import Iterator, List, Optional, Union import requests from requests.exceptions import HTTPError import time from enum import Enum from alpaca_trade_api import __version__ from .common import ( get_base_url, get_data_url, get_credentials, get_api_version, URL, FLOAT, ) from .entity import ( Bar, Entity, Account, AccountConfigurations, AccountActivity, Asset, Order, Position, Clock, Calendar, Trade, Quote, Watchlist, PortfolioHistory ) from .entity_v2 import ( BarV2, BarsV2, LatestBarsV2, LatestQuotesV2, LatestTradesV2, SnapshotV2, SnapshotsV2, TradesV2, TradeV2, QuotesV2, QuoteV2, NewsV2, NewsListV2 ) logger = logging.getLogger(__name__) Positions = List[Position] Orders = List[Order] Assets = List[Asset] AccountActivities = List[AccountActivity] Calendars = List[Calendar] Watchlists = List[Watchlist] TradeIterator = Iterator[Union[Trade, dict]] QuoteIterator = Iterator[Union[Quote, dict]] BarIterator = Iterator[Union[Bar, dict]] NewsIterator = Iterator[Union[NewsV2, dict]] DATA_V2_MAX_LIMIT = 10000 # max items per api call NEWS_MAX_LIMIT = 50 # max items per api call class RetryException(Exception): pass class APIError(Exception): """ Represent API related error. error.status_code will have http status code. """ def __init__(self, error, http_error=None): print("error is: {}".format(error)) try: super().__init__(error['message']) except: print ("APIError init failed with: {}".format(error)) self._error = error self._http_error = http_error @property def code(self): return self._error['code'] @property def status_code(self): http_error = self._http_error if http_error is not None and hasattr(http_error, 'response'): return http_error.response.status_code @property def request(self): if self._http_error is not None: return self._http_error.request @property def response(self): if self._http_error is not None: return self._http_error.response class TimeFrameUnit(Enum): Minute = "Min" Hour = "Hour" Day = "Day" Week = "Week" Month = "Month" class TimeFrame: def __init__(self, amount: int, unit: TimeFrameUnit): self.validate(amount, unit) self.__amount = amount self.__unit = unit @property def amount(self): return self.__amount @amount.setter def amount(self, value: int): self.validate(value, self.__unit) self.__amount = value @property def unit(self) -> TimeFrameUnit: return self.__unit @unit.setter def unit(self, value: TimeFrameUnit): self.validate(self.__amount, value) self.__unit = value # using "value" field for backwards compatibility @property def value(self): return f"{self.__amount}{self.__unit.value}" def __str__(self): return self.value @staticmethod def validate(amount: int, unit: TimeFrameUnit): if amount <= 0: raise ValueError("Amount must be a positive integer value.") if unit == TimeFrameUnit.Minute and amount > 59: raise ValueError("Second or Minute units can only be " + "used with amounts between 1-59.") if unit == TimeFrameUnit.Hour and amount > 23: raise ValueError("Hour units can only be used with amounts 1-23") if unit in (TimeFrameUnit.Day, TimeFrameUnit.Week) and amount != 1: raise ValueError( "Day and Week units can only be used with amount 1") if unit == TimeFrameUnit.Month and amount not in (1, 2, 3, 6, 12): raise ValueError( "Month units can only be used with amount 1, 2, 3, 6 and 12") # These are kept for backwards compatibility TimeFrame.Minute = TimeFrame(1, TimeFrameUnit.Minute) TimeFrame.Hour = TimeFrame(1, TimeFrameUnit.Hour) TimeFrame.Day = TimeFrame(1, TimeFrameUnit.Day) TimeFrame.Week = TimeFrame(1, TimeFrameUnit.Week) TimeFrame.Month = TimeFrame(1, TimeFrameUnit.Month) class Sort(Enum): Asc = "asc" Desc = "desc" def __str__(self): return self.value class REST(object): def __init__(self, key_id: str = None, secret_key: str = None, base_url: URL = None, api_version: str = None, oauth=None, raw_data: bool = False ): """ :param raw_data: should we return api response raw or wrap it with Entity objects. """ self._key_id, self._secret_key, self._oauth = get_credentials( key_id, secret_key, oauth) self._base_url: URL = URL(base_url or get_base_url()) self._api_version = get_api_version(api_version) self._session = requests.Session() self._use_raw_data = raw_data self._retry = int(os.environ.get('APCA_RETRY_MAX', 3)) self._retry_wait = int(os.environ.get('APCA_RETRY_WAIT', 3)) self._retry_codes = [int(o) for o in os.environ.get( 'APCA_RETRY_CODES', '429,504').split(',')] def _request(self, method, path, data=None, base_url: URL = None, api_version: str = None): base_url = base_url or self._base_url version = api_version if api_version else self._api_version url: URL = URL(base_url + '/' + version + path) headers = {} if self._oauth: headers['Authorization'] = 'Bearer ' + self._oauth else: headers['APCA-API-KEY-ID'] = self._key_id headers['APCA-API-SECRET-KEY'] = self._secret_key headers['User-Agent'] = 'APCA-TRADE-SDK-PY/' + __version__ opts = { 'headers': headers, # Since we allow users to set endpoint URL via env var, # human error to put non-SSL endpoint could exploit # uncanny issues in non-GET request redirecting http->https. # It's better to fail early if the URL isn't right. 'allow_redirects': False, } if method.upper() in ['GET', 'DELETE']: opts['params'] = data else: opts['json'] = data retry = self._retry if retry < 0: retry = 0 while retry >= 0: try: return self._one_request(method, url, opts, retry) except RetryException: retry_wait = self._retry_wait logger.warning( 'sleep {} seconds and retrying {} ' '{} more time(s)...'.format( retry_wait, url, retry)) time.sleep(retry_wait) retry -= 1 continue def _one_request(self, method: str, url: URL, opts: dict, retry: int): """ Perform one request, possibly raising RetryException in the case the response is 429. Otherwise, if error text contain "code" string, then it decodes to json object and returns APIError. Returns the body json in the 200 status. """ retry_codes = self._retry_codes resp = self._session.request(method, url, **opts) try: resp.raise_for_status() except HTTPError as http_error: # retry if we hit Rate Limit if resp.status_code in retry_codes and retry > 0: raise RetryException() if 'code' in resp.text: error = resp.json() if 'code' in error: raise APIError(error, http_error) else: raise if resp.text != '': return resp.json() return None def get(self, path, data=None): return self._request('GET', path, data) def post(self, path, data=None): return self._request('POST', path, data) def put(self, path, data=None): return self._request('PUT', path, data) def patch(self, path, data=None): return self._request('PATCH', path, data) def delete(self, path, data=None): return self._request('DELETE', path, data) def data_get(self, path, data=None, api_version='v1'): base_url: URL = get_data_url() return self._request( 'GET', path, data, base_url=base_url, api_version=api_version, ) def get_account(self) -> Account: """Get the account""" resp = self.get('/account') return self.response_wrapper(resp, Account) def get_account_configurations(self) -> AccountConfigurations: """Get account configs""" resp = self.get('/account/configurations') return self.response_wrapper(resp, AccountConfigurations) def update_account_configurations( self, no_shorting: bool = None, dtbp_check: str = None, trade_confirm_email: str = None, suspend_trade: bool = None) -> AccountConfigurations: """ alpaca.markets/docs/api-documentation/api-v2/account-configuration/ Update account configs :param dtbp_check: both, entry, or exit :param trade_confirm_email: all or none """ params = {} if no_shorting is not None: params['no_shorting'] = no_shorting if dtbp_check is not None: params['dtbp_check'] = dtbp_check if trade_confirm_email is not None: params['trade_confirm_email'] = trade_confirm_email if suspend_trade is not None: params['suspend_trade'] = suspend_trade resp = self.patch('/account/configurations', params) return self.response_wrapper(resp, AccountConfigurations) def list_orders(self, status: str = None, limit: int = None, after: str = None, until: str = None, direction: str = None, params=None, nested: bool = None, symbols: List[str] = None ) -> Orders: """ Get a list of orders https://docs.alpaca.markets/web-api/orders/#get-a-list-of-orders :param status: open, closed or all. Defaults to open. :param limit: Defaults to 50 and max is 500 :param after: timestamp :param until: timestamp :param direction: asc or desc. :param params: refer to documentation :param nested: should the data be nested like json :param symbols: list of str (symbols) """ if params is None: params = dict() if limit is not None: params['limit'] = limit if after is not None: params['after'] = after if until is not None: params['until'] = until if direction is not None: params['direction'] = direction if status is not None: params['status'] = status if nested is not None: params['nested'] = nested if symbols is not None: params['symbols'] = ",".join(symbols) url = '/orders' resp = self.get(url, params) if self._use_raw_data: return resp else: return [self.response_wrapper(o, Order) for o in resp] def submit_order(self, symbol: str, qty: float = None, side: str = "buy", type: str = "market", time_in_force: str = "day", limit_price: str = None, stop_price: str = None, client_order_id: str = None, extended_hours: bool = None, order_class: str = None, take_profit: dict = None, stop_loss: dict = None, trail_price: str = None, trail_percent: str = None, notional: float = None): """ :param symbol: symbol or asset ID :param qty: float. Mutually exclusive with "notional". :param side: buy or sell :param type: market, limit, stop, stop_limit or trailing_stop :param time_in_force: day, gtc, opg, cls, ioc, fok :param limit_price: str of float :param stop_price: str of float :param client_order_id: :param extended_hours: bool. If true, order will be eligible to execute in premarket/afterhours. :param order_class: simple, bracket, oco or oto :param take_profit: dict with field "limit_price" e.g {"limit_price": "298.95"} :param stop_loss: dict with fields "stop_price" and "limit_price" e.g {"stop_price": "297.95", "limit_price": "298.95"} :param trail_price: str of float :param trail_percent: str of float :param notional: float. Mutually exclusive with "qty". """ """Request a new order""" params = { 'symbol': symbol, 'side': side, 'type': type, 'time_in_force': time_in_force } if qty is not None: params['qty'] = qty if notional is not None: params['notional'] = notional if limit_price is not None: params['limit_price'] = FLOAT(limit_price) if stop_price is not None: params['stop_price'] = FLOAT(stop_price) if client_order_id is not None: params['client_order_id'] = client_order_id if extended_hours is not None: params['extended_hours'] = extended_hours if order_class is not None: params['order_class'] = order_class if take_profit is not None: if 'limit_price' in take_profit: take_profit['limit_price'] = FLOAT(take_profit['limit_price']) params['take_profit'] = take_profit if stop_loss is not None: if 'limit_price' in stop_loss: stop_loss['limit_price'] = FLOAT(stop_loss['limit_price']) if 'stop_price' in stop_loss: stop_loss['stop_price'] = FLOAT(stop_loss['stop_price']) params['stop_loss'] = stop_loss if trail_price is not None: params['trail_price'] = trail_price if trail_percent is not None: params['trail_percent'] = trail_percent resp = self.post('/orders', params) return self.response_wrapper(resp, Order) def get_order_by_client_order_id(self, client_order_id: str) -> Order: """Get an order by client order id""" params = { 'client_order_id': client_order_id, } resp = self.get('/orders:by_client_order_id', params) return self.response_wrapper(resp, Order) def get_order(self, order_id: str, nested: bool = None) -> Order: """Get an order""" params = {} if nested is not None: params['nested'] = nested resp = self.get('/orders/{}'.format(order_id), params) return self.response_wrapper(resp, Order) def replace_order( self, order_id: str, qty: str = None, limit_price: str = None, stop_price: str = None, trail: str = None, time_in_force: str = None, client_order_id: str = None, ) -> Order: """ :param order_id: :param qty: str of int :param limit_price: str of float :param stop_price: str of float :param trail: str of float, represents trailing_price or trailing_percent. determined by the original order. :param time_in_force: day, gtc, opg, cls, ioc, fok note: you cannot replace type of order. so, it was trailing_stop(e.g) it will remain trailing_stop. """ params = {} if qty is not None: params['qty'] = qty if limit_price is not None: params['limit_price'] = FLOAT(limit_price) if stop_price is not None: params['stop_price'] = FLOAT(stop_price) if trail is not None: params['trail'] = FLOAT(trail) if time_in_force is not None: params['time_in_force'] = time_in_force if client_order_id is not None: params['client_order_id'] = client_order_id resp = self.patch('/orders/{}'.format(order_id), params) return self.response_wrapper(resp, Order) def cancel_order(self, order_id: str) -> None: """Cancel an order""" self.delete('/orders/{}'.format(order_id)) def cancel_all_orders(self) -> None: """Cancel all open orders""" self.delete('/orders') def list_positions(self) -> Positions: """Get a list of open positions""" resp = self.get('/positions') if self._use_raw_data: return resp else: return [self.response_wrapper(p, Position) for p in resp] def get_position(self, symbol: str) -> Position: """Get an open position""" resp = self.get('/positions/{}'.format(symbol)) return self.response_wrapper(resp, Position) def close_position(self, symbol: str, *, qty: float = None, # percentage: float = None # currently unsupported api ) -> Position: """Liquidates the position for the given symbol at market price""" # if qty and percentage: # raise Exception("Can't close position with qty and pecentage") # elif qty: # data = {'qty': qty} # elif percentage: # data = {'percentage': percentage} # else: # data = {} if qty: data = {'qty': qty} else: data = {} resp = self.delete('/positions/{}'.format(symbol), data=data) return self.response_wrapper(resp, Position) def close_all_positions(self) -> Positions: """Liquidates all open positions at market price""" resp = self.delete('/positions') if self._use_raw_data: return resp else: return [self.response_wrapper(o, Position) for o in resp] def list_assets(self, status=None, asset_class=None) -> Assets: """Get a list of assets""" params = { 'status': status, 'asset_class': asset_class, } resp = self.get('/assets', params) if self._use_raw_data: return resp else: return [self.response_wrapper(o, Asset) for o in resp] def get_asset(self, symbol: str) -> Asset: """Get an asset""" resp = self.get('/assets/{}'.format(symbol)) return self.response_wrapper(resp, Asset) def _data_get(self, endpoint: str, symbol_or_symbols: Union[str, List[str]], api_version: str = 'v2', endpoint_base: str = 'stocks', resp_grouped_by_symbol: Optional[bool] = None, page_limit: int = DATA_V2_MAX_LIMIT, **kwargs): page_token = None total_items = 0 limit = kwargs.get('limit') if resp_grouped_by_symbol is None: resp_grouped_by_symbol = not isinstance(symbol_or_symbols, str) while True: actual_limit = None if limit: actual_limit = min(int(limit) - total_items, page_limit) if actual_limit < 1: break data = kwargs data['limit'] = actual_limit data['page_token'] = page_token path = f'/{endpoint_base}' if isinstance(symbol_or_symbols, str) and symbol_or_symbols: path += f'/{symbol_or_symbols}' else: data['symbols'] = ','.join(symbol_or_symbols) if endpoint: path += f'/{endpoint}' resp = self.data_get(path, data=data, api_version=api_version) if not resp_grouped_by_symbol: k = endpoint or endpoint_base for item in resp.get(k, []) or []: yield item total_items += 1 else: by_symbol = resp.get(endpoint, {}) or {} for sym, items in sorted(by_symbol.items()): for item in items or []: item['S'] = sym yield item total_items += 1 page_token = resp.get('next_page_token') if not page_token: break def get_trades_iter(self, symbol: Union[str, List[str]], start: Optional[str] = None, end: Optional[str] = None, limit: int = None, raw=False) -> TradeIterator: trades = self._data_get('trades', symbol, start=start, end=end, limit=limit) for trade in trades: if raw: yield trade else: yield self.response_wrapper(trade, Trade) def get_trades(self, symbol: Union[str, List[str]], start: Optional[str] = None, end: Optional[str] = None, limit: int = None, ) -> TradesV2: trades = list(self.get_trades_iter(symbol, start, end, limit, raw=True)) return TradesV2(trades) def get_quotes_iter(self, symbol: Union[str, List[str]], start: Optional[str] = None, end: Optional[str] = None, limit: int = None, raw=False) -> QuoteIterator: quotes = self._data_get('quotes', symbol, start=start, end=end, limit=limit) for quote in quotes: if raw: yield quote else: yield self.response_wrapper(quote, Quote) def get_quotes(self, symbol: Union[str, List[str]], start: Optional[str] = None, end: Optional[str] = None, limit: int = None, ) -> QuotesV2: quotes = list(self.get_quotes_iter(symbol, start, end, limit, raw=True)) return QuotesV2(quotes) def get_bars_iter(self, symbol: Union[str, List[str]], timeframe: TimeFrame, start: Optional[str] = None, end: Optional[str] = None, adjustment: str = 'raw', limit: int = None, raw=False) -> BarIterator: bars = self._data_get('bars', symbol, timeframe=timeframe, adjustment=adjustment, start=start, end=end, limit=limit) for bar in bars: if raw: yield bar else: yield self.response_wrapper(bar, Bar) def get_bars(self, symbol: Union[str, List[str]], timeframe: TimeFrame, start: Optional[str] = None, end: Optional[str] = None, adjustment: str = 'raw', limit: int = None, ) -> BarsV2: bars = list(self.get_bars_iter(symbol, timeframe, start, end, adjustment, limit, raw=True)) return BarsV2(bars) def get_latest_bar(self, symbol: str) -> BarV2: resp = self.data_get( '/stocks/{}/bars/latest'.format(symbol), api_version='v2') return self.response_wrapper(resp['bar'], BarV2) def get_latest_bars(self, symbols: List[str]) -> LatestBarsV2: resp = self.data_get( f'/stocks/bars/latest?symbols={_join_with_commas(symbols)}', api_version='v2') return self.response_wrapper(resp['bars'], LatestBarsV2) def get_latest_trade(self, symbol: str) -> TradeV2: resp = self.data_get( '/stocks/{}/trades/latest'.format(symbol), api_version='v2') return self.response_wrapper(resp['trade'], TradeV2) def get_latest_trades(self, symbols: List[str]) -> LatestTradesV2: resp = self.data_get( f'/stocks/trades/latest?symbols={_join_with_commas(symbols)}', api_version='v2') return self.response_wrapper(resp['trades'], LatestTradesV2) def get_latest_quote(self, symbol: str) -> QuoteV2: resp = self.data_get( '/stocks/{}/quotes/latest'.format(symbol), api_version='v2') return self.response_wrapper(resp['quote'], QuoteV2) def get_latest_quotes(self, symbols: List[str]) -> LatestQuotesV2: resp = self.data_get( f'/stocks/quotes/latest?symbols={_join_with_commas(symbols)}', api_version='v2') return self.response_wrapper(resp['quotes'], LatestQuotesV2) def get_snapshot(self, symbol: str) -> SnapshotV2: resp = self.data_get('/stocks/{}/snapshot'.format(symbol), api_version='v2') return self.response_wrapper(resp, SnapshotV2) def get_snapshots(self, symbols: List[str]) -> SnapshotsV2: resp = self.data_get( '/stocks/snapshots?symbols={}'.format(_join_with_commas(symbols)), api_version='v2') return self.response_wrapper(resp, SnapshotsV2) def get_crypto_trades_iter(self, symbol: str, start: Optional[str] = None, end: Optional[str] = None, limit: int = None, exchanges: Optional[List[str]] = None, raw=False) -> TradeIterator: trades = self._data_get('trades', symbol, api_version='v1beta1', endpoint_base='crypto', start=start, end=end, limit=limit, exchanges=exchanges) for trade in trades: if raw: yield trade else: yield self.response_wrapper(trade, Trade) def get_crypto_trades(self, symbol: str, start: Optional[str] = None, end: Optional[str] = None, limit: int = None, exchanges: Optional[List[str]] = None, ) -> TradesV2: return TradesV2(list(self.get_crypto_trades_iter( symbol, start, end, limit, exchanges, raw=True))) def get_crypto_quotes_iter(self, symbol: str, start: Optional[str] = None, end: Optional[str] = None, limit: int = None, exchanges: Optional[List[str]] = None, raw=False) -> QuoteIterator: quotes = self._data_get('quotes', symbol, api_version='v1beta1', endpoint_base='crypto', start=start, end=end, limit=limit, exchanges=exchanges) for quote in quotes: if raw: yield quote else: yield self.response_wrapper(quote, Quote) def get_crypto_quotes(self, symbol: str, start: Optional[str] = None, end: Optional[str] = None, limit: int = None, exchanges: Optional[List[str]] = None, ) -> QuotesV2: return QuotesV2(list(self.get_crypto_quotes_iter( symbol, start, end, limit, exchanges, raw=True))) def get_crypto_bars_iter(self, symbol: Union[str, List[str]], timeframe: TimeFrame, start: Optional[str] = None, end: Optional[str] = None, limit: int = None, exchanges: Optional[List[str]] = None, raw=False) -> BarIterator: bars = self._data_get('bars', symbol, api_version='v1beta1', endpoint_base='crypto', timeframe=timeframe, start=start, end=end, limit=limit, exchanges=exchanges) for bar in bars: if raw: yield bar else: yield self.response_wrapper(bar, Bar) def get_crypto_bars(self, symbol: Union[str, List[str]], timeframe: TimeFrame, start: Optional[str] = None, end: Optional[str] = None, limit: int = None, exchanges: Optional[List[str]] = None, ) -> BarsV2: return BarsV2(list(self.get_crypto_bars_iter( symbol, timeframe, start, end, limit, exchanges, raw=True))) def get_latest_crypto_bar(self, symbol: str, exchange: str) -> BarV2: resp = self.data_get( '/crypto/{}/bars/latest'.format(symbol), data={'exchange': exchange}, api_version='v1beta1') return self.response_wrapper(resp['bar'], BarV2) def get_latest_crypto_bars(self, symbols: List[str], exchange: str) -> LatestBarsV2: resp = self.data_get( '/crypto/bars/latest', data={'exchange': exchange, 'symbols': _join_with_commas(symbols)}, api_version='v1beta1') return self.response_wrapper(resp['bars'], LatestBarsV2) def get_latest_crypto_trade(self, symbol: str, exchange: str) -> TradeV2: resp = self.data_get( '/crypto/{}/trades/latest'.format(symbol), data={'exchange': exchange}, api_version='v1beta1') return self.response_wrapper(resp['trade'], TradeV2) def get_latest_crypto_trades(self, symbols: List[str], exchange: str) -> LatestTradesV2: resp = self.data_get( '/crypto/trades/latest', data={'exchange': exchange, 'symbols': _join_with_commas(symbols)}, api_version='v1beta1') return self.response_wrapper(resp['trades'], LatestTradesV2) def get_latest_crypto_quote(self, symbol: str, exchange: str) -> QuoteV2: resp = self.data_get( '/crypto/{}/quotes/latest'.format(symbol), data={'exchange': exchange}, api_version='v1beta1') return self.response_wrapper(resp['quote'], QuoteV2) def get_latest_crypto_quotes(self, symbols: List[str], exchange: str) -> LatestQuotesV2: resp = self.data_get( '/crypto/quotes/latest', data={'exchange': exchange, 'symbols': _join_with_commas(symbols)}, api_version='v1beta1') return self.response_wrapper(resp['quotes'], LatestQuotesV2) def get_latest_crypto_xbbo(self, symbol: str, exchanges: Optional[List[str]] = None, ) -> QuoteV2: params = {} if exchanges: params['exchanges'] = _join_with_commas(exchanges) resp = self.data_get( '/crypto/{}/xbbo/latest'.format(symbol), data=params, api_version='v1beta1') return self.response_wrapper(resp['xbbo'], QuoteV2) def get_latest_crypto_xbbos(self, symbols: List[str], exchanges: Optional[List[str]] = None, ) -> QuoteV2: params = {'symbols': _join_with_commas(symbols)} if exchanges: params['exchanges'] = _join_with_commas(exchanges) resp = self.data_get( '/crypto/xbbos/latest', data=params, api_version='v1beta1') return self.response_wrapper(resp['xbbos'], QuotesV2) def get_crypto_snapshot(self, symbol: str, exchange: str) -> SnapshotV2: resp = self.data_get('/crypto/{}/snapshot'.format(symbol), data={'exchange': exchange}, api_version='v1beta1') return self.response_wrapper(resp, SnapshotV2) def get_latest_crypto_snapshots(self, symbols: List[str], exchange: str) -> SnapshotsV2: resp = self.data_get( '/crypto/snapshots', data={'exchange': exchange, 'symbols': _join_with_commas(symbols)}, api_version='v1beta1') return self.response_wrapper(resp['snapshots'], SnapshotsV2) def get_news_iter(self, symbol: Optional[Union[str, List[str]]] = None, start: Optional[str] = None, end: Optional[str] = None, limit: int = 10, sort: Sort = Sort.Desc, include_content: bool = False, exclude_contentless: bool = False, raw=False) -> NewsIterator: symbol = symbol or [] # Avoid passing symbol as path param if isinstance(symbol, str): symbol = [symbol] news = self._data_get('', symbol, api_version='v1beta1', endpoint_base='news', start=start, end=end, limit=limit, sort=sort, include_content=include_content, exclude_contentless=exclude_contentless, resp_grouped_by_symbol=False, page_limit=NEWS_MAX_LIMIT) for n in news: if raw: yield n else: yield self.response_wrapper(n, NewsV2) def get_news(self, symbol: Optional[Union[str, List[str]]] = None, start: Optional[str] = None, end: Optional[str] = None, limit: int = 10, sort: Sort = Sort.Desc, include_content: bool = False, exclude_contentless: bool = False, ) -> NewsListV2: news = list(self.get_news_iter(symbol=symbol, start=start, end=end, limit=limit, sort=sort, include_content=include_content, exclude_contentless=exclude_contentless, raw=True)) return NewsListV2(news) def get_clock(self) -> Clock: resp = self.get('/clock') return self.response_wrapper(resp, Clock) def get_activities( self, activity_types: str = None, until: str = None, after: str = None, direction: str = None, date: str = None, page_size: int = None, page_token: str = None ) -> AccountActivities: """ go to alpaca.markets/docs/api-documentation/api-v2/account-activities/ :param activity_types: go to documnetation to see available types :param until: isoformat timestamp :param after: isoformat timestamp :param direction: asc or sesc. default is desc :param date: str. can't be sued with until/after :param page_size: :param page_token: :return: """ url = '/account/activities' params = {} if isinstance(activity_types, list): params['activity_types'] = ','.join(activity_types) elif activity_types is not None: url += '/{}'.format(activity_types) if after is not None: params['after'] = after if until is not None: params['until'] = until if direction is not None: params['direction'] = direction if date is not None: params['date'] = date if page_size is not None: params['page_size'] = page_size if page_token is not None: params['page_token'] = page_token resp = self.get(url, data=params) if self._use_raw_data: return resp else: return [self.response_wrapper(o, AccountActivity) for o in resp] def get_calendar(self, start: str = None, end: str = None) -> Calendars: """ :param start: isoformat date string eg '2006-01-02T15:04:05Z' or '2006-01-02' :param end: isoformat date string """ params = {} if start is not None: params['start'] = start if end is not None: params['end'] = end resp = self.get('/calendar', data=params) if self._use_raw_data: return resp else: return [self.response_wrapper(o, Calendar) for o in resp] def get_watchlists(self) -> Watchlists: """Get the list of watchlists registered under the account""" resp = self.get('/watchlists') if self._use_raw_data: return resp else: return [self.response_wrapper(o, Watchlist) for o in resp] def get_watchlist(self, watchlist_id: str) -> Watchlist: """Get a watchlist identified by the ID""" resp = self.get('/watchlists/{}'.format((watchlist_id))) return self.response_wrapper(resp, Watchlist) def get_watchlist_by_name(self, watchlist_name: str) -> Watchlist: """Get a watchlist identified by its name""" params = { 'name': watchlist_name, } resp = self.get('/watchlists:by_name', data=params) return self.response_wrapper(resp, Watchlist) def create_watchlist(self, watchlist_name: str, symbols=None) -> Watchlist: """Create a new watchlist with an optional initial set of assets""" params = { 'name': watchlist_name, } if symbols is not None: params['symbols'] = symbols resp = self.post('/watchlists', data=params) return self.response_wrapper(resp, Watchlist) def add_to_watchlist(self, watchlist_id: str, symbol: str) -> Watchlist: """Add an asset to the watchlist""" resp = self.post( '/watchlists/{}'.format(watchlist_id), data=dict(symbol=symbol) ) return self.response_wrapper(resp, Watchlist) def update_watchlist(self, watchlist_id: str, name: str = None, symbols=None) -> Watchlist: """Update a watchlist's name and/or asset list""" params = {} if name is not None: params['name'] = name if symbols is not None: params['symbols'] = symbols resp = self.put('/watchlists/{}'.format(watchlist_id), data=params) return self.response_wrapper(resp, Watchlist) def delete_watchlist(self, watchlist_id: str) -> None: """Delete a watchlist identified by the ID permanently""" self.delete('/watchlists/{}'.format(watchlist_id)) def delete_from_watchlist(self, watchlist_id: str, symbol: str) -> None: """Remove an asset from the watchlist's asset list""" self.delete('/watchlists/{}/{}'.format(watchlist_id, symbol)) def get_portfolio_history(self, date_start: str = None, date_end: str = None, period: str = None, timeframe=None, extended_hours: bool = None) -> PortfolioHistory: """ alpaca.markets/docs/api-documentation/api-v2/portfolio-history/ :param date_start: YYYY-MM-DD :param date_end: YYYY-MM-DD :param period: The duration of the data in + such as 1D, where can be D for day, W for week, M for month and A for year. Defaults to 1M. :param timeframe: The resolution of time window. 1Min, 5Min, 15Min, 1H, or 1D :param extended_hours: bool. If true, include extended hours in the result. """ params = {} if date_start is not None: params['date_start'] = date_start if date_end is not None: params['date_end'] = date_end if period is not None: params['period'] = period if timeframe is not None: params['timeframe'] = timeframe if extended_hours is not None: params['extended_hours'] = extended_hours resp = self.get('/account/portfolio/history', data=params) return self.response_wrapper(resp, PortfolioHistory) def __enter__(self): return self def close(self): self._session.close() def __exit__(self, exc_type, exc_val, exc_tb): self.close() def response_wrapper(self, obj, entity: Entity): """ To allow the user to get raw response from the api, we wrap all functions with this method, checking if the user has set raw_data bool. if they didn't, we wrap the response with an Entity object. :param obj: response from server :param entity: derivative object of Entity :return: """ if self._use_raw_data: return obj else: return entity(obj) def _join_with_commas(lst: List[str]) -> str: if isinstance(lst, str): raise ValueError('expected list, str found') return ','.join(lst)