15457673245

Committed 27 Apr 2025 09:35AM UTC coverage: 94.383% (+0.01%) from 94.373%

Build # 15457673245

Build Type

push

github

Committed by

xmatthias

Commit Message

test: update tests for new file structure

Run Details

22264 of 23589 relevant lines covered (94.38%)

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_bt_content,
                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,
                    strategy_files={
                        s.get_strategy_name(): s.__file__ for s in ApiBG.bt["bt"].strategylist
                    },
                )
                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"],
100	ApiBG.bt["bt"].all_bt_content,
101	min_date=min_date,
102	max_date=max_date,
103	)
104
105	if btconfig.get("export", "none") == "trades":	1✔
106	combined_res = combined_dataframes_with_rel_mean(	1✔
107	ApiBG.bt["data"], min_date, max_date
108	)
109	fn = store_backtest_results(	1✔
110	btconfig,
111	ApiBG.bt["bt"].results,
112	datetime.now().strftime("%Y-%m-%d_%H-%M-%S"),
113	market_change_data=combined_res,
114	strategy_files={
115	s.get_strategy_name(): s.__file__ for s in ApiBG.bt["bt"].strategylist
116	},
117	)
118	ApiBG.bt["bt"].results["metadata"][strategy_name]["filename"] = str(fn.stem)	1✔
119	ApiBG.bt["bt"].results["metadata"][strategy_name]["strategy"] = strategy_name	1✔
120
121	logger.info("Backtest finished.")	1✔
122
123	except ConfigurationError as e:	1✔
124	logger.error(f"Backtesting encountered a configuration Error: {e}")	×
125
126	except (Exception, OperationalException, DependencyException) as e:	1✔
127	logger.exception(f"Backtesting caused an error: {e}")	1✔
128	ApiBG.bt["bt_error"] = str(e)	1✔
129	finally:
130	ApiBG.bgtask_running = False	1✔
131
132
133	@router.post("/backtest", response_model=BacktestResponse, tags=["webserver", "backtest"])	1✔
134	async def api_start_backtest(	1✔
135	bt_settings: BacktestRequest, background_tasks: BackgroundTasks, config=Depends(get_config)
136	):
137	ApiBG.bt["bt_error"] = None	1✔
138	"""Start backtesting if not done so already"""	1✔
139	if ApiBG.bgtask_running:	1✔
140	raise RPCException("Bot Background task already running")	1✔
141
142	if ":" in bt_settings.strategy:	1✔
143	raise HTTPException(status_code=500, detail="base64 encoded strategies are not allowed.")	1✔
144
145	btconfig = deepcopy(config)	1✔
146	remove_exchange_credentials(btconfig["exchange"], True)	1✔
147	settings = dict(bt_settings)	1✔
148	if settings.get("freqai", None) is not None:	1✔
149	settings["freqai"] = dict(settings["freqai"])	×
150	# Pydantic models will contain all keys, but non-provided ones are None
151
152	btconfig = deep_merge_dicts(settings, btconfig, allow_null_overrides=False)	1✔
153	try:	1✔
154	btconfig["stake_amount"] = float(btconfig["stake_amount"])	1✔
155	except ValueError:	×
156	pass	×
157
158	# Force dry-run for backtesting
159	btconfig["dry_run"] = True	1✔
160
161	# Start backtesting
162	# Initialize backtesting object
163
164	background_tasks.add_task(__run_backtest_bg, btconfig=btconfig)	1✔
165	ApiBG.bgtask_running = True	1✔
166
167	return {	1✔
168	"status": "running",
169	"running": True,
170	"progress": 0,
171	"step": str(BacktestState.STARTUP),
172	"status_msg": "Backtest started",
173	}
174
175
176	@router.get("/backtest", response_model=BacktestResponse, tags=["webserver", "backtest"])	1✔
177	def api_get_backtest():	1✔
178	"""
179	Get backtesting result.
180	Returns Result after backtesting has been ran.
181	"""
182	from freqtrade.persistence import LocalTrade	1✔
183
184	if ApiBG.bgtask_running:	1✔
185	return {	1✔
186	"status": "running",
187	"running": True,
188	"step": (
189	ApiBG.bt["bt"].progress.action if ApiBG.bt["bt"] else str(BacktestState.STARTUP)
190	),
191	"progress": ApiBG.bt["bt"].progress.progress if ApiBG.bt["bt"] else 0,
192	"trade_count": len(LocalTrade.bt_trades),
193	"status_msg": "Backtest running",
194	}
195
196	if not ApiBG.bt["bt"]:	1✔
197	return {	1✔
198	"status": "not_started",
199	"running": False,
200	"step": "",
201	"progress": 0,
202	"status_msg": "Backtest not yet executed",
203	}
204	if ApiBG.bt["bt_error"]:	1✔
205	return {	1✔
206	"status": "error",
207	"running": False,
208	"step": "",
209	"progress": 0,
210	"status_msg": f"Backtest failed with {ApiBG.bt['bt_error']}",
211	}
212
213	return {	1✔
214	"status": "ended",
215	"running": False,
216	"status_msg": "Backtest ended",
217	"step": "finished",
218	"progress": 1,
219	"backtest_result": ApiBG.bt["bt"].results,
220	}
221
222
223	@router.delete("/backtest", response_model=BacktestResponse, tags=["webserver", "backtest"])	1✔
224	def api_delete_backtest():	1✔
225	"""Reset backtesting"""
226	if ApiBG.bgtask_running:	1✔
227	return {	1✔
228	"status": "running",
229	"running": True,
230	"step": "",
231	"progress": 0,
232	"status_msg": "Backtest running",
233	}
234	if ApiBG.bt["bt"]:	1✔
235	ApiBG.bt["bt"].cleanup()	1✔
236	del ApiBG.bt["bt"]	1✔
237	ApiBG.bt["bt"] = None	1✔
238	del ApiBG.bt["data"]	1✔
239	ApiBG.bt["data"] = None	1✔
240	logger.info("Backtesting reset")	1✔
241	return {	1✔
242	"status": "reset",
243	"running": False,
244	"step": "",
245	"progress": 0,
246	"status_msg": "Backtest reset",
247	}
248
249
250	@router.get("/backtest/abort", response_model=BacktestResponse, tags=["webserver", "backtest"])	1✔
251	def api_backtest_abort():	1✔
252	if not ApiBG.bgtask_running:	1✔
253	return {	1✔
254	"status": "not_running",
255	"running": False,
256	"step": "",
257	"progress": 0,
258	"status_msg": "Backtest ended",
259	}
260	ApiBG.bt["bt"].abort = True	1✔
261	return {	1✔
262	"status": "stopping",
263	"running": False,
264	"step": "",
265	"progress": 0,
266	"status_msg": "Backtest ended",
267	}
268
269
270	@router.get(	1✔
271	"/backtest/history", response_model=list[BacktestHistoryEntry], tags=["webserver", "backtest"]
272	)
273	def api_backtest_history(config=Depends(get_config)):	1✔
274	# Get backtest result history, read from metadata files
275	return get_backtest_resultlist(config["user_data_dir"] / "backtest_results")	1✔
276
277
278	@router.get(	1✔
279	"/backtest/history/result", response_model=BacktestResponse, tags=["webserver", "backtest"]
280	)
281	def api_backtest_history_result(filename: str, strategy: str, config=Depends(get_config)):	1✔
282	# Get backtest result history, read from metadata files
283	bt_results_base: Path = config["user_data_dir"] / "backtest_results"	1✔
284	for ext in [".zip", ".json"]:	1✔
285	fn = (bt_results_base / filename).with_suffix(ext)	1✔
286	if is_file_in_dir(fn, bt_results_base):	1✔
287	break	1✔
288	else:
289	raise HTTPException(status_code=404, detail="File not found.")	×
290
291	results: dict[str, Any] = {	1✔
292	"metadata": {},
293	"strategy": {},
294	"strategy_comparison": [],
295	}
296	load_and_merge_backtest_result(strategy, fn, results)	1✔
297	return {	1✔
298	"status": "ended",
299	"running": False,
300	"step": "",
301	"progress": 1,
302	"status_msg": "Historic result",
303	"backtest_result": results,
304	}
305
306
307	@router.delete(	1✔
308	"/backtest/history/{file}",
309	response_model=list[BacktestHistoryEntry],
310	tags=["webserver", "backtest"],
311	)
312	def api_delete_backtest_history_entry(file: str, config=Depends(get_config)):	1✔
313	# Get backtest result history, read from metadata files
314	bt_results_base: Path = config["user_data_dir"] / "backtest_results"	1✔
315	for ext in [".zip", ".json"]:	1✔
316	file_abs = (bt_results_base / file).with_suffix(ext)	1✔
317	# Ensure file is in backtest_results directory
318	if is_file_in_dir(file_abs, bt_results_base):	1✔
319	break	1✔
320	else:
321	raise HTTPException(status_code=404, detail="File not found.")	1✔
322
323	delete_backtest_result(file_abs)	1✔
324	return get_backtest_resultlist(config["user_data_dir"] / "backtest_results")	1✔
325
326
327	@router.patch(	1✔
328	"/backtest/history/{file}",
329	response_model=list[BacktestHistoryEntry],
330	tags=["webserver", "backtest"],
331	)
332	def api_update_backtest_history_entry(	1✔
333	file: str, body: BacktestMetadataUpdate, config=Depends(get_config)
334	):
335	# Get backtest result history, read from metadata files
336	bt_results_base: Path = config["user_data_dir"] / "backtest_results"	1✔
337	for ext in [".zip", ".json"]:	1✔
338	file_abs = (bt_results_base / file).with_suffix(ext)	1✔
339	# Ensure file is in backtest_results directory
340	if is_file_in_dir(file_abs, bt_results_base):	1✔
341	break	1✔
342	else:
343	raise HTTPException(status_code=404, detail="File not found.")	1✔
344
345	content = {"notes": body.notes}	1✔
346	try:	1✔
347	update_backtest_metadata(file_abs, body.strategy, content)	1✔
348	except ValueError as e:	1✔
349	raise HTTPException(status_code=400, detail=str(e))	1✔
350
351	return get_backtest_result(file_abs)	1✔
352
353
354	@router.get(	1✔
355	"/backtest/history/{file}/market_change",
356	response_model=BacktestMarketChange,
357	tags=["webserver", "backtest"],
358	)
359	def api_get_backtest_market_change(file: str, config=Depends(get_config)):	1✔
360	bt_results_base: Path = config["user_data_dir"] / "backtest_results"	1✔
361	for fn in (	1✔
362	Path(file).with_suffix(".zip"),
363	Path(f"{file}_market_change").with_suffix(".feather"),
364	):
365	file_abs = bt_results_base / fn	1✔
366	# Ensure file is in backtest_results directory
367	if is_file_in_dir(file_abs, bt_results_base):	1✔
368	break	1✔
369	else:
370	raise HTTPException(status_code=404, detail="File not found.")	1✔
371
372	df = get_backtest_market_change(file_abs)	1✔
373
374	return {	1✔
375	"columns": df.columns.tolist(),
376	"data": df.values.tolist(),
377	"length": len(df),
378	}

freqtrade / freqtrade / 15457673245

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