14507242113

Committed 02 Dec 2024 07:11PM UTC coverage: 94.422% (+0.05%) from 94.377%

Build # 14507242113

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #11028 from xzmeng/fix-none

fix: check if days is None before conversion

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

525 existing lines in 54 files now uncovered.

21684 of 22965 relevant lines covered (94.42%)

0.94 hits per line

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

96.77

/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_stats
    from freqtrade.resolvers import StrategyResolver

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

        if (
            not ApiBG.bt["bt"]
            or lastconfig.get("timeframe") != strat.timeframe
            or lastconfig.get("timeframe_detail") != btconfig.get("timeframe_detail")
            or lastconfig.get("timerange") != btconfig["timerange"]
        ):
            from freqtrade.optimize.backtesting import Backtesting

            ApiBG.bt["bt"] = Backtesting(btconfig)
            ApiBG.bt["bt"].load_bt_data_detail()
        else:
            ApiBG.bt["bt"].config = btconfig
            ApiBG.bt["bt"].init_backtest()
        # Only reload data if timeframe changed.
        if (
            not ApiBG.bt["data"]
            or not ApiBG.bt["timerange"]
            or lastconfig.get("timeframe") != strat.timeframe
            or lastconfig.get("timerange") != btconfig["timerange"]
        ):
            ApiBG.bt["data"], ApiBG.bt["timerange"] = ApiBG.bt["bt"].load_bt_data()

        lastconfig["timerange"] = btconfig["timerange"]
        lastconfig["timeframe"] = strat.timeframe
        lastconfig["enable_protections"] = btconfig.get("enable_protections")
        lastconfig["dry_run_wallet"] = btconfig.get("dry_run_wallet")

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

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

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

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

        logger.info("Backtest finished.")

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

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


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

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

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

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

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

    # Start backtesting
    # Initialize backtesting object

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

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


@router.get("/backtest", response_model=BacktestResponse, tags=["webserver", "backtest"])
def api_get_backtest():
    """
    Get backtesting result.
    Returns Result after backtesting has been ran.
    """
    from freqtrade.persistence import LocalTrade

    if ApiBG.bgtask_running:
        return {
            "status": "running",
            "running": True,
            "step": (
                ApiBG.bt["bt"].progress.action if ApiBG.bt["bt"] else str(BacktestState.STARTUP)
            ),
            "progress": ApiBG.bt["bt"].progress.progress if ApiBG.bt["bt"] else 0,
            "trade_count": len(LocalTrade.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"
    fn = (bt_results_base / filename).with_suffix(".json")

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


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

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


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

    return get_backtest_result(file_abs)


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

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

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

freqtrade / freqtrade / 14507242113

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