fix(telemetry): derive Home use from energy and preserve history gaps

.changeset/home-use-alignment.md

@@ -0,0 +1,5 @@
+---
+"ftw": patch
+---
+
+Calculate Home use from timestamped charger session energy where available and integrate other sources over the same period. Adapt the period to source reporting and show the estimated mean, period and age. Cloud polls do not refresh source time. Retain signed daily energy only for bounded balance failures that recover, while preserving raw quality, offline gaps and control inputs.

go/internal/api/api.go

@@ -2463,11 +2463,13 @@ func (s *Server) handleEnergyDaily(w http.ResponseWriter, r *http.Request) {
 	for i := days - 1; i >= 0; i-- {
 		dayStart := todayMidnight.AddDate(0, 0, -i)
 		dayKey := dayStart.Format("2006-01-02")
-		isToday := i == 0
+		// A bounded balance episode can close just after midnight. Let its
+		// final intervals arrive before treating the previous day as immutable.
+		cacheReady := dailyEnergyCacheReady(dayStart, now)
 
 		var de state.DayEnergy
-		if isToday {
-			d, err := s.deps.State.DailyEnergy(dayStart.UnixMilli(), now.UnixMilli())
+		if !cacheReady {
+			d, err := s.deps.State.DailyEnergy(dayStart.UnixMilli(), min(now.UnixMilli(), dayStart.AddDate(0, 0, 1).UnixMilli()))
 			if err != nil {
 				slog.Error("handleEnergyDaily: DailyEnergy failed", "err", err, "day", dayKey)
 				http.Error(w, "history load failed", http.StatusInternalServerError)

go/internal/api/api_energy_today.go

@@ -23,6 +23,11 @@ func energyTodayFields(d state.DayEnergy) map[string]any {
 		out[key] = d.Missing.Value(field.flag, field.wh)
 		measured[key] = d.MeasuredValue(field.flag, field.wh)
 	}
+	if d.Missing&state.EstimatedLoad != 0 {
+		// The load total includes inferred intervals. It cannot also appear
+		// in the measured-only fallback used by clients after a data gap.
+		measured["load_wh"] = nil
+	}
 	out["measured"] = measured
 	return out
 }

go/internal/api/api_history_quality.go

@@ -1,18 +1,25 @@
 package api
 
 import (
+	"strings"
+	"time"
+
 	"github.com/srcfl/ftw/go/internal/state"
 	"github.com/srcfl/ftw/go/internal/telemetry"
-	"strings"
 )
 
+func dailyEnergyCacheReady(dayStart, now time.Time) bool {
+	return !now.Before(dayStart.AddDate(0, 0, 1).Add(state.LoadEstimateRecoveryWindow + time.Minute))
+}
+
 func (s *Server) qualifyStatusHistory(resp map[string]any, siteMeter string) {
 	if s.deps.HistoryMeasurementSources == nil {
 		return
 	}
 	opts := s.deps.HistoryMeasurementSources()
 	opts.UseWatchdogTimeout = true
 	r := s.deps.Tel.ForecastMeasurementNow(siteMeter, opts)
+	resp["load_estimate"] = s.deps.Tel.HomeUseEstimate(time.Now(), siteMeter, opts)
 	var missing state.HistoryMissing
 	for kind, flag := range map[telemetry.DerType]state.HistoryMissing{telemetry.DerMeter: state.MissingGrid, telemetry.DerPV: state.MissingPV, telemetry.DerBattery: state.MissingBattery, telemetry.DerEV: state.MissingEV, telemetry.DerV2X: state.MissingV2X} {
 		if !r.Known[kind] {

go/internal/api/api_home_estimate_test.go

@@ -0,0 +1,68 @@
+package api
+
+import (
+	"encoding/json"
+	"math"
+	"testing"
+	"time"
+
+	"github.com/srcfl/ftw/go/internal/state"
+	"github.com/srcfl/ftw/go/internal/telemetry"
+)
+
+func TestStatusHomeEstimateIsSeparateFromInstantaneousPower(t *testing.T) {
+	tel := telemetry.NewStore()
+	tel.EnsureDriverHealth("site")
+	base := time.Now().Add(-20 * time.Second)
+	for i := 0; i <= 10; i++ {
+		data, _ := json.Marshal(map[string]any{"power_observed_at": base.Add(time.Duration(i) * time.Second).Format(time.RFC3339Nano)})
+		tel.Update("site", telemetry.DerMeter, 1000+float64(i)*100, nil, data)
+	}
+	srv := New(&Deps{Tel: tel, HistoryMeasurementSources: func() telemetry.ForecastOptions { return telemetry.ForecastOptions{} }})
+	resp := map[string]any{}
+	for i := 0; i < 3; i++ {
+		srv.qualifyStatusHistory(resp, "site")
+		e, ok := resp["load_estimate"].(telemetry.HomeEstimate)
+		if !ok || e.Watts == nil || math.Abs(*e.Watts-1500) > 1e-6 || e.AgeMS < 15000 || e.AgeMS > 20000 || e.Method != "reported_energy_balance_v3" || e.TimingBasis != "source" || e.EnergyBasis != "power_integral" || e.EnergyWh == nil || e.EndAgeMS < 10000 {
+			t.Fatal(e)
+		}
+		if resp["load_w"] != float64(2000) || resp["grid_w"] != float64(2000) || resp["history_quality"] != "complete" {
+			t.Fatal(resp)
+		}
+	}
+	tel.DriverHealthMut("site").SetOffline()
+	srv.qualifyStatusHistory(resp, "site")
+	if resp["load_w"] != nil || resp["load_estimate"].(telemetry.HomeEstimate).Watts != nil {
+		t.Fatal(resp)
+	}
+}
+
+func TestEnergyTodayKeepsEstimatedLoadOutOfMeasuredFields(t *testing.T) {
+	got := energyTodayFields(state.DayEnergy{Intervals: 2, Missing: state.EstimatedLoad, LoadWh: 22, ImportWh: 10})
+	if got["history_quality"] != "estimated" || got["load_wh"] != float64(22) {
+		t.Fatal(got)
+	}
+	measured := got["measured"].(map[string]any)
+	if measured["load_wh"] != nil || measured["import_wh"] != float64(10) {
+		t.Fatal(measured)
+	}
+}
+
+func TestDailyEnergyCacheWaitsForMidnightRecovery(t *testing.T) {
+	loc, err := time.LoadLocation("Europe/Stockholm")
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Include the 23/25-hour days: a local day is not always 24 hours.
+	for _, day := range []time.Time{time.Date(2026, 10, 17, 0, 0, 0, 0, loc), time.Date(2026, 10, 25, 0, 0, 0, 0, loc), time.Date(2026, 3, 29, 0, 0, 0, 0, loc)} {
+		end := day.AddDate(0, 0, 1)
+		for _, elapsed := range []time.Duration{-time.Second, 0, 30 * time.Second, 59 * time.Second, 2 * time.Minute, 179 * time.Second} {
+			if dailyEnergyCacheReady(day, end.Add(elapsed)) {
+				t.Fatal("cached before recovery can finish", day, elapsed)
+			}
+		}
+		if !dailyEnergyCacheReady(day, end.Add(3*time.Minute)) {
+			t.Fatal("closed day never became cacheable", day)
+		}
+	}
+}

go/internal/state/history_dashboard.go

@@ -42,6 +42,9 @@ func (s *Store) recordDashboardTx(ctx context.Context, tx *sql.Tx, p *HistoryPoi
 		}
 	}
 	if p == nil {
+		if err := clearLoadEstimatePendingTx(ctx, tx); err != nil {
+			return err
+		}
 		if _, err := tx.ExecContext(ctx, `INSERT INTO history_site_cursor VALUES('last',0) ON CONFLICT(key) DO UPDATE SET value=0`); err != nil {
 			return err
 		}
@@ -78,20 +81,23 @@ func (s *Store) recordDashboardTx(ctx context.Context, tx *sql.Tx, p *HistoryPoi
 		}
 	}
 	if p.TsMs > previous {
+		if previous == 0 || p.TsMs-previous > maxLoadEstimateTickMS {
+			if err := clearLoadEstimatePendingTx(ctx, tx); err != nil {
+				return err
+			}
+		}
 		if previous > 0 && p.TsMs-previous <= maxCostIntegrationGap.Milliseconds() {
 			var previousMissing HistoryMissing
 			err := tx.QueryRowContext(ctx, `SELECT value FROM history_partial_cursor WHERE key='missing'`).Scan(&previousMissing)
 			if err != nil && !errors.Is(err, sql.ErrNoRows) {
 				return err
 			}
 			var previousEligible int
-			if previousMissing != 0 {
-				err := tx.QueryRowContext(ctx, `SELECT value FROM history_partial_cursor WHERE key='residual_eligible'`).Scan(&previousEligible)
-				if err != nil && !errors.Is(err, sql.ErrNoRows) {
-					return err
-				}
+			err = tx.QueryRowContext(ctx, `SELECT value FROM history_partial_cursor WHERE key='residual_eligible'`).Scan(&previousEligible)
+			if err != nil && !errors.Is(err, sql.ErrNoRows) {
+				return err
 			}
-			if err := addSiteIntervalEstimateTx(ctx, tx, previous, p, previousMissing, previousMissing == 0 || previousEligible == 1); err != nil {
+			if err := addSiteIntervalEstimateTx(ctx, tx, previous, p, previousMissing, previousEligible == 1); err != nil {
 				return err
 			}
 			if err := addSiteIntervalTx(ctx, tx, previous, p, previousMissing); err != nil {
@@ -158,7 +164,9 @@ func addSiteIntervalTx(ctx context.Context, tx *sql.Tx, previous int64, p *Histo
 	point.ClearMissingValues()
 	p = &point
 	table, kind := "history_site_energy", "energy"
-	if p.Missing != 0 {
+	// SoC does not enter the power balance. Keep its quality bit, but do
+	// not mix otherwise known power into a failed-balance energy bucket.
+	if p.Missing&^MissingSoC != 0 {
 		table, kind = "history_site_energy_partial", "partial_energy"
 	}
 	for from := previous; from < p.TsMs; {
@@ -217,7 +225,7 @@ type siteEnergyRow struct {
 func (s *Store) walkSiteEnergy(ctx context.Context, from, to int64, visit func(siteEnergyRow) error) error {
 	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
 	defer cancel()
-	rows, err := s.history.QueryContext(ctx, `WITH energy AS (SELECT *, 'energy' AS kind FROM history_site_energy UNION ALL SELECT *, 'partial_energy' AS kind FROM history_site_energy_partial) SELECT e.start_ms,e.resolution_ms,e.first_ms,e.last_ms,e.covered_ms,e.n,e.import_wh,e.export_wh,e.pv_wh,e.bat_charge_wh,e.bat_discharge_wh,e.load_wh,e.ev_wh,COALESCE(q.missing,0),CASE WHEN e.kind='partial_energy' AND q.missing IN (136,40) THEN COALESCE((SELECT SUM(load_wh) FROM history_load_estimates x WHERE x.start_ms>=e.start_ms AND x.start_ms<e.start_ms+e.resolution_ms AND x.start_ms+x.resolution_ms<=e.start_ms+e.resolution_ms),0) ELSE 0 END,CASE WHEN e.kind='partial_energy' AND q.missing IN (136,40) THEN COALESCE((SELECT SUM(covered_ms) FROM history_load_estimates x WHERE x.start_ms>=e.start_ms AND x.start_ms<e.start_ms+e.resolution_ms AND x.start_ms+x.resolution_ms<=e.start_ms+e.resolution_ms),0) ELSE 0 END FROM energy e LEFT JOIN history_quality q ON q.kind=e.kind AND q.start_ms=e.start_ms AND q.resolution_ms=e.resolution_ms WHERE e.start_ms>=? AND e.start_ms<? AND first_ms>=? AND last_ms<=? ORDER BY e.start_ms`, bucketStart(from, EnergyLedgerDailyBucketMS), to, from, to)
+	rows, err := s.history.QueryContext(ctx, `WITH energy AS (SELECT *, 'energy' AS kind FROM history_site_energy UNION ALL SELECT *, 'partial_energy' AS kind FROM history_site_energy_partial) SELECT e.start_ms,e.resolution_ms,e.first_ms,e.last_ms,e.covered_ms,e.n,e.import_wh,e.export_wh,e.pv_wh,e.bat_charge_wh,e.bat_discharge_wh,e.load_wh,e.ev_wh,COALESCE(q.missing,0),CASE WHEN e.kind='partial_energy' AND (q.missing & ~16) IN (136,40) THEN COALESCE((SELECT SUM(load_wh) FROM history_load_estimates x WHERE x.start_ms>=e.start_ms AND x.start_ms<e.start_ms+e.resolution_ms AND x.start_ms+x.resolution_ms<=e.start_ms+e.resolution_ms),0) ELSE 0 END,CASE WHEN e.kind='partial_energy' AND (q.missing & ~16) IN (136,40) THEN COALESCE((SELECT SUM(covered_ms) FROM history_load_estimates x WHERE x.start_ms>=e.start_ms AND x.start_ms<e.start_ms+e.resolution_ms AND x.start_ms+x.resolution_ms<=e.start_ms+e.resolution_ms),0) ELSE 0 END FROM energy e LEFT JOIN history_quality q ON q.kind=e.kind AND q.start_ms=e.start_ms AND q.resolution_ms=e.resolution_ms WHERE e.start_ms>=? AND e.start_ms<? AND first_ms>=? AND last_ms<=? ORDER BY e.start_ms`, bucketStart(from, EnergyLedgerDailyBucketMS), to, from, to)
 	if err != nil {
 		return err
 	}
@@ -248,9 +256,9 @@ func (s *Store) aggregateDayEnergy(ctx context.Context, from, to int64) (DayEner
 		originalLoad += r.load
 		originalMissing |= r.missing
 		coveredMS += r.covered
-		if r.missing == MissingLoad|BalanceMismatch || r.missing == MissingLoad|MissingEV {
+		if r.missing&^MissingSoC == MissingLoad|BalanceMismatch || r.missing&^MissingSoC == MissingLoad|MissingEV {
 			if r.estimateCovered > 0 && r.estimateCovered == r.covered {
-				r.load, r.missing = r.estimateLoad, EstimatedLoad
+				r.load, r.missing = r.estimateLoad, (r.missing&MissingSoC)|EstimatedLoad
 				estimatedMS += r.covered
 			}
 		}

go/internal/state/history_load_estimate.go

@@ -5,6 +5,7 @@ import (
 	"database/sql"
 	"encoding/json"
 	"math"
+	"time"
 )
 
 // Estimates supplement, never overwrite, the original measurements and quality.
@@ -15,8 +16,20 @@ const loadEstimateSchema = `CREATE TABLE IF NOT EXISTS history_load_estimates (
  start_ms INTEGER NOT NULL,resolution_ms INTEGER NOT NULL,covered_ms INTEGER NOT NULL,
  load_wh REAL NOT NULL,method TEXT NOT NULL,PRIMARY KEY(start_ms,resolution_ms)) WITHOUT ROWID`
 
+// Pending evidence never enters totals. A valid balance must close a short
+// episode before any of its intervals can supplement the original history.
+const loadEstimatePendingSchema = `CREATE TABLE IF NOT EXISTS history_load_estimate_pending (
+ from_ms INTEGER PRIMARY KEY,to_ms INTEGER NOT NULL,load_w REAL NOT NULL) WITHOUT ROWID`
+
+// LoadEstimateRecoveryWindow bounds a recovered balance episode, including a
+// minute-scale cloud lag. It is a history estimate budget, not a sensor delay.
+const LoadEstimateRecoveryWindow = 2 * time.Minute
+const maxLoadEstimateEpisodeMS = int64(LoadEstimateRecoveryWindow / time.Millisecond)
+const maxLoadEstimateTickMS int64 = 15_000
+
 func residualHouseholdW(p *HistoryPoint) (float64, bool) {
-	if p.Missing != 0 && p.Missing != MissingLoad|BalanceMismatch {
+	missing := p.Missing &^ MissingSoC
+	if missing != 0 && missing != MissingLoad|BalanceMismatch {
 		return 0, false
 	}
 	var evidence struct {
@@ -27,7 +40,7 @@ func residualHouseholdW(p *HistoryPoint) (float64, bool) {
 	if json.Unmarshal([]byte(p.JSON), &evidence) != nil || evidence.EV == nil || evidence.V2X == nil {
 		return 0, false
 	}
-	if p.Missing != 0 && evidence.Reason != "invalid_balance" {
+	if missing != 0 && evidence.Reason != "invalid_balance" {
 		return 0, false
 	}
 	w := p.GridW - p.PVW - p.BatW - *evidence.EV - *evidence.V2X
@@ -37,16 +50,72 @@ func residualHouseholdW(p *HistoryPoint) (float64, bool) {
 func addSiteIntervalEstimateTx(ctx context.Context, tx *sql.Tx, previous int64, p *HistoryPoint, previousMissing HistoryMissing, previousEligible bool) error {
 	// An offline, duplicate, skewed or otherwise missing source is not repaired
 	// by interpreting it as zero. Both endpoints need usable source evidence.
-	if p.Missing|previousMissing != MissingLoad|BalanceMismatch || !previousEligible {
-		return nil
-	}
+	missing := (p.Missing | previousMissing) &^ MissingSoC
 	w, ok := residualHouseholdW(p)
-	if !ok {
+	if missing != MissingLoad|BalanceMismatch || !previousEligible || !ok || p.TsMs-previous > maxLoadEstimateTickMS {
+		return clearLoadEstimatePendingTx(ctx, tx)
+	}
+	var start int64
+	if err := tx.QueryRowContext(ctx, `SELECT COALESCE(MIN(from_ms),?) FROM history_load_estimate_pending`, previous).Scan(&start); err != nil {
+		return err
+	}
+	if p.TsMs-start > maxLoadEstimateEpisodeMS {
+		// Keep a marker until a valid snapshot arrives. Discarding only the
+		// rows would let a long failure look like repeated short episodes.
+		if err := clearLoadEstimatePendingTx(ctx, tx); err != nil {
+			return err
+		}
+		if p.Missing&MissingLoad == 0 {
+			return nil
+		}
+		_, err := tx.ExecContext(ctx, `INSERT INTO history_load_estimate_pending VALUES(?, ?, 0)`, start, p.TsMs)
+		return err
+	}
+	if _, err := tx.ExecContext(ctx, `INSERT INTO history_load_estimate_pending VALUES(?,?,?)`, previous, p.TsMs, w); err != nil {
+		return err
+	}
+	if p.Missing&MissingLoad != 0 {
 		return nil
 	}
-	for from := previous; from < p.TsMs; {
+	rows, err := tx.QueryContext(ctx, `SELECT from_ms,to_ms,load_w FROM history_load_estimate_pending ORDER BY from_ms`)
+	if err != nil {
+		return err
+	}
+	type interval struct {
+		from, to int64
+		watts    float64
+	}
+	var intervals []interval
+	for rows.Next() {
+		var v interval
+		if err := rows.Scan(&v.from, &v.to, &v.watts); err != nil {
+			rows.Close()
+			return err
+		}
+		intervals = append(intervals, v)
+	}
+	if err := rows.Err(); err != nil {
+		rows.Close()
+		return err
+	}
+	rows.Close()
+	for _, v := range intervals {
+		if err := writeLoadEstimateTx(ctx, tx, v.from, v.to, v.watts); err != nil {
+			return err
+		}
+	}
+	return clearLoadEstimatePendingTx(ctx, tx)
+}
+
+func clearLoadEstimatePendingTx(ctx context.Context, tx *sql.Tx) error {
+	_, err := tx.ExecContext(ctx, `DELETE FROM history_load_estimate_pending`)
+	return err
+}
+
+func writeLoadEstimateTx(ctx context.Context, tx *sql.Tx, previous, until int64, w float64) error {
+	for from := previous; from < until; {
 		start := bucketStart(from, ArchiveResolutionMS)
-		to := min(p.TsMs, start+ArchiveResolutionMS)
+		to := min(until, start+ArchiveResolutionMS)
 		covered := to - from
 		_, err := tx.ExecContext(ctx, `INSERT INTO history_load_estimates VALUES(?,?,?,?,?)
    ON CONFLICT(start_ms,resolution_ms) DO UPDATE SET covered_ms=covered_ms+excluded.covered_ms,load_wh=load_wh+excluded.load_wh`, start, ArchiveResolutionMS, covered, w*float64(covered)/3600000, "signed_source_balance_v1")