Spaces:
Runtime error
Runtime error
File size: 7,777 Bytes
79c4838 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 |
import numpy as np
class ExperimentTerminator():
def __init__(self):
# The number of samples to be used in Monte Carlo elements of the code. Increasing this number
# improves accuracy of estimates at the expense of run time.
self.mc_samples = 5_000
# The Type I Error rate for the experiment. This determines the credible interval size used
# for all calculations (e.g., alpha = 0.05 produces 95% credible intervals)
self.alpha = 0.05
def get_posterior_samples(self,
completed_trials_a,
completed_trials_b,
successes_a,
successes_b):
"""
Get samples from the posterior distribution of the probability of success in the
control group and the test group.
:param completed_trials_a: Integer giving the number of trials completed in the control group
:param completed_trials_b: Integer giving the number of trials completed in the test group
:param successes_a: Integer giving the number of successes observed so far in the control group
:param successes_b: Integer giving the number of successes observed so far in the test group
:return: A list with two arrays of samples. The first array is a set of posterior samples from
the distribution of the probability of success in the control group. The second array is the
same for the test group.
"""
posterior_samples_a = np.random.beta(successes_a,
completed_trials_a - successes_a,
self.mc_samples)
posterior_samples_b = np.random.beta(successes_b,
completed_trials_b - successes_b,
self.mc_samples)
return [posterior_samples_a, posterior_samples_b]
def get_prob_reject_null(self,
planned_trials_a,
planned_trials_b,
completed_trials_a,
completed_trials_b,
successes_a,
successes_b,
posterior_samples_a,
posterior_samples_b):
"""
Calculate the probability that the null hypothesis will be rejected by the planned end
of the experiment
:param planned_trials_a: Integer giving the number of trials planned to be completed in
the control group in the experiment
:param planned_trials_b: Integer giving the number of trials planned to be completed in
the test group in the experiment
:param completed_trials_a: Integer giving the number of trials completed in the control group
:param completed_trials_b: Integer giving the number of trials completed in the teest group
:param successes_a: Integer giving the number of successes observed so far in the control group
:param successes_b: Integer giving the number of successes observed so far in the test group
:param posterior_samples_a: Posterior samples for the control group returned by get_posterior_samples
:param posterior_samples_a: Posterior samples for the test group returned by get_posterior_samples
:return: Float with the posterior predictive probability of rejecting the null hypothesis.
"""
potential_successes_a = np.random.binomial(planned_trials_a - completed_trials_a,
posterior_samples_a,
self.mc_samples)
potential_successes_a += successes_a
potential_successes_b = np.random.binomial(planned_trials_b - completed_trials_b,
posterior_samples_b,
self.mc_samples)
potential_successes_b += successes_b
rejection = np.zeros(self.mc_samples)
for i in range(self.mc_samples):
post_samples_a = np.random.beta(potential_successes_a[i] + 1,
planned_trials_a - potential_successes_a[i] + 1,
self.mc_samples)
post_samples_b = np.random.beta(potential_successes_b[i] + 1,
planned_trials_b - potential_successes_b[i] + 1,
self.mc_samples)
post_samples_b_minus_a = post_samples_b - post_samples_a
interval = np.quantile(post_samples_b_minus_a, [self.alpha / 2, 1 - (self.alpha / 2)])
if (interval[0] > 0 or interval[1] < 0):
rejection[i] = 1
return np.mean(rejection)
def analyze_experiment(self,
planned_trials_a,
planned_trials_b,
completed_trials_a,
completed_trials_b,
successes_a,
successes_b):
"""
Based on the number of planned trials, completed trials, and successes observed so far in the
control and test groups, calculate a bunch of summary measures of the posterior distribution
and predicted posterior.
:param planned_trials_a: Integer giving the number of trials planned to be completed in
the control group in the experiment
:param planned_trials_b: Integer giving the number of trials planned to be completed in
the test group in the experiment
:param completed_trials_a: Integer giving the number of trials completed in the control group
:param completed_trials_b: Integer giving the number of trials completed in the teest group
:param successes_a: Integer giving the number of successes observed so far in the control group
:param successes_b: Integer giving the number of successes observed so far in the test group
"""
posterior_samples = self.get_posterior_samples(completed_trials_a,
completed_trials_b,
successes_a,
successes_b)
posterior_lift = (posterior_samples[1] - posterior_samples[0]) / posterior_samples[0]
conversion_rate_a = successes_a / completed_trials_a
conversion_rate_b = successes_b / completed_trials_b
expected_lift = np.mean(posterior_lift)
pr_b_gt_a = np.mean(posterior_lift >= 0)
pr_reject_null = self.get_prob_reject_null(planned_trials_a,
planned_trials_b,
completed_trials_a,
completed_trials_b,
successes_a,
successes_b,
posterior_samples[0],
posterior_samples[1])
return [conversion_rate_a,
conversion_rate_b,
expected_lift,
pr_b_gt_a,
pr_reject_null,
posterior_lift]
if __name__ == "__main__":
et = ExperimentTerminator()
exp_outcomes = et.analyze_experiment(2000, 2000, 1000, 1000, 250, 270)
print(exp_outcomes) |