Publish and enrich real-time monetary knowledge feeds utilizing Amazon MSK and Amazon Managed Service for Apache Flink

Monetary knowledge feeds are real-time streams of inventory quotes, commodity costs, choices trades, or different real-time monetary knowledge. Corporations concerned with capital markets comparable to hedge funds, funding banks, and brokerages use these feeds to tell funding selections.

Monetary knowledge feed suppliers are more and more being requested by their clients to ship the feed on to them via the AWS Cloud. That’s as a result of their clients have already got infrastructure on AWS to retailer and course of the information and wish to devour it with minimal effort and latency. As well as, the AWS Cloud’s cost-effectiveness allows even small and mid-size corporations to develop into monetary knowledge suppliers. They’ll ship and monetize knowledge feeds that they’ve enriched with their very own useful info.

An enriched knowledge feed can mix knowledge from a number of sources, together with monetary information feeds, so as to add info comparable to inventory splits, company mergers, quantity alerts, and transferring common crossovers to a fundamental feed.

On this put up, we display how one can publish an enriched real-time knowledge feed on AWS utilizing Amazon Managed Streaming for Kafka (Amazon MSK) and Amazon Managed Service for Apache Flink. You’ll be able to apply this structure sample to varied use instances inside the capital markets trade; we focus on a few of these use instances on this put up.

Apache Kafka is a high-throughput, low-latency distributed occasion streaming platform. Monetary exchanges comparable to Nasdaq and NYSE are more and more turning to Kafka to ship their knowledge feeds due to its distinctive capabilities in dealing with high-volume, high-velocity knowledge streams.

Amazon MSK is a totally managed service that makes it straightforward so that you can construct and run functions on AWS that use Kafka to course of streaming knowledge.

Apache Flink is an opensource distributed processing engine, providing highly effective programming interfaces for each stream and batch processing, with first-class help for stateful processing, occasion time semantics, checkpointing, snapshots and rollback. Apache Flink helps a number of programming languages, Java, Python, Scala, SQL, and a number of APIs with completely different stage of abstraction, which can be utilized interchangeably in the identical utility.

Amazon Managed Service for Apache Flink is a totally managed, serverless expertise in working Apache Flink functions. Clients can simply construct actual time Flink functions utilizing any of Flink’s languages and APIs.

On this put up, we use a real-time inventory quotes feed from monetary knowledge supplier Alpaca and add an indicator when the worth strikes above or beneath a sure threshold. The code supplied within the GitHub repo permits you to deploy the answer to your AWS account. This answer was constructed by AWS Associate NETSOL Applied sciences.

Resolution overview

On this answer, we deploy an Apache Flink utility that enriches the uncooked knowledge feed, an MSK cluster that incorporates the messages streams for each the uncooked and enriched feeds, and an Amazon OpenSearch Service cluster that acts as a persistent knowledge retailer for querying the information. In a separate digital personal cloud (VPC) that acts because the buyer’s VPC, we additionally deploy an Amazon EC2 occasion working a Kafka shopper that consumes the enriched knowledge feed. The next diagram illustrates this structure.

Determine 1 – Resolution structure

The next is a step-by-step breakdown of the answer:

1. The EC2 occasion in your VPC is working a Python utility that fetches inventory quotes out of your knowledge supplier via an API. On this case, we use Alpaca’s API.
2. The applying sends these quotes utilizing Kafka shopper library to your kafka subject on MSK cluster. The kafka subject shops the uncooked quotes.
3. The Apache Flink utility takes the Kafka message stream and enriches it by including an indicator every time the inventory value rises or declines 5% or extra from the earlier enterprise day’s closing value.
4. The Apache Flink utility then sends the enriched knowledge to a separate Kafka subject in your MSK cluster.
5. The Apache Flink utility additionally sends the enriched knowledge stream to Amazon OpenSearch utilizing a Flink connector for OpenSearch. Amazon Opensearch shops the information, and OpenSearch Dashboards permits functions to question the information at any level sooner or later.
6. Your buyer is working a Kafka shopper utility on an EC2 occasion in a separate VPC in their very own AWS account. This utility makes use of AWS PrivateLink to devour the enriched knowledge feed securely, in actual time.
7. All Kafka person names and passwords are encrypted and saved in AWS Secrets and techniques Supervisor. The SASL/SCRAM authentication protocol used right here makes positive all knowledge to and from the MSK cluster is encrypted in transit. Amazon MSK encrypts all knowledge at relaxation within the MSK cluster by default.

The deployment course of consists of the next high-level steps:

1. Launch the Amazon MSK cluster, Apache Flink utility, Amazon OpenSearch Service area, and Kafka producer EC2 occasion within the producer AWS account. This step often completes inside 45 minutes.
2. Arrange multi-VPC connectivity and SASL/SCRAM authentication for the MSK cluster. This step can take as much as half-hour.
3. Launch the VPC and Kafka shopper EC2 occasion within the shopper account. This step takes about 10 minutes.

Stipulations

To deploy this answer, full the next prerequisite steps:

1. Create an AWS account in case you don’t have already got one and log in. We discuss with this because the producer account.
2. Create an AWS Id and Entry Administration (IAM) person with full admin permissions. For directions, discuss with Create an IAM person.
3. Signal out and signal again in to the AWS Administration Console as this IAM admin person.
4. Create an EC2 key pair named my-ec2-keypair within the producer account. If you have already got an EC2 key pair, you possibly can skip this step.
5. Comply with the directions in ALPACA_README to join a free Primary account at Alpaca to get your Alpaca API key and secret key. Alpaca will present the real-time inventory quotes for our enter knowledge feed.
6. Set up the AWS Command Line Interface (AWS CLI) in your native improvement machine and create a profile for the admin person. For directions, see Arrange the AWS Command Line Interface (AWS CLI).
7. Set up the newest model of the AWS Cloud Growth Package (AWS CDK) globally:

 npm set up -g aws-cdk@newest

Deploy the Amazon MSK cluster

These steps create a brand new supplier VPC and launch the Amazon MSK cluster there. You additionally deploy the Apache Flink utility and launch a brand new EC2 occasion to run the applying that fetches the uncooked inventory quotes.

1. In your improvement machine, clone the GitHub repo and set up the Python packages:

   git clone https://github.com/aws-samples/msk-powered-financial-data-feed.git
   cd msk-powered-financial-data-feed
   pip set up -r necessities.txt

2. Set the next atmosphere variables to specify your producer AWS account quantity and AWS Area:

   export CDK_DEFAULT_ACCOUNT={your_AWS_account_no}
   export CDK_DEFAULT_REGION=us-east-1

3. Run the next instructions to create your config.py file:

   echo "mskCrossAccountId = <Your producer AWS account ID>" > config.py
   echo "producerEc2KeyPairName="" " >> config.py
   echo "consumerEc2KeyPairName="" " >> config.py
   echo "mskConsumerPwdParamStoreValue="" " >> config.py
   echo "mskClusterArn = '' " >> config.py

4. Run the next instructions to create your alpaca.conf file:

   echo [alpaca] > dataFeedMsk/alpaca.conf
   echo ALPACA_API_KEY=your_api_key >> dataFeedMsk/alpaca.conf
   echo ALPACA_SECRET_KEY=your_secret_key >> dataFeedMsk/alpaca.conf

5. Edit the alpaca.conf file and substitute your_api_key and your_secret_key along with your Alpaca API key.
6. Bootstrap the atmosphere for the producer account:

   cdk bootstrap aws://{your_AWS_account_no}/{your_aws_region}

7. Utilizing your editor or built-in improvement atmosphere (IDE), edit the config.py file:
   1. Replace the mskCrossAccountId parameter along with your AWS producer account quantity.
   2. You probably have an present EC2 key pair, replace the producerEc2KeyPairName parameter with the title of your key pair.
8. View the dataFeedMsk/parameters.py file:
   1. If you’re deploying in a Area apart from us-east-1, replace the Availability Zone IDs az1 and az2 accordingly. For instance, the Availability Zones for us-west-2 would us-west-2a and us-west-2b.
   2. Make it possible for the enableSaslScramClientAuth, enableClusterConfig, and enableClusterPolicy parameters within the parameters.py file are set to False.
9. Be sure you are within the listing the place the app1.py file is positioned. Then deploy as follows:

   cdk deploy --all --app "python app1.py" --profile {your_profile_name}

10. Examine that you just now have an Amazon Easy Storage Service (Amazon S3) bucket whose title begins with awsblog-dev-artifacts containing a folder with some Python scripts and the Apache Flink utility JAR file.

Deploy multi-VPC connectivity and SASL/SCRAM

Full the next steps to deploy multi-VPC connectivity and SASL/SCRAM authentication for the MSK cluster:

1. Set the enableSaslScramClientAuth, enableClusterConfig, and enableClusterPolicy parameters within the config.py file to True.
2. Be sure you’re within the listing the place the config.py file is positioned and deploy the multi-VPC connectivity and SASL/SCRAM authentication for the MSK cluster:

cdk deploy --all --app "python app1.py" --profile {your_profile_name}

This step can take as much as half-hour.

3. To verify the outcomes, navigate to your MSK cluster on the Amazon MSK console, and select the Properties

You must see PrivateLink turned on, and SASL/SCRAM because the authentication kind.

4. Copy the MSK cluster ARN.
5. Edit your config.py file and enter the ARN as the worth for the mskClusterArn parameter, then save the up to date file.

Deploy the information feed shopper

Full the steps on this part to create an EC2 occasion in a brand new shopper account to run the Kafka shopper utility. The applying will connect with the MSK cluster via PrivateLink and SASL/SCRAM.

1. Navigate to Parameter Retailer, a functionality of AWS Techniques Supervisor, in your producer account.
2. Copy the worth of the blogAws-dev-mskConsumerPwd-ssmParamStore parameter and replace the mskConsumerPwdParamStoreValue parameter within the config.py file.
3. Examine the worth of the parameter named blogAws-dev-getAzIdsParamStore and make a remark of those two values.
4. Create one other AWS account for the Kafka shopper in case you don’t have already got one, and log in.
5. Create an IAM person with admin permissions.
6. Sign off and log again in to the console utilizing this IAM admin person.
7. Be sure you are in the identical Area because the Area you used within the producer account. Then create a brand new EC2 key pair named, for instance, my-ec2-consumer-keypair, on this shopper account.
8. Replace the worth of consumerEc2KeyPairName in your config.py file with the title of the important thing pair you simply created.
9. Open the AWS Useful resource Entry Supervisor (AWS RAM) console in your shopper account.
10. Examine the Availability Zone IDs from the Techniques Supervisor parameter retailer with the Availability Zone IDs proven on the AWS RAM console.
11. Establish the corresponding Availability Zone names for the matching Availability Zone IDs.
12. Open the parameters.py file within the dataFeedMsk folder and insert these Availability Zone names into the variables crossAccountAz1 and crossAccountAz2. For instance, in Parameter Retailer, if the values are “use1-az4” and “use1-az6”, then, once you swap to the buyer account’s AWS RAM console and examine, you might discover that these values correspond to the Availability Zone names “us-east-1a” and “us-east-1b”. In that case, you must replace the parameters.py file with these Availability Zone names by setting crossAccountAz1 to “us-east-1a” and crossAccountAz2 to “us-east-1b”.
13. Set the next atmosphere variables, specifying your shopper AWS account ID:

export CDK_DEFAULT_ACCOUNT={your_aws_account_id}
export CDK_DEFAULT_REGION=us-east-1

14. Bootstrap the buyer account atmosphere. It is advisable add particular insurance policies to the AWS CDK position on this case.

    cdk bootstrap aws://{your_aws_account_id}/{your_aws_region} --cloudformation-execution-policies "arn:aws:iam::aws:coverage/AmazonMSKFullAccess,arn:aws:iam::aws:coverage/AdministratorAccess" –-profile <your-user-profile>

You now must grant the buyer account entry to the MSK cluster.

15. On the console, copy the buyer AWS account quantity to your clipboard.
16. Signal out and signal again in to your producer AWS account.
17. On the Amazon MSK console, navigate to your MSK cluster.
18. Select Properties and scroll right down to Safety settings.
19. Select Edit cluster coverage and add the buyer account root to the Principal part as follows, then save the adjustments:

    "Principal": {
        "AWS": ["arn:aws:iam::<producer-acct-no>:root", "arn:aws:iam::<consumer-acct-no>:root"]
    },
    

20. Create the IAM position that must be hooked up to the EC2 shopper occasion:

    aws iam create-role --role-name awsblog-dev-app-consumerEc2Role --assume-role-policy-document file://dataFeedMsk/ec2ConsumerPolicy.json --profile <your-user-profile>

21. Deploy the buyer account infrastructure, together with the VPC, shopper EC2 occasion, safety teams, and connectivity to the MSK cluster:

    cdk deploy --all --app "python app2.py" --profile {your_profile_name}

Run the functions and think about the information

Now that we have now the infrastructure up, we will produce a uncooked inventory quotes feed from the producer EC2 occasion to the MSK cluster, enrich it utilizing the Apache Flink utility, and devour the enriched feed from the buyer utility via PrivateLink. For this put up, we use the Flink DataStream Java API for the inventory knowledge feed processing and enrichment. We additionally use Flink aggregations and windowing capabilities to determine insights in a sure time window.

Run the managed Flink utility

Full the next steps to run the managed Flink utility:

1. In your producer account, open the Amazon Managed Service for Apache Flink console and navigate to your utility.
2. To run the applying, select Run, choose Run with newest snapshot, and select Run.
   
3. When the applying adjustments to the Working state, select Open Apache Flink dashboard.

You must see your utility beneath Working Jobs.

Run the Kafka producer utility

Full the next steps to run the Kafka producer utility:

1. On the Amazon EC2 console, find the IP deal with of the producer EC2 occasion named awsblog-dev-app-kafkaProducerEC2Instance.
2. Hook up with the occasion utilizing SSH and run the next instructions:

   sudo su
   cd atmosphere
   supply alpaca-script/bin/activate
   python3 ec2-script-live.py AMZN NVDA

It is advisable begin the script throughout market open hours. This may run the script that creates a connection to the Alpaca API. You must see traces of output exhibiting that it’s making the connection and subscribing to the given ticker symbols.

View the enriched knowledge feed in OpenSearch Dashboards

Full the next steps to create an index sample to view the enriched knowledge in your OpenSearch dashboard:

1. To search out the grasp person title for OpenSearch, open the config.py file and find the worth assigned to the openSearchMasterUsername parameter.
2. Open Secrets and techniques Supervisor and click on on awsblog-dev-app-openSearchSecrets secret to retrieve the password for OpenSearch.
3. Navigate to your OpenSearch console and discover the URL to your OpenSearch dashboard by clicking on the area title on your OpenSearch cluster. Click on on the URL and register utilizing your grasp person title and password.
4. Within the OpenSearch navigation bar on the left, choose Dashboards Administration beneath the Administration part.
5. Select Index patterns, then select Create index sample.
6. Enter amzn* within the Index sample title discipline to match the AMZN ticker, then select Subsequent step.
   
7. Choose timestamp beneath Time discipline and select Create index sample.
8. Select Uncover within the OpenSearch Dashboards navigation pane.
9. With amzn chosen on the index sample dropdown, choose the fields to view the enriched quotes knowledge.

The indicator discipline has been added to the uncooked knowledge by Amazon Managed Service for Apache Flink to point whether or not the present value route is impartial, bullish, or bearish.

Run the Kafka shopper utility

To run the buyer utility to devour the information feed, you first must get the multi-VPC brokers URL for the MSK cluster within the producer account.

1. On the Amazon MSK console, navigate to your MSK cluster and select View shopper info.
2. Copy the worth of the Personal endpoint (multi-VPC).
   
3. SSH to your shopper EC2 occasion and run the next instructions:

   sudo su
   alias kafka-consumer=/kafka_2.13-3.5.1/bin/kafka-console-consumer.sh
   kafka-consumer --bootstrap-server {$MULTI_VPC_BROKER_URL} --topic amznenhanced --from-beginning --consumer.config ./customer_sasl.properties
   

You must then see traces of output for the enriched knowledge feed like the next:

{"image":"AMZN","shut":194.64,"open":194.58,"low":194.58,"excessive":194.64,"quantity":255.0,"timestamp":"2024-07-11 19:49:00","%change":-0.8784661217630548,"indicator":"Impartial"}
{"image":"AMZN","shut":194.77,"open":194.615,"low":194.59,"excessive":194.78,"quantity":1362.0,"timestamp":"2024-07-11 19:50:00","%change":-0.8122628778040887,"indicator":"Impartial"}
{"image":"AMZN","shut":194.82,"open":194.79,"low":194.77,"excessive":194.82,"quantity":1143.0,"timestamp":"2024-07-11 19:51:00","%change":-0.7868000916660381,"indicator":"Impartial"}

Within the output above, no important adjustments are occurring to the inventory costs, so the indicator exhibits “Impartial”. The Flink utility determines the suitable sentiment based mostly on the inventory value motion.

Extra monetary providers use instances

On this put up, we demonstrated find out how to construct an answer that enriches a uncooked inventory quotes feed and identifies inventory motion patterns utilizing Amazon MSK and Amazon Managed Service for Apache Flink. Amazon Managed Service for Apache Flink gives numerous options comparable to snapshot, checkpointing, and a just lately launched Rollback API. These options can help you construct resilient real-time streaming functions.

You’ll be able to apply this method to a wide range of different use instances within the capital markets area. On this part, we focus on different instances during which you should use the identical architectural patterns.

Actual-time knowledge visualization

Utilizing real-time feeds to create charts of shares is the commonest use case for real-time market knowledge within the cloud. You’ll be able to ingest uncooked inventory costs from knowledge suppliers or exchanges into an MSK subject and use Amazon Managed Service for Apache Flink to show the excessive value, low value, and quantity over a time period. This is named aggregates and is the inspiration for displaying candlestick bar graphs. You too can use Flink to find out inventory value ranges over time.

Inventory implied volatility

Implied volatility (IV) is a measure of the market’s expectation of how a lot a inventory’s value is prone to fluctuate sooner or later. IV is forward-looking and derived from the present market value of an possibility. It is usually used to cost new choices contracts and is typically known as the inventory market’s concern gauge as a result of it tends to spike larger throughout market stress or uncertainty. With Amazon Managed Service for Apache Flink, you possibly can devour knowledge from a securities feed that can present present inventory costs and mix this with an choices feed that gives contract values and strike costs to calculate the implied volatility.

Technical indicator engine

Technical indicators are used to investigate inventory value and quantity habits, present buying and selling indicators, and determine market alternatives, which may help within the decision-making strategy of buying and selling. Though implied volatility is a technical indicator, there are lots of different indicators. There could be easy indicators comparable to “Easy Shifting Common” that characterize a measure of development in a selected inventory value based mostly on the typical of value over a time period. There are additionally extra advanced indicators comparable to Relative Power Index (RSI) that measures the momentum of a inventory’s value motion. RSI is a mathematical method that makes use of the exponential transferring common of upward actions and downward actions.

Market alert engine

Graphs and technical indicators aren’t the one instruments that you should use to make funding selections. Different knowledge sources are necessary, comparable to ticker image adjustments, inventory splits, dividend funds, and others. Traders additionally act on latest information in regards to the firm, its rivals, staff, and different potential company-related info. You need to use the compute capability supplied by Amazon Managed Service for Apache Flink to ingest, filter, remodel, and correlate the completely different knowledge sources to the inventory costs and create an alert engine that may advocate funding actions based mostly on these alternate knowledge sources. Examples can vary from invoking an motion if dividend costs enhance or lower to utilizing generative synthetic intelligence (AI) to summarize a number of correlated information objects from completely different sources right into a single alert about an occasion.

Market surveillance

Market surveillance is the monitoring and investigation of unfair or unlawful buying and selling practices within the inventory markets to keep up honest and orderly markets. Each personal corporations and authorities companies conduct market surveillance to uphold guidelines and shield traders.

You need to use Amazon Managed Service for Apache Flink streaming analytics as a robust surveillance device. Streaming analytics can detect even refined cases of market manipulation in actual time. By integrating market knowledge feeds with exterior knowledge sources, comparable to firm merger bulletins, information feeds, and social media, streaming analytics can shortly determine potential makes an attempt at market manipulation. This permits regulators to be alerted in actual time, enabling them to take immediate motion even earlier than the manipulation can absolutely unfold.

Markets threat administration

In fast-paced capital markets, end-of-day threat measurement is inadequate. Companies want real-time threat monitoring to remain aggressive. Monetary establishments can use Amazon Managed Service for Apache Flink to compute intraday value-at-risk (VaR) in actual time. By ingesting market knowledge and portfolio adjustments, Amazon Managed Service for Apache Flink supplies a low-latency, high-performance answer for steady VaR calculations.

This permits monetary establishments to proactively handle threat by shortly figuring out and mitigating intraday exposures, relatively than reacting to previous occasions. The flexibility to stream threat analytics empowers corporations to optimize portfolios and keep resilient in risky markets.

Clear up

It’s all the time an excellent follow to scrub up all of the assets you created as a part of this put up to keep away from any extra value. To scrub up your assets, full the next steps:

1. Delete the CloudFormation stacks from the buyer account.
2. Delete the CloudFormation stacks from the supplier account.

Conclusion

On this put up, we confirmed you find out how to present a real-time monetary knowledge feed that may be consumed by your clients utilizing Amazon MSK and Amazon Managed Service for Apache Flink. We used Amazon Managed Service for Apache Flink to counterpoint a uncooked knowledge feed and ship it to Amazon OpenSearch. Utilizing this answer as a template, you possibly can mixture a number of supply feeds, use Flink to calculate in actual time any technical indicator, show knowledge and volatility, or create an alert engine. You’ll be able to add worth on your clients by inserting extra monetary info inside your feed in actual time.

We hope you discovered this put up useful and encourage you to check out this answer to unravel attention-grabbing monetary trade challenges.

Concerning the Authors

Rana Dutt is a Principal Options Architect at Amazon Internet Providers. He has a background in architecting scalable software program platforms for monetary providers, healthcare, and telecom corporations, and is obsessed with serving to clients construct on AWS.

Amar Surjit is a Senior Options Architect at Amazon Internet Providers (AWS), the place he focuses on knowledge analytics and streaming providers. He advises AWS clients on architectural finest practices, serving to them design dependable, safe, environment friendly, and cost-effective real-time analytics knowledge programs. Amar works intently with clients to create modern cloud-based options that deal with their distinctive enterprise challenges and speed up their transformation journeys.

Diego Soares is a Principal Options Architect at AWS with over 20 years of expertise within the IT trade. He has a background in infrastructure, safety, and networking. Previous to becoming a member of AWS in 2021, Diego labored for Cisco, supporting monetary providers clients for over 15 years. He works with giant monetary establishments to assist them obtain their enterprise targets with AWS. Diego is obsessed with how expertise solves enterprise challenges and supplies useful outcomes by growing advanced answer architectures.