Optimizer: blend measured values into forecast inputs (#31975)

core/metrics/collector.go

@@ -170,6 +170,19 @@ func (c *Collector) EnergyProfile(from time.Time) (*[96]float64, error) {
 	return energyProfile(c.entity, from)
 }
 
+// LastSlotEnergy returns the energy in kWh of the most recently completed
+// 15min slot, or false when it has not been persisted (boot, data gap) or
+// contains recovered downtime energy.
+func (c *Collector) LastSlotEnergy() (float64, bool) {
+	ts := c.accu.clock.Now().Truncate(tariff.SlotDuration).Add(-tariff.SlotDuration)
+
+	var m meter
+	if db.Instance.Where("meter = ? AND ts = ? AND COALESCE(recovered, 0) = 0", c.entity.Id, ts.Unix()).Limit(1).Find(&m).RowsAffected == 0 {
+		return 0, false
+	}
+	return m.Energy, true
+}
+
 func (c *Collector) SetEnergyMeterTotal(v float64) error {
 	return c.process(func() {
 		c.accu.SetEnergyMeterTotal(v)

core/metrics/collector_test.go

@@ -533,3 +533,39 @@ func TestCreateEntityReconcileGuard(t *testing.T) {
 	require.NoError(t, db.Instance.First(&stored, meterRow.Id).Error)
 	require.Equal(t, Meter, stored.Group)
 }
+
+func TestCollectorLastSlotEnergy(t *testing.T) {
+	clk := clock.NewMock() // 1970-01-01 00:00:00 UTC, on a slot boundary
+
+	require.NoError(t, db.NewInstance("sqlite", ":memory:"))
+	require.NoError(t, SetupSchema())
+
+	col, err := NewCollector(Home, "last", "", WithClock(clk))
+	require.NoError(t, err)
+
+	// nothing persisted yet
+	_, ok := col.LastSlotEnergy()
+	require.False(t, ok)
+
+	// constant 1 kW: crossing into 00:30 persists slot 00:15
+	require.NoError(t, col.AddEnergy(nil, nil, 1e3))
+	clk.Add(15 * time.Minute) // 00:15
+	require.NoError(t, col.AddEnergy(nil, nil, 1e3))
+	clk.Add(15 * time.Minute) // 00:30
+	require.NoError(t, col.AddEnergy(nil, nil, 1e3))
+
+	v, ok := col.LastSlotEnergy()
+	require.True(t, ok)
+	require.InDelta(t, 0.25, v, 1e-10) // kWh
+
+	// mid-slot the previous slot stays the reference
+	clk.Add(5 * time.Minute) // 00:35
+	v, ok = col.LastSlotEnergy()
+	require.True(t, ok)
+	require.InDelta(t, 0.25, v, 1e-10)
+
+	// a recovered slot is excluded
+	require.NoError(t, db.Instance.Model(new(meter)).Where("meter = ? AND ts = ?", col.entity.Id, 15*60).Update("recovered", true).Error)
+	_, ok = col.LastSlotEnergy()
+	require.False(t, ok)
+}

core/optimizer.md

@@ -28,6 +28,14 @@ TODO
 
 Collected 15min energy profile averaged over the last 30 days.
 
+### Measured value blending
+
+The solar forecast and the base load profile are anchored to the current situation
+using the last completed 15min metrics slot, decaying linearly over 4 slots:
+
+- base load: the measured home consumption replaces the first slot and decays into the profile
+- solar: the scale factor measured production/forecasted production is applied to the first slot and decays towards 1
+
 ### End of forecast commercial value
 
 Use minimum of energy consumption cost.

core/site_optimizer.go

@@ -47,6 +47,9 @@ var optimizerChargingStrategies = []string{
 
 const defaultOptimizerChargingStrategy = string(optimizer.OptimizerStrategyChargingStrategyChargeBeforeExport)
 
+// optimizerDecaySlots is the number of slots over which measured values decay into the forecast
+const optimizerDecaySlots = 4
+
 // triggerOptimizer re-runs the optimizer immediately so a changed setting takes
 // effect without waiting for the next slot. It is a no-op when the optimizer is
 // not active or a run is already in progress; the running update reflects the
@@ -290,6 +293,13 @@ func (site *Site) optimizerUpdate(battery []types.Measurement) error {
 		return err
 	}
 
+	// blend measured energy of the last metrics slot into the first slots
+	if v := site.measuredSlotEnergy(metrics.Home); v > 0 {
+		orig := slices.Clone(gt[:min(optimizerDecaySlots, len(gt))])
+		blendMeasured(gt, v, optimizerDecaySlots)
+		site.log.DEBUG.Printf("optimizer: home slots updated with measured %.0fWh: %.0f -> %.0f", v, orig, gt[:len(orig)])
+	}
+
 	// allow empty solar forecast
 	ft := lo.RepeatBy(minLen, func(i int) float32 { return float32(0) })
 	if solarTariff != nil && len(solar) > 0 {
@@ -298,7 +308,16 @@ func (site *Site) optimizerUpdate(battery []types.Measurement) error {
 			return err
 		}
 
-		ft = prorate(scaleAndPrune(solarEnergy, site.effectiveSolarScale(), minLen), firstSlotDuration)
+		scale := site.effectiveSolarScale()
+		ftSlots := scaleAndPrune(solarEnergy, scale, minLen)
+
+		// decay the scale derived from measured vs forecasted energy of the last completed slot
+		if pv, fcst := site.measuredSlotEnergy(site.Meters.PVMetersRef...), site.measuredSlotEnergy(metrics.Forecast)*scale; pv > 0 && fcst > 0 {
+			orig := slices.Clone(ftSlots[:min(optimizerDecaySlots, len(ftSlots))])
+			blendScale(ftSlots, pv/fcst, optimizerDecaySlots)
+			site.log.DEBUG.Printf("optimizer: pv slots updated with scale %.2f: %.0f -> %.0f", pv/fcst, orig, ftSlots[:len(orig)])
+		}
+		ft = prorate(ftSlots, firstSlotDuration)
 	}
 
 	req := optimizer.OptimizationInput{
@@ -784,6 +803,45 @@ func profileSlotsFromNow(profile []float64) []float64 {
 	return profile[firstSlot:]
 }
 
+// measuredSlotEnergy returns the summed energy in Wh of the last completed
+// metrics slot for the given collector refs, 0 when not available
+func (site *Site) measuredSlotEnergy(refs ...string) float64 {
+	var sum float64
+	for _, ref := range refs {
+		c, ok := site.collectors[ref]
+		if !ok {
+			return 0
+		}
+
+		v, ok := c.LastSlotEnergy()
+		if !ok {
+			return 0
+		}
+		sum += v
+	}
+
+	return sum * 1e3
+}
+
+// blendMeasured decays the first slots from the measured value into the
+// forecast. Slot 0 uses the measured value, the forecast takes over from
+// slot decaySlots on.
+func blendMeasured[T constraints.Float](slots []T, measured T, decaySlots int) {
+	for i := range min(decaySlots, len(slots)) {
+		w := T(decaySlots-i) / T(decaySlots)
+		slots[i] = w*measured + (1-w)*slots[i]
+	}
+}
+
+// blendScale decays a scale factor towards 1 over the first slots.
+// Slot 0 is scaled by the full factor, from slot decaySlots on it is 1.
+func blendScale[T constraints.Float](slots []T, scale float64, decaySlots int) {
+	for i := range min(decaySlots, len(slots)) {
+		w := float64(decaySlots-i) / float64(decaySlots)
+		slots[i] = T(float64(slots[i]) * (w*scale + (1 - w)))
+	}
+}
+
 // prorate adjusts the first slot's energy amount according to remaining duration
 func prorate[T constraints.Float](slots []T, firstSlotDuration time.Duration) []float32 {
 	// return empty slice instead of nil to make api happy

core/site_optimizer_test.go

@@ -260,6 +260,28 @@ func TestOptimizerChargingStrategy(t *testing.T) {
 	assert.Equal(t, "attenuate_grid_peaks", site.GetOptimizerChargingStrategy())
 }
 
+func TestBlendMeasured(t *testing.T) {
+	slots := []float64{100, 100, 100, 100, 100, 100}
+	blendMeasured(slots, 200, 4)
+	assert.Equal(t, []float64{200, 175, 150, 125, 100, 100}, slots)
+
+	// fewer slots than decay length
+	short := []float32{100, 100}
+	blendMeasured(short, 200, 4)
+	assert.Equal(t, []float32{200, 175}, short)
+}
+
+func TestBlendScale(t *testing.T) {
+	slots := []float32{100, 100, 100, 100, 100, 100}
+	blendScale(slots, 2, 4)
+	assert.Equal(t, []float32{200, 175, 150, 125, 100, 100}, slots)
+
+	// fewer slots than decay length
+	short := []float64{100, 100}
+	blendScale(short, 0.5, 4)
+	assert.Equal(t, []float64{50, 62.5}, short)
+}
+
 func TestCurrentSlotSuggestion(t *testing.T) {
 	// slotHours 1 makes the per-slot Wh values map 1:1 to W
 	for _, tc := range []struct {