diff --git a/quapy/classification/svmperf.py b/quapy/classification/svmperf.py index 3fc48f6..b7ba007 100644 --- a/quapy/classification/svmperf.py +++ b/quapy/classification/svmperf.py @@ -67,8 +67,7 @@ class SVMperf(BaseEstimator, ClassifierMixin): # this would allow to run parallel instances of predict random_code = 'svmperfprocess'+'-'.join(str(local_random.randint(0, 1000000)) for _ in range(5)) if self.host_folder is None: - # tmp dir are removed after the fit terminates in multiprocessing... - self.tmpdir = tempfile.TemporaryDirectory(suffix=random_code).name + self.tmpdir = join(tempfile.gettempdir(), random_code) else: self.tmpdir = join(self.host_folder, '.' + random_code) makedirs(self.tmpdir, exist_ok=True) diff --git a/quapy/tests/test_bayesian_utils.py b/quapy/tests/test_bayesian_utils.py new file mode 100644 index 0000000..8ac6058 --- /dev/null +++ b/quapy/tests/test_bayesian_utils.py @@ -0,0 +1,69 @@ +import unittest + +import numpy as np + +from quapy.method._bayesian import ( + _validate_temperature, + _resolve_dirichlet_prior, + kl_div, + js_div, + normalized, + lambda_inverse, + lambda_forward, +) + + +class TestBayesianUtils(unittest.TestCase): + """ + Smoke tests for the pure-numpy helper functions in method/_bayesian.py. These do not require the + optional jax/stan/numpyro dependencies (unlike BayesianKDEy/BayesianMAPLS themselves), so they can + run regardless of whether `quapy[bayes]` is installed. + """ + + def test_validate_temperature(self): + self.assertEqual(_validate_temperature(2), 2.0) + self.assertEqual(_validate_temperature(0.5), 0.5) + with self.assertRaises(ValueError): + _validate_temperature(0) + with self.assertRaises(ValueError): + _validate_temperature(-1) + with self.assertRaises(ValueError): + _validate_temperature('not-a-number') + + def test_resolve_dirichlet_prior(self): + np.testing.assert_array_equal(_resolve_dirichlet_prior('uniform', n_classes=3), np.ones(3)) + np.testing.assert_array_equal(_resolve_dirichlet_prior(2.5, n_classes=3), np.full(3, 2.5)) + np.testing.assert_array_equal(_resolve_dirichlet_prior([1, 2, 3], n_classes=3), np.array([1., 2., 3.])) + with self.assertRaises(ValueError): + _resolve_dirichlet_prior([1, 2], n_classes=3) # wrong shape + with self.assertRaises(ValueError): + _resolve_dirichlet_prior('unknown-prior', n_classes=3) + + def test_kl_div_and_js_div(self): + p = np.array([0.5, 0.5]) + self.assertAlmostEqual(kl_div(p, p), 0.0, places=6) + self.assertAlmostEqual(js_div(p, p), 0.0, places=6) + + q = np.array([0.9, 0.1]) + self.assertGreater(kl_div(p, q), 0.0) + self.assertGreater(js_div(p, q), 0.0) + # JS divergence is symmetric, unlike KL + self.assertAlmostEqual(js_div(p, q), js_div(q, p), places=6) + + def test_normalized(self): + # normalized() operates on a batch of row vectors, shape (n_samples, n_features) + a = np.array([[3.0, 4.0], [0.0, 0.0]]) + normed = normalized(a) + self.assertAlmostEqual(np.linalg.norm(normed[0]), 1.0, places=6) + # an all-zero row should not raise (guarded division), and stays all-zero + np.testing.assert_array_equal(normed[1], np.array([0.0, 0.0])) + + def test_lambda_inverse_forward_roundtrip(self): + dpq, lam = 0.3, 0.4 + gamma = lambda_inverse(dpq, lam) + recovered_lam = lambda_forward(dpq, gamma) + self.assertAlmostEqual(recovered_lam, lam, places=6) + + +if __name__ == '__main__': + unittest.main() diff --git a/quapy/tests/test_calibration.py b/quapy/tests/test_calibration.py new file mode 100644 index 0000000..4bdaae4 --- /dev/null +++ b/quapy/tests/test_calibration.py @@ -0,0 +1,35 @@ +import unittest + +import numpy as np +from sklearn.linear_model import LogisticRegression + +from quapy.classification.calibration import NBVSCalibration, BCTSCalibration, TSCalibration, VSCalibration +from quapy.tests._synthetic import make_labelled_collection + + +class TestCalibration(unittest.TestCase): + + data = make_labelled_collection(n_samples=200, n_features=10, n_classes=3, random_state=23) + + def test_calibration_methods_fit_predict_proba(self): + X, y = self.data.Xy + for calib_cls in [NBVSCalibration, BCTSCalibration, TSCalibration, VSCalibration]: + model = calib_cls(LogisticRegression(max_iter=2000), val_split=5) + model.fit(X, y) + posteriors = model.predict_proba(X) + self.assertEqual(posteriors.shape, (len(y), self.data.n_classes)) + np.testing.assert_allclose(posteriors.sum(axis=1), 1.0, rtol=1e-5, + err_msg=f'{calib_cls.__name__} posteriors do not sum to 1') + predictions = model.predict(X) + self.assertEqual(len(predictions), len(y)) + + def test_calibration_with_float_val_split(self): + X, y = self.data.Xy + model = BCTSCalibration(LogisticRegression(max_iter=2000), val_split=0.3) + model.fit(X, y) + posteriors = model.predict_proba(X) + self.assertEqual(posteriors.shape, (len(y), self.data.n_classes)) + + +if __name__ == '__main__': + unittest.main() diff --git a/quapy/tests/test_confidence.py b/quapy/tests/test_confidence.py new file mode 100644 index 0000000..c8b9d47 --- /dev/null +++ b/quapy/tests/test_confidence.py @@ -0,0 +1,62 @@ +import unittest + +import numpy as np +from sklearn.linear_model import LogisticRegression + +from quapy.method.aggregative import PACC +from quapy.method.confidence import ConfidenceIntervals, ConfidenceEllipseSimplex, AggregativeBootstrap +from quapy.tests._synthetic import make_dataset + + +def _dirichlet_samples(n_classes=3, n_samples=300, random_state=0): + rng = np.random.RandomState(random_state) + return rng.dirichlet(np.ones(n_classes) * 5, size=n_samples) + + +class TestConfidenceRegions(unittest.TestCase): + + def test_confidence_intervals_contain_own_mean(self): + samples = _dirichlet_samples() + region = ConfidenceIntervals(samples) + point_estimate = region.point_estimate() + self.assertEqual(region.coverage(point_estimate), 1.) + self.assertEqual(region.n_dim, 3) + + def test_confidence_ellipse_simplex_contains_own_mean(self): + samples = _dirichlet_samples() + region = ConfidenceEllipseSimplex(samples) + point_estimate = region.point_estimate() + self.assertIn(point_estimate, region) + + def test_simplex_portion_is_cached_and_consistent(self): + # regression test for the @lru_cache-on-bound-method memory leak fix: + # results must still be memoized per instance, and two instances must not share state + region1 = ConfidenceEllipseSimplex(_dirichlet_samples(random_state=1)) + region2 = ConfidenceEllipseSimplex(_dirichlet_samples(random_state=2)) + + p1_first = region1.simplex_portion() + p1_second = region1.simplex_portion() + self.assertEqual(p1_first, p1_second) + + p2 = region2.simplex_portion() + self.assertTrue(hasattr(region1, '_simplex_portion_cache')) + self.assertTrue(hasattr(region2, '_simplex_portion_cache')) + # each instance keeps its own cached value + self.assertEqual(region1._simplex_portion_cache, p1_first) + self.assertEqual(region2._simplex_portion_cache, p2) + + def test_aggregative_bootstrap_end_to_end(self): + dataset = make_dataset(n_train=150, n_test=50, n_classes=3, n_features=12, random_state=5) + learner = LogisticRegression(max_iter=2000) + learner.fit(*dataset.training.Xy) + quantifier = AggregativeBootstrap( + PACC(learner, fit_classifier=False), n_test_samples=50, confidence_level=0.9 + ) + quantifier.fit(*dataset.training.Xy) + point_estimate, region = quantifier.predict_conf(dataset.test.X) + self.assertEqual(len(point_estimate), 3) + self.assertEqual(region.coverage(point_estimate), 1.) + + +if __name__ == '__main__': + unittest.main() diff --git a/quapy/tests/test_reader.py b/quapy/tests/test_reader.py new file mode 100644 index 0000000..a7ee21c --- /dev/null +++ b/quapy/tests/test_reader.py @@ -0,0 +1,56 @@ +import os +import tempfile +import unittest + +import numpy as np + +from quapy.data.reader import from_text, from_sparse, from_csv, reindex_labels, binarize + + +class TestReader(unittest.TestCase): + + def _write_tmp(self, content, suffix='.txt'): + fd, path = tempfile.mkstemp(suffix=suffix) + with os.fdopen(fd, 'w') as f: + f.write(content) + self.addCleanup(os.remove, path) + return path + + def test_from_text(self): + path = self._write_tmp('1\tthis is positive\n0\tthis is negative\n') + sentences, labels = from_text(path, verbose=0) + self.assertEqual(sentences, ['this is positive', 'this is negative']) + self.assertEqual(labels, [1, 0]) + + def test_from_text_skips_malformed_lines(self): + # a line without a tab separator should be skipped (and warned about), not raise + path = self._write_tmp('1\tgood line\nthis line has no label\n0\tanother good line\n') + sentences, labels = from_text(path, verbose=0) + self.assertEqual(sentences, ['good line', 'another good line']) + self.assertEqual(labels, [1, 0]) + + def test_from_sparse(self): + # format: