9394559170

Committed 26 Apr 2024 06:36AM UTC coverage: 94.656% (-0.02%) from 94.674%

Build # 9394559170

Build Type

push

github

Committed by

xmatthias

Commit Message

Loader should be passed as kwarg for clarity

Run Details

20280 of 21425 relevant lines covered (94.66%)

0.95 hits per line

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

98.04

/freqtrade/commands/deploy_commands.py

import logging
import sys
from pathlib import Path
from typing import Any, Dict, Optional, Tuple

import requests

from freqtrade.configuration import setup_utils_configuration
from freqtrade.configuration.directory_operations import copy_sample_files, create_userdata_dir
from freqtrade.constants import USERPATH_STRATEGIES
from freqtrade.enums import RunMode
from freqtrade.exceptions import ConfigurationError, OperationalException
from freqtrade.util import render_template, render_template_with_fallback


logger = logging.getLogger(__name__)


# Timeout for requests
req_timeout = 30


def start_create_userdir(args: Dict[str, Any]) -> None:
    """
    Create "user_data" directory to contain user data strategies, hyperopt, ...)
    :param args: Cli args from Arguments()
    :return: None
    """
    if "user_data_dir" in args and args["user_data_dir"]:
        userdir = create_userdata_dir(args["user_data_dir"], create_dir=True)
        copy_sample_files(userdir, overwrite=args["reset"])
    else:
        logger.warning("`create-userdir` requires --userdir to be set.")
        sys.exit(1)


def deploy_new_strategy(strategy_name: str, strategy_path: Path, subtemplate: str) -> None:
    """
    Deploy new strategy from template to strategy_path
    """
    fallback = 'full'
    attributes = render_template_with_fallback(
        templatefile=f"strategy_subtemplates/strategy_attributes_{subtemplate}.j2",
        templatefallbackfile=f"strategy_subtemplates/strategy_attributes_{fallback}.j2",
    )
    indicators = render_template_with_fallback(
        templatefile=f"strategy_subtemplates/indicators_{subtemplate}.j2",
        templatefallbackfile=f"strategy_subtemplates/indicators_{fallback}.j2",
    )
    buy_trend = render_template_with_fallback(
        templatefile=f"strategy_subtemplates/buy_trend_{subtemplate}.j2",
        templatefallbackfile=f"strategy_subtemplates/buy_trend_{fallback}.j2",
    )
    sell_trend = render_template_with_fallback(
        templatefile=f"strategy_subtemplates/sell_trend_{subtemplate}.j2",
        templatefallbackfile=f"strategy_subtemplates/sell_trend_{fallback}.j2",
    )
    plot_config = render_template_with_fallback(
        templatefile=f"strategy_subtemplates/plot_config_{subtemplate}.j2",
        templatefallbackfile=f"strategy_subtemplates/plot_config_{fallback}.j2",
    )
    additional_methods = render_template_with_fallback(
        templatefile=f"strategy_subtemplates/strategy_methods_{subtemplate}.j2",
        templatefallbackfile="strategy_subtemplates/strategy_methods_empty.j2",
    )

    strategy_text = render_template(templatefile='base_strategy.py.j2',
                                    arguments={"strategy": strategy_name,
                                               "attributes": attributes,
                                               "indicators": indicators,
                                               "buy_trend": buy_trend,
                                               "sell_trend": sell_trend,
                                               "plot_config": plot_config,
                                               "additional_methods": additional_methods,
                                               })

    logger.info(f"Writing strategy to `{strategy_path}`.")
    strategy_path.write_text(strategy_text)


def start_new_strategy(args: Dict[str, Any]) -> None:

    config = setup_utils_configuration(args, RunMode.UTIL_NO_EXCHANGE)

    if "strategy" in args and args["strategy"]:

        new_path = config['user_data_dir'] / USERPATH_STRATEGIES / (args['strategy'] + '.py')

        if new_path.exists():
            raise OperationalException(f"`{new_path}` already exists. "
                                       "Please choose another Strategy Name.")

        deploy_new_strategy(args['strategy'], new_path, args['template'])

    else:
        raise ConfigurationError("`new-strategy` requires --strategy to be set.")


def clean_ui_subdir(directory: Path):
    if directory.is_dir():
        logger.info("Removing UI directory content.")

        for p in reversed(list(directory.glob('**/*'))):  # iterate contents from leaves to root
            if p.name in ('.gitkeep', 'fallback_file.html'):
                continue
            if p.is_file():
                p.unlink()
            elif p.is_dir():
                p.rmdir()


def read_ui_version(dest_folder: Path) -> Optional[str]:
    file = dest_folder / '.uiversion'
    if not file.is_file():
        return None

    with file.open('r') as f:
        return f.read()


def download_and_install_ui(dest_folder: Path, dl_url: str, version: str):
    from io import BytesIO
    from zipfile import ZipFile

    logger.info(f"Downloading {dl_url}")
    resp = requests.get(dl_url, timeout=req_timeout).content
    dest_folder.mkdir(parents=True, exist_ok=True)
    with ZipFile(BytesIO(resp)) as zf:
        for fn in zf.filelist:
            with zf.open(fn) as x:
                destfile = dest_folder / fn.filename
                if fn.is_dir():
                    destfile.mkdir(exist_ok=True)
                else:
                    destfile.write_bytes(x.read())
    with (dest_folder / '.uiversion').open('w') as f:
        f.write(version)


def get_ui_download_url(version: Optional[str] = None) -> Tuple[str, str]:
    base_url = 'https://api.github.com/repos/freqtrade/frequi/'
    # Get base UI Repo path

    resp = requests.get(f"{base_url}releases", timeout=req_timeout)
    resp.raise_for_status()
    r = resp.json()

    if version:
        tmp = [x for x in r if x['name'] == version]
        if tmp:
            latest_version = tmp[0]['name']
            assets = tmp[0].get('assets', [])
        else:
            raise ValueError("UI-Version not found.")
    else:
        latest_version = r[0]['name']
        assets = r[0].get('assets', [])
    dl_url = ''
    if assets and len(assets) > 0:
        dl_url = assets[0]['browser_download_url']

    # URL not found - try assets url
    if not dl_url:
        assets = r[0]['assets_url']
        resp = requests.get(assets, timeout=req_timeout)
        r = resp.json()
        dl_url = r[0]['browser_download_url']

    return dl_url, latest_version


def start_install_ui(args: Dict[str, Any]) -> None:

    dest_folder = Path(__file__).parents[1] / 'rpc/api_server/ui/installed/'
    # First make sure the assets are removed.
    dl_url, latest_version = get_ui_download_url(args.get('ui_version'))

    curr_version = read_ui_version(dest_folder)
    if curr_version == latest_version and not args.get('erase_ui_only'):
        logger.info(f"UI already up-to-date, FreqUI Version {curr_version}.")
        return

    clean_ui_subdir(dest_folder)
    if args.get('erase_ui_only'):
        logger.info("Erased UI directory content. Not downloading new version.")
    else:
        # Download a new version
        download_and_install_ui(dest_folder, dl_url, latest_version)

1	import logging	1✔
2	import sys	1✔
3	from pathlib import Path	1✔
4	from typing import Any, Dict, Optional, Tuple	1✔
5
6	import requests	1✔
7
8	from freqtrade.configuration import setup_utils_configuration	1✔
9	from freqtrade.configuration.directory_operations import copy_sample_files, create_userdata_dir	1✔
10	from freqtrade.constants import USERPATH_STRATEGIES	1✔
11	from freqtrade.enums import RunMode	1✔
12	from freqtrade.exceptions import ConfigurationError, OperationalException	1✔
13	from freqtrade.util import render_template, render_template_with_fallback	1✔
14
15
16	logger = logging.getLogger(__name__)	1✔
17
18
19	# Timeout for requests
20	req_timeout = 30	1✔
21
22
23	def start_create_userdir(args: Dict[str, Any]) -> None:	1✔
24	"""
25	Create "user_data" directory to contain user data strategies, hyperopt, ...)
26	:param args: Cli args from Arguments()
27	:return: None
28	"""
29	if "user_data_dir" in args and args["user_data_dir"]:	1✔
30	userdir = create_userdata_dir(args["user_data_dir"], create_dir=True)	1✔
31	copy_sample_files(userdir, overwrite=args["reset"])	1✔
32	else:
33	logger.warning("`create-userdir` requires --userdir to be set.")	1✔
34	sys.exit(1)	1✔
35
36
37	def deploy_new_strategy(strategy_name: str, strategy_path: Path, subtemplate: str) -> None:	1✔
38	"""
39	Deploy new strategy from template to strategy_path
40	"""
41	fallback = 'full'	1✔
42	attributes = render_template_with_fallback(	1✔
43	templatefile=f"strategy_subtemplates/strategy_attributes_{subtemplate}.j2",
44	templatefallbackfile=f"strategy_subtemplates/strategy_attributes_{fallback}.j2",
45	)
46	indicators = render_template_with_fallback(	1✔
47	templatefile=f"strategy_subtemplates/indicators_{subtemplate}.j2",
48	templatefallbackfile=f"strategy_subtemplates/indicators_{fallback}.j2",
49	)
50	buy_trend = render_template_with_fallback(	1✔
51	templatefile=f"strategy_subtemplates/buy_trend_{subtemplate}.j2",
52	templatefallbackfile=f"strategy_subtemplates/buy_trend_{fallback}.j2",
53	)
54	sell_trend = render_template_with_fallback(	1✔
55	templatefile=f"strategy_subtemplates/sell_trend_{subtemplate}.j2",
56	templatefallbackfile=f"strategy_subtemplates/sell_trend_{fallback}.j2",
57	)
58	plot_config = render_template_with_fallback(	1✔
59	templatefile=f"strategy_subtemplates/plot_config_{subtemplate}.j2",
60	templatefallbackfile=f"strategy_subtemplates/plot_config_{fallback}.j2",
61	)
62	additional_methods = render_template_with_fallback(	1✔
63	templatefile=f"strategy_subtemplates/strategy_methods_{subtemplate}.j2",
64	templatefallbackfile="strategy_subtemplates/strategy_methods_empty.j2",
65	)
66
67	strategy_text = render_template(templatefile='base_strategy.py.j2',	1✔
68	arguments={"strategy": strategy_name,
69	"attributes": attributes,
70	"indicators": indicators,
71	"buy_trend": buy_trend,
72	"sell_trend": sell_trend,
73	"plot_config": plot_config,
74	"additional_methods": additional_methods,
75	})
76
77	logger.info(f"Writing strategy to `{strategy_path}`.")	1✔
78	strategy_path.write_text(strategy_text)	1✔
79
80
81	def start_new_strategy(args: Dict[str, Any]) -> None:	1✔
82
83	config = setup_utils_configuration(args, RunMode.UTIL_NO_EXCHANGE)	1✔
84
85	if "strategy" in args and args["strategy"]:	1✔
86
87	new_path = config['user_data_dir'] / USERPATH_STRATEGIES / (args['strategy'] + '.py')	1✔
88
89	if new_path.exists():	1✔
90	raise OperationalException(f"`{new_path}` already exists. "	1✔
91	"Please choose another Strategy Name.")
92
93	deploy_new_strategy(args['strategy'], new_path, args['template'])	1✔
94
95	else:
96	raise ConfigurationError("`new-strategy` requires --strategy to be set.")	1✔
97
98
99	def clean_ui_subdir(directory: Path):	1✔
100	if directory.is_dir():	1✔
101	logger.info("Removing UI directory content.")	1✔
102
103	for p in reversed(list(directory.glob('*/'))): # iterate contents from leaves to root	1✔
104	if p.name in ('.gitkeep', 'fallback_file.html'):	1✔
105	continue	1✔
106	if p.is_file():	1✔
107	p.unlink()	1✔
108	elif p.is_dir():	1✔
109	p.rmdir()	1✔
110
111
112	def read_ui_version(dest_folder: Path) -> Optional[str]:	1✔
113	file = dest_folder / '.uiversion'	1✔
114	if not file.is_file():	1✔
115	return None	1✔
116
117	with file.open('r') as f:	1✔
118	return f.read()	1✔
119
120
121	def download_and_install_ui(dest_folder: Path, dl_url: str, version: str):	1✔
122	from io import BytesIO	1✔
123	from zipfile import ZipFile	1✔
124
125	logger.info(f"Downloading {dl_url}")	1✔
126	resp = requests.get(dl_url, timeout=req_timeout).content	1✔
127	dest_folder.mkdir(parents=True, exist_ok=True)	1✔
128	with ZipFile(BytesIO(resp)) as zf:	1✔
129	for fn in zf.filelist:	1✔
130	with zf.open(fn) as x:	1✔
131	destfile = dest_folder / fn.filename	1✔
132	if fn.is_dir():	1✔
133	destfile.mkdir(exist_ok=True)	1✔
134	else:
135	destfile.write_bytes(x.read())	1✔
136	with (dest_folder / '.uiversion').open('w') as f:	1✔
137	f.write(version)	1✔
138
139
140	def get_ui_download_url(version: Optional[str] = None) -> Tuple[str, str]:	1✔
141	base_url = 'https://api.github.com/repos/freqtrade/frequi/'	1✔
142	# Get base UI Repo path
143
144	resp = requests.get(f"{base_url}releases", timeout=req_timeout)	1✔
145	resp.raise_for_status()	1✔
146	r = resp.json()	1✔
147
148	if version:	1✔
149	tmp = [x for x in r if x['name'] == version]	1✔
150	if tmp:	1✔
151	latest_version = tmp[0]['name']	1✔
152	assets = tmp[0].get('assets', [])	1✔
153	else:
154	raise ValueError("UI-Version not found.")	1✔
155	else:
156	latest_version = r[0]['name']	1✔
157	assets = r[0].get('assets', [])	1✔
158	dl_url = ''	1✔
159	if assets and len(assets) > 0:	1✔
160	dl_url = assets[0]['browser_download_url']	1✔
161
162	# URL not found - try assets url
163	if not dl_url:	1✔
164	assets = r[0]['assets_url']	1✔
165	resp = requests.get(assets, timeout=req_timeout)	1✔
166	r = resp.json()	1✔
167	dl_url = r[0]['browser_download_url']	1✔
168
169	return dl_url, latest_version	1✔
170
171
172	def start_install_ui(args: Dict[str, Any]) -> None:	1✔
173
174	dest_folder = Path(__file__).parents[1] / 'rpc/api_server/ui/installed/'	1✔
175	# First make sure the assets are removed.
176	dl_url, latest_version = get_ui_download_url(args.get('ui_version'))	1✔
177
178	curr_version = read_ui_version(dest_folder)	1✔
179	if curr_version == latest_version and not args.get('erase_ui_only'):	1✔
180	logger.info(f"UI already up-to-date, FreqUI Version {curr_version}.")	×
181	return	×
182
183	clean_ui_subdir(dest_folder)	1✔
184	if args.get('erase_ui_only'):	1✔
185	logger.info("Erased UI directory content. Not downloading new version.")	1✔
186	else:
187	# Download a new version
188	download_and_install_ui(dest_folder, dl_url, latest_version)	1✔

freqtrade / freqtrade / 9394559170

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