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 typing import Any, Callable, List, NoReturn, Union 

11 

12import ConfigSpace as CS 

13import flaml.tune.sample 

14import numpy as np 

15import numpy.typing as npt 

16import pytest 

17import scipy 

18from ConfigSpace.hyperparameters import Hyperparameter, NormalIntegerHyperparameter 

19 

20from mlos_core.spaces.converters.flaml import ( 

21 FlamlDomain, 

22 FlamlSpace, 

23 configspace_to_flaml_space, 

24) 

25 

26OptimizerSpace = Union[FlamlSpace, CS.ConfigurationSpace] 

27OptimizerParam = Union[FlamlDomain, Hyperparameter] 

28 

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

30 

31 

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

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

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

35 

36 kurtosis = scipy.stats.kurtosis(arr) 

37 

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

39 

40 frequencies = counts / len(arr) 

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

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

43 

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

45 assert p_value > 0.3 

46 assert f_p_value > 0.5 

47 

48 

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

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

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

52 assert_is_uniform(logs) 

53 

54 

55def test_is_uniform() -> None: 

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

57 np.random.seed(42) 

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

59 assert_is_uniform(uniform) 

60 

61 

62def test_is_log_uniform() -> None: 

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

64 np.random.seed(42) 

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

66 assert_is_log_uniform(log_uniform) 

67 

68 

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

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

71 raise NotImplementedError("subclass must override conversion_function") 

72 

73 

74class BaseConversion(metaclass=ABCMeta): 

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

76 

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

78 

79 @abstractmethod 

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

81 """ 

82 Sample from the given configuration space. 

83 

84 Parameters 

85 ---------- 

86 config_space : CS.ConfigurationSpace 

87 Configuration space to sample from. 

88 n_samples : int 

89 Number of samples to use, by default 1. 

90 """ 

91 

92 @abstractmethod 

93 def get_parameter_names(self, config_space: OptimizerSpace) -> List[str]: 

94 """ 

95 Get the parameter names from the given configuration space. 

96 

97 Parameters 

98 ---------- 

99 config_space : CS.ConfigurationSpace 

100 Configuration space. 

101 """ 

102 

103 @abstractmethod 

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

105 """ 

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

107 

108 Parameters 

109 ---------- 

110 points : np.array 

111 Counts of each categorical value. 

112 """ 

113 

114 @abstractmethod 

115 def test_dimensionality(self) -> None: 

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

117 

118 def test_unsupported_hyperparameter(self) -> None: 

119 input_space = CS.ConfigurationSpace() 

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

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

122 self.conversion_function(input_space) 

123 

124 def test_continuous_bounds(self) -> None: 

125 input_space = CS.ConfigurationSpace() 

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

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

128 

129 converted_space = self.conversion_function(input_space) 

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

131 "a", 

132 "b", 

133 ] 

134 point = self.sample(converted_space) 

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

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

137 

138 def test_uniform_samples(self) -> None: 

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

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

141 converted_space = self.conversion_function(input_space) 

142 

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

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

145 

146 # uniform float 

147 assert_is_uniform(uniform) 

148 

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

150 assert c.upper in integer_uniform 

151 assert c.lower in integer_uniform 

152 # integer uniform 

153 assert_is_uniform(integer_uniform) 

154 

155 def test_uniform_categorical(self) -> None: 

156 input_space = CS.ConfigurationSpace() 

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

158 converted_space = self.conversion_function(input_space) 

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

160 counts = self.categorical_counts(points) 

161 assert 35 < counts[0] < 65 

162 assert 35 < counts[1] < 65 

163 

164 def test_weighted_categorical(self) -> None: 

165 raise NotImplementedError("subclass must override") 

166 

167 def test_log_int_spaces(self) -> None: 

168 raise NotImplementedError("subclass must override") 

169 

170 def test_log_float_spaces(self) -> None: 

171 raise NotImplementedError("subclass must override") 

172 

173 

174class TestFlamlConversion(BaseConversion): 

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

176 

177 conversion_function = staticmethod(configspace_to_flaml_space) 

178 

179 def sample( 

180 self, 

181 config_space: FlamlSpace, # type: ignore[override] 

182 n_samples: int = 1, 

183 ) -> npt.NDArray: 

184 assert isinstance(config_space, dict) 

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

186 ret: npt.NDArray = np.array( 

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

188 ).T 

189 return ret 

190 

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

192 assert isinstance(config_space, dict) 

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

194 return ret 

195 

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

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

198 assert isinstance(counts, np.ndarray) 

199 return counts 

200 

201 def test_dimensionality(self) -> None: 

202 input_space = CS.ConfigurationSpace() 

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

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

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

206 output_space = configspace_to_flaml_space(input_space) 

207 assert len(output_space) == 3 

208 

209 def test_weighted_categorical(self) -> None: 

210 np.random.seed(42) 

211 input_space = CS.ConfigurationSpace() 

212 input_space.add( 

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

214 ) 

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

216 configspace_to_flaml_space(input_space) 

217 

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

219 def test_log_int_spaces(self) -> None: 

220 np.random.seed(42) 

221 # integer is supported 

222 input_space = CS.ConfigurationSpace() 

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

224 converted_space = configspace_to_flaml_space(input_space) 

225 

226 # test log integer sampling 

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

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

229 

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

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

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

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

234 assert_is_log_uniform(integer_log_uniform, base=10) 

235 

236 def test_log_float_spaces(self) -> None: 

237 np.random.seed(42) 

238 

239 # continuous is supported 

240 input_space = CS.ConfigurationSpace() 

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

242 converted_space = configspace_to_flaml_space(input_space) 

243 

244 # test log integer sampling 

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

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

247 

248 assert_is_log_uniform(float_log_uniform) 

249 

250 

251if __name__ == "__main__": 

252 # For attaching debugger debugging: 

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