# *****************************************************************************
#
# Copyright (c) 2020, the pyEX authors.
#
# This file is part of the pyEX library, distributed under the terms of
# the Apache License 2.0. The full license can be found in the LICENSE file.
#
from functools import wraps
import pandas as pd
from ..common import (
_get,
_getAsync,
_raiseIfNotStr,
_reindex,
_strOrDate,
_strToList,
_toDatetime,
json_normalize,
)
[docs]def iexTops(symbols=None, token="", version="stable", format="json"):
"""TOPS provides IEX’s aggregated best quoted bid and offer position in near real time for all securities on IEX’s displayed limit order book.
TOPS is ideal for developers needing both quote and trade data.
https://iexcloud.io/docs/api/#tops
Args:
symbol (str): Ticker to request
token (str): Access token
version (str): API version
format (str): return format, defaults to json
Returns:
dict: result
"""
symbols = _strToList(symbols)
if symbols:
return _get(
"tops?symbols=" + ",".join(symbols) + "%2b",
token=token,
version=version,
format=format,
)
return _get("tops", token=token, version=version, format=format)
[docs]@wraps(iexTops)
async def iexTopsAsync(symbols=None, token="", version="stable", format="json"):
symbols = _strToList(symbols)
if symbols:
return await _getAsync(
"tops?symbols=" + ",".join(symbols) + "%2b",
token=token,
version=version,
format=format,
)
return await _getAsync("tops", token=token, version=version, format=format)
[docs]@wraps(iexTops)
def iexTopsDF(*args, **kwargs):
return _reindex(_toDatetime(json_normalize(iexTops(*args, **kwargs))), "symbol")
[docs]def iexLast(symbols=None, token="", version="stable", format="json"):
"""Last provides trade data for executions on IEX. It is a near real time, intraday API that provides IEX last sale price, size and time.
Last is ideal for developers that need a lightweight stock quote.
https://iexcloud.io/docs/api/#last
Args:
symbol (str): Ticker to request
token (str): Access token
version (str): API version
format (str): return format, defaults to json
Returns:
dict: result
"""
symbols = _strToList(symbols)
if symbols:
return _get(
"tops/last?symbols=" + ",".join(symbols) + "%2b",
token=token,
version=version,
format=format,
)
return _get("tops/last", token=token, version=version, format=format)
[docs]@wraps(iexLast)
async def iexLastAsync(symbols=None, token="", version="stable", format="json"):
symbols = _strToList(symbols)
if symbols:
return await _getAsync(
"tops/last?symbols=" + ",".join(symbols) + "%2b",
token=token,
version=version,
format=format,
)
return await _getAsync("tops/last", token=token, version=version, format=format)
[docs]@wraps(iexLast)
def iexLastDF(*args, **kwargs):
return _reindex(_toDatetime(json_normalize(iexLast(*args, **kwargs))), "symbol")
[docs]def iexDeep(symbol=None, token="", version="stable", format="json"):
"""DEEP is used to receive real-time depth of book quotations direct from IEX.
The depth of book quotations received via DEEP provide an aggregated size of resting displayed orders at a price and side,
and do not indicate the size or number of individual orders at any price level.
Non-displayed orders and non-displayed portions of reserve orders are not represented in DEEP.
DEEP also provides last trade price and size information. Trades resulting from either displayed or non-displayed orders matching on IEX will be reported. Routed executions will not be reported.
https://iexcloud.io/docs/api/#deep
Args:
symbol (str): Ticker to request
token (str): Access token
version (str): API version
format (str): return format, defaults to json
Returns:
dict: result
"""
_raiseIfNotStr(symbol)
if symbol:
return _get(
"deep?symbols=" + symbol, token=token, version=version, format=format
)
return _get("deep", token=token, version=version, format=format)
[docs]@wraps(iexDeep)
async def iexDeepAsync(symbol=None, token="", version="stable", format="json"):
_raiseIfNotStr(symbol)
if symbol:
return await _getAsync(
"deep?symbols=" + symbol, token=token, version=version, format=format
)
return await _getAsync("deep", token=token, version=version, format=format)
[docs]@wraps(iexDeep)
def iexDeepDF(*args, **kwargs):
return _toDatetime(json_normalize(iexDeep(*args, **kwargs)))
[docs]def iexAuction(symbol=None, token="", version="stable", format="json"):
"""DEEP broadcasts an Auction Information Message every one second between the Lock-in Time and the auction match for Opening and Closing Auctions,
and during the Display Only Period for IPO, Halt, and Volatility Auctions. Only IEX listed securities are eligible for IEX Auctions.
https://iexcloud.io/docs/api/#deep-auction
Args:
symbol (str): Ticker to request
token (str): Access token
version (str): API version
format (str): return format, defaults to json
Returns:
dict: result
"""
_raiseIfNotStr(symbol)
if symbol:
return _get(
"deep/auction?symbols=" + symbol,
token=token,
version=version,
format=format,
)
return _get("deep/auction", token=token, version=version, format=format)
[docs]@wraps(iexAuction)
async def iexAuctionAsync(symbol=None, token="", version="stable", format="json"):
_raiseIfNotStr(symbol)
if symbol:
return await _getAsync(
"deep/auction?symbols=" + symbol,
token=token,
version=version,
format=format,
)
return await _getAsync("deep/auction", token=token, version=version, format=format)
[docs]@wraps(iexAuction)
def iexAuctionDF(*args, **kwargs):
return _toDatetime(json_normalize(iexAuction(*args, **kwargs)))
[docs]def iexBook(symbol=None, token="", version="stable", format="json"):
"""Book shows IEX’s bids and asks for given symbols.
https://iexcloud.io/docs/api/#deep-book
Args:
symbol (str): Ticker to request
token (str): Access token
version (str): API version
format (str): return format, defaults to json
Returns:
dict: result
"""
_raiseIfNotStr(symbol)
if symbol:
return _get(
"deep/book?symbols=" + symbol, token=token, version=version, format=format
)
return _get("deep/book", token=token, version=version, format=format)
[docs]@wraps(iexBook)
async def iexBookAsync(symbol=None, token="", version="stable", format="json"):
_raiseIfNotStr(symbol)
if symbol:
return await _getAsync(
"deep/book?symbols=" + symbol, token=token, version=version, format=format
)
return await _getAsync("deep/book", token=token, version=version, format=format)
[docs]@wraps(iexBook)
def iexBookDF(*args, **kwargs):
x = iexBook(*args, **kwargs)
data = []
for key in x:
d = x[key]
d["symbol"] = key
data.append(d)
return _toDatetime(json_normalize(data))
[docs]def iexOpHaltStatus(symbol=None, token="", version="stable", format="json"):
"""The Exchange may suspend trading of one or more securities on IEX for operational reasons and indicates such operational halt using the Operational halt status message.
IEX disseminates a full pre-market spin of Operational halt status messages indicating the operational halt status of all securities.
In the spin, IEX will send out an Operational Halt Message with “N” (Not operationally halted on IEX) for all securities that are eligible for trading at the start of the Pre-Market Session.
If a security is absent from the dissemination, firms should assume that the security is being treated as operationally halted in the IEX Trading System at the start of the Pre-Market Session.
After the pre-market spin, IEX will use the Operational halt status message to relay changes in operational halt status for an individual security.
https://iexcloud.io/docs/api/#deep-operational-halt-status
Args:
symbol (str): Ticker to request
token (str): Access token
version (str): API version
format (str): return format, defaults to json
Returns:
dict: result
"""
_raiseIfNotStr(symbol)
if symbol:
return _get(
"deep/op-halt-status?symbols=" + symbol,
token=token,
version=version,
format=format,
)
return _get("deep/op-halt-status", token=token, version=version, format=format)
[docs]@wraps(iexOpHaltStatus)
async def iexOpHaltStatusAsync(symbol=None, token="", version="stable", format="json"):
_raiseIfNotStr(symbol)
if symbol:
return await _getAsync(
"deep/op-halt-status?symbols=" + symbol,
token=token,
version=version,
format=format,
)
return await _getAsync(
"deep/op-halt-status", token=token, version=version, format=format
)
[docs]@wraps(iexOpHaltStatus)
def iexOpHaltStatusDF(*args, **kwargs):
x = iexOpHaltStatus(*args, **kwargs)
data = []
for key in x:
d = x[key]
d["symbol"] = key
data.append(d)
return _toDatetime(pd.DataFrame(data))
[docs]def iexOfficialPrice(symbol=None, token="", version="stable", format="json"):
"""The Official Price message is used to disseminate the IEX Official Opening and Closing Prices.
These messages will be provided only for IEX Listed Securities.
https://iexcloud.io/docs/api/#deep-official-price
Args:
symbol (str): Ticker to request
token (str): Access token
version (str): API version
format (str): return format, defaults to json
Returns:
dict: result
"""
_raiseIfNotStr(symbol)
if symbol:
return _get(
"deep/official-price?symbols=" + symbol,
token=token,
version=version,
format=format,
)
return _get("deep/official-price", token=token, version=version, format=format)
[docs]@wraps(iexOfficialPrice)
async def iexOfficialPriceAsync(symbol=None, token="", version="stable", format="json"):
_raiseIfNotStr(symbol)
if symbol:
return await _getAsync(
"deep/official-price?symbols=" + symbol,
token=token,
version=version,
format=format,
)
return await _getAsync(
"deep/official-price", token=token, version=version, format=format
)
[docs]@wraps(iexOfficialPrice)
def iexOfficialPriceDF(*args, **kwargs):
return _toDatetime(json_normalize(iexOfficialPrice(*args, **kwargs)))
[docs]def iexSecurityEvent(symbol=None, token="", version="stable", format="json"):
"""The Security event message is used to indicate events that apply to a security. A Security event message will be sent whenever such event occurs
https://iexcloud.io/docs/api/#deep-security-event
Args:
symbol (str): Ticker to request
token (str): Access token
version (str): API version
format (str): return format, defaults to json
Returns:
dict: result
"""
_raiseIfNotStr(symbol)
if symbol:
return _get(
"deep/security-event?symbols=" + symbol,
token=token,
version=version,
format=format,
)
return _get("deep/security-event", token=token, version=version, format=format)
[docs]@wraps(iexSecurityEvent)
async def iexSecurityEventAsync(symbol=None, token="", version="stable", format="json"):
_raiseIfNotStr(symbol)
if symbol:
return await _getAsync(
"deep/security-event?symbols=" + symbol,
token=token,
version=version,
format=format,
)
return await _getAsync(
"deep/security-event", token=token, version=version, format=format
)
[docs]@wraps(iexSecurityEvent)
def iexSecurityEventDF(*args, **kwargs):
x = iexSecurityEvent(*args, **kwargs)
data = []
for key in x:
d = x[key]
d["symbol"] = key
data.append(d)
return _toDatetime(pd.DataFrame(data))
[docs]def iexSsrStatus(symbol=None, token="", version="stable", format="json"):
"""In association with Rule 201 of Regulation SHO, the Short Sale Price Test Message is used to indicate when a short sale price test restriction is in effect for a security.
IEX disseminates a full pre-market spin of Short sale price test status messages indicating the Rule 201 status of all securities.
After the pre-market spin, IEX will use the Short sale price test status message in the event of an intraday status change.
The IEX Trading System will process orders based on the latest short sale price test restriction status.
https://iexcloud.io/docs/api/#deep-short-sale-price-test-status
Args:
symbol (str): Ticker to request
token (str): Access token
version (str): API version
format (str): return format, defaults to json
Returns:
dict: result
"""
_raiseIfNotStr(symbol)
if symbol:
return _get(
"deep/ssr-status?symbols=" + symbol,
token=token,
version=version,
format=format,
)
return _get("deep/ssr-status", token=token, version=version, format=format)
[docs]@wraps(iexSsrStatus)
async def iexSsrStatusAsync(symbol=None, token="", version="stable", format="json"):
_raiseIfNotStr(symbol)
if symbol:
return await _getAsync(
"deep/ssr-status?symbols=" + symbol,
token=token,
version=version,
format=format,
)
return await _getAsync(
"deep/ssr-status", token=token, version=version, format=format
)
[docs]@wraps(iexSsrStatus)
def iexSsrStatusDF(*args, **kwargs):
x = iexSsrStatus(*args, **kwargs)
data = []
for key in x:
d = x[key]
d["symbol"] = key
data.append(d)
return _toDatetime(pd.DataFrame(data))
[docs]def iexSystemEvent(token="", version="stable", format="json"):
"""The System event message is used to indicate events that apply to the market or the data feed.
There will be a single message disseminated per channel for each System Event type within a given trading session.
https://iexcloud.io/docs/api/#deep-system-event
Args:
token (str): Access token
version (str): API version
format (str): return format, defaults to json
Returns:
dict: result
"""
return _get("deep/system-event", token=token, version=version, format=format)
[docs]@wraps(iexSystemEvent)
async def iexSystemEventAsync(token="", version="stable", format="json"):
return await _getAsync(
"deep/system-event", token=token, version=version, format=format
)
[docs]@wraps(iexSystemEvent)
def iexSystemEventDF(*args, **kwargs):
return _toDatetime(json_normalize(iexSystemEvent(*args, **kwargs)))
[docs]def iexHist(date=None, token="", version="stable", format="json"):
"""
Args:
date (datetime): Effective date
token (str): Access token
version (str): API version
format (str): return format, defaults to json
Returns:
dict: result
"""
if date is None:
return _get("hist", token=token, version=version, format=format)
else:
date = _strOrDate(date)
return _get("hist?date=" + date, token=token, version=version, format=format)
[docs]@wraps(iexHist)
async def iexHistAsync(date=None, token="", version="stable", format="json"):
if date is None:
return await _get("hist", token=token, version=version, format=format)
else:
date = _strOrDate(date)
return await _getAsync(
"hist?date=" + date, token=token, version=version, format=format
)
[docs]@wraps(iexHist)
def iexHistDF(*args, **kwargs):
x = iexHist(*args, **kwargs)
data = []
for key in x:
dat = x[key]
for item in dat:
item["date"] = key
data.append(item)
return _reindex(_toDatetime(pd.DataFrame(data)), "date")