#!/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}/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() # 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 print_commands(): # 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__}") 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)