from __future__ import annotations
import math
from typing import Callable, Tuple
import numpy as np
import pandas as pd
import plotly.graph_objects as go
import seaborn as sns
from scipy.stats import chi2_contingency, fisher_exact, ks_2samp, mannwhitneyu
from scipy.stats.contingency import crosstab
from statsmodels.stats.power import TTestIndPower
from statsmodels.stats.weightstats import ttest_ind, ztest
from retentioneering.backend.tracker import track
from retentioneering.eventstream.types import EventstreamType
from retentioneering.tooling.stattests.constants import STATTEST_NAMES
def _effect_size_cohend(sample1: list, sample2: list) -> float:
"""
Calculate an effect size value using Cohen’s D measure (difference between two means).
Parameters
----------
sample1 : list
First group's statistics.
sample2 : list
Second group's statistics.
Returns
-------
float
Calculated an effect size value.
"""
sample_size1, sample_size2 = len(sample1), len(sample2)
standard_deviation1, standard_deviation2 = np.var(sample1, ddof=1), np.var(sample2, ddof=1)
pooled_standard_deviation = float(
math.sqrt(
((sample_size1 - 1) * standard_deviation1 + (sample_size2 - 1) * standard_deviation2)
/ (sample_size1 + sample_size2 - 2)
)
)
mean_sample1, mean_sample2 = float(np.mean(sample1)), float(np.mean(sample2))
return (mean_sample1 - mean_sample2) / pooled_standard_deviation
def _effect_size_cohenh(sample1: list, sample2: list) -> float:
"""
Calculate an effect size value using Cohen’s D measure (between proportions)
Parameters
----------
sample1 : list
First group's statistics.
sample2 : list
Second group's statistics.
Returns
-------
float
Calculated an effect size value.
"""
mean_sample1, mean_sample2 = np.mean(sample1), np.mean(sample2)
return 2 * (math.asin(math.sqrt(mean_sample1)) - math.asin(math.sqrt(mean_sample2)))
[docs]class StatTests:
"""
A class for determining statistical difference between two groups of users.
Parameters
----------
eventstream : EventstreamType
See Also
--------
.Eventstream.stattests : Call StatTests tool as an eventstream method.
Notes
-----
See :doc:`StatTests user guide</user_guides/stattests>` for the details.
"""
__eventstream: EventstreamType
test: STATTEST_NAMES
groups: Tuple[list[str | int], list[str | int]]
func: Callable
group_names: Tuple[str, str]
alpha: float
g1_data: list[str | int]
g2_data: list[str | int]
is_fitted: bool
output_template_numerical = "{0} (mean ± SD): {1:.3f} ± {2:.3f}, n = {3}"
output_template_categorical = "{0} (size): n = {1}"
p_val, power, label_min, label_max = 0.0, 0.0, "", ""
@track( # type: ignore
tracking_info={"event_name": "init"},
scope="stattests",
allowed_params=[],
)
def __init__(self, eventstream: EventstreamType) -> None:
self.__eventstream = eventstream
self.user_col = self.__eventstream.schema.user_id
self.event_col = self.__eventstream.schema.event_name
self.time_col = self.__eventstream.schema.event_timestamp
self.g1_data = list()
self.g2_data = list()
self.is_fitted = False
def _get_group_values(self) -> Tuple[list, list]:
data = self.__eventstream.to_dataframe()
# obtain two populations for each group
g1 = data[data[self.user_col].isin(self.groups[0])].copy()
g2 = data[data[self.user_col].isin(self.groups[1])].copy()
# obtain two distributions:
if self.test not in ["chi2_contingency", "fisher_exact"]:
g1_data = list(g1.groupby(self.user_col).apply(self.func).dropna().astype(float).values)
g2_data = list(g2.groupby(self.user_col).apply(self.func).dropna().astype(float).values)
else:
g1_data = list(g1.groupby(self.user_col).apply(self.func).dropna().values)
g2_data = list(g2.groupby(self.user_col).apply(self.func).dropna().values)
return g1_data, g2_data
def _get_freq_table(self, a: list, b: list) -> list:
labels = ["A"] * len(a) + ["B"] * len(b)
values = np.concatenate([a, b])
if self.test == "fisher_exact" and np.unique(values).shape[0] != 2:
raise ValueError("For Fisher exact test, there should be exactly 2 categories of observations in the data")
return crosstab(labels, values)[1]
def _get_test_results(self, data_max: list, data_min: list) -> Tuple[float, float]:
if self.test in ["ztest", "ttest", "mannwhitneyu", "ks_2samp"]:
# calculate effect size
if max(data_max) <= 1 and min(data_max) >= 0 and max(data_min) <= 1 and min(data_min) >= 0:
# if analyze proportions use Cohen's h:
effect_size = _effect_size_cohenh(data_max, data_min)
else:
# for other variables use Cohen's d:
effect_size = _effect_size_cohend(data_max, data_min)
# calculate power
power = TTestIndPower().power(
effect_size=effect_size,
nobs1=len(data_max),
ratio=len(data_min) / len(data_max),
alpha=self.alpha,
alternative="larger",
)
if self.test == "ks_2samp":
p_val = ks_2samp(data_max, data_min, alternative="less")[1]
elif self.test == "mannwhitneyu":
p_val = mannwhitneyu(data_max, data_min, alternative="greater")[1]
elif self.test == "ttest":
p_val = ttest_ind(data_max, data_min, alternative="larger")[1]
elif self.test == "ztest":
p_val = ztest(data_max, data_min, alternative="larger")[1]
elif self.test in ["chi2_contingency", "fisher_exact"]:
power = None
if self.test == "chi2_contingency":
freq_table = self._get_freq_table(data_max, data_min)
p_val = chi2_contingency(freq_table)[1]
elif self.test == "fisher_exact":
freq_table = self._get_freq_table(data_max, data_min)
p_val = fisher_exact(freq_table, alternative="greater")[1]
else:
raise ValueError(f"The argument test is not supported. Supported tests are: {STATTEST_NAMES}")
return p_val, power # type: ignore
def _get_sorted_test_results(self) -> Tuple[float, float, str, str]:
p_val_norm, power_norm = self._get_test_results(self.g1_data, self.g2_data)
p_val_rev, power_rev = self._get_test_results(self.g2_data, self.g1_data)
if p_val_norm < p_val_rev:
p_val = p_val_norm
power = power_norm
label_max = self.group_names[0]
label_min = self.group_names[1]
else:
p_val = p_val_rev
power = power_rev
label_max = self.group_names[1]
label_min = self.group_names[0]
return p_val, power, label_max, label_min
[docs] @track( # type: ignore
tracking_info={"event_name": "fit"},
scope="stattests",
allowed_params=[
"test",
"groups",
"func",
"group_names",
"alpha",
],
)
def fit(
self,
test: STATTEST_NAMES,
groups: Tuple[list[str | int], list[str | int]],
func: Callable,
group_names: Tuple[str, str] = ("group_1", "group_2"),
alpha: float = 0.05,
) -> None:
"""
Calculates the stattests internal values with the defined parameters.
Applying ``fit`` method is necessary for the following usage
of any visualization or descriptive ``StatTests`` methods.
Parameters
----------
test : {'mannwhitneyu', 'ttest', 'ztest', 'ks_2samp', 'chi2_contingency', 'fisher_exact'}
Test the null hypothesis that 2 independent samples are drawn from the same
distribution. Supported tests are:
- ``mannwhitneyu`` see :mannwhitneyu:`scipy documentation<>`.
- ``ttest`` see :statsmodel_ttest:`statsmodels documentation<>`.
- ``ztest`` see :statsmodel_ztest:`statsmodels documentation<>`.
- ``ks_2samp`` see :scipy_ks:`scipy documentation<>`.
- ``chi2_contingency`` see :scipy_chi2:`scipy documentation<>`.
- ``fisher_exact`` see :scipy_fisher:`scipy documentation<>`.
groups : tuple of list
Must contain a tuple of two elements (g_1, g_2): g_1 and g_2 are collections
of user_id`s.
func : Callable
Selected metrics. Must contain a function that takes a dataset as an argument for
a single user trajectory and returns a single numerical value.
group_names : tuple, default ('group_1', 'group_2')
Names for selected groups g_1 and g_2.
alpha : float, default 0.05
Selected level of significance.
"""
self.groups = groups
self.func = func
self.test = test
self.group_names = group_names
self.alpha = alpha
self.g1_data, self.g2_data = self._get_group_values()
self.p_val, self.power, self.label_min, self.label_max = self._get_sorted_test_results()
self.is_fitted = True
[docs] @track( # type: ignore
tracking_info={"event_name": "plot"},
scope="stattests",
allowed_params=[],
)
def plot(self) -> Tuple[go.Figure, str]:
"""
Plots a barplot comparing the metric values between two groups.
Should be used after :py:func:`fit`.
Returns
-------
go.Figure
"""
data1 = pd.DataFrame(data={"data": self.g1_data, "groups": self.group_names[0]})
data2 = pd.DataFrame(data={"data": self.g2_data, "groups": self.group_names[1]})
combined_stats = pd.concat([data1, data2]).reset_index()
compare_plot = sns.displot(data=combined_stats, x="data", hue="groups", multiple="dodge")
compare_plot.set(xlabel=None)
return compare_plot
@property
@track( # type: ignore
tracking_info={"event_name": "values"},
scope="stattests",
allowed_params=[],
)
def values(self) -> dict:
"""
Returns the comprehensive results of the comparison between the two groups.
Should be used after :py:func:`fit`.
Returns
-------
dict
"""
if not self.is_fitted:
raise ValueError("The StatTests instance needs to be fitted before returning values")
if self.test in ["ztest", "ttest", "mannwhitneyu", "ks_2samp"]:
res_dict = {
"group_one_name": self.group_names[0],
"group_one_size": len(self.g1_data),
"group_one_mean": np.array(self.g1_data).mean(),
"group_one_SD": np.array(self.g1_data).std(),
"group_two_name": self.group_names[1],
"group_two_size": len(self.g2_data),
"group_two_mean": np.array(self.g2_data).mean(),
"group_two_SD": np.array(self.g2_data).std(),
"greatest_group_name": self.label_max,
"least_group_name": self.label_min,
"is_group_one_greatest": self.label_max == self.group_names[0],
"p_val": self.p_val,
"power_estimated": self.power,
}
elif self.test in ["chi2_contingency", "fisher_exact"]:
res_dict = {
"group_one_name": self.group_names[0],
"group_one_size": len(self.g1_data),
"group_two_name": self.group_names[1],
"group_two_size": len(self.g2_data),
"p_val": self.p_val,
}
else:
raise ValueError("Wrong test passed")
return res_dict
@track( # type: ignore
tracking_info={"event_name": "display_results"},
scope="stattests",
allowed_params=[],
)
def display_results(self) -> None:
if not self.is_fitted:
raise ValueError("The StatTests instance needs to be fitted before displaying results")
values = self.values
if self.test in ["ztest", "ttest", "mannwhitneyu", "ks_2samp"]:
print(
self.output_template_numerical.format(
values["group_one_name"], values["group_one_mean"], values["group_one_SD"], values["group_one_size"]
)
)
print(
self.output_template_numerical.format(
values["group_two_name"], values["group_two_mean"], values["group_two_SD"], values["group_two_size"]
)
)
print(
"'{0}' is greater than '{1}' with p-value: {2:.5f}".format(
values["greatest_group_name"], values["least_group_name"], values["p_val"]
)
)
print("power of the test: {0:.2f}%".format(100 * values["power_estimated"]))
elif self.test in ["chi2_contingency", "fisher_exact"]:
print(self.output_template_categorical.format(values["group_one_name"], values["group_one_size"]))
print(self.output_template_categorical.format(values["group_two_name"], values["group_two_size"]))
print("Group difference test with p-value: {:.5f}".format(values["p_val"]))
else:
raise ValueError("Wrong test passed")
@property
@track( # type: ignore
tracking_info={"event_name": "params"},
scope="stattests",
allowed_params=[],
)
def params(self) -> dict:
"""
Returns the parameters used for the last fitting.
Should be used after :py:func:`fit`.
"""
return {
"test": self.test,
"groups": self.groups,
"func": self.func,
"group_names": self.group_names,
"alpha": self.alpha,
}