Автоматичне надсилання повідомлень у P2P-чат Binance через WebSocket на Python

Надсилання повідомлень продавцю про переказ коштів у P2P-чаті Binance. Підключення через WebSocket та автоматизація на Python.

У світі автоматизації криптовалютної торгівлі важливою частиною залишається взаємодія з контрагентами.

У цій статті ми розглянемо, як отримати облікові дані чату, підключитися до WebSocket і надіслати повідомлення до P2P-чату Binance, повідомляючи про переказ коштів.

🔐 Отримання облікових даних для WebSocket чату

Для початку нам потрібно одержати chatWssUrl, listenKey та listenToken, щоб підключитися до WebSocket-каналу.

📦 Функція `retrieveChatCredential`


def retrieveChatCredential():
    # You Api Key on binance
    api_key = "11111"
    # You Apy Secret on binance
    secret_key = "11111"
    
    base_url = "https://api.binance.com"

    endpoint = "/sapi/v1/c2c/chat/retrieveChatCredential"
    timestamp = int(time.time() * 1000)
    query_string = f"timestamp={timestamp}"

    signature = hmac.new(
        secret_key.encode('utf-8'),
        query_string.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()

    headers = {
        "clientType": "web",
        "X-MBX-APIKEY": api_key
    }

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"
    response = requests.get(url, headers=headers)

    if response.status_code == 200:
        return response.json()
    else:
        print("Error:", response.status_code, response.text)
        return None

Після отримання даних формується URL-адреса підключення:

Ми використовуємо бібліотеку websocket-client, щоб встановити з'єднання та отримувати події в реальному часі з P2P-чату.


chatWssUrl = res_detail['data']['chatWssUrl']
listenKey = res_detail['data']['listenKey']
listenToken = res_detail['data']['listenToken']
chat_wss_url = f"{chatWssUrl}/{listenKey}?token={listenToken}&clientType=web"

🔌 Встановлення WebSocket-з'єднання


def connect_to_websocket():
    global reconnect_attempts, ws
    reconnect_attempts = 0
    ws = websocket.WebSocketApp(
        chat_wss_url,
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    ws.run_forever()

📡 Обробники подій

on_open — підключення встановлено
on_message — надійшло повідомлення
on_error — помилка з'єднання
on_close — закриття з'єднання та автоматичний reconnect

✉️ Надсилання повідомлення в чат

Коли настає потрібний момент (наприклад, після переказу коштів продавцю), викликається функція:


def send_message(content, order_no):
    if ws_connection:
     # ============================================================
     # 💖 Donation Notice
     #
     # English:
     # ⚠️ Full functionality is available only after making a donation.
     # After your donation, provide the transaction hash and we will unlock the feature for you.
     #
     # Spanish:
     # ⚠️ La funcionalidad completa está disponible solo después de realizar una donación.
     # Después de tu donación, proporciona el hash de la transacción y desbloquearemos la función.
     # ============================================================

        try:
           ws_connection.send(json.dumps(response_message))
           print(f"Message sent to chat: {content}")
           log.logging(f"Message sent payload: {json.dumps(response_message)}")

        except Exception as e:
            log.logging(f"Message sending error: {str(e)}")

Приклад виклику:


send_message("Вітаю, гроші переведені. Будь ласка, відпустіть крипту.", "789456123ABCDEF")

Повний код реалізації:


import uuid
import websocket
import time
import hmac
import hashlib
import requests
import json

# You Api Key on binance
api_key = "11111"
# You Apy Secret on binance
secret_key = "11111"

def retrieveChatCredential():
    
    base_url = "https://api.binance.com"
    endpoint = "/sapi/v1/c2c/chat/retrieveChatCredential"
    timestamp = int(time.time() * 1000)
    query_string = f"timestamp={timestamp}"
    signature = hmac.new(
        secret_key.encode('utf-8'),
        query_string.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()

    headers = {
        "clientType": "web",
        "X-MBX-APIKEY": api_key
    }

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"
    response = requests.get(url, headers=headers)

    if response.status_code == 200:
        return response.json()
    else:
        print("Error:", response.status_code, response.text)
        return None

res_detail = retrieveChatCredential()

if res_detail:
    chatWssUrl = res_detail['data']['chatWssUrl']
    listenKey = res_detail['data']['listenKey']
    listenToken = res_detail['data']['listenToken']

    chat_wss_url = f"{chatWssUrl}/{listenKey}?token={listenToken}&clientType=web"
    #print("WebSocket URL:", chat_wss_url)
else:
    raise Exception("Failed to obtain credentials for WebSocket.")

def generate_uuid():
    return str(uuid.uuid4())

def send_message(content, order_no):
    if ws_connection:
     # ============================================================
     # 💖 Donation Notice
     #
     # English:
     # ⚠️ Full functionality is available only after making a donation.
     # After your donation, provide the transaction hash and we will unlock the feature for you.
     #
     # Spanish:
     # ⚠️ La funcionalidad completa está disponible solo después de realizar una donación.
     # Después de tu donación, proporciona el hash de la transacción y desbloquearemos la función.
     # ============================================================


        try:
           ws_connection.send(json.dumps(response_message))
           print(f"Message sent to chat: {content}  for orderNo: {order_no} ")

        except Exception as e:
             print(f"Message sent error: {str(e)}")

def on_open(ws):
    global ws_connection
    ws_connection = ws
    print("WebSocket connection  established for Chat")


def on_message(ws, message):
    try:
        parsed_message = json.loads(message)
        print(f"parsed_message: {parsed_message}")
    except json.JSONDecodeError:
        print("Error JSON:", message)
 

def on_error(ws, error):
    print("Error WebSocket:", error)

def on_close(ws, close_status_code, close_msg):
    global reconnect_attempts
    print("WebSocket connection closed.", close_msg)
   
    if reconnect_attempts < 5:
        reconnect_attempts += 1
        print(f"Attempting to reconnect ({reconnect_attempts})...")
        time.sleep(5)
        connect_to_websocket()
    else:
        print("The maximum number of reconnection attempts has been reached.")

reconnect_attempts = 0
ws_connection = None

def connect_to_websocket():
    global reconnect_attempts, ws
    reconnect_attempts = 0
    ws = websocket.WebSocketApp(
        chat_wss_url,
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    ws.run_forever()

connect_to_websocket()

🤖 Переваги автоматизації чату

📈 Прискорення угод — продавець швидше отримає повідомлення.
🔒 Безпека — менше ручних дій, менше помилок.
⏱ Економія часу — особливо при великому обсязі угод.
🧠 Інтеграція з ботами — логіка може бути вбудована у вашого P2P-бота.

🧠 Практичне застосування

Така автоматизація особливо корисна під час масштабних операцій або якщо ви працюєте з кількома ордерами. Бот може моментально повідомити продавця, що оплату відправлено, та скоротити загальний час завершення угоди.

💡 Готове рішення

У нас є готовий продукт - P2Pay Binance Bot, в якому вже реалізований необхідний функціонал. Цей бот автоматично відстежує статуси замовлень, виконує виплати продавцям криптовалюти на банківські рахунки (картки) згідно з умовами ордера та одночасно надсилає повідомлення та повідомлення в чат.

Таким чином, P2Pay Binance Bot повністю автоматизує процес обробки угод та комунікацію з учасниками, забезпечуючи зручність та безпеку.

📌 Висновок

Автоматизація P2P-чату Binance за допомогою Python і WebSocket - це інструмент для розробників, які створюють ботів або автоматизують торгові процеси. Отримавши chatWssUrl та listenToken, Ви можете підключатися до чату, слухати події та надсилати повідомлення, повідомляючи контрагентів про переклади.