DEV Community: 0x26res

Build a real-time crypto analytics dashboard with Beavers and Perspective

0x26res — Thu, 25 Jul 2024 10:12:08 +0000

Coinbase Market Data API Analytics

This example shows how you can leverage two powerful python libraries, Beavers and Perspective, to analyse data in realtime and display it in a dashboard.
This tutorial assumes you are familiar with Kafka and Python and Apache Arrow.

Architecture Overview

We will connect to Coinbase's websocket API to receive crypto market updates in real time.
In order to share this data with other services and decouple producers from consumers, we'll publish this data over Kafka, as json.
We'll then run a Beavers job that will read the data from Kafka, enrich it, and publish it in a perspective dashboard.

Initial Set Up

You'll need:

Git
Python (at least 3.10)
Docker to run a Kafka cluster

The code for this tutorial is available on github

Clone the repo

git clone https://github.com/0x26res/beavers-examples
cd beavers-example/01_coinbase_analytics/

Set Up the Virtual Environment

python3 -m venv --clear .venv
source ./.venv/bin/activate
pip install -r requirements.txt

Set Up Kafka

We use the kafka-kraft docker image to run a super simple Kafka cluster.
To start the cluster:

docker run --name=simple_kafka -p 9092:9092 -d bashj79/kafka-kraft

Once started you can create a Kafka topic called ticker

docker exec simple_kafka /opt/kafka/bin/kafka-topics.sh \
  --create \
  --topic=ticker \
  --partitions=1 \
  --bootstrap-server=localhost:9092 \
  --replication-factor=1

Publish Coinbase's Market Data on Kafka

In this step, we'll run a simple python job that listen to Coinbase's Websocket market data API, and publish the data on the ticker Kafka topic.

python ./websocket_feed.py

You should now be able to see the Coinbase data streaming on Kafka.

docker exec simple_kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --topic=ticker \
  --bootstrap-server=localhost:9092

Run the dashboard

python ./dashboard.py

You can see the dashboard in http://localhost:8082/ticker.

Introducing Beavers

In order to build our dashboard, we'll use Beavers.
Beavers is a streaming python library optimized for analytics.

At its core, Beavers uses a ~~dam~~ DAG to process incoming data.
Each node in the DAG is a Python function.

dag = Dag()

The first node in the dashboard DAG is a source node, called ticker.
Its output is a pyarrow.Table for which we need to specify the schema.

ticker = dag.pa.source_table(schema=TICKER_SCHEMA, name="ticker")

This is what is displayed in the dashboard.

Simple Transformation in Beavers

Next, we want to add a derived column to ticker.
The new columns, spread, is the difference between the best_ask and best_bid.

For this we just introduce a function:

def add_spread(table: pa.Table) -> pa.Table:
    return table.append_column(
        "spread", pc.subtract(table["best_ask"], table["best_bid"])
    )

And add the function to the DAG:

ticker_with_spread = dag.pa.table_stream(
    add_spread, schema=TICKER_WITH_SPREAD_SCHEMA
).map(ticker)

You can see it in: http://localhost:8082/ticker_with_spread

Advanced Transformation with Beavers

Now let's introduce a more advanced computation.
For each incoming ticker record, we would like to calculate the change in the last 5 minutes

For this, first we need to keep track of the history over the last 10 minutes (we do need a bit more than 5 minutes).

ticker_history = dag.state(TickerHistory()).map(ticker, dag.now())

And then do an as of join, to find the price 5 minutes ago and calculate the change

ticker_with_change = dag.pa.table_stream(
    add_5min_change, TICKER_WITH_CHANGE_SCHEMA
).map(ticker, ticker_history)

You can see it in: http://localhost:8082/ticker_with_change

Fastsandra: Faster Python Cassandra Driver

0x26res — Sun, 05 Jan 2020 21:23:52 +0000

Make the most of the Datastax Cassandra driver for python by tuning it correctly.

Quick example

Throughout this tutorial we'll use a Cassandra table called time_series to store a simple time series. Here's the code to set up the table:


CREATE KEYSPACE fastsandra
  WITH REPLICATION = { 
   'class' : 'SimpleStrategy', 
   'replication_factor' : 1 
  };


CREATE TABLE fastsandra.time_series ( 
  event_date date,
  instrument_id int,
  event_timestamp timestamp,
  value double,
  PRIMARY KEY (event_date, instrument_id, event_timestamp)
);

And here's how to query the data for given date, using Cluster.execute:

import cassandra.cluster

cluster = cassandra.cluster.Cluster() # default cluster, pointing to localhost

with cluster.connect('fastsandra') as session:    
    results = session.execute(
        "SELECT * from fastsandra.time_series where event_date = '2019-10-01'"
    )
    rows = [r for r in results]
rows[0]
# Row(event_date=Date(18170), instrument_id=1, event_timestamp=datetime.datetime(2019, 10, 1, 8, 0), value=0.2210726153252428)
rows[0].value
# 0.2210726153252428

The driver returns a bunch of rows, each of them is a named tuple.

It's very easy to convert the results into a pandas DataFrame:

import pandas as pd
pd.DataFrame(rows)

   event_date  instrument_id     event_timestamp     value
0  2019-10-01              1 2019-10-01 08:00:00  0.221073
1  2019-10-01              1 2019-10-01 08:15:00  0.661251
2  2019-10-01              1 2019-10-01 08:30:00  0.927390
3  2019-10-01              1 2019-10-01 08:45:00  0.083483
4  2019-10-01              1 2019-10-01 09:00:00  0.934817

Speed up the driver with `NumpyProtocolHandler`

If you're dealing with a lot of data, there is a big performance and memory overhead when using named tuples and plain python objects (int, float etc). It's much better to use numpy arrays and cython for this.

The good news is that you can set up the Cassandra driver so it reads the data directly into numpy array, using cython, saving both CPU and memory.

Installing the cython version of the Cassandra driver with `NumpyProtocolHandler`

First you need to install the driver correctly. Here's how to do it on ubuntu, for python 3.7:

# First make sure you have your the correct library installed on your system:
sudo apt install python3.7-dev libev4 libev-dev
# Activate your your virtual environment and first install Cython and numpy
pip install Cython==0.29.14
pip install numpy==1.17.1
# Then install the cassandra-driver (this should take a few minutes)
pip install cassandra-driver
# Check that it worked:
from cassandra.protocol import NumpyProtocolHandler
# Should print:
# <class 'cassandra.protocol.cython_protocol_handler.<locals>.CythonProtocolHandler'>
# If it doesn't print anything, it didn't work

This can be a bit tricky so here are some trouble shooting tips:

You must have the python development library installed
You must install Cython before you install the driver
If you installed cassandra-driver previously you may have to clear your pip cache
Another option is to use pip --no-cache-dir
If it still does not work, run the installation with -v and check the logs

pip -v --no-cache-dir install cassandra-driver

Using `NumpyProtocolHandler`

from cassandra.protocol import NumpyProtocolHandler
from cassandra.query import tuple_factory

with cluster.connect('fastsandra') as session:
    session.row_factory = tuple_factory  #required for Numpy results
    session.client_protocol_handler = NumpyProtocolHandler  # for a dict of NumPy arrays as result
    results = session.execute(
        "SELECT * from fastsandra.time_series where event_date = '2019-10-01'"
    )
    np_rows = [r for r in results]
df = pd.concat([pd.DataFrame(r) for r in rows])

With Session client_protocol_handler set to NumpyProtocolHandler the content of the result set is changed.
Instead of a sequence of named tuples, it is now a sequence of dict of numpy ndarray.
For example np_rows[0]['values'] is a masked array of double:

masked_array(data=[0.2210726153252428, 0.6612507337531311,
                   0.9273900637252853, ..., 0.1700777337201711,
                   0.6348330019120819, 0.23939705731588268],
             mask=[False, False, False, ..., False, False, False],
       fill_value=1e+20)

And np_rows[0]['instrument_id'] is a masked array of int32:

masked_array(data=[1, 1, 1, ..., 136, 136, 136],
             mask=[False, False, False, ..., False, False, False],
       fill_value=999999,
            dtype=int32)

This means that we save a lot of memory and CPU time by not having the intermediate objects (integer, float, named tuple).

Tuning the driver further

Numpy is much faster when the data type of the array is a native type: double, integer and timestamp.
On the other hand it is very slow when the data type is object.
Unfortunately the cassandra driver doesn't have support timestamp.
Our date and timestamp columns are objects:

np_rows[0]['event_date'].dtype
# dtype('O')
np_rows[0]['event_timestamp'].dtype
# dtype('O')

But there is an easy way to monkey patch the driver so it supports datetime (and date).

numpy_parser defines mapping between the cassandra and numpy data types

_cqltype_to_numpy = {
    cqltypes.LongType:          np.dtype('>i8'),
    cqltypes.CounterColumnType: np.dtype('>i8'),
    cqltypes.Int32Type:         np.dtype('>i4'),
    cqltypes.ShortType:         np.dtype('>i2'),
    cqltypes.FloatType:         np.dtype('>f4'),
    cqltypes.DoubleType:        np.dtype('>f8'),
}

All we need to do is add support for date and timestamp.

Add support for timestamps

import cassandra.cqltypes
import cassandra.numpy_parser as numpy_parser


numpy_parser._cqltype_to_numpy.update({
    cassandra.cqltypes.DateType: np.dtype('datetime64[ms]'),
    cassandra.cqltypes.TimestampType: np.dtype('datetime64[ms]'),
})

And if we reload the data from cassandra we can see:

np_rows[0]['event_timestamp'].dtype
# dtype('>M8[ms]')

This is very easy because the behind the scene Cassandra represent timestamp as the number of millis since epoch, the same way numpy does.

Add support for dates

For dates it requires more work. Because of the way Cassandra represents date:

# Values of the 'date'` type are encoded as 32-bit unsigned integers
# representing a number of days with epoch (January 1st, 1970) at the center of the
# range (2^31).

When loading date fields we need to do some massaging:

load unsigned int (32bit)
withdraw the EPOCH_OFFSET (2^31)
Convert to datetime64[D]

numpy_parser._cqltype_to_numpy.update({
  cassandra.cqltypes.SimpleDateType: np.dtype('>u4'),
})

def result_set_to_df(results: cassandra.cluster.ResultSet) -> pd.DataFrame:
    df = pd.DataFrame(pd.concat((pd.DataFrame(r) for r in  results)))
    for name, dtype in zip(results.column_names, results.column_types):
        if dtype == cassandra.cqltypes.SimpleDateType:
            df[name] = (df[name] - cassandra.cqltypes.SimpleDateType.EPOCH_OFFSET_DAYS).astype('datetime64[D]')
    return df

And you can run this example:

with cluster.connect('fastsandra') as session:
    session.row_factory = tuple_factory
    session.client_protocol_handler = NumpyProtocolHandler

    results = session.execute(
        "SELECT * from fastsandra.time_series where event_date = '2019-10-01'"
    )
    df = result_set_to_df(results)

The resulting DataFrame has got the following dtypes:

event_date         datetime64[ns]
instrument_id               int32
event_timestamp    datetime64[ns]
value                     float64

There is zero object overhead, everything goes straight from the wire format to numpy arrays.