Regime Surrogate

Regime-conditional surrogate that interpolates factor recommendations across regime descriptors (e.g. dataset size, noise level) instead of relying on hard regime buckets. Builds on fit_surrogate.

Install via the optional extra (same as the base surrogate):

uv pip install 'trade-study[surrogate]'

trade_study.fit_regime_surrogate(results, regime_factors, factors, *, method='gp', seed=0, n_estimators=200, cv_folds=5, warn_below_r2=0.0, cv_group_by=None)

Fit a surrogate that conditions on regime features.

Internally fits a single :class:SurrogateModel over the joint regime_factors + factors space, so observables can be interpolated across continuous regime axes.

Every config in results.configs must contain values for both the regime features and the design factors.

Parameters:

Name	Type	Description	Default
results	ResultsTable	A :class:ResultsTable from previous study runs that spans multiple regimes.	required
regime_factors	list[Factor]	Factors describing the regime (additional input dimensions of the surrogate; typically continuous).	required
factors	list[Factor]	Tunable design factors. Together with regime_factors these must cover every key referenced in results.configs.	required
method	str	Surrogate backend, "gp" or "rf". See :func:trade_study.fit_surrogate.	'gp'
seed	int	Random seed forwarded to the backend estimators.	0
n_estimators	int	Number of trees for the "rf" backend.	200
cv_folds	int	Cross-validation folds for the held-out accuracy check. See :func:trade_study.fit_surrogate.	5
warn_below_r2	float | None	Warn if any observable's cross-validated R^2 falls below this threshold. See :func:trade_study.fit_surrogate.	0.0
cv_group_by	str | Sequence[str] | None	"regime" holds out complete regime descriptor tuples; "design" groups identical joint configurations. A sequence selects grouping factor names; None retains shuffled row CV.	None

Returns:

Type	Description
RegimeSurrogate	A fitted :class:RegimeSurrogate.

Raises:

Type	Description
ValueError	If regime_factors is empty, if a name appears in both regime_factors and factors, or if the underlying :func:fit_surrogate call fails.

Source code in src/trade_study/regime.py

def fit_regime_surrogate(  # ruff: ignore[too-many-arguments]
    results: ResultsTable,
    regime_factors: list[Factor],
    factors: list[Factor],
    *,
    method: str = "gp",
    seed: int = 0,
    n_estimators: int = 200,
    cv_folds: int = 5,
    warn_below_r2: float | None = 0.0,
    cv_group_by: str | Sequence[str] | None = None,
) -> RegimeSurrogate:
    """Fit a surrogate that conditions on regime features.

    Internally fits a single :class:`SurrogateModel` over the joint
    ``regime_factors + factors`` space, so observables can be
    interpolated across continuous regime axes.

    Every config in ``results.configs`` must contain values for both the
    regime features and the design factors.

    Args:
        results: A :class:`ResultsTable` from previous study runs that
            spans multiple regimes.
        regime_factors: Factors describing the regime (additional input
            dimensions of the surrogate; typically continuous).
        factors: Tunable design factors. Together with
            ``regime_factors`` these must cover every key referenced in
            ``results.configs``.
        method: Surrogate backend, ``"gp"`` or ``"rf"``. See
            :func:`trade_study.fit_surrogate`.
        seed: Random seed forwarded to the backend estimators.
        n_estimators: Number of trees for the ``"rf"`` backend.
        cv_folds: Cross-validation folds for the held-out accuracy check. See
            :func:`trade_study.fit_surrogate`.
        warn_below_r2: Warn if any observable's cross-validated R^2 falls
            below this threshold. See :func:`trade_study.fit_surrogate`.
        cv_group_by: ``"regime"`` holds out complete regime descriptor tuples;
            ``"design"`` groups identical joint configurations. A sequence
            selects grouping factor names; None retains shuffled row CV.

    Returns:
        A fitted :class:`RegimeSurrogate`.

    Raises:
        ValueError: If ``regime_factors`` is empty, if a name appears in
            both ``regime_factors`` and ``factors``, or if the
            underlying :func:`fit_surrogate` call fails.
    """
    if not regime_factors:
        msg = "fit_regime_surrogate: regime_factors must be non-empty"
        raise ValueError(msg)
    overlap = {f.name for f in regime_factors} & {f.name for f in factors}
    if overlap:
        msg = (
            f"fit_regime_surrogate: names appear in both regime_factors and "
            f"factors: {sorted(overlap)}"
        )
        raise ValueError(msg)
    inner = fit_surrogate(
        results,
        [*regime_factors, *factors],
        method=method,
        seed=seed,
        n_estimators=n_estimators,
        cv_folds=cv_folds,
        warn_below_r2=warn_below_r2,
        cv_group_by=(
            [f.name for f in regime_factors] if cv_group_by == "regime" else cv_group_by
        ),
    )
    return RegimeSurrogate(
        inner=inner,
        regime_factors=list(regime_factors),
        factors=list(factors),
    )

fit_regime_surrogate(..., cv_group_by="regime") holds out entire tuples of regime descriptors. The underlying model exposes cv_group_by and group ids; cv_r2/cv_rmse describe held-out regimes, while row_cv_r2/row_cv_rmse describe shuffled rows. Use "design" for joint design/regime configurations or an explicit sequence of factor names for another grouping.

Inspect model.support(regime, configs) before trusting a recommendation. Prediction and recommendation warn when candidates or the query regime fall outside observed marginal support. warn_support=False disables that warning; warn_below_r2=None independently disables accuracy warnings.

trade_study.RegimeSurrogate(inner, regime_factors, factors) dataclass

Surrogate that conditions on regime features.

Wraps a single :class:SurrogateModel fit over the union of regime descriptors and design factors. Use :func:fit_regime_surrogate to construct one.

Attributes:

Name	Type	Description
inner	SurrogateModel	The underlying :class:SurrogateModel over the joint regime_factors + factors input space.
regime_factors	list[Factor]	Factors describing the regime (additional input dimensions of the surrogate).
factors	list[Factor]	Tunable design factors that are optimized at a given regime by :meth:recommend.

cv_r2 property

Per-observable held-out cross-validated R^2.

Returns:

Name	Type	Description
	dict[str, float]	Mapping from observable name to cross-validated R^2, from the
underlying	dict[str, float]	attr:inner surrogate.

cv_rmse property

Per-observable held-out cross-validated RMSE.

Returns:

Type	Description
dict[str, float]	Mapping from observable name to cross-validated RMSE, from
dict[str, float]	the underlying :attr:inner surrogate.

row_cv_r2 property

Shuffled row-validation R^2, for comparison with grouped validation.

Returns:

Type	Description
dict[str, float]	Per-observable row-validation R^2 from the underlying surrogate.

row_cv_rmse property

Shuffled row-validation RMSE in observable units.

Returns:

Type	Description
dict[str, float]	Per-observable row-validation RMSE from the underlying surrogate.

support(regime, configs)

Inspect observed support at a query regime without warnings.

Parameters:

Name	Type	Description	Default
regime	dict[str, Any]	Regime descriptor values.	required
configs	Sequence[dict[str, Any]]	Design-factor configurations to inspect.	required

Returns:

Type	Description
list[PredictionSupport]	Marginal support diagnostics for each config and observable.

Source code in src/trade_study/regime.py

def support(
    self, regime: dict[str, Any], configs: Sequence[dict[str, Any]]
) -> list[PredictionSupport]:
    """Inspect observed support at a query regime without warnings.

    Args:
        regime: Regime descriptor values.
        configs: Design-factor configurations to inspect.

    Returns:
        Marginal support diagnostics for each config and observable.
    """
    return self.inner.support([_merge(regime, c) for c in configs])

predict(regime, config, *, warn_support=True)

Predict observables at a regime + config pair.

Parameters:

Name	Type	Description	Default
regime	dict[str, Any]	Mapping of regime-feature names to values.	required
config	dict[str, Any]	Mapping of design-factor names to values.	required
warn_support	bool	Warn on values outside observed training support.	True

Returns:

Type	Description
dict[str, float]	Mapping from observable name to predicted scalar.

Source code in src/trade_study/regime.py

def predict(
    self,
    regime: dict[str, Any],
    config: dict[str, Any],
    *,
    warn_support: bool = True,
) -> dict[str, float]:
    """Predict observables at a regime + config pair.

    Args:
        regime: Mapping of regime-feature names to values.
        config: Mapping of design-factor names to values.
        warn_support: Warn on values outside observed training support.

    Returns:
        Mapping from observable name to predicted scalar.
    """
    return self.inner.predict(_merge(regime, config), warn_support=warn_support)

predict_batch(regime, configs, *, warn_support=True)

Predict observables for a batch of configs at one regime.

Parameters:

Name	Type	Description	Default
regime	dict[str, Any]	Mapping of regime-feature names to values.	required
configs	Sequence[dict[str, Any]]	Sequence of design-factor configs to score at regime.	required
warn_support	bool	Warn on values outside observed training support.	True

Returns:

Type	Description
dict[str, NDArray[float64]]	Mapping from observable name to a length-len(configs)
dict[str, NDArray[float64]]	array of predictions.

Source code in src/trade_study/regime.py

def predict_batch(
    self,
    regime: dict[str, Any],
    configs: Sequence[dict[str, Any]],
    *,
    warn_support: bool = True,
) -> dict[str, NDArray[np.float64]]:
    """Predict observables for a batch of configs at one regime.

    Args:
        regime: Mapping of regime-feature names to values.
        configs: Sequence of design-factor configs to score at
            ``regime``.
        warn_support: Warn on values outside observed training support.

    Returns:
        Mapping from observable name to a length-``len(configs)``
        array of predictions.
    """
    merged = [_merge(regime, c) for c in configs]
    return self.inner.predict_batch(merged, warn_support=warn_support)

uncertainty(regime, config, *, warn_support=True)

Predictive standard deviation per observable (GP only).

Parameters:

Name	Type	Description	Default
regime	dict[str, Any]	Mapping of regime-feature names to values.	required
config	dict[str, Any]	Mapping of design-factor names to values.	required
warn_support	bool	Warn on values outside observed training support.	True

Returns:

Name	Type	Description
	dict[str, float]	Mapping from observable name to predictive standard deviation.
Propagates	dict[str, float]	class:NotImplementedError from the underlying
	dict[str, float]	surrogate when the backend does not expose calibrated
	dict[str, float]	uncertainties (non-GP backends).

Source code in src/trade_study/regime.py

def uncertainty(
    self,
    regime: dict[str, Any],
    config: dict[str, Any],
    *,
    warn_support: bool = True,
) -> dict[str, float]:
    """Predictive standard deviation per observable (GP only).

    Args:
        regime: Mapping of regime-feature names to values.
        config: Mapping of design-factor names to values.
        warn_support: Warn on values outside observed training support.

    Returns:
        Mapping from observable name to predictive standard deviation.
        Propagates :class:`NotImplementedError` from the underlying
        surrogate when the backend does not expose calibrated
        uncertainties (non-GP backends).
    """
    return self.inner.uncertainty(_merge(regime, config), warn_support=warn_support)

recommend(regime, *, objective, mode='min', n_candidates=512, seed=0, candidates=None, warn_below_r2=0.0, risk=0.0, warn_support=True)

Recommend a design-factor config at a query regime.

Samples n_candidates configs from the design-factor space via a scrambled Sobol' sequence and returns the one whose surrogate prediction for objective is best under mode. With risk > 0 candidates are ranked pessimistically, by prediction + risk * spread when minimizing and prediction - risk * spread when maximizing, where spread is :meth:SurrogateModel.spread_batch; this avoids picking the candidate the surrogate is most optimistically wrong about.

Parameters:

Name	Type	Description	Default
regime	dict[str, Any]	Mapping of regime-feature names to values.	required
objective	str	Name of the observable to optimize. Must be one of self.inner.observable_names.	required
mode	str	"min" or "max".	'min'
n_candidates	int	Number of design-space samples to evaluate. Ignored when candidates is provided.	512
seed	int	Seed for the Sobol' sampler.	0
candidates	Sequence[dict[str, Any]] | None	Optional explicit list of design-factor configs to score; if given, overrides n_candidates.	None
warn_below_r2	float | None	Warn if objective's cross-validated R^2 is below this threshold, so a caller optimizing against a poorly-fit surrogate gets a signal right at the point of use, not just buried in fit-time logs. Pass None to disable.	0.0
risk	float	Non-negative multiple of the predictive spread added to the ranking criterion; 0 ranks on predictions alone.	0.0
warn_support	bool	Warn if candidates or the query regime lie outside observed training support. Inspect :meth:support for details.	True

Returns:

Type	Description
dict[str, Any]	The candidate config (a copy) achieving the best predicted
dict[str, Any]	objective under mode.

Raises:

Type	Description
ValueError	If objective is not a fitted observable, if mode is not "min" or "max", if risk is negative, or if there are no candidates to score.

Source code in src/trade_study/regime.py

def recommend(  # ruff: ignore[too-many-arguments]
    self,
    regime: dict[str, Any],
    *,
    objective: str,
    mode: str = "min",
    n_candidates: int = 512,
    seed: int = 0,
    candidates: Sequence[dict[str, Any]] | None = None,
    warn_below_r2: float | None = 0.0,
    risk: float = 0.0,
    warn_support: bool = True,
) -> dict[str, Any]:
    """Recommend a design-factor config at a query regime.

    Samples ``n_candidates`` configs from the design-factor space via
    a scrambled Sobol' sequence and returns the one whose surrogate
    prediction for ``objective`` is best under ``mode``. With
    ``risk > 0`` candidates are ranked pessimistically, by
    ``prediction + risk * spread`` when minimizing and
    ``prediction - risk * spread`` when maximizing, where ``spread`` is
    :meth:`SurrogateModel.spread_batch`; this avoids picking the
    candidate the surrogate is most optimistically wrong about.

    Args:
        regime: Mapping of regime-feature names to values.
        objective: Name of the observable to optimize. Must be one
            of ``self.inner.observable_names``.
        mode: ``"min"`` or ``"max"``.
        n_candidates: Number of design-space samples to evaluate.
            Ignored when ``candidates`` is provided.
        seed: Seed for the Sobol' sampler.
        candidates: Optional explicit list of design-factor configs
            to score; if given, overrides ``n_candidates``.
        warn_below_r2: Warn if ``objective``'s cross-validated R^2 is
            below this threshold, so a caller optimizing
            against a poorly-fit surrogate gets a signal right at the
            point of use, not just buried in fit-time logs. Pass
            ``None`` to disable.
        risk: Non-negative multiple of the predictive spread added to
            the ranking criterion; ``0`` ranks on predictions alone.
        warn_support: Warn if candidates or the query regime lie outside
            observed training support. Inspect :meth:`support` for details.

    Returns:
        The candidate config (a copy) achieving the best predicted
        ``objective`` under ``mode``.

    Raises:
        ValueError: If ``objective`` is not a fitted observable, if
            ``mode`` is not ``"min"`` or ``"max"``, if ``risk`` is
            negative, or if there are no candidates to score.
    """
    if mode not in _SUPPORTED_MODES:
        msg = f"mode must be one of {sorted(_SUPPORTED_MODES)}; got {mode!r}"
        raise ValueError(msg)
    if risk < 0:
        msg = f"risk must be non-negative; got {risk}"
        raise ValueError(msg)
    if objective not in self.inner.observable_names:
        msg = (
            f"objective {objective!r} is not a fitted observable; "
            f"available: {self.inner.observable_names}"
        )
        raise ValueError(msg)
    r2 = self.cv_r2.get(objective, float("nan"))
    if warn_below_r2 is not None and np.isfinite(r2) and r2 < warn_below_r2:
        warnings.warn(
            f"recommend: objective {objective!r} has cross-validated "
            f"R^2={r2:.3f} (< {warn_below_r2}); this recommendation may "
            f"not be trustworthy.",
            UserWarning,
            stacklevel=2,
        )
    pool = (
        list(candidates)
        if candidates is not None
        else build_grid(
            self.factors,
            method="sobol",
            n_samples=n_candidates,
            seed=seed,
        )
    )
    if not pool:
        msg = "recommend: no candidates to score"
        raise ValueError(msg)
    preds = self.predict_batch(regime, pool, warn_support=warn_support)[objective]
    if risk > 0:
        merged = [_merge(regime, c) for c in pool]
        spread = self.inner.spread_batch(merged, warn_support=False)[objective]
        preds = preds + risk * spread if mode == "min" else preds - risk * spread
    idx = int(np.argmin(preds)) if mode == "min" else int(np.argmax(preds))
    return dict(pool[idx])