from itertools import accumulateimport timeimport osimport jsonimport fc

from itertools import accumulate

import time

import os

import json

import fcntl

import random

from concurrent.futures import ThreadPoolExecutor, as_completed

from target_fetcher import TargetPriceFetching

def proper_round(x, precision):

return float(f"%.{precision}f" % round(x, precision))

def str_round(x, precision):

return f"%.{precision}f" % round(x, precision)

class MarketMaking:

def __init__(self, group, cooldown = 2.5, ema_gamma = 0., prev_mid = 0, only_usdt=True):

self.cooldown = cooldown

self.params = []

self.ema_gamma = 0.0

self.prev_mid = prev_mid

self.group = group

self.zi = 0

self.last_len = 0

self.price = 0

self.thread_pool = ThreadPoolExecutor(max_workers=100)

self.logs_count = 0

self.tp_fetcher = TargetPriceFetching(group=self.group, only_usdt=only_usdt)

def graceful_stop(self):

print('MM bot starts graceful stop...')

with open(f'/mnt/.launch-cache/{self.group}.m.runtime', 'r') as f:

runtime_params = json.loads(f.read())

for params in self.params:

if params['market'].market not in runtime_params:

runtime_params[params['market'].market] = {}

runtime_params[params['market'].market]['resistance_base_filled_amount'] = params['resistance_base_filled_amount']

runtime_params[params['market'].market]['resistance_quote_filled_amount'] = params['resistance_quote_filled_amount']

with open(f'/mnt/.launch-cache/{self.group}.m.runtime', 'w') as f:

f.write(f'{json.dumps(runtime_params)}\n')

f.flush()

os.fsync(f.fileno())

print('finish')

def handle_private_trades(self, params, trades):

try:

for trade in trades:

if trade.get('isTaker', False):

print(f'{params['market'].market} MM taker trade:', trade)

if trade['side'] == 'buy':

params['resistance_base_filled_amount'] -= trade['quantity']

params['resistance_quote_filled_amount'] += (trade['quantity'] * trade['price'])

else:

params['resistance_base_filled_amount'] += trade['quantity']

params['resistance_quote_filled_amount'] -= (trade['quantity'] * trade['price'])

if len(trades) > 0 and 'isTaker' in trades[0]: # just logging for exchanges isTaker supported

print(f"{params['market'].market} resistance filled: {params['resistance_base_filled_amount']} {params['resistance_quote_filled_amount']}")

except Exception as e:

print(f'{e} in handle_private_trades')

def log(self, params, name, obj, need_flush=False):

self.logs_count += 1

params['log'].write(f'{int(time.time() * 1000)} {name} {obj}\n')

if need_flush or self.logs_count % 10:

params['log'].flush()

self.logs_count = 0

def make_zone(self,

bound = 1500,

total_base = None,

total_quote = None,

qty_step = 2000,

n_levels = 20,

protection = False

):

zone = {}

zone['zone'] = float(bound)

zone['n_levels'] = int(n_levels)

zone['total_base'] = float(total_base) if total_base != None else None

zone['total_quote'] = float(total_quote) if total_quote != None else None

zone['protection'] = bool(protection)

zone['qty_step'] = float(qty_step) / 10000

return zone

# def add_market(self, market, levels, margins, bal_cut, zero_pos, min_bal, n_sup=4, ask_sup_cut=0.05, bid_sup_cut=0.05, cancel_on_start=False, ask_bal_cut=None, bid_bal_cut=None):

def add_market(self,

market,

spread_size = 200,

zero_base = None,

zero_quote = None,

taker_buyback_price = None,

taker_buyback_funds = 0,

drop_base_offset = 0,

drop_quote_offset = 0,

ask_zones = [],

bid_zones = [],

cancel_on_start = False,

retain = False,

tb_order_funds = 0,

tb_spread_size = 1,

tb_price_bound = 0,

min_order_size = 5,

min_base_balance_for_resistance = -1,

min_quote_balance_for_resistance = -1):

params = {}

ask_zones = sorted(ask_zones, key = lambda x: x['zone'])

bid_zones = sorted(bid_zones, key = lambda x: x['zone'])

params['market'] = market

add = spread_size // 2

if min_order_size > 0.1: # иначе скорее всего конфиг не в USDT заполнили случайно и делить еще раз не нужно

print(f'[warn] min_order_size not in USDT: {min_order_size}')

min_order_size /= self.tp_fetcher.get_quote_in_price(market.symbol[2])

print('min_order_size:', min_order_size)

params['min_order_size'] = min_order_size

params['tb_order_funds'] = tb_order_funds

params['tb_spread_size'] = tb_spread_size

params['tb_price_bound'] = tb_price_bound

params['retain'] = retain

params['drop_base_offset'] = params['min_pos_base'] = drop_base_offset

params['taker_buyback_price'] = taker_buyback_price

params['taker_buyback_funds'] = taker_buyback_funds

params['total_sell_base'] = 0

print(f'mbbfr={min_base_balance_for_resistance}')

params['min_base_balance_for_resistance'] = min_base_balance_for_resistance

params['min_quote_balance_for_resistance'] = min_quote_balance_for_resistance

if not os.path.isfile(f"/mnt/.launch-cache/{self.group}.m.runtime"):

empty_file = open(f"/mnt/.launch-cache/{self.group}.m.runtime", 'w')

empty_file.write('{}')

empty_file.close()

try:

with open(f"/mnt/.launch-cache/{self.group}.m.runtime", "r") as f:

runtime_params = json.loads(f.read())

params['resistance_base_filled_amount'] = runtime_params[params['market'].market]['resistance_base_filled_amount']

params['resistance_quote_filled_amount'] = runtime_params[params['market'].market]['resistance_quote_filled_amount']

except:

params['resistance_base_filled_amount'] = 0

params['resistance_quote_filled_amount'] = 0

print(f"{params['market'].market} resistance params: {params['resistance_base_filled_amount']} {params['resistance_quote_filled_amount']}")

params['prev_bid'] = 0

params['prev_ask'] = 0

params['resistance_cases_count'] = 0

for zone in ask_zones:

if not 'protection' in zone or zone['protection'] != True:

params['min_pos_base'] += zone['total_base']

params['total_sell_base'] += zone['total_base']

print(zone['total_base'])

zone['min_level_base'] = zone['total_base'] / (zone['n_levels'] * (1 + zone['qty_step'] * (zone['n_levels'] - 1) / 2))

zone['price_step'] = (zone['zone'] - add + spread_size) // zone['n_levels']

zone['levels'] = [add + zone['price_step'] * i for i in range(zone['n_levels'])]

zone['prev'] = [0 for lvl in zone['levels']]

zone['margins'] = [zone['price_step'] // 2 for i in range(zone['n_levels'])]

add = zone['zone'] + zone['price_step']

params['zero_base'] = zero_base

params['ask_zones'] = ask_zones

params['stopped'] = False

add = spread_size // 2

params['spread_size'] = spread_size

params['drop_quote_offset'] = params['max_pos_quote'] = drop_quote_offset

params['total_buy_quote'] = 0

for zone in bid_zones:

if not 'protection' in zone or zone['protection'] != True:

params['max_pos_quote'] += zone['total_quote']

params['total_buy_quote'] += zone['total_quote']

zone['min_level_quote'] = zone['total_quote'] / (zone['n_levels'] * (1 + zone['qty_step'] * (zone['n_levels'] - 1) / 2))

zone['price_step'] = (zone['zone'] - add + spread_size) // zone['n_levels']

zone['levels'] = [add + zone['price_step'] * i for i in range(zone['n_levels'])]

zone['prev'] = [0 for lvl in zone['levels']]

zone['margins'] = [zone['price_step'] // 2 for i in range(zone['n_levels'])]

if add == spread_size // 2:

zone['margins'][0] = 0

add = zone['zone'] + zone['price_step']

params['zero_quote'] = zero_quote

params['bid_zones'] = bid_zones

params['cnt'] = 0

params['cancel_on_start'] = cancel_on_start

logfile = f"/mnt/mm_telemetry/{market.symbol[1].upper()}_{market.symbol[2].upper()}_{market.market}"

if not os.path.isfile(logfile):

open(logfile, 'a').close()

with open(logfile, 'r+') as f:

fcntl.flock(f, fcntl.LOCK_EX)

lines = f.readlines()

if len(lines) > 2e6:

keep_probability = 0.05

elif len(lines) > 1e6:

keep_probability = 0.2

elif len(lines) > 2e5:

keep_probability = 0.5

else:

keep_probability = 1

if keep_probability != 1:

retained_lines = [line for line in lines if random.random() < keep_probability]

f.seek(0)

f.truncate()

f.writelines(retained_lines)

del lines

params['log'] = open(logfile, "a")

params['prev_tb_price'] = 0

bid_ask_file = f'/mnt/mm_bid_ask/{market.symbol[1].upper()}_{market.symbol[2].upper()}_{market.market}'

try:

if not os.path.isfile(bid_ask_file):

open(bid_ask_file, 'a').close()

with open(bid_ask_file, 'r') as f:

prev = f.readlines()

except Exception as e:

print('Excpetion mm bid-ask:', e)

prev = []

with open(bid_ask_file, 'r+') as f:

fcntl.flock(f, fcntl.LOCK_EX)

f.seek(0)

f.truncate(0)

f.writelines(prev[-100:])

f.flush()

os.fsync(f.fileno())

params['bid_ask_log'] = open(bid_ask_file, 'a')

self.log(params, "alert", {'op': 'start'}, True)

self.params.append(params)

def build_grid(self, params, mid_price):

"""

не хочется терять consistency в логах,

поэтому вместо логгирования раз в N итераций,

будем делать запросы в несколько тредов

"""

bg = time.time()

futures = {}

futures[self.thread_pool.submit(params['market'].get_info)] = 'info'

futures[self.thread_pool.submit(params['market'].get_new_public_trades)] = 'public_trades'

futures[self.thread_pool.submit(params['market'].get_new_private_trades)] = 'private_trades'

futures[self.thread_pool.submit(params['market'].get_depth)] = 'depth'

for f in as_completed(futures):

result = f.result()

if futures[f] == 'info':

info = result

self.log(params, 'book', info['book'])

self.log(params, 'balance', info['balance'])

elif futures[f] == 'depth': # TODO: добавить в get_info кол-ва в бестах

depth = result

elif futures[f] == 'private_trades':

self.handle_private_trades(params, result)

self.log(params, futures[f], result)

else:

self.log(params, futures[f], result)

print('info-trades-log latency', int(1000*(time.time() - bg)))

asks = []

bids = []

precision = params['market'].symbol[3]

if info['book']['ask_price'] <= mid_price * (1 - params['spread_size'] / 20000) or info['book']['bid_price'] >= mid_price * (1 + params['spread_size'] / 20000):

params['resistance_cases_count'] += 1

else:

params['resistance_cases_count'] = 0

if info['book']['ask_price'] <= mid_price * (1 - params['spread_size'] / 20000) and params['resistance_cases_count'] >= 2:

print('need to align: ask price is too low')

if params['min_quote_balance_for_resistance'] != -1:

if params['prev_ask'] == 0 or params['prev_ask'] > info['book']['ask_price']: # чтобы не бить в себя (актуально для таргета с DEX)

qty = depth['asks'][0][1]

if info['balance']['quote'] - qty * info['book']['ask_price'] >= params['min_quote_balance_for_resistance']:

print(f'{params['market'].market} resistance buy trade: {info['book']['ask_price']} {qty}')

params['market'].new_limit(proper_round(info['book']['ask_price'], precision), max(qty, 5 / info['book']['ask_price']), True)

elif info['book']['bid_price'] >= mid_price * (1 + params['spread_size'] / 20000) and params['resistance_cases_count'] >= 2:

print('need to align: bid price is too high')

if params['min_base_balance_for_resistance'] != -1:

if params['prev_bid'] == 0 or params['prev_bid'] < info['book']['bid_price']: # чтобы не бить в себя (актуально для таргета с DEX)

qty = depth['bids'][0][1]

if info['balance']['base'] - qty >= params['min_base_balance_for_resistance']:

print(f'{params['market'].market} resistance sell trade: {info['book']['bid_price']} {qty}')

params['market'].new_limit(proper_round(info['book']['bid_price'], precision), max(qty, 5 / info['book']['bid_price']), False)

params['last_mid_price'] = (info['book']['ask_price'] + info['book']['bid_price']) / 2

print(f"{params['market'].market}: cur balances: {info['balance']}, zero base: {params['zero_base']}, zero quote: {params['zero_quote']}")

print(f'mid price: {mid_price}')

real_ask = info['book']['ask_price']

real_bid = info['book']['bid_price']

if real_ask == 0:

real_ask = mid_price

price_step = proper_round(0.1 ** precision, precision)

base_balance = info['balance']['base']

quote_balance = info['balance']['quote']

ask_bound = 0

bid_bound = 1e9

cur_pos_base = (base_balance - params['zero_base'] + params['resistance_base_filled_amount'])

print(f"{cur_pos_base} ~~~ {-params['min_pos_base']}")

if cur_pos_base < -params['min_pos_base']:

params['stopped'] = True

raise Exception(f"{params['market'].market} stopped")

cur_pos_quote = (params['zero_quote'] - quote_balance - params['resistance_quote_filled_amount'])

print(f"{cur_pos_quote} -- {params['max_pos_quote']}")

if cur_pos_quote > params['max_pos_quote']:

params['stopped'] = True

raise Exception(f"{params['market'].market} stopped")

bound = mid_price

if base_balance > params['zero_base'] + 1 / mid_price:

bound = pnl_ask_bound = ask_bound = (params['zero_quote'] - quote_balance) / (base_balance - params['zero_base'])

elif base_balance < params['zero_base'] - 1 / mid_price:

bound = pnl_bid_bound = bid_bound = (quote_balance - params['zero_quote']) / (params['zero_base'] - base_balance)

print(f'ask_bound={ask_bound} bid_bound={bid_bound} <<<<<<')

position_info = {'bound': bound, 'zero_base': params['zero_base'], 'zero_quote': params['zero_quote']}

self.log(params, 'position', position_info)

available_sell = []

cur_total_sell = max(0, params['total_sell_base'] + min(0, cur_pos_base + params['drop_base_offset']))

sell_coef = cur_total_sell / max(1, params['total_sell_base'])

norm_coef = min(base_balance, cur_total_sell) / max(1, cur_total_sell)

for zone in reversed(params['ask_zones']):

# if zone['protection']:

# delta = zone['total_base']

# else:

# delta = min(cur_total_sell, zone['total_base'])

# cur_total_sell -= delta

available_sell.append(zone['total_base'])

available_sell = list(reversed(available_sell))

zi = 0

for zone in params['ask_zones']:

min_qty = zone['min_level_base']

qty_step = zone['qty_step']

if min_qty * zone['n_levels'] >= available_sell[zi]:

min_qty = available_sell[zi] / zone['n_levels']

qty_step = 0

else:

qty_step = (available_sell[zi] / (min_qty * zone['n_levels']) - 1) * 2 / (zone['n_levels'] - 1)

for i in range(zone['n_levels']):

suggest_ask = proper_round(mid_price * (1 + zone['levels'][i] / 10000), precision)

if abs(zone['prev'][i] / suggest_ask - 1) <= zone['margins'][i] / 10000:

suggest_ask = zone['prev'][i]

if len(asks) > 0 and asks[-1][0] >= suggest_ask:

suggest_ask = asks[-1][0] + price_step

zone['prev'][i] = suggest_ask

qty = min_qty * (1 + i * qty_step) * norm_coef

if not zone['protection']:

qty *= sell_coef

asks.append([float(str_round(suggest_ask, precision)), qty])

zi += 1

available_buy = []

cur_total_buy = params['total_buy_quote'] - max(0, cur_pos_quote - params['drop_quote_offset'])

buy_coef = cur_total_buy / max(1, params['total_buy_quote'])

norm_coef = min(quote_balance, cur_total_buy) / max(1, cur_total_buy)

print(f'norm coef = {norm_coef}')

print(f'buy coef = {buy_coef}')

for zone in reversed(params['bid_zones']):

# if zone['protection']:

# delta = zone['total_quote']

# else:

# delta = min(cur_total_buy, zone['total_quote'])

# cur_total_buy -= delta

available_buy.append(zone['total_quote'])

available_buy = list(reversed(available_buy))

zi = 0

for zone in params['bid_zones']:

min_qty = zone['min_level_quote']

print(f'min qty = {min_qty}')

qty_step = zone['qty_step']

if min_qty * zone['n_levels'] >= available_buy[zi]:

min_qty = available_buy[zi] / zone['n_levels']

qty_step = 0

else:

qty_step = (available_buy[zi] / (min_qty * zone['n_levels']) - 1) * 2 / (zone['n_levels'] - 1)

print('qty_step:', qty_step)

print('min_qty:', min_qty)

for i in range(zone['n_levels']):

suggest_bid = proper_round(mid_price * (1 - zone['levels'][i] / 10000), precision)

if abs(zone['prev'][i] / suggest_bid - 1) <= zone['margins'][i] / 10000:

suggest_bid = zone['prev'][i]

if len(bids) > 0 and bids[-1][0] <= suggest_bid:

suggest_bid = bids[-1][0] - price_step

print('suggest_bid:', suggest_bid)

zone['prev'][i] = suggest_bid

print('norm_coef:', norm_coef)

qty = (min_qty * (1 + i * qty_step)) / suggest_bid * norm_coef

print('qty:', qty)

if not zone['protection']:

qty *= buy_coef

bids.append([float(str_round(suggest_bid, precision)), qty])

zi += 1

if not params['retain']:

ask_bound = max(ask_bound, real_bid + price_step)

bid_bound = min(bid_bound, real_ask - price_step)

min_ask_price = 1e9

for ask in asks:

min_ask_price = min(min_ask_price, ask[0])

if min_ask_price < ask_bound:

for ask in asks:

ask[0] += ask_bound - min_ask_price

ask[0] = float(str_round(ask[0], precision))

max_bid_price = 0

for bid in bids:

max_bid_price = max(max_bid_price, bid[0])

if max_bid_price > bid_bound:

for bid in bids:

funds = bid[1] * bid[0]

bid[0] -= max_bid_price - bid_bound

bid[0] = float(str_round(bid[0], precision))

bid[1] = funds / bid[0] if bid[0] > 0 else 0

print(sorted([ask[0] for ask in asks]))

print(f'prev asks: {asks}')

print(f'prev bids: {bids}')

actual_asks, actual_bids = [], []

asks_less_than_min_size, bids_less_than_min_size = 0, 1 # в quote из-за умножения больше неточность

for ask in asks:

if ask[0] * ask[1] >= 1.01 * params['min_order_size']:

actual_asks.append(ask)

else:

asks_less_than_min_size += (ask[0] * ask[1])

for bid in bids:

if bid[0] * bid[1] >= 1.01 * params['min_order_size']:

actual_bids.append(bid)

else:

bids_less_than_min_size += (bid[0] * bid[1])

print(f"ask less: {asks_less_than_min_size}, bid less: {bids_less_than_min_size}")

print(f"cur_pos_base={cur_pos_base}, shit={-params['min_pos_base'] + asks_less_than_min_size}")

if cur_pos_base < -params['min_pos_base'] + asks_less_than_min_size:

params['stopped'] = True

raise Exception(f"{params['market'].market} stopped")

if cur_total_buy > 0 and cur_pos_quote + bids_less_than_min_size > params['max_pos_quote']:

params['stopped'] = True

raise Exception(f"{params['market'].market} stopped")

if params['tb_order_funds'] != 0:

if cur_pos_base > 0: # открыта long позиция -> закрываем продажей

""" вместо real_ask можно использовать вместо min(real_ask, user_ask_new), т.к.

плохой случай, это когда мы сейчас выставимся более узко, чем было до этого,

но т.к. мм после этого сужения на следующей итерации не будет сужаться сильнее, то значит на следующей

итерации всё выставится как надо.

и ещё плохой случай, когда цена постоянно уходит вниз, но раз цена идет вниз, значит давление на продажу,

а значит наш кэшаут вряд ли исполнится, и некритично подождать пока всё сбалансируется

"""

print(f'real ask: {real_ask} {real_bid}')

if params['prev_tb_price'] != real_ask:

params['prev_tb_price'] = real_ask - price_step

price = params['prev_tb_price']

if price <= real_bid:

price = real_ask

qty = min(cur_pos_base, params['tb_order_funds'] / price)

if params['tb_price_bound'] > 0:

price = max(price, params['tb_price_bound'])

if params['tb_spread_size'] > 0 and real_bid:

price = max(price, real_bid * (1 + params['tb_spread_size'] / 100))

if qty * price >= params['min_order_size'] + 0.1 and price >= pnl_ask_bound:

actual_asks.append([float(str_round(price, precision)), qty])

elif cur_pos_base < 0: # открыта short позиция -> закрываем покупкой

price = real_bid + price_step

if price >= real_ask:

price = real_bid

qty = min(-cur_pos_base, params['tb_order_funds'] / price)

if params['tb_price_bound'] > 0:

price = min(price, params['tb_price_bound'])

if params['tb_spread_size'] > 0 and real_ask:

price = min(price, real_ask * (1 - params['tb_spread_size'] / 100))

if qty * price >= params['min_order_size'] + 0.1 and price <= pnl_bid_bound:

actual_bids.append([float(str_round(price, precision)), qty])

print('B', flush=True)

return (actual_asks, actual_bids)

def process_grid(self, params, mid_price):

orders = params['market'].get_orders()

if len(orders['ask']) > 0:

params['prev_ask'] = min(orders['ask'], key = lambda x: x['price'])['price']

else:

params['prev_ask'] = 0

if len(orders['bid']) > 0:

params['prev_bid'] = max(orders['bid'], key = lambda x: x['price'])['price']

else:

params['prev_bid'] = 0

new_grid = self.build_grid(params, mid_price)

print('built')

print(new_grid)

return

new_ask_grid = {float(ask[0]): ask[1] for ask in new_grid[0]}

new_bid_grid = {float(bid[0]): bid[1] for bid in new_grid[1]}

bg = time.time()

self.log(params, 'orders', orders)

print('get-orders latency', int(1000*(time.time() - bg)))

ask2 = 0

bid2 = 0

bg = time.time()

if params['last_mid_price'] > 0:

for order in orders['ask']:

ask2 += order['price'] * order['quantity'] * (order['price'] / params['last_mid_price'] <= 1.02)

for order in orders['bid']:

bid2 += order['price'] * order['quantity'] * (order['price'] / params['last_mid_price'] >= 0.98)

self.log(params, 'depth', {'ask2': ask2, 'bid2': bid2}, params['cnt'] % 5 != 0)

params['cnt'] += 1

if params['cnt'] % 5 == 0:

depth = params['market'].get_depth()

ask_price = depth['asks'][0][0] if len(depth['asks']) > 0 else 1e9

bid_price = depth['bids'][0][0] if len(depth['bids']) > 0 else 0

organic = {'ask': [0 for i in range(30)], 'bid': [0 for i in range(30)]}

user = {'ask': [0 for i in range(30)], 'bid': [0 for i in range(30)]}

for ask in depth['asks']:

percent = int(100 * (ask[0] / ask_price - 1))

if percent >= 30:

break

organic['ask'][percent] += ask[1] * ask[0]

for bid in depth['bids']:

percent = int(100 * (1 - bid[0] / bid_price))

if percent >= 30:

continue

organic['bid'][percent] += bid[1] * bid[0]

if ask_price == 1e9 and len(orders['ask']) > 0:

ask_price = min([order['price'] for order in orders['ask']])

if bid_price == 0 and len(orders['bid']) > 0:

bid_price = max([order['price'] for order in orders['bid']])

for order in orders['ask']:

qty = order['quantity']

price = order['price']

if price < ask_price:

continue

percent = int(100 * (price / ask_price - 1))

if percent >= 30:

continue

organic['ask'][percent] -= qty * price

user['ask'][percent] += qty * price

for order in orders['bid']:

qty = order['quantity']

price = order['price']

if price > bid_price:

continue

percent = int(100 * (1 - price / bid_price))

if percent >= 30:

continue

organic['bid'][percent] -= qty * price

user['bid'][percent] += qty * price

organic['ask'] = list(accumulate(organic['ask']))

organic['bid'] = list(accumulate(organic['bid']))

user['ask'] = list(accumulate(user['ask']))

user['bid'] = list(accumulate(user['bid']))

self.log(params, 'organic_depth', organic)

self.log(params, 'user_depth', user, True)

step = 0.1 ** params['market'].symbol[3]

if params['taker_buyback_price'] != None and params['taker_buyback_funds'] >= 5 and min(new_ask_grid.keys()) >= ask_price + step and ask_price < mid_price:

delta = min(depth['asks'][0][1], params['taker_buyback_funds'] / ask_price)

print(f'LOL {ask_price} {delta}')

params['taker_buyback_funds'] -= delta * ask_price

params['market'].new_limit(ask_price, delta, True)

print(params['cnt'])

print(f"{params['market'].market} asks: {orders['ask']}")

print(f"{params['market'].market} bids: {orders['bid']}")

print('extra-orders-log latency', 1000*(time.time() - bg))

bg = time.time()

try:

new_bid_price, new_ask_price = max(new_bid_grid.keys()), min(new_ask_grid.keys())

bid_price_log = {'ts': int(time.time()*1000), 'bid_price': new_bid_price, 'ask_price': new_ask_price}

params['bid_ask_log'].write(f"{bid_price_log}\n")

params['bid_ask_log'].flush()

except Exception as e:

print(f'Exception in bid-ask logging: {e}', flush=True)

print('bid-ask-log latency', 1000*(time.time() - bg))

to_cancel = []

to_create = [[], []]

for order in orders['ask']:

price = order['price']

if price in new_ask_grid and abs(new_ask_grid[price] / order['quantity'] - 1) <= 0.05:

new_ask_grid.pop(order['price'])

else:

to_cancel.append(order['id'])

for order in orders['bid']:

price = order['price']

if price in new_bid_grid and abs(new_bid_grid[price] / order['quantity'] - 1) <= 0.05:

new_bid_grid.pop(price)

else:

to_cancel.append(order['id'])

for price in new_ask_grid:

ask = [price, new_ask_grid[price]]

if not ask[0] in orders['ask']:

to_create[0].append(ask)

for price in new_bid_grid:

bid = [price, new_bid_grid[price]]

if not bid[0] in orders['bid']:

to_create[1].append(bid)

print(f'New len: {len(to_create[0])} + {len(to_create[1])}, cancel len: {len(to_cancel)}')

bg = time.time()

if hasattr(params['market'], 'cancel_batch') and callable(getattr(params['market'], 'cancel_batch')):

self.thread_pool.submit(params['market'].cancel_batch, to_cancel)

else:

for oid in to_cancel:

self.zi += 1

self.thread_pool.submit(params['market'].cancel, oid)

if hasattr(params['market'], 'new_batch_limit_maker') and callable(getattr(params['market'], 'new_batch_limit_maker')):

self.thread_pool.submit(params['market'].new_batch_limit_maker, to_create[0], to_create[1])

else:

for ask in to_create[0]:

self.zi += 1

self.thread_pool.submit(params['market'].new_limit_maker, ask[0], ask[1], False)

for bid in to_create[1]:

self.zi += 1

self.thread_pool.submit(params['market'].new_limit_maker, bid[0], bid[1], True)

print('orders-update latency', 1000*(time.time() - bg))

def run_event_loop(self):

for i in range(len(self.params)):

if self.params[i]['cancel_on_start']:

self.params[i]['market'].cancel_open_orders()

while True:

try:

print('time:', int(time.time()*1000))

bg = time.time()

mid = self.tp_fetcher.target()

print('target init:', mid)

futures = {}

for params in self.params:

if not params['stopped']:

if params['market'].symbol[2].upper() != 'USDT':

quote_price = self.tp_fetcher.get_quote_in_price(params['market'].symbol[2])

print('quote price:', quote_price)

mid /= quote_price

print('target after:', mid)

future = self.thread_pool.submit(self.process_grid, params, mid)

futures[future] = params

for f in as_completed(futures):

try:

_ = f.result()

except Exception as e:

print(f'exception: {e}')

self.log(futures[f], 'orders', futures[f]['market'].get_orders(), True)

if futures[f]['stopped']:

self.log(params, 'alert', {'op': 'stop'}, True)

print('local-cycle latency', int(1000*(time.time() - bg)))

except Exception as e:

print(f'exception: {e}')

finally:

time.sleep(self.cooldown)

Untitled