diff --git a/sequentialized_barnard_tests/base.py b/sequentialized_barnard_tests/base.py index d03b392..86f39fa 100644 --- a/sequentialized_barnard_tests/base.py +++ b/sequentialized_barnard_tests/base.py @@ -1,251 +1,31 @@ """Base class definitions. -This module defines base classes for hypothesis tests. +This module re-exports shared test interfaces from statistical_comparison_core. +The canonical definitions now live in statistical_comparison_core.base. """ -from abc import ABC, abstractmethod -from dataclasses import dataclass -from enum import Enum -from typing import Any, Optional, Type, Union - -from numpy.typing import ArrayLike - - -class Decision(Enum): - """Enum class to represent the decision of a hypothesis test.""" - - AcceptNull = 0 - AcceptAlternative = 1 - FailToDecide = 2 - - -@dataclass -class TestResult: - """Data class defining the result of a hypothesis test. - - A result must contain a decision. Any auxiliary information will be thrown into - info as an optional dictionary. - """ - - __test__ = False - - decision: Decision - info: Optional[dict] = None - - -class TwoSampleBinomialHypothesis(Enum): - """Enum class to represent the hypothesis of a test that compares parameters of two - Bernoulli distributions. - - We assume we have two distributions Bernoulli(p_0) and Bernoulli(p_1). A pair of - data is drawn independently from the two distributions, one at a time. The first - hypothesis, `P0LessThanP1`, represents `p_0 < p_1`. The second hypothesis, - `P0MoreThanP1`, represents `p_0 > p_1`. Note that both of these are possible - alternative hypotheses in a statistical test, and the user will be tasked with - choosing one as their alternative when instantiating a test. - """ - - P0LessThanP1 = 0 - P0MoreThanP1 = 1 - - -Hypothesis = TwoSampleBinomialHypothesis - - -class TwoSampleTestBase(ABC): - """Abstract base class for a two-sample hypothesis test.""" - - @abstractmethod - def run_on_sequence( - self, - sequence_0: ArrayLike, - sequence_1: ArrayLike, - *args, - **kwargs, - ) -> TestResult: - """Runs the test on a pair of sequential data. - - Args: - sequence_0: Sequence of data from the first source. - sequence_1: Sequence of data from the second source. - *args: Additional positional arguments (if any). - **kwargs: Additional optional or keyward arguments (if any). - - - Returns: - TestResult: Result of the hypothesis test. - """ - pass - - -TestBase = TwoSampleTestBase - - -class SequentialTwoSampleTestBase(TwoSampleTestBase): - """Base class for a family of sequential two-sample hypothesis tests.""" - - def run_on_sequence( - self, - sequence_0: ArrayLike, - sequence_1: ArrayLike, - *args, - **kwargs, - ) -> TestResult: - """Runs the test on a pair of sequential data. - - Args: - sequence_0: Sequence of data from the first source. - sequence_1: Sequence of data from the second source. - *args: Additional positional arguments (if any). - **kwargs: Additional optional or keyward arguments (if any). - - Returns: - TestResult: Result of the hypothesis test. - - Raises: - ValueError: If the two sequences do not have the same length. - """ - self.reset() - if not (len(sequence_0) == len(sequence_1)): - raise (ValueError("The two input sequences must have the same size.")) - - result = TestResult(decision=Decision.FailToDecide) - for idx in range(len(sequence_0)): - result = self.step(sequence_0[idx], sequence_1[idx], *args, **kwargs) - if not result.decision == Decision.FailToDecide: - break - return result - - def reset(self) -> None: - """Resets internal memory states of the test.""" - pass - - @abstractmethod - def step( - self, - datum_0: Union[bool, int, float], - datum_1: Union[bool, int, float], - *args, - **kwargs, - ) -> TestResult: - """Runs the test on a single pair of data. - - Args: - datum_0: Datum from the first source. - datum_1: Datum from the second source. - *args: Additional positional arguments (if any). - **kwargs: Additional optional or keyward arguments (if any). - - Returns: - TestResult: Result of the hypothesis test. - """ - pass - - -SequentialTestBase = SequentialTwoSampleTestBase - - -class MirroredTestMixin: - """A mixin class to define mirrored hypothesis tests. - - In our terminology, a mirrored test is one that runs two one-sided tests - simultaneously, with the null and the alternaive flipped from each other. This is so - that it can yield either Decision.AcceptNull or Decision.AcceptAlternative depending - on the input data, unlike standard one-sided tests that can never 'accept' the null. - (Those standard tests will at most fail to reject the null, as represented by - Decision.FailToDecide.) - - For example, if the alternative is Hypothesis.P0MoreThanP1 and the decision is - Decision.AcceptNull, it should be interpreted as accepting Hypothesis.P0LessThanP1. - - The significance level alpha controls the following two errors simultaneously: (1) - probability of wrongly accepting the alternative when the null is true, and (2) - probability of wrongly accepting the null when the alternative is true. Note that - Bonferroni correction is not needed since the null hypothesis for one test is the - alternative for the other. - - Attributes: - It has the same attributes as the underlying base tests. - """ - - _base_class: Type[Union[TestBase, SequentialTestBase]] = ( - None # To be set by subclasses. - ) - - def __init__(self, alternative: Hypothesis, *args, **kwargs) -> None: - """Initializes the MirroredTestMixin object. - - It instantiates two private member variables, _test_for_alternative and - _test_for_null, with flipped alternative hypotheses. - - Args: - alternative: Specification of the alternative hypothesis. - *args: Additional positional arguments (if any). - **kwargs: Additional optional or keyward arguments (if any). - - Raises: - AttributeError: If alternative is not a valid Hypothesis for a mirrored test. - """ - if alternative == Hypothesis.P0MoreThanP1: - null = Hypothesis.P0LessThanP1 - elif alternative == Hypothesis.P0LessThanP1: - null = Hypothesis.P0MoreThanP1 - else: - raise ( - AttributeError(f"{alternative} is not a valid value for alternative.") - ) - - self._test_for_alternative = self._base_class(alternative, *args, **kwargs) - self._test_for_null = self._base_class(null, *args, **kwargs) - - def __getattr__(self, name: str) -> Any: - """Dynamically forward attributes from the underlying base tests. - - Args: - name: Name of the attribute. - - Raises: - AttributeError: If the attribute does not exist in the base test class. - """ - # Dynamically forward attributes from the base test. - if hasattr(self._test_for_alternative, name): - return getattr(self._test_for_alternative, name) - else: - raise AttributeError( - f"'{self.__class__.__name__}' object has no attribute '{name}'" - ) - - def __setattr__(self, name: str, value: Any) -> None: - """Dynamically set attributes to the underlying base tests. - - Args: - name: Name of the attribute. - value: Value of the attribute. - - Raises: - AttributeError: If the attribute assignment to the base tests fails. - """ - if name == "alternative": - # For alternative, make sure _test_for_alternative and _test_for_null has - # flipped alternative hypothesis. - if value == Hypothesis.P0MoreThanP1: - null = Hypothesis.P0LessThanP1 - elif value == Hypothesis.P0LessThanP1: - null = Hypothesis.P0MoreThanP1 - else: - raise (AttributeError(f"{value} is not a valid value for alternative.")) - self._test_for_alternative.alternative = value - self._test_for_null.alternative = null - else: - # If setting attributes after initialization, set on both - # _test_for_alternative and _test_for_null. - if name not in ["_test_for_alternative", "_test_for_null", "_base_class"]: - if hasattr(self._test_for_alternative, name): - setattr(self._test_for_alternative, name, value) - setattr(self._test_for_null, name, value) - else: - raise AttributeError( - f"'{self.__class__.__name__}' object has no attribute '{name}'" - ) - else: - super().__setattr__(name, value) +from statistical_comparison_core.base import ( + Decision, + Hypothesis, + MirroredTestMixin, + SequentialTestBase, + SequentialTwoSampleTestBase, + TestBase, + TestResult, + TwoSampleBinomialHypothesis, + TwoSampleMeanHypothesis, + TwoSampleTestBase, +) + +__all__ = [ + "Decision", + "Hypothesis", + "MirroredTestMixin", + "SequentialTestBase", + "SequentialTwoSampleTestBase", + "TestBase", + "TestResult", + "TwoSampleBinomialHypothesis", + "TwoSampleMeanHypothesis", + "TwoSampleTestBase", +] diff --git a/sequentialized_barnard_tests/tools/plotting.py b/sequentialized_barnard_tests/tools/plotting.py index fd30d87..187900c 100644 --- a/sequentialized_barnard_tests/tools/plotting.py +++ b/sequentialized_barnard_tests/tools/plotting.py @@ -1,5 +1,7 @@ from typing import Dict, List, Optional, Tuple, Union +import warnings + from matplotlib.cm import get_cmap import matplotlib.pyplot as plt import numpy as np @@ -8,93 +10,26 @@ from sequentialized_barnard_tests import Decision, Hypothesis from sequentialized_barnard_tests.step import MirroredStepTest +import statistical_comparison_helpers as _sch + def compact_letter_display( significant_pair_list: List[Tuple[str, str]], sorted_model_list: List[str], ) -> List[str]: """Generates Compact Letter Display (CLD) given a list of significant - pairs and a list of models. CLD is Based on "An Algorithm for a - Letter-Based Representation of All-Pairwise Comparisons" by Piepho - (2004). + pairs and a list of models. - Args: - significant_pair_list: A list containing tuples of model names that - were deemed significantly different by each A/B test. - sorted_model_list: A list of model names sorted by performance in - descending order. - - Returns: - A list of letters representing CLD for the corresponding models. + .. deprecated:: + Use ``statistical_comparison_helpers.compact_letter_display`` instead. """ - num_models = len(sorted_model_list) - - # Map model names to indices. - model_to_index = {model: idx for idx, model in enumerate(sorted_model_list)} - # Convert significant pairs from names to indices. - significant_index_pairs = [ - (model_to_index[m1], model_to_index[m2]) for m1, m2 in significant_pair_list - ] - - # --- Inner helper to remove redundant columns --- - def remove_redundant_columns(matrix): - changed = True - while changed: - changed = False - for i in range(len(matrix)): - for j in range(len(matrix)): - if i != j: - indices_i = {idx for idx, char in enumerate(matrix[i]) if char} - indices_j = {idx for idx, char in enumerate(matrix[j]) if char} - if indices_i.issubset(indices_j): - matrix.pop(i) - changed = True - break - if changed: - break - return matrix - - # --- Main algorithm --- - # Start with a single column of 'a's for all models. - letter_matrix = [["a"] * num_models] - - # For each significant pair, update the letter matrix. - for model_idx1, model_idx2 in significant_index_pairs: - while any(col[model_idx1] and col[model_idx2] for col in letter_matrix): - for col_index, letter_column in enumerate(letter_matrix): - if letter_column[model_idx1] and letter_column[model_idx2]: - new_column = letter_column.copy() - new_column[model_idx1] = "" - letter_column[model_idx2] = "" - letter_matrix[col_index] = letter_column - letter_matrix.append(new_column) - letter_matrix = remove_redundant_columns(letter_matrix) - break # re-check with the while condition - - # --- Reassign letters based on sorted columns --- - def first_nonempty_position(column): - for pos, char in enumerate(column): - if char: - return pos - return len(column) - - letter_matrix.sort(key=first_nonempty_position) - - for idx, column in enumerate(letter_matrix): - replacement_letter = chr(ord("a") + idx) - letter_matrix[idx] = [replacement_letter if char else "" for char in column] - - # --- Build final CLD output for each model --- - final_display = [] - for model_idx in range(num_models): - letters = "".join( - letter_matrix[col_idx][model_idx] - for col_idx in range(len(letter_matrix)) - if letter_matrix[col_idx][model_idx] - ) - final_display.append(letters) - - return final_display + warnings.warn( + "sequentialized_barnard_tests.tools.plotting.compact_letter_display is " + "deprecated. Use statistical_comparison_helpers.compact_letter_display instead.", + DeprecationWarning, + stacklevel=2, + ) + return _sch.compact_letter_display(significant_pair_list, sorted_model_list) def compare_success_and_get_cld( @@ -187,7 +122,7 @@ def compare_success_and_get_cld( reverse=True, ) ] - letters_list = compact_letter_display( + letters_list = _sch.compact_letter_display( input_list_to_cld, models_sorted_by_success_rates ) if verbose: @@ -243,25 +178,20 @@ def draw_samples_from_beta_posterior( beta_prior: float = 1, ) -> np.ndarray: """Draw samples from the beta posterior distribution given a success array. - These samples can be used to estimate the posterior distribution of the - success rate of a Bernoulli process. Note that the default prior parameters - of (1, 1) correspond to a uniform prior. - - Args: - success_array: A binary array with True/False indicating success/failure. - rng: A numpy random Generator instance. - num_samples: Optional number of samples to draw. Defaults to 10000. - alpha_prior: Optional alpha parameter of the beta prior. Defaults to 1. - beta_prior: Optional beta parameter of the beta prior. Defaults to 1. - Returns: - Samples drawn from the beta posterior distribution. + .. deprecated:: + Use ``statistical_comparison_helpers.draw_samples_from_beta_posterior`` instead. """ - n_trials = len(success_array) - n_successes = np.sum(success_array) - n_failures = n_trials - n_successes - posterior = stats.beta(alpha_prior + n_successes, beta_prior + n_failures) - return posterior.rvs(num_samples, random_state=rng) + warnings.warn( + "sequentialized_barnard_tests.tools.plotting.draw_samples_from_beta_posterior " + "is deprecated. Use statistical_comparison_helpers.draw_samples_from_beta_posterior instead.", + DeprecationWarning, + stacklevel=2, + ) + return _sch.draw_samples_from_beta_posterior( + success_array, rng, num_samples=num_samples, + alpha_prior=alpha_prior, beta_prior=beta_prior, + ) def plot_model_comparison( @@ -276,81 +206,31 @@ def plot_model_comparison( height: int = 4, dpi: int = 100, ) -> Union[None, plt.Figure]: - """Makes a violin plot of success rate estimates with corresponding CLD letters - for policy comparison. + """Makes a violin plot of success rate estimates with corresponding CLD letters. - Args: - model_name_list: A list of model names. - success_arrays: A list of arrays indicating success/failure for each model. - cld_letters: A list of CLD letters corresponding to each model. - rng: A numpy random Generator instance for posterior sampling. - output_path: Optional file path to save the plot. If None, the plot will not - be saved but returned as a matplotlib Figure object. Defaults to None. - title: Optional title for the plot. Defaults to None. - add_legend: Whether to show legend on the plot. Defaults to False. - unit_width: Figure width per model. Defaults to 6. - height: Figure height. Defaults to 4. - dpi: Resolution of the saved plot. Defaults to 100. - - Returns: - If output_path is None, returns a matplotlib Figure object containing - the plot. Otherwise, saves the plot to the specified path and returns None. + .. deprecated:: + Use ``statistical_comparison_helpers.plot_model_comparison`` instead. """ - num_models = len(model_name_list) - - posterior_samples = [] - means = [] - - for success_array in success_arrays: - samples = draw_samples_from_beta_posterior(success_array, rng) - posterior_samples.append(samples) - means.append(np.mean(samples)) - - fig, ax = plt.subplots(figsize=(max(unit_width, num_models), height), dpi=dpi) - - cmap = get_cmap("tab10") - colors = [cmap(i % 10) for i in range(num_models)] - - parts = ax.violinplot( - posterior_samples, - positions=np.arange(num_models), - showmeans=True, - showmedians=False, - showextrema=False, - widths=0.8, + warnings.warn( + "sequentialized_barnard_tests.tools.plotting.plot_model_comparison is " + "deprecated. Use statistical_comparison_helpers.plot_model_comparison instead.", + DeprecationWarning, + stacklevel=2, + ) + from statistical_comparison_helpers.plotting import plot_model_comparison as _plot + from statistical_comparison_helpers.posterior import Binary + + return _plot( + model_name_list, + success_arrays, + cld_letters, + rng, + score=Binary(), + plot_mode="posterior", + output_path=output_path, + title=title, + add_legend=add_legend, + unit_width=unit_width, + height=height, + dpi=dpi, ) - for pc, color in zip(parts["bodies"], colors): - pc.set_facecolor(color) - pc.set_alpha(0.6) - parts["cmeans"].set_color("black") - parts["cmeans"].set_linewidth(0.8) - - # Add CLD labels - for i, (x, y, label) in enumerate(zip(np.arange(num_models), means, cld_letters)): - ax.text( - x + 0.15, - y + 0.03, - label, - fontsize=12, - fontweight="bold", - color="black", - verticalalignment="center", - zorder=4, - ) - - ax.set_xticks(np.arange(num_models)) - ax.set_xticklabels(model_name_list, rotation=0, ha="center") - ax.set_ylim(0.0, 1.0) - ax.set_ylabel("Success Rate") - if title is not None: - ax.set_title(title) - if add_legend: - ax.legend(parts["bodies"], model_name_list, loc="best") - plt.tight_layout() - - if output_path is not None: - plt.savefig(output_path, dpi=300) - plt.close() - print(f"Saved a PNG plot to {output_path}") - else: - return fig diff --git a/setup.py b/setup.py index 2e4dd2c..5c372e1 100644 --- a/setup.py +++ b/setup.py @@ -2,11 +2,11 @@ setup( name="sequentialized_barnard_tests", - version="0.0.4", + version="0.0.6", description="Sequential statistical hypothesis testing for two-by-two contingency tables.", authors=["David Snyder", "Haruki Nishimura"], author_emails=["dasnyder@princeton.edu", "haruki.nishimura@tri.global"], - license="MIT", + license="CC-BY-NC-4.0", packages=find_packages(), package_data={ "sequentialized_barnard_tests": [ @@ -23,6 +23,8 @@ "numpy>=1.20", "pytest", "scipy", + "statistical-comparison-core", + "statistical-comparison-helpers", "tqdm", ], ) diff --git a/tests/sequentialized_barnard_tests/test_sbt_deprecation_shims.py b/tests/sequentialized_barnard_tests/test_sbt_deprecation_shims.py new file mode 100644 index 0000000..e61257d --- /dev/null +++ b/tests/sequentialized_barnard_tests/test_sbt_deprecation_shims.py @@ -0,0 +1,92 @@ +"""Tests for deprecation shims in sequentialized_barnard_tests.tools.plotting.""" + +import warnings + +import numpy as np + +import statistical_comparison_helpers as sch + + +class TestCLDShim: + def test_emits_deprecation_warning(self): + from sequentialized_barnard_tests.tools.plotting import compact_letter_display + + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + result = compact_letter_display( + [("A", "B")], ["A", "B", "C"] + ) + assert len(w) == 1 + assert issubclass(w[0].category, DeprecationWarning) + assert "deprecated" in str(w[0].message).lower() + + def test_output_matches_statistical_comparison_helpers(self): + from sequentialized_barnard_tests.tools.plotting import compact_letter_display + + pairs = [("A", "B"), ("B", "C")] + models = ["A", "B", "C", "D"] + + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + shim_result = compact_letter_display(pairs, models) + + expected = sch.compact_letter_display(pairs, models) + assert shim_result == expected + + def test_returns_list_of_str(self): + from sequentialized_barnard_tests.tools.plotting import compact_letter_display + + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + result = compact_letter_display([], ["A", "B"]) + + assert isinstance(result, list) + assert all(isinstance(x, str) for x in result) + + +class TestBetaPosteriorShim: + def test_emits_deprecation_warning(self): + from sequentialized_barnard_tests.tools.plotting import draw_samples_from_beta_posterior + + rng = np.random.default_rng(42) + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + draw_samples_from_beta_posterior(np.array([1, 0, 1]), rng, num_samples=100) + assert len(w) == 1 + assert issubclass(w[0].category, DeprecationWarning) + + def test_output_shape(self): + from sequentialized_barnard_tests.tools.plotting import draw_samples_from_beta_posterior + + rng = np.random.default_rng(42) + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + result = draw_samples_from_beta_posterior(np.array([1, 0, 1, 1]), rng, num_samples=500) + assert result.shape == (500,) + + +class TestPlotShim: + def test_emits_deprecation_warning(self): + import matplotlib + matplotlib.use("Agg") + + from sequentialized_barnard_tests.tools.plotting import plot_model_comparison + + rng = np.random.default_rng(42) + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("ignore") + warnings.simplefilter("always", DeprecationWarning) + fig = plot_model_comparison( + ["A", "B"], + [np.array([1, 0, 1]), np.array([0, 1, 0])], + ["a", "b"], + rng, + ) + assert any( + issubclass(x.category, DeprecationWarning) + and "sequentialized_barnard_tests.tools.plotting.plot_model_comparison" in str(x.message) + for x in w + ) + + import matplotlib.pyplot as plt + plt.close(fig)