FraudDetectionHopping.feature

  @hopping
  Scenario Outline: Accumulative transaction during 5 minutes interval: amount above $1500 then notify
    Given Customer has a credit card with account number "<cc>"
    When Customer transacts $<amount1> at "<event1>"
    And Customer transacts $<amount2> at "<event2>"
    And Customer transacts $<amount3> at "<event3>"
    Then Fraud flag is "<flag>"
    And total suspicious amount is $<total>
    Examples:
      |cc|amount1|amount2|amount3|event1|event2|event3|flag|total|
      |4567-8901-2345-6789|100.50|500.50|500.0|2020-12-10T13:00:00Z|2020-12-10T13:01:00Z|2020-12-10T13:05:00Z||0|
      |4567-8901-2345-6789|500.50|500.50|500.0|2020-12-10T13:00:00Z|2020-12-10T13:02:00Z|2020-12-10T13:04:00Z|Y|1501.0|
      |4567-8901-2345-6789|500.50|500.50|500.0|2020-12-10T13:00:00Z|2020-12-10T13:01:00Z|2020-12-10T13:06:00Z||0|

FraudDetectionSession.feature

  @session
  Scenario Outline: Accumulative transaction in a single period of activity with 1 hour inactivity gap: amount above $4000 then notify
    Given Customer has a credit card with account number "<cc>"
    When Customer transacts $<amount1> at "<event1>"
    And Customer transacts $<amount2> at "<event2>"
    And Customer transacts $<amount3> at "<event3>"
    And Customer transacts $<amount4> at "<event4>"
    And Customer transacts $<amount5> at "<event5>"
    Then Fraud flag is "<flag>"
    And total suspicious amount is $<total>
    Examples:
      |cc|amount1|amount2|amount3|amount4|amount5|event1|event2|event3|event4|event5|flag|total|
      |4567-8901-2345-6789|1000|1000|1000|1000|0|2020-12-10T13:00:00Z|2020-12-10T13:10:00Z|2020-12-10T13:30:00Z|2020-12-10T13:40:00Z|2020-12-10T13:50:00Z||0|
      |4567-8901-2345-6789|1000|1000|1000|1000|1000|2020-12-10T13:00:00Z|2020-12-10T13:30:00Z|2020-12-10T13:30:00Z|2020-12-10T13:40:00Z|2020-12-10T13:50:00Z|Y|5000|
      |4567-8901-2345-6789|1000|1000|1000|1000|1000|2020-12-10T13:00:00Z|2020-12-10T13:30:00Z|2020-12-10T13:30:00Z|2020-12-10T13:40:00Z|2020-12-10T14:30:00Z|Y|5000|
      |4567-8901-2345-6789|1000|1000|1000|1000|1000|2020-12-10T13:00:00Z|2020-12-10T14:01:00Z|2020-12-10T14:30:00Z|2020-12-10T14:40:00Z|2020-12-10T14:50:00Z||0|
      # this will set flag to Y because amount to is $1500. it means breaking the first scenario then we still need to informed the suspicious transaction.
      |4567-8901-2345-6789|1000|1500|1000|1000|0|2020-12-10T13:00:00Z|2020-12-10T14:01:00Z|2020-12-10T14:30:00Z|2020-12-10T14:40:00Z|2020-12-10T14:50:00Z|Y|1500|
      # this will set flag to Y because event2 and event3 break hopping windows scenario
      |4567-8901-2345-6789|1000|1000|1000|1000|0|2020-12-10T13:00:00Z|2020-12-10T13:10:00Z|2020-12-10T13:11:00Z|2020-12-10T13:40:00Z|2020-12-10T13:50:00Z|Y|2000|

FraudDetectionSingle.feature

Feature: Fraud Detection

  @single
  Scenario Outline: Single transaction scenario: amount above $1000 then notify
    Given Customer has a credit card with account number "<cc>"
    When Customer transacts $<amount> at "<event>"
    Then Fraud flag is "<flag>"
    And total suspicious amount is $<total>
    Examples:
      |cc|amount|event|flag|total|
      |4567-8901-2345-6789|500.50|2020-12-10T13:50:40Z||0|
      |4567-8901-2345-6789|1000|2020-12-10T13:50:40Z||0|
      |4567-8901-2345-6789|1000.50|2020-12-10T13:50:40Z|Y|1000.50|
      |4567-8901-2345-6789|1500|2020-12-10T13:50:40Z|Y|1500|

FraudDetectionStepDef.java

    // ... code fragment ...
    public FraudDetectionStepDef() {

        Before(scenario -> init());

        Given("^Customer has a credit card with account number \"([^\"]*)\"$", (String cc) -> {
            logger.info("Credit card no {}", cc);
            this.topicKey = cc;
        });

        When("Customer transacts ${double} at {string}", (Double amount, String iso8601str) -> {
            Instant date = Instant.parse(iso8601str);
            logger.info("Transaction with amount {} and event-date {}", amount, date);

            final CreditCardTransactionDto transaction = getTransaction(amount, date);
            transactionTopic.pipeInput(topicKey, transaction, date);
        });

        Then("Fraud flag is {string}", (String flag) -> {
            logger.info("Fraud flag is {}", flag);
            KeyValue<String, CreditCardFraudDetectionDto> kv = null;
            while(!fraudTopic.isEmpty()) {
                kv = fraudTopic.readKeyValue();
            }

            if(kv != null) {
                assertThat(kv.key).isEqualTo(topicKey);
                fraudDetectionDto = kv.value;
                assertThat(fraudDetectionDto.getFraudFlag()).isEqualTo(flag);
            } else {
                assertThat(flag).isNullOrEmpty();
            }
        });

        And("total suspicious amount is ${double}", (Double suspicious) -> {
            logger.info("suspicious amount is {}", suspicious);
            if(fraudDetectionDto != null) {
                final Set<CreditCardTransactionDto> list = fraudDetectionDto.getSuspiciousTransactions();
                final BigDecimal total = list.stream()
                        .map(value -> BigDecimal.valueOf(value.getTrxAmount()))
                        .reduce(BigDecimal.ZERO, BigDecimal::add);
                assertThat(total.doubleValue()).isEqualTo(suspicious);
            } else {
                assertThat(suspicious).isEqualTo(0.0);
            }
        });

        After(scenario -> tear());
    }

FraudDetectionTopologyHopping.java

// full code at FraudDetectionTopology.java

// hopping-windows
final KStream<String, CreditCardFraudDetectionDto> hopping =
        input.groupByKey(Grouped.with(keySerde, creditCardTransactionSerde))
            .windowedBy(TimeWindows.of(Duration.ofMinutes(5)).advanceBy(Duration.ofMinutes(1)))
            .aggregate(() -> CreditCardTransactionAggregationDto.builder().ongoingTransactions(Set.of()).build(),
                    (key, value, aggr) -> {
                        final Set<CreditCardTransactionDto> current = aggr.getOngoingTransactions();
                        Set<CreditCardTransactionDto> set = new HashSet<>(current);
                        set.add(value);
                        return aggr.toBuilder()
                                .ongoingTransactions(set)
                                .build();
                    },
                    Materialized.with(keySerde, MySerdesFactory.creditCardTransactionAggregationSerde()))
            .toStream()
            .filter((key, value) -> hoppingWindowThreshold.compareTo(value.sumOngoingTransactions()) < 0)
            .mapValues(value -> CreditCardFraudDetectionDto.builder()
                    .fraudFlag("Y")
                    .suspiciousTransactions(value.getOngoingTransactions())
                    .build())
            .filter((key, value) -> value != null)
            .map((key,value) -> new KeyValue<>(key.key(), value));

FraudDetectionTopologyJoin.java

// full code at FraudDetectionTopology.java

// suspicious trx always goes into these joins
single.outerJoin(hopping,
        valueJoiner(),
        JoinWindows.of(Duration.ofSeconds(1)),
        StreamJoined.with(keySerde, creditCardFraudDetectionSerde, creditCardFraudDetectionSerde))
    .outerJoin(session,
        valueJoiner(),
        JoinWindows.of(Duration.ofSeconds(1)),
        StreamJoined.with(keySerde, creditCardFraudDetectionSerde, creditCardFraudDetectionSerde))
    .to(CREDIT_CARD_FRAUD_DETECTION_OUTPUT,
            Produced.with(keySerde, creditCardFraudDetectionSerde));

FraudDetectionTopologySession.java

// full code at FraudDetectionTopology.java

// session-windows
final KStream<String, CreditCardFraudDetectionDto> session =
        input.groupByKey(Grouped.with(keySerde, creditCardTransactionSerde))
            .windowedBy(SessionWindows.with(Duration.ofHours(1)))
            .aggregate(() -> CreditCardTransactionAggregationDto.builder().ongoingTransactions(Set.of()).build(),
                    (key, value, aggr) -> {
                        final Set<CreditCardTransactionDto> current = aggr.getOngoingTransactions();
                        Set<CreditCardTransactionDto> set = new HashSet<>(current);
                        set.add(value);
                        return aggr.toBuilder()
                                .ongoingTransactions(set)
                                .build();
                    },
                    (key, aggOne, aggTwo) -> {
                        final Set<CreditCardTransactionDto> ongoing1 = aggOne.getOngoingTransactions();
                        final Set<CreditCardTransactionDto> ongoing2 = aggTwo.getOngoingTransactions();
                        Set<CreditCardTransactionDto> set = new HashSet<>(ongoing1);
                        set.addAll(ongoing2);
                        return aggOne.toBuilder().ongoingTransactions(set).build();
                    },
                    Materialized.with(keySerde, MySerdesFactory.creditCardTransactionAggregationSerde()))
            .toStream()
            .filter((key, value) -> sessionWindowThreshold.compareTo(value.sumOngoingTransactions()) < 0)
            .mapValues(value -> CreditCardFraudDetectionDto.builder()
                    .fraudFlag("Y")
                    .suspiciousTransactions(value.getOngoingTransactions())
                    .build())
            .filter((key, value) -> value != null)
            .map((key,value) -> new KeyValue<>(key.key(), value));

FraudDetectionTopologySingle.java

// full code at FraudDetectionTopology.java
final KStream<String, CreditCardTransactionDto> input =
        builder.stream(CREDIT_CARD_TRANSACTION_INPUT, Consumed.with(keySerde, creditCardTransactionSerde));

// single
final KStream<String, CreditCardFraudDetectionDto> single =
    input.filter((key, value) -> singleThreshold.compareTo(value.getTrxAmount()) < 0)
        .mapValues(value -> CreditCardFraudDetectionDto.builder()
            .fraudFlag("Y")
            .suspiciousTransactions(Set.of(value))
            .build());

pom.xml

        <dependency>
            <groupId>io.cucumber</groupId>
            <artifactId>cucumber-java8</artifactId>
            <version>6.8.1</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>io.cucumber</groupId>
            <artifactId>cucumber-junit</artifactId>
            <version>6.8.1</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.junit.vintage</groupId>
            <artifactId>junit-vintage-engine</artifactId>
            <version>5.6.2</version>
            <scope>test</scope>
        </dependency>