diff --git a/src/lib/doctor-report/__tests__/day-bucket-latest-and-count.test.ts b/src/lib/doctor-report/__tests__/day-bucket-latest-and-count.test.ts index feb000741..b83ac5e05 100644 --- a/src/lib/doctor-report/__tests__/day-bucket-latest-and-count.test.ts +++ b/src/lib/doctor-report/__tests__/day-bucket-latest-and-count.test.ts @@ -45,6 +45,8 @@ function bucket( clinicalTar2Count: 0, clinicalLowRiskSum: null, clinicalHighRiskSum: null, + hourMeanSum: null, + hourCount: null, }; } diff --git a/src/lib/doctor-report/collect.ts b/src/lib/doctor-report/collect.ts index 79aaacad5..0f850b658 100644 --- a/src/lib/doctor-report/collect.ts +++ b/src/lib/doctor-report/collect.ts @@ -78,6 +78,7 @@ import { loadLabResults, } from "./clinical-records"; import { dayKeyAsUtcMidnight } from "@/lib/tz/date-only"; +import { readingsMean } from "@/lib/measurements/day-mean"; const DENSE_REPORT_RAW_WINDOW_DAYS = 90; @@ -446,8 +447,19 @@ export async function collectDoctorReportData( const stats: Record = {}; for (const [type, entries] of Object.entries(byType)) { const values = entries.map((e) => e.value); + // A day of pulse is the mean of its local hours' means and the period + // average the mean of its days (`day-mean.ts`); every other type keeps the + // mean of its readings. The dense types' stats are replaced below. stats[type] = { - avg: values.reduce((a, b) => a + b, 0) / values.length, + avg: + readingsMean( + type, + entries.map((e) => ({ + value: e.value, + measuredAt: new Date(e.measuredAt), + })), + reportTz, + ) ?? 0, ...minMaxOf(values), count: values.length, latest: values[values.length - 1], diff --git a/src/lib/doctor-report/dense-buckets.ts b/src/lib/doctor-report/dense-buckets.ts index bd89d4dc1..8d05fdab9 100644 --- a/src/lib/doctor-report/dense-buckets.ts +++ b/src/lib/doctor-report/dense-buckets.ts @@ -13,6 +13,7 @@ */ import { Prisma } from "@/generated/prisma/client"; import { prisma } from "@/lib/db"; +import { isHourlyMeanTypeSql } from "@/lib/measurements/day-mean"; import type { GlucoseContext, MeasurementSource, @@ -44,6 +45,13 @@ export interface DenseMeasurementBucket { clinicalTar2Count: number; clinicalLowRiskSum: number | null; clinicalHighRiskSum: number | null; + /** + * For an hourly-mean type (pulse, see `day-mean.ts`): the sum of the + * bucket's local hours' means and the number of those hours, so the day + * value is their ratio. `null` for every other type. + */ + hourMeanSum: number | null; + hourCount: number | null; } /** @@ -65,6 +73,8 @@ export async function loadDenseMeasurementBuckets(params: { const typeList = Prisma.join( denseTypes.map((type) => Prisma.sql`${type}::"measurement_type"`), ); + // Closed code constant (the hourly-mean type list), spliced as literals. + const hourlyMeanType = Prisma.raw(isHourlyMeanTypeSql('"type"')); return prisma.$queryRaw` WITH localized AS ( SELECT @@ -78,7 +88,11 @@ export async function loadDenseMeasurementBuckets(params: { date_trunc( 'day', (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${reportTz} - ) AS local_day + ) AS local_day, + date_trunc( + 'hour', + (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${reportTz} + ) AS local_hour FROM measurements m WHERE m."user_id" = ${userId} AND m."measured_at" >= ${start} @@ -121,6 +135,43 @@ export async function loadDenseMeasurementBuckets(params: { "measured_at" DESC, "id" DESC ), + -- The hourly-mean day (pulse): each local hour's mean, then their sum + -- and count per bucket. Empty for every other type. + hour_means AS ( + SELECT + local_day, + "type", + "source", + "device_type", + "glucose_context", + AVG("value") AS hour_mean + FROM localized + WHERE ${hourlyMeanType} + GROUP BY + local_day, + local_hour, + "type", + "source", + "device_type", + "glucose_context" + ), + hour_agg AS ( + SELECT + local_day, + "type", + "source", + "device_type", + "glucose_context", + SUM(hour_mean)::double precision AS hour_mean_sum, + COUNT(*)::int AS hour_count + FROM hour_means + GROUP BY + local_day, + "type", + "source", + "device_type", + "glucose_context" + ), grouped AS ( SELECT local_day, @@ -196,7 +247,9 @@ export async function loadDenseMeasurementBuckets(params: { grouped.clinical_tar1_count AS "clinicalTar1Count", grouped.clinical_tar2_count AS "clinicalTar2Count", grouped.clinical_low_risk_sum AS "clinicalLowRiskSum", - grouped.clinical_high_risk_sum AS "clinicalHighRiskSum" + grouped.clinical_high_risk_sum AS "clinicalHighRiskSum", + hour_agg.hour_mean_sum AS "hourMeanSum", + hour_agg.hour_count AS "hourCount" FROM grouped INNER JOIN latest_rows ON latest_rows.local_day = grouped.local_day @@ -204,6 +257,12 @@ export async function loadDenseMeasurementBuckets(params: { AND latest_rows."source" = grouped."source" AND latest_rows."device_type" IS NOT DISTINCT FROM grouped."device_type" AND latest_rows."glucose_context" IS NOT DISTINCT FROM grouped."glucose_context" + LEFT JOIN hour_agg + ON hour_agg.local_day = grouped.local_day + AND hour_agg."type" = grouped."type" + AND hour_agg."source" = grouped."source" + AND hour_agg."device_type" IS NOT DISTINCT FROM grouped."device_type" + AND hour_agg."glucose_context" IS NOT DISTINCT FROM grouped."glucose_context" ORDER BY "bucketStart" ASC, grouped."type" ASC `; diff --git a/src/lib/doctor-report/measurement-series.ts b/src/lib/doctor-report/measurement-series.ts index 1f0d0a6ef..c00d76eb0 100644 --- a/src/lib/doctor-report/measurement-series.ts +++ b/src/lib/doctor-report/measurement-series.ts @@ -10,6 +10,7 @@ */ import { collapseMeasurementsToCanonical } from "@/lib/doctor-report-helpers"; import { glucoseContextBucket } from "@/lib/glucose"; +import { usesHourlyMeanDay } from "@/lib/measurements/day-statistic"; import type { DoctorReportStats } from "@/lib/doctor-report-types"; import type { GlucoseClinicalMetrics } from "@/lib/analytics/glucose-metrics"; import type { DenseMeasurementBucket } from "./dense-buckets"; @@ -61,6 +62,8 @@ export function summariseDenseBuckets( max: number; latestAt: Date; latest: number; + hourMeanSum: number; + hourCount: number; } >(); for (const row of canonical) { @@ -76,11 +79,15 @@ export function summariseDenseBuckets( max: row.maxValue, latestAt: row.latestAt, latest: row.latestValue, + hourMeanSum: row.hourMeanSum ?? 0, + hourCount: row.hourCount ?? 0, }); continue; } current.count += row.count; current.sum += row.sumValue; + current.hourMeanSum += row.hourMeanSum ?? 0; + current.hourCount += row.hourCount ?? 0; current.min = Math.min(current.min, row.minValue); current.max = Math.max(current.max, row.maxValue); if (row.latestAt > current.latestAt) { @@ -109,12 +116,26 @@ export function summariseDenseBuckets( } }; + // For an hourly-mean type (pulse, see `day-mean.ts`) a day is the mean of + // its local hours' means and the period average the mean of its days, so a + // workout hour's dense samples do not outweigh the rest of the day, nor a + // workout day the rest of the period. Count, min and max stay over every + // reading. Every other type keeps the mean of its readings. + const dayMeans = new Map(); for (const row of byTypeDay.values()) { + const value = + usesHourlyMeanDay(row.type) && row.hourCount > 0 + ? row.hourMeanSum / row.hourCount + : row.sum / row.count; (byType[row.type] ??= []).push({ - value: row.sum / row.count, + value, measuredAt: row.bucketStart.toISOString(), }); accumulate(statsAcc, row.type, row); + const dm = dayMeans.get(row.type) ?? { sum: 0, days: 0 }; + dm.sum += value; + dm.days += 1; + dayMeans.set(row.type, dm); } for (const row of canonical) { @@ -137,9 +158,13 @@ export function summariseDenseBuckets( } const stats: DenseMeasurementSummary["stats"] = {}; for (const [key, value] of statsAcc) { + const dm = dayMeans.get(key); stats[key] = { count: value.count, - avg: value.sum / value.count, + avg: + usesHourlyMeanDay(key) && dm !== undefined && dm.days > 0 + ? dm.sum / dm.days + : value.sum / value.count, min: value.min, max: value.max, latest: value.latest, diff --git a/src/lib/illness/__tests__/correlation-read-pulse-hourly.test.ts b/src/lib/illness/__tests__/correlation-read-pulse-hourly.test.ts new file mode 100644 index 000000000..502ea27e0 --- /dev/null +++ b/src/lib/illness/__tests__/correlation-read-pulse-hourly.test.ts @@ -0,0 +1,86 @@ +/** + * The illness read's live per-day pulse is the mean of the day's local hours' + * means, so a workout hour's dense samples do not read as a fever-day pulse + * spike. Other vitals keep the mean of their readings. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const NOW = new Date("2026-01-21T12:00:00Z"); + +const db = vi.hoisted(() => ({ + measurement: { findMany: vi.fn() }, + illnessDayLog: { findMany: vi.fn() }, +})); +const engine = vi.hoisted(() => ({ compute: vi.fn() })); + +vi.mock("@/lib/db", () => ({ prisma: db })); +vi.mock("@/lib/rollups/measurement-coverage", () => ({ + probeRollupCoverage: vi.fn(async () => new Map()), +})); +vi.mock("@/lib/rollups/measurement-read-wmy", () => ({ + readBestGranularityRollups: vi.fn(async () => null), +})); +vi.mock("@/lib/rollups/measurement-read", () => ({ + loadUserSourcePriority: vi.fn(async () => null), +})); +vi.mock("../correlation", async (orig) => ({ + ...(await orig()), + computeIllnessCorrelation: (input: unknown) => { + engine.compute(input); + return { status: "insufficient" }; + }, +})); + +import { computeEpisodeCorrelation } from "../correlation-read"; + +/** A workout hour (six readings at 150) and two resting hours at 60, in Kolkata local time. */ +const DAY_ROWS = [ + ...[0, 5, 10, 15, 20, 25].map((m) => ({ + value: 150, + // 04:45Z + m = 10:15 + m local (one local hour, two UTC hours). + measuredAt: new Date(Date.parse("2026-01-15T04:45:00Z") + m * 60_000), + })), + { value: 60, measuredAt: new Date("2026-01-15T06:30:00Z") }, + { value: 60, measuredAt: new Date("2026-01-15T08:30:00Z") }, +]; + +beforeEach(() => { + vi.clearAllMocks(); + db.illnessDayLog.findMany.mockResolvedValue([]); + db.measurement.findMany.mockImplementation( + async ({ where }: { where: { type: string } }) => + where.type === "PULSE" || where.type === "HEART_RATE_VARIABILITY" + ? DAY_ROWS + : [], + ); +}); + +describe("illness read, live per-day means", () => { + it("makes a pulse day the mean of its local hours' means; HRV keeps the reading mean", async () => { + await computeEpisodeCorrelation( + "u1", + { + id: "ep", + onsetAt: new Date("2026-01-16T08:00:00Z"), + resolvedAt: null, + lifecycle: "ACUTE", + }, + "Asia/Kolkata", + NOW, + ); + const { series } = engine.compute.mock.calls[0][0] as { + series: Array<{ + type: string; + episodeDays: Array<{ day: string; mean: number }>; + episodeDayMax: Array<{ day: string; mean: number }>; + }>; + }; + const pulse = series.find((s) => s.type === "PULSE")!; + // Hours' means 150, 60, 60 → 90 (the readings average 127.5). + expect(pulse.episodeDays).toEqual([{ day: "2026-01-15", mean: 90 }]); + // The day's max stays over every reading. + expect(pulse.episodeDayMax).toEqual([{ day: "2026-01-15", mean: 150 }]); + const hrv = series.find((s) => s.type === "HEART_RATE_VARIABILITY")!; + expect(hrv.episodeDays).toEqual([{ day: "2026-01-15", mean: 127.5 }]); + }); +}); diff --git a/src/lib/illness/correlation-read.ts b/src/lib/illness/correlation-read.ts index 08b5840bf..aa50005d1 100644 --- a/src/lib/illness/correlation-read.ts +++ b/src/lib/illness/correlation-read.ts @@ -30,6 +30,7 @@ import { prisma } from "@/lib/db"; import { readBestGranularityRollups } from "@/lib/rollups/measurement-read-wmy"; import { probeRollupCoverage } from "@/lib/rollups/measurement-coverage"; import { dayKeyForUserTz } from "@/lib/measurements/consolidation-tz"; +import { dayValue } from "@/lib/measurements/day-mean"; import { resolveUserTimezone } from "@/lib/measurements/consolidation-base"; import { reconstructNights } from "@/lib/insights/derived/sleep-score"; import { loadUserSourcePriority } from "@/lib/rollups/measurement-read"; @@ -250,18 +251,24 @@ async function readWindowMeans( select: { value: true, measuredAt: true }, }); if (rows.length === 0) return { points: [], source: "none" }; - const byDay = new Map(); + const byDay = new Map(); for (const row of rows) { const day = dayKeyForUserTz(row.measuredAt, tz); - const acc = byDay.get(day) ?? { sum: 0, count: 0, max: -Infinity }; - acc.sum += row.value; - acc.count += 1; + const acc = byDay.get(day) ?? { rows: [], max: -Infinity }; + acc.rows.push(row); acc.max = Math.max(acc.max, row.value); byDay.set(day, acc); } + // The day's mean is the type's day value (`day-mean.ts`): the mean of its + // local hours' means for pulse, the mean of its readings otherwise. The + // max stays over every reading. const points = [...byDay.entries()] .sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0)) - .map(([day, acc]) => ({ day, mean: acc.sum / acc.count, max: acc.max })); + .map(([day, acc]) => ({ + day, + mean: dayValue(type, acc.rows, tz) as number, + max: acc.max, + })); return { points, source: "live" }; } diff --git a/src/lib/insights/__tests__/correlation-channel-series-rollup-swap.test.ts b/src/lib/insights/__tests__/correlation-channel-series-rollup-swap.test.ts index de6d72f18..e5a301014 100644 --- a/src/lib/insights/__tests__/correlation-channel-series-rollup-swap.test.ts +++ b/src/lib/insights/__tests__/correlation-channel-series-rollup-swap.test.ts @@ -239,6 +239,50 @@ describe("rollup ↔ raw parity — multi-source day (source collapse)", () => { }); }); +describe("pulse from the DAY rollup tier", () => { + it("reads each source's stored DAY mean (the hourly-mean day), not its readings' sum over count", async () => { + mockCoverage([["PULSE", true]]); + // The stored DAY mean of pulse is the mean of the day's hours' means (90), + // while its readings average 127.5. A second source on another day is + // single-source and kept as is. + rollupFindMany.mockResolvedValue([ + { + type: "PULSE", + bucketStart: new Date(`${utcDayKey(at(8, 7))}T00:00:00.000Z`), + count: 8, + mean: 90, + sumValue: 1020, + }, + ]); + const result = await fetchMeasurementDailySeriesTiered( + USER, + "Europe/Berlin", + SINCE(), + ["PULSE"], + ); + expect(result.rollupTypes).toEqual(["PULSE"]); + expect(result.byType.get("PULSE")).toEqual([ + { day: utcDayKey(at(8, 7)), value: 90 }, + ]); + }); + + it("weighs each source's day once on a day two sources report", async () => { + mockCoverage([["PULSE", true]]); + const bucketStart = new Date(`${utcDayKey(at(8, 7))}T00:00:00.000Z`); + rollupFindMany.mockResolvedValue([ + { type: "PULSE", bucketStart, count: 900, mean: 70, sumValue: 900 * 120 }, + { type: "PULSE", bucketStart, count: 3, mean: 60, sumValue: 180 }, + ]); + const result = await fetchMeasurementDailySeriesTiered( + USER, + "Europe/Berlin", + SINCE(), + ["PULSE"], + ); + expect(result.byType.get("PULSE")![0].value).toBeCloseTo(65, 10); + }); +}); + describe("rollup ↔ raw parity — far-from-UTC profile timezone", () => { const savedTz = process.env.TZ; afterAll(() => { diff --git a/src/lib/insights/__tests__/correlation-channel-series.test.ts b/src/lib/insights/__tests__/correlation-channel-series.test.ts index 14a1dce74..e2026f620 100644 --- a/src/lib/insights/__tests__/correlation-channel-series.test.ts +++ b/src/lib/insights/__tests__/correlation-channel-series.test.ts @@ -145,6 +145,39 @@ describe("buildMeasurementDailySeries — grain consistency (v1.29.6)", () => { expect(points).toEqual([{ day: "2026-06-04", value: 62 }]); }); + + it("makes a pulse day the mean of its local hours' means, not of its readings", () => { + // A dense workout hour (six readings at 150) and two resting hours at 60. + // The plain mean would be (6 * 150 + 120) / 8 = 127.5; each hour once + // gives (150 + 60 + 60) / 3 = 90. + const rows: MeasurementSeriesRow[] = [ + ...[0, 10, 20, 30, 40, 50].map((m) => + row(`2026-06-04T10:${String(m).padStart(2, "0")}:00.000Z`, 150), + ), + row("2026-06-04T12:00:00.000Z", 60), + row("2026-06-04T14:00:00.000Z", 60), + ]; + expect(buildMeasurementDailySeries("PULSE", rows, "UTC", null)).toEqual([ + { day: "2026-06-04", value: 90 }, + ]); + // HRV keeps the plain mean of its readings. + expect( + buildMeasurementDailySeries("HEART_RATE_VARIABILITY", rows, "UTC", null), + ).toEqual([{ day: "2026-06-04", value: 127.5 }]); + }); + + it("reads the pulse hours on the profile zone's own clock", () => { + // Asia/Kolkata (UTC+05:30): 10:20Z and 10:40Z are 15:50 and 16:10 local, + // two local hours (100 and 40), though one UTC hour. + const rows: MeasurementSeriesRow[] = [ + row("2026-06-04T10:20:00.000Z", 100), + row("2026-06-04T10:25:00.000Z", 100), + row("2026-06-04T10:40:00.000Z", 40), + ]; + expect( + buildMeasurementDailySeries("PULSE", rows, "Asia/Kolkata", null), + ).toEqual([{ day: "2026-06-04", value: 70 }]); + }); }); describe("fetchMeasurementWindowSeries — desc+cap+resort (v1.30.3 QA F1/F2/F3)", () => { diff --git a/src/lib/insights/correlation-channel-series.ts b/src/lib/insights/correlation-channel-series.ts index 55d37773f..ecbca30b3 100644 --- a/src/lib/insights/correlation-channel-series.ts +++ b/src/lib/insights/correlation-channel-series.ts @@ -63,6 +63,8 @@ import type { } from "@/generated/prisma/client"; import { TRACKED_INTAKE_WHERE } from "@/lib/medications/intake-tracking"; import { dateOnlyKey } from "@/lib/tz/date-only"; +import { hourMeanRows } from "@/lib/measurements/day-mean"; +import { usesHourlyMeanDay } from "@/lib/measurements/day-statistic"; import { decryptSymptomText } from "@/lib/symptoms/server"; import { MAX_ACTIVE_SYMPTOM_DEFINITIONS, @@ -689,8 +691,15 @@ export function buildMeasurementDailySeries( if (CUMULATIVE_HK_TYPES.has(type)) { return buildCumulativeDailySeries(type, rows, tz, priorityJson); } + // A day of pulse is the mean of its local hours' means (`day-mean.ts`): + // collapse each local hour to its mean first, so the daily mean below + // weighs every hour once. Every other type passes through unchanged. return toDailyMeans( - rows.map((r) => ({ value: r.value, at: r.at })), + hourMeanRows( + type, + rows.map((r) => ({ value: r.value, measuredAt: r.at })), + tz, + ).map((r) => ({ value: r.value, at: r.measuredAt })), tz, ); } @@ -808,6 +817,7 @@ function isRollupSwapEligible(type: MeasurementType): boolean { * mean exactly up to float addition order. */ function composeRollupDailyMeans( + type: MeasurementType, rows: Array<{ bucketStart: Date; count: number; @@ -815,10 +825,14 @@ function composeRollupDailyMeans( sumValue: number | null; }>, ): DailySeriesPoint[] { + // A pulse DAY row's stored `mean` is the mean of its hours' means + // (`day-mean.ts`), which `sumValue / count` would undo; each source's day + // then weighs one. Every other type keeps the count-weighted compose. + const hourly = usesHourlyMeanDay(type); const byDay = new Map(); for (const r of rows) { if (r.count <= 0) continue; - const sum = r.sumValue ?? r.mean * r.count; + const sum = hourly ? r.mean : (r.sumValue ?? r.mean * r.count); if (!Number.isFinite(sum)) continue; // DAY buckets are minted at UTC midnight; the UTC date IS the bucket's // day key (the same convention graded-series and the derived baselines @@ -826,7 +840,7 @@ function composeRollupDailyMeans( const day = dateOnlyKey(r.bucketStart); const acc = byDay.get(day) ?? { sum: 0, count: 0 }; acc.sum += sum; - acc.count += r.count; + acc.count += hourly ? 1 : r.count; byDay.set(day, acc); } return [...byDay.entries()] @@ -920,7 +934,10 @@ export async function fetchMeasurementDailySeriesTiered( rowsByType.set(bucket.type, list); } for (const type of covered) { - const series = composeRollupDailyMeans(rowsByType.get(type) ?? []); + const series = composeRollupDailyMeans( + type, + rowsByType.get(type) ?? [], + ); if (series.length === 0) { // Coverage said "has buckets" but the window resolved empty — // backfill pending or all data older than the window. Fall back diff --git a/src/lib/insights/derived/baseline.ts b/src/lib/insights/derived/baseline.ts index c267e44c4..a53f5675b 100644 --- a/src/lib/insights/derived/baseline.ts +++ b/src/lib/insights/derived/baseline.ts @@ -282,7 +282,12 @@ export async function readDayMeanSeries( timeZone: timeZone ?? "UTC", valueRange: plausibleMetricRange(type), }); - const points = days.map((d) => ({ day: d.day, mean: d.sum / d.n })); + // A pulse day is the mean of its local hours' means (`dayMean`, see + // `day-mean.ts`); every other type's day is the mean of its readings. + const points = days.map((d) => ({ + day: d.day, + mean: d.dayMean ?? d.sum / d.n, + })); if (points.length === 0) { return { points: [], source: "none" }; } diff --git a/src/lib/measurements/__tests__/fake-day-aggregates.ts b/src/lib/measurements/__tests__/fake-day-aggregates.ts index 096727fb9..dd17f37ef 100644 --- a/src/lib/measurements/__tests__/fake-day-aggregates.ts +++ b/src/lib/measurements/__tests__/fake-day-aggregates.ts @@ -19,6 +19,8 @@ import type { SourceDayAggregateRow, } from "@/lib/measurements/day-aggregates"; import { userDayKey } from "@/lib/tz/format"; +import { dayValue } from "@/lib/measurements/day-mean"; +import { usesHourlyMeanDay } from "@/lib/measurements/day-statistic"; export function foldDayAggregates( rows: ReadonlyArray<{ measuredAt: Date; value: number }>, @@ -74,7 +76,30 @@ export async function fakeReadDayAggregates( orderBy: { measuredAt: "asc" }, select: { measuredAt: true, value: true }, })) ?? []) as Array<{ measuredAt: Date; value: number }>; - return foldDayAggregates(rows, opts); + const days = foldDayAggregates(rows, opts); + if (!usesHourlyMeanDay(opts.type)) return days; + // Mirror the SQL's hourly-mean day: the mean of the (day, segment)'s local + // hours' means. + const starts = opts.segmentStarts ?? []; + return days.map((d) => { + const members = rows.filter((r) => { + const t = r.measuredAt.getTime(); + if (t < opts.since.getTime()) return false; + if (opts.until && t > opts.until.getTime()) return false; + if ( + opts.valueRange && + (r.value < opts.valueRange.min || r.value > opts.valueRange.max) + ) { + return false; + } + return ( + userDayKey(r.measuredAt, opts.timeZone) === d.day && + starts.filter((s) => t < s.getTime()).length === d.segment + ); + }); + const dayMean = dayValue(opts.type, members, opts.timeZone); + return dayMean === null ? d : { ...d, dayMean }; + }); } /** diff --git a/src/lib/measurements/day-aggregates.ts b/src/lib/measurements/day-aggregates.ts index 48d2966b1..d29b71cfa 100644 --- a/src/lib/measurements/day-aggregates.ts +++ b/src/lib/measurements/day-aggregates.ts @@ -27,6 +27,8 @@ import type { import { Prisma } from "@/generated/prisma/client"; import { prisma } from "@/lib/db"; import { isValidTimezone } from "@/lib/tz/format"; +import { dayWeightedRowsSql } from "@/lib/measurements/day-mean"; +import { usesHourlyMeanDay } from "@/lib/measurements/day-statistic"; export interface DayAggregateRow { /** Calendar day in the requested zone, `YYYY-MM-DD`. */ @@ -37,6 +39,21 @@ export interface DayAggregateRow { sum: number; min: number; max: number; + /** + * For a type whose day is the mean of its hours' means (pulse, see + * `day-mean.ts`): `SUM(value * day_weight)` and `SUM(day_weight)` over the + * row's readings. A whole day's weights add up to one, so `weightedSum / + * weightSum` is the day value and the same ratio over several rows the mean + * of their days. Absent for every other type. + */ + weightedSum?: number; + weightSum?: number; + /** + * The row's value for an hourly-mean type, `weightedSum / weightSum`: the + * mean of its local hours' means. Absent for every other type, whose value + * stays `sum / n`. Read it as `dayMean ?? sum / n`. + */ + dayMean?: number; } export interface ReadDayAggregatesOptions { @@ -78,6 +95,10 @@ export async function readDayAggregates( " + ", )})`; + if (usesHourlyMeanDay(opts.type)) { + return readHourlyMeanDayAggregates(opts, timeZone, until, range, segment); + } + // The day and segment are computed once per row in `src` and grouped by // name, so the bound zone / boundary parameters never have to match // themselves across SELECT and GROUP BY. @@ -125,6 +146,74 @@ export async function readDayAggregates( })); } +/** + * `readDayAggregates` for an hourly-mean type: the same rows, plus the day + * weights of `day-mean.ts` cut in the same zone, so a day's hours count once + * each however densely they were sampled. + */ +async function readHourlyMeanDayAggregates( + opts: ReadDayAggregatesOptions, + timeZone: string, + until: Prisma.Sql, + range: Prisma.Sql, + segment: Prisma.Sql, +): Promise { + const rows = await (opts.db ?? prisma).$queryRaw< + Array<{ + day: string; + segment: number; + n: number; + sum: number; + min: number; + max: number; + wsum: number; + wdays: number; + }> + >` + WITH base AS ( + SELECT m.* + FROM measurements m + WHERE m."user_id" = ${opts.userId} + AND m."type" = ${opts.type}::measurement_type + AND m."deleted_at" IS NULL + AND m."measured_at" >= ${opts.since} + ${until} + ${range} + ), + src AS ( + SELECT + to_char((m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}, 'YYYY-MM-DD') AS day, + ${segment} AS segment, + m."value" AS value, + m.day_weight AS day_weight + FROM ${dayWeightedRowsSql("base", timeZone)} m + ) + SELECT + day, + segment::int AS segment, + COUNT(*)::int AS n, + SUM(value)::double precision AS sum, + MIN(value)::double precision AS min, + MAX(value)::double precision AS max, + SUM(value * day_weight)::double precision AS wsum, + SUM(day_weight)::double precision AS wdays + FROM src + GROUP BY day, segment + ORDER BY day ASC, segment DESC + `; + return rows.map((r) => ({ + day: r.day, + segment: Number(r.segment), + n: Number(r.n), + sum: Number(r.sum), + min: Number(r.min), + max: Number(r.max), + weightedSum: Number(r.wsum), + weightSum: Number(r.wdays), + dayMean: Number(r.wsum) / Number(r.wdays), + })); +} + export interface SourceDayAggregateRow { type: MeasurementType; /** Calendar day in the requested zone, `YYYY-MM-DD`. */ diff --git a/src/lib/rollups/tiered-context.ts b/src/lib/rollups/tiered-context.ts index a5b29605f..d9d815c2e 100644 --- a/src/lib/rollups/tiered-context.ts +++ b/src/lib/rollups/tiered-context.ts @@ -40,6 +40,9 @@ import { } from "./measurement-rollups"; import { probeRollupCoverage } from "./measurement-coverage"; import { readDayAggregates } from "@/lib/measurements/day-aggregates"; +import { dayWeightedRowsSql, windowMeanSql } from "@/lib/measurements/day-mean"; +import { usesHourlyMeanDay } from "@/lib/measurements/day-statistic"; +import { Prisma } from "@/generated/prisma/client"; const DAY_MS = 24 * 60 * 60 * 1000; @@ -270,9 +273,11 @@ async function readRecentDaily( since, timeZone: tz, }); + // A pulse day is the mean of its local hours' means (`dayMean`, see + // `day-mean.ts`); every other type's day is the mean of its readings. return days.map((d) => ({ date: d.day, - value: round(d.sum / d.n), + value: round(d.dayMean ?? d.sum / d.n), count: d.n, })); } @@ -316,6 +321,9 @@ async function readBandLive( tzKey?: string, ): Promise { const unit = BAND_TRUNC_UNIT[granularity]; + if (usesHourlyMeanDay(type)) { + return readHourlyMeanBandLive(userId, type, unit, from, to, tzKey); + } try { const rows = tzKey ? await prisma.$queryRaw< @@ -387,6 +395,78 @@ async function readBandLive( } } +/** + * `readBandLive` for a type whose day is the mean of its hours' means (pulse, + * `day-mean.ts`): a bucket's mean is the mean of its days, each the mean of + * its hours, so a workout's dense samples do not outweigh the rest of the + * bucket. Count, min, max and the spread stay over every reading. The bucket + * boundary is the one the plain branch uses; hours and days are cut in the + * user's zone when `tzKey` is set, in the session's (UTC) frame otherwise. + */ +async function readHourlyMeanBandLive( + userId: string, + type: MeasurementType, + unit: "day" | "week" | "month" | "year", + from: Date, + to: Date, + tzKey?: string, +): Promise { + const mean = Prisma.raw( + windowMeanSql({ + typeColumn: 'm."type"', + value: 'm."value"', + weight: "m.day_weight", + }), + ); + const bucket = tzKey + ? Prisma.sql`date_trunc(${unit}, m."measured_at" AT TIME ZONE ${tzKey})` + : Prisma.sql`date_trunc(${unit}, m."measured_at")`; + try { + const rows = await prisma.$queryRaw< + Array<{ + bucket_start: Date; + count: number; + mean: number; + min_value: number; + max_value: number; + sd: number | null; + }> + >` + WITH src AS ( + SELECT * FROM measurements m + WHERE m."user_id" = ${userId} + AND m."type" = ${type}::"measurement_type" + AND m."deleted_at" IS NULL + AND m."measured_at" >= ${from} + AND m."measured_at" < ${to} + ) + SELECT + ${bucket} AS bucket_start, + COUNT(*)::int AS count, + ${mean}::double precision AS mean, + MIN(m."value")::double precision AS min_value, + MAX(m."value")::double precision AS max_value, + STDDEV_POP(m."value")::double precision AS sd + FROM ${dayWeightedRowsSql("src", tzKey ?? null)} m + GROUP BY 1, m."type" + ORDER BY 1 + `; + return rows.map((r) => ({ + bucketStart: new Date(r.bucket_start), + count: Number(r.count), + mean: Number(r.mean), + minValue: Number(r.min_value), + maxValue: Number(r.max_value), + sd: r.sd === null ? null : Number(r.sd), + slope: null, + r2: null, + })); + } catch { + // Degrades to empty, like the plain branch. + return []; + } +} + async function readBand( userId: string, type: MeasurementType, diff --git a/tests/integration/day-aggregates-pulse-hourly.test.ts b/tests/integration/day-aggregates-pulse-hourly.test.ts new file mode 100644 index 000000000..238dcaeaf --- /dev/null +++ b/tests/integration/day-aggregates-pulse-hourly.test.ts @@ -0,0 +1,175 @@ +/** + * The per-day fold the background readers share (`readDayAggregates`) carries + * the hourly-mean day for pulse, cut on the requested zone's own hours and + * days; the vitals baseline and the tiered context read their live pulse days + * from it. + * + * Asia/Kolkata (UTC+05:30). Day one: six readings at 150 from 04:45Z to + * 04:55Z and two at 150 at 05:05Z and 05:10Z (all local 10:15 to 10:40, one + * local hour, two UTC hours), then one reading at 60 in each of local hours + * 12, 14 and 16. Hours' means 150, 60, 60, 60, day value 82.5; the readings + * average (8 x 150 + 180) / 11 = 125.45. Day two: one reading at 60. + * + * A second workout day ten days earlier (12:45Z to 13:10Z, local 18:15 to + * 18:40, then one reading at 60 at 14:00Z, 15:00Z and 16:00Z) sits in the + * tiered context's day band, which a zone this far from UTC reads live. + */ +import { beforeAll, describe, expect, it } from "vitest"; + +import type { Prisma } from "@/generated/prisma/client"; +import { readDayAggregates } from "@/lib/measurements/day-aggregates"; +import { readDayMeanSeries } from "@/lib/insights/derived/baseline"; +import { buildTieredSeries } from "@/lib/rollups/tiered-context"; +import { invalidateUserTimezone } from "@/lib/tz/resolver"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +const prisma = getPrismaClient(); +const USER = "day-aggregates-pulse-hourly"; +const TZ = "Asia/Kolkata"; +const now = new Date(); +const D = Date.UTC( + now.getUTCFullYear(), + now.getUTCMonth(), + now.getUTCDate() - 10, +); +const at = (h: number, m: number) => new Date(D + h * 3_600_000 + m * 60_000); +const dayKey = (ms: number) => + new Intl.DateTimeFormat("en-CA", { timeZone: TZ }).format(new Date(ms)); + +beforeAll(async () => { + await truncateAllTables(prisma); + await prisma.user.create({ + data: { id: USER, username: USER, timezone: TZ }, + }); + invalidateUserTimezone(USER); + const readings: Array<[Date, number]> = [ + ...[45, 47, 49, 51, 53, 55].map((m): [Date, number] => [at(4, m), 150]), + [at(5, 5), 150], + [at(5, 10), 150], + [at(6, 30), 60], + [at(8, 30), 60], + [at(10, 30), 60], + [at(24 + 4, 0), 60], + ...[45, 47, 49, 51, 53, 55, 65, 70].map((m): [Date, number] => [ + at(-10 * 24 + 12, m), + 150, + ]), + [at(-10 * 24 + 14, 0), 60], + [at(-10 * 24 + 15, 0), 60], + [at(-10 * 24 + 16, 0), 60], + ]; + const rows: Prisma.MeasurementCreateManyInput[] = []; + for (const type of ["PULSE", "HEART_RATE_VARIABILITY"] as const) { + readings.forEach(([measuredAt, value], i) => + rows.push({ + id: `${type}-${i}`, + userId: USER, + type, + unit: type === "PULSE" ? "bpm" : "ms", + source: "APPLE_HEALTH", + value, + measuredAt, + }), + ); + } + await prisma.measurement.createMany({ data: rows }); +}, 120_000); + +describe("readDayAggregates", () => { + it("carries the hourly-mean day for pulse, on the zone's own hours", async () => { + const days = await readDayAggregates({ + userId: USER, + type: "PULSE", + since: new Date(D - 86_400_000), + timeZone: TZ, + }); + expect(days.map((d) => d.day)).toEqual([ + dayKey(D + 5 * 3_600_000), + dayKey(D + 28 * 3_600_000), + ]); + expect(days[0]).toMatchObject({ n: 11, min: 60, max: 150 }); + expect(days[0].sum).toBe(8 * 150 + 3 * 60); + expect(days[0].dayMean).toBeCloseTo(82.5, 9); + expect(days[1].dayMean).toBeCloseTo(60, 9); + }); + + it("splits the hours per segment", async () => { + const days = await readDayAggregates({ + userId: USER, + type: "PULSE", + since: new Date(D - 86_400_000), + timeZone: TZ, + // Readings before 08:00Z (the workout and local hour 12) fall into + // segment 1, the rest of the day into segment 0. + segmentStarts: [at(8, 0)], + }); + const first = days.filter((d) => d.day === dayKey(D + 5 * 3_600_000)); + expect(first.map((d) => [d.segment, d.dayMean])).toEqual([ + [1, expect.closeTo(105, 9)], + [0, expect.closeTo(60, 9)], + ]); + }); + + it("leaves every other type without a day mean", async () => { + const days = await readDayAggregates({ + userId: USER, + type: "HEART_RATE_VARIABILITY", + since: new Date(D - 86_400_000), + timeZone: TZ, + }); + expect(days[0].dayMean ?? null).toBeNull(); + }); +}); + +describe("live pulse days of the readers built on it", () => { + it("the vitals baseline", async () => { + const { points, source } = await readDayMeanSeries( + USER, + "PULSE", + 30, + now, + new Map(), + TZ, + ); + expect(source).toBe("live"); + expect(points.map((p) => p.mean)).toEqual([ + expect.closeTo(82.5, 9), + expect.closeTo(82.5, 9), + expect.closeTo(60, 9), + ]); + const hrv = await readDayMeanSeries( + USER, + "HEART_RATE_VARIABILITY", + 30, + now, + new Map(), + TZ, + ); + expect(hrv.points[0].mean).toBeCloseTo((8 * 150 + 3 * 60) / 11, 9); + }); + + it("the tiered context's recent days and its live day band", async () => { + const series = await buildTieredSeries(USER, "PULSE", { + now: now.getTime(), + tz: TZ, + skipEnsureFresh: true, + }); + expect(series.recentDaily.map((p) => [p.value, p.count])).toEqual([ + [82.5, 11], + [60, 1], + ]); + const workout = series.dayBand.find((b) => b.count === 11); + expect(workout?.mean).toBeCloseTo(82.5, 2); + const hrv = await buildTieredSeries(USER, "HEART_RATE_VARIABILITY", { + now: now.getTime(), + tz: TZ, + skipEnsureFresh: true, + }); + expect(hrv.recentDaily[0].value).toBeCloseTo((8 * 150 + 180) / 11, 2); + expect(hrv.dayBand.find((b) => b.count === 11)?.mean).toBeCloseTo( + (8 * 150 + 180) / 11, + 2, + ); + }); +}); diff --git a/tests/integration/dense-read-bounded.test.ts b/tests/integration/dense-read-bounded.test.ts index 6357877b7..8fefb1505 100644 --- a/tests/integration/dense-read-bounded.test.ts +++ b/tests/integration/dense-read-bounded.test.ts @@ -29,6 +29,8 @@ import { buildGradedSeriesWithRollups } from "@/lib/insights/graded-series"; import { readDayAggregates } from "@/lib/measurements/day-aggregates"; import { foldDayAggregates } from "@/lib/measurements/__tests__/fake-day-aggregates"; import { localHmAsUtc } from "@/lib/tz/local-day"; +import { dayValue } from "@/lib/measurements/day-mean"; +import { userDayKey } from "@/lib/tz/format"; vi.mock("@/lib/db-compat", () => ({ ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), @@ -202,7 +204,31 @@ describe("pulse status over a dense stream (#1023)", () => { rows.map((r) => ({ measuredAt: r.at, value: r.value })), opts, ); - expect(sql).toEqual(memory); + // Pulse rows also carry the hourly-mean day (`day-mean.ts`); the plain + // fold fields agree with the in-memory fold, and each row's day value is + // the mean of its local hours' means over the same readings. + expect( + sql.map( + ({ dayMean: _d, weightedSum: _w, weightSum: _s, ...plain }) => plain, + ), + ).toEqual(memory); + for (const row of sql) { + const members = rows.filter( + (r) => + r.value >= 20 && + r.value <= 300 && + userDayKey(r.at, tz) === row.day && + opts.segmentStarts.filter((s) => r.at < s).length === row.segment, + ); + expect(row.dayMean).toBeCloseTo( + dayValue( + "PULSE", + members.map((r) => ({ value: r.value, measuredAt: r.at })), + tz, + )!, + 9, + ); + } // Sanity: the impossible value was dropped and the edge row split off. expect(sql.reduce((n, r) => n + r.n, 0)).toBe(5); }); diff --git a/tests/integration/doctor-report-pulse-hourly-mean.test.ts b/tests/integration/doctor-report-pulse-hourly-mean.test.ts new file mode 100644 index 000000000..050a10dd3 --- /dev/null +++ b/tests/integration/doctor-report-pulse-hourly-mean.test.ts @@ -0,0 +1,116 @@ +/** + * The doctor report's pulse figures use the hourly-mean day: a day is the mean + * of its local hours' means and the period average is the mean of its days, + * on both the raw path (a short window) and the day-bucket path (a long one). + * + * The account sits in Asia/Kolkata (UTC+05:30), so a local hour starts at half + * past a UTC hour. The workout runs 04:45Z to 05:14:30Z: one local hour + * (10:15 to 10:44:30), two UTC hours. + * + * day one 60 readings at 150 in local 10:xx, one at 60 in local 12, 14, 16 + * hours' means 150, 60, 60, 60, day value 82.5 + * (plain mean 145.7; UTC hours would give 96) + * day two one reading at 60, day value 60 + * period (82.5 + 60) / 2 = 71.25 (plain mean over readings 144.375) + * + * Count, min and max stay over every reading. + */ +import { beforeAll, describe, expect, it } from "vitest"; + +import type { Prisma } from "@/generated/prisma/client"; +import { collectDoctorReportData } from "@/lib/doctor-report-data"; +import { selectionFromLeaves } from "@/lib/report-selection/selection"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +const MODULES = { + cycle: false, + mood: false, + sleep: false, + glucose: false, + workouts: false, + recovery: false, + labs: false, + illness: false, + achievements: true, + coach: true, + insights: true, + medications: true, + doctorReport: true, + environment: true, + mcp: true, + inboundDocuments: true, + mentalHealth: true, + nutrients: true, + vaccinations: true, +} as const; + +const SELECTION = selectionFromLeaves(["PULSE"]); +const END = new Date("2026-05-01T00:00:00.000Z"); +const LONG = { + start: new Date("2026-01-01T00:00:00.000Z"), + end: END, + days: 120, +}; +const SHORT = { + start: new Date("2026-02-15T00:00:00.000Z"), + end: END, + days: 75, +}; + +let userId = ""; + +beforeAll(async () => { + const prisma = getPrismaClient(); + await truncateAllTables(prisma); + const user = await prisma.user.create({ + data: { + username: "doctor-report-pulse-hourly", + email: "doctor-report-pulse-hourly@example.test", + timezone: "Asia/Kolkata", + }, + }); + userId = user.id; + const rows: Prisma.MeasurementCreateManyInput[] = []; + const add = (iso: string, value: number) => + rows.push({ + userId, + type: "PULSE", + value, + unit: "bpm", + source: "APPLE_HEALTH", + measuredAt: new Date(iso), + }); + const workoutStart = Date.parse("2026-03-10T04:45:00.000Z"); + for (let i = 0; i < 60; i += 1) { + add(new Date(workoutStart + i * 30_000).toISOString(), 150); + } + add("2026-03-10T06:30:00.000Z", 60); + add("2026-03-10T08:30:00.000Z", 60); + add("2026-03-10T10:30:00.000Z", 60); + add("2026-03-11T03:30:00.000Z", 60); + await prisma.measurement.createMany({ data: rows }); +}, 120_000); + +describe.each([ + ["day-bucket path (long window)", LONG], + ["raw path (short window)", SHORT], +] as const)("doctor report pulse, %s", (_label, range) => { + it("averages the days, each the mean of its local hours", async () => { + const data = await collectDoctorReportData(userId, range, SELECTION, { + moduleMap: { ...MODULES }, + }); + expect(data.stats.PULSE).toMatchObject({ count: 64, min: 60, max: 150 }); + expect(data.stats.PULSE.avg).toBeCloseTo(71.25, 9); + }); +}); + +it("draws the long window's day points as hourly-mean days", async () => { + const data = await collectDoctorReportData(userId, LONG, SELECTION, { + moduleMap: { ...MODULES }, + }); + expect(data.measurements.PULSE.map((p) => p.value)).toEqual([ + expect.closeTo(82.5, 9), + expect.closeTo(60, 9), + ]); +});