From 0db52a1bff0b15652dfe049110ca046c200505f3 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Mon, 19 Oct 2026 17:57:01 +0200 Subject: [PATCH 1/5] fix(forecast): train load models on closed history intervals Core computed household load three ways: history balanced per-source energy per interval, the Energyplan tracker kept its own quarter-hour accumulator, and the built-in load model learned from one instant reading per minute. The models learned a different quantity from the Home use shown in History. Add state.HouseholdIntervals, which returns closed, valid intervals from the same source energy as daily totals. The built-in model trains once per closed quarter on its mean, and the tracker takes its load labels from the same quarters. PV keeps its accumulator for the curtailment flag and waits until history decides the quarter, so each immutable observation is written once. Co-Authored-By: Claude Opus 5.5 --- .../load-models-use-history-intervals.md | 5 + go/cmd/ftw/forecast_observation_time_test.go | 100 +++++++++--- go/cmd/ftw/forecast_tracking.go | 61 ++++++-- go/cmd/ftw/forecast_tracking_test.go | 28 ++-- go/internal/loadmodel/heating_bound_test.go | 19 +-- go/internal/loadmodel/intervals_test.go | 146 ++++++++++++++++++ go/internal/loadmodel/model.go | 4 +- go/internal/loadmodel/reconfigure_test.go | 22 --- .../loadmodel/restart_learning_test.go | 82 ---------- go/internal/loadmodel/service.go | 79 +++++++--- go/internal/loadmodel/service_test.go | 60 ------- go/internal/state/household_intervals.go | 48 ++++++ go/internal/state/household_intervals_test.go | 124 +++++++++++++++ 13 files changed, 533 insertions(+), 245 deletions(-) create mode 100644 .changeset/load-models-use-history-intervals.md create mode 100644 go/internal/loadmodel/intervals_test.go create mode 100644 go/internal/state/household_intervals.go create mode 100644 go/internal/state/household_intervals_test.go diff --git a/.changeset/load-models-use-history-intervals.md b/.changeset/load-models-use-history-intervals.md new file mode 100644 index 000000000..1d0ff5542 --- /dev/null +++ b/.changeset/load-models-use-history-intervals.md @@ -0,0 +1,5 @@ +--- +"ftw": patch +--- + +Train both load models on the closed quarter-hour household energy that daily Home use is built from. The built-in model no longer learns from one instant reading per minute, and the Energyplan forecast no longer keeps its own load collector, so forecasts learn the Home use shown in History. diff --git a/go/cmd/ftw/forecast_observation_time_test.go b/go/cmd/ftw/forecast_observation_time_test.go index 36809f76d..09f9633f5 100644 --- a/go/cmd/ftw/forecast_observation_time_test.go +++ b/go/cmd/ftw/forecast_observation_time_test.go @@ -12,17 +12,48 @@ import ( "github.com/srcfl/ftw/go/internal/telemetry" ) -func TestForecastObservationUsesCurrentTelemetryCutoff(t *testing.T) { - ctx := context.Background() +// householdHistoryStore records source ticks as Core does, so load +// observations come from the same closed intervals as daily Home use. +func householdHistoryStore(t *testing.T) *state.Store { + t.Helper() st, err := state.Open(filepath.Join(t.TempDir(), "state.db")) if err != nil { t.Fatal(err) } - defer st.Close() - if err = st.InitForecastArchive(ctx); err != nil { + t.Cleanup(func() { st.Close() }) + if err := st.EnableHistoryAggregation(); err != nil { + t.Fatal(err) + } + if err := st.InitForecastArchive(context.Background()); err != nil { t.Fatal(err) } + return st +} +// recordSiteHistory writes 5 s source ticks over [from, to]. The house uses +// gridW - pvW. +func recordSiteHistory(t *testing.T, st *state.Store, from, to time.Time, gridW, pvW float64) { + t.Helper() + for at := from; !at.After(to); at = at.Add(5 * time.Second) { + v := state.HistorySources{AtMS: at.UnixMilli(), Topology: "site", GridW: gridW, PVW: pvW} + if err := st.RecordTickWithOptionalHistory(&state.HistoryPoint{TsMs: at.UnixMilli(), JSON: `{}`, Sources: &v}, nil, nil); err != nil { + t.Fatal(err) + } + } +} + +// pvReading carries a deliberately wrong instantaneous house value: load +// labels must come from history, never from this snapshot. +func pvReading(at time.Time) telemetry.ForecastReading { + return telemetry.ForecastReading{ + At: at, Earliest: at, Latest: at, PVEarliest: at, PVLatest: at, + Valid: true, PVValid: true, HouseholdW: 9999, PVW: -800, + } +} + +func TestForecastObservationUsesCurrentTelemetryCutoff(t *testing.T) { + ctx := context.Background() + st := householdHistoryStore(t) tel := telemetry.NewStore() site := trackerSite() f := trackerFixture(time.Now()) @@ -34,19 +65,13 @@ func TestForecastObservationUsesCurrentTelemetryCutoff(t *testing.T) { observedAt = at return false } - reading := func(at time.Time) telemetry.ForecastReading { - return telemetry.ForecastReading{ - At: at, Earliest: at, Latest: at, PVEarliest: at, PVLatest: at, - Valid: true, PVValid: true, HouseholdW: 1200, PVW: -800, - } - } feed := func(at time.Time) []forecasting.Observation { - return f.observationIntervals(reading(at), site, at) + return f.observationIntervals(ctx, pvReading(at), site, at) } - // Prime the real accumulator from the start of the current quarter through - // the scheduled tick. A synchronous archive retry can delay observe until a - // newer telemetry sample has arrived. + // Prime the real PV accumulator from the start of the current quarter + // through the scheduled tick. A synchronous archive retry can delay + // observe until a newer telemetry sample has arrived. scheduledAt := time.Now().UTC() quarterStart := scheduledAt.Truncate(15 * time.Minute) if out := feed(quarterStart); len(out) != 0 { @@ -74,6 +99,7 @@ func TestForecastObservationUsesCurrentTelemetryCutoff(t *testing.T) { var completed []forecasting.Observation quarterEnd := quarterStart.Add(15 * time.Minute) + recordSiteHistory(t, st, quarterStart, quarterEnd, 400, -800) if observedAt.Before(quarterEnd) { for at := observedAt.Add(time.Minute); at.Before(quarterEnd); at = at.Add(time.Minute) { completed = append(completed, feed(at)...) @@ -85,31 +111,61 @@ func TestForecastObservationUsesCurrentTelemetryCutoff(t *testing.T) { t.Fatal(err) } completed = append(completed, persisted...) - if len(completed) != 1 || !completed[0].LoadKnown || completed[0].Quality != telemetry.ForecastIntervalQuality || + if len(completed) != 1 || !completed[0].LoadKnown || completed[0].Quality != state.HouseholdIntervalQuality || !completed[0].PVKnown || math.Abs(completed[0].LoadW-1200) > 1e-6 || math.Abs(completed[0].PVW-800) > 1e-6 || completed[0].StartMS != quarterStart.UnixMilli() || completed[0].EndMS != quarterEnd.UnixMilli() { t.Fatalf("delayed observation lost or changed the complete quarter: %+v", completed) } } -func TestForecastObservationStillRejectsLongGap(t *testing.T) { +func TestForecastPVStillRejectsLongGap(t *testing.T) { f := &forecastTracker{} site := trackerSite() start := time.Date(2026, 1, 5, 12, 0, 0, 0, time.UTC) - reading := func(at time.Time) telemetry.ForecastReading { - return telemetry.ForecastReading{At: at, Earliest: at, Latest: at, Valid: true, HouseholdW: 1200} - } - if out := f.observationIntervals(reading(start), site, start); len(out) != 0 { + ctx := context.Background() + if out := f.observationIntervals(ctx, pvReading(start), site, start); len(out) != 0 { t.Fatalf("first sample completed an interval: %+v", out) } - if out := f.observationIntervals(reading(start.Add(3*time.Minute)), site, start.Add(3*time.Minute)); len(out) != 0 { + if out := f.observationIntervals(ctx, pvReading(start.Add(3*time.Minute)), site, start.Add(3*time.Minute)); len(out) != 0 { t.Fatalf("long gap produced an interval: %+v", out) } var completed []forecasting.Observation for at := start.Add(4 * time.Minute); !at.After(start.Add(15 * time.Minute)); at = at.Add(time.Minute) { - completed = append(completed, f.observationIntervals(reading(at), site, at)...) + completed = append(completed, f.observationIntervals(ctx, pvReading(at), site, at)...) } if len(completed) != 0 { t.Fatalf("quarter with a long gap was accepted: %+v", completed) } } + +// PV waits for history to decide its quarter, so each quarter is archived +// once. If history stays silent, PV goes alone and a late load is dropped. +func TestForecastPVWaitsForTheHouseholdDecision(t *testing.T) { + ctx := context.Background() + st := householdHistoryStore(t) + f := &forecastTracker{store: st} + site := trackerSite() + start := time.Now().UTC().Truncate(15*time.Minute).Add(-2 * time.Hour) + end := start.Add(15 * time.Minute) + feed := func(at time.Time) []forecasting.Observation { return f.observationIntervals(ctx, pvReading(at), site, at) } + for at := start; at.Before(end); at = at.Add(time.Minute) { + if out := feed(at); len(out) != 0 { + t.Fatalf("partial quarter emitted: %+v", out) + } + } + recordSiteHistory(t, st, start, end.Add(-time.Minute), 400, -800) + if out := feed(end); len(out) != 0 { + t.Fatalf("PV emitted before history decided its quarter: %+v", out) + } + if out := feed(end.Add(time.Minute)); len(out) != 0 { + t.Fatalf("PV emitted inside the wait: %+v", out) + } + out := feed(end.Add(maxHouseholdDecisionWait)) + if len(out) != 1 || out[0].LoadKnown || !out[0].PVKnown || out[0].StartMS != start.UnixMilli() { + t.Fatalf("PV lost after silent history: %+v", out) + } + recordSiteHistory(t, st, end.Add(-time.Minute+5*time.Second), end.Add(time.Minute), 400, -800) + if out := feed(end.Add(maxHouseholdDecisionWait + time.Minute)); len(out) != 0 { + t.Fatalf("a decided quarter was emitted twice: %+v", out) + } +} diff --git a/go/cmd/ftw/forecast_tracking.go b/go/cmd/ftw/forecast_tracking.go index f41c6be52..2e88aaf54 100644 --- a/go/cmd/ftw/forecast_tracking.go +++ b/go/cmd/ftw/forecast_tracking.go @@ -71,9 +71,10 @@ type forecastTracker struct { mu sync.RWMutex errors []forecasting.ErrorSample observations []forecasting.Observation - accumulator telemetry.ForecastAccumulator lastRevision string pvAccumulator telemetry.ForecastAccumulator + pendingPV map[int64]forecasting.Observation // PV quarters waiting for their household decision + loadNextMS int64 // first quarter whose household load is undecided stopped bool clock func() time.Time learningMu sync.RWMutex @@ -249,8 +250,8 @@ func (f *forecastTracker) observe(ctx context.Context) { if f.configMu != nil { f.configMu.RUnlock() } - f.accumulator = telemetry.ForecastAccumulator{} f.pvAccumulator = telemetry.ForecastAccumulator{} + f.pendingPV, f.loadNextMS = nil, 0 f.lastRevision = "" return } @@ -271,7 +272,7 @@ func (f *forecastTracker) observe(ctx context.Context) { f.observationPVValid, f.observationLoadValid = r.PVValid, r.Valid f.mu.Unlock() var jobs []forecastObservationJob - for _, o := range f.observationIntervals(r, site, now) { + for _, o := range f.observationIntervals(ctx, r, site, now) { job := forecastObservationJob{observation: o, site: site, away: f.away != nil && f.away(time.UnixMilli(o.StartMS))} rows, _ := f.store.LoadForecasts(o.StartMS-time.Hour.Milliseconds(), o.EndMS) if weather := forecastRow(usableTrackingWeather(rows, o.AvailableAtMS), o.StartMS, o.AvailableAtMS); weather != nil && weather.FetchedAtMs >= site.WeatherSinceMS { @@ -286,34 +287,64 @@ func (f *forecastTracker) observe(ctx context.Context) { f.acceptObservations(flushCtx, jobs, f.flushObservations) } +// maxHouseholdDecisionWait bounds how long a PV quarter waits for history to +// close the same quarter. Past it, PV goes alone and that load is given up. +const maxHouseholdDecisionWait = 2 * time.Minute + // observationIntervals keeps the two measurement claims independent. Unknown // battery power invalidates household load, but cannot erase valid PV evidence. -func (f *forecastTracker) observationIntervals(r telemetry.ForecastReading, site forecastSite, now time.Time) []forecasting.Observation { - if site.Revision != f.lastRevision { - f.accumulator = telemetry.ForecastAccumulator{} +// Household load comes from closed history intervals, the evidence behind +// daily Home use. Observations are immutable, so PV waits until history has +// decided its quarter and each quarter is emitted once. +func (f *forecastTracker) observationIntervals(ctx context.Context, r telemetry.ForecastReading, site forecastSite, now time.Time) []forecasting.Observation { + const quarter = int64(15 * time.Minute / time.Millisecond) + nowMS := now.UnixMilli() + if site.Revision != f.lastRevision || f.loadNextMS == 0 { + // Start at the next whole quarter: one observed under an earlier + // revision, or before this process, never becomes a label. f.pvAccumulator = telemetry.ForecastAccumulator{} + f.pendingPV = map[int64]forecasting.Observation{} + f.loadNextMS = (nowMS + quarter - 1) / quarter * quarter f.lastRevision = site.Revision } - byStart := map[int64]forecasting.Observation{} - for _, v := range f.accumulator.Observe(r) { - byStart[v.Start.UnixMilli()] = forecasting.Observation{StartMS: v.Start.UnixMilli(), EndMS: v.End.UnixMilli(), AvailableAtMS: now.UnixMilli(), LoadW: v.HouseholdW, LoadKnown: true, Quality: v.Quality, ConfigVersion: site.Revision} - } pv := r pv.Valid = r.PVValid pv.HouseholdW = 0 pv.Latest = r.PVLatest for _, v := range f.pvAccumulator.Observe(pv) { + if v.End.UnixMilli() <= f.loadNextMS { + continue // its quarter is decided and may already be archived + } key := v.Start.UnixMilli() - o, ok := byStart[key] - if !ok { - o = forecasting.Observation{StartMS: key, EndMS: v.End.UnixMilli(), AvailableAtMS: now.UnixMilli(), Quality: "complete_pv_v1", ConfigVersion: site.Revision} + f.pendingPV[key] = forecasting.Observation{StartMS: key, EndMS: v.End.UnixMilli(), Quality: "complete_pv_v1", ConfigVersion: site.Revision, PVW: math.Max(0, -v.PVW), PVKnown: v.PVValid} + } + byStart := map[int64]forecasting.Observation{} + if f.store != nil { + loads, closed, err := f.store.HouseholdIntervals(ctx, f.loadNextMS, nowMS, quarter) + if err != nil { + slog.Warn("forecast: household intervals unavailable", "err", err) + } + for _, v := range loads { + byStart[v.StartMS] = forecasting.Observation{StartMS: v.StartMS, EndMS: v.EndMS, LoadW: v.LoadW, LoadKnown: true, Quality: state.HouseholdIntervalQuality, ConfigVersion: site.Revision} + } + f.loadNextMS = max(f.loadNextMS, closed) + } + for key, o := range f.pendingPV { + if o.EndMS > f.loadNextMS && f.store != nil && nowMS-o.EndMS < maxHouseholdDecisionWait.Milliseconds() { + continue + } + // A released quarter is decided: history cannot add its load later. + f.loadNextMS = max(f.loadNextMS, o.EndMS) + if load, ok := byStart[key]; ok { + load.PVW, load.PVKnown = o.PVW, o.PVKnown + o = load } - o.PVW = math.Max(0, -v.PVW) - o.PVKnown = v.PVValid byStart[key] = o + delete(f.pendingPV, key) } out := make([]forecasting.Observation, 0, len(byStart)) for _, o := range byStart { + o.AvailableAtMS = nowMS out = append(out, o) } sort.Slice(out, func(i, j int) bool { return out[i].StartMS < out[j].StartMS }) diff --git a/go/cmd/ftw/forecast_tracking_test.go b/go/cmd/ftw/forecast_tracking_test.go index 09d0e9ff5..c57c6a460 100644 --- a/go/cmd/ftw/forecast_tracking_test.go +++ b/go/cmd/ftw/forecast_tracking_test.go @@ -203,24 +203,27 @@ func TestForecastOriginFollowsCandidateCapture(t *testing.T) { } } func TestForecastIndependentPVIntervalsAndCurtailment(t *testing.T) { + ctx := context.Background() f := &forecastTracker{} site := trackerSite() start := time.Date(2026, 1, 5, 12, 0, 0, 0, time.UTC) var out []forecasting.Observation for i := 0; i <= 15; i++ { at := start.Add(time.Duration(i) * time.Minute) - out = append(out, f.observationIntervals(telemetry.ForecastReading{At: at, Latest: at, PVLatest: at, Valid: false, PVValid: true, PVW: -2000}, site, at)...) + out = append(out, f.observationIntervals(ctx, telemetry.ForecastReading{At: at, Latest: at, PVLatest: at, Valid: false, PVValid: true, PVW: -2000}, site, at)...) } if len(out) != 1 || !out[0].PVKnown || out[0].LoadKnown || out[0].PVW != 2000 { t.Fatalf("missing battery blocked PV: %+v", out) } - f = &forecastTracker{} + st := householdHistoryStore(t) + recordSiteHistory(t, st, start, start.Add(15*time.Minute), -1000, -2000) + f = &forecastTracker{store: st} out = nil for i := 0; i <= 15; i++ { at := start.Add(time.Duration(i) * time.Minute) - out = append(out, f.observationIntervals(telemetry.ForecastReading{At: at, Latest: at, PVLatest: at, Valid: true, PVValid: i != 7, PVW: -2000, HouseholdW: 1000}, site, at)...) + out = append(out, f.observationIntervals(ctx, telemetry.ForecastReading{At: at, Latest: at, PVLatest: at, Valid: true, PVValid: i != 7, PVW: -2000}, site, at)...) } - if len(out) != 1 || out[0].PVKnown || !out[0].LoadKnown { + if len(out) != 1 || out[0].PVKnown || !out[0].LoadKnown || math.Abs(out[0].LoadW-1000) > 1e-6 { t.Fatalf("curtailment became available PV or hid house: %+v", out) } } @@ -373,17 +376,24 @@ func TestForecastFullHorizonRecordFitsArchive(t *testing.T) { t.Logf("stored full horizon JSON bytes=%d", len(raw)) } func TestForecastConfigChangeBreaksInterval(t *testing.T) { - f := &forecastTracker{} + // History itself is complete across the change; the revision alone must + // keep the quarter from becoming a label for the new configuration. + st := householdHistoryStore(t) + f := &forecastTracker{store: st} site := trackerSite() start := time.Date(2026, 1, 5, 12, 0, 0, 0, time.UTC) - for i := 0; i <= 15; i++ { + recordSiteHistory(t, st, start, start.Add(31*time.Minute), -1000, -2000) + for i := 0; i <= 31; i++ { if i == 8 { site.Revision = "changed" } at := start.Add(time.Duration(i) * time.Minute) - out := f.observationIntervals(telemetry.ForecastReading{At: at, Latest: at, PVLatest: at, Valid: true, PVValid: true, HouseholdW: 1000, PVW: -2000}, site, at) - if len(out) > 0 { - t.Fatal("interval crosses config revision") + out := f.observationIntervals(context.Background(), telemetry.ForecastReading{At: at, Latest: at, PVLatest: at, Valid: true, PVValid: true, PVW: -2000}, site, at) + switch { + case i < 30 && len(out) > 0: + t.Fatalf("interval crosses config revision: %+v", out) + case i == 30 && (len(out) != 1 || out[0].StartMS != start.Add(15*time.Minute).UnixMilli() || out[0].ConfigVersion != "changed" || !out[0].LoadKnown): + t.Fatalf("first whole quarter under the new revision: %+v", out) } } } diff --git a/go/internal/loadmodel/heating_bound_test.go b/go/internal/loadmodel/heating_bound_test.go index d4a0ab293..1e9a1b324 100644 --- a/go/internal/loadmodel/heating_bound_test.go +++ b/go/internal/loadmodel/heating_bound_test.go @@ -15,7 +15,7 @@ import ( // feed trains m once per service interval in [from, to). func feed(m *Model, from, to time.Time, sample func(time.Time) (loadW, tempC float64)) { - for at := from; at.Before(to); at = at.Add(defaultSampleInterval) { + for at := from; at.Before(to); at = at.Add(TrainingInterval) { loadW, tempC := sample(at) m.Update(at, loadW, tempC) } @@ -43,7 +43,7 @@ func TestHeatingBoundLowersAnOverstatedPrior(t *testing.T) { } var steps []string - for at := monday; !at.After(saturday); at = at.Add(defaultSampleInterval) { + for at := monday; !at.After(saturday); at = at.Add(TrainingInterval) { before := m.HeatingW_per_degC m.Update(at, houseW, outdoorC) if m.HeatingW_per_degC != before { @@ -77,7 +77,7 @@ func TestHeatingBoundKeepsRealHeating(t *testing.T) { m := NewModel(8000) m.HeatingW_per_degC = coef lowest := coef - for at := start; !at.After(end); at = at.Add(defaultSampleInterval) { + for at := start; !at.After(end); at = at.Add(TrainingInterval) { loadW, tempC := house(at) m.Update(at, loadW, tempC) lowest = math.Min(lowest, m.HeatingW_per_degC) @@ -117,7 +117,7 @@ func TestHeatingBoundIgnoresMildDays(t *testing.T) { m := NewModel(5520) m.HeatingW_per_degC = 300 start := time.Date(2026, 5, 4, 0, 0, 0, 0, time.UTC) - feed(m, start, start.AddDate(0, 0, 3).Add(defaultSampleInterval), func(at time.Time) (float64, float64) { + feed(m, start, start.AddDate(0, 0, 3).Add(TrainingInterval), func(at time.Time) (float64, float64) { return 400, 15 + float64(at.Hour())/2 // 15 °C to 26.5 °C }) if m.HeatingW_per_degC != 300 { @@ -140,7 +140,7 @@ func TestHeatingBoundNeedsSixColdHours(t *testing.T) { sample := func(at time.Time) (float64, float64) { sinceMidnight := at.Sub(time.Date(at.Year(), at.Month(), at.Day(), 0, 0, 0, 0, time.UTC)) switch { - case at.Weekday() == time.Monday && sinceMidnight < 6*time.Hour-defaultSampleInterval: + case at.Weekday() == time.Monday && sinceMidnight < 6*time.Hour-TrainingInterval: return 400, 10 case at.Weekday() != time.Monday && sinceMidnight < 6*time.Hour: return 800, 10 @@ -150,12 +150,12 @@ func TestHeatingBoundNeedsSixColdHours(t *testing.T) { // The first sample of a day closes the day before. Had Monday counted, // Thursday's first sample would close a third cold day. - feed(m, monday, thursday.Add(defaultSampleInterval), sample) + feed(m, monday, thursday.Add(TrainingInterval), sample) if m.HeatingW_per_degC != 300 || m.HeatingBoundRecentN != 2 { t.Fatalf("%d cold samples counted as a day: coefficient %.2f, %d days", heatingBoundMinSamples-1, m.HeatingW_per_degC, m.HeatingBoundRecentN) } - feed(m, thursday.Add(defaultSampleInterval), friday.Add(defaultSampleInterval), sample) + feed(m, thursday.Add(TrainingInterval), friday.Add(TrainingInterval), sample) // Each bound is 800 W / 8 °C = 100 W/°C. if m.HeatingW_per_degC != 200 { t.Fatalf("three days of %d cold samples left the coefficient at %.2f, want 200", heatingBoundMinSamples, m.HeatingW_per_degC) @@ -177,8 +177,9 @@ func TestHeatingBoundSumsSurviveRestart(t *testing.T) { got := NewService(st, telemetry.NewStore(), "site", 4000, 0).Model() - if got.HeatingBoundDay != "2026-01-07" || got.HeatingBoundSamples != 420 || - got.HeatingBoundLoadSumW != 420*400 || got.HeatingBoundDeltaSumC != 420*8 || + n := int64(7 * time.Hour / TrainingInterval) + if got.HeatingBoundDay != "2026-01-07" || got.HeatingBoundSamples != n || + got.HeatingBoundLoadSumW != float64(n)*400 || got.HeatingBoundDeltaSumC != float64(n)*8 || got.HeatingBoundRecentN != 2 || got.HeatingBoundRecent != [heatingBoundDays]float64{0, 50, 50} { t.Fatalf("day sums after restart: day %q, %d samples, %.0f W, %.0f °C, bounds %v (%d)", got.HeatingBoundDay, got.HeatingBoundSamples, got.HeatingBoundLoadSumW, got.HeatingBoundDeltaSumC, diff --git a/go/internal/loadmodel/intervals_test.go b/go/internal/loadmodel/intervals_test.go new file mode 100644 index 000000000..726a94678 --- /dev/null +++ b/go/internal/loadmodel/intervals_test.go @@ -0,0 +1,146 @@ +package loadmodel + +import ( + "testing" + "time" + + "github.com/srcfl/ftw/go/internal/state" + "github.com/srcfl/ftw/go/internal/telemetry" +) + +// householdStore records source ticks exactly as Core does, so the model +// trains on the same closed intervals as daily Home use. +func householdStore(t *testing.T) *state.Store { + t.Helper() + st := openTestDB(t) + if err := st.EnableHistoryAggregation(); err != nil { + t.Fatal(err) + } + return st +} + +func recordHousehold(t *testing.T, st *state.Store, from, to time.Time, sources func(time.Time) state.HistorySources) { + t.Helper() + for at := from; !at.After(to); at = at.Add(5 * time.Second) { + v := sources(at) + v.AtMS, v.Topology = at.UnixMilli(), "site" + if err := st.RecordTickWithOptionalHistory(&state.HistoryPoint{TsMs: at.UnixMilli(), JSON: `{}`, Sources: &v}, nil, nil); err != nil { + t.Fatal(err) + } + } +} + +// trainingBase is a past, quarter-aligned hour; a first sample just before it +// places the cursor on it. +func trainingBase() time.Time { + return time.Now().UTC().Truncate(time.Hour).Add(-3 * time.Hour) +} + +func TestTrainsOnceOnEachClosedValidHouseholdInterval(t *testing.T) { + st := householdStore(t) + s := NewService(st, telemetry.NewStore(), "site", 4000, 0) + base := trainingBase() + s.sampleAt(base.Add(-time.Millisecond)) + q := TrainingInterval + recordHousehold(t, st, base, base.Add(3*q), func(at time.Time) state.HistorySources { + switch { + case at.Before(base.Add(q)): + return state.HistorySources{GridW: 1000} + case at.Before(base.Add(2 * q)): + // One battery reading is unknown: this quarter is no label. + v := state.HistorySources{GridW: 1000} + if at.Sub(base) == q+5*time.Minute { + v.Missing = state.MissingBattery + } + return v + } + // The battery supplies 200 W, so the house uses 800 W. + return state.HistorySources{GridW: 600, BatteryW: -200} + }) + s.sampleAt(base.Add(3*q + time.Minute)) + m := s.Model() + if m.Samples != 2 { + t.Fatalf("samples = %d, want the two valid quarters", m.Samples) + } + if got := m.Bucket[m.hourOfWeek(base)].Mean; got != 900 { + t.Fatalf("hour mean = %v, want the mean of 1000 W and 800 W", got) + } + s.sampleAt(base.Add(3*q + 2*time.Minute)) + if s.Model().Samples != 2 { + t.Fatal("a closed interval trained twice") + } + + // The switch happens during the fourth quarter; it trains neither + // profile, and the fifth trains the new one. + if err := s.SetProfile(ProfileAway); err != nil { + t.Fatal(err) + } + s.sampleAt(base.Add(3*q + 3*time.Minute)) + recordHousehold(t, st, base.Add(3*q+5*time.Second), base.Add(5*q), func(time.Time) state.HistorySources { + return state.HistorySources{GridW: 500} + }) + s.sampleAt(base.Add(5*q + time.Minute)) + snap := s.Snapshot() + if snap.Profiles[ProfileHome].Samples != 2 || snap.Profiles[ProfileAway].Samples != 1 { + t.Fatalf("profiles home=%d away=%d", snap.Profiles[ProfileHome].Samples, snap.Profiles[ProfileAway].Samples) + } +} + +func TestLearningCutoffSkipsIntervalsStartedBeforeIt(t *testing.T) { + st := householdStore(t) + s := NewService(st, telemetry.NewStore(), "site", 4000, 0) + base := trainingBase() + s.sampleAt(base.Add(-time.Millisecond)) + if err := s.RestartLearning(base.Add(TrainingInterval / 2)); err != nil { + t.Fatal(err) + } + // RestartLearning moves the cursor to the next interval after a sample. + s.mu.Lock() + s.nextIntervalMS = base.UnixMilli() + s.mu.Unlock() + recordHousehold(t, st, base, base.Add(2*TrainingInterval), func(time.Time) state.HistorySources { + return state.HistorySources{GridW: 1000} + }) + s.sampleAt(base.Add(2*TrainingInterval + time.Minute)) + if got := s.Model().Samples; got != 1 { + t.Fatalf("samples = %d, want only the interval after the cutoff", got) + } +} + +// An interval read before a reconfiguration or learning restart must not +// train the new model, and the reset cursor never returns to it. +func TestIntervalReadInFlightIsDiscarded(t *testing.T) { + for name, reset := range map[string]func(*Service) error{ + "reconfigure": func(s *Service) error { + return s.Reconfigure("site", telemetry.ForecastOptions{}, "UTC", "b") + }, + "restart learning": func(s *Service) error { return s.RestartLearning(time.Now()) }, + } { + t.Run(name, func(t *testing.T) { + st := householdStore(t) + s := NewService(st, telemetry.NewStore(), "site", 4000, 0) + if err := s.Reconfigure("site", telemetry.ForecastOptions{}, "UTC", "a"); err != nil { + t.Fatal(err) + } + base := trainingBase() + s.sampleAt(base.Add(-time.Millisecond)) + recordHousehold(t, st, base, base.Add(TrainingInterval), func(time.Time) state.HistorySources { + return state.HistorySources{GridW: 1000} + }) + captured, release, done := make(chan struct{}), make(chan struct{}), make(chan struct{}) + s.Temp = func(time.Time) (float64, bool) { close(captured); <-release; return 5, true } + go func() { defer close(done); s.sampleAt(base.Add(TrainingInterval + time.Minute)) }() + <-captured + if err := reset(s); err != nil { + t.Fatal(err) + } + close(release) + <-done + s.Temp = nil + s.sampleAt(time.Now()) + if got := s.Model().Samples; got != 0 { + t.Fatalf("interval from before the reset trained: %d samples", got) + } + }) + } +} diff --git a/go/internal/loadmodel/model.go b/go/internal/loadmodel/model.go index b43ee0007..c107df727 100644 --- a/go/internal/loadmodel/model.go +++ b/go/internal/loadmodel/model.go @@ -31,9 +31,9 @@ const heatingBoundStep = 0.5 // heating off must not pull a correct coefficient down. const heatingBoundDays = 3 -// heatingBoundMinSamples is six hours of cold samples at the service's rate. +// heatingBoundMinSamples is six hours of cold training intervals. // A day with less cold data does not bound the coefficient. -const heatingBoundMinSamples = int64(6 * time.Hour / defaultSampleInterval) +const heatingBoundMinSamples = int64(6 * time.Hour / TrainingInterval) const PlausibleLoadHeadroom = 1.25 diff --git a/go/internal/loadmodel/reconfigure_test.go b/go/internal/loadmodel/reconfigure_test.go index 9d8385e1f..fdb8eea1f 100644 --- a/go/internal/loadmodel/reconfigure_test.go +++ b/go/internal/loadmodel/reconfigure_test.go @@ -56,28 +56,6 @@ func TestReconfigureBindsEveryProfileAndSurvivesRestart(t *testing.T) { } } -func TestReconfigureRejectsTrainingCapturedAtPriorBoundary(t *testing.T) { - tel := telemetry.NewStore() - tel.Update("site", telemetry.DerMeter, 1000, nil, nil) - tel.RecordDriverSuccess("site") - s := NewService(nil, tel, "site", 4000, 0) - if err := s.Reconfigure("site", telemetry.ForecastOptions{}, "UTC", "a"); err != nil { - t.Fatal(err) - } - captured, release, done := make(chan struct{}), make(chan struct{}), make(chan struct{}) - s.Temp = func(time.Time) (float64, bool) { close(captured); <-release; return 5, true } - go func() { defer close(done); s.sampleAt(time.Now()) }() - <-captured - if err := s.Reconfigure("site", telemetry.ForecastOptions{HouseholdInvalidReason: "changed topology"}, "UTC", "b"); err != nil { - t.Fatal(err) - } - close(release) - <-done - if s.Model().Samples != 0 { - t.Fatal("old balance trained new topology") - } -} - func TestResetKeepsConfigBinding(t *testing.T) { s := NewService(nil, telemetry.NewStore(), "site", 4000, 0) if err := s.Reconfigure("site", telemetry.ForecastOptions{}, "UTC", "a"); err != nil { diff --git a/go/internal/loadmodel/restart_learning_test.go b/go/internal/loadmodel/restart_learning_test.go index 4055e554b..c027e6a62 100644 --- a/go/internal/loadmodel/restart_learning_test.go +++ b/go/internal/loadmodel/restart_learning_test.go @@ -1,7 +1,6 @@ package loadmodel import ( - "encoding/json" "testing" "time" @@ -51,87 +50,6 @@ func TestRestartLearningClearsEveryLoadProfileAndRestoresConfiguredPrior(t *test } } -func TestRestartLearningRejectsLoadSampleCapturedBeforeCutoff(t *testing.T) { - tel := telemetry.NewStore() - tel.Update("site", telemetry.DerMeter, 1000, nil, nil) - tel.RecordDriverSuccess("site") - s := NewService(nil, tel, "site", 4000, 0) - cutoff := time.Now().Add(time.Millisecond) - if err := s.RestartLearning(cutoff); err != nil { - t.Fatal(err) - } - - s.sampleAt(cutoff.Add(10 * time.Millisecond)) - - if got := s.Model().Samples; got != 0 { - t.Fatalf("pre-cutoff telemetry trained load model: %d samples", got) - } -} - -func TestRestartLearningRejectsMixedLoadBalanceWithOldInput(t *testing.T) { - tel := telemetry.NewStore() - now := time.Now() - cutoff := now.Add(-time.Second) - powerData := func(watts float64, measured time.Time) []byte { - data, err := json.Marshal(map[string]telemetry.ForecastPowerSample{ - "forecast_power": { - Version: 1, - Known: true, - Watts: watts, - MeasuredAtMS: measured.UnixMilli(), - ReceivedAtMS: now.UnixMilli(), - }, - }) - if err != nil { - t.Fatal(err) - } - return data - } - tel.Update("site", telemetry.DerMeter, 1000, nil, powerData(1000, cutoff.Add(-time.Second))) - tel.RecordDriverSuccess("site") - tel.Update("battery", telemetry.DerBattery, 500, nil, powerData(500, cutoff.Add(time.Second))) - tel.RecordDriverSuccess("battery") - s := NewService(nil, tel, "site", 4000, 0) - if err := s.RestartLearning(cutoff); err != nil { - t.Fatal(err) - } - - s.sampleAt(now.Add(time.Second)) - - if got := s.Model().Samples; got != 0 { - t.Fatalf("balance with pre-cutoff meter trained load model: %d samples", got) - } -} - -func TestRestartLearningInvalidatesLoadSampleInFlight(t *testing.T) { - tel := telemetry.NewStore() - tel.Update("site", telemetry.DerMeter, 1000, nil, nil) - tel.RecordDriverSuccess("site") - s := NewService(nil, tel, "site", 4000, 0) - entered := make(chan struct{}) - release := make(chan struct{}) - s.Temp = func(time.Time) (float64, bool) { - close(entered) - <-release - return 5, true - } - done := make(chan struct{}) - go func() { - defer close(done) - s.sampleAt(time.Now()) - }() - <-entered - if err := s.RestartLearning(time.Now()); err != nil { - t.Fatal(err) - } - close(release) - <-done - - if got := s.Model().Samples; got != 0 { - t.Fatalf("in-flight sample survived restart: %d", got) - } -} - func TestRestartLearningLoadRetryPersistsAllProfilesWithoutErasingNewSamples(t *testing.T) { bad := openTestDB(t) s := NewService(bad, telemetry.NewStore(), "site", 4000, 0) diff --git a/go/internal/loadmodel/service.go b/go/internal/loadmodel/service.go index d578461b2..1926f7a2c 100644 --- a/go/internal/loadmodel/service.go +++ b/go/internal/loadmodel/service.go @@ -29,9 +29,13 @@ const ( profileStateKey = "loadmodel/profile" ) -// defaultSampleInterval is how often the service trains the active model. +// defaultSampleInterval is how often the service looks for closed intervals. const defaultSampleInterval = time.Minute +// TrainingInterval is the width of one training label: the household mean +// over a closed quarter-hour of the same source energy as daily Home use. +const TrainingInterval = 15 * time.Minute + // legacyFeatureHash is the fingerprint of the feature space in force when the // envelope was introduced. State written before then — both the per-profile // keys and legacyStateKey — carries no fingerprint, but it was fitted against @@ -59,8 +63,10 @@ type Snapshot struct { Profiles map[Profile]Model `json:"profiles"` } -// Service trains the load model online from telemetry. Mirrors -// pvmodel.Service so operators + future code have one pattern. +// Service trains the load model online from closed household intervals in +// Store, the same evidence as daily Home use. Mirrors pvmodel.Service so +// operators + future code have one pattern. Tele and the forecast options are +// kept for callers; training no longer reads instantaneous telemetry. type Service struct { Store *state.Store Tele *telemetry.Store @@ -76,7 +82,7 @@ type Service struct { models map[Profile]*Model forecastOptions telemetry.ForecastOptions timezone string - lastForecastInput time.Time + nextIntervalMS int64 // first interval not yet trained; 0 waits for the next whole one configuredHeating *float64 stop chan struct{} @@ -90,7 +96,7 @@ func NewService(st *state.Store, tel *telemetry.Store, siteMeter string, peakW, Tele: tel, SiteMeter: siteMeter, SampleInterval: defaultSampleInterval, - PersistEvery: 10, + PersistEvery: 1, stop: make(chan struct{}), done: make(chan struct{}), active: ProfileHome, @@ -178,7 +184,7 @@ func (s *Service) SetSiteMeter(name string) { s.mu.Lock() s.SiteMeter = name s.generation++ - s.lastForecastInput = time.Time{} + s.nextIntervalMS = 0 s.mu.Unlock() } @@ -267,6 +273,10 @@ func (s *Service) SetProfile(profile Profile) error { peak := s.activeModelLocked().PeakW s.models[profile] = newProfileModel(peak, profile) } + if s.active != profile { + // An interval that spans the switch belongs to neither profile. + s.nextIntervalMS = 0 + } s.active = profile s.mu.Unlock() return s.persistProfile(profile) @@ -431,7 +441,7 @@ func (s *Service) Reconfigure(siteMeter string, opts telemetry.ForecastOptions, } s.SiteMeter, s.timezone, s.forecastOptions = siteMeter, zone, opts s.generation++ - s.lastForecastInput = time.Time{} + s.nextIntervalMS = 0 if changed { peak := s.activeModelLocked().PeakW epoch := s.learningStartedMSLocked() @@ -477,34 +487,55 @@ func (s *Service) SetTimezone(zone string) error { return nil } func (s *Service) sample() { s.sampleAt(time.Now()) } + +// sampleAt trains on every interval closed since the last call. Training +// starts at the first whole interval after start or reconfiguration, so an +// interval observed under an older configuration never becomes a label. func (s *Service) sampleAt(now time.Time) { - if s.Tele == nil { + if s.Store == nil { return } - s.mu.RLock() - site, opts, profile, generation, learningStartedMS := s.SiteMeter, s.forecastOptions, s.active, s.generation, s.learningStartedMSLocked() - s.mu.RUnlock() - reading := s.Tele.ForecastMeasurement(now, site, opts) - temp := math.NaN() - if s.Temp != nil { - if value, ok := s.Temp(now); ok { - temp = value + width := TrainingInterval.Milliseconds() + s.mu.Lock() + if s.nextIntervalMS == 0 { + s.nextIntervalMS = (now.UnixMilli() + width - 1) / width * width + } + profile, generation, learningStartedMS, from := s.active, s.generation, s.learningStartedMSLocked(), s.nextIntervalMS + s.mu.Unlock() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + intervals, closed, err := s.Store.HouseholdIntervals(ctx, from, now.UnixMilli(), width) + cancel() + if err != nil { + slog.Warn("loadmodel: household intervals unavailable", "err", err) + return + } + temps := make([]float64, len(intervals)) + for i, v := range intervals { + temps[i] = math.NaN() + if s.Temp != nil { + if value, ok := s.Temp(time.UnixMilli(v.StartMS + width/2)); ok { + temps[i] = value + } } } s.mu.Lock() - if s.active != profile || s.SiteMeter != site || s.generation != generation { + // A profile switch or reconfiguration during the read discards it; the + // cursor stays, so the next call reads the same closed intervals again. + if s.active != profile || s.generation != generation || s.nextIntervalMS != from { s.mu.Unlock() return } model := s.activeModelLocked() - updated := false - if reading.Valid && reading.Earliest.UnixMilli() >= learningStartedMS && reading.Latest.After(s.lastForecastInput) { - s.lastForecastInput = reading.Latest - updated = model.Update(now, reading.HouseholdW, temp) + trained := 0 + for i, v := range intervals { + if v.StartMS >= learningStartedMS && model.Update(time.UnixMilli(v.StartMS+width/2), v.LoadW, temps[i]) { + trained++ + } } + s.nextIntervalMS = max(from, closed) samples := model.Samples s.mu.Unlock() - if updated && s.PersistEvery > 0 && samples%s.PersistEvery == 0 { + if trained > 0 && s.PersistEvery > 0 && samples%s.PersistEvery < int64(trained) { if err := s.persist(); err != nil { slog.Warn("loadmodel persist", "err", err) } @@ -625,7 +656,7 @@ func (s *Service) Reset() { s.models[profile].LearningStartedMS = old.LearningStartedMS s.models[profile].MaxPlausibleW = old.MaxPlausibleW s.generation++ - s.lastForecastInput = time.Time{} + s.nextIntervalMS = 0 s.mu.Unlock() if err := s.persist(); err != nil { slog.Warn("loadmodel persist", "err", err) @@ -651,7 +682,7 @@ func (s *Service) RestartLearning(at time.Time) error { s.models[profile] = freshProfile(s.models[profile], profile, s.timezone, startedMS, s.configuredHeating) } s.generation++ - s.lastForecastInput = time.Time{} + s.nextIntervalMS = 0 } s.mu.Unlock() return s.persistLocked() diff --git a/go/internal/loadmodel/service_test.go b/go/internal/loadmodel/service_test.go index f02d91169..1bb4e1d1b 100644 --- a/go/internal/loadmodel/service_test.go +++ b/go/internal/loadmodel/service_test.go @@ -53,34 +53,6 @@ func TestSeedHeatingCoefAppliesOnColdStart(t *testing.T) { } } -func TestProfileSwitchTrainsOnlyActiveProfile(t *testing.T) { - tel := telemetry.NewStore() - tel.Update("site", telemetry.DerMeter, 1000, nil, nil) - tel.RecordDriverSuccess("site") - - s := NewService(nil, tel, "site", 4000, 17250) - now := time.Now() - s.sampleAt(now) - - if err := s.SetProfile(ProfileAway); err != nil { - t.Fatalf("set profile: %v", err) - } - tel.Update("site", telemetry.DerMeter, 200, nil, nil) - tel.RecordDriverSuccess("site") - s.sampleAt(time.Now()) - - snap := s.Snapshot() - if snap.ActiveProfile != ProfileAway { - t.Fatalf("active profile = %q, want %q", snap.ActiveProfile, ProfileAway) - } - if got := snap.Profiles[ProfileHome].Samples; got != 1 { - t.Fatalf("home samples = %d, want 1", got) - } - if got := snap.Profiles[ProfileAway].Samples; got != 1 { - t.Fatalf("away samples = %d, want 1", got) - } -} - func TestProfileAndModelsPersist(t *testing.T) { st, err := state.Open(filepath.Join(t.TempDir(), "state.db")) if err != nil { @@ -110,35 +82,3 @@ func TestProfileAndModelsPersist(t *testing.T) { t.Fatalf("restored away samples = %d, want 8", got) } } - -func TestSampleRequiresOnlineSiteMeter(t *testing.T) { - tel := telemetry.NewStore() - tel.Update("site", telemetry.DerMeter, 1000, nil, nil) - - s := NewService(nil, tel, "site", 4000, 17250) - s.sampleAt(time.Date(2026, 1, 5, 12, 0, 0, 0, time.UTC)) - - if got := s.Model().Samples; got != 0 { - t.Fatalf("samples = %d, want 0 when site meter has no online health", got) - } -} - -func TestSampleRejectsMissingDERs(t *testing.T) { - tel := telemetry.NewStore() - tel.Update("site", telemetry.DerMeter, 4000, nil, nil) - tel.RecordDriverSuccess("site") - tel.Update("battery", telemetry.DerBattery, 3000, nil, nil) - tel.DriverHealthMut("battery").SetOffline() - s := NewService(nil, tel, "site", 4000, 11000) - s.sampleAt(time.Now()) - if s.Model().Samples != 0 { - t.Fatal("missing battery trained 4kW as house load") - } - tel.RecordDriverSuccess("battery") - tel.SetDriverCommandFault("battery", true, "refused") - s.sampleAt(time.Now()) - m := s.Model() - if m.Samples != 1 || m.Bucket[m.hourOfWeek(time.Now())].Mean != 1000 { - t.Fatalf("fresh command fault dropped real battery flow: %+v", m) - } -} diff --git a/go/internal/state/household_intervals.go b/go/internal/state/household_intervals.go new file mode 100644 index 000000000..e988edccd --- /dev/null +++ b/go/internal/state/household_intervals.go @@ -0,0 +1,48 @@ +package state + +import ( + "context" + "fmt" +) + +// HouseholdIntervalQuality labels training evidence read from these intervals. +const HouseholdIntervalQuality = "source_interval_balance_v1" + +// HouseholdInterval is the mean household power over one closed interval, +// balanced from the same source energy as daily Home use. Load models train +// on these, so a forecast learns the Home use that History shows. +type HouseholdInterval struct { + StartMS, EndMS int64 + LoadW float64 +} + +// HouseholdIntervals returns closed, valid intervals in [from, to) and the end +// of the closed range, so a caller can move past intervals left out. from must +// be aligned to width. An interval is closed once committed source progress +// reaches its end; later writes cannot change it. Intervals with a missing +// source, a recording gap or a negative balance are left out. +func (s *Store) HouseholdIntervals(ctx context.Context, from, to, width int64) ([]HouseholdInterval, int64, error) { + if width <= 0 || from%width != 0 { + return nil, from, fmt.Errorf("household intervals: from %d is not aligned to %d ms", from, width) + } + through, ok, err := s.HistorySourceProgress(ctx) + if err != nil || !ok { + return nil, from, err + } + to = bucketStart(min(to, through), width) + if to <= from { + return nil, from, nil + } + buckets, err := s.sourceLoadBuckets(ctx, from, to, width) + if err != nil { + return nil, from, err + } + var out []HouseholdInterval + for _, b := range buckets { + if b.width != width || b.covered != width || b.missing&MissingPower != 0 || b.wh < 0 || !finiteHistoryValue(b.wh) { + continue + } + out = append(out, HouseholdInterval{StartMS: b.start, EndMS: b.start + width, LoadW: b.wh * 3600000 / float64(width)}) + } + return out, to, nil +} diff --git a/go/internal/state/household_intervals_test.go b/go/internal/state/household_intervals_test.go new file mode 100644 index 000000000..1487b2b05 --- /dev/null +++ b/go/internal/state/household_intervals_test.go @@ -0,0 +1,124 @@ +package state + +import ( + "context" + "testing" + "time" +) + +const quarterMS = int64(15 * time.Minute / time.Millisecond) + +func householdQuarters(t *testing.T, s *Store, from, to int64) map[int64]float64 { + t.Helper() + got, closed, err := s.HouseholdIntervals(context.Background(), from, to, quarterMS) + if err != nil { + t.Fatal(err) + } + if closed < from || closed > to || closed%quarterMS != 0 { + t.Fatalf("closed range ends at %d, outside [%d,%d]", closed, from, to) + } + out := map[int64]float64{} + for _, v := range got { + if v.EndMS-v.StartMS != quarterMS { + t.Fatalf("interval width %d", v.EndMS-v.StartMS) + } + out[v.StartMS] = v.LoadW + } + return out +} + +func TestHouseholdIntervalsMatchDailyTotal(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24*time.Hour).UnixMilli() - 6*quarterMS + end := base + 3*quarterMS + for at := base; at <= end; at += 5000 { + // House 1200 W during the first quarter, then 800 W; a battery and + // a charger take part so the balance uses every source. + grid := 1700.0 + if at >= base+quarterMS { + grid = 1300 + } + sourceTick(t, s, at, HistorySources{GridW: grid, BatteryW: 300, EVW: 200}) + } + got := householdQuarters(t, s, base, end) + if len(got) != 3 { + t.Fatalf("quarters=%v", got) + } + near(t, "first quarter W", got[base], 1200) + near(t, "second quarter W", got[base+quarterMS], 800) + d, err := s.DailyEnergy(base, end) + if err != nil || d.Missing&MissingLoad != 0 { + t.Fatal("daily total unavailable", err, d.Missing) + } + var wh float64 + for _, w := range got { + wh += w / 4 + } + near(t, "daily Wh equals the sum of training intervals", d.LoadWh, wh) +} + +func TestHouseholdIntervalsLeaveOutIncompleteEvidence(t *testing.T) { + for _, tc := range []struct { + name string + tick func(at, base int64) (HistorySources, bool) + }{ + {"missing EV", func(at, base int64) (HistorySources, bool) { + v := HistorySources{GridW: 1000} + if at >= base+quarterMS+60000 && at < base+quarterMS+75000 { + v.Missing = MissingEV + } + return v, true + }}, + {"three-minute recording gap", func(at, base int64) (HistorySources, bool) { + return HistorySources{GridW: 1000}, at <= base+quarterMS+60000 || at >= base+quarterMS+240000 + }}, + {"negative balance", func(at, base int64) (HistorySources, bool) { + v := HistorySources{GridW: 1000} + if at >= base+quarterMS && at < base+2*quarterMS { + v.GridW = -500 + } + return v, true + }}, + } { + t.Run(tc.name, func(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24*time.Hour).UnixMilli() - 6*quarterMS + end := base + 3*quarterMS + for at := base; at <= end; at += 5000 { + if v, ok := tc.tick(at, base); ok { + sourceTick(t, s, at, v) + } + } + got := householdQuarters(t, s, base, end) + if _, ok := got[base+quarterMS]; ok || len(got) != 2 { + t.Fatalf("incomplete quarter became a training label: %v", got) + } + near(t, "neighbour quarter W", got[base], 1000) + near(t, "neighbour quarter W", got[base+2*quarterMS], 1000) + }) + } +} + +func TestHouseholdIntervalsWaitForCommittedProgress(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24*time.Hour).UnixMilli() - 6*quarterMS + for at := base; at < base+quarterMS; at += 5000 { + sourceTick(t, s, at, HistorySources{GridW: 1000}) + } + if got := householdQuarters(t, s, base, base+4*quarterMS); len(got) != 0 { + t.Fatalf("open quarter returned before its end was committed: %v", got) + } + sourceTick(t, s, base+quarterMS, HistorySources{GridW: 1000}) + if got := householdQuarters(t, s, base, base+4*quarterMS); len(got) != 1 { + t.Fatalf("closed quarter missing: %v", got) + } + // A later gap closes the next quarter without a label; callers move past it. + sourceTick(t, s, base+3*quarterMS, HistorySources{GridW: 1000}) + got, closed, err := s.HouseholdIntervals(context.Background(), base+quarterMS, base+4*quarterMS, quarterMS) + if err != nil || len(got) != 0 || closed != base+3*quarterMS { + t.Fatalf("gap quarters: got=%v closed=%d err=%v", got, closed-base, err) + } + if _, _, err := s.HouseholdIntervals(context.Background(), base+1, base+quarterMS, quarterMS); err == nil { + t.Fatal("unaligned start accepted") + } +} From ff23e756e7c3765a01afc6d05712dc5c4473f944 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Mon, 19 Oct 2026 18:52:41 +0200 Subject: [PATCH 2/5] style: gofmt the tracker observation test Co-Authored-By: Claude Opus 5.5 --- go/cmd/ftw/forecast_observation_time_test.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/go/cmd/ftw/forecast_observation_time_test.go b/go/cmd/ftw/forecast_observation_time_test.go index 09f9633f5..297a02303 100644 --- a/go/cmd/ftw/forecast_observation_time_test.go +++ b/go/cmd/ftw/forecast_observation_time_test.go @@ -145,9 +145,11 @@ func TestForecastPVWaitsForTheHouseholdDecision(t *testing.T) { st := householdHistoryStore(t) f := &forecastTracker{store: st} site := trackerSite() - start := time.Now().UTC().Truncate(15*time.Minute).Add(-2 * time.Hour) + start := time.Now().UTC().Truncate(15 * time.Minute).Add(-2 * time.Hour) end := start.Add(15 * time.Minute) - feed := func(at time.Time) []forecasting.Observation { return f.observationIntervals(ctx, pvReading(at), site, at) } + feed := func(at time.Time) []forecasting.Observation { + return f.observationIntervals(ctx, pvReading(at), site, at) + } for at := start; at.Before(end); at = at.Add(time.Minute) { if out := feed(at); len(out) != 0 { t.Fatalf("partial quarter emitted: %+v", out) From ff5195a6d0fa654f0d9085794b130bfdd766455c Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Mon, 19 Oct 2026 19:07:12 +0200 Subject: [PATCH 3/5] test: use gaps longer than the history bridge window Short source gaps may become estimated history once #1753 lands, and then a training label. Keep these tests about gaps that stay unknown. Co-Authored-By: Claude Opus 5.5 --- go/internal/loadmodel/intervals_test.go | 5 +++-- go/internal/state/household_intervals_test.go | 3 ++- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/go/internal/loadmodel/intervals_test.go b/go/internal/loadmodel/intervals_test.go index 726a94678..38e1787ff 100644 --- a/go/internal/loadmodel/intervals_test.go +++ b/go/internal/loadmodel/intervals_test.go @@ -47,9 +47,10 @@ func TestTrainsOnceOnEachClosedValidHouseholdInterval(t *testing.T) { case at.Before(base.Add(q)): return state.HistorySources{GridW: 1000} case at.Before(base.Add(2 * q)): - // One battery reading is unknown: this quarter is no label. + // The battery is unknown for three minutes, longer than any gap + // history may bridge as an estimate: this quarter is no label. v := state.HistorySources{GridW: 1000} - if at.Sub(base) == q+5*time.Minute { + if since := at.Sub(base); since >= q+5*time.Minute && since < q+8*time.Minute { v.Missing = state.MissingBattery } return v diff --git a/go/internal/state/household_intervals_test.go b/go/internal/state/household_intervals_test.go index 1487b2b05..3a8c1f2c3 100644 --- a/go/internal/state/household_intervals_test.go +++ b/go/internal/state/household_intervals_test.go @@ -62,9 +62,10 @@ func TestHouseholdIntervalsLeaveOutIncompleteEvidence(t *testing.T) { name string tick func(at, base int64) (HistorySources, bool) }{ + // Longer than any gap history may bridge as an estimate. {"missing EV", func(at, base int64) (HistorySources, bool) { v := HistorySources{GridW: 1000} - if at >= base+quarterMS+60000 && at < base+quarterMS+75000 { + if at >= base+quarterMS+60000 && at < base+quarterMS+240000 { v.Missing = MissingEV } return v, true From 44297c4a80ee68854214979c531ce70557847dd9 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Mon, 19 Oct 2026 19:11:53 +0200 Subject: [PATCH 4/5] fix(loadmodel): preserve stored minute-based heating bounds --- .../load-models-use-history-intervals.md | 2 + .../loadmodel/heating_bound_migration_test.go | 103 ++++++++++++++++++ go/internal/loadmodel/heating_bound_test.go | 4 +- go/internal/loadmodel/model.go | 20 ++-- 4 files changed, 118 insertions(+), 11 deletions(-) create mode 100644 go/internal/loadmodel/heating_bound_migration_test.go diff --git a/.changeset/load-models-use-history-intervals.md b/.changeset/load-models-use-history-intervals.md index 1d0ff5542..d66af750f 100644 --- a/.changeset/load-models-use-history-intervals.md +++ b/.changeset/load-models-use-history-intervals.md @@ -3,3 +3,5 @@ --- Train both load models on the closed quarter-hour household energy that daily Home use is built from. The built-in model no longer learns from one instant reading per minute, and the Energyplan forecast no longer keeps its own load collector, so forecasts learn the Home use shown in History. + +Keep cold-day counters in their stored minute units when training moves to quarter-hours, preserving learned profiles and partial-day evidence. diff --git a/go/internal/loadmodel/heating_bound_migration_test.go b/go/internal/loadmodel/heating_bound_migration_test.go new file mode 100644 index 000000000..ccc9c648e --- /dev/null +++ b/go/internal/loadmodel/heating_bound_migration_test.go @@ -0,0 +1,103 @@ +package loadmodel + +import ( + "testing" + "time" + + "github.com/srcfl/ftw/go/internal/modelstate" + "github.com/srcfl/ftw/go/internal/telemetry" +) + +func TestRestoredMinuteHeatingBoundsKeepLearning(t *testing.T) { + for _, tc := range []struct { + name string + coldMinutes int64 + wantHeating float64 + wantDays int + }{ + {name: "thirty minutes cannot close a cold day", coldMinutes: 30, wantHeating: 500, wantDays: 2}, + {name: "old minutes and a new quarter make six hours", coldMinutes: 345, wantHeating: 300, wantDays: 3}, + } { + t.Run(tc.name, func(t *testing.T) { + st := openTestDB(t) + day := time.Date(2026, 4, 6, 0, 0, 0, 0, time.UTC) + stored := make(map[Profile]Model) + for _, profile := range Profiles() { + m := *trainedModel(t) + m.Timezone = "UTC" + m.ConfigRevision = "existing-revision" + m.LearningStartedMS = day.AddDate(0, 0, -30).UnixMilli() + m.Samples = 1200 + m.LastMs = day.Add(6 * time.Hour).UnixMilli() + m.PriorScale = newProfileModel(m.PeakW, profile).PriorScale + m.HeatingW_per_degC = 500 + // State written by the old minute sampler, including a + // partial cold day and two qualified earlier days. + m.HeatingBoundDay = day.Format("2006-01-02") + m.HeatingBoundSamples = tc.coldMinutes + m.HeatingBoundLoadSumW = float64(tc.coldMinutes) * 1000 + m.HeatingBoundDeltaSumC = float64(tc.coldMinutes) * 10 + m.HeatingBoundRecent = [heatingBoundDays]float64{0, 100, 100} + m.HeatingBoundRecentN = 2 + stored[profile] = m + js, err := modelstate.Wrap(FeatureHash(), m) + if err != nil { + t.Fatal(err) + } + if err := st.SaveConfig(stateKey(profile), js); err != nil { + t.Fatal(err) + } + } + if err := st.SaveConfig(profileStateKey, string(ProfileAway)); err != nil { + t.Fatal(err) + } + + s := NewService(st, telemetry.NewStore(), "site", 4000, 0) + snapshot := s.Snapshot() + if snapshot.ActiveProfile != ProfileAway { + t.Fatal("restore changed the active profile") + } + for _, profile := range Profiles() { + if snapshot.Profiles[profile] != stored[profile] { + t.Fatalf("restore changed learned state or partial-day evidence for %s", profile) + } + } + + quarterMidpoint := day.Add(6*time.Hour + TrainingInterval/2) + s.mu.Lock() + trained := true + for _, profile := range Profiles() { + m := s.models[profile] + if !m.Update(quarterMidpoint, 1000, 8) { + trained = false + } + } + s.mu.Unlock() + if !trained { + t.Fatal("new quarter did not train") + } + for profile, m := range s.Snapshot().Profiles { + if m.HeatingBoundSamples != tc.coldMinutes+15 || + m.HeatingBoundLoadSumW != float64(tc.coldMinutes+15)*1000 || + m.HeatingBoundDeltaSumC != float64(tc.coldMinutes+15)*10 { + t.Fatalf("%s mixed minute and quarter units: %d minutes, %g load sum, %g temperature sum", + profile, m.HeatingBoundSamples, m.HeatingBoundLoadSumW, m.HeatingBoundDeltaSumC) + } + } + if err := s.persist(); err != nil { + t.Fatal(err) + } + restarted := NewService(st, telemetry.NewStore(), "site", 4000, 0) + for profile, m := range restarted.Snapshot().Profiles { + if m != s.Snapshot().Profiles[profile] { + t.Fatalf("restart changed migrated state for %s", profile) + } + m.Update(day.AddDate(0, 0, 1), 1000, HeatingReferenceC) + if m.HeatingW_per_degC != tc.wantHeating || m.HeatingBoundRecentN != tc.wantDays { + t.Fatalf("%s closed %d old minutes plus a quarter as %d cold days at %g W/C; want %d days at %g W/C", + profile, tc.coldMinutes, m.HeatingBoundRecentN, m.HeatingW_per_degC, tc.wantDays, tc.wantHeating) + } + } + }) + } +} diff --git a/go/internal/loadmodel/heating_bound_test.go b/go/internal/loadmodel/heating_bound_test.go index 1e9a1b324..8b5145482 100644 --- a/go/internal/loadmodel/heating_bound_test.go +++ b/go/internal/loadmodel/heating_bound_test.go @@ -153,7 +153,7 @@ func TestHeatingBoundNeedsSixColdHours(t *testing.T) { feed(m, monday, thursday.Add(TrainingInterval), sample) if m.HeatingW_per_degC != 300 || m.HeatingBoundRecentN != 2 { t.Fatalf("%d cold samples counted as a day: coefficient %.2f, %d days", - heatingBoundMinSamples-1, m.HeatingW_per_degC, m.HeatingBoundRecentN) + heatingBoundMinSamples-heatingBoundSampleWeight, m.HeatingW_per_degC, m.HeatingBoundRecentN) } feed(m, thursday.Add(TrainingInterval), friday.Add(TrainingInterval), sample) // Each bound is 800 W / 8 °C = 100 W/°C. @@ -177,7 +177,7 @@ func TestHeatingBoundSumsSurviveRestart(t *testing.T) { got := NewService(st, telemetry.NewStore(), "site", 4000, 0).Model() - n := int64(7 * time.Hour / TrainingInterval) + n := int64(7 * time.Hour / time.Minute) if got.HeatingBoundDay != "2026-01-07" || got.HeatingBoundSamples != n || got.HeatingBoundLoadSumW != float64(n)*400 || got.HeatingBoundDeltaSumC != float64(n)*8 || got.HeatingBoundRecentN != 2 || got.HeatingBoundRecent != [heatingBoundDays]float64{0, 50, 50} { diff --git a/go/internal/loadmodel/model.go b/go/internal/loadmodel/model.go index c107df727..35601e3a2 100644 --- a/go/internal/loadmodel/model.go +++ b/go/internal/loadmodel/model.go @@ -31,9 +31,11 @@ const heatingBoundStep = 0.5 // heating off must not pull a correct coefficient down. const heatingBoundDays = 3 -// heatingBoundMinSamples is six hours of cold training intervals. -// A day with less cold data does not bound the coefficient. -const heatingBoundMinSamples = int64(6 * time.Hour / TrainingInterval) +// Cold-day totals keep the old minute sampler's units on disk. A new quarter +// contributes fifteen minutes, so restored partial days retain their meaning. +// A day with less than six cold hours does not bound the coefficient. +const heatingBoundMinSamples = int64(6 * time.Hour / time.Minute) +const heatingBoundSampleWeight = int64(TrainingInterval / time.Minute) const PlausibleLoadHeadroom = 1.25 @@ -89,9 +91,9 @@ type Model struct { // MaxPlausibleW remains for stored-state/API compatibility. Grid limits do not cap gross household load. MaxPlausibleW float64 `json:"max_plausible_w,omitempty"` - // Sums over the cold samples of the current local day, and the bounds - // of the last cold days, newest last. They cap the heating coefficient; - // see closeHeatingBoundDay. + // Cold-minute sums for the current local day, and the bounds of the last + // cold days, newest last. Keep minute units for stored-state compatibility. + // They cap the heating coefficient; see closeHeatingBoundDay. HeatingBoundDay string `json:"heating_bound_day,omitempty"` HeatingBoundLoadSumW float64 `json:"heating_bound_load_sum_w,omitempty"` HeatingBoundDeltaSumC float64 `json:"heating_bound_delta_sum_c,omitempty"` @@ -271,9 +273,9 @@ func (m *Model) Update(t time.Time, actualLoadW, tempC float64) (updated bool) { cold := knownTemp && tempC < HeatingReferenceC-HeatingMinDeltaT if cold { - m.HeatingBoundLoadSumW += actualLoadW - m.HeatingBoundDeltaSumC += HeatingReferenceC - tempC - m.HeatingBoundSamples++ + m.HeatingBoundLoadSumW += actualLoadW * float64(heatingBoundSampleWeight) + m.HeatingBoundDeltaSumC += (HeatingReferenceC - tempC) * float64(heatingBoundSampleWeight) + m.HeatingBoundSamples += heatingBoundSampleWeight } if cold && b.Days >= MinTrustSamples { deltaT := HeatingReferenceC - tempC From fe42b908887f0c5fcf57a6243f1ac7465d4cbd6f Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Mon, 19 Oct 2026 20:22:24 +0200 Subject: [PATCH 5/5] fix(forecast): train only on fully measured household intervals History may show Home use across a bridged or estimated gap, but a load model must not learn from it. Leave out intervals with EstimatedLoad or BalanceMismatch as well as a missing source. Co-Authored-By: Claude Opus 5.5 --- go/internal/state/household_intervals.go | 18 +++++++++----- go/internal/state/household_intervals_test.go | 24 +++++++++++++++++++ 2 files changed, 36 insertions(+), 6 deletions(-) diff --git a/go/internal/state/household_intervals.go b/go/internal/state/household_intervals.go index e988edccd..54415188b 100644 --- a/go/internal/state/household_intervals.go +++ b/go/internal/state/household_intervals.go @@ -16,11 +16,17 @@ type HouseholdInterval struct { LoadW float64 } -// HouseholdIntervals returns closed, valid intervals in [from, to) and the end -// of the closed range, so a caller can move past intervals left out. from must -// be aligned to width. An interval is closed once committed source progress -// reaches its end; later writes cannot change it. Intervals with a missing -// source, a recording gap or a negative balance are left out. +// Training is strict where display is lenient: History may show Home use +// across a bridged or estimated gap, but a model learns only from intervals +// measured in full. +const householdTrainingExcluded = MissingPower | BalanceMismatch | EstimatedLoad + +// HouseholdIntervals returns closed, fully measured intervals in [from, to) +// and the end of the closed range, so a caller can move past intervals left +// out. from must be aligned to width. An interval is closed once committed +// source progress reaches its end; later writes cannot change it. Intervals +// with a missing source, a recording gap, an estimate or a negative balance +// are left out. func (s *Store) HouseholdIntervals(ctx context.Context, from, to, width int64) ([]HouseholdInterval, int64, error) { if width <= 0 || from%width != 0 { return nil, from, fmt.Errorf("household intervals: from %d is not aligned to %d ms", from, width) @@ -39,7 +45,7 @@ func (s *Store) HouseholdIntervals(ctx context.Context, from, to, width int64) ( } var out []HouseholdInterval for _, b := range buckets { - if b.width != width || b.covered != width || b.missing&MissingPower != 0 || b.wh < 0 || !finiteHistoryValue(b.wh) { + if b.width != width || b.covered != width || b.missing&householdTrainingExcluded != 0 || b.wh < 0 || !finiteHistoryValue(b.wh) { continue } out = append(out, HouseholdInterval{StartMS: b.start, EndMS: b.start + width, LoadW: b.wh * 3600000 / float64(width)}) diff --git a/go/internal/state/household_intervals_test.go b/go/internal/state/household_intervals_test.go index 3a8c1f2c3..2bc85e709 100644 --- a/go/internal/state/household_intervals_test.go +++ b/go/internal/state/household_intervals_test.go @@ -123,3 +123,27 @@ func TestHouseholdIntervalsWaitForCommittedProgress(t *testing.T) { t.Fatal("unaligned start accepted") } } + +// History may show Home use across an estimated minute; training may not. +func TestHouseholdIntervalsLeaveOutEstimatedMinutes(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24*time.Hour).UnixMilli() - 6*quarterMS + end := base + 3*quarterMS + for at := base; at <= end; at += 5000 { + sourceTick(t, s, at, HistorySources{GridW: 1000}) + } + // A bridged or otherwise estimated minute, as a display rule records it. + if _, err := s.history.Exec(`INSERT INTO history_quality VALUES('source_energy',?,?,?) ON CONFLICT(kind,start_ms,resolution_ms) DO UPDATE SET missing=missing|excluded.missing`, + base+quarterMS+5*60000, ArchiveResolutionMS, EstimatedLoad); err != nil { + t.Fatal(err) + } + got := householdQuarters(t, s, base, end) + if _, ok := got[base+quarterMS]; ok || len(got) != 2 { + t.Fatalf("estimated minute became a training label: %v", got) + } + d, err := s.DailyEnergy(base, end) + if err != nil || d.Missing&MissingLoad != 0 { + t.Fatal("History lost Home use over an estimated minute", err, d.Missing) + } + near(t, "History Wh", d.LoadWh, 3*250) +}