6181253459

Committed 08 Sep 2023 06:04AM UTC coverage: 94.614% (+0.06%) from 94.556%

Build # 6181253459

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Merge pull request #9159 from stash86/fix-adjust

remove old codes when we only can do partial entries

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

19114 of 20202 relevant lines covered (94.61%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.61

/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.detect_environment import running_in_docker
from freqtrade.configuration.directory_operations import chown_user_directory
from freqtrade.constants import UNLIMITED_STAKE_AMOUNT
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):",
            "default": 'USD',
        },
        {
            "type": "select",
            "name": "exchange_name",
            "message": "Select exchange",
            "choices": [
                "binance",
                "binanceus",
                "bittrex",
                "gate",
                "huobi",
                "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)

1	import logging	1✔
2	import secrets	1✔
3	from pathlib import Path	1✔
4	from typing import Any, Dict, List	1✔
5
6	from questionary import Separator, prompt	1✔
7
8	from freqtrade.configuration.detect_environment import running_in_docker	1✔
9	from freqtrade.configuration.directory_operations import chown_user_directory	1✔
10	from freqtrade.constants import UNLIMITED_STAKE_AMOUNT	1✔
11	from freqtrade.exceptions import OperationalException	1✔
12	from freqtrade.exchange import MAP_EXCHANGE_CHILDCLASS, available_exchanges	1✔
13	from freqtrade.util import render_template	1✔
14
15
16	logger = logging.getLogger(__name__)	1✔
17
18
19	def validate_is_int(val):	1✔
20	try:	1✔
21	_ = int(val)	1✔
22	return True	1✔
23	except Exception:	1✔
24	return False	1✔
25
26
27	def validate_is_float(val):	1✔
28	try:	1✔
29	_ = float(val)	1✔
30	return True	1✔
31	except Exception:	1✔
32	return False	1✔
33
34
35	def ask_user_overwrite(config_path: Path) -> bool:	1✔
36	questions = [	1✔
37	{
38	"type": "confirm",
39	"name": "overwrite",
40	"message": f"File {config_path} already exists. Overwrite?",
41	"default": False,
42	},
43	]
44	answers = prompt(questions)	1✔
45	return answers['overwrite']	1✔
46
47
48	def ask_user_config() -> Dict[str, Any]:	1✔
49	"""
50	Ask user a few questions to build the configuration.
51	Interactive questions built using https://github.com/tmbo/questionary
52	:returns: Dict with keys to put into template
53	"""
54	questions: List[Dict[str, Any]] = [	1✔
55	{
56	"type": "confirm",
57	"name": "dry_run",
58	"message": "Do you want to enable Dry-run (simulated trades)?",
59	"default": True,
60	},
61	{
62	"type": "text",
63	"name": "stake_currency",
64	"message": "Please insert your stake currency:",
65	"default": 'USDT',
66	},
67	{
68	"type": "text",
69	"name": "stake_amount",
70	"message": f"Please insert your stake amount (Number or '{UNLIMITED_STAKE_AMOUNT}'):",
71	"default": "unlimited",
72	"validate": lambda val: val == UNLIMITED_STAKE_AMOUNT or validate_is_float(val),
73	"filter": lambda val: '"' + UNLIMITED_STAKE_AMOUNT + '"'
74	if val == UNLIMITED_STAKE_AMOUNT
75	else val
76	},
77	{
78	"type": "text",
79	"name": "max_open_trades",
80	"message": "Please insert max_open_trades (Integer or -1 for unlimited open trades):",
81	"default": "3",
82	"validate": lambda val: validate_is_int(val)
83	},
84	{
85	"type": "select",
86	"name": "timeframe_in_config",
87	"message": "Time",
88	"choices": ["Have the strategy define timeframe.", "Override in configuration."]
89	},
90	{
91	"type": "text",
92	"name": "timeframe",
93	"message": "Please insert your desired timeframe (e.g. 5m):",
94	"default": "5m",
95	"when": lambda x: x["timeframe_in_config"] == 'Override in configuration.'
96
97	},
98	{
99	"type": "text",
100	"name": "fiat_display_currency",
101	"message": "Please insert your display Currency (for reporting):",
102	"default": 'USD',
103	},
104	{
105	"type": "select",
106	"name": "exchange_name",
107	"message": "Select exchange",
108	"choices": [
109	"binance",
110	"binanceus",
111	"bittrex",
112	"gate",
113	"huobi",
114	"kraken",
115	"kucoin",
116	"okx",
117	Separator("------------------"),
118	"other",
119	],
120	},
121	{
122	"type": "confirm",
123	"name": "trading_mode",
124	"message": "Do you want to trade Perpetual Swaps (perpetual futures)?",
125	"default": False,
126	"filter": lambda val: 'futures' if val else 'spot',
127	"when": lambda x: x["exchange_name"] in ['binance', 'gate', 'okx'],
128	},
129	{
130	"type": "autocomplete",
131	"name": "exchange_name",
132	"message": "Type your exchange name (Must be supported by ccxt)",
133	"choices": available_exchanges(),
134	"when": lambda x: x["exchange_name"] == 'other'
135	},
136	{
137	"type": "password",
138	"name": "exchange_key",
139	"message": "Insert Exchange Key",
140	"when": lambda x: not x['dry_run']
141	},
142	{
143	"type": "password",
144	"name": "exchange_secret",
145	"message": "Insert Exchange Secret",
146	"when": lambda x: not x['dry_run']
147	},
148	{
149	"type": "password",
150	"name": "exchange_key_password",
151	"message": "Insert Exchange API Key password",
152	"when": lambda x: not x['dry_run'] and x['exchange_name'] in ('kucoin', 'okx')
153	},
154	{
155	"type": "confirm",
156	"name": "telegram",
157	"message": "Do you want to enable Telegram?",
158	"default": False,
159	},
160	{
161	"type": "password",
162	"name": "telegram_token",
163	"message": "Insert Telegram token",
164	"when": lambda x: x['telegram']
165	},
166	{
167	"type": "password",
168	"name": "telegram_chat_id",
169	"message": "Insert Telegram chat id",
170	"when": lambda x: x['telegram']
171	},
172	{
173	"type": "confirm",
174	"name": "api_server",
175	"message": "Do you want to enable the Rest API (includes FreqUI)?",
176	"default": False,
177	},
178	{
179	"type": "text",
180	"name": "api_server_listen_addr",
181	"message": ("Insert Api server Listen Address (0.0.0.0 for docker, "
182	"otherwise best left untouched)"),
183	"default": "127.0.0.1" if not running_in_docker() else "0.0.0.0",
184	"when": lambda x: x['api_server']
185	},
186	{
187	"type": "text",
188	"name": "api_server_username",
189	"message": "Insert api-server username",
190	"default": "freqtrader",
191	"when": lambda x: x['api_server']
192	},
193	{
194	"type": "password",
195	"name": "api_server_password",
196	"message": "Insert api-server password",
197	"when": lambda x: x['api_server']
198	},
199	]
200	answers = prompt(questions)	1✔
201
202	if not answers:	1✔
203	# Interrupted questionary sessions return an empty dict.
204	raise OperationalException("User interrupted interactive questions.")	1✔
205	# Ensure default is set for non-futures exchanges
206	answers['trading_mode'] = answers.get('trading_mode', "spot")	1✔
207	answers['margin_mode'] = (	1✔
208	'isolated'
209	if answers.get('trading_mode') == 'futures'
210	else ''
211	)
212	# Force JWT token to be a random string
213	answers['api_server_jwt_key'] = secrets.token_hex()	1✔
214	answers['api_server_ws_token'] = secrets.token_urlsafe(25)	1✔
215
216	return answers	1✔
217
218
219	def deploy_new_config(config_path: Path, selections: Dict[str, Any]) -> None:	1✔
220	"""
221	Applies selections to the template and writes the result to config_path
222	:param config_path: Path object for new config file. Should not exist yet
223	:param selections: Dict containing selections taken by the user.
224	"""
225	from jinja2.exceptions import TemplateNotFound	1✔
226	try:	1✔
227	exchange_template = MAP_EXCHANGE_CHILDCLASS.get(	1✔
228	selections['exchange_name'], selections['exchange_name'])
229
230	selections['exchange'] = render_template(	1✔
231	templatefile=f"subtemplates/exchange_{exchange_template}.j2",
232	arguments=selections
233	)
234	except TemplateNotFound:	×
235	selections['exchange'] = render_template(	×
236	templatefile="subtemplates/exchange_generic.j2",
237	arguments=selections
238	)
239
240	config_text = render_template(templatefile='base_config.json.j2',	1✔
241	arguments=selections)
242
243	logger.info(f"Writing config to `{config_path}`.")	1✔
244	logger.info(	1✔
245	"Please make sure to check the configuration contents and adjust settings to your needs.")
246
247	config_path.write_text(config_text)	1✔
248
249
250	def start_new_config(args: Dict[str, Any]) -> None:	1✔
251	"""
252	Create a new strategy from a template
253	Asking the user questions to fill out the template accordingly.
254	"""
255
256	config_path = Path(args['config'][0])	1✔
257	chown_user_directory(config_path.parent)	1✔
258	if config_path.exists():	1✔
259	overwrite = ask_user_overwrite(config_path)	1✔
260	if overwrite:	1✔
261	config_path.unlink()	1✔
262	else:
263	raise OperationalException(	1✔
264	f"Configuration file `{config_path}` already exists. "
265	"Please delete it or use a different configuration file name.")
266	selections = ask_user_config()	1✔
267	deploy_new_config(config_path, selections)	1✔

freqtrade / freqtrade / 6181253459

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous