Binance LeaderBoard kullanıcılarının işlemlerine göre al sat yapan bot yazalım.

Konu başlığından da anlaşılacağı üzerine bugün yine bot yazıyoruz 🙂 Otomatik trade yapan botumuz iki bölümden oluşacak. Birincisi Binance Futures Top Traders kısmında bulunan bizim seçtiğimiz (haftalık PNL en yüksek ve takipçisi en fazla olanları seçtim) kişilerin işlemlerini anlık olarak telegram üzerinden bize bildirecek. İkincisi ise yine anlık bildirilen işlemler, yine bizim seçtiğimiz kişiler doğrultusunda otomatik olarak işleme girecek. Bir nevi copytrade mantığı olacak ama işin güzel tarafı şu, otomatik olarak işleme girme durumunu biz belirleyeceğiz. Yani istersek gelen bildirimler üzerinden değerlendirme yapıp kendimiz de işleme girebiliriz.

Öncelikle requirements.txt dosyamız ile kullanacağımız python kütüphanelerimizi betiğimize ekleyelim.

requirements.txt

pandas==2.0.3
python-telegram-bot==20.4
requests==2.28.2
configparser==6.0.0

Bir setup.py dosyası oluşturuyoruz ve çalıştırdığımız vakit telegram bot token ve chat id alanlarını gireceğimiz input değerlerini config.ini dosyamıza yazdırıyoruz.

setup.py

import os
import sys
import configparser

def banner():
    os.system('cls' if os.name == 'nt' else 'clear')
    print("Kurulum Ayarları")

banner()

config_file                     = 'config.ini'

telegram_info                   = configparser.RawConfigParser()
telegram_info.add_section('telegram')

xbot                            = input("[+] Telegram Bot Token: ")
telegram_info.set('telegram', 'bottoken', xbot)

xchat                           = input("[+] Telegram ChatID : ")
telegram_info.set('telegram', 'chatid', xchat)

with open(config_file, 'w') as setup:
    telegram_info.write(setup)

print("[+] Kurulum işlemi tamamlandı!")

Binance web üzerinden oluşturacağımız isteğimize ilişkin fonksiyon butiğimizi yazalım:

misc.py

def get_header(account_info_url: str):
    return {
        'authority'                     : 'www.binance.com',
        'accept'                        : '*/*',
        'accept-language'               : 'en-US,en;q=0.9',
        'bnc-uuid'                      : '0202c537-8c2b-463a-bdef-33761d21986a',
        'clienttype'                    : 'web',
        'csrftoken'                     : 'd41d8cd98f00b204e9800998ecf8427e',
        'device-info'                   : 'eyJzY3JlZW5fcmVzb2x1dGlvbiI6IjE5MjAsMTA4MCIsImF2YWlsYWJsZV9zY3JlZW5fcmVzb2x1dGlvbiI6IjE5MjAsMTA0MCIsInN5c3RlbV92ZXJzaW9uIjoiV2luZG93cyAxMCIsImJyYW5kX21vZGVsIjoidW5rbm93biIsInN5c3RlbV9sYW5nIjoicnUtUlUiLCJ0aW1lem9uZSI6IkdNVCszIiwidGltZXpvbmVPZmZzZXQiOi0xODAsInVzZXJfYWdlbnQiOiJNb3ppbGxhLzUuMCAoV2luZG93cyBOVCAxMC4wOyBXaW42NDsgeDY0KSBBcHBsZVdlYktpdC81MzcuMzYgKEtIVE1MLCBsaWtlIEdlY2tvKSBDaHJvbWUvMTAxLjAuNDk1MS42NyBTYWZhcmkvNTM3LjM2IiwibGlzdF9wbHVnaW4iOiJQREYgVmlld2VyLENocm9tZSBQREYgVmlld2VyLENocm9taXVtIFBERiBWaWV3ZXIsTWljcm9zb2Z0IEVkZ2UgUERGIFZpZXdlcixXZWJLaXQgYnVpbHQtaW4gUERGIiwiY2FudmFzX2NvZGUiOiI1ZjhkZDMyNCIsIndlYmdsX3ZlbmRvciI6Ikdvb2dsZSBJbmMuIChJbnRlbCkiLCJ3ZWJnbF9yZW5kZXJlciI6IkFOR0xFIChJbnRlbCwgSW50ZWwoUikgVUhEIEdyYXBoaWNzIDYyMCBEaXJlY3QzRDExIHZzXzVfMCBwc181XzAsIEQzRDExKSIsImF1ZGlvIjoiMTI0LjA0MzQ3NTI3NTE2MDc0IiwicGxhdGZvcm0iOiJXaW4zMiIsIndlYl90aW1lem9uZSI6IkV1cm9wZS9Nb3Njb3ciLCJkZXZpY2VfbmFtZSI6IkNocm9tZSBWMTAxLjAuNDk1MS42NyAoV2luZG93cykiLCJmaW5nZXJwcmludCI6IjE5YWFhZGNmMDI5ZTY1MzU3N2Q5OGYwMmE0NDE4Nzc5IiwiZGV2aWNlX2lkIjoiIiwicmVsYXRlZF9kZXZpY2VfaWRzIjoiMTY1MjY4OTg2NTQwMGdQNDg1VEtmWnVCeUhONDNCc2oifQ==',
        'fvideo-id'                     : '3214483f88c0abbba34e5ecf5edbeeca1e56e405',
        'lang'                          : 'en',
        'origin'                        : 'https://www.binance.com',
        'referer'                       : account_info_url,
        'sec-ch-ua'                     : '" Not A;Brand";v="99", "Chromium";v="101", "Google Chrome";v="101"',
        'sec-ch-ua-mobile'              : '?0',
        'sec-ch-ua-platform'            : '"Windows"',
        'sec-fetch-dest'                : 'empty',
        'sec-fetch-mode'                : 'cors',
        'sec-fetch-site'                : 'same-origin',
        'user-agent'                    : 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/101.0.4951.67 Safari/537.36',
        'x-trace-id'                    : 'e9d5223c-5d71-4834-8563-c253a1fc3ae8',
        'x-ui-request-trace'            : 'e9d5223c-5d71-4834-8563-c253a1fc3ae8',
    }

def get_json(uid: str):
    return     {
        'encryptedUid'                  : uid,
        'tradeType'                     : 'PERPETUAL',
}

Daha sonra telegram üzerinden mesaj gelmesi sağlayan kodumuzu yazalım:

message.py

import requests
import os
import sys
import configparser

telegram_info                       = configparser.RawConfigParser()
telegram_info.read('config.ini')

try:
    telegram_bot_token              = telegram_info['telegram']['bottoken']
    telegram_chatid                 = telegram_info['telegram']['chatid']
except KeyError:
    os.system("cls" if os.name == "nt" else "clear")
    print("Run setup.py first")
    sys.exit(1)

def telegram_send_message(message):
    apiURL                          = f'https://api.telegram.org/bot{telegram_bot_token}/sendMessage'
    try:
        response                    = requests.post(apiURL, json={'chat_id': telegram_chatid, 'text': message, 'parse_mode': 'html', 'disable_web_page_preview': True})
        print(response.text)
    except Exception as e:
        print(e)

binance futures üzerinden işleme giren ve sıralamada yer alan kullanıcıların kullanıcı adlarını, girdiği pozisyonları ve fiyatları çeken kodumuzu yazıyoruz.

binance.py

import requests
import time
from message import telegram_send_message

def get_position(headers, json_data, max_retries=5):
    retry_count                     = 0
    while retry_count <= max_retries:
        try:
            return requests.post("https://www.binance.com/bapi/futures/v1/public/future/leaderboard/getOtherPosition",
                                 headers=headers, json=json_data)
        except requests.exceptions.ConnectionError as e:
            print(f"Connection error occurred: {e}")
            telegram_send_message(f"Connection error occurred: {e}")
            if retry_count >= max_retries:
                telegram_send_message("Max retry count reached. Waiting for 10 minutes before next try...")
                time.sleep(600)
                retry_count         = 0
            else:
                print("Retrying in 5 seconds...")
                time.sleep(5)
                retry_count        += 1

def get_nickname(headers, json_data, max_retries=5):
    retry_count                     = 0
    while retry_count <= max_retries:
        try:
            return requests.post("https://www.binance.com/bapi/futures/v2/public/future/leaderboard/getOtherLeaderboardBaseInfo",
                                 headers=headers, json=json_data)
        except requests.exceptions.ConnectionError as e:
            print(f"Connection error occurred: {e}")
            telegram_send_message(f"Connection error occurred: {e}")
            if retry_count >= max_retries:
                telegram_send_message("Max retry count reached. Waiting for 10 minutes before next try...")
                time.sleep(600)
                retry_count         = 0
            else:
                print("Retrying in 5 seconds...")
                time.sleep(5)
                retry_count        += 1

def get_markprice(symbol):
    api_url                         = "https://fapi.binance.com/fapi/v1/premiumIndex"
    req_data                        = requests.get(api_url, params={"symbol": symbol})
    try:
        data                        = req_data.json()
        return data['markPrice']
    except Exception:
        return "Market price retrieval error"

son olarak main.py kodumuz ile botumuzu bitiriyoruz.

import pandas as pd
import time
import json
import requests
import datetime
from misc import get_header, get_json
from message import telegram_send_message
from binance import get_position, get_nickname, get_markprice

# Binance User ID alanı, buraya katip etmek istediğimiz kullanıcıların id lerini çekiyoruz.
TARGETED_ACCOUNT_UIDs                = [
                                        "151722250827C6026238FE5B77213AB7",
                                        "899D809A6394484913659EFBA0DB5595",
                                        "7D727A8B66302DBEB2528287D49B5528",
                                        "0280FF0272DE98EB7739B1A9ABF53660",
                                        "F7F33ED8C5115D28E772A48E9B181C62"
                                        ]

ACCOUNT_INFO_URL_TEMPLATE           = 'https://www.binance.com/en/futures-activity/leaderboard/user?encryptedUid={}'

# USDT cinsinden tahmini giriş boyutunun hesaplanması da dahil olmak üzere DataFrame'in değiştirilmesi
def modify_data(data) -> pd.DataFrame:
    df                              = pd.DataFrame(data)
    position                        = pd.DataFrame(df['data'][0]).set_index('symbol')
    position['estimatedEntrySize']  = round((abs(position['amount']) / position['leverage']) * position['entryPrice'], 2)
    position['pnl']                 = round(position['pnl'], 2)
    position.loc[(position['amount'] > 0), 'estimatedPosition'] = 'LONG'
    position.loc[(position['amount'] < 0), 'estimatedPosition'] = 'SHORT'
    position['updateTime']          = position['updateTime'].apply(lambda x: datetime.datetime(*x[:-1], x[-1] // 1000))
    position['updateTime']          = position['updateTime'].dt.strftime('%Y-%m-%d %H:%M:%S')
    position_result                 = position[['estimatedPosition', 'leverage', 'estimatedEntrySize', 'amount',
                                                'entryPrice', 'markPrice', 'pnl', 'updateTime']]
    return position_result

previous_symbols                    = {}
previous_position_results           = {}
is_first_runs                       = {uid: True for uid in TARGETED_ACCOUNT_UIDs}

def send_new_position_message(symbol, row, nickname):
    estimated_position              = row['estimatedPosition']
    leverage                        = row['leverage']
    estimated_entry_size            = row['estimatedEntrySize']
    entry_price                     = row['entryPrice']
    pnl                             = row['pnl']
    updatetime                      = row['updateTime']
    message                         = f"[<b>{nickname}</b>]\n<b>Yeni pozisyon açıldı!</b>\n\n" \
                                      f"Pozisyonu: <b>{symbol} {estimated_position} {leverage}X</b>\n\n" \
                                      f"Base currency - USDT\n" \
                                      f"------------------------------\n" \
                                      f"Giriş Fiyatı: {entry_price}\n" \
                                      f"Est. Büyüklüğü: {estimated_entry_size}\n" \
                                      f"PnL: {pnl}\n\n" \
                                      f"Son güncelleme:\n{updatetime} (UTC+0)\n" \
                                      f"<a href='{ACCOUNT_INFO_URL}'><b>Profil Adresi</b></a>"
    telegram_send_message(message)

def send_closed_position_message(symbol, row, nickname):
    estimated_position              = row['estimatedPosition']
    leverage                        = row['leverage']
    updatetime                      = row['updateTime']
    message                         = f"[<b>{nickname}</b>]\n<b>Pozisyon kapandı!</b>\n\n" \
                                      f"Pozisyon: <b>{symbol} {estimated_position} {leverage}X</b>\n" \
                                      f"Fiyat: {get_markprice(symbol)} USDT\n\n" \
                                      f"Son güncelleme:\n{updatetime} (UTC+0)\n" \
                                      f"<a href='{ACCOUNT_INFO_URL}'><b>Profil adresi</b></a>"
    telegram_send_message(message)

def send_current_positions(position_result, nickname):
    if position_result.empty:
        telegram_send_message(f"[<b>{nickname}</b>]\n<b>Açık pozisyon yok</b>")
    else:
        telegram_send_message(f"[<b>{nickname}</b>]\n<b>Pozisyonlar:</b>")
        for symbol, row in position_result.iterrows():
            estimated_position      = row['estimatedPosition']
            leverage                = row['leverage']
            estimated_entry_size    = row['estimatedEntrySize']
            entry_price             = row['entryPrice']
            pnl                     = row['pnl']
            updatetime              = row['updateTime']
            message                 = f"Pozisyon: <b>{symbol} {estimated_position} {leverage}X</b>\n\n" \
                                      f"Base currency - USDT\n" \
                                      f"------------------------------\n" \
                                      f"Giriş fiyatı: {entry_price}\n" \
                                      f"Est. Büyüklüğü: {estimated_entry_size}\n" \
                                      f"PnL: {pnl}\n\n" \
                                      f"Son güncelleme:\n{updatetime} (UTC+0)\n" \
                                      f"<a href='{ACCOUNT_INFO_URL}'><b>Profil adresi</b></a>"
            telegram_send_message(message)

while True:
    try:
        for TARGETED_ACCOUNT_UID in TARGETED_ACCOUNT_UIDs:
            ACCOUNT_INFO_URL        = ACCOUNT_INFO_URL_TEMPLATE.format(TARGETED_ACCOUNT_UID)
            headers                 = get_header(ACCOUNT_INFO_URL)
            json_data               = get_json(TARGETED_ACCOUNT_UID)

            response_nickname       = get_nickname(headers, json_data)
            response                = get_position(headers, json_data)
            if response is not None and response_nickname is not None:
                nickname            = json.loads(response_nickname.text)['data']['nickName']
                leaderboard_data    = json.loads(response.text)
                position_result     = modify_data(leaderboard_data)

                new_symbols         = position_result.index.difference(previous_symbols.get(TARGETED_ACCOUNT_UID, pd.Index([])))
                if not is_first_runs[TARGETED_ACCOUNT_UID] and not new_symbols.empty:
                    for symbol in new_symbols:
                        send_new_position_message(symbol, position_result.loc[symbol], nickname)

                closed_symbols      = previous_symbols.get(TARGETED_ACCOUNT_UID, pd.Index([])).difference(position_result.index)
                if not is_first_runs[TARGETED_ACCOUNT_UID] and not closed_symbols.empty:
                    for symbol in closed_symbols:
                        if symbol in previous_position_results.get(TARGETED_ACCOUNT_UID, pd.DataFrame()).index:
                            send_closed_position_message(symbol, previous_position_results[TARGETED_ACCOUNT_UID].loc[symbol], nickname)

                if is_first_runs[TARGETED_ACCOUNT_UID]:
                    send_current_positions(position_result, nickname)

                previous_position_results[TARGETED_ACCOUNT_UID] = position_result.copy()
                previous_symbols[TARGETED_ACCOUNT_UID] = position_result.index.copy()
                is_first_runs[TARGETED_ACCOUNT_UID] = False
                
        time.sleep(300)
    except Exception as e:
        print(f"Error occurred: {e}")
        message                     = f"Error occurred for UID <b>{TARGETED_ACCOUNT_UID}</b>:\n{e}\n\n" \
                                      f"Retrying after 60s"
        telegram_send_message(message)
        time.sleep(60)

Örnek çıktılarımız:

147220cookie-checkBinance LeaderBoard kullanıcılarının işlemlerine göre al sat yapan bot yazalım.yes