Trading Bot

┌─────────────────┐     ┌──────────────────┐     ┌──────────────────┐
│  Market Data     │────▶│  Signal / Logic   │────▶│  Order Execution │
│  (WS + REST)     │     │  (Strategy)       │     │  (REST API)      │
└─────────────────┘     └──────────────────┘     └──────────────────┘
        ▲                                                 │
        │           ┌──────────────────┐                  │
        └───────────│  Fill Monitoring  │◀─────────────────┘
                    │  (WebSocket)      │
                    └──────────────────┘

import json, asyncio, websockets

async def stream_market_data():
    async with websockets.connect("wss://api.whitebit.com/ws") as ws:
        # Subscribe to last price updates
        await ws.send(json.dumps({
            "id": 1,
            "method": "lastprice_subscribe",
            "params": ["BTC_USDT"]
        }))
        # Subscribe to orderbook depth
        await ws.send(json.dumps({
            "id": 2,
            "method": "depth_subscribe",
            # params: [market, limit, price_interval, multi_subscription_flag]
            "params": ["BTC_USDT", 20, "0", True]
        }))

        async for message in ws:
            data = json.loads(message)
            method = data.get("method")
            if method == "lastprice_update":
                print(f"Price: {data['params']}")
            elif method == "depth_update":
                print(f"Orderbook update: {len(data['params'][1].get('bids', []))} bids")

import base64, hashlib, hmac, json, threading, time, requests

API_KEY = "YOUR_API_KEY"
API_SECRET = "YOUR_API_SECRET"
BASE_URL = "https://whitebit.com"

_nonce_lock = threading.Lock()
_last_nonce = 0

def _next_nonce():
    """Monotonic millisecond nonce — safe under concurrent callers."""
    global _last_nonce
    with _nonce_lock:
        n = max(int(time.time() * 1000), _last_nonce + 1)
        _last_nonce = n
        return str(n)

def send_request(path, data=None):
    """Send a signed request. Returns the decoded JSON body.
    Raises requests.HTTPError on HTTP 4xx/5xx — caller can inspect
    `.response.status_code` and `.response.json()`."""
    data = dict(data or {})
    data["request"] = path
    data["nonce"] = _next_nonce()
    data_json = json.dumps(data)
    payload_b64 = base64.b64encode(data_json.encode()).decode()
    signature = hmac.new(
        API_SECRET.encode(), payload_b64.encode(), hashlib.sha512
    ).hexdigest()
    headers = {
        "Content-Type": "application/json",
        "X-TXC-APIKEY": API_KEY,
        "X-TXC-PAYLOAD": payload_b64,
        "X-TXC-SIGNATURE": signature,
    }
    resp = requests.post(BASE_URL + path, headers=headers, data=data_json)
    resp.raise_for_status()
    return resp.json()

async def authenticate_ws(ws):
    """Fetch a WebSocket token via REST and authorize the WS connection."""
    token_resp = send_request("/api/v4/profile/websocket_token")
    token = token_resp["websocket_token"]
    await ws.send(json.dumps({
        "id": 999,
        "method": "authorize",
        "params": [token, "public"],
    }))
    ack = json.loads(await ws.recv())
    if ack.get("id") != 999 or ack.get("result", {}).get("status") != "success":
        raise RuntimeError(f"WS authorize failed: {ack}")

import threading

def kill_switch_heartbeat(market, timeout_seconds, refresh_interval):
    """Refresh kill-switch on a regular interval. If this thread stops,
    the kill-switch expires and all orders are canceled automatically."""
    timeout_str = str(timeout_seconds)  # API requires timeout as string
    while True:
        try:
            send_request("/api/v4/order/kill-switch", {
                "market": market,
                "timeout": timeout_str,
            })
        except Exception as e:
            print(f"Kill-switch refresh failed: {e} — retrying in 5s")
            time.sleep(5)
            continue
        time.sleep(refresh_interval)

# Refresh every 45 seconds with a 60-second timeout (15-second safety margin).
# Runs as a daemon thread (not an asyncio task) so refreshes continue even
# when the event loop is busy processing WebSocket messages.
heartbeat = threading.Thread(
    target=kill_switch_heartbeat,
    args=("BTC_USDT", 60, 45),
    daemon=True
)
heartbeat.start()

async def monitor_fills(ws):
    # Subscribe to trade fills
    await ws.send(json.dumps({
        "id": 10,
        "method": "deals_subscribe",
        "params": [["BTC_USDT"]]
    }))
    # Subscribe to order state changes
    await ws.send(json.dumps({
        "id": 11,
        "method": "ordersPending_subscribe",
        "params": ["BTC_USDT"]
    }))

    async for message in ws:
        data = json.loads(message)
        method = data.get("method")

        if method == "deals_update":
            # params is a flat 11-element tuple — not an array of objects.
            # Elements: [deal_id, time, market, order_id, price, amount, fee,
            #            client_order_id, side_int, role_int, fee_asset]
            p = data["params"]
            deal_id, ts, market, order_id, price, amount, fee, client_oid, side_int, role_int, fee_asset = p
            side = "sell" if side_int == 1 else "buy"
            print(f"Fill: {side} {amount} @ {price} (order {order_id})")

        elif method == "ordersPending_update":
            # params is [event_id, order_object].
            # event_id: 1=new, 2=update, 3=finish (cancel or execute)
            event_id, order = data["params"]
            event_label = {1: "new", 2: "update", 3: "finish"}.get(event_id, "?")
            print(f"Order {event_label}: {order['id']} — remaining: {order.get('left', 'N/A')}")

import random

def send_request_with_retry(path, data=None, max_retries=5):
    """Send a request with exponential backoff on HTTP 429 or 5xx.
    Requires `send_request` to raise on HTTP errors (via resp.raise_for_status())."""
    for attempt in range(max_retries):
        try:
            return send_request(path, data)
        except requests.HTTPError as e:
            status = e.response.status_code
            if status == 429 or 500 <= status < 600:
                delay = min(30, 2 ** attempt) + random.uniform(0, 1)
                print(f"HTTP {status}, retrying in {delay:.1f}s...")
                time.sleep(delay)
                continue
            raise  # 4xx other than 429 — don't retry
    raise Exception(f"Max retries exceeded for {path}")

PING_INTERVAL = 50   # seconds between pings
PONG_TIMEOUT = 10    # reconnect if no pong arrives within 10 seconds

async def keepalive(ws):
    """Send ping messages and monitor for pong responses."""
    while True:
        try:
            await ws.send(json.dumps({"id": 0, "method": "ping", "params": []}))
            pong = await asyncio.wait_for(ws.recv(), timeout=PONG_TIMEOUT)
            data = json.loads(pong)
            if data.get("result") != "pong":
                # Not a pong — process as regular message and wait for pong
                pass
        except (asyncio.TimeoutError, websockets.exceptions.ConnectionClosed):
            print("Connection lost — triggering reconnection")
            break
        await asyncio.sleep(PING_INTERVAL)

async def connect_with_retry(url, max_delay=60):
    """Connect to WebSocket with exponential backoff."""
    delay = 1
    while True:
        try:
            ws = await websockets.connect(url)
            print("WebSocket connected")
            return ws
        except Exception as e:
            jitter = random.uniform(0, 1)
            print(f"Connection failed: {e}. Retrying in {delay + jitter:.1f}s...")
            await asyncio.sleep(delay + jitter)
            delay = min(max_delay, delay * 2)

async def run_bot():
    """Main bot loop with automatic reconnection and resubscription."""
    subscriptions = [
        {"method": "lastprice_subscribe", "params": ["BTC_USDT"]},
        {"method": "depth_subscribe", "params": ["BTC_USDT", 20, "0", True]},
        {"method": "deals_subscribe", "params": [["BTC_USDT"]]},
        {"method": "ordersPending_subscribe", "params": ["BTC_USDT"]},
    ]

    while True:
        ws = await connect_with_retry("wss://api.whitebit.com/ws")
        try:
            # Private channels require authentication first
            await authenticate_ws(ws)

            # Re-subscribe to all channels
            for i, sub in enumerate(subscriptions):
                await ws.send(json.dumps({"id": i + 1, **sub}))

            # Process messages — `handle_message` is a strategy-specific
            # callback the caller provides. The grid-bot walkthrough below
            # shows the strategy-specific replacement trigger.
            async for message in ws:
                data = json.loads(message)
                await handle_message(data)

        except websockets.exceptions.ConnectionClosed:
            print("Disconnected — reconnecting...")
            continue

def reconcile_state(market, local_orders):
    """Query active orders; reconcile against local state."""
    # 1. Active orders on the exchange right now
    active = send_request("/api/v4/orders", {"market": market})
    server_ids = {o["orderId"] for o in active}

    # 2. Drop any local order that the server no longer knows about
    disappeared = set(local_orders) - server_ids
    for oid in disappeared:
        local_orders.pop(oid)
        print(f"Reconcile: {oid} vanished — assuming filled or canceled")

    # 3. Orders the server knows about that are missing from local state
    #    (e.g., after a restart)
    unknown_locally = server_ids - set(local_orders)
    for oid in unknown_locally:
        print(f"Reconcile: adopting server-side order {oid}")

async def handle_message(data):
    """Replacement trigger. deals_update fires per PARTIAL execution —
    log fills there, never replace. Replace only on ordersPending
    event 3 (finish) with left == 0 (fully executed, not canceled)."""
    if data.get("method") == "ordersPending_update":
        event_id, order = data["params"]
        if event_id == 3 and float(order.get("left", "0")) == 0:
            filled = grid_orders.pop(order["id"], None)
            if filled is None:
                return  # not a grid order
            up = filled["side"] == "buy"
            send_request("/api/v4/order/new", {
                "market": MARKET,
                "side": "sell" if up else "buy",
                "amount": ORDER_AMOUNT,
                "price": str(float(filled["price"]) + (GRID_SPACING if up else -GRID_SPACING)),
            })