CryptoBot ESP32 Micropython

CryptoBot

 

Данные берутся с биржи QUTRADE.

В будущем будут добавленны и другие биржи.

Для нормальной работы надо прописать свой SSID и пароль (Настройки Wi-Fi).

WIFI_SSID = "My_network"
WIFI_PASS = "112233344" 

Частота обновления раз в 5 секунд настраивается параметром CHECK_INTERVAL . 

Для начала торгов надо получить api_key и secret_key после регистрации на бирже  QUTRADE.

Принцип работы бота , при изменении цены на 1% бот выставляет ордер на покупку и ордер на прдажу +-1%. 

main.py     

import urequests as requests
import ujson as json
import time

from led_driver import print_like_console , print_multiline_console

import uhashlib as hashlib
import ure
import ubinascii

WIFI_SSID = "My_network"
WIFI_PASS = "112233344" 

import network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(WIFI_SSID, WIFI_PASS)
while not wlan.isconnected():
    pass
print("Wi-Fi OK", wlan.ifconfig())
print_like_console('Wi-Fi OK', x=0, y=0, delay=0.05)



# === Настройки ===
base_url = 'https://qutrade.io'
method = '/api/v1/new_order/'
api_key = 'KEY'
secret_key = 'SECRET'

def urlencode(params):
    """Упрощённый urlencode для MicroPython"""
    out = []
    for k, v in params.items():
        out.append(f"{k}={v}")
    return "&".join(out)

def hmac_sha256(key, msg):
    """HMAC-SHA256 без модуля hmac (для MicroPython)"""
    block_size = 64  # SHA256 block size
    if len(key) > block_size:
        key = hashlib.sha256(key).digest()
    key = key + b'\x00' * (block_size - len(key))
    
    o_key_pad = bytes([b ^ 0x5c for b in key])
    i_key_pad = bytes([b ^ 0x36 for b in key])
    
    inner = hashlib.sha256(i_key_pad + msg).digest()
    hmac_result = hashlib.sha256(o_key_pad + inner).digest()
    return ubinascii.hexlify(hmac_result).decode()


last_tonce = 0  # глобальная переменная

def get_tonce():
    """Генерирует уникальный tonce для каждого запроса."""
    global last_tonce
    new_tonce = round(time.time() * 1000)
    if new_tonce <= last_tonce:
        new_tonce = last_tonce + 1
    last_tonce = new_tonce
    return new_tonce


def qtrader(pair, side, amo, price):
    print(f"[TRADE] {pair} {side.upper()} {amo:.4f} @ {price:.6f}")
    params = {
        'pair': pair,
        'type': 'limit',
        'side': side,
        'amount': amo,
        'price': price,
        'tonce': get_tonce()
    }
    params_string = urlencode(params)
    signature = hmac_sha256(secret_key.encode(), params_string.encode())

    headers = {
        'API-Key': api_key,
        'API-Sign': signature,
        'Content-Type': 'application/x-www-form-urlencoded'
    }

    try:
        r = requests.post(base_url + '/api/v1/new_order/', data=params_string, headers=headers)
        print(f"{side.upper()} {pair} {amo} @ {price}")
        print("Ответ:", r.text)
        r.close()
    except Exception as e:
        print("Ошибка qtrader:", e)


# === Настройки ===
PAIRS = ["matic_usdt"]            # пара
MIN_ORDER = 0.2                   # минимальный объем ордера в USDT
VOLATILITY_THRESHOLD = 0.01       # порог изменения (1%)
CHECK_INTERVAL = 5                # пауза между проверками


# === Получение данных о парах ===
def get_datas(pairs):
    base_url = 'https://qutrade.io/api/v1/market_data/'
    params = 'pair=' + ','.join(pairs)
    try:
        r = requests.get(base_url + '?' + params)
        data = r.json()
        r.close()
    except Exception as e:
        print("Ошибка при запросе:", e)
        return {}

    result = {}
    if "list" not in data:
        print("Некорректный ответ:", data)
        return {}

    for pair in pairs:
        try:
            info = data["list"][pair]
            price = float(info["price"])
            bid = float(info["bid"])
            ask = float(info["ask"])
            result[pair] = {"price": price, "bid": bid, "ask": ask}
        except Exception:
            continue

    return result

# === Подсчёт процента изменения ===
def percc(new, old):
    return (new - old) / old * 100

# === Инициализация ===
print("Запуск торгового бота...")
print_like_console('Starting', x=0, y=0, delay=0.05)
last_buy = {}
last_sell = {}
prices = {}

data = get_datas(PAIRS)
for p in data:
    prices[p] = data[p]["price"]
    last_buy[p] = prices[p]
    last_sell[p] = prices[p]
    print(f"{p}: {prices[p]} USDT")

time.sleep(2)
rounds = 0


# === Основной цикл ===
while True:
    try:
        lines = []
        y = get_datas(PAIRS)
        for p in y.keys():
            lp = y[p]["price"]

            if lp == last_buy[p] or lp == last_sell[p]:
                print(f"[{p}] цена стабильно: {lp}")
                #print_like_console(f"[{p}] : {lp}", x=0, y=0, delay=0.05)
                lines.append(f"{p.split('_')[0]} : {lp}")
                continue

            # если цена упала
            if lp < last_buy[p] * (1 - VOLATILITY_THRESHOLD):
                print(f"🔴 Цена падает {p}: {lp}")
                down = lp * (1 - VOLATILITY_THRESHOLD)
                up = lp * (1 + VOLATILITY_THRESHOLD)
                last_buy[p] = down
                last_sell[p] = up
                amo = MIN_ORDER / down
                qtrader(p, "buy", amo, down)
                qtrader(p, "sell", amo, up)
                lines.append(f"[B] {p} @ {down:.6f}")
                lines.append(f"[S] {p} @ {up:.6f}")
                rounds = 0

            # если цена выросла
            elif lp > last_sell[p] * (1 + VOLATILITY_THRESHOLD):
                print(f"🟢 Цена растёт {p}: {lp}")
                down = lp * (1 - VOLATILITY_THRESHOLD)
                up = lp * (1 + VOLATILITY_THRESHOLD)
                last_buy[p] = down
                last_sell[p] = up
                amo = MIN_ORDER / down
                qtrader(p, "buy", amo, down)
                qtrader(p, "sell", amo, up)
                lines.append(f"[B] {p} @ {down:.6f}")
                lines.append(f"[S] {p} @ {up:.6f}")
                rounds = 0

        rounds += 1
        print(f"Round {rounds} завершён\n")
        #print_like_console(f"Round {rounds}", x=0, y=0, delay=0.05)
        lines.append(f"Round {rounds}")
        print_multiline_console(lines, 0.05)
        time.sleep(CHECK_INTERVAL)

    except Exception as e:
        print("Ошибка в цикле:", e)
        time.sleep(5)