fix: mpc retry logic

entsoe/api_client.go

@@ -57,8 +57,8 @@ func DownloadPublicationMarketData(ctx context.Context, securityToken string, ur
 		return nil, err
 	}
 
-	// If current time is >= 13:00, also download data for the next day
-	if now.Hour() >= 13 {
+	// If current time is >= 14:00, also download data for the next day
+	if now.Hour() >= 14 {
 		tomorrow := now.AddDate(0, 0, 1)
 		urlNextDay := buildPublicationMarketDataURL(securityToken, urlFormat, tomorrow)
 

scheduler/mpc.go

@@ -72,12 +72,20 @@ func (s *MinerScheduler) RunMPCOptimize(ctx context.Context) {
 		return
 	}
 
-	// Step 5: Save optimization results
+	// Step 5: Save optimization results to memory
 	s.mu.Lock()
 	s.mpcDecisions = decisions
-	s.lastMPCExecutionError = nil // Clear any previous execution error
+	s.lastExecutedDecision = nil // Clear last executed decision for new optimization
 	s.mu.Unlock()
 
+	// Step 5.1: Persist decisions to database (only when not in dry run mode)
+	if !config.DryRun {
+		if err := s.saveMPCDecisions(ctx, decisions); err != nil {
+			s.logger.Printf("Warning: Failed to save MPC decisions to database: %v", err)
+			// Continue execution even if persistence fails
+		}
+	}
+
 	// Log summary
 	s.logger.Printf("MPC optimization completed with %d decisions", len(decisions))
 	totalProfit := 0.0
@@ -91,7 +99,13 @@ func (s *MinerScheduler) RunMPCOptimize(ctx context.Context) {
 
 	// Record execution status
 	s.mu.Lock()
-	s.lastMPCExecutionError = err
+	if err != nil {
+		// Execution failed, set lastExecutedDecision to nil
+		s.lastExecutedDecision = nil
+	} else {
+		// Execution succeeded, store the executed decision
+		s.lastExecutedDecision = &decisions[0]
+	}
 	s.mu.Unlock()
 
 	if err != nil {
@@ -524,15 +538,7 @@ func (s *MinerScheduler) executeMPCDecision(decision *mpc.ControlDecision, dryRu
 // This ensures the decision is applied even if previous execution failed
 func (s *MinerScheduler) runMPCExecution() {
 
-	// Check if last execution failed
 	s.mu.RLock()
-	lastError := s.lastMPCExecutionError
-	if lastError == nil {
-		// Last execution succeeded, no need to retry
-		s.mu.RUnlock()
-		return
-	}
-
 	config := s.GetConfig()
 
 	// Check if Plant Modbus Address is configured
@@ -541,27 +547,58 @@ func (s *MinerScheduler) runMPCExecution() {
 		return
 	}
 
-	// Get current MPC decision
+	// Get current MPC decision based on current timestamp
 	if len(s.mpcDecisions) == 0 {
 		s.mu.RUnlock()
 		return
 	}
-	currentDecision := s.mpcDecisions[0]
+
+	now := time.Now().Unix()
+	var currentDecision *mpc.ControlDecision
+
+	// Find the decision that matches the current hour
+	for i := range s.mpcDecisions {
+		decision := &s.mpcDecisions[i]
+		// Check if current time falls within this decision's hour
+		// Each decision covers a 1-hour window starting from its timestamp
+		if now >= decision.Timestamp && now < decision.Timestamp+3600 {
+			currentDecision = decision
+			break
+		}
+	}
+
+	if currentDecision == nil {
+		// No matching decision found for current timestamp
+		s.mu.RUnlock()
+		return
+	}
+
+	lastExecuted := s.lastExecutedDecision
 	s.mu.RUnlock()
 
-	s.logger.Printf("Retrying MPC decision execution (previous error: %v)", lastError)
+	// Check if this decision has already been executed
+	if lastExecuted != nil && currentDecision.Timestamp == lastExecuted.Timestamp {
+		// Decision already executed, no need to retry
+		return
+	}
+
+	s.logger.Printf("Executing MPC decision for timestamp %d (hour %d)", currentDecision.Timestamp, currentDecision.Hour)
 
 	// Execute the current decision
-	err := s.executeMPCDecision(&currentDecision, config.DryRun)
+	err := s.executeMPCDecision(currentDecision, config.DryRun)
 
 	s.mu.Lock()
-	s.lastMPCExecutionError = err
-	s.mu.Unlock()
-
 	if err != nil {
-		s.logger.Printf("Error retrying MPC decision: %v (will retry again in 1 minute)", err)
+		// Execution failed, set lastExecutedDecision to nil
+		s.lastExecutedDecision = nil
+		s.mu.Unlock()
+		s.logger.Printf("Error executing MPC decision: %v (will retry again in 1 minute)", err)
 		return
 	}
 
-	s.logger.Printf("Successfully retried MPC decision execution")
+	// Execution succeeded, store the executed decision
+	s.lastExecutedDecision = currentDecision
+	s.mu.Unlock()
+
+	s.logger.Printf("Successfully executed MPC decision")
 }

scheduler/mpc_persistence.go

@@ -0,0 +1,209 @@
+package scheduler
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"time"
+
+	"github.com/devskill-org/ems/mpc"
+)
+
+// saveMPCDecisions persists MPC decisions to the database
+func (s *MinerScheduler) saveMPCDecisions(ctx context.Context, decisions []mpc.ControlDecision) error {
+	if s.db == nil {
+		return fmt.Errorf("database connection not available")
+	}
+
+	if len(decisions) == 0 {
+		return nil
+	}
+
+	// Use first decision timestamp as minimum
+	// Decisions are ordered by timestamp because:
+	// 1. MPC forecast is built from a map and explicitly sorted by hour
+	// 2. MPC controller reconstructs path in the same order as forecast
+	// 3. Timestamp increases monotonically with hour
+	minTimestamp := decisions[0].Timestamp
+
+	// Begin transaction
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return fmt.Errorf("failed to begin transaction: %w", err)
+	}
+	defer tx.Rollback()
+
+	// Delete existing decisions with timestamp >= minTimestamp
+	_, err = tx.ExecContext(ctx, `DELETE FROM mpc_decisions WHERE timestamp >= $1`, minTimestamp)
+	if err != nil {
+		return fmt.Errorf("failed to delete existing decisions: %w", err)
+	}
+
+	// Prepare upsert statement
+	stmt, err := tx.PrepareContext(ctx, `
+		INSERT INTO mpc_decisions (
+			timestamp,
+			hour,
+			battery_charge,
+			battery_discharge,
+			grid_import,
+			grid_export,
+			battery_soc,
+			profit,
+			import_price,
+			export_price,
+			solar_forecast,
+			load_forecast,
+			cloud_coverage,
+			weather_symbol
+		) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
+		ON CONFLICT (timestamp) DO UPDATE SET
+			hour = EXCLUDED.hour,
+			battery_charge = EXCLUDED.battery_charge,
+			battery_discharge = EXCLUDED.battery_discharge,
+			grid_import = EXCLUDED.grid_import,
+			grid_export = EXCLUDED.grid_export,
+			battery_soc = EXCLUDED.battery_soc,
+			profit = EXCLUDED.profit,
+			import_price = EXCLUDED.import_price,
+			export_price = EXCLUDED.export_price,
+			solar_forecast = EXCLUDED.solar_forecast,
+			load_forecast = EXCLUDED.load_forecast,
+			cloud_coverage = EXCLUDED.cloud_coverage,
+			weather_symbol = EXCLUDED.weather_symbol
+	`)
+	if err != nil {
+		return fmt.Errorf("failed to prepare statement: %w", err)
+	}
+	defer stmt.Close()
+
+	// Insert all decisions
+	for _, decision := range decisions {
+		_, err := stmt.ExecContext(ctx,
+			decision.Timestamp,
+			decision.Hour,
+			decision.BatteryCharge,
+			decision.BatteryDischarge,
+			decision.GridImport,
+			decision.GridExport,
+			decision.BatterySOC,
+			decision.Profit,
+			decision.ImportPrice,
+			decision.ExportPrice,
+			decision.SolarForecast,
+			decision.LoadForecast,
+			decision.CloudCoverage,
+			decision.WeatherSymbol,
+		)
+		if err != nil {
+			return fmt.Errorf("failed to insert decision for hour %d: %w", decision.Hour, err)
+		}
+	}
+
+	// Commit transaction
+	if err := tx.Commit(); err != nil {
+		return fmt.Errorf("failed to commit transaction: %w", err)
+	}
+
+	s.logger.Printf("Saved %d MPC decisions to database", len(decisions))
+	return nil
+}
+
+// loadLatestMPCDecisions loads MPC decisions from the database with timestamp >= now
+func (s *MinerScheduler) loadLatestMPCDecisions(ctx context.Context) ([]mpc.ControlDecision, error) {
+	if s.db == nil {
+		return nil, fmt.Errorf("database connection not available")
+	}
+
+	// Get current Unix timestamp
+	now := ctx.Value("now")
+	var nowTimestamp int64
+	if now != nil {
+		if t, ok := now.(int64); ok {
+			nowTimestamp = t
+		}
+	}
+	if nowTimestamp == 0 {
+		nowTimestamp = s.getCurrentTimestamp()
+	}
+
+	// Load decisions with timestamp >= now, ordered by timestamp
+	rows, err := s.db.QueryContext(ctx, `
+		SELECT
+			timestamp,
+			hour,
+			battery_charge,
+			battery_discharge,
+			grid_import,
+			grid_export,
+			battery_soc,
+			profit,
+			import_price,
+			export_price,
+			solar_forecast,
+			load_forecast,
+			cloud_coverage,
+			weather_symbol
+		FROM mpc_decisions
+		WHERE timestamp >= $1
+		ORDER BY timestamp ASC
+	`, nowTimestamp)
+	if err != nil {
+		return nil, fmt.Errorf("failed to query decisions: %w", err)
+	}
+	defer rows.Close()
+
+	var decisions []mpc.ControlDecision
+	for rows.Next() {
+		var decision mpc.ControlDecision
+		var cloudCoverage sql.NullFloat64
+		var weatherSymbol sql.NullString
+
+		err := rows.Scan(
+			&decision.Timestamp,
+			&decision.Hour,
+			&decision.BatteryCharge,
+			&decision.BatteryDischarge,
+			&decision.GridImport,
+			&decision.GridExport,
+			&decision.BatterySOC,
+			&decision.Profit,
+			&decision.ImportPrice,
+			&decision.ExportPrice,
+			&decision.SolarForecast,
+			&decision.LoadForecast,
+			&cloudCoverage,
+			&weatherSymbol,
+		)
+		if err != nil {
+			return nil, fmt.Errorf("failed to scan decision: %w", err)
+		}
+
+		if cloudCoverage.Valid {
+			decision.CloudCoverage = cloudCoverage.Float64
+		}
+		if weatherSymbol.Valid {
+			decision.WeatherSymbol = weatherSymbol.String
+		}
+
+		decisions = append(decisions, decision)
+	}
+
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating decisions: %w", err)
+	}
+
+	if len(decisions) == 0 {
+		s.logger.Printf("No future MPC decisions found in database")
+		return nil, nil
+	}
+
+	s.logger.Printf("Loaded %d MPC decisions from database (starting from timestamp %d)", len(decisions), nowTimestamp)
+
+	return decisions, nil
+}
+
+// getCurrentTimestamp returns the current Unix timestamp
+func (s *MinerScheduler) getCurrentTimestamp() int64 {
+	return time.Now().Unix()
+}