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']
)