forked from CyberPunkMetalHead/binance-trading-bot-new-coins
-
-
Notifications
You must be signed in to change notification settings - Fork 6
/
main.py
156 lines (126 loc) · 5.1 KB
/
main.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
148
149
150
151
152
153
154
155
156
import asyncio
import logging
import traceback
from datetime import datetime
from typing import List
from bot import Bot
from util import Config, Util
Config.load_global_config()
# setup logging
Util.setup_logging(name="new-coin-bot", level=Config.PROGRAM_OPTIONS["LOG_LEVEL"])
logging.getLogger("requests").setLevel(logging.WARNING)
logging.getLogger("urllib3").setLevel(logging.WARNING)
total_time = 0
total_iter = 0
def setup() -> List[Bot]:
Config.NOTIFICATION_SERVICE.info("Creating bots..")
# Create bots based on config
b = []
for broker in Config.ENABLED_BROKERS:
Config.NOTIFICATION_SERVICE.info("Creating bot [{}]".format(broker))
b.append(Bot(broker))
if len(b) > 0:
b[0].upgrade_update()
return b
async def forever(routines: List):
while True:
current_time = datetime.now()
if Config.FRONTLOAD_ENABLED:
while (
current_time.second >= Config.FRONTLOAD_START
or current_time.second
<= Config.FRONTLOAD_DURATION - (60 - Config.FRONTLOAD_START)
) and Config.auto_rate_current_weight < (Config.auto_rate_limit * 0.9):
# FRONTLOAD PERIOD
await main(routines, current_time)
current_time = datetime.now()
# STANDARD PERIOD
await main(routines, current_time)
sleep_time = get_sleep_time(current_time)
await asyncio.sleep(sleep_time)
async def main(bots_: List, current_time: datetime):
await _main(bots_)
time_taken = datetime.now() - current_time
Config.NOTIFICATION_SERVICE.debug(
"Loop finished in [{}] seconds".format(time_taken.microseconds / 1000000)
)
Config.total_time += time_taken.microseconds / 1000000
Config.total_iter += 1
Config.NOTIFICATION_SERVICE.debug(
"Request Weight: {}".format(Config.auto_rate_current_weight)
)
async def _main(bots_: List):
coroutines = [b.run_async() for b in bots_]
# This returns the results one by one.
for future in asyncio.as_completed(coroutines):
await future
def get_sleep_time(current_time: datetime) -> int:
if (
Config.auto_rate_current_weight
>= Config.auto_rate_limit * Config.RATE_INTERVENTION_PERCENTAGE / 100
):
increase_time = True
if Config.auto_rate_current_weight >= Config.auto_rate_limit * 0.95:
resume_time = datetime(
current_time.year,
current_time.month,
current_time.day,
current_time.hour,
current_time.minute + 1,
0,
0,
)
Config.NOTIFICATION_SERVICE.info(f"Bot request count above [85%] of rate limit")
if current_time.minute != Config.auto_rate_increased_minute:
Config.auto_rate_increased_minute = current_time.minute
increase_time = True
else:
increase_time = False
else:
resume_time = datetime(
current_time.year,
current_time.month,
current_time.day,
current_time.hour,
current_time.minute
if Config.FRONTLOAD_ENABLED
else current_time.minute + 1,
Config.FRONTLOAD_START if Config.FRONTLOAD_ENABLED else 0,
500000 if Config.FRONTLOAD_START else 0,
)
Config.NOTIFICATION_SERVICE.info(f"Bot request count above [{Config.RATE_INTERVENTION_PERCENTAGE}%] of "
f"rate limit")
sleep_time = min(max((resume_time - current_time).seconds, 1), 59)
if Config.AUTO_INCREASE_FREQUENCY and increase_time:
Config.NOTIFICATION_SERVICE.info(
f"Increasing FREQUENCY from [{Config.FREQUENCY_SECONDS}] to "
f"[{Config.FREQUENCY_SECONDS + Config.AUTO_INCREASE_AMOUNT}] seconds"
)
Config.FREQUENCY_SECONDS += Config.AUTO_INCREASE_AMOUNT
Config.NOTIFICATION_SERVICE.info(
"Sleeping for [{}] seconds until [{}] to avoid exceeding rate limits\n".format(
sleep_time, resume_time
)
)
elif current_time.second + Config.FREQUENCY_SECONDS > Config.FRONTLOAD_START:
sleep_time = Config.FRONTLOAD_START - current_time.second
else:
sleep_time = Config.FREQUENCY_SECONDS
Config.NOTIFICATION_SERVICE.debug("Sleeping for [{}] seconds\n".format(sleep_time))
return min(max(sleep_time, 1), 59)
if __name__ == "__main__":
Config.NOTIFICATION_SERVICE.info("Starting...")
loop = asyncio.get_event_loop()
bots = setup()
try:
loop.create_task(forever(bots))
loop.run_forever()
except KeyboardInterrupt as e:
Config.NOTIFICATION_SERVICE.info("Exiting program...")
except Exception as e:
Config.NOTIFICATION_SERVICE.error(traceback.format_exc())
finally:
for bot in bots:
bot.save()
print("AVG TIME PER LOOP: {}".format(Config.total_time / Config.total_iter))
print("TOTAL LOOPS: {}".format(Config.total_iter))