rest_client.py

#!/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, username=None, password=None):

        self._serverurl = serverurl
        self._session = requests.Session()
        self._session.auth = (username, password)

    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))
            # 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.
        :return: json object
        """
        return self._post("start")

    def stop(self):
        """
        Stop the bot. Use start to restart
        :return: json object
        """
        return self._post("stop")

    def stopbuy(self):
        """
        Stop buying (but handle sells gracefully).
        use reload_conf to reset
        :return: json object
        """
        return self._post("stopbuy")

    def reload_conf(self):
        """
        Reload configuration
        :return: json object
        """
        return self._post("reload_conf")

    def balance(self):
        """
        Get the account balance
        :return: json object
        """
        return self._get("balance")

    def count(self):
        """
        Returns the amount of open trades
        :return: json object
        """
        return self._get("count")

    def daily(self, days=None):
        """
        Returns the amount of open trades
        :return: json object
        """
        return self._get("daily", params={"timescale": days} if days else None)

    def edge(self):
        """
        Returns information about edge
        :return: json object
        """
        return self._get("edge")

    def profit(self):
        """
        Returns the profit summary
        :return: json object
        """
        return self._get("profit")

    def performance(self):
        """
        Returns the performance of the different coins
        :return: json object
        """
        return self._get("performance")

    def status(self):
        """
        Get the status of open trades
        :return: json object
        """
        return self._get("status")

    def version(self):
        """
        Returns the version of the bot
        :return: json object containing the version
        """
        return self._get("version")

    def whitelist(self):
        """
        Show the current whitelist
        :return: json object
        """
        return self._get("whitelist")

    def blacklist(self, *args):
        """
        Show the current blacklist
        :param add: List of coins to add (example: "BNB/BTC")
        :return: 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
        :return: 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)
        :return: 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")
    username = config.get("api_server", {}).get("username")
    password = config.get("api_server", {}).get("password")

    server_url = f"http://{url}:{port}"
    client = FtRestClient(server_url, username, password)

    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
    response = getattr(client, command)(*args["command_arguments"])
    return response


if __name__ == "__main__":
    args = add_arguments()
    main(args)