Bir ara meraktan binance'de herhangi bir pump olduğunu görebileceğim bir ürün arıyordum bununla ilgili web siteleri var ama onlarda ücretliydi bende ai'lara gidip kendime bir detektör hazırladım bayağı bir zamandırda ufak tefek konular için açar bakarım
Hafıza : Son 5 dakikalık (WINDOW_SECONDS = 300) veriyi hafızasında tutar. Mevcut fiyatı 5 dakika önceki fiyatla kıyaslar.
Fiyat Eşiği: Eğer fiyat 5 dakika içinde %3'ten fazla (PRICE_PCT_THRESHOLD) artmışsa takibe alır.
Hacim Eşiği: Sadece fiyat artışı yetmez; aynı zamanda o anki işlem hacminin, önceki ortalamanın 3 katından fazla (VOLUME_MULTIPLIER) olması gerekir. Bu, artışın "gerçek bir alım dalgası" olup olmadığını anlamaya yarar.
#!/usr/bin/env python3
"""
binance_pump_detector.py
Basit, pip gerektirmeyen bir Binance pump algılama scripti.
- Public Binance endpoint /api/v3/ticker/24hr kullanır (API anahtarı gerekmez)
- Sadece USDT paritelerini izler
- Fiyat artışı ve anlık hacim (quoteVolume delta) artışına göre alarm verir
Kullanım:
python3 binance_pump_detector.py
Konfigürasyon satırlarını değiştirerek eşikleri ayarlayın.
"""
import time
import urllib.request
import ssl
import json
from collections import deque, defaultdict
from statistics import mean
# ========== CONFIG ==========
POLL_INTERVAL = 15 # saniye başına kaç kez çekilsin
WINDOW_SECONDS = 300 # kaç saniyelik geçmişe bakılsın (ör: 300 = 5 dakika)
PRICE_PCT_THRESHOLD = 3.0 # bu süre içinde fiyat % kaç artarsa pump sayılacak
VOLUME_MULTIPLIER = 3.0 # anlık hacim (quote delta) önceki ortalamanın kaç katı olmalı
MIN_QUOTE_VOLUME = 2000.0 # anahtar: $ cinsinden minimum quote hacmi (küçük tokenlar için aşağı çekilebilir)
MAX_SYMBOLS = 1000 # güvenlik için bir sınır
BINANCE_24HR_URL = '
https://api.binance.com/api/v3/ticker/24hr'
# SSL doğrulama bypass (bazı ortamlarda self-signed sertifika zinciri hatasını aşmak için)
ssl_ctx = ssl.create_default_context()
ssl_ctx.check_hostname = False
ssl_ctx.verify_mode = ssl.CERT_NONE
# ANSI renk kodları
RED = "\033[91m"
GREEN = "\033[92m"
YELLOW = "\033[93m"
CYAN = "\033[96m"
RESET = "\033[0m"
# ========== Veri yapıları ==========
# Her sembol için geçmişteki (timestamp, price, quoteVolume_cumulative) kayıtlarını tutacağız
history = defaultdict(lambda: deque())
# Her sembol için son N "delta quoteVolume" kayıtları (per poll) - ortalamanın hesabı için
delta_queues = defaultdict(lambda: deque(maxlen=10))
# -------- Helper: HTTP GET ve JSON parse (basit, requests gerektirmez) --------
def fetch_24hr_all():
req = urllib.request.Request(BINANCE_24HR_URL, headers={'User-Agent': 'pump-detector/1.0'})
with urllib.request.urlopen(req, timeout=20, context=ssl_ctx) as resp:
data = resp.read()
return json.loads(data)
# -------- Pump detection logic --------
def process_snapshot(snapshot, now_ts):
count = 0
alerts = []
for item in snapshot:
symbol = item.get('symbol')
if not symbol or not symbol.endswith('USDT'):
continue
try:
price = float(item.get('lastPrice', 0))
quote_cum = float(item.get('quoteVolume', 0))
except Exception:
continue
h = history[symbol]
h.append((now_ts, price, quote_cum))
while h and (now_ts - h[0][0]) > WINDOW_SECONDS:
h.popleft()
if len(h) >= 2:
t_old, price_old, quote_old = h[0]
if price_old <= 0:
continue
price_pct = (price - price_old) / price_old * 100.0
quote_delta = quote_cum - quote_old
if quote_delta < 0:
quote_delta = 0.0
if len(h) >= 3:
_, _, quote_prev_cum = h[-2]
last_poll_delta = quote_cum - quote_prev_cum
if last_poll_delta < 0:
last_poll_delta = 0.0
dq = delta_queues[symbol]
if last_poll_delta > 0:
dq.append(last_poll_delta)
prev_mean = mean(dq) if dq else 0.0
else:
prev_mean = 0.0
if prev_mean < 1e-6:
multiplier = float('inf') if quote_delta >= MIN_QUOTE_VOLUME else 0.0
else:
multiplier = (quote_delta / prev_mean) if prev_mean > 0 else 0.0
is_price_ok = price_pct >= PRICE_PCT_THRESHOLD
is_volume_ok = (multiplier >= VOLUME_MULTIPLIER) and (quote_delta >= MIN_QUOTE_VOLUME)
if is_price_ok and is_volume_ok:
alerts.append({
'symbol': symbol,
'price_old': price_old,
'price_now': price,
'price_pct': round(price_pct, 2),
'quote_delta': round(quote_delta, 2),
'multiplier': float('inf') if prev_mean==0 else round(multiplier, 2),
'timestamp': now_ts,
})
count += 1
if count >= MAX_SYMBOLS:
break
return alerts
# -------- Alert output --------
def print_alerts(alerts):
for a in alerts:
print(f"{RED}*** PUMP DETECTED ***{RESET}")
print(f"{RED}{a['symbol']}{RESET} | {YELLOW}{a['price_old']}{RESET} -> {GREEN}{a['price_now']}{RESET} | %: {GREEN}{a['price_pct']}{RESET} | quote_delta: {YELLOW}{a['quote_delta']}{RESET} | mult: {YELLOW}{a['multiplier']}{RESET}")
print(f"{RED}---------------------------{RESET}")
# ========== Main loop ==========
def main():
print(f"{CYAN}Starting Binance USDT pump detector{RESET}")
print(f"Poll interval: {POLL_INTERVAL}s, window: {WINDOW_SECONDS}s")
while True:
now_ts = int(time.time())
try:
snapshot = fetch_24hr_all()
except Exception as e:
print(f"{RED}Fetch error:{RESET}", e)
time.sleep(5)
continue
alerts = process_snapshot(snapshot, now_ts)
if alerts:
print_alerts(alerts)
time.sleep(POLL_INTERVAL)
if __name__ == '__main__':
try:
main()
except KeyboardInterrupt:
print(f"{YELLOW}\nExiting on user interrupt{RESET}")
