15035825052

Committed 09 Apr 2025 08:08AM UTC coverage: 94.373% (+0.007%) from 94.366%

Build # 15035825052

Build Type

push

github

Committed by

web-flow

Commit Message

Update dependabot.yml

Run Details

22172 of 23494 relevant lines covered (94.37%)

0.94 hits per line

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

96.93

/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

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.ft_types import get_BacktestResultType_default
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


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_results
    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["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_results(
                    btconfig,
                    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.bt_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"
    for ext in [".zip", ".json"]:
        fn = (bt_results_base / filename).with_suffix(ext)
        if is_file_in_dir(fn, bt_results_base):
            break
    else:
        raise HTTPException(status_code=404, detail="File not found.")

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

freqtrade / freqtrade / 15035825052

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