-
-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathcrypto_arbitrage_tracker_alert.py
More file actions
127 lines (111 loc) · 4.06 KB
/
Copy pathcrypto_arbitrage_tracker_alert.py
File metadata and controls
127 lines (111 loc) · 4.06 KB
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
import requests
import time
import os
from datetime import datetime
def clear_terminal():
os.system('cls' if os.name == 'nt' else 'clear')
def display_header():
yellow = "\033[93m"
red = "\033[91m"
turquoise = "\033[96m"
reset = "\033[0m"
date = datetime.utcnow().strftime("%Y-%m-%d")
time_utc = datetime.utcnow().strftime("%H:%M UTC")
print(yellow + "=" * 95)
print("ARBITRAGE PRICE TRACKER | Monitor crypto exchange prices and spot opportunities in real-time!")
print("=" * 95)
print(red + "DISCLAIMER: This software is for informational purposes only. Use at your own risk.\n"
"Crypto prices may differ between exchanges. Always verify before trading." + yellow)
print("=" * 95)
print(turquoise + f"DATE: {date} | TIME: {time_utc}" + yellow)
print("=" * 95)
print(f"{'Cryptocurrency':<18} | {'Exchange':<12} | {'Price (USD)':<15}")
print("=" * 95 + reset)
def get_top_50_cryptos():
try:
url = "https://api.coingecko.com/api/v3/coins/markets"
params = {
"vs_currency": "usd",
"order": "market_cap_desc",
"per_page": 50,
"page": 1,
"sparkline": "false"
}
response = requests.get(url, params=params)
return response.json()
except:
return []
def get_binance_price(symbol):
try:
url = f"https://api.binance.com/api/v3/ticker/price?symbol={symbol}USDT"
data = requests.get(url).json()
return float(data['price']) if 'price' in data else None
except:
return None
def get_coinbase_price(symbol):
try:
url = f"https://api.coinbase.com/v2/prices/{symbol}-USD/spot"
data = requests.get(url).json()
return float(data['data']['amount']) if 'data' in data else None
except:
return None
def get_kraken_price(symbol):
try:
special_cases = {"BTC": "XBT", "DOGE": "XDG", "USDT": "USDTZ"}
kraken_symbol = special_cases.get(symbol, symbol)
url = f"https://api.kraken.com/0/public/Ticker?pair={kraken_symbol}USD"
data = requests.get(url).json()
if 'result' in data and len(data['result']) > 0:
pair = list(data['result'].keys())[0]
return float(data['result'][pair]['c'][0])
return None
except:
return None
def detect_arbitrage(prices, name):
values = [v for v in prices.values() if v is not None and 0.01 < v < 100000]
if len(values) < 2:
return
min_price = min(values)
max_price = max(values)
diff = (max_price - min_price) / min_price
if diff >= 0.5:
print(f"\033[93m[⚠️ POSSIBLE ERROR] Check data for {name} | "
f"Buy @ ${min_price:.2f}, Sell @ ${max_price:.2f} | Spread: {diff*100:.2f}%\033[0m")
elif diff >= 0.01:
print(f"\033[92m[ALERT] Arbitrage Opportunity: {name} | "
f"Buy @ ${min_price:.2f}, Sell @ ${max_price:.2f} | Spread: {diff*100:.2f}%\033[0m")
def display_crypto(crypto):
name = crypto['name']
symbol = crypto['symbol'].upper()
binance_price = get_binance_price(symbol)
coinbase_price = get_coinbase_price(symbol)
kraken_price = get_kraken_price(symbol)
if not any([binance_price, coinbase_price, kraken_price]):
return
print(f"{name:<18}")
if binance_price:
print(f"{'':<18} | Binance | {binance_price:<15.4f} USD")
if coinbase_price:
print(f"{'':<18} | Coinbase | {coinbase_price:<15.4f} USD")
if kraken_price:
print(f"{'':<18} | Kraken | {kraken_price:<15.4f} USD")
detect_arbitrage({
"Binance": binance_price,
"Coinbase": coinbase_price,
"Kraken": kraken_price
}, name)
print("-" * 95)
time.sleep(0.66)
def main():
cryptos = get_top_50_cryptos()
total = len(cryptos)
batch_size = 10
while True:
for i in range(0, total, batch_size):
clear_terminal()
display_header()
for crypto in cryptos[i:i+batch_size]:
display_crypto(crypto)
time.sleep(2)
if __name__ == "__main__":
main()