stable/scripts/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 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 ""
        # 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 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 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)