-
Notifications
You must be signed in to change notification settings - Fork 7
/
Copy pathmodels.py
153 lines (120 loc) · 6.06 KB
/
models.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
import dateutil.parser
import datetime
BITMEX_MULTIPLIER = 0.00000001 # Converts satoshi numbers to Bitcoin on Bitmex
BITMEX_TF_MINUTES = {"1m": 1, "5m": 5, "1h": 60, "1d": 1440}
class Balance:
def __init__(self, info, exchange):
if exchange == "binance_futures":
self.initial_margin = float(info['initialMargin'])
self.maintenance_margin = float(info['maintMargin'])
self.margin_balance = float(info['marginBalance'])
self.wallet_balance = float(info['walletBalance'])
self.unrealized_pnl = float(info['unrealizedProfit'])
elif exchange == "binance_spot":
self.free = float(info['free'])
self.locked = float(info['locked'])
elif exchange == "bitmex":
self.initial_margin = info['initMargin'] * BITMEX_MULTIPLIER
self.maintenance_margin = info['maintMargin'] * BITMEX_MULTIPLIER
self.margin_balance = info['marginBalance'] * BITMEX_MULTIPLIER
self.wallet_balance = info['walletBalance'] * BITMEX_MULTIPLIER
self.unrealized_pnl = info['unrealisedPnl'] * BITMEX_MULTIPLIER
class Candle:
def __init__(self, candle_info, timeframe, exchange):
if exchange in ["binance_futures", "binance_spot"]:
self.timestamp = candle_info[0]
self.open = float(candle_info[1])
self.high = float(candle_info[2])
self.low = float(candle_info[3])
self.close = float(candle_info[4])
self.volume = float(candle_info[5])
elif exchange == "bitmex":
self.timestamp = dateutil.parser.isoparse(candle_info['timestamp'])
self.timestamp = self.timestamp - datetime.timedelta(minutes=BITMEX_TF_MINUTES[timeframe])
self.timestamp = int(self.timestamp.timestamp() * 1000)
self.open = candle_info['open']
self.high = candle_info['high']
self.low = candle_info['low']
self.close = candle_info['close']
self.volume = candle_info['volume']
elif exchange == "parse_trade":
self.timestamp = candle_info['ts']
self.open = candle_info['open']
self.high = candle_info['high']
self.low = candle_info['low']
self.close = candle_info['close']
self.volume = candle_info['volume']
def tick_to_decimals(tick_size: float) -> int:
tick_size_str = "{0:.8f}".format(tick_size)
while tick_size_str[-1] == "0":
tick_size_str = tick_size_str[:-1]
split_tick = tick_size_str.split(".")
if len(split_tick) > 1:
return len(split_tick[1])
else:
return 0
class Contract:
def __init__(self, contract_info, exchange):
if exchange == "binance_futures":
self.symbol = contract_info['symbol']
self.base_asset = contract_info['baseAsset']
self.quote_asset = contract_info['quoteAsset']
self.price_decimals = contract_info['pricePrecision']
self.quantity_decimals = contract_info['quantityPrecision']
self.tick_size = 1 / pow(10, contract_info['pricePrecision'])
self.lot_size = 1 / pow(10, contract_info['quantityPrecision'])
elif exchange == "binance_spot":
self.symbol = contract_info['symbol']
self.base_asset = contract_info['baseAsset']
self.quote_asset = contract_info['quoteAsset']
# The actual lot size and tick size on Binance spot can be found in the 'filters' fields
# contract_info['filters'] is a list
for b_filter in contract_info['filters']:
if b_filter['filterType'] == 'PRICE_FILTER':
self.tick_size = float(b_filter['tickSize'])
self.price_decimals = tick_to_decimals(float(b_filter['tickSize']))
if b_filter['filterType'] == 'LOT_SIZE':
self.lot_size = float(b_filter['stepSize'])
self.quantity_decimals = tick_to_decimals(float(b_filter['stepSize']))
elif exchange == "bitmex":
self.symbol = contract_info['symbol']
self.base_asset = contract_info['rootSymbol']
self.quote_asset = contract_info['quoteCurrency']
self.price_decimals = tick_to_decimals(contract_info['tickSize'])
self.quantity_decimals = tick_to_decimals(contract_info['lotSize'])
self.tick_size = contract_info['tickSize']
self.lot_size = contract_info['lotSize']
self.quanto = contract_info['isQuanto']
self.inverse = contract_info['isInverse']
self.multiplier = contract_info['multiplier'] * BITMEX_MULTIPLIER
if self.inverse:
self.multiplier *= -1
self.exchange = exchange
class OrderStatus:
def __init__(self, order_info, exchange):
if exchange == "binance_futures":
self.order_id = order_info['orderId']
self.status = order_info['status'].lower()
self.avg_price = float(order_info['avgPrice'])
self.executed_qty = float(order_info['executedQty'])
elif exchange == "binance_spot":
self.order_id = order_info['orderId']
self.status = order_info['status'].lower()
self.avg_price = float(order_info['avgPrice'])
self.executed_qty = float(order_info['executedQty'])
elif exchange == "bitmex":
self.order_id = order_info['orderID']
self.status = order_info['ordStatus'].lower()
self.avg_price = order_info['avgPx']
self.executed_qty = order_info['cumQty']
class Trade:
def __init__(self, trade_info):
self.time: int = trade_info['time']
self.contract: Contract = trade_info['contract']
self.strategy: str = trade_info['strategy']
self.side: str = trade_info['side']
self.entry_price: float = trade_info['entry_price']
self.status: str = trade_info['status']
self.pnl: float = trade_info['pnl']
self.quantity = trade_info['quantity']
self.entry_id = trade_info['entry_id']