15036016265

Committed 14 May 2025 06:14PM UTC coverage: 94.366% (-0.05%) from 94.417%

Build # 15036016265

Build Type

push

github

Committed by

xmatthias

Commit Message

fix: default max_open_trades to inf instead of -1

Without this, the auto-conversion doesn't backpopulate to the config

closes #11752

Run Details

22361 of 23696 relevant lines covered (94.37%)

0.94 hits per line

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

96.91

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

freqtrade / freqtrade / 15036016265

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