stable/scripts/rest_client.py
2019-05-18 09:57:10 +02:00

203 lines
5.8 KiB
Python
Executable File

#!/usr/bin/env python3
"""
Simple command line client into RPC commands
Can be used as an alternate to Telegram
Should not import anything from freqtrade,
so it can be used as a standalone script.
"""
import argparse
import json
import logging
import inspect
from typing import List
from urllib.parse import urlencode, urlparse, urlunparse
from pathlib import Path
import requests
from requests.exceptions import ConnectionError
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
)
logger = logging.getLogger("ft_rest_client")
class FtRestClient():
def __init__(self, serverurl):
self._serverurl = serverurl
self._session = requests.Session()
def _call(self, method, apipath, params: dict = None, data=None, files=None):
if str(method).upper() not in ('GET', 'POST', 'PUT', 'DELETE'):
raise ValueError('invalid method <{0}>'.format(method))
basepath = f"{self._serverurl}/{apipath}"
hd = {"Accept": "application/json",
"Content-Type": "application/json"
}
# Split url
schema, netloc, path, par, query, fragment = urlparse(basepath)
# URLEncode query string
query = urlencode(params) if params else None
# recombine url
url = urlunparse((schema, netloc, path, par, query, fragment))
try:
resp = self._session.request(method, url, headers=hd, data=json.dumps(data),
# auth=self.session.auth
)
# return resp.text
return resp.json()
except ConnectionError:
logger.warning("Connection error")
def _get(self, apipath, params: dict = None):
return self._call("GET", apipath, params=params)
def _post(self, apipath, params: dict = None, data: dict = None):
return self._call("POST", apipath, params=params, data=data)
def balance(self):
"""
get the account balance
:returns: json object
"""
return self._get("balance")
def count(self):
"""
Returns the amount of open trades
:returns: json object
"""
return self._get("count")
def daily(self, days=None):
"""
Returns the amount of open trades
:returns: json object
"""
return self._get("daily", params={"timescale": days} if days else None)
def profit(self):
"""
Returns the profit summary
:returns: json object
"""
return self._get("profit")
def status(self):
"""
Get the status of open trades
:returns: json object
"""
return self._get("status")
def whitelist(self):
"""
Show the current whitelist
:returns: json object
"""
return self._get("whitelist")
def blacklist(self, *args):
"""
Show the current blacklist
:param add: List of coins to add (example: "BNB/BTC")
:returns: json object
"""
if not args:
return self._get("blacklist")
else:
return self._post("blacklist", data={"blacklist": args})
def version(self):
"""
Returns the version of the bot
:returns: json object containing the version
"""
return self._get("version")
def add_arguments():
parser = argparse.ArgumentParser()
parser.add_argument("command",
help="Positional argument defining the command to execute.")
parser.add_argument('--show',
help='Show possible methods with this client',
dest='show',
action='store_true',
default=False
)
parser.add_argument('-c', '--config',
help='Specify configuration file (default: %(default)s). ',
dest='config',
type=str,
metavar='PATH',
default='config.json'
)
parser.add_argument("command_arguments",
help="Positional arguments for the parameters for [command]",
nargs="*",
default=[]
)
args = parser.parse_args()
# if len(argv) == 1:
# print('\nThis script accepts the following arguments')
# print('- daily (int) - Where int is the number of days to report back. daily 3')
# print('- start - this will start the trading thread')
# print('- stop - this will start the trading thread')
# print('- there will be more....\n')
return vars(args)
def load_config(configfile):
file = Path(configfile)
if file.is_file():
with file.open("r") as f:
config = json.load(f)
return config
return {}
def main(args):
if args.get("show"):
# Print dynamic help for the different commands
client = FtRestClient(None)
print("Possible commands:")
for x, y in inspect.getmembers(client):
if not x.startswith('_'):
print(f"{x} {getattr(client, x).__doc__}")
return
config = load_config(args["config"])
url = config.get("api_server", {}).get("server_url", "127.0.0.1")
port = config.get("api_server", {}).get("listen_port", "8080")
server_url = f"http://{url}:{port}"
client = FtRestClient(server_url)
m = [x for x, y in inspect.getmembers(client) if not x.startswith('_')]
command = args["command"]
if command not in m:
logger.error(f"Command {command} not defined")
return
print(getattr(client, command)(*args["command_arguments"]))
if __name__ == "__main__":
args = add_arguments()
main(args)