Coverage for mlos_core/mlos_core/tests/spaces/spaces_test.py: 94%

1# 

2# Copyright (c) Microsoft Corporation. 

3# Licensed under the MIT License. 

4# 

5"""Tests for mlos_core.spaces.""" 

6 

7# pylint: disable=missing-function-docstring 

8 

9from abc import ABCMeta, abstractmethod 

10from collections.abc import Callable 

11from typing import Any, NoReturn 

12 

13import ConfigSpace as CS 

14import flaml.tune.sample 

15import numpy as np 

16import numpy.typing as npt 

17import pytest 

18import scipy 

19from ConfigSpace.hyperparameters import Hyperparameter, NormalIntegerHyperparameter 

20 

21from mlos_core.spaces.converters.flaml import ( 

22 FlamlDomain, 

23 FlamlSpace, 

24 configspace_to_flaml_space, 

25) 

26 

27OptimizerSpace = FlamlSpace | CS.ConfigurationSpace 

28OptimizerParam = FlamlDomain | Hyperparameter 

29 

30NP_E: float = np.e # type: ignore[misc,unused-ignore] # false positive (read deleted variable) 

31 

32 

33def assert_is_uniform(arr: npt.NDArray) -> None: 

34 """Implements a few tests for uniformity.""" 

35 _values, counts = np.unique(arr, return_counts=True) 

36 

37 kurtosis = scipy.stats.kurtosis(arr) 

38 

39 _chi_sq, p_value = scipy.stats.chisquare(counts) 

40 

41 frequencies = counts / len(arr) 

42 assert np.isclose(frequencies.sum(), 1) 

43 _f_chi_sq, f_p_value = scipy.stats.chisquare(frequencies) 

44 

45 assert np.isclose(kurtosis, -1.2, atol=0.1) 

46 assert p_value > 0.3 

47 assert f_p_value > 0.5 

48 

49 

50def assert_is_log_uniform(arr: npt.NDArray, base: float = NP_E) -> None: 

51 """Checks whether an array is log uniformly distributed.""" 

52 logs = np.log(arr) / np.log(base) 

53 assert_is_uniform(logs) 

54 

55 

56def test_is_uniform() -> None: 

57 """Test our uniform distribution check function.""" 

58 np.random.seed(42) 

59 uniform = np.random.uniform(1, 20, 1000) 

60 assert_is_uniform(uniform) 

61 

62 

63def test_is_log_uniform() -> None: 

64 """Test our log uniform distribution check function.""" 

65 np.random.seed(42) 

66 log_uniform = np.exp(np.random.uniform(np.log(1), np.log(20), 1000)) 

67 assert_is_log_uniform(log_uniform) 

68 

69 

70def invalid_conversion_function(*args: Any) -> NoReturn: 

71 """A quick dummy function for the base class to make pylint happy.""" 

72 raise NotImplementedError("subclass must override conversion_function") 

73 

74 

75class BaseConversion(metaclass=ABCMeta): 

76 """Base class for testing optimizer space conversions.""" 

77 

78 conversion_function: Callable[..., OptimizerSpace] = invalid_conversion_function 

79 

80 @abstractmethod 

81 def sample(self, config_space: OptimizerSpace, n_samples: int = 1) -> npt.NDArray: 

82 """ 

83 Sample from the given configuration space. 

84 

85 Parameters 

86 ---------- 

87 config_space : CS.ConfigurationSpace 

88 Configuration space to sample from. 

89 n_samples : int 

90 Number of samples to use, by default 1. 

91 """ 

92 

93 @abstractmethod 

94 def get_parameter_names(self, config_space: OptimizerSpace) -> list[str]: 

95 """ 

96 Get the parameter names from the given configuration space. 

97 

98 Parameters 

99 ---------- 

100 config_space : CS.ConfigurationSpace 

101 Configuration space. 

102 """ 

103 

104 @abstractmethod 

105 def categorical_counts(self, points: npt.NDArray) -> npt.NDArray: 

106 """ 

107 Get the counts of each categorical value in the given points. 

108 

109 Parameters 

110 ---------- 

111 points : np.array 

112 Counts of each categorical value. 

113 """ 

114 

115 @abstractmethod 

116 def test_dimensionality(self) -> None: 

117 """Check that the dimensionality of the converted space is correct.""" 

118 

119 def test_unsupported_hyperparameter(self) -> None: 

120 input_space = CS.ConfigurationSpace() 

121 input_space.add(NormalIntegerHyperparameter("a", mu=50, sigma=5, lower=0, upper=99)) 

122 with pytest.raises(ValueError, match="NormalIntegerHyperparameter"): 

123 self.conversion_function(input_space) 

124 

125 def test_continuous_bounds(self) -> None: 

126 input_space = CS.ConfigurationSpace() 

127 input_space.add(CS.UniformFloatHyperparameter("a", lower=100, upper=200)) 

128 input_space.add(CS.UniformIntegerHyperparameter("b", lower=-10, upper=-5)) 

129 

130 converted_space = self.conversion_function(input_space) 

131 assert self.get_parameter_names(converted_space) == [ # pylint: disable=unreachable 

132 "a", 

133 "b", 

134 ] 

135 point = self.sample(converted_space) 

136 assert 100 <= point[0] <= 200 

137 assert -10 <= point[1] <= -5 

138 

139 def test_uniform_samples(self) -> None: 

140 c = CS.UniformIntegerHyperparameter("c", lower=1, upper=20) 

141 input_space = CS.ConfigurationSpace({"a": (1.0, 5.0), "c": c}) 

142 converted_space = self.conversion_function(input_space) 

143 

144 np.random.seed(42) # pylint: disable=unreachable 

145 uniform, integer_uniform = self.sample(converted_space, n_samples=1000).T 

146 

147 # uniform float 

148 assert_is_uniform(uniform) 

149 

150 # Check that we get both ends of the sampled range returned to us. 

151 assert c.upper in integer_uniform 

152 assert c.lower in integer_uniform 

153 # integer uniform 

154 assert_is_uniform(integer_uniform) 

155 

156 def test_uniform_categorical(self) -> None: 

157 input_space = CS.ConfigurationSpace() 

158 input_space.add(CS.CategoricalHyperparameter("c", choices=["foo", "bar"])) 

159 converted_space = self.conversion_function(input_space) 

160 points = self.sample(converted_space, n_samples=100) # pylint: disable=unreachable 

161 counts = self.categorical_counts(points) 

162 assert 35 < counts[0] < 65 

163 assert 35 < counts[1] < 65 

164 

165 def test_weighted_categorical(self) -> None: 

166 raise NotImplementedError("subclass must override") 

167 

168 def test_log_int_spaces(self) -> None: 

169 raise NotImplementedError("subclass must override") 

170 

171 def test_log_float_spaces(self) -> None: 

172 raise NotImplementedError("subclass must override") 

173 

174 

175class TestFlamlConversion(BaseConversion): 

176 """Tests for ConfigSpace to Flaml parameter conversions.""" 

177 

178 conversion_function = staticmethod(configspace_to_flaml_space) 

179 

180 def sample( 

181 self, 

182 config_space: OptimizerSpace, 

183 n_samples: int = 1, 

184 ) -> npt.NDArray: 

185 assert isinstance(config_space, dict) # FlamlSpace 

186 assert isinstance(next(iter(config_space.values())), flaml.tune.sample.Domain) 

187 ret: npt.NDArray = np.array( 

188 [domain.sample(size=n_samples) for domain in config_space.values()] 

189 ).T 

190 return ret 

191 

192 def get_parameter_names(self, config_space: FlamlSpace) -> list[str]: # type: ignore[override] 

193 assert isinstance(config_space, dict) 

194 ret: list[str] = list(config_space.keys()) 

195 return ret 

196 

197 def categorical_counts(self, points: npt.NDArray) -> npt.NDArray: 

198 _vals, counts = np.unique(points, return_counts=True) 

199 assert isinstance(counts, np.ndarray) 

200 return counts 

201 

202 def test_dimensionality(self) -> None: 

203 input_space = CS.ConfigurationSpace() 

204 input_space.add(CS.UniformIntegerHyperparameter("a", lower=1, upper=10)) 

205 input_space.add(CS.CategoricalHyperparameter("b", choices=["bof", "bum"])) 

206 input_space.add(CS.CategoricalHyperparameter("c", choices=["foo", "bar"])) 

207 output_space = configspace_to_flaml_space(input_space) 

208 assert len(output_space) == 3 

209 

210 def test_weighted_categorical(self) -> None: 

211 np.random.seed(42) 

212 input_space = CS.ConfigurationSpace() 

213 input_space.add( 

214 CS.CategoricalHyperparameter("c", choices=["foo", "bar"], weights=[0.9, 0.1]) 

215 ) 

216 with pytest.raises(ValueError, match="non-uniform"): 

217 configspace_to_flaml_space(input_space) 

218 

219 @pytest.mark.skip(reason="FIXME: flaml sampling is non-log-uniform") 

220 def test_log_int_spaces(self) -> None: 

221 np.random.seed(42) 

222 # integer is supported 

223 input_space = CS.ConfigurationSpace() 

224 input_space.add(CS.UniformIntegerHyperparameter("d", lower=1, upper=20, log=True)) 

225 converted_space = configspace_to_flaml_space(input_space) 

226 

227 # test log integer sampling 

228 integer_log_uniform = self.sample(converted_space, n_samples=1000) 

229 integer_log_uniform = np.array(integer_log_uniform).ravel() 

230 

231 # FIXME: this fails - flaml is calling np.random.uniform() on base 10 

232 # logs of the bounds as expected but for some reason the resulting 

233 # samples are more skewed towards the lower end of the range 

234 # See Also: https://github.com/microsoft/FLAML/issues/1104 

235 assert_is_log_uniform(integer_log_uniform, base=10) 

236 

237 def test_log_float_spaces(self) -> None: 

238 np.random.seed(42) 

239 

240 # continuous is supported 

241 input_space = CS.ConfigurationSpace() 

242 input_space.add(CS.UniformFloatHyperparameter("b", lower=1, upper=5, log=True)) 

243 converted_space = configspace_to_flaml_space(input_space) 

244 

245 # test log integer sampling 

246 float_log_uniform = self.sample(converted_space, n_samples=1000) 

247 float_log_uniform = np.array(float_log_uniform).ravel() 

248 

249 assert_is_log_uniform(float_log_uniform) 

250 

251 

252if __name__ == "__main__": 

253 # For attaching debugger debugging: 

254 pytest.main(["-vv", "-k", "test_log_int_spaces", __file__])