@ -155,33 +155,55 @@ def load_backtest_metadata(filename: Path | str) -> dict[str, Any]:
raise OperationalException ( " Unexpected error while loading backtest metadata. " ) from e
def load_backtest_stats ( filename : Path | str ) - > BacktestResultType :
def _normalize_filename ( file_or_directory : Path | str , filename : Path | str | None ) - > Path :
"""
Normalize the filename by ensuring it is a Path object .
: param file_or_directory : The directory or file to normalize .
: param filename : The filename to normalize .
: return : A Path object representing the normalized filename .
"""
if isinstance ( file_or_directory , str ) :
file_or_directory = Path ( file_or_directory )
if file_or_directory . is_dir ( ) :
if not filename :
filename = get_latest_backtest_filename ( file_or_directory )
if Path ( filename ) . is_file ( ) :
fn = Path ( filename )
else :
fn = file_or_directory / filename
else :
fn = file_or_directory
return fn
def load_backtest_stats (
file_or_directory : Path | str , filename : Path | str | None = None
) - > BacktestResultType :
"""
Load backtest statistics file .
: param filename : pathlib . Path object , or string pointing to the file .
: param file_or_directory : pathlib . Path object , or string pointing to the directory ,
or absolute / relative path to the backtest results file .
: param filename : Optional filename to load from ( if different from the main filename ) .
Only valid when loading from a directory .
: return : a dictionary containing the resulting file .
"""
if isinstance ( filename , str ) :
filename = Path ( filename )
if filename . is_dir ( ) :
filename = filename / get_latest_backtest_filename ( filename )
if not filename . is_file ( ) :
raise ValueError ( f " File { filename } does not exist. " )
logger . info ( f " Loading backtest result from { filename } " )
fn = _normalize_filename ( file_or_directory , filename )
if filename . suffix == " .zip " :
if not fn . is_file ( ) :
raise ValueError ( f " File or directory { fn } does not exist. " )
logger . info ( f " Loading backtest result from { fn } " )
if fn . suffix == " .zip " :
data = json_load (
StringIO (
load_file_from_zip ( filename , filename . with_suffix ( " .json " ) . name ) . decode ( " utf-8 " )
)
StringIO ( load_file_from_zip ( fn , fn . with_suffix ( " .json " ) . name ) . decode ( " utf-8 " ) )
)
else :
with filename . open ( ) as file :
with f n. open ( ) as file :
data = json_load ( file )
# Legacy list format does not contain metadata.
if isinstance ( data , dict ) :
data [ " metadata " ] = load_backtest_metadata ( f ile name )
data [ " metadata " ] = load_backtest_metadata ( f n)
return data
@ -362,16 +384,21 @@ def _load_backtest_data_df_compatibility(df: pd.DataFrame) -> pd.DataFrame:
return df
def load_backtest_data ( filename : Path | str , strategy : str | None = None ) - > pd . DataFrame :
def load_backtest_data (
file_or_directory : Path | str , strategy : str | None = None , filename : Path | str | None = None
) - > pd . DataFrame :
"""
Load backtest data file .
: param filename : pathlib . Path object , or string pointing to a file or directory
Load backtest data file , returns a dataframe with the individual trades .
: param file_or_directory : pathlib . Path object , or string pointing to the directory ,
or absolute / relative path to the backtest results file .
: param strategy : Strategy to load - mainly relevant for multi - strategy backtests
Can also serve as protection to load the correct result .
: param filename : Optional filename to load from ( if different from the main filename ) .
Only valid when loading from a directory .
: return : a dataframe with the analysis results
: raise : ValueError if loading goes wrong .
"""
data = load_backtest_stats ( file name)
data = load_backtest_stats ( file _or_directory, file name)
if not isinstance ( data , list ) :
# new, nested format
if " strategy " not in data :
@ -430,20 +457,23 @@ def load_file_from_zip(zip_path: Path, filename: str) -> bytes:
raise ValueError ( f " Bad zip file: { zip_path } . " ) from None
def load_backtest_analysis_data ( backtest_dir : Path , name : Literal [ " signals " , " rejected " , " exited " ] ) :
def load_backtest_analysis_data (
file_or_directory : Path ,
name : Literal [ " signals " , " rejected " , " exited " ] ,
filename : Path | str | None = None ,
) :
"""
Load backtest analysis data either from a pickle file or from within a zip file
: param backtest_dir : Directory containing backtest results
: param file_or_directory : pathlib . Path object , or string pointing to the directory ,
or absolute / relative path to the backtest results file .
: param name : Name of the analysis data to load ( signals , rejected , exited )
: param filename : Optional filename to load from ( if different from the main filename ) .
Only valid when loading from a directory .
: return : Analysis data
"""
import joblib
if backtest_dir . is_dir ( ) :
lbf = Path ( get_latest_backtest_filename ( backtest_dir ) )
zip_path = backtest_dir / lbf
else :
zip_path = backtest_dir
zip_path = _normalize_filename ( file_or_directory , filename )
if zip_path . suffix == " .zip " :
# Load from zip file
@ -458,10 +488,10 @@ def load_backtest_analysis_data(backtest_dir: Path, name: Literal["signals", "re
else :
# Load from separate pickle file
if backtest_dir . is_dir ( ) :
scpf = Path ( backtest_dir , f " { zip_path . stem } _ { name } .pkl " )
if file_or_directory . is_dir ( ) :
scpf = Path ( file_or_directory , f " { zip_path . stem } _ { name } .pkl " )
else :
scpf = Path ( backtest_dir. parent / f " { backtest_dir . stem } _ { name } .pkl " )
scpf = Path ( file_or_directory. parent / f " { file_or_directory . stem } _ { name } .pkl " )
try :
with scpf . open ( " rb " ) as scp :
Matthias
6 months ago
committed by
GitHub