from cluster_experiments.power_analysis import *

NormalPowerAnalysis

Class used to run Power analysis, using the central limit theorem to estimate power based on standard errors of the estimator, and the fact that the coefficients of a regression are normally distributed. It does so by running simulations. In each simulation: 1. Assign treatment to dataframe randomly 2. Add pre-experiment data if needed 3. Get standard error from analysis

Finally it returns the power of the analysis by counting how many times the effect was detected.

Parameters:

Name	Type	Description	Default
splitter	RandomSplitter	RandomSplitter class to randomly assign treatment to dataframe.	required
analysis	ExperimentAnalysis	ExperimentAnalysis class to use for analysis.	required
cupac_model	Optional[BaseEstimator]	Sklearn estimator class to add pre-experiment data to dataframe. If None, no pre-experiment data will be added.	None
target_col	str	Name of the column with the outcome variable.	'target'
treatment_col	str	Name of the column with the treatment variable.	'treatment'
treatment	str	value of treatment_col considered to be treatment (not control)	'B'
control	str	value of treatment_col considered to be control (not treatment)	'A'
n_simulations	int	Number of simulations to run.	100
alpha	float	Significance level.	0.05
features_cupac_model	Optional[List[str]]	Covariates to be used in cupac model	None
seed	Optional[int]	Optional. Seed to use for the splitter.	None

Usage:

from datetime import date

import numpy as np
import pandas as pd
from cluster_experiments.experiment_analysis import GeeExperimentAnalysis
from cluster_experiments.power_analysis import NormalPowerAnalysis
from cluster_experiments.random_splitter import ClusteredSplitter

N = 1_000
users = [f"User {i}" for i in range(1000)]
clusters = [f"Cluster {i}" for i in range(100)]
dates = [f"{date(2022, 1, i):%Y-%m-%d}" for i in range(1, 32)]
df = pd.DataFrame(
    {
        "cluster": np.random.choice(clusters, size=N),
        "target": np.random.normal(0, 1, size=N),
        "user": np.random.choice(users, size=N),
        "date": np.random.choice(dates, size=N),
    }
)

experiment_dates = [f"{date(2022, 1, i):%Y-%m-%d}" for i in range(15, 32)]
sw = ClusteredSplitter(
    cluster_cols=["cluster", "date"],
)

analysis = GeeExperimentAnalysis(
    cluster_cols=["cluster", "date"],
)

pw = NormalPowerAnalysis(
    splitter=sw, analysis=analysis, n_simulations=50
)

power = pw.power_analysis(df, average_effect=0.1)
print(f"{power = }")

Source code in cluster_experiments/power_analysis.py

class NormalPowerAnalysis:
    """
    Class used to run Power analysis, using the central limit theorem to estimate power based on standard errors of the estimator,
    and the fact that the coefficients of a regression are normally distributed.
    It does so by running simulations. In each simulation:
    1. Assign treatment to dataframe randomly
    2. Add pre-experiment data if needed
    3. Get standard error from analysis

    Finally it returns the power of the analysis by counting how many times the effect was detected.

    Args:
        splitter: RandomSplitter class to randomly assign treatment to dataframe.
        analysis: ExperimentAnalysis class to use for analysis.
        cupac_model: Sklearn estimator class to add pre-experiment data to dataframe. If None, no pre-experiment data will be added.
        target_col: Name of the column with the outcome variable.
        treatment_col: Name of the column with the treatment variable.
        treatment: value of treatment_col considered to be treatment (not control)
        control: value of treatment_col considered to be control (not treatment)
        n_simulations: Number of simulations to run.
        alpha: Significance level.
        features_cupac_model: Covariates to be used in cupac model
        seed: Optional. Seed to use for the splitter.

    Usage:
    ```python
    from datetime import date

    import numpy as np
    import pandas as pd
    from cluster_experiments.experiment_analysis import GeeExperimentAnalysis
    from cluster_experiments.power_analysis import NormalPowerAnalysis
    from cluster_experiments.random_splitter import ClusteredSplitter

    N = 1_000
    users = [f"User {i}" for i in range(1000)]
    clusters = [f"Cluster {i}" for i in range(100)]
    dates = [f"{date(2022, 1, i):%Y-%m-%d}" for i in range(1, 32)]
    df = pd.DataFrame(
        {
            "cluster": np.random.choice(clusters, size=N),
            "target": np.random.normal(0, 1, size=N),
            "user": np.random.choice(users, size=N),
            "date": np.random.choice(dates, size=N),
        }
    )

    experiment_dates = [f"{date(2022, 1, i):%Y-%m-%d}" for i in range(15, 32)]
    sw = ClusteredSplitter(
        cluster_cols=["cluster", "date"],
    )

    analysis = GeeExperimentAnalysis(
        cluster_cols=["cluster", "date"],
    )

    pw = NormalPowerAnalysis(
        splitter=sw, analysis=analysis, n_simulations=50
    )

    power = pw.power_analysis(df, average_effect=0.1)
    print(f"{power = }")
    ```
    """

    def __init__(
        self,
        splitter: RandomSplitter,
        analysis: ExperimentAnalysis,
        cupac_model: Optional[BaseEstimator] = None,
        target_col: str = "target",
        treatment_col: str = "treatment",
        treatment: str = "B",
        control: str = "A",
        n_simulations: int = 100,
        alpha: float = 0.05,
        features_cupac_model: Optional[List[str]] = None,
        seed: Optional[int] = None,
        hypothesis: str = "two-sided",
        time_col: Optional[str] = None,
    ):
        self.splitter = splitter
        self.analysis = analysis
        self.n_simulations = n_simulations
        self.target_col = target_col
        self.treatment = treatment
        self.control = control
        self.treatment_col = treatment_col
        self.alpha = alpha
        self.hypothesis = hypothesis
        self.time_col = time_col

        self.cupac_handler = CupacHandler(
            cupac_model=cupac_model,
            target_col=target_col,
            features_cupac_model=features_cupac_model,
        )
        if seed is not None:
            random.seed(seed)  # seed for splitter
            np.random.seed(seed)  # numpy seed
            # may need to seed other stochasticity sources if added

        self.check_inputs()

    def _split(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Split dataframe.
        Args:
            df: Dataframe with outcome variable
        """
        treatment_df = self.splitter.assign_treatment_df(df)
        self.log_nulls(treatment_df)
        treatment_df = treatment_df.query(
            f"{self.treatment_col}.notnull()", engine="python"
        ).query(
            f"{self.treatment_col}.isin(['{self.treatment}', '{self.control}'])",
            engine="python",
        )
        return treatment_df

    def _get_standard_error(
        self,
        df: pd.DataFrame,
        n_simulations: int,
        verbose: bool,
    ) -> Generator[float, None, None]:
        for _ in tqdm(range(n_simulations), disable=not verbose):
            split_df = self._split(df)
            yield self.analysis.get_standard_error(split_df)

    def _normal_power_calculation(
        self, alpha: float, std_error: float, average_effect: float
    ) -> float:
        """Returns the power of the analysis using the normal distribution.
        Arguments:
            alpha: significance level
            std_error: standard error of the analysis
            average_effect: effect size of the analysis
        """
        if HypothesisEntries(self.analysis.hypothesis) == HypothesisEntries.LESS:
            z_alpha = norm.ppf(alpha)
            return float(norm.cdf(z_alpha - average_effect / std_error))

        if HypothesisEntries(self.analysis.hypothesis) == HypothesisEntries.GREATER:
            z_alpha = norm.ppf(1 - alpha)
            return 1 - float(norm.cdf(z_alpha - average_effect / std_error))

        if HypothesisEntries(self.analysis.hypothesis) == HypothesisEntries.TWO_SIDED:
            z_alpha = norm.ppf(1 - alpha / 2)
            norm_cdf_right = norm.cdf(z_alpha - average_effect / std_error)
            norm_cdf_left = norm.cdf(-z_alpha - average_effect / std_error)
            return float(norm_cdf_left + (1 - norm_cdf_right))

        raise ValueError(f"{self.analysis.hypothesis} is not a valid HypothesisEntries")

    def _normal_mde_calculation(
        self, alpha: float, std_error: float, power: float
    ) -> float:
        """
        Returns the minimum detectable effect of the analysis using the normal distribution.
        Args:
            alpha: Significance level.
            std_error: Standard error of the analysis.
            power: Power of the analysis.
        """
        if HypothesisEntries(self.analysis.hypothesis) == HypothesisEntries.LESS:
            z_alpha = norm.ppf(alpha)
            z_beta = norm.ppf(1 - power)
        elif HypothesisEntries(self.analysis.hypothesis) == HypothesisEntries.GREATER:
            z_alpha = norm.ppf(1 - alpha)
            z_beta = norm.ppf(power)
        elif HypothesisEntries(self.analysis.hypothesis) == HypothesisEntries.TWO_SIDED:
            # we are neglecting norm_cdf_left
            z_alpha = norm.ppf(1 - alpha / 2)
            z_beta = norm.ppf(power)
        else:
            raise ValueError(
                f"{self.analysis.hypothesis} is not a valid HypothesisEntries"
            )

        return float(z_alpha + z_beta) * std_error

    def mde_power_line(
        self,
        df: pd.DataFrame,
        pre_experiment_df: Optional[pd.DataFrame] = None,
        verbose: bool = False,
        powers: Iterable[float] = (),
        n_simulations: Optional[int] = None,
        alpha: Optional[float] = None,
    ) -> Dict[float, float]:
        """
        Returns the minimum detectable effect of the analysis.

        Args:
            df: Dataframe with outcome and treatment variables.
            pre_experiment_df: Dataframe with pre-experiment data.
            verbose: Whether to show progress bar.
            power: Power of the analysis.
            n_simulations: Number of simulations to run.
            alpha: Significance level.
        """
        alpha = self.alpha if alpha is None else alpha
        std_error = self._get_average_standard_error(
            df=df,
            pre_experiment_df=pre_experiment_df,
            verbose=verbose,
            n_simulations=n_simulations,
        )
        return {
            power: self._normal_mde_calculation(
                alpha=alpha, std_error=std_error, power=power
            )
            for power in powers
        }

    def mde(
        self,
        df: pd.DataFrame,
        pre_experiment_df: Optional[pd.DataFrame] = None,
        verbose: bool = False,
        power: float = 0.8,
        n_simulations: Optional[int] = None,
        alpha: Optional[float] = None,
    ) -> float:
        """
        Returns the minimum detectable effect of the analysis.

        Args:
            df: Dataframe with outcome and treatment variables.
            pre_experiment_df: Dataframe with pre-experiment data.
            verbose: Whether to show progress bar.
            power: Power of the analysis.
            n_simulations: Number of simulations to run.
            alpha: Significance level.
        """
        return self.mde_power_line(
            df=df,
            pre_experiment_df=pre_experiment_df,
            verbose=verbose,
            powers=[power],
            n_simulations=n_simulations,
            alpha=alpha,
        )[power]

    def _get_average_standard_error(
        self,
        df: pd.DataFrame,
        pre_experiment_df: Optional[pd.DataFrame] = None,
        verbose: bool = False,
        n_simulations: Optional[int] = None,
    ) -> float:
        """
        Gets standard error to be used in normal power calculation.

        Args:
            df: Dataframe with outcome and treatment variables.
            pre_experiment_df: Dataframe with pre-experiment data.
            verbose: Whether to show progress bar.
            average_effects: Average effects to test.
            n_simulations: Number of simulations to run.
            alpha: Significance level.
        """
        n_simulations = self.n_simulations if n_simulations is None else n_simulations

        df = df.copy()
        df = self.cupac_handler.add_covariates(df, pre_experiment_df)

        std_errors = list(self._get_standard_error(df, n_simulations, verbose))
        std_error_mean = float(np.mean(std_errors))

        return std_error_mean

    def run_average_standard_error(
        self,
        df: pd.DataFrame,
        pre_experiment_df: Optional[pd.DataFrame] = None,
        verbose: bool = False,
        n_simulations: Optional[int] = None,
        experiment_length: Iterable[int] = (),
    ) -> Generator[Tuple[float, int], None, None]:
        """
        Run power analysis by simulation, using standard errors from the analysis.

        Args:
            df: Dataframe with outcome and treatment variables.
            pre_experiment_df: Dataframe with pre-experiment data.
            verbose: Whether to show progress bar.
            n_simulations: Number of simulations to run.
            experiment_length: Length of the experiment in days.
        """
        n_simulations = self.n_simulations if n_simulations is None else n_simulations

        for n_days in experiment_length:
            df_time = df.copy()
            experiment_start = df_time[self.time_col].min()
            df_time = df_time.loc[
                df_time[self.time_col] < experiment_start + pd.Timedelta(days=n_days)
            ]
            std_error_mean = self._get_average_standard_error(
                df=df_time,
                pre_experiment_df=pre_experiment_df,
                verbose=verbose,
                n_simulations=n_simulations,
            )
            yield std_error_mean, n_days

    def power_time_line(
        self,
        df: pd.DataFrame,
        pre_experiment_df: Optional[pd.DataFrame] = None,
        verbose: bool = False,
        average_effects: Iterable[float] = (),
        experiment_length: Iterable[int] = (),
        n_simulations: Optional[int] = None,
        alpha: Optional[float] = None,
    ) -> List[Dict]:
        """
        Run power analysis by simulation, using standard errors from the analysis.

        Args:
            df: Dataframe with outcome and treatment variables.
            pre_experiment_df: Dataframe with pre-experiment data.
            verbose: Whether to show progress bar.
            average_effects: Average effects to test.
            experiment_length: Length of the experiment in days.
            n_simulations: Number of simulations to run.
            alpha: Significance level.
        """
        alpha = self.alpha if alpha is None else alpha

        results = []
        for std_error_mean, n_days in self.run_average_standard_error(
            df=df,
            pre_experiment_df=pre_experiment_df,
            verbose=verbose,
            n_simulations=n_simulations,
            experiment_length=experiment_length,
        ):
            for effect in average_effects:
                power = self._normal_power_calculation(
                    alpha=alpha, std_error=std_error_mean, average_effect=effect
                )
                results.append(
                    {"effect": effect, "power": power, "experiment_length": n_days}
                )

        return results

    def mde_time_line(
        self,
        df: pd.DataFrame,
        pre_experiment_df: Optional[pd.DataFrame] = None,
        verbose: bool = False,
        powers: Iterable[float] = (),
        experiment_length: Iterable[int] = (),
        n_simulations: Optional[int] = None,
        alpha: Optional[float] = None,
    ) -> List[Dict]:
        alpha = self.alpha if alpha is None else alpha

        results = []
        for std_error_mean, n_days in self.run_average_standard_error(
            df=df,
            pre_experiment_df=pre_experiment_df,
            verbose=verbose,
            n_simulations=n_simulations,
            experiment_length=experiment_length,
        ):
            for power in powers:
                mde = self._normal_mde_calculation(
                    alpha=alpha, std_error=std_error_mean, power=power
                )
                results.append(
                    {"power": power, "mde": mde, "experiment_length": n_days}
                )
        return results

    def power_line(
        self,
        df: pd.DataFrame,
        pre_experiment_df: Optional[pd.DataFrame] = None,
        verbose: bool = False,
        average_effects: Iterable[float] = (),
        n_simulations: Optional[int] = None,
        alpha: Optional[float] = None,
    ) -> Dict[float, float]:
        """
        Run power analysis by simulation, using standard errors from the analysis.
        Args:
            df: Dataframe with outcome and treatment variables.
            pre_experiment_df: Dataframe with pre-experiment data.
            verbose: Whether to show progress bar.
            average_effects: Average effects to test.
            n_simulations: Number of simulations to run.
            alpha: Significance level.
        """
        alpha = self.alpha if alpha is None else alpha

        std_error_mean = self._get_average_standard_error(
            df=df,
            pre_experiment_df=pre_experiment_df,
            verbose=verbose,
            n_simulations=n_simulations,
        )

        return {
            effect: self._normal_power_calculation(
                alpha=alpha, std_error=std_error_mean, average_effect=effect
            )
            for effect in average_effects
        }

    def power_analysis(
        self,
        df: pd.DataFrame,
        pre_experiment_df: Optional[pd.DataFrame] = None,
        verbose: bool = False,
        average_effect: float = 0.0,
        n_simulations: Optional[int] = None,
        alpha: Optional[float] = None,
    ) -> float:
        """
        Run power analysis by simulation, using standard errors from the analysis.
        Args:
            df: Dataframe with outcome and treatment variables.
            pre_experiment_df: Dataframe with pre-experiment data.
            verbose: Whether to show progress bar.
            average_effect: Average effect of treatment.
            n_simulations: Number of simulations to run.
            alpha: Significance level.
        """
        return self.power_line(
            df=df,
            pre_experiment_df=pre_experiment_df,
            verbose=verbose,
            average_effects=[average_effect],
            n_simulations=n_simulations,
            alpha=alpha,
        )[average_effect]

    def log_nulls(self, df: pd.DataFrame) -> None:
        """Warns about dropping nulls in treatment column"""
        n_nulls = len(df.query(f"{self.treatment_col}.isnull()", engine="python"))
        if n_nulls > 0:
            logging.warning(
                f"There are {n_nulls} null values in treatment, dropping them"
            )

    @classmethod
    def from_dict(cls, config_dict: dict) -> "NormalPowerAnalysis":
        """Constructs PowerAnalysis from dictionary"""
        config = PowerConfig(**config_dict)
        return cls.from_config(config)

    @classmethod
    def from_config(cls, config: PowerConfig) -> "NormalPowerAnalysis":
        """Constructs PowerAnalysis from PowerConfig"""
        splitter_cls = _get_mapping_key(splitter_mapping, config.splitter)
        analysis_cls = _get_mapping_key(analysis_mapping, config.analysis)
        cupac_cls = _get_mapping_key(cupac_model_mapping, config.cupac_model)
        return cls(
            splitter=splitter_cls.from_config(config),
            analysis=analysis_cls.from_config(config),
            cupac_model=cupac_cls.from_config(config),
            target_col=config.target_col,
            treatment_col=config.treatment_col,
            treatment=config.treatment,
            control=config.control,
            n_simulations=config.n_simulations,
            alpha=config.alpha,
            features_cupac_model=config.features_cupac_model,
            seed=config.seed,
            hypothesis=config.hypothesis,
            time_col=config.time_col,
        )

    def check_treatment_col(self):
        """Checks consistency of treatment column"""
        assert (
            self.analysis.treatment_col == self.treatment_col
        ), f"treatment_col in analysis ({self.analysis.treatment_col}) must be the same as treatment_col in PowerAnalysis ({self.treatment_col})"

        assert (
            self.analysis.treatment_col == self.splitter.treatment_col
        ), f"treatment_col in analysis ({self.analysis.treatment_col}) must be the same as treatment_col in splitter ({self.splitter.treatment_col})"

    def check_target_col(self):
        assert (
            self.analysis.target_col == self.target_col
        ), f"target_col in analysis ({self.analysis.target_col}) must be the same as target_col in PowerAnalysis ({self.target_col})"

    def check_treatment(self):
        assert (
            self.analysis.treatment == self.treatment
        ), f"treatment in analysis ({self.analysis.treatment}) must be the same as treatment in PowerAnalysis ({self.treatment})"

        assert (
            self.analysis.treatment in self.splitter.treatments
        ), f"treatment in analysis ({self.analysis.treatment}) must be in treatments in splitter ({self.splitter.treatments})"

        assert (
            self.control in self.splitter.treatments
        ), f"control in power analysis ({self.control}) must be in treatments in splitter ({self.splitter.treatments})"

    def check_covariates(self):
        if hasattr(self.analysis, "covariates"):
            cupac_in_covariates = (
                self.cupac_handler.cupac_outcome_name in self.analysis.covariates
            )

            assert cupac_in_covariates or not self.cupac_handler.is_cupac, (
                f"covariates in analysis must contain {self.cupac_handler.cupac_outcome_name} if cupac_model is not None. "
                f"If you want to use cupac_model, you must add the cupac outcome to the covariates of the analysis "
                f"You may want to do covariates=['{self.cupac_handler.cupac_outcome_name}'] in your analysis method or your config"
            )

            if hasattr(self.splitter, "cluster_cols"):
                if set(self.analysis.covariates).intersection(
                    set(self.splitter.cluster_cols)
                ):
                    logging.warning(
                        f"covariates in analysis ({self.analysis.covariates}) are also cluster_cols in splitter ({self.splitter.cluster_cols}). "
                        f"Be specially careful when using switchback splitters, since the time splitter column is being overriden"
                    )

    def check_clusters(self):
        has_analysis_clusters = hasattr(self.analysis, "cluster_cols")
        has_splitter_clusters = hasattr(self.splitter, "cluster_cols")
        not_cluster_cols_cond = not has_analysis_clusters or not has_splitter_clusters
        assert (
            not_cluster_cols_cond
            or self.analysis.cluster_cols == self.splitter.cluster_cols
        ), f"cluster_cols in analysis ({self.analysis.cluster_cols}) must be the same as cluster_cols in splitter ({self.splitter.cluster_cols})"

        assert (
            has_splitter_clusters
            or not has_analysis_clusters
            or not self.analysis.cluster_cols
            or isinstance(self.splitter, RepeatedSampler)
        ), "analysis has cluster_cols but splitter does not."

        assert (
            has_analysis_clusters
            or not has_splitter_clusters
            or not self.splitter.cluster_cols
        ), "splitter has cluster_cols but analysis does not."

        has_time_col = hasattr(self.splitter, "time_col")
        assert not (
            has_time_col
            and has_splitter_clusters
            and self.splitter.time_col not in self.splitter.cluster_cols
        ), "in switchback splitters, time_col must be in cluster_cols"

    def check_inputs(self):
        self.check_covariates()
        self.check_treatment_col()
        self.check_target_col()
        self.check_treatment()
        self.check_clusters()

check_treatment_col()

Checks consistency of treatment column

Source code in cluster_experiments/power_analysis.py

def check_treatment_col(self):
    """Checks consistency of treatment column"""
    assert (
        self.analysis.treatment_col == self.treatment_col
    ), f"treatment_col in analysis ({self.analysis.treatment_col}) must be the same as treatment_col in PowerAnalysis ({self.treatment_col})"

    assert (
        self.analysis.treatment_col == self.splitter.treatment_col
    ), f"treatment_col in analysis ({self.analysis.treatment_col}) must be the same as treatment_col in splitter ({self.splitter.treatment_col})"

from_config(config) classmethod

Constructs PowerAnalysis from PowerConfig

Source code in cluster_experiments/power_analysis.py

@classmethod
def from_config(cls, config: PowerConfig) -> "NormalPowerAnalysis":
    """Constructs PowerAnalysis from PowerConfig"""
    splitter_cls = _get_mapping_key(splitter_mapping, config.splitter)
    analysis_cls = _get_mapping_key(analysis_mapping, config.analysis)
    cupac_cls = _get_mapping_key(cupac_model_mapping, config.cupac_model)
    return cls(
        splitter=splitter_cls.from_config(config),
        analysis=analysis_cls.from_config(config),
        cupac_model=cupac_cls.from_config(config),
        target_col=config.target_col,
        treatment_col=config.treatment_col,
        treatment=config.treatment,
        control=config.control,
        n_simulations=config.n_simulations,
        alpha=config.alpha,
        features_cupac_model=config.features_cupac_model,
        seed=config.seed,
        hypothesis=config.hypothesis,
        time_col=config.time_col,
    )

from_dict(config_dict) classmethod

Constructs PowerAnalysis from dictionary

Source code in cluster_experiments/power_analysis.py

@classmethod
def from_dict(cls, config_dict: dict) -> "NormalPowerAnalysis":
    """Constructs PowerAnalysis from dictionary"""
    config = PowerConfig(**config_dict)
    return cls.from_config(config)

log_nulls(df)

Warns about dropping nulls in treatment column

Source code in cluster_experiments/power_analysis.py

def log_nulls(self, df: pd.DataFrame) -> None:
    """Warns about dropping nulls in treatment column"""
    n_nulls = len(df.query(f"{self.treatment_col}.isnull()", engine="python"))
    if n_nulls > 0:
        logging.warning(
            f"There are {n_nulls} null values in treatment, dropping them"
        )

mde(df, pre_experiment_df=None, verbose=False, power=0.8, n_simulations=None, alpha=None)

Returns the minimum detectable effect of the analysis.

Parameters:

Name	Type	Description	Default
df	DataFrame	Dataframe with outcome and treatment variables.	required
pre_experiment_df	Optional[DataFrame]	Dataframe with pre-experiment data.	None
verbose	bool	Whether to show progress bar.	False
power	float	Power of the analysis.	0.8
n_simulations	Optional[int]	Number of simulations to run.	None
alpha	Optional[float]	Significance level.	None

Source code in cluster_experiments/power_analysis.py

def mde(
    self,
    df: pd.DataFrame,
    pre_experiment_df: Optional[pd.DataFrame] = None,
    verbose: bool = False,
    power: float = 0.8,
    n_simulations: Optional[int] = None,
    alpha: Optional[float] = None,
) -> float:
    """
    Returns the minimum detectable effect of the analysis.

    Args:
        df: Dataframe with outcome and treatment variables.
        pre_experiment_df: Dataframe with pre-experiment data.
        verbose: Whether to show progress bar.
        power: Power of the analysis.
        n_simulations: Number of simulations to run.
        alpha: Significance level.
    """
    return self.mde_power_line(
        df=df,
        pre_experiment_df=pre_experiment_df,
        verbose=verbose,
        powers=[power],
        n_simulations=n_simulations,
        alpha=alpha,
    )[power]

mde_power_line(df, pre_experiment_df=None, verbose=False, powers=(), n_simulations=None, alpha=None)

Returns the minimum detectable effect of the analysis.

Parameters:

Name	Type	Description	Default
df	DataFrame	Dataframe with outcome and treatment variables.	required
pre_experiment_df	Optional[DataFrame]	Dataframe with pre-experiment data.	None
verbose	bool	Whether to show progress bar.	False
power		Power of the analysis.	required
n_simulations	Optional[int]	Number of simulations to run.	None
alpha	Optional[float]	Significance level.	None

Source code in cluster_experiments/power_analysis.py

def mde_power_line(
    self,
    df: pd.DataFrame,
    pre_experiment_df: Optional[pd.DataFrame] = None,
    verbose: bool = False,
    powers: Iterable[float] = (),
    n_simulations: Optional[int] = None,
    alpha: Optional[float] = None,
) -> Dict[float, float]:
    """
    Returns the minimum detectable effect of the analysis.

    Args:
        df: Dataframe with outcome and treatment variables.
        pre_experiment_df: Dataframe with pre-experiment data.
        verbose: Whether to show progress bar.
        power: Power of the analysis.
        n_simulations: Number of simulations to run.
        alpha: Significance level.
    """
    alpha = self.alpha if alpha is None else alpha
    std_error = self._get_average_standard_error(
        df=df,
        pre_experiment_df=pre_experiment_df,
        verbose=verbose,
        n_simulations=n_simulations,
    )
    return {
        power: self._normal_mde_calculation(
            alpha=alpha, std_error=std_error, power=power
        )
        for power in powers
    }

power_analysis(df, pre_experiment_df=None, verbose=False, average_effect=0.0, n_simulations=None, alpha=None)

Run power analysis by simulation, using standard errors from the analysis. Args: df: Dataframe with outcome and treatment variables. pre_experiment_df: Dataframe with pre-experiment data. verbose: Whether to show progress bar. average_effect: Average effect of treatment. n_simulations: Number of simulations to run. alpha: Significance level.

Source code in cluster_experiments/power_analysis.py

def power_analysis(
    self,
    df: pd.DataFrame,
    pre_experiment_df: Optional[pd.DataFrame] = None,
    verbose: bool = False,
    average_effect: float = 0.0,
    n_simulations: Optional[int] = None,
    alpha: Optional[float] = None,
) -> float:
    """
    Run power analysis by simulation, using standard errors from the analysis.
    Args:
        df: Dataframe with outcome and treatment variables.
        pre_experiment_df: Dataframe with pre-experiment data.
        verbose: Whether to show progress bar.
        average_effect: Average effect of treatment.
        n_simulations: Number of simulations to run.
        alpha: Significance level.
    """
    return self.power_line(
        df=df,
        pre_experiment_df=pre_experiment_df,
        verbose=verbose,
        average_effects=[average_effect],
        n_simulations=n_simulations,
        alpha=alpha,
    )[average_effect]

power_line(df, pre_experiment_df=None, verbose=False, average_effects=(), n_simulations=None, alpha=None)

Run power analysis by simulation, using standard errors from the analysis. Args: df: Dataframe with outcome and treatment variables. pre_experiment_df: Dataframe with pre-experiment data. verbose: Whether to show progress bar. average_effects: Average effects to test. n_simulations: Number of simulations to run. alpha: Significance level.

Source code in cluster_experiments/power_analysis.py

def power_line(
    self,
    df: pd.DataFrame,
    pre_experiment_df: Optional[pd.DataFrame] = None,
    verbose: bool = False,
    average_effects: Iterable[float] = (),
    n_simulations: Optional[int] = None,
    alpha: Optional[float] = None,
) -> Dict[float, float]:
    """
    Run power analysis by simulation, using standard errors from the analysis.
    Args:
        df: Dataframe with outcome and treatment variables.
        pre_experiment_df: Dataframe with pre-experiment data.
        verbose: Whether to show progress bar.
        average_effects: Average effects to test.
        n_simulations: Number of simulations to run.
        alpha: Significance level.
    """
    alpha = self.alpha if alpha is None else alpha

    std_error_mean = self._get_average_standard_error(
        df=df,
        pre_experiment_df=pre_experiment_df,
        verbose=verbose,
        n_simulations=n_simulations,
    )

    return {
        effect: self._normal_power_calculation(
            alpha=alpha, std_error=std_error_mean, average_effect=effect
        )
        for effect in average_effects
    }

power_time_line(df, pre_experiment_df=None, verbose=False, average_effects=(), experiment_length=(), n_simulations=None, alpha=None)

Run power analysis by simulation, using standard errors from the analysis.

Parameters:

Name	Type	Description	Default
df	DataFrame	Dataframe with outcome and treatment variables.	required
pre_experiment_df	Optional[DataFrame]	Dataframe with pre-experiment data.	None
verbose	bool	Whether to show progress bar.	False
average_effects	Iterable[float]	Average effects to test.	()
experiment_length	Iterable[int]	Length of the experiment in days.	()
n_simulations	Optional[int]	Number of simulations to run.	None
alpha	Optional[float]	Significance level.	None

Source code in cluster_experiments/power_analysis.py

def power_time_line(
    self,
    df: pd.DataFrame,
    pre_experiment_df: Optional[pd.DataFrame] = None,
    verbose: bool = False,
    average_effects: Iterable[float] = (),
    experiment_length: Iterable[int] = (),
    n_simulations: Optional[int] = None,
    alpha: Optional[float] = None,
) -> List[Dict]:
    """
    Run power analysis by simulation, using standard errors from the analysis.

    Args:
        df: Dataframe with outcome and treatment variables.
        pre_experiment_df: Dataframe with pre-experiment data.
        verbose: Whether to show progress bar.
        average_effects: Average effects to test.
        experiment_length: Length of the experiment in days.
        n_simulations: Number of simulations to run.
        alpha: Significance level.
    """
    alpha = self.alpha if alpha is None else alpha

    results = []
    for std_error_mean, n_days in self.run_average_standard_error(
        df=df,
        pre_experiment_df=pre_experiment_df,
        verbose=verbose,
        n_simulations=n_simulations,
        experiment_length=experiment_length,
    ):
        for effect in average_effects:
            power = self._normal_power_calculation(
                alpha=alpha, std_error=std_error_mean, average_effect=effect
            )
            results.append(
                {"effect": effect, "power": power, "experiment_length": n_days}
            )

    return results

run_average_standard_error(df, pre_experiment_df=None, verbose=False, n_simulations=None, experiment_length=())

Run power analysis by simulation, using standard errors from the analysis.

Parameters:

Name	Type	Description	Default
df	DataFrame	Dataframe with outcome and treatment variables.	required
pre_experiment_df	Optional[DataFrame]	Dataframe with pre-experiment data.	None
verbose	bool	Whether to show progress bar.	False
n_simulations	Optional[int]	Number of simulations to run.	None
experiment_length	Iterable[int]	Length of the experiment in days.	()

Source code in cluster_experiments/power_analysis.py

def run_average_standard_error(
    self,
    df: pd.DataFrame,
    pre_experiment_df: Optional[pd.DataFrame] = None,
    verbose: bool = False,
    n_simulations: Optional[int] = None,
    experiment_length: Iterable[int] = (),
) -> Generator[Tuple[float, int], None, None]:
    """
    Run power analysis by simulation, using standard errors from the analysis.

    Args:
        df: Dataframe with outcome and treatment variables.
        pre_experiment_df: Dataframe with pre-experiment data.
        verbose: Whether to show progress bar.
        n_simulations: Number of simulations to run.
        experiment_length: Length of the experiment in days.
    """
    n_simulations = self.n_simulations if n_simulations is None else n_simulations

    for n_days in experiment_length:
        df_time = df.copy()
        experiment_start = df_time[self.time_col].min()
        df_time = df_time.loc[
            df_time[self.time_col] < experiment_start + pd.Timedelta(days=n_days)
        ]
        std_error_mean = self._get_average_standard_error(
            df=df_time,
            pre_experiment_df=pre_experiment_df,
            verbose=verbose,
            n_simulations=n_simulations,
        )
        yield std_error_mean, n_days

PowerAnalysis

Class used to run Power analysis. It does so by running simulations. In each simulation: 1. Assign treatment to dataframe randomly 2. Perturbate dataframe 3. Add pre-experiment data if needed 4. Run analysis

Finally it returns the power of the analysis by counting how many times the effect was detected.

Parameters:

Name	Type	Description	Default
perturbator	Perturbator	Perturbator class to perturbate dataframe with treatment assigned.	required
splitter	RandomSplitter	RandomSplitter class to randomly assign treatment to dataframe.	required
analysis	ExperimentAnalysis	ExperimentAnalysis class to use for analysis.	required
cupac_model	Optional[BaseEstimator]	Sklearn estimator class to add pre-experiment data to dataframe. If None, no pre-experiment data will be added.	None
target_col	str	Name of the column with the outcome variable.	'target'
treatment_col	str	Name of the column with the treatment variable.	'treatment'
treatment	str	value of treatment_col considered to be treatment (not control)	'B'
control	str	value of treatment_col considered to be control (not treatment)	'A'
n_simulations	int	Number of simulations to run.	100
alpha	float	Significance level.	0.05
features_cupac_model	Optional[List[str]]	Covariates to be used in cupac model	None
seed	Optional[int]	Optional. Seed to use for the splitter.	None

Usage:

from datetime import date

import numpy as np
import pandas as pd
from cluster_experiments.experiment_analysis import GeeExperimentAnalysis
from cluster_experiments.perturbator import ConstantPerturbator
from cluster_experiments.power_analysis import PowerAnalysis
from cluster_experiments.random_splitter import ClusteredSplitter

N = 1_000
users = [f"User {i}" for i in range(1000)]
clusters = [f"Cluster {i}" for i in range(100)]
dates = [f"{date(2022, 1, i):%Y-%m-%d}" for i in range(1, 32)]
df = pd.DataFrame(
    {
        "cluster": np.random.choice(clusters, size=N),
        "target": np.random.normal(0, 1, size=N),
        "user": np.random.choice(users, size=N),
        "date": np.random.choice(dates, size=N),
    }
)

experiment_dates = [f"{date(2022, 1, i):%Y-%m-%d}" for i in range(15, 32)]
sw = ClusteredSplitter(
    cluster_cols=["cluster", "date"],
)

perturbator = ConstantPerturbator()

analysis = GeeExperimentAnalysis(
    cluster_cols=["cluster", "date"],
)

pw = PowerAnalysis(
    perturbator=perturbator, splitter=sw, analysis=analysis, n_simulations=50
)

power = pw.power_analysis(df, average_effect=0.1)
print(f"{power = }")

Source code in cluster_experiments/power_analysis.py

class PowerAnalysis:
    """
    Class used to run Power analysis. It does so by running simulations. In each simulation:
    1. Assign treatment to dataframe randomly
    2. Perturbate dataframe
    3. Add pre-experiment data if needed
    4. Run analysis

    Finally it returns the power of the analysis by counting how many times the effect was detected.

    Args:
        perturbator: Perturbator class to perturbate dataframe with treatment assigned.
        splitter: RandomSplitter class to randomly assign treatment to dataframe.
        analysis: ExperimentAnalysis class to use for analysis.
        cupac_model: Sklearn estimator class to add pre-experiment data to dataframe. If None, no pre-experiment data will be added.
        target_col: Name of the column with the outcome variable.
        treatment_col: Name of the column with the treatment variable.
        treatment: value of treatment_col considered to be treatment (not control)
        control: value of treatment_col considered to be control (not treatment)
        n_simulations: Number of simulations to run.
        alpha: Significance level.
        features_cupac_model: Covariates to be used in cupac model
        seed: Optional. Seed to use for the splitter.

    Usage:
    ```python
    from datetime import date

    import numpy as np
    import pandas as pd
    from cluster_experiments.experiment_analysis import GeeExperimentAnalysis
    from cluster_experiments.perturbator import ConstantPerturbator
    from cluster_experiments.power_analysis import PowerAnalysis
    from cluster_experiments.random_splitter import ClusteredSplitter

    N = 1_000
    users = [f"User {i}" for i in range(1000)]
    clusters = [f"Cluster {i}" for i in range(100)]
    dates = [f"{date(2022, 1, i):%Y-%m-%d}" for i in range(1, 32)]
    df = pd.DataFrame(
        {
            "cluster": np.random.choice(clusters, size=N),
            "target": np.random.normal(0, 1, size=N),
            "user": np.random.choice(users, size=N),
            "date": np.random.choice(dates, size=N),
        }
    )

    experiment_dates = [f"{date(2022, 1, i):%Y-%m-%d}" for i in range(15, 32)]
    sw = ClusteredSplitter(
        cluster_cols=["cluster", "date"],
    )

    perturbator = ConstantPerturbator()

    analysis = GeeExperimentAnalysis(
        cluster_cols=["cluster", "date"],
    )

    pw = PowerAnalysis(
        perturbator=perturbator, splitter=sw, analysis=analysis, n_simulations=50
    )

    power = pw.power_analysis(df, average_effect=0.1)
    print(f"{power = }")
    ```
    """

    def __init__(
        self,
        perturbator: Perturbator,
        splitter: RandomSplitter,
        analysis: ExperimentAnalysis,
        cupac_model: Optional[BaseEstimator] = None,
        target_col: str = "target",
        treatment_col: str = "treatment",
        treatment: str = "B",
        control: str = "A",
        n_simulations: int = 100,
        alpha: float = 0.05,
        features_cupac_model: Optional[List[str]] = None,
        seed: Optional[int] = None,
        hypothesis: str = "two-sided",
    ):
        self.perturbator = perturbator
        self.splitter = splitter
        self.analysis = analysis
        self.n_simulations = n_simulations
        self.target_col = target_col
        self.treatment = treatment
        self.control = control
        self.treatment_col = treatment_col
        self.alpha = alpha
        self.hypothesis = hypothesis

        self.cupac_handler = CupacHandler(
            cupac_model=cupac_model,
            target_col=target_col,
            features_cupac_model=features_cupac_model,
        )
        if seed is not None:
            random.seed(seed)  # seed for splitter
            np.random.seed(seed)  # seed for the binary perturbator
            # may need to seed other stochasticity sources if added

        self.check_inputs()

    def _simulate_perturbed_df(
        self,
        df: pd.DataFrame,
        pre_experiment_df: Optional[pd.DataFrame] = None,
        verbose: bool = False,
        average_effect: Optional[float] = None,
        n_simulations: int = 100,
    ) -> Generator[pd.DataFrame, None, None]:
        """Yields splitted + perturbated dataframe for each iteration of the simulation."""
        df = df.copy()
        df = self.cupac_handler.add_covariates(df, pre_experiment_df)

        for _ in tqdm(range(n_simulations), disable=not verbose):
            yield self._split_and_perturbate(df, average_effect)

    def simulate_pvalue(
        self,
        df: pd.DataFrame,
        pre_experiment_df: Optional[pd.DataFrame] = None,
        verbose: bool = False,
        average_effect: Optional[float] = None,
        n_simulations: int = 100,
    ) -> Generator[float, None, None]:
        """
        Yields p-values for each iteration of the simulation.
        In general, this is to be used in power_analysis method. However,
        if you're interested in the distribution of p-values, you can use this method to generate them.
        Args:
            df: Dataframe with outcome variable.
            pre_experiment_df: Dataframe with pre-experiment data.
            verbose: Whether to show progress bar.
            average_effect: Average effect of treatment. If None, it will use the perturbator average effect.
            n_simulations: Number of simulations to run.
        """
        for perturbed_df in self._simulate_perturbed_df(
            df,
            pre_experiment_df=pre_experiment_df,
            verbose=verbose,
            average_effect=average_effect,
            n_simulations=n_simulations,
        ):
            yield self.analysis.get_pvalue(perturbed_df)

    def running_power_analysis(
        self,
        df: pd.DataFrame,
        pre_experiment_df: Optional[pd.DataFrame] = None,
        verbose: bool = False,
        average_effect: Optional[float] = None,
        n_simulations: int = 100,
    ) -> Generator[float, None, None]:
        """
        Yields running power for each iteration of the simulation.
        if you're interested in getting the power at each iteration, you can use this method to generate them.
        Args:
            df: Dataframe with outcome variable.
            pre_experiment_df: Dataframe with pre-experiment data.
            verbose: Whether to show progress bar.
            average_effect: Average effect of treatment. If None, it will use the perturbator average effect.
            n_simulations: Number of simulations to run.
        """
        n_rejected = 0
        for i, perturbed_df in enumerate(
            self._simulate_perturbed_df(
                df,
                pre_experiment_df=pre_experiment_df,
                verbose=verbose,
                average_effect=average_effect,
                n_simulations=n_simulations,
            )
        ):
            p_value = self.analysis.get_pvalue(perturbed_df)
            n_rejected += int(p_value < self.alpha)
            yield n_rejected / (i + 1)

    def simulate_point_estimate(
        self,
        df: pd.DataFrame,
        pre_experiment_df: Optional[pd.DataFrame] = None,
        verbose: bool = False,
        average_effect: Optional[float] = None,
        n_simulations: int = 100,
    ) -> Generator[float, None, None]:
        """
        Yields point estimates for each iteration of the simulation.
        In general, this is to be used in power_analysis method. However,
        if you're interested in the distribution of point estimates, you can use this method to generate them.

        This is an experimental feature and it might change in the future.

        Args:
            df: Dataframe with outcome and treatment variables.
            pre_experiment_df: Dataframe with pre-experiment data.
            verbose: Whether to show progress bar.
            average_effect: Average effect of treatment. If None, it will use the perturbator average effect.
            n_simulations: Number of simulations to run.
        """
        for perturbed_df in self._simulate_perturbed_df(
            df,
            pre_experiment_df=pre_experiment_df,
            verbose=verbose,
            average_effect=average_effect,
            n_simulations=n_simulations,
        ):
            yield self.analysis.get_point_estimate(perturbed_df)

    def power_analysis(
        self,
        df: pd.DataFrame,
        pre_experiment_df: Optional[pd.DataFrame] = None,
        verbose: bool = False,
        average_effect: Optional[float] = None,
        n_simulations: Optional[int] = None,
        alpha: Optional[float] = None,
        n_jobs: int = 1,
    ) -> float:
        """
        Run power analysis by simulation
        Args:
            df: Dataframe with outcome and treatment variables.
            pre_experiment_df: Dataframe with pre-experiment data.
            verbose: Whether to show progress bar.
            average_effect: Average effect of treatment. If None, it will use the perturbator average effect.
            n_simulations: Number of simulations to run.
            alpha: Significance level.
            n_jobs: Number of jobs to run in parallel. If 1, it will run in serial.
        """
        n_simulations = self.n_simulations if n_simulations is None else n_simulations
        alpha = self.alpha if alpha is None else alpha

        df = df.copy()
        df = self.cupac_handler.add_covariates(df, pre_experiment_df)

        if n_jobs == 1:
            return self._non_parallel_loop(
                df, average_effect, n_simulations, alpha, verbose
            )
        elif n_jobs > 1 or n_jobs == -1:
            return self._parallel_loop(
                df, average_effect, n_simulations, alpha, verbose, n_jobs
            )
        else:
            raise ValueError("n_jobs must be greater than 0, or -1.")

    def _split(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Split dataframe.
        Args:
            df: Dataframe with outcome variable
        """
        treatment_df = self.splitter.assign_treatment_df(df)
        self.log_nulls(treatment_df)
        treatment_df = treatment_df.query(
            f"{self.treatment_col}.notnull()", engine="python"
        ).query(
            f"{self.treatment_col}.isin(['{self.treatment}', '{self.control}'])",
            engine="python",
        )

        return treatment_df

    def _perturbate(
        self, treatment_df: pd.DataFrame, average_effect: Optional[float]
    ) -> pd.DataFrame:
        """
        Perturbate dataframe using perturbator.
        Args:
            df: Dataframe with outcome variable
            average_effect: Average effect of treatment. If None, it will use the perturbator average effect.
        """

        perturbed_df = self.perturbator.perturbate(
            treatment_df, average_effect=average_effect
        )
        return perturbed_df

    def _split_and_perturbate(
        self, df: pd.DataFrame, average_effect: Optional[float]
    ) -> pd.DataFrame:
        treatment_df = self._split(df)
        perturbed_df = self._perturbate(
            treatment_df=treatment_df, average_effect=average_effect
        )
        return perturbed_df

    def _run_simulation(self, args: Tuple[pd.DataFrame, Optional[float]]) -> float:
        df, average_effect = args
        perturbed_df = self._split_and_perturbate(df, average_effect)
        return self.analysis.get_pvalue(perturbed_df)

    def _non_parallel_loop(
        self,
        df: pd.DataFrame,
        average_effect: Optional[float],
        n_simulations: int,
        alpha: float,
        verbose: bool,
    ) -> float:
        """
        Run power analysis by simulation in serial
        Args:
            df: Dataframe with outcome and treatment variables.
            average_effect: Average effect of treatment. If None, it will use the perturbator average effect.
            n_simulations: Number of simulations to run.
            alpha: Significance level.
        """
        n_detected_mde = 0
        for _ in tqdm(range(n_simulations), disable=not verbose):
            p_value = self._run_simulation((df, average_effect))
            if verbose:
                print(f"p_value of simulation run: {p_value:.3f}")
            n_detected_mde += p_value < alpha

        return n_detected_mde / n_simulations

    def _parallel_loop(
        self,
        df: pd.DataFrame,
        average_effect: Optional[float],
        n_simulations: int,
        alpha: float,
        verbose: bool,
        n_jobs: int,
    ) -> float:
        """
        Run power analysis by simulation in parallel
        Args:
            df: Dataframe with outcome and treatment variables.
            average_effect: Average effect of treatment. If None, it will use the perturbator average effect.
            n_simulations: Number of simulations to run.
            alpha: Significance level.
            n_jobs: Number of jobs to run in parallel.
        """
        from multiprocessing import Pool, cpu_count

        n_jobs = n_jobs if n_jobs != -1 else cpu_count()

        n_detected_mde = 0
        with Pool(processes=n_jobs) as pool:
            args = [(df, average_effect) for _ in range(n_simulations)]
            results = pool.imap_unordered(self._run_simulation, args)
            for p_value in tqdm(results, total=n_simulations, disable=not verbose):
                n_detected_mde += p_value < alpha

        return n_detected_mde / n_simulations

    def power_line(
        self,
        df: pd.DataFrame,
        pre_experiment_df: Optional[pd.DataFrame] = None,
        verbose: bool = False,
        average_effects: Iterable[float] = (),
        n_simulations: Optional[int] = None,
        alpha: Optional[float] = None,
        n_jobs: int = 1,
    ) -> Dict[float, float]:
        """Runs power analysis with multiple average effects

        Args:
            df: Dataframe with outcome and treatment variables.
            pre_experiment_df: Dataframe with pre-experiment data.
            verbose: Whether to show progress bar.
            average_effects: Average effects to test.
            n_simulations: Number of simulations to run.
            alpha: Significance level.
            n_jobs: Number of jobs to run in parallel.

        Returns:
            Dictionary with average effects as keys and power as values.
        """
        return {
            effect: self.power_analysis(
                df=df,
                pre_experiment_df=pre_experiment_df,
                verbose=verbose,
                average_effect=effect,
                n_simulations=n_simulations,
                alpha=alpha,
                n_jobs=n_jobs,
            )
            for effect in tqdm(
                list(average_effects), disable=not verbose, desc="Effects loop"
            )
        }

    def log_nulls(self, df: pd.DataFrame) -> None:
        """Warns about dropping nulls in treatment column"""
        n_nulls = len(df.query(f"{self.treatment_col}.isnull()", engine="python"))
        if n_nulls > 0:
            logging.warning(
                f"There are {n_nulls} null values in treatment, dropping them"
            )

    @classmethod
    def from_dict(cls, config_dict: dict) -> "PowerAnalysis":
        """Constructs PowerAnalysis from dictionary"""
        config = PowerConfig(**config_dict)
        return cls.from_config(config)

    @classmethod
    def from_config(cls, config: PowerConfig) -> "PowerAnalysis":
        """Constructs PowerAnalysis from PowerConfig"""
        perturbator_cls = _get_mapping_key(perturbator_mapping, config.perturbator)
        splitter_cls = _get_mapping_key(splitter_mapping, config.splitter)
        analysis_cls = _get_mapping_key(analysis_mapping, config.analysis)
        cupac_cls = _get_mapping_key(cupac_model_mapping, config.cupac_model)
        return cls(
            perturbator=perturbator_cls.from_config(config),
            splitter=splitter_cls.from_config(config),
            analysis=analysis_cls.from_config(config),
            cupac_model=cupac_cls.from_config(config),
            target_col=config.target_col,
            treatment_col=config.treatment_col,
            treatment=config.treatment,
            control=config.control,
            n_simulations=config.n_simulations,
            alpha=config.alpha,
            features_cupac_model=config.features_cupac_model,
            seed=config.seed,
            hypothesis=config.hypothesis,
        )

    def check_treatment_col(self):
        """Checks consistency of treatment column"""
        assert (
            self.analysis.treatment_col == self.perturbator.treatment_col
        ), f"treatment_col in analysis ({self.analysis.treatment_col}) must be the same as treatment_col in perturbator ({self.perturbator.treatment_col})"

        assert (
            self.analysis.treatment_col == self.treatment_col
        ), f"treatment_col in analysis ({self.analysis.treatment_col}) must be the same as treatment_col in PowerAnalysis ({self.treatment_col})"

        assert (
            self.analysis.treatment_col == self.splitter.treatment_col
        ), f"treatment_col in analysis ({self.analysis.treatment_col}) must be the same as treatment_col in splitter ({self.splitter.treatment_col})"

    def check_target_col(self):
        assert (
            self.analysis.target_col == self.perturbator.target_col
        ), f"target_col in analysis ({self.analysis.target_col}) must be the same as target_col in perturbator ({self.perturbator.target_col})"

        assert (
            self.analysis.target_col == self.target_col
        ), f"target_col in analysis ({self.analysis.target_col}) must be the same as target_col in PowerAnalysis ({self.target_col})"

    def check_treatment(self):
        assert (
            self.analysis.treatment == self.perturbator.treatment
        ), f"treatment in analysis ({self.analysis.treatment}) must be the same as treatment in perturbator ({self.perturbator.treatment})"

        assert (
            self.analysis.treatment == self.treatment
        ), f"treatment in analysis ({self.analysis.treatment}) must be the same as treatment in PowerAnalysis ({self.treatment})"

        assert (
            self.analysis.treatment in self.splitter.treatments
        ), f"treatment in analysis ({self.analysis.treatment}) must be in treatments in splitter ({self.splitter.treatments})"

        assert (
            self.control in self.splitter.treatments
        ), f"control in power analysis ({self.control}) must be in treatments in splitter ({self.splitter.treatments})"

    def check_covariates(self):
        if hasattr(self.analysis, "covariates"):
            cupac_in_covariates = (
                self.cupac_handler.cupac_outcome_name in self.analysis.covariates
            )

            assert cupac_in_covariates or not self.cupac_handler.is_cupac, (
                f"covariates in analysis must contain {self.cupac_handler.cupac_outcome_name} if cupac_model is not None. "
                f"If you want to use cupac_model, you must add the cupac outcome to the covariates of the analysis "
                f"You may want to do covariates=['{self.cupac_handler.cupac_outcome_name}'] in your analysis method or your config"
            )

            if hasattr(self.splitter, "cluster_cols"):
                if set(self.analysis.covariates).intersection(
                    set(self.splitter.cluster_cols)
                ):
                    logging.warning(
                        f"covariates in analysis ({self.analysis.covariates}) are also cluster_cols in splitter ({self.splitter.cluster_cols}). "
                        f"Be specially careful when using switchback splitters, since the time splitter column is being overriden"
                    )

    def check_clusters(self):
        has_analysis_clusters = hasattr(self.analysis, "cluster_cols")
        has_splitter_clusters = hasattr(self.splitter, "cluster_cols")
        not_cluster_cols_cond = not has_analysis_clusters or not has_splitter_clusters
        assert (
            not_cluster_cols_cond
            or self.analysis.cluster_cols == self.splitter.cluster_cols
        ), f"cluster_cols in analysis ({self.analysis.cluster_cols}) must be the same as cluster_cols in splitter ({self.splitter.cluster_cols})"

        assert (
            has_splitter_clusters
            or not has_analysis_clusters
            or not self.analysis.cluster_cols
            or isinstance(self.splitter, RepeatedSampler)
        ), "analysis has cluster_cols but splitter does not."

        assert (
            has_analysis_clusters
            or not has_splitter_clusters
            or not self.splitter.cluster_cols
        ), "splitter has cluster_cols but analysis does not."

        has_time_col = hasattr(self.splitter, "time_col")
        assert not (
            has_time_col
            and has_splitter_clusters
            and self.splitter.time_col not in self.splitter.cluster_cols
        ), "in switchback splitters, time_col must be in cluster_cols"

    def check_inputs(self):
        self.check_covariates()
        self.check_treatment_col()
        self.check_target_col()
        self.check_treatment()
        self.check_clusters()

check_treatment_col()

Checks consistency of treatment column

Source code in cluster_experiments/power_analysis.py

def check_treatment_col(self):
    """Checks consistency of treatment column"""
    assert (
        self.analysis.treatment_col == self.perturbator.treatment_col
    ), f"treatment_col in analysis ({self.analysis.treatment_col}) must be the same as treatment_col in perturbator ({self.perturbator.treatment_col})"

    assert (
        self.analysis.treatment_col == self.treatment_col
    ), f"treatment_col in analysis ({self.analysis.treatment_col}) must be the same as treatment_col in PowerAnalysis ({self.treatment_col})"

    assert (
        self.analysis.treatment_col == self.splitter.treatment_col
    ), f"treatment_col in analysis ({self.analysis.treatment_col}) must be the same as treatment_col in splitter ({self.splitter.treatment_col})"

from_config(config) classmethod

Constructs PowerAnalysis from PowerConfig

Source code in cluster_experiments/power_analysis.py

@classmethod
def from_config(cls, config: PowerConfig) -> "PowerAnalysis":
    """Constructs PowerAnalysis from PowerConfig"""
    perturbator_cls = _get_mapping_key(perturbator_mapping, config.perturbator)
    splitter_cls = _get_mapping_key(splitter_mapping, config.splitter)
    analysis_cls = _get_mapping_key(analysis_mapping, config.analysis)
    cupac_cls = _get_mapping_key(cupac_model_mapping, config.cupac_model)
    return cls(
        perturbator=perturbator_cls.from_config(config),
        splitter=splitter_cls.from_config(config),
        analysis=analysis_cls.from_config(config),
        cupac_model=cupac_cls.from_config(config),
        target_col=config.target_col,
        treatment_col=config.treatment_col,
        treatment=config.treatment,
        control=config.control,
        n_simulations=config.n_simulations,
        alpha=config.alpha,
        features_cupac_model=config.features_cupac_model,
        seed=config.seed,
        hypothesis=config.hypothesis,
    )

from_dict(config_dict) classmethod

Constructs PowerAnalysis from dictionary

Source code in cluster_experiments/power_analysis.py

@classmethod
def from_dict(cls, config_dict: dict) -> "PowerAnalysis":
    """Constructs PowerAnalysis from dictionary"""
    config = PowerConfig(**config_dict)
    return cls.from_config(config)

log_nulls(df)

Warns about dropping nulls in treatment column

Source code in cluster_experiments/power_analysis.py

def log_nulls(self, df: pd.DataFrame) -> None:
    """Warns about dropping nulls in treatment column"""
    n_nulls = len(df.query(f"{self.treatment_col}.isnull()", engine="python"))
    if n_nulls > 0:
        logging.warning(
            f"There are {n_nulls} null values in treatment, dropping them"
        )

power_analysis(df, pre_experiment_df=None, verbose=False, average_effect=None, n_simulations=None, alpha=None, n_jobs=1)

Run power analysis by simulation Args: df: Dataframe with outcome and treatment variables. pre_experiment_df: Dataframe with pre-experiment data. verbose: Whether to show progress bar. average_effect: Average effect of treatment. If None, it will use the perturbator average effect. n_simulations: Number of simulations to run. alpha: Significance level. n_jobs: Number of jobs to run in parallel. If 1, it will run in serial.

Source code in cluster_experiments/power_analysis.py

def power_analysis(
    self,
    df: pd.DataFrame,
    pre_experiment_df: Optional[pd.DataFrame] = None,
    verbose: bool = False,
    average_effect: Optional[float] = None,
    n_simulations: Optional[int] = None,
    alpha: Optional[float] = None,
    n_jobs: int = 1,
) -> float:
    """
    Run power analysis by simulation
    Args:
        df: Dataframe with outcome and treatment variables.
        pre_experiment_df: Dataframe with pre-experiment data.
        verbose: Whether to show progress bar.
        average_effect: Average effect of treatment. If None, it will use the perturbator average effect.
        n_simulations: Number of simulations to run.
        alpha: Significance level.
        n_jobs: Number of jobs to run in parallel. If 1, it will run in serial.
    """
    n_simulations = self.n_simulations if n_simulations is None else n_simulations
    alpha = self.alpha if alpha is None else alpha

    df = df.copy()
    df = self.cupac_handler.add_covariates(df, pre_experiment_df)

    if n_jobs == 1:
        return self._non_parallel_loop(
            df, average_effect, n_simulations, alpha, verbose
        )
    elif n_jobs > 1 or n_jobs == -1:
        return self._parallel_loop(
            df, average_effect, n_simulations, alpha, verbose, n_jobs
        )
    else:
        raise ValueError("n_jobs must be greater than 0, or -1.")

power_line(df, pre_experiment_df=None, verbose=False, average_effects=(), n_simulations=None, alpha=None, n_jobs=1)

Runs power analysis with multiple average effects

Parameters:

Name	Type	Description	Default
df	DataFrame	Dataframe with outcome and treatment variables.	required
pre_experiment_df	Optional[DataFrame]	Dataframe with pre-experiment data.	None
verbose	bool	Whether to show progress bar.	False
average_effects	Iterable[float]	Average effects to test.	()
n_simulations	Optional[int]	Number of simulations to run.	None
alpha	Optional[float]	Significance level.	None
n_jobs	int	Number of jobs to run in parallel.	1

Returns:

Type	Description
Dict[float, float]	Dictionary with average effects as keys and power as values.

Source code in cluster_experiments/power_analysis.py

def power_line(
    self,
    df: pd.DataFrame,
    pre_experiment_df: Optional[pd.DataFrame] = None,
    verbose: bool = False,
    average_effects: Iterable[float] = (),
    n_simulations: Optional[int] = None,
    alpha: Optional[float] = None,
    n_jobs: int = 1,
) -> Dict[float, float]:
    """Runs power analysis with multiple average effects

    Args:
        df: Dataframe with outcome and treatment variables.
        pre_experiment_df: Dataframe with pre-experiment data.
        verbose: Whether to show progress bar.
        average_effects: Average effects to test.
        n_simulations: Number of simulations to run.
        alpha: Significance level.
        n_jobs: Number of jobs to run in parallel.

    Returns:
        Dictionary with average effects as keys and power as values.
    """
    return {
        effect: self.power_analysis(
            df=df,
            pre_experiment_df=pre_experiment_df,
            verbose=verbose,
            average_effect=effect,
            n_simulations=n_simulations,
            alpha=alpha,
            n_jobs=n_jobs,
        )
        for effect in tqdm(
            list(average_effects), disable=not verbose, desc="Effects loop"
        )
    }

running_power_analysis(df, pre_experiment_df=None, verbose=False, average_effect=None, n_simulations=100)

Yields running power for each iteration of the simulation. if you're interested in getting the power at each iteration, you can use this method to generate them. Args: df: Dataframe with outcome variable. pre_experiment_df: Dataframe with pre-experiment data. verbose: Whether to show progress bar. average_effect: Average effect of treatment. If None, it will use the perturbator average effect. n_simulations: Number of simulations to run.

Source code in cluster_experiments/power_analysis.py

def running_power_analysis(
    self,
    df: pd.DataFrame,
    pre_experiment_df: Optional[pd.DataFrame] = None,
    verbose: bool = False,
    average_effect: Optional[float] = None,
    n_simulations: int = 100,
) -> Generator[float, None, None]:
    """
    Yields running power for each iteration of the simulation.
    if you're interested in getting the power at each iteration, you can use this method to generate them.
    Args:
        df: Dataframe with outcome variable.
        pre_experiment_df: Dataframe with pre-experiment data.
        verbose: Whether to show progress bar.
        average_effect: Average effect of treatment. If None, it will use the perturbator average effect.
        n_simulations: Number of simulations to run.
    """
    n_rejected = 0
    for i, perturbed_df in enumerate(
        self._simulate_perturbed_df(
            df,
            pre_experiment_df=pre_experiment_df,
            verbose=verbose,
            average_effect=average_effect,
            n_simulations=n_simulations,
        )
    ):
        p_value = self.analysis.get_pvalue(perturbed_df)
        n_rejected += int(p_value < self.alpha)
        yield n_rejected / (i + 1)

simulate_point_estimate(df, pre_experiment_df=None, verbose=False, average_effect=None, n_simulations=100)

Yields point estimates for each iteration of the simulation. In general, this is to be used in power_analysis method. However, if you're interested in the distribution of point estimates, you can use this method to generate them.

This is an experimental feature and it might change in the future.

Parameters:

Name	Type	Description	Default
df	DataFrame	Dataframe with outcome and treatment variables.	required
pre_experiment_df	Optional[DataFrame]	Dataframe with pre-experiment data.	None
verbose	bool	Whether to show progress bar.	False
average_effect	Optional[float]	Average effect of treatment. If None, it will use the perturbator average effect.	None
n_simulations	int	Number of simulations to run.	100

Source code in cluster_experiments/power_analysis.py

def simulate_point_estimate(
    self,
    df: pd.DataFrame,
    pre_experiment_df: Optional[pd.DataFrame] = None,
    verbose: bool = False,
    average_effect: Optional[float] = None,
    n_simulations: int = 100,
) -> Generator[float, None, None]:
    """
    Yields point estimates for each iteration of the simulation.
    In general, this is to be used in power_analysis method. However,
    if you're interested in the distribution of point estimates, you can use this method to generate them.

    This is an experimental feature and it might change in the future.

    Args:
        df: Dataframe with outcome and treatment variables.
        pre_experiment_df: Dataframe with pre-experiment data.
        verbose: Whether to show progress bar.
        average_effect: Average effect of treatment. If None, it will use the perturbator average effect.
        n_simulations: Number of simulations to run.
    """
    for perturbed_df in self._simulate_perturbed_df(
        df,
        pre_experiment_df=pre_experiment_df,
        verbose=verbose,
        average_effect=average_effect,
        n_simulations=n_simulations,
    ):
        yield self.analysis.get_point_estimate(perturbed_df)

simulate_pvalue(df, pre_experiment_df=None, verbose=False, average_effect=None, n_simulations=100)

Yields p-values for each iteration of the simulation. In general, this is to be used in power_analysis method. However, if you're interested in the distribution of p-values, you can use this method to generate them. Args: df: Dataframe with outcome variable. pre_experiment_df: Dataframe with pre-experiment data. verbose: Whether to show progress bar. average_effect: Average effect of treatment. If None, it will use the perturbator average effect. n_simulations: Number of simulations to run.

Source code in cluster_experiments/power_analysis.py

def simulate_pvalue(
    self,
    df: pd.DataFrame,
    pre_experiment_df: Optional[pd.DataFrame] = None,
    verbose: bool = False,
    average_effect: Optional[float] = None,
    n_simulations: int = 100,
) -> Generator[float, None, None]:
    """
    Yields p-values for each iteration of the simulation.
    In general, this is to be used in power_analysis method. However,
    if you're interested in the distribution of p-values, you can use this method to generate them.
    Args:
        df: Dataframe with outcome variable.
        pre_experiment_df: Dataframe with pre-experiment data.
        verbose: Whether to show progress bar.
        average_effect: Average effect of treatment. If None, it will use the perturbator average effect.
        n_simulations: Number of simulations to run.
    """
    for perturbed_df in self._simulate_perturbed_df(
        df,
        pre_experiment_df=pre_experiment_df,
        verbose=verbose,
        average_effect=average_effect,
        n_simulations=n_simulations,
    ):
        yield self.analysis.get_pvalue(perturbed_df)

PowerAnalysisWithPreExperimentData

Bases: PowerAnalysis

This is intended to work mainly for diff-in-diff or synthetic control-like estimators, and NOT for cases of CUPED/CUPAC. Same as PowerAnalysis, but allowing a perturbation only at experiment period and keeping pre-experiment df intact. Using this class, the pre experiment df is also available when the class is instantiated.

Source code in cluster_experiments/power_analysis.py

class PowerAnalysisWithPreExperimentData(PowerAnalysis):
    """
    This is intended to work mainly for diff-in-diff or synthetic control-like estimators, and NOT for cases of CUPED/CUPAC.
    Same as PowerAnalysis, but allowing a perturbation only at experiment period and keeping pre-experiment df intact.
    Using this class, the pre experiment df is also available when the class is instantiated.
    """

    def _perturbate(
        self, treatment_df: pd.DataFrame, average_effect: Optional[float]
    ) -> pd.DataFrame:
        if not hasattr(self.analysis, "_split_pre_experiment_df"):
            raise AttributeError(
                "The PowerAnalysisWithPreExperimentData is intended to work mainly for diff-in-diff or synthetic control-like estimators."
                "For other cases use the PowerAnalysis"
            )

        df, pre_experiment_df = self.analysis._split_pre_experiment_df(treatment_df)

        perturbed_df = self.perturbator.perturbate(df, average_effect=average_effect)

        return pd.concat([perturbed_df, pre_experiment_df])