Average pulse by day in the dashboard, insights and series stats

The dashboard summaries, the insights aggregate and the series stats strip
still took a plain mean over pulse readings for their 7-day, 30-day, prior-month
and all-time figures, while the year-ago figure and the chart already weighed
each day once, so the year delta compared two different statistics. They now
go through the day-mean helper: each day is the mean of its hours' means and a
window the mean of its days, in the frame each reader buckets its days by.
Count, min, max, the median and the spread stay over every reading; every other
type is unchanged.

src/app/api/measurements/series/route.ts

@@ -27,6 +27,12 @@ import { reconstructSleepNights } from "@/lib/analytics/sleep-night";
 import { VALUE_RANGES } from "@/lib/validations/measurement";
 import { loadUserSourcePriority } from "@/lib/rollups/measurement-read";
 import { usesHourlyMeanDay } from "@/lib/measurements/day-statistic";
+import {
+  dayWeightedRows,
+  readingsMean,
+  windowMeanSql,
+  zoneDayFrame,
+} from "@/lib/measurements/day-mean";
 import { seriesRowsFrom } from "@/lib/measurements/series-canonical";
 import { resolveUserTimezone, userDayKey } from "@/lib/tz/resolver";
 import { convertGlucose, resolveGlucoseUnit } from "@/lib/glucose";
@@ -487,17 +493,50 @@ export const GET = apiHandler(async (request: NextRequest) => {
     // strip's mean/min/max/stdDev/count are the same figures the raw path
     // computed: `AVG`/`MIN`/`MAX` over `value`, `STDDEV_POP` matching the
     // JS population-variance helper, exact row count. Aggregate-only —
-    // no row transfer.
-    const [rawAgg] = await prisma.$queryRawUnsafe<
-      Array<{
-        n: number;
-        mean: number | null;
-        min: number | null;
-        max: number | null;
-        sd: number | null;
-      }>
-    >(
-      `
+    // no row transfer. For an hourly-mean type (pulse) the mean is the mean
+    // of the local days' values, each day the mean of its local hours'
+    // means, so it is the mean of the day points the chart draws; count,
+    // min, max and the spread stay over every reading.
+    const [rawAgg] = usesHourlyMeanDay(type)
+      ? await prisma.$queryRawUnsafe<
+          Array<{
+            n: number;
+            mean: number | null;
+            min: number | null;
+            max: number | null;
+            sd: number | null;
+          }>
+        >(
+          `
+      WITH src AS (
+        SELECT m.*
+        FROM ${seriesRows}
+        WHERE m."measured_at" >= $2
+      )
+      SELECT
+        COUNT(*)::int                            AS n,
+        ${windowMeanSql({ typeColumn: 'm."type"', value: 'm."value"', weight: "m.day_weight" })}::double precision
+                                                 AS mean,
+        MIN(m."value")::double precision         AS min,
+        MAX(m."value")::double precision         AS max,
+        STDDEV_POP(m."value")::double precision  AS sd
+      FROM ${dayWeightedRows("src", zoneDayFrame("$3"))} m
+      GROUP BY m."type"
+    `,
+          user.id,
+          since,
+          userTz,
+        )
+      : await prisma.$queryRawUnsafe<
+          Array<{
+            n: number;
+            mean: number | null;
+            min: number | null;
+            max: number | null;
+            sd: number | null;
+          }>
+        >(
+          `
       SELECT
         COUNT(*)::int                            AS n,
         AVG(m."value")::double precision         AS mean,
@@ -507,9 +546,9 @@ export const GET = apiHandler(async (request: NextRequest) => {
       FROM ${seriesRows}
       WHERE m."measured_at" >= $2
     `,
-      user.id,
-      since,
-    );
+          user.id,
+          since,
+        );
     const round2 = (v: number) => Math.round(v * 100) / 100;
     const dayId = (d: Date) =>
       grain === "day"
@@ -687,6 +726,25 @@ export const GET = apiHandler(async (request: NextRequest) => {
           valueMax: row.valueMax ?? null,
         };
       });
+      // An hourly-mean type's strip mean is the mean of its local days'
+      // values, each day the mean of its local hours' means, so a dense
+      // workout hour does not outweigh the rest of the window. Count, min,
+      // max and the spread stay over every reading.
+      if (usesHourlyMeanDay(type) && rows.length > 0) {
+        const userTz = await resolveUserTimezone(user.id);
+        const values = rows.map((r) => r.value);
+        const rawSummary = summarize(
+          rows.map((r) => ({ date: r.measuredAt, value: r.value })),
+        );
+        const mean = readingsMean(type, rows, userTz);
+        statsOverride = {
+          mean: mean === null ? 0 : Math.round(mean * 100) / 100,
+          min: rawSummary.min ?? 0,
+          max: rawSummary.max ?? 0,
+          stdDev: stdDev(values),
+          count: rawSummary.count,
+        };
+      }
     } else {
       points = rows.map((r) => ({
         id: r.id,

src/lib/analytics/__tests__/summaries-slice.test.ts

@@ -45,11 +45,9 @@ vi.mock("@/lib/logging/context", () => ({
 // 90-day FILTER caps the slice itself writes — is exercised without
 // coupling the slice unit test to the rank builder's enum-whitelist
 // internals.
-vi.mock("@/lib/analytics/source-rank-sql", () => ({
-  buildSourceRankCase: vi.fn(() => "90"),
-  canonicalMeasurementsFrom: vi.fn(
-    (_rank: string, sinceInterval?: string) =>
-      `(
+vi.mock("@/lib/analytics/source-rank-sql", () => {
+  const cte = (_rank: string, sinceInterval?: string) =>
+    `
         SELECT mm.*
         FROM measurements mm
         WHERE mm."user_id" = $1
@@ -58,13 +56,15 @@ vi.mock("@/lib/analytics/source-rank-sql", () => ({
             sinceInterval
               ? `AND mm."measured_at" >= NOW() - INTERVAL '${sinceInterval}'`
               : ""
-          }
-      ) m`,
-  ),
-}));
+          }`;
+  return {
+    buildSourceRankCase: vi.fn(() => "90"),
+    canonicalMeasurementsCte: vi.fn(cte),
+  };
+});
 
 import { prisma } from "@/lib/db";
-import { canonicalMeasurementsFrom } from "@/lib/analytics/source-rank-sql";
+import { canonicalMeasurementsCte } from "@/lib/analytics/source-rank-sql";
 import { startOfUtcDay } from "@/lib/tz/start-of-utc-day";
 import { computeSummariesSlice } from "../summaries-slice";
 
@@ -79,7 +79,7 @@ const ROLLUP_FIND_MANY = prisma.measurementRollup
   .findMany as unknown as ReturnType<typeof vi.fn>;
 const ROLLUP_FIND_FIRST = prisma.measurementRollup
   .findFirst as unknown as ReturnType<typeof vi.fn>;
-const CANONICAL_FROM = canonicalMeasurementsFrom as unknown as ReturnType<
+const CANONICAL_FROM = canonicalMeasurementsCte as unknown as ReturnType<
   typeof vi.fn
 >;
 
@@ -609,11 +609,12 @@ describe("computeSummariesSlice", () => {
       expect(narrowsCap?.length).toBeGreaterThanOrEqual(1);
 
       // v1.11.1 — the cold-fallback `windowed` scan delegates its outer
-      // 90-day cap to `canonicalMeasurementsFrom(rank, "90 days")`. The
+      // 90-day cap to `canonicalMeasurementsCte(rank, "90 days")` (a CTE
+      // now, so the day weights can read it twice). The
       // helper lives in `@/lib/analytics/source-rank-sql` (stubbed
       // above), so pin the contract at the call boundary: the slice
       // must ask for the 90-day window.
-      expect(canonicalMeasurementsFrom).toHaveBeenCalledWith(
+      expect(canonicalMeasurementsCte).toHaveBeenCalledWith(
         expect.any(String),
         "90 days",
       );
@@ -621,7 +622,7 @@ describe("computeSummariesSlice", () => {
       // 15-year scan-DoS floor (far beyond any real history) so a
       // coverage-miss cold read can't trigger an unbounded full-partition
       // scan. It no longer calls the helper without an interval.
-      expect(canonicalMeasurementsFrom).toHaveBeenCalledWith(
+      expect(canonicalMeasurementsCte).toHaveBeenCalledWith(
         expect.any(String),
         "15 years",
       );

src/lib/analytics/summaries-slice.ts

@@ -60,7 +60,18 @@ import type { MeasurementType } from "@/generated/prisma/client";
 import { Prisma } from "@/generated/prisma/client";
 
 import { prisma } from "@/lib/db";
-import { windowWeighting } from "@/lib/measurements/day-statistic";
+import {
+  usesHourlyMeanDay,
+  windowWeighting,
+} from "@/lib/measurements/day-statistic";
+import {
+  SESSION_DAY_FRAME,
+  dayWeightedRows,
+  dayWeightedRowsSql,
+  foldMeanSql,
+  isHourlyMeanTypeSql,
+  windowMeanSql,
+} from "@/lib/measurements/day-mean";
 import type { DataSummary } from "@/lib/analytics/trends";
 import { measurementTypeEnum } from "@/lib/validations/measurement";
 import { annotate } from "@/lib/logging/context";
@@ -80,7 +91,7 @@ import {
 import { startOfUtcDay } from "@/lib/tz/start-of-utc-day";
 import {
   buildSourceRankCase,
-  canonicalMeasurementsFrom,
+  canonicalMeasurementsCte,
 } from "@/lib/analytics/source-rank-sql";
 import {
   isFullyCovered,
@@ -137,11 +148,33 @@ interface AllTimeAggregateRow {
 interface RollupAggregateRow {
   type: string;
   count: number;
+  /** Days that hold a reading: the weight of the mean for an hourly-mean type. */
+  days: number;
   min: number;
   max: number;
   mean: number;
 }
 
+/**
+ * The window means over canonical readings that carry a `day_weight` (see
+ * `day-mean.ts`): the mean of the day values for an hourly-mean type (pulse),
+ * the unchanged `AVG(value)` for every other type. Days are UTC days, the
+ * frame the canonical-source pick and the rollup tier bucket by.
+ */
+function windowedMean(filter: string): string {
+  return windowMeanSql({
+    typeColumn: 'm."type"',
+    value: 'm."value"',
+    weight: "m.day_weight",
+    filter,
+  });
+}
+const AVG7_SQL = windowedMean(`m."measured_at" >= NOW() - INTERVAL '7 days'`);
+const AVG30_SQL = windowedMean(`m."measured_at" >= NOW() - INTERVAL '30 days'`);
+const AVG30_LAST_MONTH_SQL = windowedMean(
+  `m."measured_at" >= NOW() - INTERVAL '60 days' AND m."measured_at" <  NOW() - INTERVAL '30 days'`,
+);
+
 interface NarrowAggregateRow {
   type: string;
   avg7: number | null;
@@ -455,26 +488,23 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
       // inner DISTINCT ON picks the ladder-winning source for each day, and the
       // join keeps only that source's readings so the 90-day AVG / median / slope
       // never blend two devices that both reported the same vital.
+      // The window means weigh each day once for an hourly-mean type
+      // (`windowedMean`); the median stays over every reading.
       prisma.$queryRawUnsafe<NarrowAggregateRow[]>(
         `
+      WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "90 days")}
+      )
       SELECT
         m."type"::text                                                AS type,
-        AVG(m."value") FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '7 days'
-        )::double precision                                           AS avg7,
-        AVG(m."value") FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '30 days'
-        )::double precision                                           AS avg30,
-        AVG(m."value") FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '60 days'
-            AND m."measured_at" <  NOW() - INTERVAL '30 days'
-        )::double precision                                           AS avg30_last_month,
+        ${AVG7_SQL}::double precision                                 AS avg7,
+        ${AVG30_SQL}::double precision                                AS avg30,
+        ${AVG30_LAST_MONTH_SQL}::double precision                     AS avg30_last_month,
         PERCENTILE_CONT(0.5) WITHIN GROUP (
           ORDER BY m."value"
         ) FILTER (
           WHERE m."measured_at" >= NOW() - INTERVAL '90 days'
         )::double precision                                           AS median
-      FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")}
+      FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m
       GROUP BY m."type"
     `,
         userId,
@@ -527,12 +557,19 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
       SELECT
         "type"::text                                       AS type,
         SUM("count")::int                                  AS count,
+        COUNT(*)::int                                      AS days,
         MIN("min_value")::double precision                 AS min,
         MAX("max_value")::double precision                 AS max,
-        (
-          SUM("count" * "mean")::double precision
-          / NULLIF(SUM("count")::double precision, 0)
-        )                                                  AS mean
+        -- An hourly-mean type (pulse): every day once, the DAY mean already
+        -- being the mean of its hours' means. Every other type: every
+        -- reading once.
+        (CASE WHEN ${isHourlyMeanTypeSql('"type"')}
+          THEN AVG("mean")::double precision
+          ELSE (
+            SUM("count" * "mean")::double precision
+            / NULLIF(SUM("count")::double precision, 0)
+          )
+        END)                                               AS mean
       FROM collapsed
       GROUP BY "type"
     `,
@@ -576,16 +613,27 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
       // day first (a hash aggregate, a few rows per day however dense the
       // stream), each day then keeps its ladder-canonical source, so the
       // sort never sees more than one row per source per day.
+      // An hourly-mean type folds each source's day as the mean of its
+      // hours' means (`wsum` / `wdays`, see `day-mean.ts`) and the days into
+      // the remainder once each, the statistic the DAY rollup stores.
       prisma.$queryRaw<
         Array<{
           type: string;
           count: number;
+          days: number;
           min: number;
           max: number;
           mean: number;
         }>
       >`
-      WITH per_source AS (
+      WITH src AS (
+        SELECT *
+        FROM measurements m
+        WHERE m."user_id" = ${userId}
+          AND m."deleted_at" IS NULL
+          AND m."measured_at" < ${foldSplit}
+      ),
+      per_source AS (
         SELECT
           m."type",
           m."source",
@@ -593,25 +641,33 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
           COUNT(*)::int                      AS cnt,
           SUM(m."value")::double precision   AS total,
           MIN(m."value")::double precision   AS min_value,
-          MAX(m."value")::double precision   AS max_value
-        FROM measurements m
-        WHERE m."user_id" = ${userId}
-          AND m."deleted_at" IS NULL
-          AND m."measured_at" < ${foldSplit}
+          MAX(m."value")::double precision   AS max_value,
+          SUM(m."value" * m.day_weight)::double precision AS wsum,
+          SUM(m.day_weight)::double precision             AS wdays
+        FROM ${dayWeightedRowsSql("src", null, { bySource: true })} m
         GROUP BY m."type", m."source", 3
       ),
       canon AS (
         SELECT DISTINCT ON (p."type", p.day)
-          p."type", p.cnt, p.total, p.min_value, p.max_value
+          p."type", p.cnt, p.total, p.min_value, p.max_value, p.wsum, p.wdays
         FROM per_source p
         ORDER BY p."type", p.day, (${Prisma.raw(rankPerSource)}), p."source"::text
       )
       SELECT
         c."type"::text                                  AS type,
         SUM(c.cnt)::int                                 AS count,
+        COUNT(*)::int                                   AS days,
         MIN(c.min_value)::double precision              AS min,
         MAX(c.max_value)::double precision              AS max,
-        (SUM(c.total) / SUM(c.cnt))::double precision   AS mean
+        ${Prisma.raw(
+          foldMeanSql({
+            typeColumn: 'c."type"',
+            total: "c.total",
+            count: "c.cnt",
+            weightedSum: "c.wsum",
+            weightSum: "c.wdays",
+          }),
+        )}::double precision                            AS mean
       FROM canon c
       GROUP BY c."type"
     `,
@@ -619,11 +675,12 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
 
   const preFoldByType = new Map<
     string,
-    { count: number; min: number; max: number; mean: number }
+    { count: number; days: number; min: number; max: number; mean: number }
   >();
   for (const row of preFoldRemainder) {
     preFoldByType.set(row.type, {
       count: Number(row.count),
+      days: Number(row.days),
       min: Number(row.min),
       max: Number(row.max),
       mean: Number(row.mean),
@@ -754,8 +811,13 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
     const allCount = row.count + (pre?.count ?? 0);
     const allMin = pre ? Math.min(row.min, pre.min) : row.min;
     const allMax = pre ? Math.max(row.max, pre.max) : row.max;
+    // Both halves weigh their days once for an hourly-mean type (pulse), so
+    // they combine by day count; every other type combines by readings.
     const allMean = pre
-      ? (row.mean * row.count + pre.mean * pre.count) / allCount
+      ? usesHourlyMeanDay(row.type)
+        ? (row.mean * Number(row.days) + pre.mean * pre.days) /
+          (Number(row.days) + pre.days)
+        : (row.mean * row.count + pre.mean * pre.count) / allCount
       : row.mean;
     summaries[row.type] = {
       count: allCount,
@@ -889,31 +951,29 @@ async function computeFromLiveAggregate(
   const [allTime, windowed, latests] = await Promise.all([
     prisma.$queryRawUnsafe<AllTimeAggregateRow[]>(
       `
+      WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, ALL_TIME_LIVE_CAP_INTERVAL)}
+      )
       SELECT
         m."type"::text                                                AS type,
         COUNT(*)                                                      AS count,
         MIN(m."value")::double precision                              AS min_value,
         MAX(m."value")::double precision                              AS max_value,
-        AVG(m."value")::double precision                              AS mean_value
-      FROM ${canonicalMeasurementsFrom(rankUnqualified, ALL_TIME_LIVE_CAP_INTERVAL)}
+        ${windowMeanSql({ typeColumn: 'm."type"', value: 'm."value"', weight: "m.day_weight" })}::double precision
+                                                                      AS mean_value
+      FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m
       GROUP BY m."type"
     `,
       userId,
     ),
     prisma.$queryRawUnsafe<WindowedAggregateRow[]>(
       `
+      WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "90 days")}
+      )
       SELECT
         m."type"::text                                                AS type,
-        AVG(m."value") FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '7 days'
-        )::double precision                                           AS avg7,
-        AVG(m."value") FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '30 days'
-        )::double precision                                           AS avg30,
-        AVG(m."value") FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '60 days'
-            AND m."measured_at" <  NOW() - INTERVAL '30 days'
-        )::double precision                                           AS avg30_last_month,
+        ${AVG7_SQL}::double precision                                 AS avg7,
+        ${AVG30_SQL}::double precision                                AS avg30,
+        ${AVG30_LAST_MONTH_SQL}::double precision                     AS avg30_last_month,
         PERCENTILE_CONT(0.5) WITHIN GROUP (
           ORDER BY m."value"
         ) FILTER (
@@ -956,7 +1016,7 @@ async function computeFromLiveAggregate(
         ) FILTER (
           WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days') AT TIME ZONE 'UTC'
         )::double precision                                           AS r2_30
-      FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")}
+      FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m
       GROUP BY m."type"
     `,
       userId,

src/lib/insights/comprehensive-aggregator.ts

@@ -114,6 +114,11 @@ import {
   type AccumulatorBucketRow,
 } from "@/lib/rollups/measurement-read";
 import { windowWeighting } from "@/lib/measurements/day-statistic";
+import {
+  SESSION_DAY_FRAME,
+  dayWeightedRows,
+  windowMeanSql,
+} from "@/lib/measurements/day-mean";
 import { startOfUtcDay } from "@/lib/tz/start-of-utc-day";
 import {
   buildSourceRankCase,
@@ -126,6 +131,26 @@ import {
 } from "@/lib/rollups/measurement-coverage";
 import { dateOnlyKey } from "@/lib/tz/date-only";
 
+/**
+ * Window means over canonical readings carrying a `day_weight` (see
+ * `day-mean.ts`): the mean of the day values for an hourly-mean type (pulse),
+ * the unchanged `AVG(value)` otherwise. Days are UTC days, the frame the
+ * canonical-source pick buckets by.
+ */
+function windowedMean(filter: string): string {
+  return windowMeanSql({
+    typeColumn: 'm."type"',
+    value: 'm."value"',
+    weight: "m.day_weight",
+    filter,
+  });
+}
+const AVG7_SQL = windowedMean(`m."measured_at" >= NOW() - INTERVAL '7 days'`);
+const AVG30_SQL = windowedMean(`m."measured_at" >= NOW() - INTERVAL '30 days'`);
+const AVG30_LAST_MONTH_SQL = windowedMean(
+  `m."measured_at" >= NOW() - INTERVAL '60 days' AND m."measured_at" <  NOW() - INTERVAL '30 days'`,
+);
+
 /**
  * Heavy aggregate row — count/min/max/mean alongside the non-composable
  * windowed columns. Used on the cold-mount fallback path where the
@@ -419,6 +444,9 @@ async function buildFromRollups(
       -- instead of twice per rebuild. Output is identical.
       WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "90 days")}
       ),
+      -- The z-score's mean and spread stay over every reading: they measure
+      -- how far one reading sits from the others, a per-reading dispersion,
+      -- not the level the window means below report.
       window_stats AS (
         SELECT
           m."type",
@@ -440,18 +468,11 @@ async function buildFromRollups(
               (ROUND(((m."value" - ws.mean_value) / ws.stddev_value)::numeric, 2))
             ) > 2
         )                                                             AS anomaly_count,
-        AVG(m."value") FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '7 days'
-        )::double precision                                           AS avg7,
-        AVG(m."value") FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '30 days'
-        )::double precision                                           AS avg30,
+        ${AVG7_SQL}::double precision                                 AS avg7,
+        ${AVG30_SQL}::double precision                                AS avg30,
         -- Prior 30d window (days [30, 60) ago) used by tile delta callout.
-        AVG(m."value") FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '60 days'
-            AND m."measured_at" <  NOW() - INTERVAL '30 days'
-        )::double precision                                           AS avg30_last_month
-      FROM cm m
+        ${AVG30_LAST_MONTH_SQL}::double precision                     AS avg30_last_month
+      FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m
       JOIN window_stats ws ON ws."type" = m."type"
       GROUP BY m."type", ws.stddev_value
     `,
@@ -662,6 +683,9 @@ async function buildFromLiveAggregate(
       -- so the DISTINCT-ON pick runs once. Output is identical.
       WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "90 days")}
       ),
+      -- The z-score's mean and spread stay over every reading: they measure
+      -- how far one reading sits from the others, a per-reading dispersion,
+      -- not the level the window means below report.
       window_stats AS (
         SELECT
           m."type",
@@ -675,7 +699,8 @@ async function buildFromLiveAggregate(
         COUNT(*)                                                      AS count,
         MIN(m."value")::double precision                              AS min_value,
         MAX(m."value")::double precision                              AS max_value,
-        AVG(m."value")::double precision                              AS mean_value,
+        ${windowMeanSql({ typeColumn: 'm."type"', value: 'm."value"', weight: "m.day_weight" })}::double precision
+                                                                      AS mean_value,
         ws.stddev_value::double precision                             AS stddev_value,
         COUNT(*) FILTER (
           WHERE ws.stddev_value IS NOT NULL
@@ -684,16 +709,9 @@ async function buildFromLiveAggregate(
               (ROUND(((m."value" - ws.mean_value) / ws.stddev_value)::numeric, 2))
             ) > 2
         )                                                             AS anomaly_count,
-        AVG(m."value") FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '7 days'
-        )::double precision                                           AS avg7,
-        AVG(m."value") FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '30 days'
-        )::double precision                                           AS avg30,
-        AVG(m."value") FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '60 days'
-            AND m."measured_at" <  NOW() - INTERVAL '30 days'
-        )::double precision                                           AS avg30_last_month,
+        ${AVG7_SQL}::double precision                                 AS avg7,
+        ${AVG30_SQL}::double precision                                AS avg30,
+        ${AVG30_LAST_MONTH_SQL}::double precision                     AS avg30_last_month,
         -- A2-M2 — the regression windows anchor on the UTC-midnight day
         -- boundary, NOT the wall-clock instant, so this cold-fallback path
         -- returns the SAME boundary-day membership the warm rollup path
@@ -731,7 +749,7 @@ async function buildFromLiveAggregate(
         ) FILTER (
           WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days') AT TIME ZONE 'UTC'
         )::double precision                                           AS r2_30
-      FROM cm m
+      FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m
       JOIN window_stats ws ON ws."type" = m."type"
       GROUP BY m."type", ws.stddev_value
     `,