Sliding window analytical use case

Prerequisites

Revisit the use case building framework to get an understanding on how to create a use case out of business context.

Revisit low code development to get an understanding on what is the anatomy of a use case.

Overview

This use case builds upon the first use case with the additional

These instructions should only be used to configure the Joule Platform for development or demonstration purposes.

Use case definition

Filtered stream of Nasdaq major bank quotes

Overview of use case processing sequence

1

Use case objective

Provide trading consumer application for the current business day with bid and ask quotes and company information for all major banks with a market cap of over $350 billion trading on the nasdaq stock market and when the spread widens to over 1.5%.

The use case should only be processing for a single defined market business day.

use case:
  name: nasdaq_banking_quotes
  constraints:
    valid from: '2024-10-01T09:25:00.000Z'
    valid to: '2024-10-01T16:35:00.000Z'
  sources:
    - live_nasdaq_quotes
  stream name: basic_sliding_window_analytics
  sinks:
    - nasdaq_major_bank_topic
    - nasdaq_major_bank_quotes_file

Change the the valid from and to the current dates

2

Processing steps

1. Enrich events with industry and market cap context information

2. Filter events by 'Major Banks' industry and with market cap greater than $350 billion

3. Send a stock record with following attributes for every event; symbol, company_name, market_cap, bid, ask

Stream definition

stream:
  name: basic_sliding_window_analytics
  eventTimeType: EVENT_TIME
  
  initialisation:
  # Import contextual company data in to Joule
  data import:
    schema: reference_data
    csv:
    - table: nasdaq_companies
      file: data/csv/nasdaq.csv
      drop table: true
      index:
        fields:
        - symbol
        unique: true
          
  processing unit:
    pipeline:
    # Filter events by major banks to reduce number of enrichment queries
    - filter:
        expression: "(typeof industry !== 'undefined' && 
                      industry == 'Major Banks')"
                    
    # Enrich filtered event with company information
    - enricher:
        fields:
          company_info:
            by query: 
               select * from reference_data.nasdaq_companies where symbol = ?
            query fields:
            - symbol
            with values:
            - company_name
            - market_cap
            using: JouleDB
    
    #
    # Sliding window basic analytic functions
    #    
    - time window:
        emitting type: quoteAggregates
        aggregate functions:
          MIN:
          - ask
          MAX:
          - bid
          SUM:
          - volume
          MEAN:
          - volatility
        policy:
          type: slidingTime
          slide: 500
          window size: 2500
          
  emit:
    select: symbol, company_name, market_cap, ask_MIN, bid_MAX, volume_SUM, volatility_MEAN
    having: "((bid_MAX - ask_MIN) / bid_MAX) > 0.015"
  group by:
  - symbol

3

Data sources

Subscribe to live nasdaq quote data (note we are using simulated data)

Source definition

kafkaConsumer:
    name: nasdaq_quotes_stream
    cluster address: joule-gs-redpanda-0:9092
    consumerGroupId: nasdaq
    topics:
      - quotes

    deserializer:
      parser: com.fractalworks.examples.banking.data.QuoteToStreamEventParser
      key deserializer: org.apache.kafka.common.serialization.IntegerDeserializer
      value deserializer: com.fractalworks.streams.transport.kafka.serializers.object.ObjectDeserializer

    properties:
      partition.assignment.strategy: org.apache.kafka.clients.consumer.StickyAssignor
      max.poll.records" : 7000
      fetch.max.bytes : 1048576

4

Validate output

A quick and easy way to validate your use case processing is to send the resulting events to a CSV file.

The use case will output both to a Kafka topic and CSV file using the current configuration.

file:
  name: nasdaq_major_bank_quotes_file
  filename: nasdaq_major_banks
  path: "./data/output/test_output"
  batchSize: 1024
  timeout: 1000
  formatter:
    csv formatter:
      contentType: text/csv
      encoding: UTF-8
      delimiter: "|"

5

Publish events to consumers

Once you are satisfied with the results its time to switch to publishing events on to the Kafka consumer topic.

A quick recap of how events will be transformed to AVRO data structures:

1. The user emit projection is rransformed to provided domain data type using an AVRO schema, see below.

2. The resulting events are then published on to the nasdaq_major_bank_quotes Kafka topic.

AVRO schema

6

{
  "type" : "record",
  "name" : "StockRecord",
  "namespace" : "com.fractalworks.examples.banking.data",
  "fields" : [
    {"name" : "symbol", "type" : "string"},
    {"name" : "company_name", "type" : "string"},
    {"name" : "market_cap", "type" : "long"},
    {"name" : "bid", "type" : "double"},
    {"name" : "ask", "type" : "double"}
    ]
}

Sink definition

kafkaPublisher:
  name: nasdaq_major_bank_topic
  cluster address: joule-gs-redpanda-0:9092
  topic: nasdaq_major_bank_quotes
  partitionKeys:
    - symbol

  serializer:
    key serializer: org.apache.kafka.common.serialization.IntegerSerializer
    avro setting:
      local schema: ./conf/avro/stockrecord.avsc

7

Deploying the use case

Now we have the use case definition we can now deployable the definitions to Joule. Following the same getting started deployment steps for this project.

Go to the "Build your first use case" folder under the Joule - Banking demo / Tutorials Postman examples within the getting started project

8

Reviewing results

First take a look at the generated CSV file by getting the first six lines of the file:

head -6 data/output/test_output/nasdaq_major_banks.csv

The command should return output similar to the below:

event_type|sub_type|event_time|ingest_time|ask|symbol|market_cap|bid|company_name
nasdaq_view|null|1732727040721|1732727040721|7.972548147055079|UBS|5.043039879E10|8.93326025899351|UBS Group AG Registered Ordinary Shares
nasdaq_view|null|1732727040722|1732727040722|2.1130250530209174|EBC|3.695943868E9|2.6803928208677|Eastern Bankshares Inc. Common Stock
nasdaq_view|null|1732727040724|1732727040724|2.0621971972218733|FNB|3.874144198E9|2.420388724487513|F.N.B. Corporation Common Stock
nasdaq_view|null|1732727040896|1732727040896|8.332975002835752|UBS|5.043039879E10|8.632278213149188|UBS Group AG Registered Ordinary Shares
nasdaq_view|null|1732727040777|1732727040777|8.378182322639656|UBS|5.043039879E10|8.58597984478865|UBS Group AG Registered Ordinary Shares

What we have learnt

As a first use case we have covered a number of key features:

• Subscribe and consume events Subscribe, consume, parse and present events ready for pipeline processing using Kafka.

• Initialise Joule with contextual data Load local CSV contextual data in to the JouleDB

• Filters and enrichment Apply filter for a subset of events using Javascript expressions and apply event enrichment with company data loaded in to JouleDB

• Results filter expression Using the "having" clause with a Javascript expression to only send events based upon a spread ratio breach

• Publishing events Send processed events to either a CSV file or on to a Kafka topic as a defined AVRO domain data structure