diff --git a/api/experimentation/dataclasses.py b/api/experimentation/dataclasses.py index 0c780a1529f8..c46f7cfc23ff 100644 --- a/api/experimentation/dataclasses.py +++ b/api/experimentation/dataclasses.py @@ -56,6 +56,25 @@ class ExposuresSummary: timeseries: ExposuresTimeseries +@dataclass(frozen=True) +class ConversionBucket: + variant: str + bucket: datetime + converted_identities: int + + +@dataclass(frozen=True) +class ConversionsTimeseriesPoint: + bucket: str + converted_identities: dict[str, int] + + +@dataclass(frozen=True) +class ConversionsTimeseries: + granularity: ExposureGranularity + points: list[ConversionsTimeseriesPoint] + + @dataclass(frozen=True) class MetricSpec: metric_id: int @@ -66,13 +85,18 @@ class MetricSpec: @dataclass(frozen=True) class ResultsAggregates: - """Sufficient statistics gathered from the warehouse for one experiment: - the specs they were computed from, per-variant identity counts, and per - metric the per-variant ``VariantStats``. Bundled so the keys can't drift.""" + """Everything one results refresh reads from the warehouse: the specs it + was computed from, per-variant identity counts, per metric the per-variant + ``VariantStats``, and the time-bucketed rows behind the over-time charts. + Bundled so the keys can't drift.""" specs: list[MetricSpec] exposure_counts: dict[str, int] metric_stats: dict[int, dict[str, VariantStats]] + granularity: ExposureGranularity + exposure_buckets: list[ExposureBucket] + # Keyed by metric id, one entry per charted metric. + conversion_buckets: dict[int, list[ConversionBucket]] @dataclass(frozen=True) @@ -80,12 +104,20 @@ class MetricResult: metric_id: int variants: dict[str, VariantStats] inference: dict[str, Inference | None] + # Only occurrence metrics chart a conversion rate; None for the rest. + conversions_timeseries: ConversionsTimeseries | None @dataclass(frozen=True) class ResultsSummary: srm_p_value: float | None metrics: list[MetricResult] + # Denominator for the conversion charts, computed in the same run as the + # metrics so both sides of the rate share one as_of. Exposures are bucketed + # by first exposure and conversions by first conversion, so only running + # totals divide: a bucket's own conversions over its own new identities + # compares different people and can exceed 100%. + exposures_timeseries: ExposuresTimeseries @dataclass(frozen=True) diff --git a/api/experimentation/results_query.py b/api/experimentation/results_query.py index df6d7a191d8d..430dd99da4dc 100644 --- a/api/experimentation/results_query.py +++ b/api/experimentation/results_query.py @@ -15,16 +15,18 @@ from collections.abc import Sequence from dataclasses import dataclass +from datetime import datetime from typing import Any -from experimentation.dataclasses import MetricSpec +from experimentation.constants import EXPOSURE_EVENT_NAME +from experimentation.dataclasses import ConversionBucket, MetricSpec from experimentation.models import MetricAggregation from experimentation.stats import VariantStats _FLOAT_VALUE = "toFloat64OrZero(m.value)" # Events are delivered at-least-once, so dedup keeps duplicates from inflating -# counts. Shared by the exposures and results queries. +# counts. Shared by the exposures, results and conversions queries. _EXPOSURES_CTE = """ WITH exposures AS ( SELECT @@ -50,10 +52,33 @@ GROUP BY variant""" ) -_METRIC_JOIN = """ LEFT JOIN events AS m + +def exposure_window_params( + *, + environment_key: str, + feature_name: str, + window_start: datetime, + window_end: datetime, +) -> dict[str, object]: + """The parameters ``_EXPOSURES_CTE`` binds, for every query that starts + from it.""" + return { + "environment_key": environment_key, + "exposure_event": EXPOSURE_EVENT_NAME, + "feature_name": feature_name, + "window_start": window_start, + "window_end": window_end, + } + + +def _metric_join(events_param: str) -> str: + """Join each exposed identity to its metric events. ``events_param`` names + the bound list of event names, so a query can join only the events it + aggregates.""" + return f""" LEFT JOIN events AS m ON m.identifier = e.identifier AND m.environment_key = %(environment_key)s - AND m.event IN %(metric_events)s + AND m.event IN %({events_param})s AND m.timestamp >= %(window_start)s AND m.timestamp < %(window_end)s""" @@ -104,6 +129,23 @@ def outer_select(self) -> str: a = self._alias return f"sum({a}) AS {a}_sum, sum({a} * {a}) AS {a}_sum_squares" + @property + def conversion_alias(self) -> str: + return f"c{self.index}" + + def first_conversion_select(self) -> str | None: + """Per-identity timestamp of the first post-exposure conversion, NULL + when the identity never converted. Same attribution condition as + unit_select, so bucket totals add up to the metric's ``sum``. + + None for value metrics: a count or sum accrues per event rather than + once per identity, so a first-conversion timestamp can't chart it.""" + if self.spec.aggregation != MetricAggregation.OCCURRENCE: + return None + return ( + f"minIfOrNull(m.timestamp, {self._condition()}) AS {self.conversion_alias}" + ) + def decode(self, n: int, row: Sequence[Any], index: dict[str, int]) -> VariantStats: """Read this slot's two columns (sum, sum_squares) from a row by name.""" return VariantStats( @@ -134,7 +176,7 @@ def build_query(self) -> str: e.variant AS variant, {unit_selects} FROM exposures AS e -{_METRIC_JOIN} +{_metric_join("metric_events")} WHERE e.quarantined = 0 GROUP BY e.identifier, e.variant ) @@ -144,13 +186,92 @@ def build_query(self) -> str: GROUP BY variant""" ) - def add_metric_params(self, params: dict[str, object]) -> None: - """Add per-metric query parameters into an existing params dict.""" + def build_conversions_query(self, *, bucket_function: str) -> str | None: + """Per variant and charted metric, how many identities first converted + in each time bucket. None when no attached metric charts, since there + is nothing to query.""" + slots = self._charted_slots + if not slots: + return None + + first_conversion_selects = ",\n ".join( + select for s in slots if (select := s.first_conversion_select()) + ) + indexes = ", ".join(str(s.index) for s in slots) + aliases = ", ".join(s.conversion_alias for s in slots) + + return ( + _EXPOSURES_CTE + + f""", +first_conversions AS ( + SELECT + e.variant AS variant, + {first_conversion_selects} + FROM exposures AS e +{_metric_join("conversion_events")} + WHERE e.quarantined = 0 + GROUP BY e.identifier, e.variant +) +SELECT + variant, + metric_index, + {bucket_function}(first_conversion, 'UTC') AS bucket, + count() AS converted_identities +FROM first_conversions +ARRAY JOIN [{indexes}] AS metric_index, [{aliases}] AS first_conversion +WHERE first_conversion IS NOT NULL +GROUP BY variant, metric_index, bucket +ORDER BY bucket""" + ) + + @property + def _charted_slots(self) -> list[_MetricSlot]: + return [s for s in self._slots if s.first_conversion_select() is not None] + + def params( + self, + *, + environment_key: str, + feature_name: str, + window_start: datetime, + window_end: datetime, + ) -> dict[str, object]: + """Every parameter the results and conversions queries bind: the + exposure window, each metric's event, and the event lists each join + narrows to.""" + params = exposure_window_params( + environment_key=environment_key, + feature_name=feature_name, + window_start=window_start, + window_end=window_end, + ) if not self._slots: - return + return params params["metric_events"] = [s.spec.event for s in self._slots] + params["conversion_events"] = [s.spec.event for s in self._charted_slots] for slot in self._slots: params[f"metric_{slot.index}_event"] = slot.spec.event + return params + + def decode_conversion_rows( + self, rows: Sequence[Sequence[Any]], column_names: Sequence[str] + ) -> dict[int, list[ConversionBucket]]: + """Group conversions-query rows by the metric behind each slot index. + Every charted metric gets a key, empty when nobody converted yet.""" + index = {name: position for position, name in enumerate(column_names)} + buckets: dict[int, list[ConversionBucket]] = { + slot.spec.metric_id: [] for slot in self._charted_slots + } + for row in rows: + metric_id = self._slots[int(row[index["metric_index"]])].spec.metric_id + buckets[metric_id].append( + ConversionBucket( + variant=str(row[index["variant"]]), + bucket=row[index["bucket"]], + converted_identities=int(row[index["converted_identities"]]), + ) + ) + return buckets def decode_rows( self, rows: list[Any], column_names: Sequence[str] diff --git a/api/experimentation/services.py b/api/experimentation/services.py index 8496f9c3feaa..058d7c6ea317 100644 --- a/api/experimentation/services.py +++ b/api/experimentation/services.py @@ -27,7 +27,6 @@ from experimentation.constants import ( CONTROL_VARIANT_KEY, EXPERIMENT_FLAG, - EXPOSURE_EVENT_NAME, EXPOSURE_HOURLY_BUCKET_MAX_WINDOW, RESULTS_MIN_CONVERSIONS_PER_VARIANT, RESULTS_MIN_IDENTITIES_PER_VARIANT, @@ -35,6 +34,9 @@ WAREHOUSE_CONNECTION_FLAG, ) from experimentation.dataclasses import ( + ConversionBucket, + ConversionsTimeseries, + ConversionsTimeseriesPoint, ExposureBucket, ExposuresSummary, ExposuresTimeseries, @@ -64,7 +66,11 @@ WarehouseDeliveryOutcome, WarehouseType, ) -from experimentation.results_query import _EXPOSURES_CTE, ResultsQueryBuilder +from experimentation.results_query import ( + _EXPOSURES_CTE, + ResultsQueryBuilder, + exposure_window_params, +) from experimentation.stats import ( Inference, VariantStats, @@ -347,27 +353,58 @@ def build_exposures_summary( excluded_identities=sum( b.first_exposed_identities for b in buckets if b.quarantined ), - timeseries=ExposuresTimeseries( - granularity=granularity, - points=_timeseries_points([b for b in buckets if not b.quarantined]), - ), + timeseries=_exposures_timeseries(buckets, granularity=granularity), ) -def _timeseries_points( +def _exposures_timeseries( buckets: Sequence[ExposureBucket], -) -> list[ExposuresTimeseriesPoint]: - new_identities_by_bucket: dict[datetime, dict[str, int]] = {} - for b in buckets: - new_identities_by_bucket.setdefault(b.bucket, {})[b.variant] = ( - b.first_exposed_identities - ) + *, + granularity: ExposureGranularity, +) -> ExposuresTimeseries: + return ExposuresTimeseries( + granularity=granularity, + points=[ + ExposuresTimeseriesPoint(bucket=bucket, new_identities=counts) + for bucket, counts in _counts_by_bucket( + (b.bucket, b.variant, b.first_exposed_identities) + for b in buckets + if not b.quarantined + ) + ], + ) + + +def _conversions_timeseries( + metric_id: int, + aggregates: ResultsAggregates, +) -> ConversionsTimeseries | None: + buckets = aggregates.conversion_buckets.get(metric_id) + if buckets is None: + return None + return ConversionsTimeseries( + granularity=aggregates.granularity, + points=[ + ConversionsTimeseriesPoint(bucket=bucket, converted_identities=counts) + for bucket, counts in _counts_by_bucket( + (b.bucket, b.variant, b.converted_identities) for b in buckets + ) + ], + ) + + +def _counts_by_bucket( + rows: typing.Iterable[tuple[datetime, str, int]], +) -> list[tuple[str, dict[str, int]]]: + """Group rows into one per-variant dict per bucket, in bucket order. The + bucket becomes an ISO string because the summary lands in a JSONField as + is, and datetimes wouldn't serialise.""" + counts_by_bucket: dict[datetime, dict[str, int]] = {} + for bucket, variant, count in rows: + counts_by_bucket.setdefault(bucket, {})[variant] = count return [ - ExposuresTimeseriesPoint( - bucket=bucket_start.isoformat(), - new_identities=new_identities_by_bucket[bucket_start], - ) - for bucket_start in sorted(new_identities_by_bucket) + (bucket.isoformat(), counts_by_bucket[bucket]) + for bucket in sorted(counts_by_bucket) ] @@ -394,13 +431,12 @@ def get_exposure_buckets( EXPOSURE_BUCKETS_QUERY.format( bucket_function=_EXPOSURE_BUCKET_FUNCTIONS[granularity] ), - { - "environment_key": environment_key, - "exposure_event": EXPOSURE_EVENT_NAME, - "feature_name": feature_name, - "window_start": window_start, - "window_end": window_end, - }, + exposure_window_params( + environment_key=environment_key, + feature_name=feature_name, + window_start=window_start, + window_end=window_end, + ), ) return [ ExposureBucket( @@ -413,37 +449,64 @@ def get_exposure_buckets( ] -def get_metric_variant_stats( +def get_results_aggregates( *, environment_key: str, feature_name: str, window_start: datetime, window_end: datetime, specs: Sequence[MetricSpec], + granularity: ExposureGranularity, ) -> ResultsAggregates: - """Run the warehouse query, returning per-variant identity counts and, per - metric, per-variant sufficient statistics.""" - builder = ResultsQueryBuilder(specs) - params: dict[str, object] = { - "environment_key": environment_key, - "exposure_event": EXPOSURE_EVENT_NAME, - "feature_name": feature_name, - "window_start": window_start, - "window_end": window_end, - } - builder.add_metric_params(params) + """Run the warehouse reads behind one results refresh: per-variant identity + counts and sufficient statistics, exposure buckets, and per charted metric + the buckets of first post-exposure conversions. - rows, columns = _get_clickhouse_client( + Three separate reads, so events landing mid-run can leave the chart's last + point a few identities off the table until the next refresh.""" + builder = ResultsQueryBuilder(specs) + params = builder.params( + environment_key=environment_key, + feature_name=feature_name, + window_start=window_start, + window_end=window_end, + ) + client = _get_clickhouse_client( send_receive_timeout=CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS, - ).execute(builder.build_query(), params, with_column_types=True) + ) + + rows, columns = client.execute( + builder.build_query(), params, with_column_types=True + ) exposure_counts, metric_stats = builder.decode_rows( rows, [name for name, _type in columns] ) + conversion_buckets: dict[int, list[ConversionBucket]] = {} + conversions_query = builder.build_conversions_query( + bucket_function=_EXPOSURE_BUCKET_FUNCTIONS[granularity] + ) + if conversions_query is not None: + rows, columns = client.execute( + conversions_query, params, with_column_types=True + ) + conversion_buckets = builder.decode_conversion_rows( + rows, [name for name, _type in columns] + ) + return ResultsAggregates( specs=list(specs), exposure_counts=exposure_counts, metric_stats=metric_stats, + granularity=granularity, + exposure_buckets=get_exposure_buckets( + environment_key=environment_key, + feature_name=feature_name, + window_start=window_start, + window_end=window_end, + granularity=granularity, + ), + conversion_buckets=conversion_buckets, ) @@ -470,9 +533,15 @@ def build_results_summary( inference=_metric_inference( spec, aggregates.metric_stats.get(spec.metric_id, {}) ), + conversions_timeseries=_conversions_timeseries( + spec.metric_id, aggregates + ), ) for spec in aggregates.specs ], + exposures_timeseries=_exposures_timeseries( + aggregates.exposure_buckets, granularity=aggregates.granularity + ), ) @@ -482,15 +551,15 @@ def compute_results_summary( window_start: "datetime", window_end: "datetime", ) -> ResultsSummary: - """Gather an experiment's metric statistics from the warehouse and reduce - them to the stored results payload.""" - specs = _experiment_metric_specs(experiment) - aggregates = get_metric_variant_stats( + """Gather an experiment's metric statistics and chart rows from the + warehouse and reduce them to the stored results payload.""" + aggregates = get_results_aggregates( environment_key=experiment.environment.api_key, feature_name=experiment.feature.name, window_start=window_start, window_end=window_end, - specs=specs, + specs=_experiment_metric_specs(experiment), + granularity=_select_exposure_granularity(window_start, window_end), ) return build_results_summary( aggregates, diff --git a/api/tests/unit/experimentation/test_models.py b/api/tests/unit/experimentation/test_models.py index 006fe325733d..805574ba2c10 100644 --- a/api/tests/unit/experimentation/test_models.py +++ b/api/tests/unit/experimentation/test_models.py @@ -169,8 +169,10 @@ def _results_summary() -> ResultsSummary: "control": VariantStats(n=1000, sum=100.0, sum_squares=100.0) }, inference={}, + conversions_timeseries=None, ) ], + exposures_timeseries=ExposuresTimeseries(granularity="day", points=[]), ) @@ -198,8 +200,10 @@ def test_experiment_results__record_refresh__stores_payload_and_clears_error( "control": {"n": 1000, "sum": 100.0, "sum_squares": 100.0} }, "inference": {}, + "conversions_timeseries": None, } ], + "exposures_timeseries": {"granularity": "day", "points": []}, } assert results.as_of == as_of assert results.last_error_at is None diff --git a/api/tests/unit/experimentation/test_services.py b/api/tests/unit/experimentation/test_services.py index e318751dd05e..4b4862249f2e 100644 --- a/api/tests/unit/experimentation/test_services.py +++ b/api/tests/unit/experimentation/test_services.py @@ -1,4 +1,4 @@ -from dataclasses import asdict +from dataclasses import asdict, replace from datetime import datetime, timezone from unittest.mock import MagicMock @@ -16,6 +16,9 @@ from environments.models import Environment from experimentation import services from experimentation.dataclasses import ( + ConversionBucket, + ConversionsTimeseries, + ConversionsTimeseriesPoint, ExposureBucket, ExposuresSummary, ExposuresTimeseries, @@ -38,7 +41,7 @@ WarehouseConnectionStatus, WarehouseType, ) -from experimentation.results_query import _MetricSlot +from experimentation.results_query import ResultsQueryBuilder, _MetricSlot from experimentation.services import ( annotate_warehouse_event_stats, verify_clickhouse_connection, @@ -796,6 +799,9 @@ def _aggregates( specs=specs, exposure_counts=exposure_counts, metric_stats=metric_stats, + granularity="day", + exposure_buckets=[], + conversion_buckets={}, ) @@ -809,6 +815,16 @@ def _result_columns(metric_count: int) -> list[tuple[str, str]]: return columns +def _conversion_columns() -> list[tuple[str, str]]: + """Column metadata for the conversions query, in SELECT order.""" + return [ + ("variant", "String"), + ("metric_index", "UInt8"), + ("bucket", "DateTime('UTC')"), + ("converted_identities", "UInt64"), + ] + + def test_get_metric_variant_stats__metrics__queries_and_maps_rows( mocker: MockerFixture, ) -> None: @@ -829,7 +845,11 @@ def test_get_metric_variant_stats__metrics__queries_and_maps_rows( ), ] mock_client = mocker.Mock() - mock_client.execute.return_value = (rows, _result_columns(4)) + mock_client.execute.side_effect = [ + (rows, _result_columns(4)), + ([], _conversion_columns()), + [], + ] mock_get_client = mocker.patch( "experimentation.services._get_clickhouse_client", return_value=mock_client, @@ -844,12 +864,13 @@ def test_get_metric_variant_stats__metrics__queries_and_maps_rows( window_end = datetime(2026, 6, 10, tzinfo=timezone.utc) # When - aggregates = services.get_metric_variant_stats( + aggregates = services.get_results_aggregates( environment_key="env-key-123", feature_name="my-feature", window_start=window_start, window_end=window_end, specs=specs, + granularity="day", ) # Then per-variant counts and sufficient statistics are mapped per metric @@ -866,8 +887,9 @@ def test_get_metric_variant_stats__metrics__queries_and_maps_rows( assert aggregates.metric_stats[13]["variant_a"] == VariantStats( n=1000, sum=210.0, sum_squares=520.0 ) - # And the query joins post-exposure metric events and excludes quarantined - sql, params = mock_client.execute.call_args.args + # And the results query joins post-exposure metric events and excludes + # quarantined identities + sql, params = mock_client.execute.call_args_list[0].args assert "LEFT JOIN events AS m" in sql assert "m.timestamp >= e.first_exposure" in sql assert "m.timestamp >= %(window_start)s" in sql @@ -897,7 +919,9 @@ def test_get_metric_variant_stats__metrics__queries_and_maps_rows( assert params["metric_2_event"] == "page_view" assert params["metric_3_event"] == "session" assert params["window_end"] == window_end - mock_get_client.assert_called_once_with( + # And the conversions join is narrowed to the occurrence metric's event + assert params["conversion_events"] == ["purchase"] + mock_get_client.assert_called_with( send_receive_timeout=services.CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS, ) @@ -912,7 +936,11 @@ def test_get_metric_variant_stats__three_variants__maps_all_variants( ("variant_b", 950, 110.0, 110.0, 5100.0, 29000.0), ] mock_client = mocker.Mock() - mock_client.execute.return_value = (rows, _result_columns(2)) + mock_client.execute.side_effect = [ + (rows, _result_columns(2)), + ([], _conversion_columns()), + [], + ] mocker.patch( "experimentation.services._get_clickhouse_client", return_value=mock_client, @@ -923,12 +951,13 @@ def test_get_metric_variant_stats__three_variants__maps_all_variants( ] # When - aggregates = services.get_metric_variant_stats( + aggregates = services.get_results_aggregates( environment_key="env-key-123", feature_name="my-feature", window_start=datetime(2026, 6, 1, tzinfo=timezone.utc), window_end=datetime(2026, 6, 10, tzinfo=timezone.utc), specs=specs, + granularity="day", ) # Then all three variants are decoded into counts and metric stats @@ -948,28 +977,32 @@ def test_get_metric_variant_stats__no_metrics__counts_variants_only( ) -> None: # Given an experiment with no attached metrics mock_client = mocker.Mock() - mock_client.execute.return_value = ( - [("control", 1000), ("variant_a", 900)], - _result_columns(0), - ) + mock_client.execute.side_effect = [ + ([("control", 1000), ("variant_a", 900)], _result_columns(0)), + [], + ] mocker.patch( "experimentation.services._get_clickhouse_client", return_value=mock_client, ) # When - aggregates = services.get_metric_variant_stats( + aggregates = services.get_results_aggregates( environment_key="env-key-123", feature_name="my-feature", window_start=datetime(2026, 6, 1, tzinfo=timezone.utc), window_end=datetime(2026, 6, 10, tzinfo=timezone.utc), specs=[], + granularity="day", ) - # Then only the per-variant counts are returned, with no metric join + # Then only the per-variant counts are returned, with no metric join, and + # no conversions query is run assert aggregates.exposure_counts == {"control": 1000, "variant_a": 900} assert aggregates.metric_stats == {} - sql, params = mock_client.execute.call_args.args + assert aggregates.conversion_buckets == {} + assert mock_client.execute.call_count == 2 + sql, params = mock_client.execute.call_args_list[0].args assert "SELECT variant, count() AS n" in sql assert "LEFT JOIN" not in sql assert "metric_events" not in params @@ -990,7 +1023,11 @@ def test_get_metric_variant_stats__shuffled_columns__maps_by_name( ] rows = [(30000.0, "control", 100.0, 1000, 5000.0, 100.0)] mock_client = mocker.Mock() - mock_client.execute.return_value = (rows, columns) + mock_client.execute.side_effect = [ + (rows, columns), + ([], _conversion_columns()), + [], + ] mocker.patch( "experimentation.services._get_clickhouse_client", return_value=mock_client, @@ -1001,12 +1038,13 @@ def test_get_metric_variant_stats__shuffled_columns__maps_by_name( ] # When - aggregates = services.get_metric_variant_stats( + aggregates = services.get_results_aggregates( environment_key="env-key-123", feature_name="my-feature", window_start=datetime(2026, 6, 1, tzinfo=timezone.utc), window_end=datetime(2026, 6, 10, tzinfo=timezone.utc), specs=specs, + granularity="day", ) # Then values are decoded by column name, not position @@ -1061,6 +1099,178 @@ def test_metric_slot_unit_select__unknown_aggregation__raises() -> None: _MetricSlot(spec=_spec(aggregation="median"), index=0).unit_select() +def test_build_conversions_query__mixed_slots__occurrence_slots_only() -> None: + # Given an occurrence metric, a sum metric and a second occurrence metric + builder = ResultsQueryBuilder( + [ + _spec(metric_id=7, event="purchase", aggregation="occurrence"), + _spec(metric_id=9, event="revenue", aggregation="sum"), + _spec(metric_id=11, event="signup", aggregation="occurrence"), + ] + ) + + # When + sql = builder.build_conversions_query(bucket_function="toStartOfDay") + + # Then each occurrence slot records the identity's first post-exposure + # conversion, with the same attribution condition as the results query + assert sql is not None + assert ( + "minIfOrNull(m.timestamp, m.event = %(metric_0_event)s" + " AND m.timestamp >= e.first_exposure) AS c0" in sql + ) + assert ( + "minIfOrNull(m.timestamp, m.event = %(metric_2_event)s" + " AND m.timestamp >= e.first_exposure) AS c2" in sql + ) + assert " AS c1" not in sql + # And conversions are counted per slot per UTC bucket, skipping identities + # that never converted and identities seen in more than one variant + assert "ARRAY JOIN [0, 2] AS metric_index, [c0, c2] AS first_conversion" in sql + assert "toStartOfDay(first_conversion, 'UTC') AS bucket" in sql + assert "WHERE first_conversion IS NOT NULL" in sql + assert "GROUP BY variant, metric_index, bucket" in sql + assert "WHERE e.quarantined = 0" in sql + # And the join is narrowed to the charted metrics' events + assert "AND m.event IN %(conversion_events)s" in sql + + +def test_build_conversions_query__no_occurrence_slots__returns_none() -> None: + # Given only value metrics, which have no conversion rate to chart + builder = ResultsQueryBuilder( + [ + _spec(metric_id=9, event="revenue", aggregation="sum"), + _spec(metric_id=13, event="session", aggregation="mean"), + ] + ) + + # When / Then + assert builder.build_conversions_query(bucket_function="toStartOfDay") is None + + +def test_decode_conversion_rows__rows__groups_by_metric_behind_slot_index() -> None: + # Given occurrence slots at index 0 and 2, rows for slot 0 only, and + # columns in a different order than the SELECT + builder = ResultsQueryBuilder( + [ + _spec(metric_id=7, event="purchase", aggregation="occurrence"), + _spec(metric_id=9, event="revenue", aggregation="sum"), + _spec(metric_id=11, event="signup", aggregation="occurrence"), + ] + ) + bucket = datetime(2026, 6, 1, tzinfo=timezone.utc) + columns = ["converted_identities", "variant", "bucket", "metric_index"] + rows = [(12, "control", bucket, 0), (3, "variant_a", bucket, 0)] + + # When + buckets = builder.decode_conversion_rows(rows, columns) + + # Then rows are decoded by column name under the metric behind their slot, + # and the charted metric nobody converted on still gets an empty entry + assert buckets == { + 7: [ + ConversionBucket("control", bucket, converted_identities=12), + ConversionBucket("variant_a", bucket, converted_identities=3), + ], + 11: [], + } + + +def test_get_results_aggregates__occurrence_metric__gathers_chart_rows( + mocker: MockerFixture, +) -> None: + # Given the warehouse answers the results query, then one conversion row + # per variant for the occurrence metric in slot 0, then one exposure row + bucket = datetime(2026, 6, 1, tzinfo=timezone.utc) + mock_client = mocker.Mock() + mock_client.execute.side_effect = [ + ([("control", 1000, 12.0, 12.0, 500.0, 900.0)], _result_columns(2)), + ( + [("control", 0, bucket, 12), ("variant_a", 0, bucket, 15)], + _conversion_columns(), + ), + [(0, "control", bucket, 1000)], + ] + mocker.patch( + "experimentation.services._get_clickhouse_client", + return_value=mock_client, + ) + specs = [ + _spec(metric_id=7, event="purchase", aggregation="occurrence"), + _spec(metric_id=9, event="revenue", aggregation="sum"), + ] + window_start = datetime(2026, 6, 1, tzinfo=timezone.utc) + window_end = datetime(2026, 6, 10, tzinfo=timezone.utc) + + # When + aggregates = services.get_results_aggregates( + environment_key="env-key-123", + feature_name="my-feature", + window_start=window_start, + window_end=window_end, + specs=specs, + granularity="day", + ) + + # Then the chart rows are mapped, conversions under the occurrence metric only + assert aggregates.granularity == "day" + assert aggregates.conversion_buckets == { + 7: [ + ConversionBucket("control", bucket, converted_identities=12), + ConversionBucket("variant_a", bucket, converted_identities=15), + ] + } + assert aggregates.exposure_buckets == [ + ExposureBucket("control", bucket, first_exposed_identities=1000) + ] + # And the conversions query buckets first conversions by UTC day, joining + # only the occurrence metric's events over the results query's window + sql, params = mock_client.execute.call_args_list[1].args + assert "toStartOfDay(first_conversion, 'UTC') AS bucket" in sql + assert params == { + "environment_key": "env-key-123", + "exposure_event": "$flag_exposure", + "feature_name": "my-feature", + "window_start": window_start, + "window_end": window_end, + "metric_events": ["purchase", "revenue"], + "conversion_events": ["purchase"], + "metric_0_event": "purchase", + "metric_1_event": "revenue", + } + + +def test_get_results_aggregates__value_metrics_only__skips_conversions_query( + mocker: MockerFixture, +) -> None: + # Given only a value metric is attached + mock_client = mocker.Mock() + mock_client.execute.side_effect = [ + ([("control", 1000, 500.0, 900.0)], _result_columns(1)), + [], + ] + mocker.patch( + "experimentation.services._get_clickhouse_client", + return_value=mock_client, + ) + + # When + aggregates = services.get_results_aggregates( + environment_key="env-key-123", + feature_name="my-feature", + window_start=datetime(2026, 6, 1, tzinfo=timezone.utc), + window_end=datetime(2026, 6, 10, tzinfo=timezone.utc), + specs=[_spec(metric_id=9, event="revenue", aggregation="sum")], + granularity="day", + ) + + # Then nothing is charted and only the results and exposures queries run + assert aggregates.conversion_buckets == {} + assert mock_client.execute.call_count == 2 + _, params = mock_client.execute.call_args_list[0].args + assert params["conversion_events"] == [] + + def test_build_results_summary__healthy_arms__infers_each_treatment() -> None: # Given a 10% control and a 12% treatment, both well above the floor control = VariantStats(n=1000, sum=100.0, sum_squares=100.0) @@ -1261,6 +1471,138 @@ def test_build_results_summary__computed__serialises_to_wire_shape() -> None: "ci_high", "chance_to_win", } + # And the chart series sit alongside, empty for a run with no bucket rows + assert payload["exposures_timeseries"] == {"granularity": "day", "points": []} + assert payload["metrics"][0]["conversions_timeseries"] is None + + +def test_build_results_summary__exposure_rows__attaches_exposures_timeseries() -> None: + # Given exposure bucket rows arriving unordered, with a quarantined row + day_1 = datetime(2026, 6, 1, tzinfo=timezone.utc) + day_3 = datetime(2026, 6, 3, tzinfo=timezone.utc) + aggregates = ResultsAggregates( + specs=[], + exposure_counts={"control": 1000, "variant_a": 1000}, + metric_stats={}, + granularity="day", + exposure_buckets=[ + ExposureBucket("control", day_3, first_exposed_identities=400), + ExposureBucket("control", day_1, first_exposed_identities=600), + ExposureBucket("variant_a", day_1, first_exposed_identities=1000), + ExposureBucket("", day_1, first_exposed_identities=5, quarantined=True), + ], + conversion_buckets={}, + ) + + # When + summary = services.build_results_summary(aggregates, expected_shares={}) + + # Then the chart denominator is bucketed in order, quarantined identities + # left out + assert summary.exposures_timeseries == ExposuresTimeseries( + granularity="day", + points=[ + ExposuresTimeseriesPoint( + bucket=day_1.isoformat(), + new_identities={"control": 600, "variant_a": 1000}, + ), + ExposuresTimeseriesPoint( + bucket=day_3.isoformat(), + new_identities={"control": 400}, + ), + ], + ) + + +def test_build_results_summary__occurrence_metrics__attach_conversions() -> None: + # Given conversion rows for two occurrence metrics, arriving unordered + day_1 = datetime(2026, 6, 1, tzinfo=timezone.utc) + day_3 = datetime(2026, 6, 3, tzinfo=timezone.utc) + aggregates = ResultsAggregates( + specs=[ + _spec(metric_id=7, event="purchase"), + _spec(metric_id=11, event="signup"), + ], + exposure_counts={"control": 1000, "variant_a": 1000}, + metric_stats={}, + granularity="day", + exposure_buckets=[], + conversion_buckets={ + 7: [ + ConversionBucket("variant_a", day_3, converted_identities=20), + ConversionBucket("control", day_1, converted_identities=100), + ConversionBucket("variant_a", day_1, converted_identities=100), + ], + 11: [ConversionBucket("control", day_1, converted_identities=7)], + }, + ) + + # When + summary = services.build_results_summary(aggregates, expected_shares={}) + + # Then each metric gets only its own rows, per variant, in bucket order + assert summary.metrics[0].conversions_timeseries == ConversionsTimeseries( + granularity="day", + points=[ + ConversionsTimeseriesPoint( + bucket=day_1.isoformat(), + converted_identities={"control": 100, "variant_a": 100}, + ), + ConversionsTimeseriesPoint( + bucket=day_3.isoformat(), + converted_identities={"variant_a": 20}, + ), + ], + ) + assert summary.metrics[1].conversions_timeseries == ConversionsTimeseries( + granularity="day", + points=[ + ConversionsTimeseriesPoint( + bucket=day_1.isoformat(), + converted_identities={"control": 7}, + ), + ], + ) + + +def test_build_results_summary__value_metric__timeseries_none() -> None: + # Given a sum metric in a run that gathered bucket rows + aggregates = ResultsAggregates( + specs=[_spec(metric_id=9, event="revenue", aggregation="sum")], + exposure_counts={"control": 1000}, + metric_stats={}, + granularity="day", + exposure_buckets=[], + conversion_buckets={}, + ) + + # When + summary = services.build_results_summary(aggregates, expected_shares={}) + + # Then a value metric has no conversion rate to chart + assert summary.metrics[0].conversions_timeseries is None + + +def test_build_results_summary__occurrence_metric_no_conversions__empty_series() -> ( + None +): + # Given an occurrence metric that was charted but nobody converted on yet + aggregates = ResultsAggregates( + specs=[_spec(metric_id=7, event="purchase")], + exposure_counts={"control": 1000}, + metric_stats={}, + granularity="day", + exposure_buckets=[], + conversion_buckets={7: []}, + ) + + # When + summary = services.build_results_summary(aggregates, expected_shares={}) + + # Then the chart is present and empty, not absent + assert summary.metrics[0].conversions_timeseries == ConversionsTimeseries( + granularity="day", points=[] + ) @pytest.mark.django_db @@ -1494,12 +1836,22 @@ def test_compute_results_summary__experiment__queries_warehouse_and_builds( } }, ) - mock_stats = mocker.patch( - "experimentation.services.get_metric_variant_stats", - return_value=aggregates, - ) window_start = datetime(2026, 6, 1, tzinfo=timezone.utc) window_end = datetime(2026, 6, 10, tzinfo=timezone.utc) + mock_gather = mocker.patch( + "experimentation.services.get_results_aggregates", + return_value=replace( + aggregates, + exposure_buckets=[ + ExposureBucket("control", window_start, first_exposed_identities=1000) + ], + conversion_buckets={ + metric.id: [ + ConversionBucket("control", window_start, converted_identities=100) + ] + }, + ), + ) # When summary = services.compute_results_summary( @@ -1508,19 +1860,37 @@ def test_compute_results_summary__experiment__queries_warehouse_and_builds( window_end=window_end, ) - # Then the warehouse is queried with the experiment's metric specs - mock_stats.assert_called_once_with( + # Then the warehouse is read once for the experiment's metric specs, + # bucketed by day because the window is longer than 72 hours + mock_gather.assert_called_once_with( environment_key=environment.api_key, feature_name=feature.name, window_start=window_start, window_end=window_end, specs=expected_specs, + granularity="day", ) # And the summary carries the metric result with an SRM verdict from the - # configured 50/50 split + # configured 50/50 split, plus both chart series assert summary.srm_p_value == pytest.approx(1.0) assert summary.metrics[0].metric_id == metric.id assert summary.metrics[0].inference["variant_a"] is not None + assert summary.exposures_timeseries == ExposuresTimeseries( + granularity="day", + points=[ + ExposuresTimeseriesPoint( + bucket=window_start.isoformat(), new_identities={"control": 1000} + ) + ], + ) + assert summary.metrics[0].conversions_timeseries == ConversionsTimeseries( + granularity="day", + points=[ + ConversionsTimeseriesPoint( + bucket=window_start.isoformat(), converted_identities={"control": 100} + ) + ], + ) def test_apply_experiment_rollout__no_segment__creates_segment_and_override( diff --git a/api/tests/unit/experimentation/test_tasks.py b/api/tests/unit/experimentation/test_tasks.py index 8155cca3ac2e..b144b251eb45 100644 --- a/api/tests/unit/experimentation/test_tasks.py +++ b/api/tests/unit/experimentation/test_tasks.py @@ -453,8 +453,10 @@ def _results_summary() -> ResultsSummary: "control": VariantStats(n=1000, sum=100.0, sum_squares=100.0) }, inference={}, + conversions_timeseries=None, ) ], + exposures_timeseries=ExposuresTimeseries(granularity="day", points=[]), ) diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 6045c4df06ce..831f765dcf69 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -838,7 +838,7 @@ Attributes: ### `warehouse.connection.connected` Logged at `info` from: - - `api/experimentation/services.py:1190` + - `api/experimentation/services.py:1259` Attributes: - `environment.id` @@ -847,8 +847,8 @@ Attributes: ### `warehouse.connection.event_names_failed` Logged at `warning` from: - - `api/experimentation/services.py:267` - - `api/experimentation/services.py:1290` + - `api/experimentation/services.py:273` + - `api/experimentation/services.py:1359` Attributes: - `environment.id` @@ -858,7 +858,7 @@ Attributes: ### `warehouse.connection.event_stats_failed` Logged at `warning` from: - - `api/experimentation/services.py:1253` + - `api/experimentation/services.py:1322` Attributes: - `environment.id` @@ -867,7 +867,7 @@ Attributes: ### `warehouse.connection.test_event_sent` Logged at `info` from: - - `api/experimentation/services.py:964` + - `api/experimentation/services.py:1033` Attributes: - `environment.id` @@ -876,7 +876,7 @@ Attributes: ### `warehouse.connection.verification_failed` Logged at `warning` from: - - `api/experimentation/services.py:1165` + - `api/experimentation/services.py:1234` Attributes: - `environment.id` @@ -886,7 +886,7 @@ Attributes: ### `warehouse.connection.verification_succeeded` Logged at `info` from: - - `api/experimentation/services.py:1175` + - `api/experimentation/services.py:1244` Attributes: - `environment.id` @@ -895,7 +895,7 @@ Attributes: ### `warehouse.delivery.all_objects_rejected` Logged at `error` from: - - `api/experimentation/services.py:1120` + - `api/experimentation/services.py:1189` Attributes: - `connection.id` @@ -906,7 +906,7 @@ Attributes: ### `warehouse.delivery.budget_exhausted` Logged at `info` from: - - `api/experimentation/services.py:1009` + - `api/experimentation/services.py:1078` Attributes: - `connection.id` @@ -917,7 +917,7 @@ Attributes: ### `warehouse.delivery.completed` Logged at `info` from: - - `api/experimentation/services.py:1130` + - `api/experimentation/services.py:1199` Attributes: - `connection.id` @@ -930,7 +930,7 @@ Attributes: ### `warehouse.delivery.failed` Logged at `error` from: - - `api/experimentation/services.py:1103` + - `api/experimentation/services.py:1172` Attributes: - `connection.id` @@ -941,7 +941,7 @@ Attributes: ### `warehouse.delivery.object_rejected` Logged at `error` from: - - `api/experimentation/services.py:1038` + - `api/experimentation/services.py:1107` Attributes: - `connection.id` @@ -953,7 +953,7 @@ Attributes: ### `warehouse.srm.overallocated` Logged at `error` from: - - `api/experimentation/services.py:560` + - `api/experimentation/services.py:629` Attributes: - `environment.id` @@ -963,7 +963,7 @@ Attributes: ### `warehouse.srm.unkeyed_variant` Logged at `error` from: - - `api/experimentation/services.py:546` + - `api/experimentation/services.py:615` Attributes: - `environment.id`