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}")

# Place 5 buy orders at descending prices.
# Note: `market` goes on each order item, not at the top level.
orders = [
    {"side": "buy", "amount": "0.001", "price": str(60000 - i * 100), "market": "BTC_USDT"}
    for i in range(5)
]
result = send_request("/api/v4/order/bulk", {"orders": orders})

# Bulk response is an array of {result, error} pairs. `result` is null on failure.
for i, item in enumerate(result):
    if item.get("result"):
        print(f"Order {i}: placed (ID {item['result']['orderId']})")
    else:
        err = item.get("error") or {}
        print(f"Order {i}: failed — code {err.get('code')}, {err.get('message', 'unknown')}")

# Activate kill-switch with 60-second timeout.
# Note: the API expects `timeout` as a STRING, not an integer.
ks = send_request("/api/v4/order/kill-switch", {
    "market": "BTC_USDT",
    "timeout": "60"  # seconds as string, valid range: "5"–"600"
})
print(f"Kill-switch active, cancellationTime: {ks.get('cancellationTime')}")

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}")
        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()

# Deactivate by setting timeout to null (None in Python → null in JSON)
send_request("/api/v4/order/kill-switch", {
    "market": "BTC_USDT",
    "timeout": None
})

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 = 60    # seconds to wait for pong before considering connection dead

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 worked example
            # below shows one complete implementation.
            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}")

import asyncio
import base64
import hashlib
import hmac
import json
import random
import threading
import time

import requests
import websockets

# ── Configuration ──────────────────────────────────────────
API_KEY = "YOUR_API_KEY"
API_SECRET = "YOUR_API_SECRET"
BASE_URL = "https://whitebit.com"
WS_URL = "wss://api.whitebit.com/ws"

# Demo market — DBTC/DUSDT costs nothing and uses the real API endpoints.
# Swap to "BTC_USDT" (or another real pair) after validating the strategy.
# Always re-check market constraints (`GET /api/v4/public/markets`) when
# switching markets — precision and min/max totals are per-pair.
MARKET = "DBTC_DUSDT"
GRID_SIZE = 10          # number of grid levels per side
GRID_SPACING = 50       # price distance between levels (in quote currency)
ORDER_AMOUNT = "0.001"  # order size per grid level (in base currency)
KS_TIMEOUT = "120"      # kill-switch timeout in seconds (API requires string)
KS_REFRESH = 90         # kill-switch refresh interval in seconds


# ── REST API helper (same as earlier in the guide) ────────
_nonce_lock = threading.Lock()
_last_nonce = 0

def _next_nonce():
    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):
    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 and authorize the connection."""
    token = send_request("/api/v4/profile/websocket_token")["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}")


# ── Grid state ─────────────────────────────────────────────
grid_orders = {}  # order_id -> {"side": str, "price": str, "level": int}
# Accessed only from the asyncio event loop. Do not access from threads.


def calculate_grid_levels(center_price):
    """Calculate buy and sell grid prices around a center price."""
    levels = []
    for i in range(1, GRID_SIZE + 1):
        levels.append({
            "side": "buy",
            "price": str(center_price - i * GRID_SPACING),
            "level": -i,
        })
        levels.append({
            "side": "sell",
            "price": str(center_price + i * GRID_SPACING),
            "level": i,
        })
    return levels


def place_grid_orders(center_price):
    """Place initial grid orders using the bulk endpoint."""
    global grid_orders
    levels = calculate_grid_levels(center_price)

    # Split into batches of 20 (bulk endpoint limit)
    for batch_start in range(0, len(levels), 20):
        batch = levels[batch_start:batch_start + 20]
        orders = [
            {"side": lvl["side"], "amount": ORDER_AMOUNT, "price": lvl["price"], "market": MARKET}
            for lvl in batch
        ]
        result = send_request("/api/v4/order/bulk", {"orders": orders})
        for i, item in enumerate(result):
            if item.get("result"):
                oid = item["result"]["orderId"]
                grid_orders[oid] = batch[i]
                print(f"Grid order placed: {batch[i]['side']} @ {batch[i]['price']}"
                      f" (ID {oid})")
            else:
                err = item.get("error") or {}
                print(f"Grid order failed: code {err.get('code')} {err.get('message')}")


def replace_filled_order(filled_order):
    """When a grid order fills, place the opposite order one level away."""
    level = filled_order["level"]
    filled_price = float(filled_order["price"])

    if filled_order["side"] == "buy":
        # Buy filled — place sell one grid level above
        new_side = "sell"
        new_price = str(filled_price + GRID_SPACING)
    else:
        # Sell filled — place buy one grid level below
        new_side = "buy"
        new_price = str(filled_price - GRID_SPACING)

    result = send_request("/api/v4/order/new", {
        "market": MARKET,
        "side": new_side,
        "amount": ORDER_AMOUNT,
        "price": new_price,
    })
    if "orderId" in result:
        grid_orders[result["orderId"]] = {
            "side": new_side,
            "price": new_price,
            "level": -level,
        }
        print(f"Replacement order: {new_side} @ {new_price} (ID {result['orderId']})")
    else:
        print(f"Replacement failed: code {result.get('code')} {result.get('message')}")


# ── Kill-switch heartbeat ──────────────────────────────────
def start_kill_switch():
    """Start a daemon thread that refreshes the kill-switch."""
    import threading

    def heartbeat():
        while True:
            try:
                send_request("/api/v4/order/kill-switch", {
                    "market": MARKET,
                    "timeout": KS_TIMEOUT,
                })
            except Exception as e:
                print(f"Kill-switch refresh error: {e}")
            time.sleep(KS_REFRESH)

    t = threading.Thread(target=heartbeat, daemon=True)
    t.start()
    print(f"Kill-switch active: {KS_TIMEOUT}s timeout, refreshing every {KS_REFRESH}s")


# ── WebSocket message handler ──────────────────────────────
async def handle_message(data):
    """Process WebSocket messages for fills and order updates."""
    method = data.get("method")

    if method == "deals_update":
        # Unpack the 11-element tuple — see Fill monitoring section above.
        p = data.get("params", [])
        if len(p) == 11:
            _, _, _, order_id, _, _, _, _, _, _, _ = p
            if order_id in grid_orders:
                filled = grid_orders.pop(order_id)
                print(f"Grid fill: {filled['side']} @ {filled['price']}")
                replace_filled_order(filled)

    elif method == "ordersPending_update":
        # params is [event_id, order_object]. Act only on event_id == 3 (finish).
        event_id, order = data["params"]
        if event_id == 3:
            oid = order["id"]
            if oid in grid_orders and float(order.get("left", "0")) == 0:
                filled = grid_orders.pop(oid)
                replace_filled_order(filled)


# ── Main bot loop ──────────────────────────────────────────
async def run_grid_bot():
    """Initialize grid orders and run the WebSocket monitoring loop."""
    # 1. Get current market price
    ticker = requests.get(f"{BASE_URL}/api/v4/public/ticker").json()
    last_price = float(ticker["BTC_USDT"]["last_price"])
    center = round(last_price / GRID_SPACING) * GRID_SPACING  # snap to grid
    print(f"Center price: {center} (last: {last_price})")

    # 2. Clean slate — cancel any orders left from a previous run.
    send_request("/api/v4/order/cancel/all", {"market": MARKET, "type": ["spot"]})

    # 3. Place initial grid orders
    place_grid_orders(center)

    # 4. Activate kill-switch
    start_kill_switch()

    # 5. Connect to WebSocket and monitor fills
    subscriptions = [
        {"method": "deals_subscribe", "params": [[MARKET]]},
        {"method": "ordersPending_subscribe", "params": [MARKET]},
    ]

    while True:
        try:
            async with websockets.connect(WS_URL) as ws:
                # Private channels require authentication first
                await authenticate_ws(ws)

                for i, sub in enumerate(subscriptions):
                    await ws.send(json.dumps({"id": i + 1, **sub}))

                async for message in ws:
                    data = json.loads(message)
                    await handle_message(data)

        except websockets.exceptions.ConnectionClosed:
            delay = random.uniform(1, 5)
            print(f"Disconnected — reconnecting in {delay:.1f}s...")
            await asyncio.sleep(delay)


if __name__ == "__main__":
    asyncio.run(run_grid_bot())