This video introduces the concept of cohort analysis using a real-world scenario: SpaceStream's user retention challenge. You will learn what a cohort is and why tracking user activity over time is essential for subscription-based services. The video explains three key metrics: **retention rate** (the proportion of the original cohort active in each month), **churn rate** (the fraction that becomes inactive), and **survival rate** (the proportion of users who have stayed active every month up to the current point). The walkthrough demonstrates how to structure your dataset in pandas, with columns for each month and rows for each user, and shows step-by-step calculations for each metric using pandas operations. By the end, you will see how to interpret the results and understand user behavior trends over multiple months.

Je bent nu verantwoordelijk voor het berekenen van geavanceerde retentiemetingen voor SpaceStream, een intergalactische holovisiondienst. Als Lead Data Analyst analyseer je een cohort van 5 gebruikers over drie maanden, waarbij je bijhoudt wie loyaal blijft en wie afhaakt. Het doel is om drie cruciale statistieken per maand te berekenen: **retentiepercentage**, **churn-percentage** en **overlevingspercentage**.

Begin met het onderzoeken van de meegeleverde dataset, waarin elke gebruiker als actief (`1`) of inactief (`0`) is gemarkeerd voor elke maand. De kolommen `month_0`, `month_1` en `month_2` geven de activiteit over drie opeenvolgende maanden weer. Je oplossing vereist het gebruik van `pandas` om deze dataset te verwerken en de benodigde statistieken per maand te extraheren.

import pandas as pd

data = {
    "user_id": [1, 2, 3, 4, 5],
    "month_0": [1, 1, 1, 1, 1],  # Everyone starts active
    "month_1": [1, 0, 1, 0, 1],  # 3 users active
    "month_2": [1, 0, 0, 0, 0],  # 1 user active
}
df = pd.DataFrame(data)

# Calculating retention rate: fraction of original cohort active in each month
cohort_size = len(df)
retention_rate = [df[f"month_{i}"].sum() / cohort_size for i in range(3)]

# Calculating churn rate: 1 - retention rate
churn_rate = [1 - r for r in retention_rate]

# Calculating survival rate: fraction of users still active in ALL months up to i
survival_rate = []
for i in range(3):
    still_active = df[[f"month_{j}" for j in range(i + 1)]].all(axis=1).sum()
    survival_rate.append(still_active / cohort_size)

print("retention_rate:", retention_rate)
print("churn_rate:", churn_rate)
print("survival_rate:", survival_rate)

Deze code berekent de benodigde statistieken voor elke maand. Het **retentiepercentage** geeft aan welk deel van het oorspronkelijke cohort in een bepaalde maand actief is. Het **churn-percentage** is simpelweg één min het retentiepercentage en geeft het aandeel weer dat niet langer actief is. Het **overlevingspercentage** kijkt naar gebruikers die continu actief zijn gebleven vanaf het begin tot en met de huidige maand – hiervoor moet een gebruiker in elke maand tot nu toe een `1` hebben.


import unittest
import user_code
import ast
import re   
import importlib
import csv
import unittest
import pandas as pd
import importlib

class TestTask(unittest.TestCase):
    def test_basic_case(self):
        import user_code
        importlib.reload(user_code)
        data = {
            "user_id": [1, 2, 3, 4, 5],
            "month_0": [1, 1, 1, 1, 1],
            "month_1": [1, 0, 1, 0, 1],
            "month_2": [1, 0, 0, 0, 0],
        }
        df = pd.DataFrame(data)
        func = getattr(user_code, 'calculate_cohort_metrics', None)
        _dynamic_test(self, callable(func), "Function is defined", "Function calculate_cohort_metrics is not defined")
        if not callable(func):
            return
        result = func(df)
        _dynamic_test(self, result is not None and isinstance(result, (tuple, list)) and len(result) == 3,
                      "Function returns three lists",
                      f"Function must return three lists, got: {result}")
        if result is None or not isinstance(result, (tuple, list)) or len(result) != 3:
            return
        retention_rate, churn_rate, survival_rate = result
        _dynamic_test(self, all(isinstance(x, list) for x in [retention_rate, churn_rate, survival_rate]),
                      "All return values are lists",
                      f"Returned values are not all lists: {result}")
        _dynamic_test(self, retention_rate == [1.0, 0.6, 0.2],
                      "Correct retention_rate for given data",
                      f"Expected retention_rate [1.0, 0.6, 0.2], got {retention_rate}")
        _dynamic_test(self, churn_rate == [0.0, 0.4, 0.8],
                      "Correct churn_rate for given data",
                      f"Expected churn_rate [0.0, 0.4, 0.8], got {churn_rate}")
        _dynamic_test(self, survival_rate == [1.0, 0.6, 0.2],
                      "Correct survival_rate for given data",
                      f"Expected survival_rate [1.0, 0.6, 0.2], got {survival_rate}")

    def test_survival_differs_from_retention(self):
        import user_code
        importlib.reload(user_code)
        # User 2 is inactive in month_1 but returns in month_2
        data = {
            "user_id": [1, 2, 3, 4],
            "month_0": [1, 1, 1, 1],
            "month_1": [1, 0, 1, 1],
            "month_2": [1, 1, 0, 1],
        }
        df = pd.DataFrame(data)
        func = getattr(user_code, 'calculate_cohort_metrics', None)
        result = func(df)
        retention_rate, churn_rate, survival_rate = result
        # retention: [1.0, 0.75, 0.75]
        # churn:    [0.0, 0.25, 0.25]
        # survival: [1.0, 0.75, 0.5] (only users 1 and 4 are active in all months)
        _dynamic_test(self, retention_rate == [1.0, 0.75, 0.75],
                      "Retention rate correct when users return after inactivity",
                      f"Expected [1.0, 0.75, 0.75], got {retention_rate}")
        _dynamic_test(self, churn_rate == [0.0, 0.25, 0.25],
                      "Churn rate correct when users return after inactivity",
                      f"Expected [0.0, 0.25, 0.25], got {churn_rate}")
        _dynamic_test(self, survival_rate == [1.0, 0.75, 0.5],
                      "Survival rate correct when users return after inactivity",
                      f"Expected [1.0, 0.75, 0.5], got {survival_rate}")

    def test_all_churned(self):
        import user_code
        importlib.reload(user_code)
        data = {
            "user_id": [1, 2, 3],
            "month_0": [1, 1, 1],
            "month_1": [0, 0, 0],
            "month_2": [0, 0, 0],
        }
        df = pd.DataFrame(data)
        func = getattr(user_code, 'calculate_cohort_metrics', None)
        result = func(df)
        retention_rate, churn_rate, survival_rate = result
        _dynamic_test(self, retention_rate == [1.0, 0.0, 0.0],
                      "Retention rate is zero after all churn",
                      f"Expected [1.0, 0.0, 0.0], got {retention_rate}")
        _dynamic_test(self, churn_rate == [0.0, 1.0, 1.0],
                      "Churn rate is one after all churn",
                      f"Expected [0.0, 1.0, 1.0], got {churn_rate}")
        _dynamic_test(self, survival_rate == [1.0, 0.0, 0.0],
                      "Survival rate is zero after all churn",
                      f"Expected [1.0, 0.0, 0.0], got {survival_rate}")

    def test_single_user(self):
        import user_code
        importlib.reload(user_code)
        data = {
            "user_id": [42],
            "month_0": [1],
            "month_1": [1],
            "month_2": [0],
        }
        df = pd.DataFrame(data)
        func = getattr(user_code, 'calculate_cohort_metrics', None)
        result = func(df)
        retention_rate, churn_rate, survival_rate = result
        _dynamic_test(self, retention_rate == [1.0, 1.0, 0.0],
                      "Retention rate correct for single user",
                      f"Expected [1.0, 1.0, 0.0], got {retention_rate}")
        _dynamic_test(self, churn_rate == [0.0, 0.0, 1.0],
                      "Churn rate correct for single user",
                      f"Expected [0.0, 0.0, 1.0], got {churn_rate}")
        _dynamic_test(self, survival_rate == [1.0, 1.0, 0.0],
                      "Survival rate correct for single user",
                      f"Expected [1.0, 1.0, 0.0], got {survival_rate}")

    def test_month_column_order(self):
        import user_code
        importlib.reload(user_code)
        data = {
            "user_id": [1,2,3],
            "month_2": [1,0,1],
            "month_0": [1,1,1],
            "month_1": [1,0,1],
        }
        df = pd.DataFrame(data)[["user_id", "month_2", "month_0", "month_1"]]  # out of order
        func = getattr(user_code, 'calculate_cohort_metrics', None)
        result = func(df)
        # Should sort columns by month number
        # month_0: [1,1,1] sum=3
        # month_1: [1,0,1] sum=2
        # month_2: [1,0,1] sum=2
        retention_rate, churn_rate, survival_rate = result
        _dynamic_test(self, retention_rate == [1.0, 0.6666666666666666, 0.6666666666666666],
                      "Handles month columns out of order",
                      f"Expected [1.0, 0.666..., 0.666...], got {retention_rate}")
        _dynamic_test(self, churn_rate == [0.0, 0.33333333333333337, 0.33333333333333337],
                      "Handles churn for out of order columns",
                      f"Expected [0.0, 0.333..., 0.333...], got {churn_rate}")
        _dynamic_test(self, survival_rate == [1.0, 0.6666666666666666, 0.6666666666666666],
                      "Handles survival for out of order columns",
                      f"Expected [1.0, 0.666..., 0.666...], got {survival_rate}")

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()


test_main.py

Beheers cohortanalyse-technieken met behulp van Python. Deze cursus behandelt het structureren van cohortgegevens, geavanceerde segmentatie, retentie-analyses en toepasbare zakelijke inzichten, met praktische uitdagingen in elke sectie.

Leer hoe u gegevens kunt structureren, preprocessen en voorbereiden voor geavanceerde cohortanalyse, inclusief cohorttoewijzing en feature engineering.

Duik in geavanceerde segmentatiestrategieën en berekeningen van retentiemetingen om diepgaandere inzichten uit cohortgegevens te verkrijgen.

Visualiseer cohortgegevens en haal bruikbare zakelijke inzichten uit met behulp van geavanceerde plot- en interpretatietechnieken.

Challenge: Het in Kaart Brengen van de Levenscyclus van een Spacestream-Verkenner

Oplossing