freqtrade_origin/freqtrade/commands/build_config_commands.py

import logging
import secrets
from pathlib import Path
from typing import Any, Dict, List

from questionary import Separator, prompt

from freqtrade.configuration import sanitize_config
from freqtrade.configuration.config_setup import setup_utils_configuration
from freqtrade.configuration.detect_environment import running_in_docker
from freqtrade.configuration.directory_operations import chown_user_directory
from freqtrade.constants import UNLIMITED_STAKE_AMOUNT
from freqtrade.enums import RunMode
from freqtrade.exceptions import OperationalException
from freqtrade.exchange import MAP_EXCHANGE_CHILDCLASS, available_exchanges
from freqtrade.util import render_template


logger = logging.getLogger(__name__)


def validate_is_int(val):
    try:
        _ = int(val)
        return True
    except Exception:
        return False


def validate_is_float(val):
    try:
        _ = float(val)
        return True
    except Exception:
        return False


def ask_user_overwrite(config_path: Path) -> bool:
    questions = [
        {
            "type": "confirm",
            "name": "overwrite",
            "message": f"File {config_path} already exists. Overwrite?",
            "default": False,
        },
    ]
    answers = prompt(questions)
    return answers["overwrite"]


def ask_user_config() -> Dict[str, Any]:
    """
    Ask user a few questions to build the configuration.
    Interactive questions built using https://github.com/tmbo/questionary
    :returns: Dict with keys to put into template
    """
    questions: List[Dict[str, Any]] = [
        {
            "type": "confirm",
            "name": "dry_run",
            "message": "Do you want to enable Dry-run (simulated trades)?",
            "default": True,
        },
        {
            "type": "text",
            "name": "stake_currency",
            "message": "Please insert your stake currency:",
            "default": "USDT",
        },
        {
            "type": "text",
            "name": "stake_amount",
            "message": f"Please insert your stake amount (Number or '{UNLIMITED_STAKE_AMOUNT}'):",
            "default": "unlimited",
            "validate": lambda val: val == UNLIMITED_STAKE_AMOUNT or validate_is_float(val),
            "filter": lambda val: (
                '"' + UNLIMITED_STAKE_AMOUNT + '"' if val == UNLIMITED_STAKE_AMOUNT else val
            ),
        },
        {
            "type": "text",
            "name": "max_open_trades",
            "message": "Please insert max_open_trades (Integer or -1 for unlimited open trades):",
            "default": "3",
            "validate": lambda val: validate_is_int(val),
        },
        {
            "type": "select",
            "name": "timeframe_in_config",
            "message": "Time",
            "choices": ["Have the strategy define timeframe.", "Override in configuration."],
        },
        {
            "type": "text",
            "name": "timeframe",
            "message": "Please insert your desired timeframe (e.g. 5m):",
            "default": "5m",
            "when": lambda x: x["timeframe_in_config"] == "Override in configuration.",
        },
        {
            "type": "text",
            "name": "fiat_display_currency",
            "message": (
                "Please insert your display Currency for reporting "
                "(leave empty to disable FIAT conversion):"
            ),
            "default": "USD",
        },
        {
            "type": "select",
            "name": "exchange_name",
            "message": "Select exchange",
            "choices": [
                "binance",
                "binanceus",
                "gate",
                "htx",
                "kraken",
                "kucoin",
                "okx",
                Separator("------------------"),
                "other",
            ],
        },
        {
            "type": "confirm",
            "name": "trading_mode",
            "message": "Do you want to trade Perpetual Swaps (perpetual futures)?",
            "default": False,
            "filter": lambda val: "futures" if val else "spot",
            "when": lambda x: x["exchange_name"] in ["binance", "gate", "okx"],
        },
        {
            "type": "autocomplete",
            "name": "exchange_name",
            "message": "Type your exchange name (Must be supported by ccxt)",
            "choices": available_exchanges(),
            "when": lambda x: x["exchange_name"] == "other",
        },
        {
            "type": "password",
            "name": "exchange_key",
            "message": "Insert Exchange Key",
            "when": lambda x: not x["dry_run"],
        },
        {
            "type": "password",
            "name": "exchange_secret",
            "message": "Insert Exchange Secret",
            "when": lambda x: not x["dry_run"],
        },
        {
            "type": "password",
            "name": "exchange_key_password",
            "message": "Insert Exchange API Key password",
            "when": lambda x: not x["dry_run"] and x["exchange_name"] in ("kucoin", "okx"),
        },
        {
            "type": "confirm",
            "name": "telegram",
            "message": "Do you want to enable Telegram?",
            "default": False,
        },
        {
            "type": "password",
            "name": "telegram_token",
            "message": "Insert Telegram token",
            "when": lambda x: x["telegram"],
        },
        {
            "type": "password",
            "name": "telegram_chat_id",
            "message": "Insert Telegram chat id",
            "when": lambda x: x["telegram"],
        },
        {
            "type": "confirm",
            "name": "api_server",
            "message": "Do you want to enable the Rest API (includes FreqUI)?",
            "default": False,
        },
        {
            "type": "text",
            "name": "api_server_listen_addr",
            "message": (
                "Insert Api server Listen Address (0.0.0.0 for docker, "
                "otherwise best left untouched)"
            ),
            "default": "127.0.0.1" if not running_in_docker() else "0.0.0.0",
            "when": lambda x: x["api_server"],
        },
        {
            "type": "text",
            "name": "api_server_username",
            "message": "Insert api-server username",
            "default": "freqtrader",
            "when": lambda x: x["api_server"],
        },
        {
            "type": "password",
            "name": "api_server_password",
            "message": "Insert api-server password",
            "when": lambda x: x["api_server"],
        },
    ]
    answers = prompt(questions)

    if not answers:
        # Interrupted questionary sessions return an empty dict.
        raise OperationalException("User interrupted interactive questions.")
    # Ensure default is set for non-futures exchanges
    answers["trading_mode"] = answers.get("trading_mode", "spot")
    answers["margin_mode"] = "isolated" if answers.get("trading_mode") == "futures" else ""
    # Force JWT token to be a random string
    answers["api_server_jwt_key"] = secrets.token_hex()
    answers["api_server_ws_token"] = secrets.token_urlsafe(25)

    return answers


def deploy_new_config(config_path: Path, selections: Dict[str, Any]) -> None:
    """
    Applies selections to the template and writes the result to config_path
    :param config_path: Path object for new config file. Should not exist yet
    :param selections: Dict containing selections taken by the user.
    """
    from jinja2.exceptions import TemplateNotFound

    try:
        exchange_template = MAP_EXCHANGE_CHILDCLASS.get(
            selections["exchange_name"], selections["exchange_name"]
        )

        selections["exchange"] = render_template(
            templatefile=f"subtemplates/exchange_{exchange_template}.j2", arguments=selections
        )
    except TemplateNotFound:
        selections["exchange"] = render_template(
            templatefile="subtemplates/exchange_generic.j2", arguments=selections
        )

    config_text = render_template(templatefile="base_config.json.j2", arguments=selections)

    logger.info(f"Writing config to `{config_path}`.")
    logger.info(
        "Please make sure to check the configuration contents and adjust settings to your needs."
    )

    config_path.write_text(config_text)


def start_new_config(args: Dict[str, Any]) -> None:
    """
    Create a new strategy from a template
    Asking the user questions to fill out the template accordingly.
    """

    config_path = Path(args["config"][0])
    chown_user_directory(config_path.parent)
    if config_path.exists():
        overwrite = ask_user_overwrite(config_path)
        if overwrite:
            config_path.unlink()
        else:
            raise OperationalException(
                f"Configuration file `{config_path}` already exists. "
                "Please delete it or use a different configuration file name."
            )
    selections = ask_user_config()
    deploy_new_config(config_path, selections)


def start_show_config(args: Dict[str, Any]) -> None:
    config = setup_utils_configuration(args, RunMode.UTIL_EXCHANGE, set_dry=False)

    # TODO: Sanitize from sensitive info before printing

    print("Your combined configuration is:")
    config_sanitized = sanitize_config(
        config["original_config"], show_sensitive=args.get("show_sensitive", False)
    )

    from rich import print_json

    print_json(data=config_sanitized)