Cv

systematica.walk_forward.cv

WalkForwardCV

WalkForwardCV(
    id: str = None,
    loader_or_data: Callable = <function load_clean_data>,
    timeframe: str = '1d',
    start: str = None,
    end: str = None,
    s1: str = 'BTCUSDT',
    s2: str = 'ETHUSDT',
    suffix: str = 'h5',
    use_close: bool = True,
    freq: str = None,
    signal_model: str = None,
    long_entries: float = nan,
    long_exits: float = nan,
    short_entries: float = nan,
    short_exits: float = nan,
    clean: bool = True,
    select_symbols: str | int | List[str | int] = None,
    portfolio_config: Dict[str, Any] = _Nothing.NOTHING,
    model_params: Dict[str, Any] = _Nothing.NOTHING,
    metrics: str | List[str] = None,
    metrics_kwargs: Dict[str, Any] = _Nothing.NOTHING,
    use_rolling: bool = False,
    storage: str | optuna.storages._base.BaseStorage = None,
    pruner: optuna.pruners._base.BasePruner = <optuna.pruners._successive_halving.SuccessiveHalvingPruner object>,
    sampler: optuna.samplers._base.BaseSampler = <optuna.samplers._tpe.sampler.TPESampler object>,
    study_name: str = None,
    direction: str | List[str] = 'maximize',
    load_if_exists: bool = False,
    n_trials: int = 100,
    n_completed_trials: int = None,
    timeout: float = None,
    n_jobs: int = 1,
    catch: Iterable[Type[Exception]] | Type[Exception] = _Nothing.NOTHING,
    callbacks: Iterable[Callable[[optuna.study.study.Study, optuna.trial._frozen.FrozenTrial], None]] | None = None,
    gc_after_trial: bool = False,
    show_progress_bar: bool = False,
    verbose: bool = True,
    tracker: Type = None,
    model: Any = None,
    preprocess_func: Callable = None,
)

Analyzes a statistical arbitrage model using cross-validation.

Method generated by attrs for class WalkForwardCV.

Ancestors

• systematica.walk_forward.base.BaseWalkForward
• abc.ABC
• systematica.generic.base.Model
• systematica.generic.base.Feature

Descendants

• systematica.api.analytics.arbitrage_index.ArbitrageIndexCV
• systematica.api.analytics.meta_model.MetaModelCV
• systematica.api.analytics.momentum.AverageMomentumFactorCV
• systematica.api.analytics.momentum.MomentumFactorCV
• systematica.api.analytics.ou_process.OUProcessCV
• systematica.api.analytics.range_breakout.RangeBreakoutCV
• systematica.api.analytics.volatility.VolatilityFactorCV
• systematica.api.analytics.volume_profile.VolumeProfileFactorCV

Static methods

check_is_valid

check_is_valid(
    model: Any,
)

Validates CV model.

Raises:

Type	Description
AttributeError	If the model does not use splitter as a parameter.

reshape_nested_dataframe

reshape_nested_dataframe(
    df: pandas.core.frame.DataFrame,
    index: pandas.core.indexes.base.Index,
) ‑> pandas.core.frame.DataFrame

Reshape a DataFrame by transposing, unstacking, reindexing, and restacking.

Parameters:

Name	Type	Default	Description
df	pd.DataFrame	--	The DataFrame to reshape. Must have a MultiIndex with a ‘symbol’ level.

Returns:

Type	Description
pd.DataFrame	The reshaped DataFrame.

Instance variables

• index_train: pandas.core.indexes.base.Index: Computes and returns the train set index of the dataset. The train set index is derived from the end bounds values returned by the splitter.

• index_test: pandas.core.indexes.base.Index: Computes and returns the test set index of the dataset. The test set index is derived from the end bounds values returned by the splitter.

• all_portfolio_metrics: List[str]: Get available cross-validation metrics.

• close_reset_stacked_by_set:

• high_reset_stacked_by_set:

• low_reset_stacked_by_set:

• model_output_reset_stacked_by_set:

• open_reset_stacked_by_set:

• preprocess_data_reset_stacked_by_set:

• price_reset_stacked_by_set:

• returns_reset_stacked_by_set:

• split_preprocess_data:

• splitter:

Methods

get_splits

get_splits(
    self,
    df: pandas.core.series.Series | pandas.core.frame.DataFrame,
) ‑> pandas.core.frame.DataFrame

Splits based on the cross-validation splitter.

Returns:

Type	Description
pd.DataFrame	DataFrame containing the split data.

set_index

set_index(
    self,
    key: str,
    set_datetime_index: bool,
) ‑> None | pandas.core.indexes.datetimes.DatetimeIndex

Set valid datetime index

Parameters:

Name	Type	Default	Description
key	str	--	If train, extract train-set. If test, extract test-set.
set_datetime_index	bool	--	If True, set end-test datetime index. Ignored otherwise.

Returns:

Type	Description
None | tp.DatetimeIndex	Datetime index.

signals_reset_stacked_by_set

signals_reset_stacked_by_set(
    self,
    validate_model: bool = True,
) ‑> pandas.core.series.Series

Resets the model output stacked by set and attaches bounds to the index.

This method uses the splitter to reset signals, stacking it by set and attaching bounds to the index. it then select the test-set.

Parameters:

Name	Type	Default	Description
validate_model	bool	True	Check if model parameters are valid. Defaults to True.

Returns:

Type	Description
pd.Series	The signals test-set.

compare_train_test_split

compare_train_test_split(
    self,
    train_test_registry: List[systematica.registries.base.Register] = None,
) ‑> pandas.core.frame.DataFrame

Compare train and test set across all splits.

Calculate all train/test comparison registered features.

Parameters:

Name	Type	Default	Description
train_test_stats_config	tp.List[StateConfig]	“None.	State representation config. if None, uses TRAIN_TEST_STATS_CONFIG. Defaults to None.

Returns:

Type	Description
pd.DataFrame	Computed features.

get_annualized_return

get_annualized_return(
    self,
    key: str = 'test',
    raw_output: bool = True,
    set_datetime_index: bool = True,
) ‑> pandas.core.series.Series | pandas.core.frame.DataFrame

Computes the annualized returns for the test set.

Parameters:

Name	Type	Default	Description
key	str	test	If train, extract train-set. If test, extract test-set. Defaults to test.
raw_output	bool	True	If True, returns raw results. Default is True, which returns processed results.
set_datetime_index	bool	True	If True, set end-test datetime index. Ignored otherwise. Defaults to True.

Returns:

Type	Description
pd.Series	Series containing the annualized returns for the test set.

get_portfolio_metrics

get_portfolio_metrics(
    self,
    metrics: str | List[str] = None,
    metrics_kwargs: Dict[str, Any] = None,
    to_numpy: bool = False,
    validate_model: bool = True,
    portfolio_config: Dict[str, Any] = None,
) ‑> pandas.core.frame.DataFrame | numpy.ndarray

Computes portfolio performance metrics for different test sets.

Parameters:

Name	Type	Default	Description
metrics	str	--	| tp.List[str], default None Portfolio performance metrics to compute. If None, uses the default metrics.
metrics_kwargs	tp.Kwargs	None	Additional keyword arguments for the metrics.
to_numpy	bool	False	If True, returns a np.ndarray. Default is False, returning a pd.DataFrame.
validate_model	bool	True	Check if model parameters are valid. Defaults to True.
portfolio_config	tp.Kwargs	None	Additional configuration for the portfolio simulation. Defaults to None.

Returns:

Type	Description
pd.DataFrame | tp.Array	Portfolio performance metrics.