import json
from datetime import datetime
from decimal import Decimal
from typing import *
from mlthon.basics.defs import Exchange, InstrumentType
from mlthon.basics.mldecimal import MLDecimal
from mlthon.basics.qty import Qty
from mlthon.instruments.instrument import Instrument
from mlthon.instruments.future import Future
from mlthon.instruments.perpetual import Perpetual
[docs]class InstrumentUtils:
[docs] @staticmethod
def exch_type_sym_to_instr_name(exchange: Exchange, instr_type: InstrumentType, symbol: str) -> str:
"""
Transfer (exchange, instrument_type, symbol) into instrument_name (e.g. byt:perp:btc-usd).
Parameters
----------
exchange: Exchange
exchange of this instrument.
instr_type: InstrumentType
instrument type of this instrument.
symbol: str
symbol (internal symbol in MLTech) of this instrument.
Returns
-------
a string, the instrument name.
"""
# mltech internal sumbol is uppcase by default
return exchange.exchange_code + ":" + instr_type.type_id + ":" + symbol.upper()
[docs] @staticmethod
def exch_syms_types_to_instr_names(exchange: Exchange, type_by_symbol: Dict[str, InstrumentType]) -> List[str]:
"""Return a list of instrument names, given the exchange, symbols and types.
Parameters
----------
exchange: Exchange
exchange of these instruments.
type_by_symbol: dict
a dictionary, whose keys are the symbols and values are the instrument types, e.g.
{"BTC-USD": InstrumentType:Perpetual}.
Returns
-------
a list of strings, each of which is the corresponding instrument name.
"""
return [InstrumentUtils.exch_type_sym_to_instr_name(exchange, instr_type, symbol)
for symbol, instr_type in type_by_symbol.items()]
[docs] @staticmethod
def get_instrument_from_raw_msg(raw_msg: str) -> Union[Instrument, Future, Perpetual]:
"""Parse the raw payload (json-formatted string) and generate the corresponding Instrument object.
Parameters
----------
raw_msg: str
raw message for instrument, a string in json format, example:
```{
"instrument_type":"SPOT",
"exchange_id":"BLN",
"mlt_symbol":"BTC-USD",
"price_tick_rules_json":
{
"default": "0.1",
"rules": []
},
"quantity_tick_rules_json":
{
"default": "0.00000001",
"rules": []
},
"multiplier":"1.0000000000",
"valid_start_ts":1630003158,
"instrument_group_id":"spot:BTC-USD",
"base_currency":"BTC",
"quote_currency":"USD",
"min_quantity":"0.0005000000",
"exch_instr_id":"BTC-USD",
"quantification":"base",
"instrument_id":10000000000,
"price_precision":1,
"quantity_precision":4
}```
Returns
-------
an Instrument (or its subclass) object
"""
json_object = json.loads(raw_msg)
return InstrumentUtils.get_instrument_from_json(json_object)
[docs] @staticmethod
def get_instrument_from_json(json_object: dict) -> Union[Instrument, Future, Perpetual]:
""" Parse a json object with instrument info, and generate the corresponding Instrument object.
Parameters
----------
json_object: dict
a json object (dict), with all the information of an instrument. Example:
{
"instrument_type":"spot", # instrument type ID (e.g. perp), not the type's whole name (e.g. PERPETUAL)
"exchange_id":"BLN",
"mlt_symbol":"BTC-USD",
"price_tick_rules_json":
{
"default": "0.1",
"rules": []
},
"quantity_tick_rules_json":
{
"default": "0.00000001",
"rules": []
},
"multiplier":"1.0000000000",
"valid_start_ts":1630003158,
"instrument_group_id":"spot:BTC-USD",
"base_currency":"BTC",
"quote_currency":"USD",
"min_quantity":"0.0005000000",
"exch_instr_id":"BTC-USD",
"quantification":"base",
"instrument_id":10000000000,
"price_precision":1,
"quantity_precision":4
}
Returns
-------
an Instrument (or its subclass) object
"""
# field name changes.
json_object["exchange"] = Exchange.parse_exchange_code(json_object.pop("exchange_id"))
json_object["min_quantity"] = Qty.from_str(json_object["min_quantity"])
json_object["multiplier"] = MLDecimal.from_str(json_object["multiplier"])
# TODO: We don't need tick rules for now. But keep in here in case we need it in the future
json_object["price_tick_rules"] = json_object.pop("price_tick_rules_json")
json_object["quantity_tick_rules"] = json_object.pop("quantity_tick_rules_json")
if "expiration_date_utc" in json_object: # for futures
json_object["expiration_date_utc"] = datetime.strptime(json_object.pop("expiration_date_utc"),
"%Y-%m-%d %H:%M:%S %Z")
json_object["instrument_type"] = InstrumentType.parse_str(json_object["instrument_type"])
if json_object["instrument_type"] == InstrumentType.Spot:
return Instrument(**json_object)
elif json_object["instrument_type"] == InstrumentType.Perpetual:
return Perpetual(**json_object)
elif json_object["instrument_type"] == InstrumentType.Future:
return Future(**json_object)
else:
raise RuntimeError("Unsupported instrument type: " + str(json_object["instrument_type"]))