Cv - Systematica - Blockforce Capital

`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.

Overview

​WalkForwardCV

​Ancestors

​Descendants

​Static methods

​check_is_valid

​reshape_nested_dataframe

​Instance variables

​Methods

​get_splits

​set_index

​signals_reset_stacked_by_set

​compare_train_test_split

​get_annualized_return

​get_portfolio_metrics

`WalkForwardCV`

Ancestors

Descendants

Static methods

`check_is_valid`

`reshape_nested_dataframe`

Instance variables

Methods

`get_splits`

`set_index`

`signals_reset_stacked_by_set`

`compare_train_test_split`

`get_annualized_return`

`get_portfolio_metrics`