Skip to main content

Code Examples

Signed request helpers, REST examples, WebSocket subscriptions, and integration patterns.

Disclaimer

The following examples are simplified onboarding examples intended for API exploration and integration learning.

They are NOT production-ready trading bots.

Production systems should additionally implement:

  • reconnect handling

  • rate limit tracking

  • persistence

  • retry policies

  • risk management

  • monitoring

  • security hardening

Python helper example for signed requests​

import base64
import hashlib
import json
import time
from typing import Optional

import requests
from nacl.signing import SigningKey


BASE_URL = "https://api.exness.com"
API_KEY = "<EXN_API_KEY>"

# Private key bytes must be loaded securely.
# Do not hardcode private keys in production code.
PRIVATE_KEY_HEX = "<ed25519_private_key_hex>"


def b64url_no_padding(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).decode("ascii").rstrip("=")


def body_hash(body: bytes) -> str:
    return b64url_no_padding(hashlib.sha256(body).digest())


def sign_request(
    method: str,
    path: str,
    body: Optional[dict] = None,
    idempotency_key: str = "",
) -> dict:
    method = method.upper()
    timestamp = int(time.time() * 1000)

    if body is None:
        body_bytes = b""
    else:
        body_bytes = json.dumps(body, separators=(",", ":"), ensure_ascii=False).encode("utf-8")

    payload = {
        "api_key": API_KEY,
        "idempotency_key": idempotency_key,
        "timestamp": timestamp,
        "sign_version": 1,
        "method": method,
        "path": path,
        "body_hash": body_hash(body_bytes),
    }

    payload_bytes = json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8")

    signing_key = SigningKey(bytes.fromhex(PRIVATE_KEY_HEX))
    signature = signing_key.sign(payload_bytes).signature

    return {
        "headers": {
            "EXN-API-KEY": API_KEY,
            "EXN-IDEMPOTENCY-KEY": idempotency_key,
            "EXN-TIMESTAMP": str(timestamp),
            "EXN-SIGN-VERSION": "1",
            "EXN-DATA": b64url_no_padding(payload_bytes),
            "EXN-SIGN": b64url_no_padding(signature),
            "Content-Type": "application/json",
        },
        "body_bytes": body_bytes,
    }


def signed_get(path: str):
    signed = sign_request("GET", path, body=None, idempotency_key="")
    return requests.get(BASE_URL + path, headers=signed["headers"])


def signed_post(path: str, body: dict, idempotency_key: str):
    signed = sign_request("POST", path, body=body, idempotency_key=idempotency_key)
    return requests.post(BASE_URL + path, headers=signed["headers"], data=signed["body_bytes"])

Python: open position and wait for transaction event​

import asyncio
import json
import uuid

import websockets


def open_position():
    client_request_id = str(uuid.uuid4())

    body = {
        "instrument": "EURUSD",
        "side": "buy",
        "volume": "0.01",
        "comment": "Test position",
    }

    ack = api_post(
        path=f"/v1/trading/accounts/{ACCOUNT_ID}/positions",
        body=body,
        idempotency_key=client_request_id,
    )

    print("ACK:", ack)

    return ack["operation_id"], ack.get("client_request_id", client_request_id)


async def listen_transactions(expected_operation_id: str):
    ws_path = f"/v1/server-events/accounts/{ACCOUNT_ID}/ws/events"
    ws_url = WS_BASE_URL + ws_path

    async with websockets.connect(
        ws_url,
        additional_headers=signed_ws_headers(ws_path),
    ) as ws:
        await ws.send(json.dumps({
            "id": "subscribe-transactions",
            "subscribe": {
                "event": "transactions"
            }
        }))

        while True:
            raw_message = await ws.recv()
            message = json.loads(raw_message)

            print("WS:", message)

            if message.get("event_type") == "transaction_event":
                if message.get("operation_id") == expected_operation_id:
                    print("Matched transaction:", message)
                    return message


async def main():
    operation_id, client_request_id = open_position()
    transaction = await listen_transactions(operation_id)

    if transaction.get("status") == "success":
        print("Position opened successfully")
    else:
        print("Operation was not successful:", transaction)


asyncio.run(main())

Python: place pending order​

import uuid


client_request_id = str(uuid.uuid4())

body = {
    "type": "limit",
    "instrument": "EURUSD",
    "side": "buy",
    "volume": "0.01",
    "price": "1.0800",
    "stop_loss_price": "1.0750",
    "take_profit_price": "1.0900",
    "comment": "Buy limit for test",
}

ack = api_post(
    path=f"/v1/trading/accounts/{ACCOUNT_ID}/orders",
    body=body,
    idempotency_key=client_request_id,
)

print("ACK:", ack)

JavaScript: helper for signed WebSocket requests​

const crypto = require("crypto");


const BASE_URL = "https://api.exness.com";
const WS_BASE_URL = "wss://api.exness.com";

const ACCOUNT_ID = "<account_id>";

const API_KEY = "<EXN_API_KEY>";

// Private key must be loaded securely.
// Do not hardcode private keys in production code.
const PRIVATE_KEY_HEX = "<ED25519_PRIVATE_KEY_HEX>";


function b64urlNoPadding(buffer) {
  return Buffer.from(buffer)
    .toString("base64url")
    .replace(/=+$/, "");
}


function bodyHash(bodyBuffer) {
  return b64urlNoPadding(crypto.createHash("sha256").update(bodyBuffer).digest());
}


function compactJson(data) {
  return Buffer.from(JSON.stringify(data));
}


function signRequest(method, path, body = null, idempotencyKey = "") {
  const timestamp = Date.now();

  const bodyBuffer = body === null
    ? Buffer.from("")
    : compactJson(body);

  const payload = {
    api_key: API_KEY,
    idempotency_key: idempotencyKey,
    timestamp,
    sign_version: 1,
    method: method.toUpperCase(),
    path,
    body_hash: bodyHash(bodyBuffer)
  };

  const payloadBuffer = compactJson(payload);

  const privateKey = crypto.createPrivateKey({
    key: Buffer.from(PRIVATE_KEY_HEX, "hex"),
    format: "der",
    type: "pkcs8"
  });

  const signature = crypto.sign(null, payloadBuffer, privateKey);

  return {
    headers: {
      "EXN-API-KEY": API_KEY,
      "EXN-IDEMPOTENCY-KEY": idempotencyKey,
      "EXN-TIMESTAMP": String(timestamp),
      "EXN-SIGN-VERSION": "1",
      "EXN-DATA": b64urlNoPadding(payloadBuffer),
      "EXN-SIGN": b64urlNoPadding(signature),
      "Content-Type": "application/json"
    },
    bodyBuffer
  };
}


function signedWsHeaders(path) {
  return signRequest("GET", path, null, "").headers;
}

Python: subscribe to live XAUUSD ticks​

import base64
import hashlib
import json
import time

from nacl.signing import SigningKey
import websocket


API_KEY = "<EXN_API_KEY>"
PRIVATE_KEY_HEX = "<ED25519_PRIVATE_KEY_HEX>"
ACCOUNT_ID = "<account_id>"
WS_BASE_URL = "wss://<resolved_access_point_host>"


def b64url_no_padding(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).decode("ascii").rstrip("=")


def generate_signed_ws_headers(signing_path: str) -> list[str]:
    timestamp = int(time.time() * 1000)
    body_bytes = b""
    body_hash = b64url_no_padding(hashlib.sha256(body_bytes).digest())

    payload = {
        "api_key": API_KEY,
        "idempotency_key": "",
        "timestamp": timestamp,
        "sign_version": 1,
        "method": "GET",
        "path": signing_path,
        "body_hash": body_hash,
    }

    payload_bytes = json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
    signing_key = SigningKey(bytes.fromhex(PRIVATE_KEY_HEX.strip())[:32])
    signature = signing_key.sign(payload_bytes).signature

    return [
        f"EXN-API-KEY: {API_KEY}",
        "EXN-IDEMPOTENCY-KEY: ",
        f"EXN-TIMESTAMP: {timestamp}",
        "EXN-SIGN-VERSION: 1",
        f"EXN-DATA: {b64url_no_padding(payload_bytes)}",
        f"EXN-SIGN: {b64url_no_padding(signature)}",
    ]


def stream_xauusd_ticks() -> None:
    ticks_path = f"/v1/server-events/accounts/{ACCOUNT_ID}/ws/ticks"
    ws_url = f"{WS_BASE_URL}{ticks_path}"
    ws = websocket.create_connection(ws_url, header=generate_signed_ws_headers(ticks_path))

    ws.send(json.dumps({
        "id": "ticks-xauusd-1",
        "subscribe": {
            "event": "ticks",
            "instruments": ["XAUUSD"],
        },
    }))

    while True:
        message = json.loads(ws.recv())
        print(message)


stream_xauusd_ticks()

JavaScript: ticks bot​

const WebSocket = require("ws");


const ticksPath = `/v1/server-events/accounts/${ACCOUNT_ID}/ws/ticks`;
const ticksUrl = `${WS_BASE_URL}${ticksPath}`;

const ws = new WebSocket(ticksUrl, {
  headers: signedWsHeaders(ticksPath)
});

ws.on("open", () => {
  ws.send(JSON.stringify({
    id: "ticks-1",
    subscribe: {
      event: "ticks",
      instruments: ["EURUSD", "GBPUSD"]
    }
  }));
});

ws.on("message", (data) => {
  const message = JSON.parse(data.toString());

  if (message.instrument === "EURUSD") {
    console.log("EURUSD tick:", {
      bid: message.bid,
      ask: message.ask,
      timestamp: message.timestamp
    });
  }
});

ws.on("error", (error) => {
  console.error("WebSocket error:", error);
});

ws.on("close", (code, reason) => {
  console.log("WebSocket closed:", code, reason.toString());
});

JavaScript: account equity monitor​

const WebSocket = require("ws");


const eventsPath = `/v1/server-events/accounts/${ACCOUNT_ID}/ws/events`;
const eventsUrl = `${WS_BASE_URL}${eventsPath}`;

const ws = new WebSocket(eventsUrl, {
  headers: signedWsHeaders(eventsPath)
});

ws.on("open", () => {
  ws.send(JSON.stringify({
    id: "account-state-1",
    subscribe: {
      event: "account_state"
    }
  }));
});

ws.on("message", (data) => {
  const message = JSON.parse(data.toString());

  if (message.event_type === "account_state_event") {
    const accountState = message.payload.account_state;

    console.log("Equity:", accountState.equity);
    console.log("Balance:", accountState.balance);
    console.log("Used margin:", accountState.used_margin);
  }
});

ws.on("error", (error) => {
  console.error("WebSocket error:", error);
});

ws.on("close", (code, reason) => {
  console.log("WebSocket closed:", code, reason.toString());
});

Python: get available instruments and subscribe to updates​

import asyncio
import json

import websockets


def get_available_instruments():
    response = api_get(f"/v1/configuration/accounts/{ACCOUNT_ID}/instruments")
    return response["instruments"]


async def subscribe_instruments(instruments):
    ws_path = f"/v1/server-events/accounts/{ACCOUNT_ID}/ws/events"
    ws_url = WS_BASE_URL + ws_path

    async with websockets.connect(
        ws_url,
        additional_headers=signed_ws_headers(ws_path),
    ) as ws:
        await ws.send(json.dumps({
            "id": "instrument-subscription-1",
            "subscribe": {
                "event": "instruments",
                "instruments": instruments
            }
        }))

        while True:
            message = json.loads(await ws.recv())

            if message.get("event_type") == "instrument_event":
                print("Instrument update:", message)


async def main():
    instruments = get_available_instruments()
    selected = instruments[:2]

    print("Subscribing to instrument updates:", selected)
    await subscribe_instruments(selected)


asyncio.run(main())

Python: Subscribe to HMR​

import asyncio
import json

import websockets


def get_available_instruments():
    response = api_get(f"/v1/configuration/accounts/{ACCOUNT_ID}/instruments")
    return response["instruments"]


async def subscribe_hmr(instruments):
    ws_path = f"/v1/server-events/accounts/{ACCOUNT_ID}/ws/events"
    ws_url = WS_BASE_URL + ws_path

    async with websockets.connect(
        ws_url,
        additional_headers=signed_ws_headers(ws_path),
    ) as ws:
        await ws.send(json.dumps({
            "id": "hmr-subscription-1",
            "subscribe": {
                "event": "hmr",
                "instruments": instruments
            }
        }))

        hmr_periods_by_id = {}

        while True:
            message = json.loads(await ws.recv())

            if message.get("event_type") == "hmr_snapshot":
                hmr_periods_by_id = {
                    period["period_id"]: period
                    for period in message.get("periods", [])
                }

                print("HMR snapshot received")
                print("Current HMR periods:", hmr_periods_by_id)

            elif message.get("event_type") == "hmr_update":
                for period in message.get("upserted_periods", []):
                    hmr_periods_by_id[period["period_id"]] = period

                for period_id in message.get("removed_period_ids", []):
                    hmr_periods_by_id.pop(period_id, None)

                print("HMR update received")
                print("Current HMR periods:", hmr_periods_by_id)


async def main():
    available = get_available_instruments()

    selected = [
        instrument
        for instrument in ["EURUSD", "XAUUSD"]
        if instrument in available
    ]

    if not selected:
        raise RuntimeError("No selected HMR instruments are available for this account")

    print("Subscribing to HMR for:", selected)
    await subscribe_hmr(selected)


asyncio.run(main())

Python: Subscribe to Transactions​

import asyncio
import json

import websockets


async def subscribe_transactions():
    ws_path = f"/v1/server-events/accounts/{ACCOUNT_ID}/ws/events"
    ws_url = WS_BASE_URL + ws_path

    async with websockets.connect(
        ws_url,
        additional_headers=signed_ws_headers(ws_path),
    ) as ws:
        await ws.send(json.dumps({
            "id": "transactions-subscription-1",
            "subscribe": {
                "event": "transactions"
            }
        }))

        while True:
            message = json.loads(await ws.recv())

            if message.get("event_type") == "trading_state_snapshot":
                print("Initial trading snapshot:", message)

            elif message.get("event_type") == "transaction_event":
                print("Transaction event:", message)


asyncio.run(subscribe_transactions())

Python: Correlate ACK and transaction_event​

ack = {
    "operation_id": "789",
    "client_request_id": "order-20260316-0001",
    "status": "accepted"
}

event = {
    "event_type": "transaction_event",
    "operation_id": "789",
    "client_request_id": "order-20260316-0001",
    "status": "success"
}

if ack["operation_id"] == event.get("operation_id"):
    print("Matched by operation_id")

if ack.get("client_request_id") == event.get("client_request_id"):
    print("Matched by client_request_id")

Python: Poll Operation Status​

operation_id = "789"

status = api_get(
    f"/v1/trading/accounts/{ACCOUNT_ID}/operations/{operation_id}"
)

print("Operation status:", status)

if status["status"] == "confirmed":
    print("Operation completed successfully")

elif status["status"] in ("rejected", "failed"):
    print("Operation did not complete successfully")
    print("Error code:", status.get("error_code"))
    print("Error message:", status.get("error_message"))

else:
    print("Operation is still pending")

Python: Get Rate Limits​

limits = api_get(
    f"/v1/configuration/accounts/{ACCOUNT_ID}/rate-limits"
)

print("Rate limits:", limits)