Source code for crix.models

from datetime import datetime
from decimal import Decimal
from enum import Enum
from typing import NamedTuple, List, Optional, Union


[docs]class Symbol(NamedTuple): name: str base: str base_precision: int quote: str quote_precision: int description: str level_aggregation: List[int] maker_fee: Decimal taker_fee: Decimal min_lot: Decimal max_lot: Decimal min_price: Decimal max_price: Decimal min_notional: Decimal tick_lot: Decimal tick_price: Decimal is_trading: bool
[docs] @staticmethod def from_json(info: dict) -> 'Symbol': """ Construct object from dictionary """ return Symbol( name=info['symbolName'], base=info['base'], base_precision=info['basePrecision'], quote=info['quote'], quote_precision=info['quotePrecision'], description=info['desc'], level_aggregation=info['levelAggregation'], maker_fee=Decimal(info['makerFee']), taker_fee=Decimal(info['takerFee']), min_lot=Decimal(info['minLot']), max_lot=Decimal(info['maxLot']), min_price=Decimal(info['minPrice']), max_price=Decimal(info['maxPrice']), min_notional=Decimal(info['minNotional']), tick_lot=Decimal(info['tickLot']), tick_price=Decimal(info['tickPrice']), is_trading=info['trading'], )
[docs]class Ticker(NamedTuple): symbol_name: str open_time: datetime open: Decimal close: Decimal high: Decimal low: Decimal volume: Decimal # in base resolution: str
[docs] @staticmethod def from_json(info: dict) -> 'Ticker': """ Construct object from dictionary """ return Ticker( symbol_name=info['symbolName'], open_time=datetime.fromtimestamp(info['openTime'] / 1000.0), open=Decimal(info['open']), close=Decimal(info['close']), high=Decimal(info['high']), low=Decimal(info['low']), volume=Decimal(info['volume']), resolution=info['resolution'] )
[docs] @staticmethod def from_json_history(info: dict) -> 'Ticker': """ Construct object from dictionary (for a fixed resolution) """ return Ticker( symbol_name=info['currency'], open_time=datetime.fromtimestamp(info['timestamp']), open=Decimal(str(info['open'])), close=Decimal(str(info['close'])), high=Decimal(str(info['high'])), low=Decimal(str(info['low'])), volume=Decimal(str(info['volume'])), resolution='D', )
[docs]class Ticker24(NamedTuple): symbol_name: str open_time: datetime open: Decimal close: Decimal high: Decimal low: Decimal volume: Decimal # in base resolution: str # extended info first_id: int last_id: int prev_close_price: Decimal price_change: Decimal price_change_percent: Decimal
[docs] @staticmethod def from_json(info: dict) -> 'Ticker24': """ Construct object from dictionary """ return Ticker24( symbol_name=info['symbolName'], open_time=datetime.fromtimestamp(info['openTime'] / 1000.0), open=Decimal(info['open']), close=Decimal(info['close']), high=Decimal(info['high']), low=Decimal(info['low']), volume=Decimal(info['volume']), resolution=info['resolution'], first_id=info['firstId'], last_id=info['lastId'], prev_close_price=Decimal(info['prevClosePrice'] or '0'), price_change=Decimal(info['priceChange'] or '0'), price_change_percent=Decimal(info['priceChangePercent'] or '0'), )
[docs]class Offer(NamedTuple): count: int price: Decimal quantity: Decimal
[docs] @staticmethod def from_json(info: dict) -> 'Offer': """ Construct object from dictionary """ return Offer( count=info['c'], price=Decimal(info['p']), quantity=Decimal(info['q']) )
[docs]class Depth(NamedTuple): symbol_name: str is_aggregated: bool last_update_id: int level_aggregation: int asks: List[Offer] bids: List[Offer]
[docs] @staticmethod def from_json(info: dict) -> 'Depth': """ Construct object from dictionary """ return Depth( symbol_name=info['symbolName'], level_aggregation=info['levelAggregation'], last_update_id=info['lastUpdateId'], is_aggregated=info['aggregated'], asks=[Offer.from_json(offer) for offer in (info['asks'] or [])], bids=[Offer.from_json(offer) for offer in (info['bids'] or [])], )
[docs]class Resolution(Enum): one_minute = '1' five_minutes = '5' fifteen_minutes = '15' half_an_hour = '30' hour = '60' two_hours = '120' four_hours = '240' day = 'D' week = 'W'
[docs]class TimeInForce(Enum): good_till_cancel = 0 immediate_or_cancel = 1 fill_or_kill = 2 good_till_date = 3
[docs]class OrderStatus(Enum): new = 0 complete = 1 cancel = 2
[docs]class OrderType(Enum): limit = 0 market = 1 stop_loss = 2 stop_loss_limit = 3 stop_loss_range = 4 # deprecated take_profit = 5 take_profit_limit = 6
[docs]class Order(NamedTuple): id: int user_id: int type: OrderType symbol_name: str is_buy: bool quantity: Decimal price: Decimal stop_price: Decimal filled_quantity: Decimal time_in_force: TimeInForce expire_time: Optional[datetime] status: OrderStatus created_at: datetime # matches server time when order was placed to the order book last_updated_at: datetime # match server time when order status changed
[docs] @staticmethod def from_json(info: dict) -> 'Order': """ Construct object from dictionary """ return Order( id=info['orderId'], user_id=info['userId'], type=OrderType(info.get('type', 0)), symbol_name=info['symbolName'], is_buy=info['isBuy'], quantity=Decimal(info['quantity'] or '0'), price=Decimal(info['price'] or '0'), stop_price=Decimal(info['stopPrice'] or '0'), filled_quantity=Decimal(info['filledQuantity'] or '0'), time_in_force=TimeInForce(info['timeInForce']), expire_time=datetime.fromtimestamp(info['expireTime'] / 1000) if info['expireTime'] and info[ 'expireTime'] > 0 else None, status=OrderStatus(info['status']), created_at=datetime.fromtimestamp(info.get('createdAt', 0) / 1000), last_updated_at=datetime.fromtimestamp(info.get('lastUpdateAt', 0) / 1000), )
[docs]class NewOrder(NamedTuple): type: OrderType symbol: str price: Decimal quantity: Decimal is_buy: bool time_in_force: TimeInForce = TimeInForce.good_till_cancel stop_price: Optional[Decimal] = None expire_time: Optional[datetime] = None
[docs] def to_json(self) -> dict: """ Build JSON package ready to send to the API endpoint """ req = { "type": self.type.value, "isBuy": self.is_buy, "price": str(self.price), "quantity": str(self.quantity), "symbolName": self.symbol, "timeInForce": self.time_in_force.value, } if self.stop_price is not None: req["stopPrice"] = str(self.stop_price) if self.expire_time is not None: req["expireTime"] = int(self.expire_time.timestamp() * 1000) return req
[docs] @staticmethod def limit(symbol: str, is_buy: bool, price: Union[Decimal, float, str], quantity: Union[Decimal, float, str], **args) -> 'NewOrder': """ Helper to create basic limit order :param symbol: symbol name as defined by the exchange :param is_buy: order direction :param price: order price :param quantity: number of items in the order :param args: additional parameters proxied to the NewOrder constructor :return: new order """ return NewOrder(type=OrderType.limit, symbol=symbol, price=Decimal(price), quantity=Decimal(quantity), is_buy=is_buy, **args)
[docs] @staticmethod def market(symbol: str, is_buy: bool, quantity: Union[Decimal, float, str], **args) -> 'NewOrder': """ Helper to create basic market order :param symbol: symbol name as defined by the exchange :param is_buy: order direction :param quantity: number of items :param args: additional parameters proxied to the NewOrder constructor :return: new order """ return NewOrder(type=OrderType.market, symbol=symbol, price=Decimal('0'), quantity=Decimal(quantity), is_buy=is_buy, time_in_force=TimeInForce.immediate_or_cancel, **args)
[docs]class Trade(NamedTuple): id: int user_id: int # matches user account that made an order created_at: datetime order_filled: bool # matches order filling state (complete or not) is_buy: bool order_id: int # matches original order id (buy or sell) price: Decimal quantity: Decimal fee: Decimal fee_currency: str symbol_name: str
[docs] @staticmethod def from_json(info: dict) -> 'Trade': """ Construct object from dictionary """ return Trade( id=info['id'], created_at=datetime.fromtimestamp(info['createdAt'] / 1000), order_filled=info.get('orderFilled', False), is_buy=info['isBuy'], order_id=info.get('orderId', 0), price=Decimal(info['price']), quantity=Decimal(info['quantity']), fee=Decimal(info['fee'] or '0'), fee_currency=info.get('feeCurrency', ''), symbol_name=info['symbolName'], user_id=info.get('userId', 0), )
[docs]class Account(NamedTuple): id: int user_id: int balance: Decimal locked_balance: Decimal currency_name: str deposit_address: str
[docs] @staticmethod def from_json(info: dict) -> 'Account': """ Construct object from dictionary """ return Account( id=info['id'], user_id=info['userId'], balance=Decimal(info['balance'] or '0'), locked_balance=Decimal(info['lockedBalance'] or '0'), currency_name=info['currencyName'], deposit_address=info['depositAddress'] )