Need help using json API of a broker with python [Xtb

I am trying to create a gui where I can log in to my XTB account. The broker has an API: http://developers.xstore.pro/documentation/#getAllSymbols

I have found a login function someone else wrote on stackoverflow a couple of years ago. I have divided it for parts where I know what is going on (this is the very, very first time I use json, ssl and socket modules, so I'm kinda white about it all).

def xtb_log_in(self):
    self.user_id_xtb = self.xtb_login_entry.get()
    self.password_xtb = self.xtb_password_entry.get()


    ###################################################
    # preparing connection
    host = 'xapi.xtb.com'
    port = 5112
    host = socket.getaddrinfo(host, port)[0][4][0]
    s = socket.socket()
    s.connect((host, port))
    s = ssl.wrap_socket(s)
    ###################################################

    ###################################################
    # sending command parameters
    parameters = {
        "command": "login",
        "arguments": {
            "userId": self.user_id_xtb,
            "password": self.password_xtb
        }
    }
    packet = json.dumps(parameters, indent=4)
    s.send(packet.encode("UTF-8"))
    ###################################################

    END = b'\n\n'
    response = s.recv(8192)
    if END in response:
        print('Print login: {}'.format(response[:response.find(END)]))

    ###################################################
    # sending command parameters
    parameters = {
        "command": "logout"
    }
    packet = json.dumps(parameters, indent=4)
    s.send(packet.encode("UTF-8"))
    ###################################################

    response = s.recv(8192)
    if END in response:
        print('Print logout: {}'.format(response[:response.find(END)]))

Now when I tried to do the same with getallsymbols command I think I have failed. Experimentally I have attached these lines to the function above:

parameters = {
    "command": "getAllSymbols"
}
packet = json.dumps(parameters, indent=1)
s.send(packet.encode("UTF-8"))
response = s.recv(8192)

But unfortunately it didn't work - I think. How do I use this and other commands properly and get access to the data they are returning? I just need a brief explanation of how this things work. I would be grateful if someone fixed this:

parameters = {
    "command": "getAllSymbols"
}
packet = json.dumps(parameters, indent=1)
s.send(packet.encode("UTF-8"))
response = s.recv(8192)

And would explain to me what is going on in there.

CodePudding user response：

I tested this code with my old account and it works for me.

But it needs account number from xStation instead of login but password is the same as for login to xStation

First it needs to login.

Next it needs to run getAllSymbols or other function.

And finally it needs logout

This code works for me. You have to put your account number and password

import socket
import ssl
import json

###################################################
# preparing connection
host = 'xapi.xtb.com'
port = 5112
host = socket.getaddrinfo(host, port)[0][4][0]
s = socket.socket()
s.connect((host, port))
s = ssl.wrap_socket(s)

END = b'\n\n'
###################################################


###################################################
# sending command parameters
parameters = {
    "command": "login",
    "arguments": {
        "userId":   '19xxxxx',     # account number
        "password": 'xxxxxxxxxx'
    }
}
packet = json.dumps(parameters)
s.send(packet.encode("UTF-8"))

# -------------------------------------------------

response = s.recv(8192)
if END in response:
    print('login:', response[:response.find(END)])
else:
    print('login:', response)

###################################################
# sending command parameters
parameters = {
    "command": "getAllSymbols"
}

packet = json.dumps(parameters)
s.send(packet.encode("UTF-8"))

# -------------------------------------------------

response = s.recv(8192)
if END in response:
    print('getAllSymbols:', response[:response.find(END)])
else:
    print('getAllSymbols:', response)

###################################################
# sending command parameters
parameters = {
    "command": "getSymbol",
    "arguments": {
        "symbol": "EURPLN"
    }
}

# -------------------------------------------------

response = s.recv(8192)
if END in response:
    print('getSymbol:', response[:response.find(END)])
else:
    print('getSymbol:', response)

###################################################
# sending command parameters
parameters = {
    "command": "logout"
}
packet = json.dumps(parameters)
s.send(packet.encode("UTF-8"))

# -------------------------------------------------

response = s.recv(8192)
if END in response:
    print('logout:', response[:response.find(END)])
else:
    print('logout:', response)

###################################################

EDIT:

Code in functions:

import socket
import ssl
import json

# --- constants ---

HOST = 'xapi.xtb.com'
PORT = 5112

END = b'\n\n'

# --- functions ---

def send(parameters):
    packet = json.dumps(parameters)
    s.send(packet.encode("UTF-8"))

    response = b''
    
    while True:
        response  = s.recv(8192)
        if END in response:
            break
        
    return json.loads(response[:response.find(END)])

def login(user_id, password):
        
    parameters = {
        "command": "login",
        "arguments": {
            "userId":   user_id, 
            "password": password
        }
    }

    return send(parameters)
    
def logout():
        
    parameters = {
        "command": "logout",
    }

    return send(parameters)

def getAllSymbols():

    parameters = {
        "command": "getAllSymbols"
    }

    return send(parameters)

def getSymbol(symbol):
    parameters = {
        "command": "getSymbol",
        "arguments": {
            "symbol": symbol
        }
    }

    return send(parameters)

# --- main ---

#HOST = socket.getaddrinfo(HOST, PORT)[0][4][0]
s = socket.socket()
s.connect((HOST, PORT))
s = ssl.wrap_socket(s)

# ---

print('login ...')

response = login('19xxxxx', 'xxxxxxxxxx')
if response['status'] is False:
    print('Problem:', response)
    exit()
    
# ---

print('getAllSymbols ...')

response = getAllSymbols()    
if response['status'] is False:
    print('Problem:', response)
    exit()

print('getAllSymbols:', response)
data = response['returnData']
# ... do something with data ...

# ---

print('getSymbol ...')

response = getSymbol("EURPLN")    
if response['status'] is False:
    print('Problem:', response)
    exit()

print('getSymbol:', response)
data = response['returnData']
# ... do something with data ...

# ---

print('logout ...')

logout()

# ---

s.close()

BTW:

In API documentation at the top of page you can see Download wrappers where you can get Python module.