15035809744

Committed 25 Apr 2025 06:28AM UTC coverage: 94.385% (+0.03%) from 94.358%

Build # 15035809744

Build Type

push

github

Committed by

xmatthias

Commit Message

test: update test for calculate_market_change

Run Details

22257 of 23581 relevant lines covered (94.39%)

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

freqtrade / freqtrade / 15035809744

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