Add smoke tests for previously-untested modules, fix wasteful svmperf tmpdir creation
- Add smoke tests covering data/reader.py, method/_threshold_optim.py, classification/calibration.py, method/confidence.py, and the pure-numpy helpers in method/_bayesian.py (skipping the jax/stan-dependent model code itself, consistent with how the aggregative-method registry already treats it as optional) - SVMperf: stop creating a tempfile.TemporaryDirectory() just to discard it immediately for its .name; generate the path directly instead of doing a pointless create/delete/recreate cycle (cleanup already happens via the class's own __del__) Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
parent
b29966797a
commit
ed02be2c8d
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
@ -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()
|
||||
|
|
@ -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()
|
||||
|
|
@ -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: <label> <col:val> <col:val> ... (1-indexed columns)
|
||||
path = self._write_tmp('1 1:0.5 2:1.0\n-1 2:2.0\n', suffix='.dat')
|
||||
X, y = from_sparse(path)
|
||||
self.assertEqual(X.shape[0], 2)
|
||||
np.testing.assert_array_equal(y, np.array([2, 0])) # labels shifted by +1
|
||||
|
||||
def test_from_csv(self):
|
||||
path = self._write_tmp('a,1.0,2.0\nb,3.0,4.0\n', suffix='.csv')
|
||||
X, y = from_csv(path)
|
||||
np.testing.assert_array_equal(X, np.array([[1.0, 2.0], [3.0, 4.0]]))
|
||||
np.testing.assert_array_equal(y, np.array(['a', 'b']))
|
||||
|
||||
def test_reindex_labels(self):
|
||||
indexed, classnames = reindex_labels(['B', 'B', 'A', 'C'])
|
||||
np.testing.assert_array_equal(indexed, np.array([1, 1, 0, 2]))
|
||||
np.testing.assert_array_equal(classnames, np.array(['A', 'B', 'C']))
|
||||
|
||||
def test_binarize(self):
|
||||
binarized = binarize([1, 2, 3, 1, 1, 0], pos_class=2)
|
||||
np.testing.assert_array_equal(binarized, np.array([0, 1, 0, 0, 0, 0]))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
|
@ -0,0 +1,33 @@
|
|||
import unittest
|
||||
|
||||
from sklearn.linear_model import LogisticRegression
|
||||
|
||||
from quapy.functional import check_prevalence_vector
|
||||
from quapy.method.aggregative import T50, MAX, X, MS, MS2
|
||||
from quapy.tests._synthetic import make_dataset
|
||||
|
||||
|
||||
class TestThresholdOptim(unittest.TestCase):
|
||||
|
||||
dataset = make_dataset(n_train=140, n_test=40, n_classes=2, n_features=12, random_state=17, name='synthetic-binary')
|
||||
|
||||
def test_compute_tpr_fpr_edge_cases(self):
|
||||
# regression test for the TP/FN vs TP/FP parameter-naming mix-up in _compute_tpr
|
||||
model = T50()
|
||||
self.assertEqual(model._compute_tpr(TP=5, FN=5), 0.5)
|
||||
self.assertEqual(model._compute_tpr(TP=0, FN=0), 1) # guarded division by zero
|
||||
self.assertEqual(model._compute_fpr(FP=3, TN=7), 0.3)
|
||||
self.assertEqual(model._compute_fpr(FP=0, TN=0), 0) # guarded division by zero
|
||||
|
||||
def test_threshold_methods_fit_predict(self):
|
||||
learner = LogisticRegression(max_iter=2000)
|
||||
learner.fit(*self.dataset.training.Xy)
|
||||
for model_cls in [T50, MAX, X, MS, MS2]:
|
||||
model = model_cls(learner, fit_classifier=False, val_split=None)
|
||||
model.fit(*self.dataset.training.Xy)
|
||||
estim_prevalences = model.predict(self.dataset.test.X)
|
||||
self.assertTrue(check_prevalence_vector(estim_prevalences), f'{model_cls.__name__} failed')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Loading…
Reference in New Issue