Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion src/hyperactive/base/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
# copyright: hyperactive developers, MIT License (see LICENSE file)

from hyperactive.base._experiment import BaseExperiment
from hyperactive.base._history import SearchHistory
from hyperactive.base._optimizer import BaseOptimizer

__all__ = ["BaseExperiment", "BaseOptimizer"]
__all__ = ["BaseExperiment", "BaseOptimizer", "SearchHistory"]
58 changes: 58 additions & 0 deletions src/hyperactive/base/_experiment.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,15 @@

# copyright: hyperactive developers, MIT License (see LICENSE file)

from __future__ import annotations

import time

import numpy as np
from skbase.base import BaseObject

from hyperactive.base._history import SearchHistory


class BaseExperiment(BaseObject):
"""Base class for experiment."""
Expand All @@ -21,6 +27,7 @@ class BaseExperiment(BaseObject):

def __init__(self):
super().__init__()
self._history = SearchHistory()

def __call__(self, params):
"""Score parameters. Same as score call, returns only a first element."""
Expand Down Expand Up @@ -77,8 +84,20 @@ def evaluate(self, params):
f"Parameters passed to {type(self)}.evaluate do not match: "
f"expected {paramnames}, got {list(params.keys())}."
)

start_time = time.perf_counter()
res, metadata = self._evaluate(params)
eval_time = time.perf_counter() - start_time

res = np.float64(res)

self._history.record(
params=params,
score=res,
metadata=metadata,
eval_time=eval_time,
)

return res, metadata

def _evaluate(self, params):
Expand Down Expand Up @@ -141,3 +160,42 @@ def score(self, params):
metadata = eval_res[1]

return sign * value, metadata

@property
def data(self) -> SearchHistory:
"""Access the collected data from optimization runs.

Tracks all evaluations during optimization. Data accumulates across
multiple optimization runs on the same experiment instance.

Returns
-------
SearchHistory
The data object with the following attributes and methods:

Attributes:
- ``history``: list[dict] - all recorded evaluations
- ``n_trials``: int - total number of trials
- ``n_runs``: int - number of optimization runs
- ``best_trial``: dict | None - trial with highest score
- ``best_score``: float | None - highest score
- ``best_params``: dict | None - parameters of best trial

Methods:
- ``get_run(run_id)``: get trials from specific run
- ``clear()``: reset all data
- ``new_run()``: signal start of new run (call before each run)

Examples
--------
>>> experiment.data.history # all evaluations as list of dicts
>>> experiment.data.best_score # highest score
>>> experiment.data.get_run(0) # evaluations from first run
>>> experiment.data.clear() # reset data

To convert to a pandas DataFrame::

import pandas as pd
df = pd.DataFrame(experiment.data.history)
"""
return self._history
163 changes: 163 additions & 0 deletions src/hyperactive/base/_history.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,163 @@
"""Search history for tracking optimization trials."""

# copyright: hyperactive developers, MIT License (see LICENSE file)

from __future__ import annotations


class SearchHistory:
"""Container for tracking optimization trial history.

This class collects data from each evaluation during optimization runs.
History accumulates across multiple optimization runs on the same experiment.

Attributes
----------
trials : list[dict]
List of all recorded trials. Each trial is a dict containing:
- iteration: int, global iteration index
- run_id: int, which optimization run (0-indexed)
- params: dict, the evaluated parameters
- score: float, the evaluation score (raw, not sign-corrected)
- metadata: dict, additional metadata from the experiment
- eval_time: float, evaluation time in seconds
"""

def __init__(self):
self._trials: list[dict] = []
self._current_run_id: int = 0

def record(
self,
params: dict,
score: float,
metadata: dict | None,
eval_time: float,
) -> None:
"""Record a single trial.

Parameters
----------
params : dict
The evaluated parameters.
score : float
The evaluation score (raw, not sign-corrected).
metadata : dict or None
Additional metadata from the experiment.
eval_time : float
Evaluation time in seconds.
"""
self._trials.append({
"iteration": len(self._trials),
"run_id": self._current_run_id,
"params": dict(params),
"score": float(score),
"metadata": dict(metadata) if metadata else {},
"eval_time": float(eval_time),
})

def new_run(self) -> None:
"""Signal the start of a new optimization run.

Increments the run_id counter. Subsequent trials will be tagged
with the new run_id.
"""
self._current_run_id += 1

def clear(self) -> None:
"""Clear all history data and reset run counter."""
self._trials = []
self._current_run_id = 0

@property
def history(self) -> list[dict]:
"""Return all recorded evaluations as a list.

Returns
-------
list[dict]
List of all evaluations. Each entry contains iteration, run_id,
params, score, metadata, and eval_time.
"""
return self._trials

@property
def n_trials(self) -> int:
"""Return the total number of recorded trials.

Returns
-------
int
Number of trials across all runs.
"""
return len(self._trials)

@property
def n_runs(self) -> int:
"""Return the number of optimization runs.

Returns
-------
int
Number of runs (0 if no trials recorded yet).
"""
return self._current_run_id + 1

@property
def best_trial(self) -> dict | None:
"""Return the trial with the highest score.

Returns
-------
dict or None
The trial dict with the highest score, or None if no trials.
"""
if not self._trials:
return None
return max(self._trials, key=lambda t: t["score"])

@property
def best_score(self) -> float | None:
"""Return the highest score across all trials.

Returns
-------
float or None
The highest score, or None if no trials.
"""
best = self.best_trial
return best["score"] if best else None

@property
def best_params(self) -> dict | None:
"""Return the parameters of the best trial.

Returns
-------
dict or None
Parameters of the trial with highest score, or None if no trials.
"""
best = self.best_trial
return best["params"] if best else None

def get_run(self, run_id: int) -> list[dict]:
"""Return all trials from a specific run.

Parameters
----------
run_id : int
The run identifier (0-indexed).

Returns
-------
list[dict]
List of trials from the specified run.
"""
return [t for t in self._trials if t["run_id"] == run_id]

def __len__(self) -> int:
"""Return the number of trials."""
return len(self._trials)

def __repr__(self) -> str:
return f"SearchHistory(n_trials={self.n_trials}, n_runs={self.n_runs})"
Loading
Loading