from typing import List, Dict, Optional
from pyokx import publicAPI, ENFORCED_INSTRUMENT_TYPES
from pyokx.OkxEnum import InstType
from pyokx.data_structures import Instrument
[docs]
class InstrumentSearcher:
"""
Provides functionality to search for instruments within a provided list of instruments based on various criteria
such as instrument ID, type, or underlying asset.
Args:
instruments (List[Instrument]): A list of instruments to search within.
Methods:
find_by_instId: Returns an instrument matching a specific instrument ID.
find_by_type: Returns all instruments of a specific type.
find_by_underlying: Returns all instruments with a specific underlying asset.
"""
def __init__(self, instTypes=ENFORCED_INSTRUMENT_TYPES, _instrument_map=None):
"""
InstrumentSearcher is a class that allows you to search for instruments by instId, instType, or underlying
:param instruments:
Usage:
```
okx_instrument_searcher = InstrumentSearcher(all_futures_instruments)
print(f'{okx_instrument_searcher.find_by_instId("BTC-USDT-240329") = }')
print(f'{okx_instrument_searcher.find_by_type(InstType.FUTURES) = }')
print(f'{okx_instrument_searcher.find_by_underlying("BTC-USDT") = }')
print(f'{"BTC-USDT-240329" in okx_instrument_searcher._instrument_map = }')
```
"""
if not isinstance(instTypes, list):
instTypes = [instTypes]
_instType = []
for instType in instTypes:
if isinstance(instType, InstType):
_instType.append(instType.value)
else:
_instType.append(instType)
instTypes = _instType
# assert all([instType in ENFORCED_INSTRUMENT_TYPES for instType in instTypes]), f'Invalid instTypes: {instTypes}'
self.instTypes = instTypes
if _instrument_map is None:
self.instruments = self.request_instruments()
self._instrument_map = self._create_map(self.instruments)
else:
self._instrument_map = _instrument_map
self.instruments = list(_instrument_map.values())
[docs]
def request_instruments(self):
instruments = []
for instTypes in self.instTypes:
returned_data = publicAPI.get_instruments(instType=instTypes)
if len(returned_data['data']) == 0:
if returned_data["code"] != '0':
print(f'{returned_data["code"] = }')
print(f'{returned_data["msg"] = }')
_instruments = []
else:
_instruments = []
for data in returned_data['data']:
_instruments.append(Instrument(**data))
instruments.extend(_instruments)
return instruments
def _create_map(self, instruments: List[Instrument]) -> Dict[str, Instrument]:
""" Create a map for quicker search by instId """
return {instrument.instId: instrument for instrument in instruments}
[docs]
def find_by_instId(self, instId: str) -> Optional[Instrument]:
""" Find an instrument by its instId """
instrument = self._instrument_map.get(instId)
if instrument:
return instrument if isinstance(instrument, Instrument) else Instrument(**instrument)
else:
return None
[docs]
def find_by_type(self, instType: InstType) -> List[Instrument]:
""" Find all instruments of a specific type """
if isinstance(instType, InstType):
instType = instType.value
return [instrument for instrument in self.instruments if instrument.instType == instType]
[docs]
def find_by_underlying(self, underlying: str) -> List[Instrument]:
""" Find all instruments of a specific underlying """
return [instrument for instrument in self.instruments if instrument.uly == underlying]
[docs]
def get_instrument_ids(self, instType: InstType = None) -> List[str]:
""" Get all instrument IDs """
if instType:
return [instrument.instId for instrument in self.find_by_type(instType)]
return list(self._instrument_map.keys())
[docs]
async def update_instruments(self):
""" Update the instruments list """
self.instruments = self.request_instruments()
self._instrument_map = self._create_map(self.instruments)
return self._instrument_map
if __name__ == "__main__":
instTypes = [
InstType.FUTURES,
InstType.SWAP,
InstType.SPOT,
InstType.MARGIN,
]
okx_instrument_searcher = InstrumentSearcher(instTypes=instTypes)
for instrument_type in instTypes:
result = okx_instrument_searcher.get_instrument_ids(instType=instrument_type)
print(f'InstType {instrument_type}: {result}')