Fix anomaly detectors

anomaly-detector/src/main/java/com/anomaly/detector/AnomalyDetector.java

@@ -21,10 +21,16 @@ import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeW
 import org.apache.flink.streaming.api.windowing.time.Time;
 import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
 import org.apache.flink.util.Collector;
+import org.apache.flink.api.common.state.MapState;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.api.common.typeinfo.TypeHint;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.Configuration;
 
 import java.util.*;
 import java.time.Instant;
 import java.io.IOException;
+import java.io.Serializable;
 
 public class AnomalyDetector {
 
@@ -81,13 +87,13 @@ public class AnomalyDetector {
         // 1. Amount anomaly - sudden high-value transactions
         DataStream<TransactionAlert> amountAlerts = transactionStream
                 .keyBy(Transaction::getCardId)
-                .window(SlidingProcessingTimeWindows.of(Time.minutes(10), Time.minutes(1)))
+                .window(SlidingProcessingTimeWindows.of(Time.minutes(5), Time.minutes(1)))
                 .process(new AmountAnomalyDetector());
 
         // 2. Location anomaly - sudden change in location
         DataStream<TransactionAlert> locationAlerts = transactionStream
                 .keyBy(Transaction::getCardId)
-                .window(SlidingProcessingTimeWindows.of(Time.minutes(10), Time.minutes(1)))
+                .window(SlidingProcessingTimeWindows.of(Time.minutes(5), Time.minutes(1)))
                 .process(new LocationAnomalyDetector());
 
         // 3. Frequency anomaly - unusual number of transactions in short time
@@ -139,9 +145,9 @@ public class AnomalyDetector {
 
             double stdDeviation = calculateStdDeviation(transactionList, averageAmount);
 
-            // Check for anomalies (transactions that are more than 3 standard deviations from mean)
+            // Check for anomalies (transactions that are more than 1.7 standard deviations from mean)
             for (Transaction transaction : transactionList) {
-                if (stdDeviation > 0 && Math.abs(transaction.getAmount() - averageAmount) > 3 * stdDeviation) {
+                if (stdDeviation > 0 && Math.abs(transaction.getAmount() - averageAmount) > 2 * stdDeviation && transaction.getAmount() > averageAmount && transaction.getAmount() > 1000) {
                     out.collect(new TransactionAlert(
                             "AMOUNT_ANOMALY",
                             transaction.getCardId(),
@@ -169,57 +175,122 @@ public class AnomalyDetector {
     public static class LocationAnomalyDetector
             extends ProcessWindowFunction<Transaction, TransactionAlert, String, TimeWindow> {
 
-        // Map to store frequent locations for each card
+        private transient MapState<String, Set<LocationPoint>> knownLocations;
-        private final Map<String, Set<LocationPoint>> cardLocations = new HashMap<>();
+        private static final int MAX_KNOWN_LOCATIONS = 5; // Limit known locations to avoid memory issues
+        private static final double ANOMALY_DISTANCE_THRESHOLD = 50.0; // Threshold in km
+        private static final int MIN_LOCATIONS_FOR_DETECTION = 3; // Minimum known locations before detecting anomalies
+
+        @Override
+        public void open(Configuration parameters) throws Exception {
+            MapStateDescriptor<String, Set<LocationPoint>> descriptor =
+                new MapStateDescriptor<>(
+                    "knownLocations",
+                    TypeInformation.of(String.class),
+                    TypeInformation.of(new TypeHint<Set<LocationPoint>>() {})
+                );
+            knownLocations = getRuntimeContext().getMapState(descriptor);
+        }
 
         @Override
         public void process(String cardId, Context context, Iterable<Transaction> transactions,
-                           Collector<TransactionAlert> out) {
+                           Collector<TransactionAlert> out) throws Exception {
             List<Transaction> transactionList = new ArrayList<>();
             transactions.forEach(transactionList::add);
 
             if (transactionList.isEmpty()) return;
 
             // Get or create location set for this card
-            Set<LocationPoint> frequentLocations = cardLocations.computeIfAbsent(cardId, k -> new HashSet<>());
+            Set<LocationPoint> cardKnownLocations;
+            if (knownLocations.contains(cardId)) {
+                cardKnownLocations = knownLocations.get(cardId);
+                System.out.println("Card " + cardId + " has " + cardKnownLocations.size() + " known locations");
+            } else {
+                cardKnownLocations = new HashSet<>();
+                System.out.println("New card detected: " + cardId + ", initializing known locations");
+            }
 
             // Process each transaction
             for (Transaction transaction : transactionList) {
                 LocationPoint currentPoint = new LocationPoint(transaction.getLatitude(), transaction.getLongitude());
 
-                // If we have at least 3 frequent locations for this card
+                // First few transactions establish the baseline locations
-                if (frequentLocations.size() >= 3) {
+                if (cardKnownLocations.size() < MIN_LOCATIONS_FOR_DETECTION) {
-                    boolean isNearKnownLocation = false;
+                    System.out.println("Building baseline for card " + cardId + ", adding location #" +
+                                      (cardKnownLocations.size() + 1) + " to known locations");
 
-                    // Check if current location is near any known frequent location
+                    // Check if this location is already very close to a known location before adding
-                    for (LocationPoint knownPoint : frequentLocations) {
+                    boolean isVeryCloseToKnown = false;
-                        if (calculateDistance(currentPoint, knownPoint) < 50) { // Less than 50km
+                    for (LocationPoint knownPoint : cardKnownLocations) {
-                            isNearKnownLocation = true;
+                        if (calculateDistance(currentPoint, knownPoint) < 2.0) { // Within 2km = same area
+                            isVeryCloseToKnown = true;
+                            System.out.println("Location is very close to existing baseline location, not adding duplicate");
                             break;
                         }
                     }
 
-                    // If not near any known location, it might be an anomaly
+                    // Only add distinct baseline locations
-                    if (!isNearKnownLocation) {
+                    if (!isVeryCloseToKnown) {
-                        out.collect(new TransactionAlert(
+                        cardKnownLocations.add(currentPoint);
-                                "LOCATION_ANOMALY",
+                    }
-                                transaction.getCardId(),
+
-                                transaction.getUserId(),
+                    // We're still building the baseline, don't check for anomalies yet
-                                transaction.getAmount(),
+                    continue;
-                                transaction.getLatitude(),
+                }
-                                transaction.getLongitude(),
+
-                                transaction.getTimestamp(),
+                // Check distance to known locations
-                                "Unusual transaction location detected at: " + 
+                double closestDistance = Double.MAX_VALUE;
-                                transaction.getLatitude() + ", " + transaction.getLongitude()
+                LocationPoint closestPoint = null;
-                        ));
+
+                for (LocationPoint knownPoint : cardKnownLocations) {
+                    double distance = calculateDistance(currentPoint, knownPoint);
+                    if (distance < closestDistance) {
+                        closestDistance = distance;
+                        closestPoint = knownPoint;
                     }
                 }
 
-                // Add current location to frequent locations (max 10 locations per card)
+                System.out.println("CARD " + cardId + ": Transaction at " + currentPoint + ", closest known location: " +
-                if (frequentLocations.size() < 10) {
+                                  closestPoint + " (" + String.format("%.2f", closestDistance) + " km)");
-                    frequentLocations.add(currentPoint);
+
+                // Detect anomaly if transaction is far from all known locations
+                if (closestDistance > ANOMALY_DISTANCE_THRESHOLD) {
+                    System.out.println("⚠️ LOCATION ANOMALY DETECTED: Distance " +
+                                      String.format("%.2f", closestDistance) + "km exceeds threshold of " +
+                                      ANOMALY_DISTANCE_THRESHOLD + "km");
+
+                    out.collect(new TransactionAlert(
+                            "LOCATION_ANOMALY",
+                            transaction.getCardId(),
+                            transaction.getUserId(),
+                            transaction.getAmount(),
+                            transaction.getLatitude(),
+                            transaction.getLongitude(),
+                            transaction.getTimestamp(),
+                            "Unusual transaction location: " + String.format("%.2f", closestDistance) +
+                            "km from nearest known location"
+                    ));
+
+                    // Don't automatically add anomalous locations to known locations
+                } else {
+                    // Check if this location is already very close to a known location
+                    boolean isVeryCloseToKnown = false;
+                    for (LocationPoint knownPoint : cardKnownLocations) {
+                        if (calculateDistance(currentPoint, knownPoint) < 2.0) { // Within 2km = same area
+                            isVeryCloseToKnown = true;
+                            break;
+                        }
+                    }
+
+                    // Only add distinct new locations, up to our maximum
+                    if (!isVeryCloseToKnown && cardKnownLocations.size() < MAX_KNOWN_LOCATIONS) {
+                        cardKnownLocations.add(currentPoint);
+                        System.out.println("Added new location to known locations: " + currentPoint);
+                    }
                 }
             }
+
+            // Update the state
+            knownLocations.put(cardId, cardKnownLocations);
         }
 
         // Calculate distance between two points using Haversine formula (in km)
@@ -238,7 +309,8 @@ public class AnomalyDetector {
             return R * c;
         }
 
-        private static class LocationPoint {
+        private static class LocationPoint implements Serializable {
+            private static final long serialVersionUID = 1L;
             private final double latitude;
             private final double longitude;
 
@@ -260,6 +332,14 @@ public class AnomalyDetector {
             public int hashCode() {
                 return Objects.hash(latitude, longitude);
             }
+
+            @Override
+            public String toString() {
+                return "LocationPoint{" +
+                        "lat=" + latitude +
+                        ", lon=" + longitude +
+                        '}';
+            }
         }
     }