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

96.79

/freqtrade/rpc/api_server/api_backtest.py

import asyncio
import logging
from copy import deepcopy
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List

from fastapi import APIRouter, BackgroundTasks, Depends
from fastapi.exceptions import HTTPException

from freqtrade.configuration.config_validation import validate_config_consistency
from freqtrade.constants import Config
from freqtrade.data.btanalysis import (delete_backtest_result, get_backtest_market_change,
                                       get_backtest_result, get_backtest_resultlist,
                                       load_and_merge_backtest_result, update_backtest_metadata)
from freqtrade.enums import BacktestState
from freqtrade.exceptions import ConfigurationError, DependencyException, OperationalException
from freqtrade.exchange.common import remove_exchange_credentials
from freqtrade.misc import deep_merge_dicts, is_file_in_dir
from freqtrade.rpc.api_server.api_schemas import (BacktestHistoryEntry, BacktestMarketChange,
                                                  BacktestMetadataUpdate, BacktestRequest,
                                                  BacktestResponse)
from freqtrade.rpc.api_server.deps import get_config
from freqtrade.rpc.api_server.webserver_bgwork import ApiBG
from freqtrade.rpc.rpc import RPCException
from freqtrade.types import get_BacktestResultType_default


logger = logging.getLogger(__name__)

# Private API, protected by authentication and webserver_mode dependency
router = APIRouter()


def __run_backtest_bg(btconfig: Config):
    from freqtrade.data.metrics import combined_dataframes_with_rel_mean
    from freqtrade.optimize.optimize_reports import generate_backtest_stats, store_backtest_stats
    from freqtrade.resolvers import StrategyResolver

    asyncio.set_event_loop(asyncio.new_event_loop())
    try:
        # Reload strategy
        lastconfig = ApiBG.bt['last_config']
        strat = StrategyResolver.load_strategy(btconfig)
        validate_config_consistency(btconfig)

        if (
            not ApiBG.bt['bt']
            or lastconfig.get('timeframe') != strat.timeframe
            or lastconfig.get('timeframe_detail') != btconfig.get('timeframe_detail')
            or lastconfig.get('timerange') != btconfig['timerange']
        ):
            from freqtrade.optimize.backtesting import Backtesting
            ApiBG.bt['bt'] = Backtesting(btconfig)
            ApiBG.bt['bt'].load_bt_data_detail()
        else:
            ApiBG.bt['bt'].config = btconfig
            ApiBG.bt['bt'].init_backtest()
        # Only reload data if timeframe changed.
        if (
            not ApiBG.bt['data']
            or not ApiBG.bt['timerange']
            or lastconfig.get('timeframe') != strat.timeframe
            or lastconfig.get('timerange') != btconfig['timerange']
        ):
            ApiBG.bt['data'], ApiBG.bt['timerange'] = ApiBG.bt[
                'bt'].load_bt_data()

        lastconfig['timerange'] = btconfig['timerange']
        lastconfig['timeframe'] = strat.timeframe
        lastconfig['protections'] = btconfig.get('protections', [])
        lastconfig['enable_protections'] = btconfig.get('enable_protections')
        lastconfig['dry_run_wallet'] = btconfig.get('dry_run_wallet')

        ApiBG.bt['bt'].enable_protections = btconfig.get('enable_protections', False)
        ApiBG.bt['bt'].strategylist = [strat]
        ApiBG.bt['bt'].results = get_BacktestResultType_default()
        ApiBG.bt['bt'].load_prior_backtest()

        ApiBG.bt['bt'].abort = False
        strategy_name = strat.get_strategy_name()
        if (ApiBG.bt['bt'].results and
                strategy_name in ApiBG.bt['bt'].results['strategy']):
            # When previous result hash matches - reuse that result and skip backtesting.
            logger.info(f'Reusing result of previous backtest for {strategy_name}')
        else:
            min_date, max_date = ApiBG.bt['bt'].backtest_one_strategy(
                strat, ApiBG.bt['data'], ApiBG.bt['timerange'])

            ApiBG.bt['bt'].results = generate_backtest_stats(
                ApiBG.bt['data'], ApiBG.bt['bt'].all_results,
                min_date=min_date, max_date=max_date)

        if btconfig.get('export', 'none') == 'trades':
            combined_res = combined_dataframes_with_rel_mean(ApiBG.bt['data'], min_date, max_date)
            fn = store_backtest_stats(
                btconfig['exportfilename'],
                ApiBG.bt['bt'].results,
                datetime.now().strftime("%Y-%m-%d_%H-%M-%S"),
                market_change_data=combined_res
                )
            ApiBG.bt['bt'].results['metadata'][strategy_name]['filename'] = str(fn.stem)
            ApiBG.bt['bt'].results['metadata'][strategy_name]['strategy'] = strategy_name

        logger.info("Backtest finished.")

    except ConfigurationError as e:
        logger.error(f"Backtesting encountered a configuration Error: {e}")

    except (Exception, OperationalException, DependencyException) as e:
        logger.exception(f"Backtesting caused an error: {e}")
        ApiBG.bt['bt_error'] = str(e)
    finally:
        ApiBG.bgtask_running = False


@router.post('/backtest', response_model=BacktestResponse, tags=['webserver', 'backtest'])
async def api_start_backtest(
        bt_settings: BacktestRequest, background_tasks: BackgroundTasks,
        config=Depends(get_config)):
    ApiBG.bt['bt_error'] = None
    """Start backtesting if not done so already"""
    if ApiBG.bgtask_running:
        raise RPCException('Bot Background task already running')

    if ':' in bt_settings.strategy:
        raise HTTPException(status_code=500, detail="base64 encoded strategies are not allowed.")

    btconfig = deepcopy(config)
    remove_exchange_credentials(btconfig['exchange'], True)
    settings = dict(bt_settings)
    if settings.get('freqai', None) is not None:
        settings['freqai'] = dict(settings['freqai'])
    # Pydantic models will contain all keys, but non-provided ones are None

    btconfig = deep_merge_dicts(settings, btconfig, allow_null_overrides=False)
    try:
        btconfig['stake_amount'] = float(btconfig['stake_amount'])
    except ValueError:
        pass

    # Force dry-run for backtesting
    btconfig['dry_run'] = True

    # Start backtesting
    # Initialize backtesting object

    background_tasks.add_task(__run_backtest_bg, btconfig=btconfig)
    ApiBG.bgtask_running = True

    return {
        "status": "running",
        "running": True,
        "progress": 0,
        "step": str(BacktestState.STARTUP),
        "status_msg": "Backtest started",
    }


@router.get('/backtest', response_model=BacktestResponse, tags=['webserver', 'backtest'])
def api_get_backtest():
    """
    Get backtesting result.
    Returns Result after backtesting has been ran.
    """
    from freqtrade.persistence import LocalTrade
    if ApiBG.bgtask_running:
        return {
            "status": "running",
            "running": True,
            "step": (ApiBG.bt['bt'].progress.action if ApiBG.bt['bt']
                     else str(BacktestState.STARTUP)),
            "progress": ApiBG.bt['bt'].progress.progress if ApiBG.bt['bt'] else 0,
            "trade_count": len(LocalTrade.trades),
            "status_msg": "Backtest running",
        }

    if not ApiBG.bt['bt']:
        return {
            "status": "not_started",
            "running": False,
            "step": "",
            "progress": 0,
            "status_msg": "Backtest not yet executed"
        }
    if ApiBG.bt['bt_error']:
        return {
            "status": "error",
            "running": False,
            "step": "",
            "progress": 0,
            "status_msg": f"Backtest failed with {ApiBG.bt['bt_error']}"
        }

    return {
        "status": "ended",
        "running": False,
        "status_msg": "Backtest ended",
        "step": "finished",
        "progress": 1,
        "backtest_result": ApiBG.bt['bt'].results,
    }


@router.delete('/backtest', response_model=BacktestResponse, tags=['webserver', 'backtest'])
def api_delete_backtest():
    """Reset backtesting"""
    if ApiBG.bgtask_running:
        return {
            "status": "running",
            "running": True,
            "step": "",
            "progress": 0,
            "status_msg": "Backtest running",
        }
    if ApiBG.bt['bt']:
        ApiBG.bt['bt'].cleanup()
        del ApiBG.bt['bt']
        ApiBG.bt['bt'] = None
        del ApiBG.bt['data']
        ApiBG.bt['data'] = None
        logger.info("Backtesting reset")
    return {
        "status": "reset",
        "running": False,
        "step": "",
        "progress": 0,
        "status_msg": "Backtest reset",
    }


@router.get('/backtest/abort', response_model=BacktestResponse, tags=['webserver', 'backtest'])
def api_backtest_abort():
    if not ApiBG.bgtask_running:
        return {
            "status": "not_running",
            "running": False,
            "step": "",
            "progress": 0,
            "status_msg": "Backtest ended",
        }
    ApiBG.bt['bt'].abort = True
    return {
        "status": "stopping",
        "running": False,
        "step": "",
        "progress": 0,
        "status_msg": "Backtest ended",
    }


@router.get('/backtest/history', response_model=List[BacktestHistoryEntry],
            tags=['webserver', 'backtest'])
def api_backtest_history(config=Depends(get_config)):
    # Get backtest result history, read from metadata files
    return get_backtest_resultlist(config['user_data_dir'] / 'backtest_results')


@router.get('/backtest/history/result', response_model=BacktestResponse,
            tags=['webserver', 'backtest'])
def api_backtest_history_result(filename: str, strategy: str, config=Depends(get_config)):
    # Get backtest result history, read from metadata files
    bt_results_base: Path = config['user_data_dir'] / 'backtest_results'
    fn = (bt_results_base / filename).with_suffix('.json')

    results: Dict[str, Any] = {
        'metadata': {},
        'strategy': {},
        'strategy_comparison': [],
    }
    if not is_file_in_dir(fn, bt_results_base):
        raise HTTPException(status_code=404, detail="File not found.")
    load_and_merge_backtest_result(strategy, fn, results)
    return {
        "status": "ended",
        "running": False,
        "step": "",
        "progress": 1,
        "status_msg": "Historic result",
        "backtest_result": results,
    }


@router.delete('/backtest/history/{file}', response_model=List[BacktestHistoryEntry],
               tags=['webserver', 'backtest'])
def api_delete_backtest_history_entry(file: str, config=Depends(get_config)):
    # Get backtest result history, read from metadata files
    bt_results_base: Path = config['user_data_dir'] / 'backtest_results'
    file_abs = (bt_results_base / file).with_suffix('.json')
    # Ensure file is in backtest_results directory
    if not is_file_in_dir(file_abs, bt_results_base):
        raise HTTPException(status_code=404, detail="File not found.")

    delete_backtest_result(file_abs)
    return get_backtest_resultlist(config['user_data_dir'] / 'backtest_results')


@router.patch('/backtest/history/{file}', response_model=List[BacktestHistoryEntry],
              tags=['webserver', 'backtest'])
def api_update_backtest_history_entry(file: str, body: BacktestMetadataUpdate,
                                      config=Depends(get_config)):
    # Get backtest result history, read from metadata files
    bt_results_base: Path = config['user_data_dir'] / 'backtest_results'
    file_abs = (bt_results_base / file).with_suffix('.json')
    # Ensure file is in backtest_results directory
    if not is_file_in_dir(file_abs, bt_results_base):
        raise HTTPException(status_code=404, detail="File not found.")
    content = {
        'notes': body.notes
    }
    try:
        update_backtest_metadata(file_abs, body.strategy, content)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

    return get_backtest_result(file_abs)


@router.get('/backtest/history/{file}/market_change', response_model=BacktestMarketChange,
            tags=['webserver', 'backtest'])
def api_get_backtest_market_change(file: str, config=Depends(get_config)):
    bt_results_base: Path = config['user_data_dir'] / 'backtest_results'
    file_abs = (bt_results_base / f"{file}_market_change").with_suffix('.feather')
    # Ensure file is in backtest_results directory
    if not is_file_in_dir(file_abs, bt_results_base):
        raise HTTPException(status_code=404, detail="File not found.")
    df = get_backtest_market_change(file_abs)

    return {
        'columns': df.columns.tolist(),
        'data': df.values.tolist(),
        'length': len(df),
    }

1	import asyncio	1✔
2	import logging	1✔
3	from copy import deepcopy	1✔
4	from datetime import datetime	1✔
5	from pathlib import Path	1✔
6	from typing import Any, Dict, List	1✔
7
8	from fastapi import APIRouter, BackgroundTasks, Depends	1✔
9	from fastapi.exceptions import HTTPException	1✔
10
11	from freqtrade.configuration.config_validation import validate_config_consistency	1✔
12	from freqtrade.constants import Config	1✔
13	from freqtrade.data.btanalysis import (delete_backtest_result, get_backtest_market_change,	1✔
14	get_backtest_result, get_backtest_resultlist,
15	load_and_merge_backtest_result, update_backtest_metadata)
16	from freqtrade.enums import BacktestState	1✔
17	from freqtrade.exceptions import ConfigurationError, DependencyException, OperationalException	1✔
18	from freqtrade.exchange.common import remove_exchange_credentials	1✔
19	from freqtrade.misc import deep_merge_dicts, is_file_in_dir	1✔
20	from freqtrade.rpc.api_server.api_schemas import (BacktestHistoryEntry, BacktestMarketChange,	1✔
21	BacktestMetadataUpdate, BacktestRequest,
22	BacktestResponse)
23	from freqtrade.rpc.api_server.deps import get_config	1✔
24	from freqtrade.rpc.api_server.webserver_bgwork import ApiBG	1✔
25	from freqtrade.rpc.rpc import RPCException	1✔
26	from freqtrade.types import get_BacktestResultType_default	1✔
27
28
29	logger = logging.getLogger(__name__)	1✔
30
31	# Private API, protected by authentication and webserver_mode dependency
32	router = APIRouter()	1✔
33
34
35	def __run_backtest_bg(btconfig: Config):	1✔
36	from freqtrade.data.metrics import combined_dataframes_with_rel_mean	1✔
37	from freqtrade.optimize.optimize_reports import generate_backtest_stats, store_backtest_stats	1✔
38	from freqtrade.resolvers import StrategyResolver	1✔
39
40	asyncio.set_event_loop(asyncio.new_event_loop())	1✔
41	try:	1✔
42	# Reload strategy
43	lastconfig = ApiBG.bt['last_config']	1✔
44	strat = StrategyResolver.load_strategy(btconfig)	1✔
45	validate_config_consistency(btconfig)	1✔
46
47	if (	1✔
48	not ApiBG.bt['bt']
49	or lastconfig.get('timeframe') != strat.timeframe
50	or lastconfig.get('timeframe_detail') != btconfig.get('timeframe_detail')
51	or lastconfig.get('timerange') != btconfig['timerange']
52	):
53	from freqtrade.optimize.backtesting import Backtesting	1✔
54	ApiBG.bt['bt'] = Backtesting(btconfig)	1✔
55	ApiBG.bt['bt'].load_bt_data_detail()	1✔
56	else:
57	ApiBG.bt['bt'].config = btconfig	1✔
58	ApiBG.bt['bt'].init_backtest()	1✔
59	# Only reload data if timeframe changed.
60	if (	1✔
61	not ApiBG.bt['data']
62	or not ApiBG.bt['timerange']
63	or lastconfig.get('timeframe') != strat.timeframe
64	or lastconfig.get('timerange') != btconfig['timerange']
65	):
66	ApiBG.bt['data'], ApiBG.bt['timerange'] = ApiBG.bt[	1✔
67	'bt'].load_bt_data()
68
69	lastconfig['timerange'] = btconfig['timerange']	1✔
70	lastconfig['timeframe'] = strat.timeframe	1✔
71	lastconfig['protections'] = btconfig.get('protections', [])	1✔
72	lastconfig['enable_protections'] = btconfig.get('enable_protections')	1✔
73	lastconfig['dry_run_wallet'] = btconfig.get('dry_run_wallet')	1✔
74
75	ApiBG.bt['bt'].enable_protections = btconfig.get('enable_protections', False)	1✔
76	ApiBG.bt['bt'].strategylist = [strat]	1✔
77	ApiBG.bt['bt'].results = get_BacktestResultType_default()	1✔
78	ApiBG.bt['bt'].load_prior_backtest()	1✔
79
80	ApiBG.bt['bt'].abort = False	1✔
81	strategy_name = strat.get_strategy_name()	1✔
82	if (ApiBG.bt['bt'].results and	1✔
83	strategy_name in ApiBG.bt['bt'].results['strategy']):
84	# When previous result hash matches - reuse that result and skip backtesting.
85	logger.info(f'Reusing result of previous backtest for {strategy_name}')	1✔
86	else:
87	min_date, max_date = ApiBG.bt['bt'].backtest_one_strategy(	1✔
88	strat, ApiBG.bt['data'], ApiBG.bt['timerange'])
89
90	ApiBG.bt['bt'].results = generate_backtest_stats(	1✔
91	ApiBG.bt['data'], ApiBG.bt['bt'].all_results,
92	min_date=min_date, max_date=max_date)
93
94	if btconfig.get('export', 'none') == 'trades':	1✔
95	combined_res = combined_dataframes_with_rel_mean(ApiBG.bt['data'], min_date, max_date)	1✔
96	fn = store_backtest_stats(	1✔
97	btconfig['exportfilename'],
98	ApiBG.bt['bt'].results,
99	datetime.now().strftime("%Y-%m-%d_%H-%M-%S"),
100	market_change_data=combined_res
101	)
102	ApiBG.bt['bt'].results['metadata'][strategy_name]['filename'] = str(fn.stem)	1✔
103	ApiBG.bt['bt'].results['metadata'][strategy_name]['strategy'] = strategy_name	1✔
104
105	logger.info("Backtest finished.")	1✔
106
107	except ConfigurationError as e:	1✔
108	logger.error(f"Backtesting encountered a configuration Error: {e}")	×
109
110	except (Exception, OperationalException, DependencyException) as e:	1✔
111	logger.exception(f"Backtesting caused an error: {e}")	1✔
112	ApiBG.bt['bt_error'] = str(e)	1✔
113	finally:
114	ApiBG.bgtask_running = False	1✔
115
116
117	@router.post('/backtest', response_model=BacktestResponse, tags=['webserver', 'backtest'])	1✔
118	async def api_start_backtest(	1✔
119	bt_settings: BacktestRequest, background_tasks: BackgroundTasks,
120	config=Depends(get_config)):
121	ApiBG.bt['bt_error'] = None	1✔
122	"""Start backtesting if not done so already"""	1✔
123	if ApiBG.bgtask_running:	1✔
124	raise RPCException('Bot Background task already running')	1✔
125
126	if ':' in bt_settings.strategy:	1✔
127	raise HTTPException(status_code=500, detail="base64 encoded strategies are not allowed.")	1✔
128
129	btconfig = deepcopy(config)	1✔
130	remove_exchange_credentials(btconfig['exchange'], True)	1✔
131	settings = dict(bt_settings)	1✔
132	if settings.get('freqai', None) is not None:	1✔
133	settings['freqai'] = dict(settings['freqai'])	×
134	# Pydantic models will contain all keys, but non-provided ones are None
135
136	btconfig = deep_merge_dicts(settings, btconfig, allow_null_overrides=False)	1✔
137	try:	1✔
138	btconfig['stake_amount'] = float(btconfig['stake_amount'])	1✔
139	except ValueError:	×
140	pass	×
141
142	# Force dry-run for backtesting
143	btconfig['dry_run'] = True	1✔
144
145	# Start backtesting
146	# Initialize backtesting object
147
148	background_tasks.add_task(__run_backtest_bg, btconfig=btconfig)	1✔
149	ApiBG.bgtask_running = True	1✔
150
151	return {	1✔
152	"status": "running",
153	"running": True,
154	"progress": 0,
155	"step": str(BacktestState.STARTUP),
156	"status_msg": "Backtest started",
157	}
158
159
160	@router.get('/backtest', response_model=BacktestResponse, tags=['webserver', 'backtest'])	1✔
161	def api_get_backtest():	1✔
162	"""
163	Get backtesting result.
164	Returns Result after backtesting has been ran.
165	"""
166	from freqtrade.persistence import LocalTrade	1✔
167	if ApiBG.bgtask_running:	1✔
168	return {	1✔
169	"status": "running",
170	"running": True,
171	"step": (ApiBG.bt['bt'].progress.action if ApiBG.bt['bt']
172	else str(BacktestState.STARTUP)),
173	"progress": ApiBG.bt['bt'].progress.progress if ApiBG.bt['bt'] else 0,
174	"trade_count": len(LocalTrade.trades),
175	"status_msg": "Backtest running",
176	}
177
178	if not ApiBG.bt['bt']:	1✔
179	return {	1✔
180	"status": "not_started",
181	"running": False,
182	"step": "",
183	"progress": 0,
184	"status_msg": "Backtest not yet executed"
185	}
186	if ApiBG.bt['bt_error']:	1✔
187	return {	1✔
188	"status": "error",
189	"running": False,
190	"step": "",
191	"progress": 0,
192	"status_msg": f"Backtest failed with {ApiBG.bt['bt_error']}"
193	}
194
195	return {	1✔
196	"status": "ended",
197	"running": False,
198	"status_msg": "Backtest ended",
199	"step": "finished",
200	"progress": 1,
201	"backtest_result": ApiBG.bt['bt'].results,
202	}
203
204
205	@router.delete('/backtest', response_model=BacktestResponse, tags=['webserver', 'backtest'])	1✔
206	def api_delete_backtest():	1✔
207	"""Reset backtesting"""
208	if ApiBG.bgtask_running:	1✔
209	return {	1✔
210	"status": "running",
211	"running": True,
212	"step": "",
213	"progress": 0,
214	"status_msg": "Backtest running",
215	}
216	if ApiBG.bt['bt']:	1✔
217	ApiBG.bt['bt'].cleanup()	1✔
218	del ApiBG.bt['bt']	1✔
219	ApiBG.bt['bt'] = None	1✔
220	del ApiBG.bt['data']	1✔
221	ApiBG.bt['data'] = None	1✔
222	logger.info("Backtesting reset")	1✔
223	return {	1✔
224	"status": "reset",
225	"running": False,
226	"step": "",
227	"progress": 0,
228	"status_msg": "Backtest reset",
229	}
230
231
232	@router.get('/backtest/abort', response_model=BacktestResponse, tags=['webserver', 'backtest'])	1✔
233	def api_backtest_abort():	1✔
234	if not ApiBG.bgtask_running:	1✔
235	return {	1✔
236	"status": "not_running",
237	"running": False,
238	"step": "",
239	"progress": 0,
240	"status_msg": "Backtest ended",
241	}
242	ApiBG.bt['bt'].abort = True	1✔
243	return {	1✔
244	"status": "stopping",
245	"running": False,
246	"step": "",
247	"progress": 0,
248	"status_msg": "Backtest ended",
249	}
250
251
252	@router.get('/backtest/history', response_model=List[BacktestHistoryEntry],	1✔
253	tags=['webserver', 'backtest'])
254	def api_backtest_history(config=Depends(get_config)):	1✔
255	# Get backtest result history, read from metadata files
256	return get_backtest_resultlist(config['user_data_dir'] / 'backtest_results')	1✔
257
258
259	@router.get('/backtest/history/result', response_model=BacktestResponse,	1✔
260	tags=['webserver', 'backtest'])
261	def api_backtest_history_result(filename: str, strategy: str, config=Depends(get_config)):	1✔
262	# Get backtest result history, read from metadata files
263	bt_results_base: Path = config['user_data_dir'] / 'backtest_results'	1✔
264	fn = (bt_results_base / filename).with_suffix('.json')	1✔
265
266	results: Dict[str, Any] = {	1✔
267	'metadata': {},
268	'strategy': {},
269	'strategy_comparison': [],
270	}
271	if not is_file_in_dir(fn, bt_results_base):	1✔
272	raise HTTPException(status_code=404, detail="File not found.")	×
273	load_and_merge_backtest_result(strategy, fn, results)	1✔
274	return {	1✔
275	"status": "ended",
276	"running": False,
277	"step": "",
278	"progress": 1,
279	"status_msg": "Historic result",
280	"backtest_result": results,
281	}
282
283
284	@router.delete('/backtest/history/{file}', response_model=List[BacktestHistoryEntry],	1✔
285	tags=['webserver', 'backtest'])
286	def api_delete_backtest_history_entry(file: str, config=Depends(get_config)):	1✔
287	# Get backtest result history, read from metadata files
288	bt_results_base: Path = config['user_data_dir'] / 'backtest_results'	1✔
289	file_abs = (bt_results_base / file).with_suffix('.json')	1✔
290	# Ensure file is in backtest_results directory
291	if not is_file_in_dir(file_abs, bt_results_base):	1✔
292	raise HTTPException(status_code=404, detail="File not found.")	1✔
293
294	delete_backtest_result(file_abs)	1✔
295	return get_backtest_resultlist(config['user_data_dir'] / 'backtest_results')	1✔
296
297
298	@router.patch('/backtest/history/{file}', response_model=List[BacktestHistoryEntry],	1✔
299	tags=['webserver', 'backtest'])
300	def api_update_backtest_history_entry(file: str, body: BacktestMetadataUpdate,	1✔
301	config=Depends(get_config)):
302	# Get backtest result history, read from metadata files
303	bt_results_base: Path = config['user_data_dir'] / 'backtest_results'	1✔
304	file_abs = (bt_results_base / file).with_suffix('.json')	1✔
305	# Ensure file is in backtest_results directory
306	if not is_file_in_dir(file_abs, bt_results_base):	1✔
307	raise HTTPException(status_code=404, detail="File not found.")	1✔
308	content = {	1✔
309	'notes': body.notes
310	}
311	try:	1✔
312	update_backtest_metadata(file_abs, body.strategy, content)	1✔
313	except ValueError as e:	1✔
314	raise HTTPException(status_code=400, detail=str(e))	1✔
315
316	return get_backtest_result(file_abs)	1✔
317
318
319	@router.get('/backtest/history/{file}/market_change', response_model=BacktestMarketChange,	1✔
320	tags=['webserver', 'backtest'])
321	def api_get_backtest_market_change(file: str, config=Depends(get_config)):	1✔
322	bt_results_base: Path = config['user_data_dir'] / 'backtest_results'	1✔
323	file_abs = (bt_results_base / f"{file}_market_change").with_suffix('.feather')	1✔
324	# Ensure file is in backtest_results directory
325	if not is_file_in_dir(file_abs, bt_results_base):	1✔
326	raise HTTPException(status_code=404, detail="File not found.")	1✔
327	df = get_backtest_market_change(file_abs)	1✔
328
329	return {	1✔
330	'columns': df.columns.tolist(),
331	'data': df.values.tolist(),
332	'length': len(df),
333	}

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