From 4e7beb797cc01bdfbe27d0a83bda2dfcd9be14c7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 26 Sep 2026 16:07:45 +0200 Subject: [PATCH] fix(analytics): all-time extremes and range deltas count one source per day The all-time weight, blood-pressure and pulse figures the insight features carry averaged every raw row, so a device that samples densely outweighed the readings the source ladder prefers on the same days. They now keep one source per day, like the dashboard's all-time figures. The period-over-period range delta averaged readings. For step-like totals the current window still holds the last day and a half as many small samples, so the average collapsed; it now compares average daily totals. For sleep it averaged stage segments and in-bed envelopes; it now compares average nights of time asleep. The API descriptions say which. --- docs/api/openapi.yaml | 17 ++- .../__tests__/range-delta-per-period.test.ts | 93 ++++++++++++ src/lib/analytics/range-delta.ts | 137 ++++++++++++++++++ src/lib/insights/__tests__/features.test.ts | 8 +- src/lib/insights/feature-blocks.ts | 72 +++++++-- src/lib/openapi/routes/insights/schemas.ts | 15 +- .../rollup-per-day-canonical.test.ts | 83 +++++++++++ 7 files changed, 396 insertions(+), 29 deletions(-) create mode 100644 src/lib/analytics/__tests__/range-delta-per-period.test.ts diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 89ecc1799..5f77cff92 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -54940,8 +54940,8 @@ components: description: Trailing-window length in days for the chosen range. granularity: type: string - description: Rollup granularity the read resolved against (`DAY` / `WEEK` / `MONTH` / `YEAR`, or `none` on a coverage - miss). + description: Rollup granularity the read resolved against (`DAY` / `WEEK` / `MONTH` / `YEAR`), `live` for SLEEP_DURATION + (nights are reconstructed from the stage rows), or `none` when there is no data. current: description: Aggregate over the current window `[now-N, now)`. $ref: "#/components/schemas/AnalyticsWindowAggregate" @@ -54968,9 +54968,10 @@ components: - delta - deltaPct additionalProperties: false - description: Single-metric period-over-period aggregate. Reads the current and previous comparable windows from the WMY - rollup tier and composes a count-weighted-mean delta. `count/min/max/mean/sum` are linearly composable across - buckets; SD/slope/r² are intentionally excluded (not composable). + description: Single-metric period-over-period aggregate. Reads the current and previous comparable windows from the + rollup tier and composes a count-weighted-mean delta, one source per day. Step-like totals compare average daily + totals and SLEEP_DURATION compares average nights, not per-reading means. `count/min/max/mean/sum` are linearly + composable across buckets; SD/slope/r² are intentionally excluded (not composable). AnalyticsWindowAggregate: type: object properties: @@ -54978,7 +54979,8 @@ components: type: integer minimum: -9007199254740991 maximum: 9007199254740991 - description: Reading count composed across buckets. + description: Reading count composed across buckets. For step-like totals it is the number of days with a total, for + SLEEP_DURATION the number of nights. min: anyOf: - type: number @@ -54993,7 +54995,8 @@ components: anyOf: - type: number - type: "null" - description: Count-weighted mean across buckets; null when empty. + description: Count-weighted mean across buckets; null when empty. For step-like totals (steps, energy, distance, + flights) it is the average daily total, for SLEEP_DURATION the average time asleep per night. sum: anyOf: - type: number diff --git a/src/lib/analytics/__tests__/range-delta-per-period.test.ts b/src/lib/analytics/__tests__/range-delta-per-period.test.ts new file mode 100644 index 000000000..445aba5bc --- /dev/null +++ b/src/lib/analytics/__tests__/range-delta-per-period.test.ts @@ -0,0 +1,93 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + readRollupBuckets: vi.fn(), + findMany: vi.fn(), +})); + +vi.mock("@/lib/db", () => ({ + prisma: { measurement: { findMany: mocks.findMany } }, +})); +vi.mock("@/lib/rollups/measurement-rollups", () => ({ + readRollupBuckets: mocks.readRollupBuckets, +})); +vi.mock("@/lib/rollups/measurement-read", () => ({ + loadUserSourcePriority: vi.fn(async () => null), +})); +vi.mock("@/lib/rollups/measurement-read-wmy", () => ({ + aggregateWmyBuckets: vi.fn(), + readBestGranularityRollups: vi.fn(async () => null), +})); +vi.mock("@/lib/tz/resolver", () => ({ + resolveUserTimezone: vi.fn(async () => "UTC"), +})); + +import { computeRangeDelta } from "@/lib/analytics/range-delta"; + +const DAY_MS = 86_400_000; +const NOW = Date.UTC(2026, 5, 30, 12); + +function dayRow(daysAgo: number, total: number, samples: number) { + const bucketStart = new Date(Date.UTC(2026, 5, 30) - daysAgo * DAY_MS); + return { + bucketStart, + count: samples, + mean: total / samples, + minValue: 0, + maxValue: total, + sd: null, + slope: null, + r2: null, + computedAt: bucketStart, + }; +} + +beforeEach(() => { + mocks.readRollupBuckets.mockReset(); + mocks.findMany.mockReset(); +}); + +describe("computeRangeDelta — per-day and per-night metrics", () => { + it("compares average daily step totals, not the mean of the readings", async () => { + // Six complete days of 8,000 in the previous week, and in the current + // week five drained days plus today still arriving as 200 samples. + mocks.readRollupBuckets.mockResolvedValue([ + ...[8, 9, 10, 11, 12, 13].map((d) => dayRow(d, 8000, 1)), + ...[1, 2, 3, 4, 5].map((d) => dayRow(d, 8000, 1)), + dayRow(0, 4000, 200), + ]); + + const result = await computeRangeDelta("u", "ACTIVITY_STEPS", "7d", NOW); + + expect(result.granularity).toBe("DAY"); + expect(result.previous).toMatchObject({ count: 6, mean: 8000 }); + expect(result.current.count).toBe(6); + expect(result.current.mean).toBeCloseTo((5 * 8000 + 4000) / 6, 6); + expect(result.current.sum).toBeCloseTo(44_000, 6); + }); + + it("compares average nights of sleep, not the mean of stage rows", async () => { + const row = (iso: string, minutes: number, sleepStage: string) => ({ + value: minutes, + measuredAt: new Date(iso), + sleepStage, + source: "APPLE_HEALTH", + deviceType: null, + }); + // Two nights: 7 h asleep (split into three stage rows plus an in-bed + // envelope) and 6 h asleep. + mocks.findMany.mockResolvedValue([ + row("2026-06-28T01:00:00.000Z", 480, "IN_BED"), + row("2026-06-28T02:00:00.000Z", 180, "CORE"), + row("2026-06-28T04:00:00.000Z", 120, "DEEP"), + row("2026-06-28T06:00:00.000Z", 120, "REM"), + row("2026-06-29T06:00:00.000Z", 360, "ASLEEP"), + ]); + + const result = await computeRangeDelta("u", "SLEEP_DURATION", "7d", NOW); + + expect(result.granularity).toBe("live"); + expect(result.current.count).toBe(2); + expect(result.current.mean).toBeCloseTo(390, 6); + }); +}); diff --git a/src/lib/analytics/range-delta.ts b/src/lib/analytics/range-delta.ts index 388321fab..d2673064c 100644 --- a/src/lib/analytics/range-delta.ts +++ b/src/lib/analytics/range-delta.ts @@ -25,14 +25,34 @@ * a window aggregate built from WEEK / MONTH buckets equals the per-row * aggregate over the same span. SD / slope / r² are intentionally NOT part of * the delta — they do not compose, matching the rest of the WMY reader tier. + * + * Two kinds of metric are not averaged per reading: + * - step-like totals (steps, energy, distance, flights, …) compare the + * average DAILY total: a day's readings are pieces of one total, and the + * last day and a half still arrives as many small samples before the + * drain folds them, so a per-reading mean collapsed the current window; + * - sleep compares the average NIGHT (time asleep, reconstructed from the + * stage rows): a per-row mean averaged stage segments and in-bed + * envelopes, which is no amount of sleep at all. + * For both, `count` is the number of days or nights. */ import type { MeasurementType } from "@/generated/prisma/client"; +import { prisma } from "@/lib/db"; +import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping"; +import { + reconstructSleepNights, + type SleepStageRow, +} from "@/lib/analytics/sleep-night"; +import { readRollupBuckets } from "@/lib/rollups/measurement-rollups"; +import { loadUserSourcePriority } from "@/lib/rollups/measurement-read"; import { aggregateWmyBuckets, readBestGranularityRollups, type RollupBucketRow, } from "@/lib/rollups/measurement-read-wmy"; +import { startOfUtcDay } from "@/lib/tz/start-of-utc-day"; +import { resolveUserTimezone } from "@/lib/tz/resolver"; import { ANALYTICS_RANGES, rangeWindowDays, @@ -86,6 +106,99 @@ export function sliceWindowDelta( }; } +/** + * Aggregate one value per day or night (a daily total, a night's time + * asleep): `count` is the number of days or nights, `mean` their average. + */ +export function aggregatePerPeriod(values: readonly number[]): WindowAggregate { + if (values.length === 0) { + return { count: 0, min: null, max: null, mean: null, sum: null }; + } + let sum = 0; + let min = Infinity; + let max = -Infinity; + for (const v of values) { + sum += v; + if (v < min) min = v; + if (v > max) max = v; + } + return { count: values.length, min, max, mean: sum / values.length, sum }; +} + +/** Split `(at, value)` points into the current and previous halves. */ +function splitPerPeriod( + points: ReadonlyArray<{ at: Date; value: number }>, + windowDays: number, + now: number, +): { current: WindowAggregate; previous: WindowAggregate } { + const currentStart = now - windowDays * DAY_MS; + const previousStart = now - 2 * windowDays * DAY_MS; + const current: number[] = []; + const previous: number[] = []; + for (const p of points) { + const t = p.at.getTime(); + if (t >= currentStart && t < now) current.push(p.value); + else if (t >= previousStart && t < currentStart) previous.push(p.value); + } + return { + current: aggregatePerPeriod(current), + previous: aggregatePerPeriod(previous), + }; +} + +/** The canonical daily totals of a step-like metric, one point per day. */ +async function readDailyTotals( + userId: string, + type: MeasurementType, + windowDays: number, + now: number, +): Promise> { + const rows = await readRollupBuckets( + userId, + type, + "DAY", + startOfUtcDay(new Date(now - 2 * windowDays * DAY_MS)), + new Date(now), + ); + return rows.map((r) => ({ at: r.bucketStart, value: r.mean * r.count })); +} + +/** Time asleep per reconstructed night, one point per night. */ +async function readNightlySleep( + userId: string, + windowDays: number, + now: number, +): Promise> { + // A night's rows start the evening before its wake day: read one day more. + const rows = (await prisma.measurement.findMany({ + where: { + userId, + type: "SLEEP_DURATION", + deletedAt: null, + measuredAt: { + gte: new Date(now - (2 * windowDays + 1) * DAY_MS), + lt: new Date(now), + }, + }, + orderBy: { measuredAt: "asc" }, + select: { + value: true, + measuredAt: true, + sleepStage: true, + source: true, + deviceType: true, + }, + })) as SleepStageRow[]; + if (rows.length === 0) return []; + const [tz, priorityJson] = await Promise.all([ + resolveUserTimezone(userId), + loadUserSourcePriority(userId), + ]); + return reconstructSleepNights(rows, tz, priorityJson) + .filter((n) => n.asleepMinutes > 0) + .map((n) => ({ at: n.measuredAt, value: n.asleepMinutes })); +} + /** * Compose the period-over-period delta from the two window aggregates. * Pure — pinned by unit test. Guards both the missing-data case (either @@ -119,6 +232,30 @@ export async function computeRangeDelta( now: number = Date.now(), ): Promise { const windowDays = rangeWindowDays(range); + + if (type === "SLEEP_DURATION" || CUMULATIVE_HK_TYPES.has(type)) { + const points = + type === "SLEEP_DURATION" + ? await readNightlySleep(userId, windowDays, now) + : await readDailyTotals(userId, type, windowDays, now); + const { current, previous } = splitPerPeriod(points, windowDays, now); + const { delta, deltaPct } = composeDelta(current, previous); + return { + range, + windowDays, + granularity: + points.length === 0 + ? "none" + : type === "SLEEP_DURATION" + ? "live" + : "DAY", + current, + previous, + delta, + deltaPct, + }; + } + // Read the full 2N span in one go so the current and previous halves share // a granularity; the router picks the coarsest tier that resolves 2N. const resolved = await readBestGranularityRollups( diff --git a/src/lib/insights/__tests__/features.test.ts b/src/lib/insights/__tests__/features.test.ts index 2fa796a92..07b10173f 100644 --- a/src/lib/insights/__tests__/features.test.ts +++ b/src/lib/insights/__tests__/features.test.ts @@ -3,7 +3,7 @@ import { describe, expect, it, vi, beforeEach } from "vitest"; vi.mock("@/lib/db", () => ({ prisma: { user: { findUnique: vi.fn() }, - measurement: { findMany: vi.fn(), groupBy: vi.fn() }, + measurement: { findMany: vi.fn() }, measurementRollup: { findMany: vi.fn() }, moodEntry: { findMany: vi.fn(), findFirst: vi.fn() }, moodEntryRollup: { findMany: vi.fn(), findFirst: vi.fn() }, @@ -15,6 +15,8 @@ vi.mock("@/lib/db", () => ({ workout: { findMany: vi.fn() }, ecgRecording: { findMany: vi.fn() }, $queryRawUnsafe: vi.fn(), + // The all-time extremes read (one grouped aggregate). + $queryRaw: vi.fn(), }, })); @@ -49,7 +51,6 @@ const prismaMock = prisma as unknown as { user: { findUnique: ReturnType }; measurement: { findMany: ReturnType; - groupBy: ReturnType; }; measurementRollup: { findMany: ReturnType }; moodEntry: { @@ -66,6 +67,7 @@ const prismaMock = prisma as unknown as { measurementReminder: { findMany: ReturnType }; workout: { findMany: ReturnType }; ecgRecording: { findMany: ReturnType }; + $queryRaw: ReturnType; }; const dayMs = 24 * 60 * 60 * 1000; @@ -123,7 +125,7 @@ beforeEach(() => { gender: "MALE", }); prismaMock.measurement.findMany.mockResolvedValue([]); - prismaMock.measurement.groupBy.mockResolvedValue([]); + prismaMock.$queryRaw.mockResolvedValue([]); prismaMock.measurementRollup.findMany.mockResolvedValue([]); prismaMock.moodEntry.findMany.mockResolvedValue([]); prismaMock.moodEntry.findFirst.mockResolvedValue(null); diff --git a/src/lib/insights/feature-blocks.ts b/src/lib/insights/feature-blocks.ts index 062d5b5ef..fcdf7ec74 100644 --- a/src/lib/insights/feature-blocks.ts +++ b/src/lib/insights/feature-blocks.ts @@ -10,6 +10,9 @@ * point (`extractFeatures`) stays in the hub. */ import { prisma } from "@/lib/db"; +import { Prisma } from "@/generated/prisma/client"; +import { buildSourceRankCase } from "@/lib/analytics/source-rank-sql"; +import { loadUserSourcePriority } from "@/lib/rollups/measurement-read"; import { classifyAgainstEffectiveRange } from "@/lib/labs/reference-range"; import { calendarDaysUntil } from "@/lib/measurement-reminders/due-day"; import { resolveLabFields } from "@/lib/labs/serialise"; @@ -33,10 +36,16 @@ interface AllTimeExtremes { /** * v1.18.11 P1 — full-history min / max / mean per measurement type via ONE - * grouped SQL aggregation, with NO row materialisation in JS. Used to fill the - * `allTime*` feature fields honestly when the bulk feature read is bounded to a - * recent window: the windowed `summarize()` covers trends + recent windows, and - * this covers the long-horizon extremes the prompt labels "allTime". + * grouped SQL aggregation, with NO row materialisation in JS. Fills the + * `allTime*` feature fields; the windowed `summarize()` covers trends + recent + * windows, and this covers the long-horizon figures the prompt labels + * "allTime". + * + * One source per (type, day): each source is folded per day first, each day + * keeps its ladder-canonical source, then the days fold into the figures — + * the same pick the dashboard's all-time figures make. Averaging every raw + * row let a device that samples densely (a watch's pulse) outweigh the + * readings the source ladder prefers on the same days. * * Only the four types that expose `allTime*` fields are aggregated (weight, * systolic, diastolic, pulse). Returns a map keyed by `MeasurementType`; a type @@ -46,19 +55,52 @@ export async function readAllTimeExtremes( userId: string, types: readonly MeasurementType[], ): Promise> { - const rows = await prisma.measurement.groupBy({ - by: ["type"], - where: { userId, deletedAt: null, type: { in: [...types] } }, - _avg: { value: true }, - _min: { value: true }, - _max: { value: true }, - }); const out = new Map(); + if (types.length === 0) return out; + const priorityJson = await loadUserSourcePriority(userId); + const rank = Prisma.raw( + buildSourceRankCase(priorityJson, 'p."type"', 'p."source"'), + ); + const typeList = Prisma.join( + types.map((t) => Prisma.sql`${t}::measurement_type`), + ); + const rows = await prisma.$queryRaw< + Array<{ type: string; mean: number; min: number; max: number }> + >` + WITH per_source AS ( + SELECT + m."type", + m."source", + date_trunc('day', m."measured_at") AS day, + 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."type" IN (${typeList}) + 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 + FROM per_source p + ORDER BY p."type", p.day, (${rank}), p."source"::text + ) + SELECT + c."type"::text AS type, + (SUM(c.total) / SUM(c.cnt))::double precision AS mean, + MIN(c.min_value)::double precision AS min, + MAX(c.max_value)::double precision AS max + FROM canon c + GROUP BY c."type" + `; for (const r of rows) { - out.set(r.type, { - mean: r._avg.value ?? null, - min: r._min.value ?? null, - max: r._max.value ?? null, + out.set(r.type as MeasurementType, { + mean: r.mean === null ? null : Number(r.mean), + min: r.min === null ? null : Number(r.min), + max: r.max === null ? null : Number(r.max), }); } return out; diff --git a/src/lib/openapi/routes/insights/schemas.ts b/src/lib/openapi/routes/insights/schemas.ts index a629ac1e2..2f7b934e6 100644 --- a/src/lib/openapi/routes/insights/schemas.ts +++ b/src/lib/openapi/routes/insights/schemas.ts @@ -913,13 +913,20 @@ export const analyticsRangeQuery = z export const analyticsWindowAggregate = z .object({ - count: z.number().int().describe("Reading count composed across buckets."), + count: z + .number() + .int() + .describe( + "Reading count composed across buckets. For step-like totals it is the number of days with a total, for SLEEP_DURATION the number of nights.", + ), min: z.number().nullable().describe("Window minimum; null when empty."), max: z.number().nullable().describe("Window maximum; null when empty."), mean: z .number() .nullable() - .describe("Count-weighted mean across buckets; null when empty."), + .describe( + "Count-weighted mean across buckets; null when empty. For step-like totals (steps, energy, distance, flights) it is the average daily total, for SLEEP_DURATION the average time asleep per night.", + ), sum: z .number() .nullable() @@ -941,7 +948,7 @@ export const analyticsRangeResponse = z granularity: z .string() .describe( - "Rollup granularity the read resolved against (`DAY` / `WEEK` / `MONTH` / `YEAR`, or `none` on a coverage miss).", + "Rollup granularity the read resolved against (`DAY` / `WEEK` / `MONTH` / `YEAR`), `live` for SLEEP_DURATION (nights are reconstructed from the stage rows), or `none` when there is no data.", ), current: analyticsWindowAggregate.describe( "Aggregate over the current window `[now-N, now)`.", @@ -965,7 +972,7 @@ export const analyticsRangeResponse = z .meta({ id: "AnalyticsRangeResponse", description: - "Single-metric period-over-period aggregate. Reads the current and previous comparable windows from the WMY rollup tier and composes a count-weighted-mean delta. `count/min/max/mean/sum` are linearly composable across buckets; SD/slope/r² are intentionally excluded (not composable).", + "Single-metric period-over-period aggregate. Reads the current and previous comparable windows from the rollup tier and composes a count-weighted-mean delta, one source per day. Step-like totals compare average daily totals and SLEEP_DURATION compares average nights, not per-reading means. `count/min/max/mean/sum` are linearly composable across buckets; SD/slope/r² are intentionally excluded (not composable).", }); export const insightsPregenerateRequest = z.object({}).meta({ diff --git a/tests/integration/rollup-per-day-canonical.test.ts b/tests/integration/rollup-per-day-canonical.test.ts index 8cc597e26..a720de64b 100644 --- a/tests/integration/rollup-per-day-canonical.test.ts +++ b/tests/integration/rollup-per-day-canonical.test.ts @@ -20,6 +20,8 @@ import { recomputeUserRollups, } from "@/lib/rollups/measurement-rollups"; import { readLiveBuckets } from "@/lib/measurements/daily-series-read"; +import { readAllTimeExtremes } from "@/lib/insights/feature-blocks"; +import { computeSummariesSlice } from "@/lib/analytics/summaries-slice"; import type { MeasurementType } from "@/generated/prisma/client"; vi.mock("@/lib/db-compat", () => ({ @@ -283,4 +285,85 @@ describe("coarse buckets — canonical source per day", () => { expect(rollupDay.mean).toBe(110); expect(liveMonth.value).toBe(110); }); + + it("all-time weight figures keep one source per day", async () => { + const user = await seedUser(); + const prisma = getPrismaClient(); + await prisma.measurement.createMany({ + data: [ + // Day A: the scale (WITHINGS leads the ladder) and a typed reading. + { + userId: user.id, + type: "WEIGHT", + value: 80, + unit: "kg", + source: "WITHINGS", + measuredAt: day(3, 7), + }, + { + userId: user.id, + type: "WEIGHT", + value: 90, + unit: "kg", + source: "MANUAL", + measuredAt: day(3, 9), + }, + // Day B: a typed reading only. + { + userId: user.id, + type: "WEIGHT", + value: 70, + unit: "kg", + source: "MANUAL", + measuredAt: day(4, 9), + }, + ], + }); + + const extremes = await readAllTimeExtremes(user.id, ["WEIGHT"]); + expect(extremes.get("WEIGHT")).toEqual({ mean: 75, min: 70, max: 80 }); + }); + + it("rows older than the rollup window count one source per day in the all-time figures", async () => { + const user = await seedUser(); + const prisma = getPrismaClient(); + const now = Date.now(); + const recent = new Date(now - 2 * DAY_MS); + const old = new Date(now - 6 * 365 * DAY_MS); + await prisma.measurement.createMany({ + data: [ + { + userId: user.id, + type: "WEIGHT", + value: 82, + unit: "kg", + source: "MANUAL", + measuredAt: recent, + }, + // Six years ago, one day seen by two sources. + { + userId: user.id, + type: "WEIGHT", + value: 90, + unit: "kg", + source: "WITHINGS", + measuredAt: old, + }, + { + userId: user.id, + type: "WEIGHT", + value: 100, + unit: "kg", + source: "MANUAL", + measuredAt: new Date(old.getTime() + 60_000), + }, + ], + }); + await recomputeUserRollups(user.id, { granularities: ["DAY"] }); + + const slice = await computeSummariesSlice(user.id); + expect(slice.summaries.WEIGHT.count).toBe(2); + expect(slice.summaries.WEIGHT.max).toBe(90); + expect(slice.summaries.WEIGHT.mean).toBe(86); + }); });