-
Notifications
You must be signed in to change notification settings - Fork 2k
/
pyunit_automl_distributions.py
141 lines (121 loc) · 6.52 KB
/
pyunit_automl_distributions.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
import sys, os
import numpy as np
import pandas as pd
import h2o
sys.path.insert(1, os.path.join("..", "..", ".."))
from h2o.automl import *
from tests import pyunit_utils as pu
def test_automl_distributions():
scenarios = [
dict(response="binomial", distribution="binomial",
algos=['DRF', 'DeepLearning', 'GBM', 'GLM', 'StackedEnsemble', 'XGBoost'], max_models=12, fail=True),
dict(response="binomial", distribution="bernoulli",
algos=['DRF', 'DeepLearning', 'GBM', 'GLM', 'StackedEnsemble', 'XGBoost'], max_models=12),
dict(response="quasibinomial", distribution="quasibinomial", algos=['GBM', 'GLM', 'StackedEnsemble'],
max_models=17, fail=True), # needed to be able to build SE
dict(response="quasibinomial", distribution="fractionalbinomial", algos=['GLM'], fail=True),
dict(response="multinomial", distribution="multinomial",
algos=['DRF', 'DeepLearning', 'GBM', 'GLM', 'StackedEnsemble', 'XGBoost'], max_models=12),
dict(response="gaussian", distribution="gaussian",
algos=['DRF', 'DeepLearning', 'GBM', 'GLM', 'StackedEnsemble', 'XGBoost'], max_models=12),
dict(response="ordinal", distribution="poisson",
algos=['DeepLearning', 'GBM', 'GLM', 'StackedEnsemble', 'XGBoost'], nrows=400),
dict(response="gaussian", distribution="gamma",
algos=['DeepLearning', 'GBM', 'GLM', 'StackedEnsemble', 'XGBoost'], nrows=400),
dict(response="gaussian", distribution="laplace", algos=['DeepLearning', 'GBM']),
dict(response="gaussian", distribution=dict(type="quantile", quantile_alpha=0.25), algos=['DeepLearning', 'GBM']),
dict(response="gaussian", distribution=dict(type="huber", huber_alpha=.3),
algos=['DeepLearning', 'GBM'], max_models=12),
dict(response="gaussian", distribution=dict(type="tweedie", tweedie_power=1.5),
algos=['DeepLearning', 'GBM', 'GLM', 'StackedEnsemble', 'XGBoost']),
dict(response="ordinal_factors", distribution="ordinal", algos=[], fail=True),
dict(response="gaussian", distribution=dict(type="custom", custom_distribution_func="FILLED_LATER_IN_THE_TEST"), algos=["GBM"]),
]
seed = 9803190
def make_data(size):
np.random.seed(seed=seed)
a = np.tile([0., 0.7333], int(size / 2))
a = a[np.argsort(np.random.uniform(size=size))]
b = 2. * a ** 0.1 + 3. + 8 * np.random.uniform(size=size)
d = np.tile([1, 2, 3, 4, 5, 6, 7, 8], int(size / 8))
c = 3 * d - 2 + np.random.uniform(size=size)
e = np.tile(["class 0", "class A"], int(size / 2))
f = np.tile(["class 0", "class A", "class alpha", "class aleph"], int(size / 4))
df = h2o.H2OFrame(pd.DataFrame(dict(quasibinomial=a, gaussian=b, noise=c, ordinal=d, binomial=e, multinomial=f)))
df["ordinal_factors"] = df["ordinal"].asfactor()
return df
def get_distribution(model_id):
ap = h2o.get_model(model_id).actual_params
if "metalearner_params" in ap.keys():
ap = h2o.get_model(model_id).metalearner().actual_params
distribution = ap.get("distribution", ap.get("family"))
if distribution == "binomial":
distribution = "bernoulli"
return distribution
tests = []
for scenario in scenarios:
def _(scenario):
distribution_name = scenario["distribution"]
if isinstance(distribution_name, dict):
distribution_name = distribution_name["type"]
def test_scenario():
expected_dist = distribution_name
df = make_data(scenario.get("nrows", 264))
# Hack so we don't remove the custom distribution function
if expected_dist == "custom":
from h2o.utils.distributions import CustomDistributionGaussian
custom_dist = h2o.upload_custom_distribution(CustomDistributionGaussian)
scenario["distribution"]["custom_distribution_func"] = custom_dist
aml = H2OAutoML(max_models=scenario.get("max_models", 12),
distribution=scenario["distribution"],
seed=seed,
max_runtime_secs_per_model=1)
try:
aml.train(y=scenario["response"], training_frame=df)
except Exception:
assert scenario.get('fail', False), "This distribution should not have failed."
return
assert not scenario.get('fail', False), "This distribution should have failed."
if aml.leaderboard.nrow == 0:
algos = []
else:
algos = list(set(get_leaderboard(aml, "algo").as_data_frame()["algo"].unique()))
for expected in ['DeepLearning', "DRF", 'GBM', 'GLM', 'StackedEnsemble', 'XGBoost']:
assert expected in algos, "Expected {expected} but no found.".format(expected=expected)
for model_id in aml.leaderboard.as_data_frame()["model_id"]:
distribution = get_distribution(model_id)
assert distribution == expected_dist or \
h2o.get_model(model_id).algo not in [a.lower() for a in scenario["algos"]], (
"{model}: Expected distribution {s_dist} but {distribution} found!".format(
model=model_id,
s_dist=expected_dist,
distribution=distribution
))
test_scenario.__name__ = "test_{}_distribution".format(distribution_name)
return test_scenario
tests.append(_(scenario))
return tests
def test_python_api():
def test_parameterized_distribution_without_param():
aml = H2OAutoML(distribution=dict(type="huber"))
aml = H2OAutoML(distribution="tweedie")
aml = H2OAutoML(distribution="quantile")
def test_parameterized_distribution_without_param2():
try:
aml = H2OAutoML(distribution="custom")
assert False, "should have failed"
except ValueError:
pass
try:
aml = H2OAutoML(distribution=dict(type="custom"))
assert False, "should have failed"
except ValueError:
pass
return [
test_parameterized_distribution_without_param,
test_parameterized_distribution_without_param2,
]
pu.run_tests(
test_automl_distributions() +
test_python_api()
)