In this challenge, you will analyze user feature usage metrics alongside retention data to pinpoint which features are most predictive of user retention. Your goal is to calculate the correlation between each feature's usage and user retention, identify the top three features with the strongest positive correlation, visualize these relationships, and recommend targeted interventions to boost engagement with these retention-driving features. This process will help you translate raw data into actionable product insights that can drive meaningful user engagement improvements.

When you identify features highly correlated with retention, consider interventions such as: enhancing onboarding flows to highlight these features, adding contextual prompts or tooltips to encourage their use, or designing targeted in-app campaigns to showcase their value. Tailoring interventions to the specific strengths of these features can maximize their impact on user retention.

Note


import unittest
import user_code
import ast
import re   
import importlib
import csv
import unittest
import importlib
import pandas as pd
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt

class TestTask(unittest.TestCase):
    def setUp(self):
        self.df = pd.DataFrame({
            'feature_A': [5, 2, 3, 7, 4, 6, 1],
            'feature_B': [1, 4, 2, 2, 3, 3, 5],
            'feature_C': [0, 1, 0, 1, 1, 0, 1],
            'feature_D': [8, 2, 5, 7, 6, 4, 3],
            'retained':   [1, 0, 0, 1, 1, 1, 0]
        })
        self.feature_cols = ['feature_A', 'feature_B', 'feature_C', 'feature_D']
        self.retention_col = 'retained'
        import user_code
        importlib.reload(user_code)
        self.user_code = user_code

    def test_correlation_computation(self):
        # Check that correlation is computed for all features
        result = self.user_code.analyze_retention_drivers(self.df, self.feature_cols, self.retention_col)
        _dynamic_test(
            self,
            result is not None and len(result) == 3,
            "Function returns a list of three tuples.",
            f"Expected 3 recommendations, got: {result}"
        )
        # Calculate correct correlations
        expected_corrs = {
            f: self.df[f].corr(self.df[self.retention_col]) for f in self.feature_cols
        }
        actual_features = [t[0] for t in result]
        actual_corrs = [t[1] for t in result]
        # Check that all returned features are from feature_cols
        _dynamic_test(
            self,
            all(f in self.feature_cols for f in actual_features),
            "All returned features are in feature_cols.",
            f"Returned features: {actual_features}"
        )
        # Check that correlation values match
        for i, f in enumerate(actual_features):
            _dynamic_test(
                self,
                abs(actual_corrs[i] - expected_corrs[f]) < 1e-6,
                f"Correlation for '{f}' is correct.",
                f"Expected correlation {expected_corrs[f]}, got {actual_corrs[i]} for {f}"
            )

    def test_top_three_features_selected(self):
        # Check that the top 3 features by absolute correlation are selected
        result = self.user_code.analyze_retention_drivers(self.df, self.feature_cols, self.retention_col)
        expected_corrs = {
            f: self.df[f].corr(self.df[self.retention_col]) for f in self.feature_cols
        }
        sorted_features = sorted(expected_corrs.items(), key=lambda x: abs(x[1]), reverse=True)
        expected_top = [f[0] for f in sorted_features[:3]]
        actual_top = [t[0] for t in result]
        _dynamic_test(
            self,
            actual_top == expected_top,
            "Top 3 features by absolute correlation are selected in order.",
            f"Expected top 3: {expected_top}, got: {actual_top}"
        )

    def test_recommendation_logic(self):
        # Check that recommendations match the thresholds
        result = self.user_code.analyze_retention_drivers(self.df, self.feature_cols, self.retention_col)
        for feature, corr, rec in result:
            if corr > 0.5:
                _dynamic_test(
                    self,
                    "onboarding" in rec and "prompt" in rec,
                    f"Recommendation for {feature} is correct for corr={corr}",
                    f"Expected onboarding/prompt for corr={corr}: {rec}"
                )
            elif corr > 0.3:
                _dynamic_test(
                    self,
                    "contextual" in rec and "tip" in rec,
                    f"Recommendation for {feature} is correct for corr={corr}",
                    f"Expected contextual tips for corr={corr}: {rec}"
                )
            else:
                _dynamic_test(
                    self,
                    "campaign" in rec,
                    f"Recommendation for {feature} is correct for corr={corr}",
                    f"Expected campaign for corr={corr}: {rec}"
                )

    def test_return_format(self):
        # Check the return format is a list of 3 tuples (feature, corr, rec)
        result = self.user_code.analyze_retention_drivers(self.df, self.feature_cols, self.retention_col)
        _dynamic_test(
            self,
            isinstance(result, list) and len(result) == 3 and all(isinstance(t, tuple) and len(t) == 3 for t in result),
            "Return value is a list of 3 tuples (feature, corr, rec)",
            f"Return value format incorrect: {result}"
        )

    def test_visualization_created(self):
        # Check that a bar chart is created (by checking for figure after function call)
        plt.close('all')
        self.user_code.analyze_retention_drivers(self.df, self.feature_cols, self.retention_col)
        figs = [f for f in map(plt.figure, plt.get_fignums()) if f]
        _dynamic_test(
            self,
            len(figs) >= 1,
            "Bar chart is created.",
            "No bar chart was created by the function."
        )

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

Learn to analyze user interactions at the feature level, map adoption curves, detect high-value behavioral paths, and cluster users by product usage using Python and machine learning.

Dive deep into user event logs to map feature adoption, uncover retention-driving behaviors, and segment users by their actual product usage patterns using Python.

Challenge: Retention Driver Analysis

Solution