Python Examples

Examples for interacting with the XBTFX Trading API using Python and the requests library.

Setup

import requests

BASE_URL = "https://interface.xbtfx.com"
API_KEY = "your_api_key_here"

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}

Check Auth Status

response = requests.get(f"{BASE_URL}/v1/auth/status", headers=headers)
auth = response.json()

print(f"Login: {auth['login']}")
print(f"Status: {auth['status']}")
print(f"Mode: {auth['margin_mode']}")

Get Account Info

response = requests.get(f"{BASE_URL}/v1/account", headers=headers)
account = response.json()

print(f"Balance: {account['balance']}")
print(f"Equity: {account['equity']}")
print(f"Free Margin: {account['margin_free']}")
print(f"Leverage: 1:{account['leverage']}")

Open a Trade

trade = {
    "symbol": "EURUSD",
    "side": "buy",
    "volume": 0.1,
    "sl": 1.08000,
    "tp": 1.09500,
}

response = requests.post(f"{BASE_URL}/v1/trade", headers=headers, json=trade)
result = response.json()

if result.get("retcode") in (10008, 10009):
    print(f"Trade opened: deal={result['deal']}, price={result['price']}")
else:
    print(f"Trade failed: {result}")

Open a Trade with Idempotency Key

import uuid

trade = {
    "symbol": "BTCUSD",
    "side": "sell",
    "volume": 0.01,
}

idempotency_headers = {
    **headers,
    "Idempotency-Key": str(uuid.uuid4()),
}

response = requests.post(
    f"{BASE_URL}/v1/trade", headers=idempotency_headers, json=trade
)
result = response.json()
print(result)

Get Open Positions

response = requests.get(f"{BASE_URL}/v1/positions", headers=headers)
positions = response.json()

for pos in positions:
    print(
        f"#{pos['ticket']} {pos['symbol']} {pos['side']} "
        f"{pos['volume']} lots  P&L: {pos['profit']}"
    )

Close a Position

close_request = {
    "ticket": 12345678,
}

response = requests.post(f"{BASE_URL}/v1/close", headers=headers, json=close_request)
result = response.json()

if result.get("retcode") in (10008, 10009):
    print(f"Closed at {result['price']}, deal={result['deal']}")
else:
    print(f"Close failed: {result}")

Modify SL/TP

modify_request = {
    "ticket": 12345678,
    "sl": 1.08200,
    "tp": 1.09800,
}

response = requests.post(f"{BASE_URL}/v1/modify", headers=headers, json=modify_request)
result = response.json()
print(result)

Get Trade History

# Using a named period
response = requests.get(
    f"{BASE_URL}/v1/history", headers=headers, params={"period": "last_week"}
)
history = response.json()

for deal in history["deals"]:
    print(
        f"{deal['time']} {deal['symbol']} {deal['side']} {deal['entry']} "
        f"{deal['volume']} @ {deal['price']}  P&L: {deal['profit']}"
    )

WebSocket: Stream Quotes

Requires the websockets library (pip install websockets).

import asyncio
import json
import websockets

API_KEY = "your_api_key_here"
WS_URL = "wss://interface.xbtfx.com/v1/ws"


async def stream_quotes():
    async with websockets.connect(WS_URL) as ws:
        # Authenticate
        await ws.send(json.dumps({
            "type": "auth",
            "api_key": API_KEY,
        }))

        auth_response = json.loads(await ws.recv())
        if auth_response.get("status") != "ok":
            print("Authentication failed")
            return

        # Subscribe to symbols
        await ws.send(json.dumps({
            "type": "subscribe",
            "symbols": ["EURUSD", "GBPUSD", "BTCUSD"],
        }))

        # Process incoming quotes
        async for message in ws:
            msg = json.loads(message)
            if msg["type"] == "quote":
                print(f"{msg['symbol']}: {msg['bid']} / {msg['ask']}")


asyncio.run(stream_quotes())