-
Notifications
You must be signed in to change notification settings - Fork 1
/
monitor-profit.py
executable file
·218 lines (184 loc) · 7.83 KB
/
monitor-profit.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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
#!/usr/bin/env python3
"""\
monitor-profit.py - P&L at Bybit
docs: https://bybit-exchange.github.io/docs/v5/intro
works with Cross-Margin
pip install pybit
server time error:
pybit._http_manager - ERROR - invalid request, please check your server timestamp or recv_window param.
req_timestamp[1707815756893],server_timestamp[1707815762060],recv_window[5000] (ErrCode: 10002).
"""
__project__ = "Trading Bot"
__part__ = 'PnL @ Bybit' # Futures only
__author__ = "Sergey V Musenko"
__email__ = "sergey@musenko.com"
__copyright__= "© 2024, musenko.com"
__license__ = "MIT"
__credits__ = ["Sergey Musenko"]
__date__ = "2024-01-18"
__version__ = "0.2"
__status__ = "dev"
from config import *
from functions import *
import os, sys
import time
from datetime import datetime
from termcolor import colored
from pybit.unified_trading import HTTP
from simple_telegram import *
import statistics
# get my secret LOCAL_CONFIG:
import socket
if socket.gethostname() == 'sereno':
from config_local import *
def side_colored(side): return colored(f"{side:4}", 'red' if side == 'Sell' else 'green')
def tp_colored(tp, l=7, d=2): return colored(f"{tp:<{l}.0{d}f}", 'cyan' if tp > 0 else None)
def pnl_colored(pnl, l=7, d=3, alarm=False): return colored(f"{pnl:<{l}.0{d}f}", 'light_red' if pnl < 0 else 'green', attrs=(['blink'] if alarm else None))
def liq_colored(liq, l=6, d=2, alarm=False): return colored(f"{liq:<{l}.0{d}f}", None if not liq else ('light_red' if alarm else 'yellow'), attrs=(['blink'] if alarm else None))
pnl_avg = {}
pnl_avg_len = 5 # collect avg for {n} last pnl values
alarms_lowest = {}
def main():
global alarms_lowest, pnl_avg, pnl_avg_len, min_PnL, min_LIQ
time_mark = datetime.today().strftime('%Y-%m-%d %H:%M:%S')
print(f'{" loading...":40}\r', end='', flush=True)
# GET POSITIONS (category: spot, linear, inverse, option)
try:
session = HTTP(api_key=api_key, api_secret=api_secret)
depo = session.get_wallet_balance(accountType="UNIFIED")['result']['list'][0] # , coin="BTC"
positions = session.get_positions(category="linear", settleCoin="USDT") # , symbol="XAIUSDT"
except Exception:
print('Sorry, read error, retry after sleep')
# send_to_telegram(TMapiToken, TMchatID, 'Connection lost, retry after sleep', print_exception=False)
return
# start printout
os.system('clear')
mark_alert = '. Alert if PnL'+colored(f"<{min_PnL}%", 'light_red') if min_PnL<0 else ""
mark_loss = '. Kill if PnL'+colored(f"<{min_Loss}%", 'light_red') if min_Loss<0 else ""
print(f'{time_mark} {__part__}{mark_alert}{mark_loss}.')
# deposit margin
atype = depo['accountType']
# wallet = float(depo['totalWalletBalance'])
marginbalance = float(depo['totalMarginBalance'])
#marginmaintenance = 100 * (float(depo['totalMaintenanceMargin']) / marginbalance)
marginini = float(depo['totalInitialMargin'])
margininipcnt = round(100 * marginini / marginbalance, 2)
margininipcntclr = 'light_red' if margininipcnt >= 50 else 'yellow' if margininipcnt >= 30 else 'cyan' if margininipcnt >= 10 else 'green'
marginavl = f"{round(marginbalance - marginini, 2):.02f}"
print(f"Margin: {round(marginbalance, 2):.02f}, Available: {colored(marginavl, margininipcntclr)}, Used: {colored(str(margininipcnt) + '%', margininipcntclr)}. All in USDT.")
pos_profit = 0
alarms_list = []
close_list = []
position_orders = []
# proceed positions
if positions and positions['result']['list']:
for pos in positions['result']['list']:
symbol = pos['symbol']
side = pos['side']
sidemark = side_colored(side)
sign = sign_sell if side == 'Sell' else sign_buy
symbolside = symbol + side
val = float(pos['positionValue'] or 0)
prc = float(pos['markPrice'] or 0)
lvrg = float(pos['leverage'] or 0) # leverage = плечо
rpnl = float(pos['curRealisedPnl'] or 0)
pnl = rpnl + float(pos['unrealisedPnl'] or 0) # includes start commission
liq = float(pos['liqPrice'] or 0)
created = int(pos['createdTime'])
posIdx = pos['positionIdx'] # needed to close pos on loss
pos_profit += pnl
# mark pnl change side, use avg per previos values
pnl_direction = colored('-', 'light_blue')
if symbolside in pnl_avg:
symbolside_pnl_avg = statistics.fmean(pnl_avg[symbolside])
if pnl != symbolside_pnl_avg:
pnl_direction = colored('↗', 'green') if pnl > symbolside_pnl_avg else colored('↘', 'light_red') # ↑↓
else:
pnl_avg[symbolside] = [] # init avg collection list
if len(pnl_avg[symbolside]) >= pnl_avg_len: # keep {n} last elements in list
pnl_avg[symbolside].pop(0) # remove oldest element
pnl_avg[symbolside].append(pnl) # save pnl as last element
# alarms, gather to `alarms` list then use it later to send Telegram message
liq_alarm = liq > 0 and abs(prc - liq)/max(prc, liq) < (min_LIQ/100) # liquidation alarm, use min_LIQ config
if liq_alarm:
alarms_list.append(f'{sign} {symbol} {min_LIQ}% to LIQUIDATION')
pnl_alarm = min_PnL * val / 100 # pnl alarm, use min_PnL config
if pnl < pnl_alarm:
if symbolside not in alarms_lowest:
alarms_lowest[symbolside] = pnl_alarm # first time
if pnl < alarms_lowest[symbolside]: # new minimum found! notify again
alarms_lowest[symbolside] = pnl
alarms_list.append(f'{sign} {symbol} PnL: {pnl}')
kill_loss = min_Loss * val / 100 # pnl alarm, use min_PnL config
if min_Loss < 0 and pnl < kill_loss:
close_list.append([symbol, side, posIdx, pnl])
PnL = pnl_colored(pnl, 8, 3, pnl < pnl_alarm)
liq = liq_colored(round(liq, 2), alarm=liq_alarm)
tp = tp_colored(round(float(pos['takeProfit'] or 0), 2))
sl = liq_colored(round(float(pos['stopLoss'] or 0), 2))
PnLp = round(100 * pnl / val, 1)
# gather positions here then sort by pnl and printout
position_orders.append([
pnl,
f"{sidemark} {symbol.replace('1000', '').lstrip('0'):10} PnL: {pnl_direction} {PnL} {PnLp:>4.1f}% VAL: {round(val, 2):<6.02f} PRC: {round(prc, 2):<6.02f} LIQ: {liq} TP: {tp} SL: {sl}"
])
# print out sorted, loosers first
position_orders.sort(key=lambda x: x[0])
i = 1
for order in position_orders:
print(f"{str(i):>2}. {order[1]}")
i += 1
print(f"TOTAL P&L: {pnl_colored(pos_profit, 8, 3)}")
# send alarms to Telegram
if alarms_list:
message = f'{sign_alarm} <b>Bybit Futures Alarm:</b>'
i = 1
for al in alarms_list:
message += f"\n{i}. {al}"
i += 1
send_to_telegram(TMapiToken, TMchatID, message)
# close loss positions and send to Telegram
if min_Loss < 0 and close_list:
message = "Close LOSS positions:"
print(f"\n{message}")
i = 1
for pos in close_list:
symbol, posside, positionIdx, pnl = pos
side = "Buy" if posside == 'Sell' else "Sell" # side contrary to position side
print(f"{i:2}. {side_colored(side)} {symbol} with PnL {pnl_colored(pnl)}... ", end='')
message += f"\n{i:2}. {side} {symbol} ... "
try:
################### kill position
session.place_order(category="linear", symbol=symbol, side=side, positionIdx=positionIdx, orderType="Market", qty="0", reduceOnly=True)
''' Azzrael, in according to a docs pass these 3:
qty=0.0, reduceOnly=True, closeOnTrigger=True
'''
###################
print(colored('closed', 'red'))
message += "closed"
except Exception as ex:
print(f'ERROR {ex.status_code}, position NOT closed!')
message += "ERROR!"
i += 1
send_to_telegram(TMapiToken, TMchatID, message)
del session # unset, I do not know if it keeps connection or what ever
print()
if __name__ == '__main__':
# send_to_telegram(TMapiToken, TMchatID, 'start')
while True:
main()
sys.stdout.flush()
try:
s = sleep_time
while s>0:
print(f' Sleep {s} sec... \r', end='')
key = input_timeout() # it makes sleep 1s
if key == 10 or key == 32: # reload now
break
elif key == 27: # exit
raise KeyboardInterrupt
s -= 1
except KeyboardInterrupt: # exit on Ctrl+C
print(f"\r{'Bye...':20}")
break
# that's all folks!