Desarrollo de un cuadro de mandos para Finnhub Stock

$ terraform init
$ terraform apply

EXTERNAL_IP= "********"

$ docker-compose up -d --env-file /home/ubuntu/.env

websockets
websocket-client
finnhub-python
kafka-python
avro

{
  "KAFKA_SERVER": "kafka",
  "KAFKA_PORT": "29092",
  "KAFKA_TOPIC_NAME": "market",
  "FINNHUB_API_KEY": "**********" (1)
}

import os
import websocket
import json
import io
import avro.schema
import avro.io

from kafka import KafkaProducer

# https://finnhub.io/docs/api/websocket-trades
class FinnhubProducer:
    def __init__(self):
        """
        Producer class that connects to the finnhub websocket, encodes & validates the JSON payload
        in avro format against pre-defined schema then sends data to kafka.
        """

        # define config from config file
        self.config = self.load_config('config.json')

        # define the kafka producer here. This assumes there is a kafka server already setup at the address and port
        #self.producer = KafkaProducer(bootstrap_servers=f"{self.config['KAFKA_SERVER']}:{self.config['KAFKA_PORT']}",api_version=(0, 10, 1))


        kafka_servers=[config['KAFKA_SERVER'] + config['KAFKA_PORT']]
        self.producer = KafkaProducer(bootstrap_servers = kafka_servers,api_version=(0, 10, 1))

        # define the avro schema here. This assumes the schema is already defined in the src/schemas folder
        # this helps us enforce the schema when we send data to kafka
        self.avro_schema = avro.schema.parse(open('trades.avsc').read())
        print("AVRO schema loaded")

        # define the websocket client
        self.ws = websocket.WebSocketApp(f"wss://ws.finnhub.io?token={self.config['FINNHUB_API_KEY']}",
                                         on_message=self.on_message,
                                         on_error=self.on_error,
                                         on_close=self.on_close)
        self.ws.on_open = self.on_open
        self.ws.run_forever()

    def load_config(self, config_file):
        with open(config_file, 'r') as f:
            config = json.load(f)
        return config

    def avro_encode(self, data, schema):
        """
        Avro encode data using the provided schema.

        Parameters
        ----------
        data : dict
            Data to encode.
        schema : avro.schema.Schema
            Avro schema to use for encoding.

        Returns
        -------
        bytes : Encoded data.
        """

        writer = avro.io.DatumWriter(schema)
        bytes_writer = io.BytesIO()
        encoder = avro.io.BinaryEncoder(bytes_writer)
        writer.write(data, encoder)
        return bytes_writer.getvalue()

    def on_message(self, ws, message):
        """
        Callback function that is called when a message is received from the websocket.

        Parameters
        ----------
        ws : websocket.WebSocketApp
            Websocket client.
        message : str
            Message received from the websocket.
        """
        message = json.loads(message)
        avro_message = self.avro_encode(
            {
                'data': message['data'],
                'type': message['type']
            },
            self.avro_schema
        )
        self.producer.send(self.config['KAFKA_TOPIC_NAME'], avro_message)

    def on_error(self, ws, error):
        """
        Websocket error callback. This currently just prints the error to the console.
        In a production environment, this should be logged to a file or sent to a monitoring service.

        Parameters
        ----------
        ws : websocket.WebSocketApp
            Websocket client.
        error : str
            Error message.
        """
        print(error)

    def on_close(self, ws):
        """
        Websocket close callback. This currently just prints a message to the console.
        In a production environment, this should be logged to a file or sent to a monitoring service.

        Parameters
        ----------
        ws : websocket.WebSocketApp
            Websocket client.
        """
        print("### closed ###")

    def on_open(self, ws):
        """
        Websocket open callback. This subscribes to the MSFT stock topic on the websocket.

        Parameters
        ----------
        ws : websocket.WebSocketApp
            Websocket client.
        """
        print("sending subscribe message")
        self.ws.send('{"type":"subscribe","symbol":"BINANCE:BTCUSDT"}')
        print("subscribed to AAPL")


if __name__ == "__main__":
    FinnhubProducer()

websockets
websocket-client
finnhub-python
kafka-python
avro

from kafka import KafkaConsumer
import json
import io
import avro.io
import avro.schema

with open('config.json', 'r') as f:
    config = json.load(f)

# define the consumer to read from the Kafka topic

# define kafka servers from the config file (KAFKA_SERVER variable)
kafka_servers=[config['KAFKA_SERVER'] + config['KAFKA_PORT']]
consumer = KafkaConsumer(
    config['KAFKA_TOPIC_NAME'],
    bootstrap_servers = kafka_servers,
    api_version=(0, 10, 1)
    )

# define the Avro schema that corresponds to the encoded data
schema = avro.schema.parse(open('trades.avsc').read())

for message in consumer:
    # asssume 'byte_string' contains the Avro-encoded byte string,
    # we need to decode it using avro library
    bytes_reader = io.BytesIO(message.value)
    decoder = avro.io.BinaryDecoder(bytes_reader)
    reader = avro.io.DatumReader(schema)
    data = reader.read(decoder)
    print(data)

{
  "KAFKA_SERVER": "kafka",
  "KAFKA_PORT": "29092",
  "KAFKA_TOPIC_NAME": "market"
}

{
  "type" : "record",
  "name" : "message",
  "namespace" : "FinnhubProducer",
  "fields" : [ {
    "name" : "data",
    "type" : {
      "type" : "array",
      "items" : {
        "type" : "record",
        "name" : "data",
        "fields" : [ {
          "name" : "c",
          "type":[
            {
               "type":"array",
               "items":["null","string"],
               "default":[]
            },
            "null"
          ],
          "doc" : "Trade conditions"
        },
        {
          "name" : "p",
          "type" : "double",
          "doc" : "Price at which the stock was traded"
        },
        {
          "name" : "s",
          "type" : "string",
          "doc" : "Symbol of a stock"
        },
        {
          "name" : "t",
          "type" : "long",
          "doc" : "Timestamp at which the stock was traded"
        },
        {
          "name" : "v",
          "type" : "double",
          "doc" : "Volume at which the stock was traded"
        } ]
      },
      "doc" : "Trades messages"
    },
    "doc"  : "Contains data inside a message"
  },
  {
    "name" : "type",
    "type" : "string",
    "doc"  : "Type of message"
  } ],
  "doc" : "A schema for upcoming Finnhub messages"
}

pyspark==3.2.1
kafka-python
cassandra-driver

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.avro.functions import from_avro
from pyspark.sql.streaming import *

from uuid import uuid1

# create a Spark session
spark = SparkSession \
    .builder \
    .master("spark://master:7077") \
    .appName("StreamProcessor") \
    .config("spark.cassandra.connection.host", 'cassandra1') \
    .config("spark.cassandra.connection.port", '9042') \
    .config("spark.cassandra.auth.username", 'cassandra') \
    .config("spark.cassandra.auth.password", 'cassandra') \
    .getOrCreate()

# suppress all the INFO logs except for errors
spark.sparkContext.setLogLevel("ERROR")

# define the Avro schema that corresponds to the encoded data
tradesSchema = open('./trades.avsc', 'r').read()

@udf(returnType=StringType())
def makeUUID():
    return str(uuid1())

# define the stream to read from the Kafka topic market
inputDF = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:29092") \
    .option("subscribe", "market") \
    .option("minPartitions", "1") \
    .option("maxOffsetsPerTrigger", "1000") \
    .option("useDeprecatedOffsetFetching", "false") \
    .load()



# explode the data column and select the columns we need
expandedDF = inputDF \
    .withColumn("avroData", from_avro(col("value"), tradesSchema)) \
    .select(col("avroData.*")) \
    .select(explode(col("data")), col("type")) \
    .select(col("col.*"), col("type"))

# create the final dataframe with the columns we need plus the ingest timestamp
finalDF = expandedDF \
    .withColumn("uuid", makeUUID()) \
    .withColumnRenamed("c", "trade_conditions") \
    .withColumnRenamed("p", "price") \
    .withColumnRenamed("s", "symbol") \
    .withColumnRenamed("t", "trade_timestamp") \
    .withColumnRenamed("v", "volume") \
    .withColumn("trade_timestamp", (col("trade_timestamp") / 1000).cast("timestamp")) \
    .withColumn("ingest_timestamp", current_timestamp().alias("ingest_timestamp"))

# write the final dataframe to Cassandra
# spark handles the streaming and batching for us
query = finalDF \
    .writeStream \
    .trigger(processingTime="5 seconds") \
    .foreachBatch(lambda batchDF, batchId: \
        batchDF.write \
            .format("org.apache.spark.sql.cassandra") \
            .option("table", "trades") \
            .option("keyspace", "market") \
            .mode("append") \
            .save()) \
    .outputMode("update") \
    .start()

# create a summary dataframe with the average price * volume
summaryDF = finalDF \
    .withColumn("price_volume_multiply", col("price") * col("volume")) \
    .withWatermark("trade_timestamp", "15 seconds") \
    .groupBy("symbol") \
    .agg(avg("price_volume_multiply").alias("price_volume_multiply"))

# add UUID and ingest timestamp to the summary dataframe and rename agg column
finalsummaryDF = summaryDF \
    .withColumn("uuid", makeUUID()) \
    .withColumn("ingest_timestamp", current_timestamp().alias("ingest_timestamp")) \
    .withColumnRenamed("avg(price_volume_multiply)", "price_volume_multiply")

# write the summary dataframe to Cassandra in 5 second batches
query2 = finalsummaryDF \
    .writeStream \
    .trigger(processingTime="5 seconds") \
    .foreachBatch(lambda batchDF, batchId: \
        batchDF.write \
            .format("org.apache.spark.sql.cassandra") \
            .option("table", "running_averages_15_sec") \
            .option("keyspace", "market") \
            .mode("append") \
            .save()) \
    .outputMode("update") \
    .start()

# wait for the stream to terminate - i.e. wait forever
spark.streams.awaitAnyTermination()

$ conda activate python3_10