Как построить торгового бота на Binance API

INTENT_CREATED
→ VALIDATED
→ READY_TO_SEND
→ SUBMITTING
→ ACKNOWLEDGED
→ PARTIALLY_FILLED
→ FILLED / CANCELED / EXPIRED / REJECTED

SUBMITTING
→ SUBMISSION_UNKNOWN
→ RECONCILING
→ ACKNOWLEDGED / TERMINAL / MANUAL_REVIEW

SUBMISSION_UNKNOWN → новый POST без reconciliation

авторизованные события source account
→ нормализованная позиция или fill
→ преобразование sizing и risk для destination
→ persistent outbox destination
→ Futures order destination
→ сверка fills и позиции

from __future__ import annotations

import hashlib
import hmac
import json
import os
import sqlite3
import time
from dataclasses import dataclass
from decimal import Decimal
from typing import Any
from urllib.parse import urlencode

import requests


MODE = os.getenv("BINANCE_MODE", "validate").lower()
if MODE not in {"validate", "testnet", "production"}:
    raise ValueError("BINANCE_MODE must be validate, testnet, or production")

BASE_URLS = {
    "validate": "https://api.binance.com",
    "testnet": "https://testnet.binance.vision",
    "production": "https://api.binance.com",
}
ORDER_PATHS = {
    "validate": "/api/v3/order/test",
    "testnet": "/api/v3/order",
    "production": "/api/v3/order",
}

BASE_URL = BASE_URLS[MODE]
ORDER_PATH = ORDER_PATHS[MODE]
API_KEY = os.environ["BINANCE_API_KEY"]
API_SECRET = os.environ["BINANCE_API_SECRET"].encode()
DB_PATH = os.getenv("BOT_DB_PATH", "orders.sqlite3")
RECV_WINDOW_MS = 5_000

session = requests.Session()
session.headers["X-MBX-APIKEY"] = API_KEY


@dataclass(frozen=True)
class LimitIntent:
    intent_id: str
    client_order_id: str
    symbol: str
    side: str
    quantity: Decimal
    price: Decimal


def connect_db() -> sqlite3.Connection:
    db = sqlite3.connect(DB_PATH)
    db.row_factory = sqlite3.Row
    db.execute("PRAGMA journal_mode=WAL")
    db.execute(
        """
        CREATE TABLE IF NOT EXISTS order_intents (
            intent_id TEXT PRIMARY KEY,
            client_order_id TEXT NOT NULL UNIQUE,
            symbol TEXT NOT NULL,
            side TEXT NOT NULL,
            quantity TEXT NOT NULL,
            price TEXT NOT NULL,
            state TEXT NOT NULL,
            exchange_order_id INTEGER,
            exchange_status TEXT,
            last_error TEXT,
            created_at_ms INTEGER NOT NULL,
            updated_at_ms INTEGER NOT NULL
        )
        """
    )
    return db


def now_ms() -> int:
    return time.time_ns() // 1_000_000


def signed_params(params: dict[str, Any]) -> dict[str, Any]:
    signed = dict(params)
    signed["timestamp"] = now_ms()
    signed["recvWindow"] = RECV_WINDOW_MS
    query = urlencode(signed)
    signed["signature"] = hmac.new(
        API_SECRET,
        query.encode(),
        hashlib.sha256,
    ).hexdigest()
    return signed


def request_json(
    method: str,
    path: str,
    *,
    params: dict[str, Any],
    timeout: float = 10.0,
) -> tuple[requests.Response, dict[str, Any]]:
    response = session.request(
        method,
        f"{BASE_URL}{path}",
        params=signed_params(params),
        timeout=timeout,
    )
    try:
        payload = response.json()
    except ValueError:
        payload = {"msg": response.text[:500]}
    return response, payload


def decimal_multiple(value: Decimal, step: Decimal) -> bool:
    if step == 0:
        return True
    return value % step == 0


def validate_range(
    *,
    name: str,
    value: Decimal,
    minimum: Decimal,
    maximum: Decimal,
    step: Decimal,
) -> None:
    if minimum != 0 and value < minimum:
        raise ValueError(f"{name} {value} is below minimum {minimum}")
    if maximum != 0 and value > maximum:
        raise ValueError(f"{name} {value} is above maximum {maximum}")
    if not decimal_multiple(value, step):
        raise ValueError(f"{name} {value} is not aligned to step {step}")


def load_symbol_rules(symbol: str) -> dict[str, Any]:
    response = session.get(
        f"{BASE_URL}/api/v3/exchangeInfo",
        params={"symbol": symbol},
        timeout=5,
    )
    response.raise_for_status()
    payload = response.json()
    symbols = payload.get("symbols", [])
    if len(symbols) != 1:
        raise ValueError(f"symbol {symbol} was not returned by exchangeInfo")
    return symbols[0]


def validate_limit_intent(intent: LimitIntent) -> None:
    rules = load_symbol_rules(intent.symbol)
    if rules["status"] != "TRADING":
        raise ValueError(f"{intent.symbol} status is {rules['status']}")
    if "LIMIT" not in rules.get("orderTypes", []):
        raise ValueError(f"LIMIT orders are not enabled for {intent.symbol}")

    filters = {
        item["filterType"]: item
        for item in rules.get("filters", [])
    }

    price_filter = filters["PRICE_FILTER"]
    lot_filter = filters["LOT_SIZE"]

    validate_range(
        name="price",
        value=intent.price,
        minimum=Decimal(price_filter["minPrice"]),
        maximum=Decimal(price_filter["maxPrice"]),
        step=Decimal(price_filter["tickSize"]),
    )
    validate_range(
        name="quantity",
        value=intent.quantity,
        minimum=Decimal(lot_filter["minQty"]),
        maximum=Decimal(lot_filter["maxQty"]),
        step=Decimal(lot_filter["stepSize"]),
    )

    notional = intent.price * intent.quantity

    if "MIN_NOTIONAL" in filters:
        minimum = Decimal(filters["MIN_NOTIONAL"]["minNotional"])
        if notional < minimum:
            raise ValueError(f"notional {notional} is below {minimum}")

    if "NOTIONAL" in filters:
        item = filters["NOTIONAL"]
        minimum = Decimal(item["minNotional"])
        maximum = Decimal(item["maxNotional"])
        if notional < minimum or (maximum != 0 and notional > maximum):
            raise ValueError(
                f"notional {notional} is outside {minimum}..{maximum}"
            )

    # Account-dependent and dynamic filters remain authoritative on Binance:
    # PERCENT_PRICE(_BY_SIDE), MAX_POSITION, MAX_NUM_ORDERS, and others.
    # Do not claim local validation proves that the order will be accepted.


def persist_intent(db: sqlite3.Connection, intent: LimitIntent) -> None:
    timestamp = now_ms()
    with db:
        db.execute(
            """
            INSERT INTO order_intents (
                intent_id, client_order_id, symbol, side,
                quantity, price, state, created_at_ms, updated_at_ms
            ) VALUES (?, ?, ?, ?, ?, ?, 'INTENT_CREATED', ?, ?)
            ON CONFLICT(intent_id) DO NOTHING
            """,
            (
                intent.intent_id,
                intent.client_order_id,
                intent.symbol,
                intent.side,
                str(intent.quantity),
                str(intent.price),
                timestamp,
                timestamp,
            ),
        )


def set_state(
    db: sqlite3.Connection,
    intent_id: str,
    state: str,
    *,
    exchange_order_id: int | None = None,
    exchange_status: str | None = None,
    error: str | None = None,
) -> None:
    with db:
        db.execute(
            """
            UPDATE order_intents
            SET state = ?,
                exchange_order_id = COALESCE(?, exchange_order_id),
                exchange_status = COALESCE(?, exchange_status),
                last_error = ?,
                updated_at_ms = ?
            WHERE intent_id = ?
            """,
            (
                state,
                exchange_order_id,
                exchange_status,
                error,
                now_ms(),
                intent_id,
            ),
        )


def query_order(
    symbol: str,
    client_order_id: str,
) -> dict[str, Any] | None:
    response, payload = request_json(
        "GET",
        "/api/v3/order",
        params={
            "symbol": symbol,
            "origClientOrderId": client_order_id,
        },
    )
    if response.ok:
        return payload
    if payload.get("code") == -2013:
        return None
    response.raise_for_status()
    return None


def reconcile(
    db: sqlite3.Connection,
    intent_id: str,
) -> None:
    row = db.execute(
        "SELECT * FROM order_intents WHERE intent_id = ?",
        (intent_id,),
    ).fetchone()
    if row is None:
        raise KeyError(intent_id)

    set_state(db, intent_id, "RECONCILING")

    try:
        order = query_order(row["symbol"], row["client_order_id"])
    except requests.RequestException as exc:
        set_state(
            db,
            intent_id,
            "SUBMISSION_UNKNOWN",
            error=f"reconciliation failed: {type(exc).__name__}: {exc}",
        )
        return

    if order is None:
        # Production must also wait for User Data Stream and query recent
        # trades before deciding that the order never existed.
        set_state(
            db,
            intent_id,
            "MANUAL_REVIEW",
            error="order not found; do not resubmit automatically",
        )
        return

    status = str(order.get("status", "UNKNOWN"))
    terminal = status in {
        "FILLED", "CANCELED", "REJECTED", "EXPIRED",
        "EXPIRED_IN_MATCH",
    }
    state = status if terminal else "ACKNOWLEDGED"
    set_state(
        db,
        intent_id,
        state,
        exchange_order_id=order.get("orderId"),
        exchange_status=status,
    )


def submit(
    db: sqlite3.Connection,
    intent: LimitIntent,
) -> None:
    persist_intent(db, intent)
    row = db.execute(
        "SELECT * FROM order_intents WHERE intent_id = ?",
        (intent.intent_id,),
    ).fetchone()
    if row is None:
        raise RuntimeError("intent was not persisted")

    if row["state"] in {
        "ACKNOWLEDGED", "PARTIALLY_FILLED", "FILLED",
        "CANCELED", "REJECTED", "EXPIRED", "EXPIRED_IN_MATCH",
        "VALIDATED", "MANUAL_REVIEW",
    }:
        return

    if row["state"] in {"SUBMITTING", "SUBMISSION_UNKNOWN", "RECONCILING"}:
        reconcile(db, intent.intent_id)
        return

    validate_limit_intent(intent)
    set_state(db, intent.intent_id, "VALIDATED")

    order_params = {
        "symbol": intent.symbol,
        "side": intent.side,
        "type": "LIMIT",
        "timeInForce": "GTC",
        "quantity": str(intent.quantity),
        "price": str(intent.price),
        "newClientOrderId": intent.client_order_id,
    }

    set_state(db, intent.intent_id, "SUBMITTING")

    try:
        response, payload = request_json(
            "POST",
            ORDER_PATH,
            params=order_params,
        )
    except (requests.Timeout, requests.ConnectionError) as exc:
        set_state(
            db,
            intent.intent_id,
            "SUBMISSION_UNKNOWN",
            error=f"{type(exc).__name__}: {exc}",
        )
        return

    if MODE == "validate" and response.ok:
        set_state(db, intent.intent_id, "VALIDATED")
        return

    if response.status_code >= 500 or payload.get("code") == -1007:
        set_state(
            db,
            intent.intent_id,
            "SUBMISSION_UNKNOWN",
            error=json.dumps(payload),
        )
        return

    if response.status_code in {418, 429}:
        set_state(
            db,
            intent.intent_id,
            "REJECTED",
            error=(
                f"rate limit response; Retry-After="
                f"{response.headers.get('Retry-After')}"
            ),
        )
        return

    if not response.ok:
        set_state(
            db,
            intent.intent_id,
            "REJECTED",
            error=json.dumps(payload),
        )
        return

    set_state(
        db,
        intent.intent_id,
        "ACKNOWLEDGED",
        exchange_order_id=payload.get("orderId"),
        exchange_status=str(payload.get("status", "NEW")),
    )


def reconcile_incomplete(db: sqlite3.Connection) -> None:
    rows = db.execute(
        """
        SELECT intent_id
        FROM order_intents
        WHERE state IN ('SUBMITTING', 'SUBMISSION_UNKNOWN', 'RECONCILING')
        """
    ).fetchall()
    for row in rows:
        reconcile(db, row["intent_id"])


if __name__ == "__main__":
    if MODE == "production" and os.getenv("ALLOW_REAL_ORDERS") != "YES":
        raise RuntimeError(
            "Set ALLOW_REAL_ORDERS=YES only after testnet validation"
        )

    database = connect_db()
    reconcile_incomplete(database)

    example = LimitIntent(
        intent_id="strategy-1:decision-20260802-0001",
        client_order_id="s1-20260802-0001",
        symbol="BTCUSDT",
        side="BUY",
        quantity=Decimal("0.00100000"),
        price=Decimal("50000.00000000"),
    )
    submit(database, example)

trading_intent_total{strategy}
trading_intent_state_total{strategy,state}
order_submission_total{strategy,status}
order_submission_unknown_total{strategy}
order_reconciliation_duration_ms{strategy}
order_manual_review_total{strategy,reason}
duplicate_intent_blocked_total{strategy}
order_reject_total{symbol,code,filter}
user_data_stream_lag_ms{account}
market_data_age_ms{stream}
market_data_gap_total{stream}
symbol_filters_age_seconds{symbol}
rate_limit_usage{type,interval,ip}
open_order_count{account,symbol}
open_position_notional{strategy,symbol}
risk_reject_total{strategy,rule}
circuit_breaker_state{strategy}