dxtrade_api.py

'''
Implement DxTrade API

If using a different broker than FTMO you must modify self.base_url below

You can run this module standalone to execute the test routines in main()
but you must first enter your demo account credentials in line:
ftmo_conn = DXT("accountnumber", "password")

'''

import requests
import json
import uuid
import time	# for time.sleep() used only for testing ( main() )
import logging

class DXT:
    def __init__(self, username, password):
        self.token = None
        self.account_id = None
        self.username = username
        self.password = password
        self.server = "ftmo"
        self.cookies = {}
        self.s = requests.Session()
        self.base_url = "https://dxtrade.ftmo.com/dxsca-web/"
        
        logging.basicConfig(
	    level=logging.INFO,
	    format="%(asctime)s [%(levelname)s] %(message)s",
	    handlers=[ logging.FileHandler("TradeCopier-O2D.log"), logging.StreamHandler() ]
        )

    def login(self):
        url = self.base_url + "login"
        payload = json.dumps({
            "username": self.username,
            "domain": "default", 
            "password": self.password
        })
        headers = {
            'Content-Type': 'application/json',
            'Accept': 'application/json'
        }
        response = self.s.request("POST", url, headers=headers, data=payload)
        if response.status_code == 200:
            self.token = response.json()["sessionToken"]
            return True
        else:
            logging.critical("Login failed with status code: %s", response.status_code)
            return False;

    def ping(self):
        url = self.base_url + "ping"
        headers = {
            'Content-Type': 'application/json',
            'Accept': 'application/json',
            'Authorization': 'DXAPI ' + self.token
        }
        response = self.s.request("POST", url, headers=headers)
        if response.status_code == 200:
            #logging.info("Ping successful")
            return True
        else:
            logging.error("Ping failed with status code: %s", response.status_code)
            return False

    def account_balance(self):
        url = self.base_url + "accounts/default:" + self.username + "/metrics?include-positions=false"
        headers = {
            'Content-Type': 'application/json',
            'Accept': 'application/json',
            'Authorization': 'DXAPI ' + self.token
        }
        response = self.s.request("GET", url, headers=headers)
        if response.status_code == 200:
            # returns a dict named 'metrics' containing a list of one item, a dict
            return response.json()['metrics'][0]['balance']
        else:
            logging.error("account balance query failed with status code: %s", response.status_code)
            return None

    def get_quote(self, currpair):
        url = self.base_url + "marketdata"
        print(f"URL:{url}")
        payload = json.dumps({
            "account": "default:" + self.username,
            "symbols": [ currpair.replace("_","") ],
            "eventTypes": [ { 
                "type": "Quote",
                "format": "COMPACT"
            } ]
        })
        headers = {
            'Content-Type': 'application/json',
            'Accept': 'application/json',
            'Authorization': 'DXAPI ' + self.token
        }
        response = self.s.request("POST", url, headers=headers, data=payload)
        if response.status_code == 200:
            # returns a dict named 'metrics' containing a list of one item, a dict
            # return bid and ask prices
            return response.json()['events'][0]['bid'], response.json()['events'][0]['ask']
        else:
            logging.error("get-quote query failed with status code: %s", response.status_code)
            return None

    '''
    placeOrder is called to open a new order or close an existing order
    when opening:
        positionCode is not given, it is generated by this function and returned
    when closing:
        positionCode of the original order is required, buy/sell must be reversed
    '''
    def placeOrder(self, currpair, lotsize, buy_or_sell, open_or_close, positionCode):
        url = self.base_url + "accounts/default:" + self.username + "/orders"
        orderCode = str(uuid.uuid4()).replace("-","")
        payload = json.dumps({
            "account": "default:" + self.username,
            "orderCode": orderCode,
            "positionCode": positionCode,
            "type": "MARKET",
            "instrument": currpair.replace("_",""),
            "quantity": lotsize,
            "positionEffect": "OPEN" if open_or_close else "CLOSE",
            "side": "BUY" if buy_or_sell == 1 else "SELL",
            "tif": "GTC"
        })
        headers = {
            'Content-Type': 'application/json',
            'Accept': 'application/json',
            'Authorization': 'DXAPI ' + self.token
        }
        response = self.s.request("POST", url, headers=headers, data=payload)
        if response.status_code == 200:
            # clean this up
            positionCode = response.json()["orderId"]
            logging.info("Order successful, order code: %s, position code: %s", orderCode, positionCode)
            return orderCode, str(positionCode)
        else:
            logging.error("Order failed with status code: %s", response.status_code)
            return None, None
            
    def openOrder(self, currpair, lotsize, buy_or_sell):
        orderCode, positionCode = self.placeOrder(currpair=currpair, lotsize=lotsize, buy_or_sell=buy_or_sell, open_or_close=1, positionCode="")
        return positionCode

    def closeOrder(self, currpair, lotsize, buy_or_sell, positionCode):
        # note buy_or_sell is inverted
        orderCode, positionCode = self.placeOrder(currpair=currpair, lotsize=lotsize, buy_or_sell=not buy_or_sell, open_or_close=0, positionCode=positionCode)
        return False if orderCode is None else True
        
    def get_positions(self):
        url = self.base_url + "accounts/default:" + self.username + "/positions"
        headers = {
            'Content-Type': 'application/json',
            'Accept': 'application/json',
            'Authorization': 'DXAPI ' + self.token
        }
        response = self.s.request("GET", url, headers=headers)
        if response.status_code == 200:
            # returns a dict named 'positions' containing a list of positions, each position is a dict
            order_list = response.json()['positions']
            # create a simplified list-of-dicts to return the results
            positions = []
            for order_item in order_list:
                positions.append({ 
                    "symbol": order_item["symbol"], 
                    "positionCode": order_item["positionCode"], 
                    "side": 1 if order_item["side"] == "BUY" else 0,
                    "quantity": order_item["quantity"]
                })
            return positions
        else:
            logging.error("query failed with status code: %s", response.status_code)
            return None

    def close_all(self):
        for position in self.get_positions():
            logging.info("close_all closing order for symbol %s position %s side %s", position['symbol'], position['positionCode'], position['side'])
            self.closeOrder(
                currpair=position['symbol'], 
                lotsize=position['quantity'], 
                buy_or_sell=position['side'], 
                positionCode=position['positionCode']
            )

if __name__ == "__main__":
    ftmo_conn = DXT("accountnumber", "password")
    ftmo_conn.login()
    ftmo_conn.ping()
    print(f"FTMO balance:{ftmo_conn.account_balance()}")
    
    # get a quote
    bid,ask = ftmo_conn.get_quote("EURUSD")
    print(f"Quote for EURUSD - Bid:{bid}, Ask:{ask}")
    
    # create these 3 orders
    positions = {}
    for order_param in [ ("EURCAD", 5000, 0), ("EURUSD", 6000, 1), ("EURGBP", 7000, 1) ]:
        symbol, lotsize, side = order_param
        positionCode = ftmo_conn.openOrder(symbol, lotsize, side)
        if positionCode is not None:
            positions[symbol] = (lotsize, side, positionCode)
    time.sleep(10)
    
    curr_positions = ftmo_conn.get_positions()
    print(f"After opening orders: {len(curr_positions)} open positions:")
    for list_item in curr_positions:
        print(list_item)

    # close 1 order        
    if "EURCAD" in positions:
        (lotsize, side, positionCode) = positions["EURCAD"]
        ftmo_conn.closeOrder("EURCAD", lotsize, side, positionCode)
    time.sleep(10)
        
    curr_positions = ftmo_conn.get_positions()
    print(f"After closing 1 order: {len(curr_positions)} open positions:")
    for list_item in curr_positions:
        print(list_item)
        
    ftmo_conn.close_all()
    time.sleep(10)
    
    curr_positions = ftmo_conn.get_positions()
    print(f"After closing all orders: {len(curr_positions)} open positions:")
    for list_item in curr_positions:
        print(list_item)