Passed
Push — master ( 9f3b80...964643 )
by Simon
01:34
created

RandomSearchSk._run()   A

Complexity

Conditions 3

Size

Total Lines 35
Code Lines 26

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
eloc 26
dl 0
loc 35
rs 9.256
c 0
b 0
f 0
cc 3
nop 6
1
"""Grid search optimizer."""
2
3
# copyright: hyperactive developers, MIT License (see LICENSE file)
4
5
from collections.abc import Sequence
6
7
import numpy as np
8
9
from sklearn.model_selection import ParameterSampler
10
11
from hyperactive.base import BaseOptimizer
12
13
14
class RandomSearchSk(BaseOptimizer):
15
    """Random search optimizer leveraging sklearn's ``ParameterSampler``.
16
17
    Parameters
18
    ----------
19
    param_distributions : dict[str, list | scipy.stats.rv_frozen]
20
        Search space specification. Discrete lists are sampled uniformly;
21
        scipy distribution objects are sampled via their ``rvs`` method.
22
    n_iter : int, default=10
23
        Number of parameter sets to evaluate.
24
    random_state : int | np.random.RandomState | None, default=None
25
        Controls the pseudo-random generator for reproducibility.
26
    error_score : float, default=np.nan
27
        Score assigned when the experiment raises an exception.
28
    experiment : BaseExperiment, optional
29
        Callable returning a scalar score when invoked with keyword
30
        arguments matching a parameter set.
31
32
    Attributes
33
    ----------
34
    best_params_ : dict[str, Any]
35
        Hyper-parameter configuration with the best (lowest) score.
36
    best_score_ : float
37
        Score achieved by ``best_params_``.
38
    best_index_ : int
39
        Index of ``best_params_`` in the sampled sequence.
40
    """
41
42
    def __init__(
43
        self,
44
        param_distributions=None,
45
        n_iter=10,
46
        random_state=None,
47
        error_score=np.nan,
48
        experiment=None,
49
    ):
50
        self.experiment = experiment
51
        self.param_distributions = param_distributions
52
        self.n_iter = n_iter
53
        self.random_state = random_state
54
        self.error_score = error_score
55
56
        super().__init__()
57
58
    @staticmethod
59
    def _is_distribution(obj) -> bool:
60
        """Return True if *obj* looks like a scipy frozen distribution."""
61
        return callable(getattr(obj, "rvs", None))
62
63
    def _check_param_distributions(self, param_distributions):
64
        """Validate ``param_distributions`` similar to sklearn ≤1.0.x."""
65
        if hasattr(param_distributions, "items"):
66
            param_distributions = [param_distributions]
67
68
        for p in param_distributions:
69
            for name, v in p.items():
70
                if self._is_distribution(v):
71
                    # Assume scipy frozen distribution – nothing to check
72
                    continue
73
74
                if isinstance(v, np.ndarray) and v.ndim > 1:
75
                    raise ValueError("Parameter array should be one-dimensional.")
76
77
                if isinstance(v, str) or not isinstance(v, (np.ndarray, Sequence)):
78
                    raise ValueError(
79
                        f"Parameter distribution for ({name}) must be a list, numpy "
80
                        f"array, or scipy.stats ``rv_frozen``, but got ({type(v)})."
81
                        " Single values need to be wrapped in a sequence."
82
                    )
83
84
                if len(v) == 0:
85
                    raise ValueError(
86
                        f"Parameter values for ({name}) need to be a non-empty sequence."
87
                    )
88
89
    def _run(
90
        self,
91
        experiment,
92
        param_distributions,
93
        n_iter,
94
        random_state,
95
        error_score,
96
    ):
97
        """Sample ``n_iter`` points and return the best parameter set."""
98
        self._check_param_distributions(param_distributions)
99
100
        sampler = ParameterSampler(
101
            param_distributions=param_distributions,
102
            n_iter=n_iter,
103
            random_state=random_state,
104
        )
105
        candidate_params = list(sampler)
106
107
        scores: list[float] = []
108
        for candidate_param in candidate_params:
109
            try:
110
                score = experiment(**candidate_param)
111
            except Exception:  # noqa: B904
112
                score = error_score
113
            scores.append(score)
114
115
        best_index = int(np.argmin(scores))  # lower-is-better convention
116
        best_params = candidate_params[best_index]
117
118
        # public attributes for external consumers
119
        self.best_index_ = best_index
120
        self.best_score_ = float(scores[best_index])
121
        self.best_params_ = best_params
122
123
        return best_params
124
125
    @classmethod
126
    def get_test_params(cls, parameter_set: str = "default"):
127
        """Provide deterministic toy configurations for unit tests."""
128
        from hyperactive.experiment.integrations import SklearnCvExperiment
129
        from hyperactive.experiment.toy import Ackley
130
131
        # 1) ML example (Iris + SVC)
132
        sklearn_exp = SklearnCvExperiment.create_test_instance()
133
        param_dist_1 = {
134
            "C": [0.01, 0.1, 1, 10],
135
            "gamma": np.logspace(-4, 1, 6),
136
        }
137
        params_sklearn = {
138
            "experiment": sklearn_exp,
139
            "param_distributions": param_dist_1,
140
            "n_iter": 5,
141
            "random_state": 42,
142
        }
143
144
        # 2) continuous optimisation example (Ackley)
145
        ackley_exp = Ackley.create_test_instance()
146
        param_dist_2 = {
147
            "x0": np.linspace(-5, 5, 50),
148
            "x1": np.linspace(-5, 5, 50),
149
        }
150
        params_ackley = {
151
            "experiment": ackley_exp,
152
            "param_distributions": param_dist_2,
153
            "n_iter": 20,
154
            "random_state": 0,
155
        }
156
157
        return [params_sklearn, params_ackley]
158