Study

Multi-phase study orchestration.

trade_study.Phase(name, grid, filter_fn=None, n_trials=100, n_reps=1, world=None, scorer=None) dataclass

A single phase in a multi-phase study.

Attributes:

Name	Type	Description
name	str	Phase identifier (e.g. "discovery", "refinement").
grid	list[dict[str, Any]] | str | GridCallable	Explicit config list, "carry" to re-use filtered configs from the previous phase, "adaptive" for optuna-driven search, or a callable (ResultsTable, list[Observable]) -> list[dict] that dynamically generates the grid from the previous phase's results.
filter_fn	Callable[[ResultsTable, list[Observable]], NDArray[intp]] | None	Optional callable that takes a ResultsTable and returns indices of configs to pass to the next phase. If None, phase is terminal.
n_trials	int	For adaptive mode, number of optuna trials.
n_reps	int	Number of times to evaluate each design point, forwarded to grid and adaptive runners. Adaptive results already contain per-trial means; grid results retain raw replicate rows.
world	Simulator | None	Optional phase-level simulator override. When set, this phase uses world instead of the Study-level simulator. Useful for multi-fidelity workflows (cheap surrogate first, expensive model later).
scorer	Scorer | None	Optional phase-level scorer override. When set, this phase uses scorer instead of the Study-level scorer.

trade_study.Study(world, scorer, observables, phases, annotations=list(), factors=list(), checkpoint_key=None) dataclass

Multi-phase model criticism study.

Attributes:

Name	Type	Description
world	Simulator	Simulator generating (truth, observations).
scorer	Scorer	Scorer evaluating observables against truth.
observables	list[Observable]	Observable definitions.
phases	list[Phase]	Ordered list of study phases.
annotations	list[Annotation]	External information (costs, constraints).
factors	list[Any]	Factor definitions (needed for adaptive mode).
checkpoint_key	str | None	Caller-managed model/data revision for checkpoint compatibility. Change this when simulator/scorer instance state, external data, globals or callable behavior changes. Structural definitions and inspectable code are checked automatically; opaque callables require an explicit key.

run(*, n_jobs=1, callback=None, checkpoint_dir=None)

Execute all phases sequentially.

Parameters:

Name	Type	Description	Default
n_jobs	int	Number of parallel workers for grid phases.	1
callback	ProgressCallback | None	Optional progress callback invoked after each trial with (trial_index, total_trials, trial_result).	None
checkpoint_dir	str | Path | None	Optional directory for per-phase checkpoints. Each completed phase is saved there; on a rerun, phases already on disk are loaded instead of run, and filters are replayed on them. Results produced elsewhere (for example reduced from cluster shards) can be written into a phase's directory with :func:save_results to stand in for running it.	None

Raises:

Type	Description
ValueError	If a callable grid is used on the first phase (no previous results to pass).

Source code in src/trade_study/study.py

def run(
    self,
    *,
    n_jobs: int = 1,
    callback: ProgressCallback | None = None,
    checkpoint_dir: str | Path | None = None,
) -> None:
    """Execute all phases sequentially.

    Args:
        n_jobs: Number of parallel workers for grid phases.
        callback: Optional progress callback invoked after each trial
            with ``(trial_index, total_trials, trial_result)``.
        checkpoint_dir: Optional directory for per-phase checkpoints.
            Each completed phase is saved there; on a rerun,
            phases already on disk are loaded instead of run, and
            filters are replayed on them. Results produced elsewhere
            (for example reduced from cluster shards) can be written
            into a phase's directory with :func:`save_results` to
            stand in for running it.

    Raises:
        ValueError: If a callable grid is used on the first phase
            (no previous results to pass).
    """
    carry_grid: list[dict[str, Any]] | None = None
    prev_result: ResultsTable | None = None
    checkpoint = Path(checkpoint_dir) if checkpoint_dir is not None else None
    if checkpoint is not None:
        self._write_index(checkpoint)

    for index, phase in enumerate(self.phases):
        saved = (
            self._phase_dir(checkpoint, index) if checkpoint is not None else None
        )
        if saved is not None and (saved / "meta.json").exists():
            result = self._load_phase(saved)
            self._results[phase.name] = result
            prev_result = result
            carry_grid = self._carry(phase)
            continue

        # Resolve phase-level overrides (multi-fidelity support)
        world = phase.world if phase.world is not None else self.world
        scorer = phase.scorer if phase.scorer is not None else self.scorer

        if isinstance(phase.grid, str) and phase.grid == "adaptive":
            result = run_adaptive(
                world,
                scorer,
                self.factors,
                self.observables,
                n_trials=phase.n_trials,
                n_reps=phase.n_reps,
            )
        elif callable(phase.grid):
            if prev_result is None:
                msg = (
                    f"Phase {phase.name!r}: callable grid requires a previous phase"
                )
                raise ValueError(msg)
            grid = phase.grid(prev_result, self.observables)
            result = run_grid(
                world,
                scorer,
                grid,
                self.observables,
                annotations=self.annotations or None,
                n_jobs=n_jobs,
                n_reps=phase.n_reps,
                callback=callback,
                checkpoint_path=saved / "trials.sqlite" if saved else None,
                checkpoint_key=self.checkpoint_key,
            )
        else:
            grid = (
                phase.grid if isinstance(phase.grid, list) else (carry_grid or [])
            )
            result = run_grid(
                world,
                scorer,
                grid,
                self.observables,
                annotations=self.annotations or None,
                n_jobs=n_jobs,
                n_reps=phase.n_reps,
                callback=callback,
                checkpoint_path=saved / "trials.sqlite" if saved else None,
                checkpoint_key=self.checkpoint_key,
            )

        self._results[phase.name] = result
        prev_result = result
        if saved is not None:
            save_results(result, saved)
        carry_grid = self._carry(phase)

save(path)

Save every completed phase's results as a checkpoint.

Parameters:

Name	Type	Description	Default
path	str | Path	Checkpoint directory, compatible with run(checkpoint_dir=...).	required

Source code in src/trade_study/study.py

def save(self, path: str | Path) -> None:
    """Save every completed phase's results as a checkpoint.

    Args:
        path: Checkpoint directory, compatible with
            ``run(checkpoint_dir=...)``.
    """
    root = Path(path)
    self._write_index(root)
    for index, phase in enumerate(self.phases):
        if phase.name in self._results:
            save_results(self._results[phase.name], self._phase_dir(root, index))

load(path)

Load the completed phases of a checkpoint into this study.

Parameters:

Name	Type	Description	Default
path	str | Path	Directory written by :meth:save or by run(checkpoint_dir=...).	required

Returns:

Type	Description
list[str]	Names of the phases loaded.

Raises:

Type	Description
FileNotFoundError	If path holds no checkpoint.

Source code in src/trade_study/study.py

def load(self, path: str | Path) -> list[str]:
    """Load the completed phases of a checkpoint into this study.

    Args:
        path: Directory written by :meth:`save` or by
            ``run(checkpoint_dir=...)``.

    Returns:
        Names of the phases loaded.

    Raises:
        FileNotFoundError: If ``path`` holds no checkpoint.
    """
    root = Path(path)
    if not (root / "study.json").exists():
        msg = f"No study checkpoint at {root}"
        raise FileNotFoundError(msg)
    self._write_index(root)
    loaded = []
    for index, phase in enumerate(self.phases):
        saved = self._phase_dir(root, index)
        if (saved / "meta.json").exists():
            self._results[phase.name] = self._load_phase(saved)
            loaded.append(phase.name)
    return loaded

results(phase)

Get results for a specific phase.

Returns:

Type	Description
ResultsTable	ResultsTable for the given phase.

Source code in src/trade_study/study.py

def results(self, phase: str) -> ResultsTable:
    """Get results for a specific phase.

    Returns:
        ResultsTable for the given phase.
    """
    return self._results[phase]

front(phase)

Get Pareto front indices for a phase.

Returns:

Type	Description
NDArray[intp]	Integer array of Pareto-optimal row indices.

Source code in src/trade_study/study.py

def front(self, phase: str) -> NDArray[np.intp]:
    """Get Pareto front indices for a phase.

    Returns:
        Integer array of Pareto-optimal row indices.
    """
    r = self._results[phase]
    dirs = [o.direction for o in self.observables]
    wts = [o.weight for o in self.observables]
    return extract_front(r.scores, dirs, wts)

front_hypervolume(phase, ref_point)

Compute hypervolume of the Pareto front for a phase.

Returns:

Type	Description
float	Hypervolume value.

Source code in src/trade_study/study.py

def front_hypervolume(
    self,
    phase: str,
    ref_point: NDArray[np.floating[Any]],
) -> float:
    """Compute hypervolume of the Pareto front for a phase.

    Returns:
        Hypervolume value.
    """
    r = self._results[phase]
    dirs = [o.direction for o in self.observables]
    wts = [o.weight for o in self.observables]
    front_idx = extract_front(r.scores, dirs, wts)
    return hypervolume(r.scores[front_idx], ref_point, dirs, wts)

compare_phases(ref_point=None)

Compare the fronts of the completed phases.

Each phase is summarized over its design points: replicate rows are averaged first when the phase ran with n_reps > 1, as for filtering, and rows with a non-finite score are left out of the front. Each phase after the first is compared with the previous completed phase by IGD+ in both directions:

• igd_plus_gain: IGD+ of the previous front against this phase's front, i.e. how far the previous front falls short of this one. Positive when this phase found better trade-offs.
• igd_plus_loss: IGD+ of this phase's front against the previous front, i.e. how far this phase falls short of the previous one. Zero when this front weakly dominates the previous front.

Parameters:

Name	Type	Description	Default
ref_point	NDArray[floating[Any]] | None	Hypervolume reference point, in the observables' units. Defaults to the worst value of each observable over every phase's front, pushed outward by 10% of its range (by 1 when the range is zero), so hypervolumes are comparable across phases.	None

Returns:

Type	Description
list[dict[str, Any]]	One dict per completed phase, in phase order, with phase,
list[dict[str, Any]]	n_trials (design points), n_front, hypervolume,
list[dict[str, Any]]	best (best value of each observable over the phase's design
list[dict[str, Any]]	points) and igd_plus_gain/igd_plus_loss (None for the
list[dict[str, Any]]	first phase). Phases without a finite front report NaN.

Source code in src/trade_study/study.py

def compare_phases(
    self,
    ref_point: NDArray[np.floating[Any]] | None = None,
) -> list[dict[str, Any]]:
    """Compare the fronts of the completed phases.

    Each phase is summarized over its design points: replicate rows are
    averaged first when the phase ran with ``n_reps > 1``, as for
    filtering, and rows with a non-finite score are left out of the
    front. Each phase after the first is compared with the previous
    completed phase by IGD+ in both directions:

    * ``igd_plus_gain``: IGD+ of the previous front against this
      phase's front, i.e. how far the previous front falls short of this
      one. Positive when this phase found better trade-offs.
    * ``igd_plus_loss``: IGD+ of this phase's front against the previous
      front, i.e. how far this phase falls short of the previous one.
      Zero when this front weakly dominates the previous front.

    Args:
        ref_point: Hypervolume reference point, in the observables'
            units. Defaults to the worst value of each observable over
            every phase's front, pushed outward by 10% of its range (by 1
            when the range is zero), so hypervolumes are comparable
            across phases.

    Returns:
        One dict per completed phase, in phase order, with ``phase``,
        ``n_trials`` (design points), ``n_front``, ``hypervolume``,
        ``best`` (best value of each observable over the phase's design
        points) and ``igd_plus_gain``/``igd_plus_loss`` (``None`` for the
        first phase). Phases without a finite front report NaN.
    """
    dirs = [o.direction for o in self.observables]
    wts = [o.weight for o in self.observables]
    tables = [
        (phase.name, self._design_points(phase))
        for phase in self.phases
        if phase.name in self._results
    ]
    fronts = {
        name: _finite_front(table.scores, dirs, wts) for name, table in tables
    }
    reference = (
        np.asarray(ref_point, dtype=float)
        if ref_point is not None
        else _padded_worst(list(fronts.values()), dirs)
    )
    rows: list[dict[str, Any]] = []
    previous: NDArray[np.floating[Any]] | None = None
    for name, table in tables:
        front = fronts[name]
        row: dict[str, Any] = {
            "phase": name,
            "n_trials": len(table.configs),
            "n_front": len(front),
            "hypervolume": (
                hypervolume(front, reference, dirs, wts) if len(front) else np.nan
            ),
            "best": _best_values(table, self.observables),
            "igd_plus_gain": None,
            "igd_plus_loss": None,
        }
        if previous is not None:
            comparable = len(front) > 0 and len(previous) > 0
            row["igd_plus_gain"] = (
                igd_plus(previous, front, dirs, wts) if comparable else np.nan
            )
            row["igd_plus_loss"] = (
                igd_plus(front, previous, dirs, wts) if comparable else np.nan
            )
        rows.append(row)
        previous = front
    return rows

stack(phase, *, maximize=False)

Compute score-based stacking weights for a phase.

Returns:

Type	Description
NDArray[floating[Any]]	Array of stacking weights.

Source code in src/trade_study/study.py

def stack(
    self,
    phase: str,
    *,
    maximize: bool = False,
) -> NDArray[np.floating[Any]]:
    """Compute score-based stacking weights for a phase.

    Returns:
        Array of stacking weights.
    """
    r = self._results[phase]
    return stack_scores(r.scores.T, maximize=maximize)

summary()

Per-phase summary: n_trials, n_front, observable ranges.

Returns:

Type	Description
dict[str, dict[str, Any]]	Dictionary mapping phase names to summary statistics.

Source code in src/trade_study/study.py

def summary(self) -> dict[str, dict[str, Any]]:
    """Per-phase summary: n_trials, n_front, observable ranges.

    Returns:
        Dictionary mapping phase names to summary statistics.
    """
    out: dict[str, dict[str, Any]] = {}
    for name, r in self._results.items():
        dirs = [o.direction for o in self.observables]
        wts = [o.weight for o in self.observables]
        front_idx = extract_front(r.scores, dirs, wts)
        out[name] = {
            "n_trials": len(r.configs),
            "n_front": len(front_idx),
            "observable_ranges": {
                obs: {
                    "min": float(np.nanmin(r.scores[:, i])),
                    "max": float(np.nanmax(r.scores[:, i])),
                }
                for i, obs in enumerate(r.observable_names)
            },
        }
    return out

trade_study.top_k_pareto_filter(k, objective_names=None)

Create a filter that keeps the top-K configs by Pareto rank.

Parameters:

Name	Type	Description	Default
k	int	Maximum number of configs to keep.	required
objective_names	list[str] | None	Subset of observables to use for ranking. If None, uses all observables.	None

Returns:

Type	Description
Callable[[ResultsTable, list[Observable]], NDArray[intp]]	Filter function compatible with Phase.filter_fn.

Source code in src/trade_study/study.py

def top_k_pareto_filter(
    k: int,
    objective_names: list[str] | None = None,
) -> Callable[[ResultsTable, list[Observable]], NDArray[np.intp]]:
    """Create a filter that keeps the top-K configs by Pareto rank.

    Args:
        k: Maximum number of configs to keep.
        objective_names: Subset of observables to use for ranking.
            If None, uses all observables.

    Returns:
        Filter function compatible with Phase.filter_fn.
    """

    def _filter(
        results: ResultsTable,
        observables: list[Observable],
    ) -> NDArray[np.intp]:
        if objective_names is not None:
            cols = [results.observable_names.index(n) for n in objective_names]
            scores = results.scores[:, cols]
            subset = [o for o in observables if o.name in objective_names]
            dirs = [o.direction for o in subset]
            wts = [o.weight for o in subset]
        else:
            scores = results.scores
            dirs = [o.direction for o in observables]
            wts = [o.weight for o in observables]

        ranks = pareto_rank(scores, dirs, wts)
        order = np.argsort(ranks)
        return order[:k]

    return _filter

trade_study.weighted_sum_filter(weights, k)

Create a filter that keeps the top-K configs by weighted sum.

Scalarises multiple objectives into a single score via a weighted sum and keeps the k best configs. Scores are min-max normalised before weighting so that objectives on different scales are comparable. MAXIMIZE objectives are negated before normalisation so that lower normalised values are always better.

Parameters:

Name	Type	Description	Default
weights	dict[str, float]	Mapping from observable name to its scalarisation weight. Only the named observables are used; the rest are ignored.	required
k	int	Maximum number of configs to keep.	required

Returns:

Type	Description
Callable[[ResultsTable, list[Observable]], NDArray[intp]]	Filter function compatible with Phase.filter_fn.

Source code in src/trade_study/study.py

def weighted_sum_filter(
    weights: dict[str, float],
    k: int,
) -> Callable[[ResultsTable, list[Observable]], NDArray[np.intp]]:
    """Create a filter that keeps the top-K configs by weighted sum.

    Scalarises multiple objectives into a single score via a weighted sum
    and keeps the ``k`` best configs.  Scores are min-max normalised
    before weighting so that objectives on different scales are
    comparable.  MAXIMIZE objectives are negated before normalisation so
    that lower normalised values are always better.

    Args:
        weights: Mapping from observable name to its scalarisation weight.
            Only the named observables are used; the rest are ignored.
        k: Maximum number of configs to keep.

    Returns:
        Filter function compatible with ``Phase.filter_fn``.
    """

    def _filter(
        results: ResultsTable,
        observables: list[Observable],
    ) -> NDArray[np.intp]:
        obs_lookup = {o.name: o for o in observables}
        cols = [results.observable_names.index(n) for n in weights]
        raw = results.scores[:, cols].copy()

        # Flip MAXIMIZE objectives so lower is always better
        for j, name in enumerate(weights):
            if obs_lookup[name].direction == Direction.MAXIMIZE:
                raw[:, j] = -raw[:, j]

        # Min-max normalise each column to [0, 1]
        col_min = np.nanmin(raw, axis=0)
        col_max = np.nanmax(raw, axis=0)
        span = col_max - col_min
        span[span == 0] = 1.0  # avoid division by zero for constant cols
        normed = (raw - col_min) / span

        w = np.array([weights[n] for n in weights])
        scalar = normed @ w
        order = np.argsort(scalar)
        return order[:k].astype(np.intp)

    return _filter

trade_study.feasibility_filter(constraints)

Create a filter that keeps only designs satisfying all constraints.

Parameters:

Name	Type	Description	Default
constraints	list[Constraint]	Constraint objects to evaluate against results.	required

Returns:

Type	Description
Callable[[ResultsTable, list[Observable]], NDArray[intp]]	Filter function compatible with Phase.filter_fn.

Source code in src/trade_study/study.py

def feasibility_filter(
    constraints: list[Constraint],
) -> Callable[[ResultsTable, list[Observable]], NDArray[np.intp]]:
    """Create a filter that keeps only designs satisfying all constraints.

    Args:
        constraints: Constraint objects to evaluate against results.

    Returns:
        Filter function compatible with ``Phase.filter_fn``.
    """

    def _filter(
        results: ResultsTable,
        _observables: list[Observable],
    ) -> NDArray[np.intp]:
        mask = results.feasible(constraints)
        return np.nonzero(mask)[0].astype(np.intp)

    return _filter

Checkpoint identity

Study.run(checkpoint_dir=...), Study.save() and Study.load() share a versioned manifest. It checks phase order, grids, replication/trial counts, factors, objectives, annotations, filters, and inspectable simulator/scorer code. Incompatible definitions and legacy manifests are refused; use a new checkpoint directory when intentionally changing the study.

Set Study(..., checkpoint_key="model-v2-data-v1") to identify simulator/scorer instance configuration, external data, globals, and other behavior that cannot be inferred from code. Update the key whenever those inputs change. Opaque callables require a key. A matching key does not override structural mismatches. A checkpoint is not a serialization of live simulator/scorer objects.

Grid phases also write a trials.sqlite ledger inside their phase directory. An interrupted phase resumes its unfinished evaluations without repeating completed design-point/replicate tasks. Adaptive phases retain phase-level checkpointing; use a persistent AdaptiveSession for external adaptive trial recovery. Final phase results keep the existing save_results() format.

Externally evaluated phase results can still be supplied with save_results() after initializing the manifest using Study.save(). Their observable schema must match the study.