-
Notifications
You must be signed in to change notification settings - Fork 3
/
test.py
50 lines (36 loc) · 1.24 KB
/
test.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
import httpx
import time
from flask import Flask
import threading
class TokenTracker:
def __init__(self):
self.previous_tokens = set()
self.new_tokens = set()
self.printtokens = set()
self.r = 1
def getraydium(self):
response = httpx.get("https://api.raydium.io/v2/sdk/token/raydium.mainnet.json")
if response.status_code == 200:
return [j["mint"] for j in response.json()["unNamed"]]
else:
print("ERR")
return None
def track(self):
while True:
time.sleep(15)
current_tokens = set(self.getraydium())
self.new_tokens = current_tokens - self.previous_tokens
if self.new_tokens:
self.printtokens = self.new_tokens.copy()
self.r += 1
print(f"New tokens added {self.new_tokens}")
self.previous_tokens = current_tokens
tr = TokenTracker()
tracking_thread = threading.Thread(target=tr.track)
tracking_thread.start()
app = Flask(__name__)
@app.route("/new_tokens", methods=["GET"])
def get_new_tokens():
return list(tr.printtokens), 200
if __name__ == "__main__":
app.run(port=5000)