最大值最小值

src/main/java/com/dashboard/aws/lambda/service/MeasureAverageService.java

@@ -48,47 +48,58 @@ public class MeasureAverageService {
 
             // 3: 查询上一小时的聚合结果（用于回退）
             LocalDateTime prevHour = start.minusHours(1);
-            Map<String, Double> prevHourData = queryPrevHourAggregates(conn, schema, targetTable, prevHour);
+            Map<String, AggregateResult> prevHourData = queryPrevHourAggregates(conn, schema, targetTable, prevHour);
 
-            // 4: 计算当前小时平均值（如果有）
-            Map<String, Double> resultMap = new HashMap<>();
+            // 4: 计算当前小时平均、最大、最小值
+            Map<String, AggregateResult> resultMap = new HashMap<>();
             for (Map.Entry<String, List<Record>> entry : currentHourData.entrySet()) {
                 String deviceId = entry.getKey();
-                double avg = computeTimeWeightedAverage(entry.getValue(), startTs, endTs);
-                resultMap.put(deviceId, avg);
+                List<Record> records = entry.getValue();
+
+                double avg = computeTimeWeightedAverage(records, startTs, endTs);
+                double max = computeMaxValue(records);
+                double min = computeMinValue(records);
+
+                resultMap.put(deviceId, new AggregateResult(avg, max, min));
             }
 
             // 5: 对比集合 A，确保所有设备都有值（否则用上一小时或 0）
             for (String deviceId : deviceSet) {
                 if (!resultMap.containsKey(deviceId)) {
-                    Double prevVal = prevHourData.get(deviceId);
-                    double value = (prevVal != null) ? prevVal : 0.0;
-                    resultMap.put(deviceId, value);
+                    AggregateResult prev = prevHourData.get(deviceId);
+                    if (prev != null) {
+                        resultMap.put(deviceId, prev);
+                    } else {
+                        resultMap.put(deviceId, new AggregateResult(0.0, 0.0, 0.0));
+                    }
                 }
             }
 
             // 6: 写入本小时聚合表
             String insertSql = String.format(
-                    "INSERT INTO %s.%s (device_id, average_value, date_year, date_month, date_day, date_hour, " +
-                            "time_start, time_end, aggregated_at) " +
-                            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", schema, targetTable
+                    "INSERT INTO %s.%s (device_id, average_value, max_value, min_value, " +
+                            "date_year, date_month, date_day, date_hour, time_start, time_end, aggregated_at) " +
+                            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                    schema, targetTable
             );
 
             try (PreparedStatement ps = conn.prepareStatement(insertSql)) {
-                for (Map.Entry<String, Double> entry : resultMap.entrySet()) {
+                for (Map.Entry<String, AggregateResult> entry : resultMap.entrySet()) {
                     String deviceId = entry.getKey();
-                    double avg = entry.getValue();
+                    AggregateResult r = entry.getValue();
                     LocalDateTime dt = start;
 
                     ps.setString(1, deviceId);
-                    ps.setString(2, String.format("%.6f", avg));
-                    ps.setInt(3, dt.getYear());
-                    ps.setInt(4, dt.getMonthValue());
-                    ps.setInt(5, dt.getDayOfMonth());
-                    ps.setInt(6, dt.getHour());
-                    ps.setLong(7, startTs);
-                    ps.setLong(8, endTs);
-                    ps.setTimestamp(9, Timestamp.valueOf(now));
+                    ps.setString(2, String.format("%.6f", r.avg));
+                    ps.setString(3, String.format("%.6f", r.max));
+                    ps.setString(4, String.format("%.6f", r.min));
+                    ps.setInt(5, dt.getYear());
+                    ps.setInt(6, dt.getMonthValue());
+                    ps.setInt(7, dt.getDayOfMonth());
+                    ps.setInt(8, dt.getHour());
+                    ps.setLong(9, startTs);
+                    ps.setLong(10, endTs);
+                    ps.setTimestamp(11, Timestamp.valueOf(now));
 
                     ps.addBatch();
                 }
@@ -126,10 +137,10 @@ public class MeasureAverageService {
     }
 
     // 查询上一小时聚合结果
-    private Map<String, Double> queryPrevHourAggregates(Connection conn, String schema, String table, LocalDateTime prevHour) throws SQLException {
-        Map<String, Double> map = new HashMap<>();
+    private Map<String, AggregateResult> queryPrevHourAggregates(Connection conn, String schema, String table, LocalDateTime prevHour) throws SQLException {
+        Map<String, AggregateResult> map = new HashMap<>();
         String sql = String.format(
-                "SELECT device_id, average_value FROM %s.%s " +
+                "SELECT device_id, average_value, max_value, min_value FROM %s.%s " +
                         "WHERE date_year = ? AND date_month = ? AND date_day = ? AND date_hour = ?",
                 schema, table
         );
@@ -140,12 +151,17 @@ public class MeasureAverageService {
             ps.setInt(4, prevHour.getHour());
             ResultSet rs = ps.executeQuery();
             while (rs.next()) {
-                map.put(rs.getString("device_id"), rs.getDouble("average_value"));
+                map.put(rs.getString("device_id"),
+                        new AggregateResult(
+                                rs.getDouble("average_value"),
+                                rs.getDouble("max_value"),
+                                rs.getDouble("min_value")));
             }
         }
         return map;
     }
 
+    // 时间加权平均
     private double computeTimeWeightedAverage(List<Record> records, long startTs, long endTs) {
         if (records == null || records.isEmpty()) return 0.0;
 
@@ -156,15 +172,22 @@ public class MeasureAverageService {
             Record rec = records.get(i);
             long nextTs = (i < records.size() - 1) ? records.get(i + 1).timestamp : endTs;
             double intervalMinutes = (nextTs - rec.timestamp) / 60000.0; // 转分钟
-            logger.info("intervalMinutes: {}, rec.value: {}", intervalMinutes, rec.value);
             sum += rec.value * intervalMinutes;
         }
 
-        logger.info("totalMinutes: {}, sum: {}", totalMinutes, sum);
-
         return totalMinutes > 0 ? sum / totalMinutes : 0.0;
     }
 
+    // 最大值
+    private double computeMaxValue(List<Record> records) {
+        return records.stream().mapToDouble(r -> r.value).max().orElse(0.0);
+    }
+
+    // 最小值
+    private double computeMinValue(List<Record> records) {
+        return records.stream().mapToDouble(r -> r.value).min().orElse(0.0);
+    }
+
     private static class Record {
         long timestamp;
         double value;
@@ -174,4 +197,16 @@ public class MeasureAverageService {
             this.value = value;
         }
     }
-}
+
+    private static class AggregateResult {
+        double avg;
+        double max;
+        double min;
+
+        AggregateResult(double avg, double max, double min) {
+            this.avg = avg;
+            this.max = max;
+            this.min = min;
+        }
+    }
+}