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

188 lines
5.6 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 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")
COMMANDS_NO_ARGS = ["start",
"stop",
"stopbuy",
"reload_conf",
]
INFO_COMMANDS = {"version": [],
"count": [],
"daily": ["timescale"],
"profit": [],
"status": [],
"balance": []
}
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)
# recombine url
url = urlunparse((schema, netloc, path, par, query, fragment))
print(url)
try:
resp = self.session.request(method, url, headers=hd, data=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 _call_command_noargs(self, command):
logger.info(f"Running command `{command}` at {self.serverurl}")
r = self._call("POST", command)
logger.info(r)
def _call_info(self, command, command_args):
logger.info(
f"Running command `{command}` with parameters `{command_args}` at {self.serverurl}")
args = INFO_COMMANDS[command]
if len(args) < len(command_args):
logger.error(f"Command {command} does only support {len(args)} arguments.")
return
params = {}
for idx, arg in enumerate(command_args):
params[args[idx]] = arg
logger.debug(params)
r = self._call("GET", command, params)
logger.info(r)
def version(self):
"""
Returns the version of the bot
:returns: json object containing the version
"""
return self._get("version")
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 add_arguments():
parser = argparse.ArgumentParser()
parser.add_argument("command",
help="Positional argument defining the command to execute.")
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):
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"]))
# Call commands without arguments
# if args["command"] in COMMANDS_NO_ARGS:
# client._call_command_noargs(args["command"])
# if args["command"] in INFO_COMMANDS:
# client._call_info(args["command"], args["command_arguments"])
if __name__ == "__main__":
args = add_arguments()
main(args)