stable/scripts/rest_client.py

263 lines
7.1 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")
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}/api/v1/{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 ""
# 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 start(self):
"""
Start the bot if it's in stopped state.
:returns: json object
"""
return self._post("start")
def stop(self):
"""
Stop the bot. Use start to restart
:returns: json object
"""
return self._post("stop")
def stopbuy(self):
"""
Stop buying (but handle sells gracefully).
use reload_conf to reset
:returns: json object
"""
return self._post("stopbuy")
def reload_conf(self):
"""
Reload configuration
:returns: json object
"""
return self._post("reload_conf")
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 edge(self):
"""
Returns information about edge
:returns: json object
"""
return self._get("edge")
def profit(self):
"""
Returns the profit summary
:returns: json object
"""
return self._get("profit")
def performance(self):
"""
Returns the performance of the different coins
:returns: json object
"""
return self._get("performance")
def status(self):
"""
Get the status of open trades
:returns: json object
"""
return self._get("status")
def version(self):
"""
Returns the version of the bot
:returns: json object containing the version
"""
return self._get("version")
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 forcebuy(self, pair, price=None):
"""
Buy an asset
:param pair: Pair to buy (ETH/BTC)
:param price: Optional - price to buy
:returns: json object of the trade
"""
data = {"pair": pair,
"price": price
}
return self._post("forcebuy", data=data)
def forcesell(self, tradeid):
"""
Force-sell a trade
:param tradeid: Id of the trade (can be received via status command)
:returns: json object
"""
return self._post("forcesell", data={"tradeid": tradeid})
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()
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 print_commands():
# Print dynamic help for the different commands using the commands doc-strings
client = FtRestClient(None)
print("Possible commands:")
for x, y in inspect.getmembers(client):
if not x.startswith('_'):
print(f"{x} {getattr(client, x).__doc__}")
def main(args):
if args.get("help"):
print_commands()
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")
print_commands()
return
print(getattr(client, command)(*args["command_arguments"]))
if __name__ == "__main__":
args = add_arguments()
main(args)