This video provides a walkthrough of the challenge requirements for preparing a dataset for cohort analysis. You will see how to preprocess a sample transactional dataset, assign users to cohorts based on their first purchase month, and engineer key features such as cohort index and order counts. The video covers strategies for handling raw data, creating cohort labels, and generating new columns to support downstream cohort analysis.

Anwendung des Wissens aus den vorherigen Kapiteln zu Kohortenzuordnung und Feature Engineering. Aufgabe: Vorbereitung eines Beispiel-Transaktionsdatensatzes für die Kohortenanalyse.




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 setUp(self):
        self.data = {
            'user_id': [1, 1, 2, 2, 2, 3],
            'order_id': [101, 102, 201, 202, 203, 301],
            'order_date': [
                '2024-01-15', '2024-02-10',
                '2024-01-20', '2024-03-05', '2024-03-25',
                '2024-02-01'
            ]
        }
        self.df = pd.DataFrame(self.data)

    def test_cohort_month_column(self):
        import user_code
        importlib.reload(user_code)
        result = user_code.prepare_cohort_data(self.df.copy())
        _dynamic_test(
            self,
            'cohort_month' in result.columns,
            "DataFrame includes 'cohort_month' column.",
            "Missing 'cohort_month' column in DataFrame."
        )
        # Check correct cohort_month values for each user
        for user in result['user_id'].unique():
            user_rows = result[result['user_id'] == user]
            cohort_months = user_rows['cohort_month'].unique()
            _dynamic_test(
                self,
                len(cohort_months) == 1,
                f"All transactions for user {user} have same cohort_month.",
                f"User {user} has multiple cohort_month values: {cohort_months}"
            )
            first_order = pd.to_datetime(user_rows['order_date']).min()
            expected = first_order.strftime('%Y-%m')
            _dynamic_test(
                self,
                cohort_months[0] == expected,
                f"cohort_month for user {user} is correct.",
                f"Expected cohort_month {expected} for user {user}, got {cohort_months[0]}"
            )

    def test_cohort_index_column(self):
        import user_code
        importlib.reload(user_code)
        result = user_code.prepare_cohort_data(self.df.copy())
        _dynamic_test(
            self,
            'cohort_index' in result.columns,
            "DataFrame includes 'cohort_index' column.",
            "Missing 'cohort_index' column in DataFrame."
        )
        # cohort_index for first order of each user must be 0
        for user in result['user_id'].unique():
            user_rows = result[result['user_id'] == user].sort_values('order_date')
            cohort_index_first = user_rows.iloc[0]['cohort_index']
            _dynamic_test(
                self,
                cohort_index_first == 0,
                f"cohort_index for first order of user {user} is 0.",
                f"Expected cohort_index 0 for first order of user {user}, got {cohort_index_first}"
            )
            # Check cohort_index increments by 1 per month
            cohort_month = pd.to_datetime(user_rows.iloc[0]['order_date']).to_period('M')
            for idx, row in user_rows.iterrows():
                order_month = pd.to_datetime(row['order_date']).to_period('M')
                expected_index = (order_month - cohort_month).n
                _dynamic_test(
                    self,
                    row['cohort_index'] == expected_index,
                    f"cohort_index for user {user} order {row['order_id']} is correct.",
                    f"Expected cohort_index {expected_index} for user {user}, got {row['cohort_index']}"
                )

    def test_total_orders_column(self):
        import user_code
        importlib.reload(user_code)
        result = user_code.prepare_cohort_data(self.df.copy())
        _dynamic_test(
            self,
            'total_orders' in result.columns,
            "DataFrame includes 'total_orders' column.",
            "Missing 'total_orders' column in DataFrame."
        )
        for user in result['user_id'].unique():
            user_rows = result[result['user_id'] == user].sort_values('order_date')
            for idx, (i, row) in enumerate(user_rows.iterrows()):
                expected = idx + 1
                _dynamic_test(
                    self,
                    row['total_orders'] == expected,
                    f"total_orders for user {user} order {row['order_id']} is correct.",
                    f"Expected total_orders {expected} for user {user}, got {row['total_orders']}"
                )

    def test_columns_and_shape(self):
        import user_code
        importlib.reload(user_code)
        result = user_code.prepare_cohort_data(self.df.copy())
        expected_cols = set(list(self.df.columns) + ['cohort_month', 'cohort_index', 'total_orders'])
        _dynamic_test(
            self,
            set(result.columns) == expected_cols,
            "Returned DataFrame has all required columns.",
            f"Expected columns {expected_cols}, got {set(result.columns)}"
        )
        _dynamic_test(
            self,
            len(result) == len(self.df),
            "Returned DataFrame has same number of rows as input.",
            f"Expected {len(self.df)} rows, got {len(result)}"
        )

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

Beherrschung von Kohortenanalysetechniken mit Python. Dieser Kurs behandelt die Strukturierung von Kohortendaten, fortgeschrittene Segmentierung, Retentionsanalysen und umsetzbare Geschäftseinblicke mit praxisnahen Herausforderungen in jedem Abschnitt.

Erfahren Sie, wie Sie Daten für fortgeschrittene Kohortenanalysen strukturieren, vorverarbeiten und vorbereiten, einschließlich Kohortenzuordnung und Feature Engineering.

Tauchen Sie ein in fortgeschrittene Segmentierungsstrategien und die Berechnung von Retentionsmetriken, um tiefere Einblicke aus Kohortendaten zu gewinnen.

Visualisierung von Kohortendaten und Gewinnung umsetzbarer geschäftlicher Erkenntnisse mithilfe fortgeschrittener Visualisierungs- und Interpretationstechniken.

Challenge: Vorbereitung von Kohortendaten

Lösung