Correlating Anomalies via Temporal Overlap Similarity

OpenSearch anomalies such as service degradation, job delays, and incident bursts are represented as time intervals, not isolated points. If two detectors fire on the same incident, their anomaly intervals will substantially overlap in time (might with a little timestamp jitter due to different interval, detector start time, and causal relationship). Our similarity therefore measures:

* how much the time windows overlap (after a small tolerance δ to account for jitter),
* optionally, whether the duration is consistent.

This PR implements threshold-graph + connected components based on similarity.

Major algorithm:
- De-dupe input anomalies by id (stable insertion order).
- For every pair (i,j):
  - Dilate both time intervals by ±delta to tolerate bucket alignment drift.
  - Require dilated overlap >= minOverlap (cheap early filter).
  - Compute temporal overlap:
    - IoU (Jaccard over time) on dilated intervals
    - Overlap coefficient (overlap / min(lenA,lenB)) for containment cases
  - Detect strong containment (ovl >= tauContain and duration ratio <= rhoMax).
  - Pick temporal term by mode:
    - IOU: use IoU
    - OVL: use overlap coefficient
    - HYBRID: if strong containment, blend ((1-lam)*IoU + lam*OVL); else use IoU
  - Compute duration penalty exp(-|durA-durB|/kappa).
    - If strong containment, relax the penalty via pow(basePen, containmentRelax)
      (or disable penalty entirely when containmentRelax == 0).
  - Similarity = temporalTerm * penalty; add an undirected edge if similarity >= alpha.
- Run DFS connected-components on the threshold graph to form clusters.
- Output deterministically: sort members in each cluster by anomaly id.
- Attach an event window per cluster as [min(start), max(end)] across its members.

Testing done:
1. UT
2. Tests on real world data

Signed-off-by: Kaituo Li <kaituo@amazon.com>

Author: kaituo

CHANGELOG.md

@@ -4,7 +4,11 @@ All notable changes to this project are documented in this file.
 Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.1.0/)
 
 ## [Unreleased 3.x](https://github.com/opensearch-project/anomaly-detection/compare/3.4...HEAD)
+
+
 ### Features
+- Correlating Anomalies via Temporal Overlap Similarity ([#1641](https://github.com/opensearch-project/anomaly-detection/pull/1641))
+
 ### Enhancements
 ### Bug Fixes
 ### Infrastructure

src/main/java/org/opensearch/ad/correlation/Anomaly.java

@@ -0,0 +1,83 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * 
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ * 
+ * Modifications Copyright OpenSearch Contributors. See
+ * GitHub history for details.
+ * 
+ */
+package org.opensearch.ad.correlation;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Objects;
+
+/**
+ * Anomaly class for anomaly correlation.
+ * @param id This uniquely identifies the emitting source of the anomaly (e.g., model id).
+ * @param configId The id of the detector.
+ * @param dataStartTime The start time of the anomaly.
+ * @param dataEndTime The end time of the anomaly.
+ */
+public final class Anomaly {
+    // e.g., model id
+    private final String id;
+    private final String configId;
+    private final Instant dataStartTime;
+    private final Instant dataEndTime;
+
+    public Anomaly(String id, String configId, Instant dataStartTime, Instant dataEndTime) {
+        this.id = Objects.requireNonNull(id, "id");
+        this.configId = Objects.requireNonNull(configId, "configId");
+        this.dataStartTime = Objects.requireNonNull(dataStartTime, "dataStartTime");
+        this.dataEndTime = Objects.requireNonNull(dataEndTime, "dataEndTime");
+
+        if (!dataEndTime.isAfter(dataStartTime)) {
+            throw new IllegalArgumentException("dataEndTime must be after dataStartTime");
+        }
+    }
+
+    public String getId() {
+        return id;
+    }
+
+    public Instant getDataStartTime() {
+        return dataStartTime;
+    }
+
+    public Instant getDataEndTime() {
+        return dataEndTime;
+    }
+
+    public Duration getDuration() {
+        return Duration.between(dataStartTime, dataEndTime);
+    }
+
+    public Instant getMidpoint() {
+        long halfNanos = getDuration().toNanos() / 2;
+        return dataStartTime.plusNanos(halfNanos);
+    }
+
+    public String getConfigId() {
+        return configId;
+    }
+
+    @Override
+    public String toString() {
+        return "Anomaly{"
+            + "id='"
+            + id
+            + '\''
+            + ", detectorName='"
+            + configId
+            + '\''
+            + ", dataStartTime="
+            + dataStartTime
+            + ", dataEndTime="
+            + dataEndTime
+            + '}';
+    }
+}