feat(nimbus): split retention results into fixed weekly metrics

experimenter/experimenter/experiments/constants.py

@@ -692,7 +692,8 @@ class AnalysisWindow(models.TextChoices):
 
     DAILY_ACTIVE_USERS = "client_level_daily_active_users_v2"
     DAYS_OF_USE = "days_of_use"
-    RETENTION = "retained"
+    RETENTION_WEEK_2 = "week_2_retention"
+    RETENTION_WEEK_4 = "week_4_retention"
     RETENTION_3_DAYS = "active_in_last_3_days"
     RETENTION_3_DAYS_DESKTOP = "active_in_last_3_days_legacy"
     SEARCH_COUNT = "search_count"
@@ -710,8 +711,17 @@ class AnalysisWindow(models.TextChoices):
     KPI_METRICS = [
         {
             "group": "other_metrics",
-            "slug": RETENTION,
+            "friendly_name": "Week 2 Retention",
+            "slug": RETENTION_WEEK_2,
             "display_type": "percentage",
+            "description": "Users who were active in Firefox during the second week after enrollment.",  # noqa
+        },
+        {
+            "group": "other_metrics",
+            "friendly_name": "Week 4 Retention",
+            "slug": RETENTION_WEEK_4,
+            "display_type": "percentage",
+            "description": "Users who were active in Firefox during the fourth week after enrollment.",  # noqa
         },
         {
             "group": "other_metrics",

experimenter/experimenter/experiments/migrations/0339_force_weekly_retention_results_refetch.py

@@ -0,0 +1,38 @@
+from django.db import migrations
+
+BATCH_SIZE = 100
+
+
+def stored_analysis_start_time(results_data):
+    metadata = (results_data.get("v3") or {}).get("metadata")
+    return metadata.get("analysis_start_time") if isinstance(metadata, dict) else None
+
+
+def force_results_refetch(apps, schema_editor):
+    NimbusExperiment = apps.get_model("experiments", "NimbusExperiment")
+
+    stale_experiments = []
+    for experiment in NimbusExperiment.objects.exclude(results_data=None).iterator(
+        chunk_size=BATCH_SIZE
+    ):
+        if stored_analysis_start_time(experiment.results_data) is None:
+            continue
+
+        experiment.results_data["v3"]["metadata"]["analysis_start_time"] = None
+        stale_experiments.append(experiment)
+
+        if len(stale_experiments) == BATCH_SIZE:
+            NimbusExperiment.objects.bulk_update(stale_experiments, ["results_data"])
+            stale_experiments.clear()
+
+    NimbusExperiment.objects.bulk_update(stale_experiments, ["results_data"])
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("experiments", "0338_nimbusemail_rollout_phase_alter_nimbusemail_type"),
+    ]
+
+    operations = [
+        migrations.RunPython(force_results_refetch, migrations.RunPython.noop),
+    ]

experimenter/experimenter/experiments/tests/test_migrations.py

@@ -97,3 +97,83 @@ def test_migration(self):
         self.assertEqual(non_rollout.status, "Live")
         self.assertEqual(non_rollout.status_next, "Live")
         self.assertEqual(non_rollout.publish_status, "Review")
+
+
+class TestForceWeeklyRetentionResultsRefetchMigration(MigratorTestCase):
+    migrate_from = (
+        "experiments",
+        "0338_nimbusemail_rollout_phase_alter_nimbusemail_type",
+    )
+    migrate_to = (
+        "experiments",
+        "0339_force_weekly_retention_results_refetch",
+    )
+
+    def prepare(self):
+        User = self.old_state.apps.get_model("auth", "User")
+        NimbusExperiment = self.old_state.apps.get_model(
+            "experiments", "NimbusExperiment"
+        )
+
+        owner, _ = User.objects.get_or_create(
+            username="test@example.com",
+            defaults={"email": "test@example.com"},
+        )
+
+        NimbusExperiment.objects.create(
+            slug="with-analysis-start-time",
+            name="With analysis start time",
+            application="firefox-desktop",
+            owner=owner,
+            results_data={
+                "v3": {
+                    "metadata": {
+                        "analysis_start_time": "2026-07-01T00:00:00+00:00",
+                        "outcomes": {},
+                    },
+                    "overall": {},
+                }
+            },
+        )
+        NimbusExperiment.objects.create(
+            slug="without-metadata",
+            name="Without metadata",
+            application="firefox-desktop",
+            owner=owner,
+            results_data={"v3": {"overall": {}}},
+        )
+        NimbusExperiment.objects.create(
+            slug="null-metadata",
+            name="Null metadata",
+            application="firefox-desktop",
+            owner=owner,
+            results_data={"v3": {"metadata": None, "overall": {}}},
+        )
+        NimbusExperiment.objects.create(
+            slug="no-results",
+            name="No results",
+            application="firefox-desktop",
+            owner=owner,
+            results_data=None,
+        )
+
+    def test_migration(self):
+        NimbusExperiment = self.new_state.apps.get_model(
+            "experiments", "NimbusExperiment"
+        )
+
+        cleared = NimbusExperiment.objects.get(slug="with-analysis-start-time")
+        self.assertIsNone(cleared.results_data["v3"]["metadata"]["analysis_start_time"])
+        self.assertEqual(cleared.results_data["v3"]["metadata"]["outcomes"], {})
+        self.assertEqual(cleared.results_data["v3"]["overall"], {})
+
+        untouched = NimbusExperiment.objects.get(slug="without-metadata")
+        self.assertEqual(untouched.results_data, {"v3": {"overall": {}}})
+
+        null_metadata = NimbusExperiment.objects.get(slug="null-metadata")
+        self.assertEqual(
+            null_metadata.results_data, {"v3": {"metadata": None, "overall": {}}}
+        )
+
+        no_results = NimbusExperiment.objects.get(slug="no-results")
+        self.assertIsNone(no_results.results_data)

experimenter/experimenter/jetstream/client.py

@@ -346,6 +346,12 @@ def get_experiment_data(experiment: NimbusExperiment):
             raw_segment_data = JetstreamData(segment_data)
             raw_data[window][AnalysisBasis.ENROLLMENTS][segment] = raw_segment_data
             data = raw_segment_data.model_copy(deep=True)
+            if data:
+                data.separate_weekly_retention_data(
+                    raw_data.get(AnalysisWindow.WEEKLY, {})
+                    .get(AnalysisBasis.ENROLLMENTS, {})
+                    .get(segment)
+                )
             (
                 result_metrics,
                 primary_metrics_set,
@@ -363,36 +369,6 @@ def get_experiment_data(experiment: NimbusExperiment):
             if data and window == AnalysisWindow.OVERALL:
                 # Append some values onto the incoming Jetstream data
                 data.append_population_percentages()
-                weekly_data = (
-                    raw_data.get(AnalysisWindow.WEEKLY, {})
-                    .get(AnalysisBasis.ENROLLMENTS, {})
-                    .get(segment)
-                )
-                week_2_retention = data.get_retention_by_window(
-                    2,
-                    weekly_data,
-                    Metric.RETENTION,
-                )
-                has_retention = any(
-                    point.metric == Metric.RETENTION for point in weekly_data or []
-                )
-
-                if has_retention and not week_2_retention and segment == Segment.ALL:
-                    runtime_errors.append(
-                        AnalysisError(
-                            experiment=experiment.slug,
-                            filename="experimenter/jetstream/client.py",
-                            func_name="get_experiment_data",
-                            log_level="WARNING",
-                            message=(
-                                "Week 2 retention is unavailable because this "
-                                "experiment did not run long enough."
-                            ),
-                            metric=Metric.RETENTION,
-                            timestamp=timezone.now(),
-                        )
-                    )
-                data.extend(week_2_retention)
                 # Append 3-day retention from daily data
                 data.append_retention_3_days(
                     raw_data.get(AnalysisWindow.DAILY, {})
@@ -406,12 +382,6 @@ def get_experiment_data(experiment: NimbusExperiment):
                 data.append_conversion_count(primary_metrics_set)
 
             elif data and window == AnalysisWindow.WEEKLY:
-                data.replace_retention_weeks(
-                    raw_data.get(AnalysisWindow.WEEKLY, {})
-                    .get(AnalysisBasis.ENROLLMENTS, {})
-                    .get(segment)
-                )
-
                 # Append 3-day retention from daily data
                 data.append_retention_3_days(
                     raw_data.get(AnalysisWindow.DAILY, {})
@@ -430,6 +400,7 @@ def get_experiment_data(experiment: NimbusExperiment):
                     .get(AnalysisBasis.ENROLLMENTS, {})
                     .get(segment)
                 )
+
                 ResultsObjectModel = create_results_object_model(data)
 
                 data = ResultsObjectModel(result_metrics, data, experiment, window)
@@ -442,6 +413,12 @@ def get_experiment_data(experiment: NimbusExperiment):
             raw_segment_data = JetstreamData(segment_data)
             raw_data[window][AnalysisBasis.EXPOSURES][segment] = raw_segment_data
             data = raw_segment_data.model_copy(deep=True)
+            if data:
+                data.separate_weekly_retention_data(
+                    raw_data.get(AnalysisWindow.WEEKLY, {})
+                    .get(AnalysisBasis.EXPOSURES, {})
+                    .get(segment)
+                )
             (
                 result_metrics,
                 primary_metrics_set,
@@ -459,11 +436,6 @@ def get_experiment_data(experiment: NimbusExperiment):
             if data and window == AnalysisWindow.OVERALL:
                 # Append some values onto Jetstream data
                 data.append_population_percentages()
-                data.append_retention_data(
-                    raw_data.get(AnalysisWindow.WEEKLY, {})
-                    .get(AnalysisBasis.EXPOSURES, {})
-                    .get(segment)
-                )
                 # Append 3-day retention from daily data
                 data.append_retention_3_days(
                     raw_data.get(AnalysisWindow.DAILY, {})
@@ -477,12 +449,6 @@ def get_experiment_data(experiment: NimbusExperiment):
                 data.append_conversion_count(primary_metrics_set)
 
             elif data and window == AnalysisWindow.WEEKLY:
-                data.replace_retention_weeks(
-                    raw_data.get(AnalysisWindow.WEEKLY, {})
-                    .get(AnalysisBasis.EXPOSURES, {})
-                    .get(segment)
-                )
-
                 # Append 3-day retention from daily data
                 data.append_retention_3_days(
                     raw_data.get(AnalysisWindow.DAILY, {})
@@ -539,17 +505,14 @@ def include_error_after_analysis_start(err):
                     errors_experiment_overall.append(err)
 
     for e in runtime_errors:
-        if isinstance(e, AnalysisError):
-            analysis_error = e
-        else:
-            analysis_error = AnalysisError(
-                experiment=experiment.slug,
-                filename="experimenter/jetstream/client.py",
-                func_name="load_data_from_gcs",
-                log_level="WARNING",
-                message=e,
-                timestamp=timezone.now(),
-            )
+        analysis_error = AnalysisError(
+            experiment=experiment.slug,
+            filename="experimenter/jetstream/client.py",
+            func_name="load_data_from_gcs",
+            log_level="WARNING",
+            message=e,
+            timestamp=timezone.now(),
+        )
         errors_experiment_overall.append(analysis_error.model_dump())
 
     errors_by_metric["experiment"] = errors_experiment_overall

experimenter/experimenter/jetstream/models.py

@@ -28,6 +28,7 @@ class BranchComparison(StrEnum):
 
 class Metric(StrEnum):
     RETENTION = "retained"
+    WEEKLY_RETENTION = "week_{}_retention"
     RETENTION_3_DAYS = "active_in_last_3_days"
     RETENTION_3_DAYS_LEGACY = "active_in_last_3_days_legacy"
     SEARCH = "search_count"
@@ -78,7 +79,6 @@ class Group(StrEnum):
     Group.SEARCH: SEARCH_METRICS,
     Group.USAGE: USAGE_METRICS,
 }
-RETENTION_2_WEEKS_WINDOW_INDEX = 2
 RETENTION_3_DAYS_WINDOW_INDEX = 4
 RETENTION_3_DAYS_METRICS = (Metric.RETENTION_3_DAYS, Metric.RETENTION_3_DAYS_LEGACY)
 
@@ -154,29 +154,34 @@ def get_retention_by_window(self, window_index, data, metric):
             jetstream_data_point
             for jetstream_data_point in data
             if jetstream_data_point.window_index == str(window_index)
-            and jetstream_data_point.metric == metric.value
+            and jetstream_data_point.metric == metric
         ]
 
-    def append_retention_data(self, weekly_data):
-        # Only use two-week retention data.
-        retention_data = self.get_retention_by_window(
-            RETENTION_2_WEEKS_WINDOW_INDEX, weekly_data, Metric.RETENTION
-        )
+    def separate_weekly_retention_data(self, weekly_data):
+        # Replace "retained" with one week_N_retention metric per week in "weekly_data",
+        # skipping week 1. Points are copied so the shared "weekly_data" is untouched.
+        retention_data = []
 
-        self.extend(retention_data)
+        for jetstream_data_point in weekly_data or []:
+            if (
+                jetstream_data_point.metric == Metric.RETENTION
+                and jetstream_data_point.window_index != "1"
+            ):
+                retention_data_point = jetstream_data_point.model_copy()
+                retention_data_point.metric = Metric.WEEKLY_RETENTION.format(
+                    retention_data_point.window_index
+                )
+                retention_data.append(retention_data_point)
 
-    def replace_retention_weeks(self, weekly_data):
-        # Remove all weekly retention data except for week 2.
-        retention_data = self.get_retention_by_window(
-            RETENTION_2_WEEKS_WINDOW_INDEX, weekly_data, Metric.RETENTION
-        )
+        self.remove_retention_data()
+        self.extend(retention_data)
 
+    def remove_retention_data(self):
         self.root = [
             jetstream_data_point
             for jetstream_data_point in self.root
             if jetstream_data_point.metric != Metric.RETENTION
         ]
-        self.extend(retention_data)
 
     def get_retention_3_days_by_window(self, window_index, daily_data):
         retention_data = []
@@ -306,11 +311,10 @@ def __init__(
 
                 # Need window index for weekly DataPoint objects and for storing
                 # significance for each window. Overall should always be 1 because
-                # there is only ever one overall window, except retained data which
-                # is pulled from week 2.
+                # there is only ever one overall window.
                 window_index = (
                     "1"
-                    if window == AnalysisWindow.OVERALL and metric != Metric.RETENTION
+                    if window == AnalysisWindow.OVERALL
                     else jetstream_data_point.window_index
                 )