fix(history): retain estimated daily use across brief balance failures

.changeset/repair-daily-balance.md

@@ -0,0 +1,5 @@
+---
+"ftw": patch
+---
+
+Keep signed source-balance estimates for short measurement imbalances, preserving original history quality and separating estimated household energy from missing data.

go/internal/api/api.go

@@ -2466,7 +2466,15 @@ func (s *Server) handleEnergyDaily(w http.ResponseWriter, r *http.Request) {
 		if de.Intervals == 0 {
 			de.Missing |= state.MissingPower
 		}
+		periodEnd := min(now.UnixMilli(), dayStart.AddDate(0, 0, 1).UnixMilli())
+		coverage, err := s.deps.State.HistoryEnergyCoverage(r.Context(), dayStart.UnixMilli(), periodEnd)
+		if err != nil {
+			http.Error(w, "history coverage unavailable", http.StatusInternalServerError)
+			return
+		}
 		out = append(out, map[string]any{
+			"coverage_ms":       coverage,
+			"period_ms":         periodEnd - dayStart.UnixMilli(),
 			"history_quality":   de.Missing.Quality(),
 			"day":               dayKey,
 			"import_wh":         de.Missing.Value(state.MissingGrid, de.ImportWh),

go/internal/state/history_dashboard.go

@@ -84,6 +84,16 @@ func (s *Store) recordDashboardTx(ctx context.Context, tx *sql.Tx, p *HistoryPoi
 			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
+				}
+			}
+			if err := addSiteIntervalEstimateTx(ctx, tx, previous, p, previousMissing, previousMissing == 0 || previousEligible == 1); err != nil {
+				return err
+			}
 			if err := addSiteIntervalTx(ctx, tx, previous, p, previousMissing); err != nil {
 				return err
 			}
@@ -93,6 +103,13 @@ func (s *Store) recordDashboardTx(ctx context.Context, tx *sql.Tx, p *HistoryPoi
 		}
 	}
 	if p.TsMs > previous {
+		eligible := 0
+		if _, ok := residualHouseholdW(p); ok {
+			eligible = 1
+		}
+		if _, err := tx.ExecContext(ctx, `INSERT INTO history_partial_cursor VALUES('residual_eligible',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, eligible); err != nil {
+			return err
+		}
 		if _, err := tx.ExecContext(ctx, `INSERT INTO history_partial_cursor VALUES('missing',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, p.Missing); err != nil {
 			return err
 		}
@@ -187,6 +204,8 @@ func siteEVWh(p *HistoryPoint, hours float64) float64 {
 }
 
 type siteEnergyRow struct {
+	estimateLoad                              float64
+	estimateCovered                           int64
 	missing                                   HistoryMissing
 	start, width, first, last, covered, n     int64
 	imp, exp, pv, charge, discharge, load, ev float64
@@ -198,14 +217,14 @@ 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) 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 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)
 	if err != nil {
 		return err
 	}
 	defer rows.Close()
 	for rows.Next() {
 		var r siteEnergyRow
-		if err := rows.Scan(&r.start, &r.width, &r.first, &r.last, &r.covered, &r.n, &r.imp, &r.exp, &r.pv, &r.charge, &r.discharge, &r.load, &r.ev, &r.missing); err != nil {
+		if err := rows.Scan(&r.start, &r.width, &r.first, &r.last, &r.covered, &r.n, &r.imp, &r.exp, &r.pv, &r.charge, &r.discharge, &r.load, &r.ev, &r.missing, &r.estimateLoad, &r.estimateCovered); err != nil {
 			return err
 		}
 		if err := visit(r); err != nil {
@@ -217,12 +236,24 @@ func (s *Store) walkSiteEnergy(ctx context.Context, from, to int64, visit func(s
 
 func (s *Store) aggregateDayEnergy(ctx context.Context, from, to int64) (DayEnergy, error) {
 	var out DayEnergy
+	var originalLoad float64
+	var originalMissing HistoryMissing
+	var estimatedMS, coveredMS int64
 	err := s.walkSiteEnergy(ctx, from, to, func(r siteEnergyRow) error {
 		out.ImportWh += r.imp
 		out.ExportWh += r.exp
 		out.PVWh += r.pv
 		out.BatChargedWh += r.charge
 		out.BatDischargedWh += r.discharge
+		originalLoad += r.load
+		originalMissing |= r.missing
+		coveredMS += r.covered
+		if r.missing == MissingLoad|BalanceMismatch || r.missing == MissingLoad|MissingEV {
+			if r.estimateCovered > 0 && r.estimateCovered == r.covered {
+				r.load, r.missing = r.estimateLoad, EstimatedLoad
+				estimatedMS += r.covered
+			}
+		}
 		out.LoadWh += r.load
 		out.Intervals += r.n
 		out.Missing |= r.missing
@@ -231,5 +262,9 @@ func (s *Store) aggregateDayEnergy(ctx context.Context, from, to int64) (DayEner
 		}
 		return nil
 	})
+	if out.LoadWh < 0 || (estimatedMS > 120000 && estimatedMS > coveredMS/20) {
+		out.LoadWh = originalLoad
+		out.Missing = originalMissing
+	}
 	return out, err
 }

go/internal/state/history_load_estimate.go

@@ -0,0 +1,70 @@
+package state
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"math"
+)
+
+// Estimates supplement, never overwrite, the original measurements and quality.
+// Signed residuals are integrated before taking a daily total: clipping each
+// negative instantaneous balance biases the total upwards. Forecast training
+// and control continue to reject those instantaneous samples.
+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`
+
+func residualHouseholdW(p *HistoryPoint) (float64, bool) {
+	if p.Missing != 0 && p.Missing != MissingLoad|BalanceMismatch {
+		return 0, false
+	}
+	var evidence struct {
+		EV     *float64 `json:"ev_w"`
+		V2X    *float64 `json:"v2x_w"`
+		Reason string   `json:"history_reason"`
+	}
+	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" {
+		return 0, false
+	}
+	w := p.GridW - p.PVW - p.BatW - *evidence.EV - *evidence.V2X
+	return w, !math.IsNaN(w) && !math.IsInf(w, 0)
+}
+
+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
+	}
+	w, ok := residualHouseholdW(p)
+	if !ok {
+		return nil
+	}
+	for from := previous; from < p.TsMs; {
+		start := bucketStart(from, ArchiveResolutionMS)
+		to := min(p.TsMs, 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")
+		if err != nil {
+			return err
+		}
+		from = to
+	}
+	return nil
+}
+
+// HistoryEnergyCoverage reports actual integrated time, excluding writer stalls
+// and offline gaps. It does not extrapolate energy into unrecorded periods.
+func (s *Store) HistoryEnergyCoverage(ctx context.Context, from, to int64) (int64, error) {
+	var covered int64
+	err := s.history.QueryRowContext(ctx, `SELECT COALESCE(SUM(covered_ms),0) FROM (
+ SELECT start_ms,first_ms,last_ms,covered_ms FROM history_site_energy
+ UNION ALL SELECT start_ms,first_ms,last_ms,covered_ms FROM history_site_energy_partial
+ ) WHERE start_ms>=? AND start_ms<? AND first_ms>=? AND last_ms<=?`, from, to, from, to).Scan(&covered)
+	return covered, err
+}

go/internal/state/history_load_estimate_test.go

@@ -0,0 +1,95 @@
+package state
+
+import (
+	"context"
+	"fmt"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+func TestSignedBalanceEstimateKeepsNegativeTransientsOutOfTraining(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "state.db")
+	s, err := Open(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := s.EnableHistoryAggregation(); err != nil {
+		t.Fatal(err)
+	}
+	base := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli()
+	// Same real 1 kW house, asynchronously reported battery step: household
+	// residual goes -1 kW then +3 kW. Summing signed source energy gives 1 kW.
+	for i, w := range []float64{1000, -1000, 3000, 1000} {
+		p := HistoryPoint{TsMs: base + int64(i)*10000, GridW: 1000, BatW: 1000 - w, LoadW: w, JSON: `{"ev_w":0,"v2x_w":0,"history_reason":""}`}
+		if w < 0 {
+			p.Missing = MissingLoad | BalanceMismatch
+			p.LoadW = 0
+			p.JSON = `{"ev_w":0,"v2x_w":0,"history_reason":"invalid_balance"}`
+		}
+		dashboardTick(t, s, fmt.Sprint(i), &p)
+	}
+	check := func(s *Store) {
+		t.Helper()
+		d, e := s.DailyEnergy(base, base+60000)
+		if e != nil {
+			t.Fatal(e)
+		}
+		near(t, "signed house energy", d.LoadWh, 1000*30.0/3600)
+		covered, err := s.HistoryEnergyCoverage(context.Background(), base, base+60000)
+		if err != nil || covered != 30000 {
+			t.Fatal(covered, err)
+		}
+
+		if d.Missing != EstimatedLoad || d.Missing.Quality() != "estimated" {
+			t.Fatal(d)
+		}
+		var mask HistoryMissing
+		if err := s.history.QueryRow(`SELECT missing FROM history_quality WHERE kind='partial_dashboard' LIMIT 1`).Scan(&mask); err != nil || mask != MissingLoad|BalanceMismatch {
+			t.Fatal(mask, err)
+		}
+	}
+	check(s)
+	if err := s.maintainDashboard(context.Background(), time.Now().Add(40*24*time.Hour)); err != nil {
+		t.Fatal(err)
+	}
+	check(s)
+	if err := s.Close(); err != nil {
+		t.Fatal(err)
+	}
+	s, err = Open(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+	check(s)
+}
+
+func TestResidualEstimateRejectsMissingSkewedAndDuplicateSources(t *testing.T) {
+	for _, tc := range []struct {
+		mask   HistoryMissing
+		reason string
+	}{
+		{MissingLoad | MissingBattery, "missing:battery"}, {MissingLoad | MissingEV, "missing:ev"},
+		{MissingLoad | BalanceMismatch, "time_skew"}, {MissingLoad | BalanceMismatch, "duplicate_flow:battery"},
+	} {
+		p := HistoryPoint{Missing: tc.mask, JSON: fmt.Sprintf(`{"ev_w":0,"v2x_w":0,"history_reason":%q}`, tc.reason)}
+		if _, ok := residualHouseholdW(&p); ok {
+			t.Fatal(tc)
+		}
+	}
+	// A physically impossible daily total must remain unknown.
+	s := freshStore(t)
+	if err := s.EnableHistoryAggregation(); err != nil {
+		t.Fatal(err)
+	}
+	base := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli()
+	for i := 0; i < 4; i++ {
+		p := HistoryPoint{TsMs: base + int64(i)*10000, GridW: 1000, BatW: 4000, Missing: MissingLoad | BalanceMismatch, JSON: `{"ev_w":0,"v2x_w":0,"history_reason":"invalid_balance"}`}
+		dashboardTick(t, s, fmt.Sprint(i), &p)
+	}
+	d, err := s.DailyEnergy(base, base+60000)
+	if err != nil || d.Missing&MissingLoad == 0 {
+		t.Fatal(d, err)
+	}
+}

go/internal/state/history_plain.go

@@ -37,7 +37,7 @@ var plainHistoryTables = []string{
 	"ts_drivers", "ts_metrics", "ts_latest",
 	"energy_daily", "energy_ledger_meta", "energy_assets", "energy_ledger_entries", "energy_ledger_cursors",
 	"ts_buckets", "ts_aggregate_hours", "ts_series_hour", "ts_archive_days",
-	"history_dashboard", "history_site_energy", "history_site_cursor", "history_quality", "history_daily_quality", "history_dashboard_partial", "history_site_energy_partial", "history_partial_cursor", "energy_daily_partial",
+	"history_load_estimates", "history_dashboard", "history_site_energy", "history_site_cursor", "history_quality", "history_daily_quality", "history_dashboard_partial", "history_site_energy_partial", "history_partial_cursor", "energy_daily_partial",
 	"history_migrations", "history_sqlite_progress",
 }
 

go/internal/state/history_quality.go

@@ -19,6 +19,7 @@ const (
 	MissingEV
 	MissingV2X
 	BalanceMismatch
+	EstimatedLoad
 )
 const MissingPower = MissingGrid | MissingPV | MissingBattery | MissingLoad | MissingEV | MissingV2X
 const historyQualitySchema = `CREATE TABLE IF NOT EXISTS history_quality(kind TEXT NOT NULL,start_ms INTEGER NOT NULL,resolution_ms INTEGER NOT NULL,missing INTEGER NOT NULL,PRIMARY KEY(kind,start_ms,resolution_ms)) WITHOUT ROWID`
@@ -54,6 +55,9 @@ func (m HistoryMissing) Quality() string {
 	if m&MissingPower != 0 {
 		return "incomplete"
 	}
+	if m&EstimatedLoad != 0 {
+		return "estimated"
+	}
 	return "complete"
 }
 func putHistoryQuality(ctx context.Context, tx *sql.Tx, kind string, start, width int64, missing HistoryMissing) error {

go/internal/state/history_schema.go

@@ -4,7 +4,7 @@ import "strings"
 
 var historySchema = append(append([]string{}, sqliteLegacyHistoryStmts...),
 	`CREATE TABLE IF NOT EXISTS ts_latest (driver_id INTEGER NOT NULL,metric_id INTEGER NOT NULL,ts_ms INTEGER NOT NULL,value REAL NOT NULL,PRIMARY KEY(driver_id,metric_id)) WITHOUT ROWID`,
-	dashboardSchema, siteEnergySchema, siteEnergyCursorSchema, historyQualitySchema, historyDailyQualitySchema,
+	loadEstimateSchema, dashboardSchema, siteEnergySchema, siteEnergyCursorSchema, historyQualitySchema, historyDailyQualitySchema,
 	strings.Replace(dashboardSchema, "history_dashboard", "history_dashboard_partial", 1), strings.Replace(siteEnergySchema, "history_site_energy", "history_site_energy_partial", 1), strings.Replace(siteEnergyCursorSchema, "history_site_cursor", "history_partial_cursor", 1),
 	`CREATE TABLE IF NOT EXISTS energy_daily_partial(day TEXT PRIMARY KEY,import_wh REAL NOT NULL,export_wh REAL NOT NULL,pv_wh REAL NOT NULL,bat_charged_wh REAL NOT NULL,bat_discharged_wh REAL NOT NULL,load_wh REAL NOT NULL,computed_at_ms INTEGER NOT NULL) STRICT`,
 	`CREATE INDEX IF NOT EXISTS idx_partial_dashboard_last ON history_dashboard_partial(last_ms)`,

go/internal/state/history_sqlite.go

@@ -26,7 +26,7 @@ var historyTables = []string{
 }
 var sqliteHistoryBaseTables = append(append([]string{}, historyTables...), "ts_series_hour", "ts_archive_days", "ts_buckets", "ts_bucket_days", "ts_legacy_bucket_days", "ts_aggregate_hours", "ts_latest", "history_dashboard", "history_site_energy", "history_site_cursor")
 var historyQualityTables = []string{"history_quality", "history_daily_quality", "history_dashboard_partial", "history_site_energy_partial", "history_partial_cursor", "energy_daily_partial"}
-var sqliteHistoryTables = append(append([]string{}, sqliteHistoryBaseTables...), historyQualityTables...)
+var sqliteHistoryTables = append(append(append([]string{}, sqliteHistoryBaseTables...), historyQualityTables...), "history_load_estimates")
 
 func ensureHistorySchema(exec func(string) error) error {
 	for _, stmt := range historySchema {
@@ -366,7 +366,7 @@ func historyOrder(table string) string {
 		return " ORDER BY path"
 	case "ts_latest":
 		return " ORDER BY driver_id, metric_id"
-	case "history_dashboard", "history_site_energy", "history_dashboard_partial", "history_site_energy_partial":
+	case "history_load_estimates", "history_dashboard", "history_site_energy", "history_dashboard_partial", "history_site_energy_partial":
 		return " ORDER BY start_ms, resolution_ms"
 	case "ts_buckets":
 		return " ORDER BY driver_id, metric_id, start_ms, resolution_ms"
@@ -487,12 +487,20 @@ func (s *Store) exportHistoryToSQLite(ctx context.Context, path string) error {
 		}
 		qualityTables++
 	}
-	tables := sqliteHistoryTables
+	tables := append(append([]string{}, sqliteHistoryBaseTables...), historyQualityTables...)
 	if qualityTables == 0 {
 		tables = sqliteHistoryBaseTables
 	} else if qualityTables != len(historyQualityTables) {
 		return errors.New("incomplete history quality tables; keep the source and restore a verified full backup")
 	}
+	// Estimates are additive and optional in published older backups.
+	var estimatesExist int
+	if err := src.QueryRowContext(ctx, `SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='history_load_estimates'`).Scan(&estimatesExist); err != nil {
+		return err
+	}
+	if estimatesExist != 0 {
+		tables = append(append([]string{}, tables...), "history_load_estimates")
+	}
 	// This is an unpublished scratch file. Flush each small commit, then
 	// pause, so the next goal/session sync does not inherit a dirty backlog.
 	dest, err := openBackupDestination(path)

web/energy-history-integration.test.mjs

@@ -80,7 +80,7 @@ test('raw energy rows load only when opened, flag rejected data and page locally
     provenance: index === 0 ? 'implausible_energy' : 'counter',
   }));
   const days = [{
-    day: '2026-07-24', import_wh: 4000, export_wh: 1000,
+    history_quality: 'estimated', day: '2026-07-24', import_wh: 4000, export_wh: 1000,
     pv_wh: 9000, load_wh: 12000, bat_charged_wh: 2000, bat_discharged_wh: 1500,
   }];
   let dailyRequests = 0;
@@ -132,8 +132,19 @@ test('raw energy rows load only when opened, flag rejected data and page locally
   assert.equal(elements.get('energy-history-prev').disabled, true);
   assert.equal(elements.get('energy-history-next').disabled, false);
 
+  assert.match(elements.get('energy-history-summary').innerHTML, /Includes estimated home use/);
+  assert.match(elements.get('energy-history-insight').innerHTML, /Home use includes estimates/);
+  assert.match(elements.get('energy-history-legend').innerHTML, /Home use includes estimates/);
   elements.get('energy-history-next').click();
   assert.equal(elements.get('energy-history-page').textContent, 'Page 2 of 3');
   assert.equal((elements.get('energy-history-rows').innerHTML.match(/<tr>/g) || []).length, 50);
   assert.equal(ledgerRequests, 1);
+  days[0].coverage_ms = 7_200_000;
+  days[0].period_ms = 10_800_000;
+  details.open = false;
+  window.ftwEnergyHistoryLoad();
+  await new Promise((resolve) => setImmediate(resolve));
+  await new Promise((resolve) => setImmediate(resolve));
+  assert.match(elements.get('energy-history-summary').innerHTML, /Recorded periods only/);
+  assert.match(elements.get('energy-history-insight').innerHTML, /Some periods were not recorded/);
 });