fix(forecast): train load models on closed history intervals

.changeset/load-models-use-history-intervals.md

@@ -0,0 +1,7 @@
+---
+"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.
+
+Keep cold-day counters in their stored minute units when training moves to quarter-hours, preserving learned profiles and partial-day evidence.

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,63 @@ 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)
+	}
+}

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 })

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)
 		}
 	}
 }