In this chapter, you will bring together everything you have learned by completing a full A/B test workflow from start to finish. This comprehensive challenge will guide you through each stage, reinforcing your understanding and practical skills. Here is a checklist of the steps you should follow for a complete A/B test analysis:

- Define a clear hypothesis for your A/B test;
- Simulate or obtain sample data for both control and variant groups;
- Calculate and compare relevant metrics for each group;
- Perform a statistical test to determine significance;
- Compute confidence intervals for your key metric(s);
- Visualize the results for clear communication;
- Interpret your findings and summarize your conclusions in a concise report.

By following these steps, you will demonstrate your ability to conduct and communicate a robust A/B test analysis, preparing you for real-world experimentation scenarios.

import unittest
import user_code
import ast
import re   
import importlib
import csv
import numpy as np
import pandas as pd
import sys
from io import StringIO

class TestTask(unittest.TestCase):
    def test_simulate_ab_test_data_output(self):
        import user_code
        importlib.reload(user_code)
        # Use larger sample sizes to reduce variance
        data = user_code.simulate_ab_test_data(
            size_control=1000,
            size_variant=2000,
            conv_rate_control=0.1,
            conv_rate_variant=0.3,
            seed=123
        )
        _dynamic_test(
            self,
            isinstance(data, pd.DataFrame),
            "Function returns a DataFrame.",
            f"Expected pandas DataFrame, got {type(data)}."
        )
        control = data[data['group'] == 'control']
        variant = data[data['group'] == 'variant']
        _dynamic_test(
            self,
            len(control) == 1000 and len(variant) == 2000,
            "Correct sample sizes in each group.",
            f"Expected 1000 control and 2000 variant rows, got {len(control)} and {len(variant)}."
        )
        # Check conversion rates are within reasonable bounds (allow for statistical noise)
        conv_rate_c = control['conversion'].mean()
        conv_rate_v = variant['conversion'].mean()
        _dynamic_test(
            self,
            0.07 <= conv_rate_c <= 0.13,
            "Control group conversion rate is within reasonable range of input rate.",
            f"Expected ~0.1 (+/-0.03), got {conv_rate_c}."
        )
        _dynamic_test(
            self,
            0.26 <= conv_rate_v <= 0.34,
            "Variant group conversion rate is within reasonable range of input rate.",
            f"Expected ~0.3 (+/-0.04), got {conv_rate_v}."
        )

    def test_analyze_ab_test(self):
        import user_code
        importlib.reload(user_code)
        # Use deterministic (non-random) data for reproducible test
        data = pd.DataFrame({
            'group': ['control'] * 20 + ['variant'] * 20,
            'conversion': [0,1,0,0,0,1,0,0,0,0, 1,0,1,0,0,1,0,0,0,1,  # 6/20 = 0.3
                           1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1]   # 20/20 = 1.0
        })
        result = user_code.analyze_ab_test(data)
        _dynamic_test(
            self,
            isinstance(result, dict),
            "analyze_ab_test returns a dict.",
            f"Expected dict, got {type(result)}."
        )
        _dynamic_test(
            self,
            all(k in result for k in ['conv_rate_control','conv_rate_variant','difference','z_score','p_value','ci_low','ci_high']),
            "All required keys present in result.",
            f"Missing keys in result: {result.keys()}."
        )
        # Check conversion rates
        _dynamic_test(
            self,
            np.isclose(result['conv_rate_control'], 0.3),
            "Correct control group conversion rate.",
            f"Expected 0.3, got {result['conv_rate_control']}."
        )
        _dynamic_test(
            self,
            np.isclose(result['conv_rate_variant'], 1.0),
            "Correct variant group conversion rate.",
            f"Expected 1.0, got {result['conv_rate_variant']}."
        )
        # Difference
        _dynamic_test(
            self,
            np.isclose(result['difference'], 0.7),
            "Correct difference in conversion rates.",
            f"Expected 0.7, got {result['difference']}."
        )
        # p-value should be < 0.05 for this big difference
        _dynamic_test(
            self,
            result['p_value'] < 0.05,
            "p-value is significant for large difference.",
            f"Expected p-value < 0.05, got {result['p_value']}."
        )
        # CI should contain the difference
        _dynamic_test(
            self,
            result['ci_low'] < result['difference'] < result['ci_high'],
            "Difference inside confidence interval.",
            f"Difference is not inside CI: {result['ci_low']}, {result['difference']}, {result['ci_high']}."
        )

    def test_plot_ab_results(self):
        import user_code
        importlib.reload(user_code)
        import matplotlib
        matplotlib.use('Agg')
        data = pd.DataFrame({
            'group': ['control'] * 10 + ['variant'] * 10,
            'conversion': [0,1,0,0,0,1,0,0,0,0, 1,1,1,0,0,1,1,0,0,1]
        })
        # Should not raise
        try:
            user_code.plot_ab_results(data)
            _dynamic_test(
                self,
                True,
                "plot_ab_results runs without error.",
                "plot_ab_results raised an exception."
            )
        except Exception as e:
            _dynamic_test(
                self,
                False,
                "plot_ab_results runs without error.",
                f"plot_ab_results raised an exception: {e}"
            )

    def test_interpretation_and_reporting(self):
        import user_code
        importlib.reload(user_code)
        # Simulate data and run analysis (large sample for statistical stability)
        data = user_code.simulate_ab_test_data(
            size_control=2000,
            size_variant=2000,
            conv_rate_control=0.12,
            conv_rate_variant=0.15,
            seed=42
        )
        result = user_code.analyze_ab_test(data)
        # Capture print output
        buffer = StringIO()
        sys.stdout = buffer
        user_code.plot_ab_results(data)
        interpretation = (
            f"Control group conversion rate: {result['conv_rate_control']:.3f}\\n"
            f"Variant group conversion rate: {result['conv_rate_variant']:.3f}\\n"
            f"Difference: {result['difference']:.3f}\\n"
            f"95% Confidence Interval: [{result['ci_low']:.3f}, {result['ci_high']:.3f}]\\n"
            f"p-value: {result['p_value']:.4f}\\n"
        )
        print(interpretation)
        if result['p_value'] < 0.05:
            conclusion = "The result is statistically significant. The variant outperforms the control."
        else:
            conclusion = "The result is not statistically significant. There is no evidence the variant outperforms the control."
        print(conclusion)
        sys.stdout = sys.__stdout__
        output = buffer.getvalue()
        _dynamic_test(
            self,
            "Control group conversion rate" in output and "Variant group conversion rate" in output,
            "Interpretation summary includes conversion rates.",
            f"Interpretation summary missing conversion rates. Output: {output}"
        )
        if result['p_value'] < 0.05:
            _dynamic_test(
                self,
                "statistically significant" in output.lower(),
                "Correct conclusion for significant result.",
                f"Expected 'statistically significant' in output. Output: {output}"
            )
        else:
            _dynamic_test(
                self,
                "not statistically significant" in output.lower(),
                "Correct conclusion for nonsignificant result.",
                f"Expected 'not statistically significant' in output. Output: {output}"
            )

def _dynamic_test(test_case, condition, success_message, failure_message):
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)

def normalize_text(text):
    text = text.lower()
    text = re.sub(r"\\s{2,}", " ", text)
    text = re.sub(r"\\s*([,:?])\\s*", r"\\1 ", text)
    return text.strip()

def change_var(code: str, var_name: str, value: str) -> str:
    tree = ast.parse(code)
    lines = code.splitlines()
    changed = False
    # Collect all assignment nodes to modify
    assign_nodes = [
        (i, node)
        for i, node in enumerate(tree.body)
        if isinstance(node, ast.Assign)
        and any(isinstance(target, ast.Name) and target.id == var_name for target in node.targets)
    ]

    # If nothing to change, return unmodified code
    if not assign_nodes:
        return code

    # Perform replacements for all matching assignments (from last to first to not break line offsets)
    for i, node in reversed(assign_nodes):
        start_line = node.lineno - 1
        line = lines[start_line]
        indent = ' ' * (len(line) - len(line.lstrip()))
        lines[start_line] = f"{indent}{var_name} = {value}"
        next_line = len(lines)
        for next_node in tree.body[i+1:]:
            if hasattr(next_node, 'lineno'):
                next_line = next_node.lineno - 1
                break
        if next_line > start_line + 1:
            lines[start_line+1:next_line] = []
        changed = True

    return '\\n'.join(lines) if changed else code

if __name__ == "__main__":
    unittest.main()


Challenge: Full A/B Test Workflow

解答