Place MetaTrader 5 orders from Python — no MetaTrader installed

python3 -m venv .venv && source .venv/bin/activate
pip install requests

export FXAPIS_KEY="fx_test_…"          # your API key
export MT5_LOGIN="26177561"
export MT5_SERVER="VantageMarkets-Demo"
export MT5_PASSWORD="…"                # the trading password
export MT5_SYMBOL="EURUSD"             # as your broker names it

"""Connect an MT5 account through fxapis, open a small trade, and close it.

Run it against a broker DEMO account. Every fxapis key reaches a real broker.
"""
import os
import sys
import time
import uuid

import requests

API = os.environ.get("FXAPIS_API", "https://api.fxapis.com")
NEEDS_ATTENTION = {"invalid_credentials", "trading_disabled", "needs_2fa", "needs_certificate"}
# Codes that mean "nothing was placed yet — ask again with the SAME key".
RETRY_SAME_KEY = {"SEND_FAILED", "IDEMPOTENCY_IN_FLIGHT", "ACCOUNT_NOT_READY", "NO_RUNTIME", "RATE_LIMITED"}

session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['FXAPIS_KEY']}"


class ApiError(Exception):
    def __init__(self, status, code, message, details=None):
        super().__init__(f"{status} {code}: {message}")
        self.status = status
        self.code = code
        self.details = details or []


def error_of(response):
    """The error object of a failed response, or None if the body is not ours (a proxy page)."""
    try:
        return response.json().get("error")
    except ValueError:
        return None


def raise_for(response):
    error = error_of(response) or {}
    raise ApiError(response.status_code, error.get("code", "HTTP_ERROR"),
                   error.get("message", response.text[:200]), error.get("details"))


def connect_account(login, server, password):
    """Stores the account at fxapis. It does not log in yet."""
    response = session.post(f"{API}/v1/accounts", timeout=30, json={
        "login": login,
        "server": server,
        "password": password,
        "mode": "warm_on_demand",
        "label": "python-guide",
    })
    if response.status_code == 201:
        return response.json()["data"]
    error = error_of(response) or {}
    if error.get("code") == "ACCOUNT_EXISTS":
        # Connected on an earlier run: find it instead of connecting it twice.
        for account in session.get(f"{API}/v1/accounts", timeout=30).json()["data"]:
            if account["login"] == login and account["server"] == server:
                return account
    raise_for(response)


def bring_online(account_id, timeout=120):
    """Starts the broker login, then polls until the account is ready to trade."""
    response = session.post(f"{API}/v1/accounts/{account_id}/warm", timeout=30)
    if response.status_code != 202:
        raise_for(response)  # 409 NEEDS_OPERATOR or NO_SECRET: a person has to act

    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        status = session.get(f"{API}/v1/accounts/{account_id}/status", timeout=30).json()["data"]
        if status["state"] == "ready":
            return status
        if status["state"] in NEEDS_ATTENTION:
            # Not retryable. Repeated failed logins are how brokers lock accounts.
            raise RuntimeError(f"account needs attention: {status['state']} ({status['detail']})")
        time.sleep(2)
    raise TimeoutError(f"account not ready after {timeout}s")


def wait_for_outcome(order_id, timeout=300):
    """After ORDER_UNRESOLVED: never resend. Poll the order until the broker's record settles it."""
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        order = session.get(f"{API}/v1/orders/{order_id}", timeout=30).json()["data"]
        if order["state"] != "unknown":
            return order
        time.sleep(5)
    raise RuntimeError(f"order {order_id} is still unknown. Do not resend it; check it again later.")


def send_order(path, body, attempts=6):
    """Places an order (or a close) exactly once, however many times the request is retried."""
    key = str(uuid.uuid4())  # one key per order, reused on every retry
    for attempt in range(attempts):
        try:
            response = session.post(f"{API}{path}", json=body, timeout=90,
                                    headers={"Idempotency-Key": key})
        except requests.RequestException:
            time.sleep(min(2 ** attempt, 10))  # we never saw the answer; the same key makes asking again safe
            continue

        if response.status_code == 201:
            return response.json()["data"]

        error = error_of(response)
        if error is None and response.status_code >= 500:
            time.sleep(min(2 ** attempt, 10))  # not our error body: a gateway hiccup
            continue
        if error is None:
            raise_for(response)

        code = error["code"]
        if code == "ORDER_UNRESOLVED":
            return wait_for_outcome(error["details"][0]["orderId"])
        if code in RETRY_SAME_KEY:
            wait = int(response.headers.get("retry-after", 0)) or min(2 ** attempt, 10)
            time.sleep(wait)
            continue
        raise_for(response)  # ORDER_REJECTED, INVALID_REQUEST, QUOTA_EXCEEDED, …
    raise RuntimeError(f"gave up after {attempts} attempts; retrying later with key {key} is still safe")


def main():
    login, server = os.environ["MT5_LOGIN"], os.environ["MT5_SERVER"]
    symbol = os.environ.get("MT5_SYMBOL", "EURUSD")

    account = connect_account(login, server, os.environ["MT5_PASSWORD"])
    account_id = account["id"]
    print(f"account {account_id} ({account['state']})")

    bring_online(account_id)
    print("online")

    try:
        order = send_order(f"/v1/accounts/{account_id}/orders/market",
                           {"symbol": symbol, "side": "buy", "volume": "0.01"})
    except ApiError as error:
        if error.code == "ORDER_REJECTED":
            print(f"rejected by the broker: {error}")
            return 1
        raise
    print(f"order {order['id']}: {order['state']} at {order['filledPrice']}")
    if order["state"] not in ("filled", "partially_filled"):
        return 1

    # Positions refresh on their own about every 15 s; reconcile (optional) refreshes them now.
    session.post(f"{API}/v1/accounts/{account_id}/reconcile", timeout=60)
    positions = session.get(f"{API}/v1/accounts/{account_id}/positions", timeout=30).json()["data"]
    for p in positions:
        print(f"  #{p['brokerPositionId']} {p['side']} {p['volume']} {p['symbol']} "
              f"@ {p['openPrice']} profit {p['profit']} (seen {p['observedAt']})")

    ticket = order["brokerPositionId"]
    close = send_order(f"/v1/accounts/{account_id}/positions/{ticket}/close", {})
    print(f"closed: {close['state']} at {close['filledPrice']}")

    # Offline now rather than after the idle window. The stored password is kept.
    session.post(f"{API}/v1/accounts/{account_id}/cool", timeout=30)

    if os.environ.get("FXAPIS_DISCONNECT") == "1":
        # Erases the stored password for good. Only when you are done with this account.
        result = session.post(f"{API}/v1/accounts/{account_id}/disconnect", timeout=30).json()["data"]
        print(f"disconnected, credentials removed: {result['credentialsRemoved']}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

account 0d7c1f5e-4b6a-4c1e-9f0a-2d9e8c7b6a51 (created)
online
order b1c7…9e2a: filled at 1.13426
  #2109584418 buy 0.01 EURUSD @ 1.13426 profit -0.07 (seen 2026-09-30T10:14:03.512Z)
closed: filled at 1.13419

session.headers["Authorization"] = f"Bearer {os.environ['FXAPIS_KEY']}"