Build an MT5 trade copier with the API

every 2 s:
  POST /v1/accounts/{master}/reconcile        pull the master's latest deals and positions
  GET  /v1/accounts/{master}/deals?since=…    what happened since last time
  GET  /v1/accounts/{master}/positions        current stops, for SL/TP changes

  deal entry "in"   → POST /v1/execution-waves                     (one call, every follower)
  deal entry "out"  → POST /v1/accounts/{follower}/positions/{ticket}/close   (per follower)
  stop moved        → POST /v1/accounts/{follower}/positions/{ticket}/modify  (per follower)

follower volume = master deal volume × multiplier, rounded DOWN to the volume step

pip install requests
export FXAPIS_KEY="fx_test_…"
export MASTER_ACCOUNT_ID="0d7c1f5e-4b6a-4c1e-9f0a-2d9e8c7b6a51"
# follower account id = volume multiplier (for example follower balance ÷ master balance)
export FOLLOWERS="5b2e8a10-93c4-4f7d-8e61-0a9c2b4d6e13=0.5,c41d7e22-6f0b-4a95-b3e8-7d1f5a2c9b80=2"
python3 copier.py

"""A minimal MT5 trade copier on fxapis: one master, many followers.

Copies entries, full and partial closes, and stop-loss / take-profit changes.
Run it on broker DEMO accounts first. Every fxapis key reaches real brokers.
"""
import os
import sqlite3
import time
from datetime import datetime, timezone
from decimal import ROUND_DOWN, Decimal

import requests

API = os.environ.get("FXAPIS_API", "https://api.fxapis.com")
MASTER = os.environ["MASTER_ACCOUNT_ID"]
# follower account id -> volume multiplier, e.g. "id1=0.5,id2=2"
FOLLOWERS = {
    account_id: Decimal(multiplier)
    for account_id, multiplier in (item.split("=") for item in os.environ["FOLLOWERS"].split(","))
}
VOLUME_STEP = Decimal("0.01")   # check your symbols' volume step and minimum
MIN_VOLUME = Decimal("0.01")
POLL_SECONDS = 2
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']}"

db = sqlite3.connect("copier.db")
db.executescript("""
create table if not exists handled_deals (deal_id text primary key);
create table if not exists links (          -- a follower position copied from a master position
    master_position text, follower text, follower_position text, volume text,
    primary key (master_position, follower));
create table if not exists pending (        -- follower orders whose result is still unknown
    order_id text primary key, master_position text, follower text, volume text);
create table if not exists stops (master_position text primary key, stop_loss text, take_profit text);
create table if not exists cursor (id integer primary key check (id = 1), since text);
""")


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


def api(method, path, attempts=5, **kwargs):
    """Reads and non-order writes: retried on 429 and 5xx, which is safe for them."""
    for attempt in range(attempts):
        try:
            r = session.request(method, f"{API}{path}", timeout=60, **kwargs)
        except requests.RequestException:
            time.sleep(min(2 ** attempt, 10))
            continue
        if r.status_code < 300:
            return r.json().get("data")
        if r.status_code == 429 or r.status_code >= 500:
            time.sleep(int(r.headers.get("retry-after", 0)) or min(2 ** attempt, 10))
            continue
        error = r.json().get("error", {})
        raise ApiError(r.status_code, error.get("code"), error.get("message"), error.get("details"))
    raise RuntimeError(f"{method} {path} kept failing")


def place(path, body, key, attempts=6):
    """Orders and closes: the same key on every retry, and ORDER_UNRESOLVED is never resent."""
    for attempt in range(attempts):
        try:
            r = session.post(f"{API}{path}", json=body, headers={"Idempotency-Key": key}, timeout=90)
        except requests.RequestException:
            time.sleep(min(2 ** attempt, 10))
            continue
        if r.status_code in (200, 201):          # 200: a multi-account order replayed by its key
            return r.json()["data"]
        try:
            error = r.json()["error"]
        except (ValueError, KeyError):
            time.sleep(min(2 ** attempt, 10))
            continue
        if error["code"] in RETRY_SAME_KEY:
            time.sleep(int(r.headers.get("retry-after", 0)) or min(2 ** attempt, 10))
            continue
        raise ApiError(r.status_code, error["code"], error["message"], error.get("details"))
    raise RuntimeError(f"gave up on {key}; it is safe to retry with the same key")


def scaled(volume, multiplier):
    lots = (Decimal(volume) * multiplier / VOLUME_STEP).to_integral_value(rounding=ROUND_DOWN) * VOLUME_STEP
    return str(lots) if lots >= MIN_VOLUME else None     # too small to copy: skip, never round up


def new_deals():
    row = db.execute("select since from cursor").fetchone()
    if row is None:
        # First start: copy from now on, not the master's history.
        since = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
        db.execute("insert into cursor (id, since) values (1, ?)", (since,))
        db.commit()
        return []
    deals, cursor = [], None
    while True:
        params = {"since": row[0], "limit": 200}
        if cursor:
            params["cursor"] = cursor
        r = session.get(f"{API}/v1/accounts/{MASTER}/deals", params=params, timeout=60)
        r.raise_for_status()
        body = r.json()
        deals.extend(body["data"])
        if not body["page"]["hasMore"]:
            break
        cursor = body["page"]["nextCursor"]
    handled = {d for (d,) in db.execute("select deal_id from handled_deals")}
    fresh = [d for d in deals if d["brokerDealId"] not in handled and d["dealType"] in ("buy", "sell")]
    return sorted(fresh, key=lambda d: (d["dealtAt"], d["brokerDealId"]))


def link(order, master_position, follower, volume):
    if order["state"] in ("filled", "partially_filled") and order.get("brokerPositionId"):
        db.execute("insert or ignore into links values (?, ?, ?, ?)",
                   (master_position, follower, order["brokerPositionId"], order.get("filledVolume") or volume))
    elif order["state"] == "unknown":
        db.execute("insert or ignore into pending values (?, ?, ?, ?)",
                   (order["id"], master_position, follower, volume))


def copy_entry(deal, master_positions):
    weights = {f: v for f, v in ((f, scaled(deal["volume"], m)) for f, m in FOLLOWERS.items()) if v}
    if not weights:
        return
    source = master_positions.get(deal["brokerPositionId"], {})
    body = {
        "accountIds": list(weights),
        "symbol": deal["symbol"],
        "side": deal["dealType"],
        "volume": next(iter(weights.values())),
        "weights": weights,
        "barrierPolicy": "release-ready",
        "clientWaveId": f"copy:{deal['brokerDealId']}",
    }
    for level in ("stopLoss", "takeProfit"):
        if source.get(level):
            body[level] = source[level]
    wave = place("/v1/execution-waves", body, key=f"copy:open:{deal['brokerDealId']}")

    while wave["state"] not in ("settled", "cancelled", "abandoned"):
        time.sleep(0.5)
        wave = api("GET", f"/v1/execution-waves/{wave['id']}")
    print(f"copied deal {deal['brokerDealId']}: {wave['summary']} spread {wave['dispatchSpreadMs']} ms")

    for leg in wave["legs"]:
        if leg["orderId"] and leg["state"] in ("filled", "unresolved"):
            order = api("GET", f"/v1/orders/{leg['orderId']}")
            link(order, deal["brokerPositionId"], leg["accountId"], leg["volume"])
        elif leg["state"] in ("rejected", "skipped"):
            print(f"  {leg['accountId']} not copied: {leg['state']} ({leg['stateDetail']})")


def copy_exit(deal, master_positions):
    master_position = deal["brokerPositionId"]
    still_open = master_positions.get(master_position)
    links = db.execute("select follower, follower_position, volume from links where master_position = ?",
                       (master_position,)).fetchall()
    for follower, ticket, volume in links:
        body = {}
        if still_open and still_open.get("volume"):
            # Partial close: close the same fraction of the follower's position.
            before = Decimal(still_open["volume"]) + Decimal(deal["volume"])
            part = scaled(volume, Decimal(deal["volume"]) / before)
            if part is None:
                continue                                  # this follower's share is below the minimum
            if Decimal(part) < Decimal(volume):
                body = {"volume": part}
        try:
            order = place(f"/v1/accounts/{follower}/positions/{ticket}/close", body,
                          key=f"copy:close:{deal['brokerDealId']}:{follower}")
        except ApiError as error:
            if error.code == "POSITION_NOT_FOUND":       # already closed (a stop hit, or closed by hand)
                db.execute("delete from links where master_position = ? and follower = ?", (master_position, follower))
                continue
            if error.code == "ORDER_UNRESOLVED":         # never resend: it is being confirmed
                print(f"  close on {follower} unresolved: order {error.details[0]['orderId']}")
                continue
            print(f"  close on {follower} failed: {error}")
            continue
        if body:
            remaining = Decimal(volume) - Decimal(order.get("filledVolume") or body["volume"])
            db.execute("update links set volume = ? where master_position = ? and follower = ?",
                       (str(remaining), master_position, follower))
        else:
            db.execute("delete from links where master_position = ? and follower = ?", (master_position, follower))


def mirror_stops(master_positions):
    for ticket, position in master_positions.items():
        levels = (position.get("stopLoss"), position.get("takeProfit"))
        row = db.execute("select stop_loss, take_profit from stops where master_position = ?", (ticket,)).fetchone()
        if row is not None and tuple(row) != levels:
            for follower, follower_position in db.execute(
                    "select follower, follower_position from links where master_position = ?", (ticket,)).fetchall():
                try:
                    # null removes a level; the same absolute prices as the master
                    api("POST", f"/v1/accounts/{follower}/positions/{follower_position}/modify",
                        json={"stopLoss": levels[0], "takeProfit": levels[1]})
                except ApiError as error:
                    print(f"  stops on {follower} not moved: {error}")
        db.execute("insert or replace into stops values (?, ?, ?)", (ticket, *levels))


def resolve_pending():
    for order_id, master_position, follower, volume in db.execute("select * from pending").fetchall():
        order = api("GET", f"/v1/orders/{order_id}")
        if order["state"] != "unknown":
            db.execute("delete from pending where order_id = ?", (order_id,))
            link(order, master_position, follower, volume)


def tick():
    api("POST", f"/v1/accounts/{MASTER}/reconcile")      # pull the master's latest deals and positions
    master_positions = {p["brokerPositionId"]: p for p in api("GET", f"/v1/accounts/{MASTER}/positions")}
    for deal in new_deals():
        if deal["entry"] == "in":
            copy_entry(deal, master_positions)
        elif deal["entry"] == "out":
            copy_exit(deal, master_positions)
        else:
            print(f"deal {deal['brokerDealId']} has entry {deal['entry']}: not copied, review it")
        db.execute("insert or ignore into handled_deals values (?)", (deal["brokerDealId"],))
        db.execute("update cursor set since = ? where id = 1", (deal["dealtAt"],))
        db.commit()
    mirror_stops(master_positions)
    resolve_pending()
    db.commit()


if __name__ == "__main__":
    while True:
        try:
            tick()
        except Exception as error:      # keep copying; the state in SQLite makes a retry safe
            print(f"tick failed: {error}")
        time.sleep(POLL_SECONDS)