This video introduces multi-dimensional cohort segmentation, explaining how you can analyze user behavior across more than one attribute—such as combining signup month with acquisition channel or geography. You'll see why adding extra dimensions uncovers deeper business insights, and you'll walk through a practical example using Python and pandas to segment users by both when they signed up and how they were acquired. Real-world cases show how companies use this approach to discover patterns that single-dimensional analysis might miss. Each code sample in video should include no more than 10 lines of code.

Multi-dimensional cohort segmentation lets you group users by more than one attribute, such as both the month they signed up and the channel through which they were acquired. While traditional cohort analysis might focus on a single factor - like signup date - multi-dimensional segmentation helps you answer more complex questions. For example, you could see if users from a specific marketing campaign in a certain month behave differently than those from another channel or region. This approach is valuable for businesses because it highlights patterns and trends that are not visible when analyzing only one dimension. By segmenting cohorts using multiple factors, you can tailor marketing strategies, improve customer retention, and allocate resources more effectively.

import pandas as pd

# Sample data
data = {
    "user_id": [1, 2, 3, 4, 5, 6],
    "signup_date": [
        "2023-01-15", "2023-01-15", "2023-02-10",
        "2023-02-15", "2023-01-25", "2023-02-18"
    ],
    "acquisition_channel": [
        "Email", "Email", "Social", "Ad", "Ad", "Social"
    ]
}

df = pd.DataFrame(data)
df["signup_month"] = pd.to_datetime(df["signup_date"]).dt.to_period("M")

# Multi-dimensional cohort segmentation by signup_month and acquisition_channel
cohorts = df.groupby(["signup_month", "acquisition_channel"])["user_id"].nunique().reset_index()
cohorts = cohorts.rename(columns={"user_id": "num_users"})

print(cohorts)

By segmenting cohorts using both `signup_month` and `acquisition_channel`, you can spot hidden trends that single-dimensional analysis might miss. For instance, you might find that users acquired via **"Email"** in January are more engaged or have higher retention than those acquired via **"Ad"** in the same month. This level of detail allows you to make data-driven decisions about where to invest your marketing budget, how to personalize onboarding experiences, and which channels yield the most valuable customers. **Multi-dimensional segmentation** is a powerful tool for uncovering insights that drive business growth.


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

class TestTask(unittest.TestCase):
    def setUp(self):
        self.expected = pd.DataFrame({
            'signup_month': [pd.Period('2023-01'), pd.Period('2023-01'), pd.Period('2023-01'), pd.Period('2023-02'), pd.Period('2023-02'), pd.Period('2023-02')],
            'acquisition_channel': ['Ad', 'Email', 'Social', 'Ad', 'Email', 'Social'],
            'num_users': [1, 1, 1, 1, 1, 1]
        })
        # Sort for comparison
        self.expected = self.expected.sort_values(['signup_month', 'acquisition_channel']).reset_index(drop=True)

    def test_cohorts_exists_and_type(self):
        import user_code
        importlib.reload(user_code)
        cohorts = getattr(user_code, 'cohorts', None)
        _dynamic_test(
            self,
            cohorts is not None and isinstance(cohorts, pd.DataFrame),
            "The 'cohorts' DataFrame exists and is a pandas DataFrame.",
            "The 'cohorts' DataFrame is missing or not a pandas DataFrame."
        )

    def test_cohorts_columns(self):
        import user_code
        importlib.reload(user_code)
        cohorts = getattr(user_code, 'cohorts', None)
        expected_cols = ['signup_month', 'acquisition_channel', 'num_users']
        _dynamic_test(
            self,
            cohorts is not None and list(cohorts.columns) == expected_cols,
            f"The 'cohorts' DataFrame has correct columns: {expected_cols}.",
            f"The 'cohorts' DataFrame columns are incorrect. Got: {list(cohorts.columns)}"
        )

    def test_cohorts_values(self):
        import user_code
        importlib.reload(user_code)
        cohorts = getattr(user_code, 'cohorts', None)
        # Sort both DataFrames for comparison
        if cohorts is not None:
            cohorts_sorted = cohorts.sort_values(['signup_month', 'acquisition_channel']).reset_index(drop=True)
            # Compare values
            same_shape = cohorts_sorted.shape == self.expected.shape
            same_cols = list(cohorts_sorted.columns) == list(self.expected.columns)
            same_values = cohorts_sorted.equals(self.expected)
            _dynamic_test(
                self,
                same_shape and same_cols and same_values,
                "The 'cohorts' DataFrame has the correct grouped values.",
                f"The grouped values in 'cohorts' are incorrect.\\nExpected:\\n{self.expected}\\nGot:\\n{cohorts_sorted}"
            )
        else:
            _dynamic_test(
                self,
                False,
                "The 'cohorts' DataFrame has the correct grouped values.",
                "The 'cohorts' DataFrame is missing."
            )

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


Multi-Dimensional Cohort Segmentation

解答