Trading Strategy

For each of these components, we define its purpose, associated columns and artifacts, as well as a mention of team ownership and a sense of stochasticity.

Component: TweetSource

This component refers to the source data used to conduct all tests defined. This benchmark data is unchanged and is used as the input to the SentimentClassifier and the EntityExtractor components for our tests.

Columns:

• tweet_id: Unique identifier for each tweet.

• tweet_content: Tweets related to financial markets.

Component: SentimentClassifier

This component is responsible for analyzing the sentiment of the tweets. Given a tweet, it will output a probability vector of the sentiment of the tweet. The vector has three elements: positive, neutral, and negative. The sum of the vector elements is 1. The SentimentClassifier is owned by a separate team and may be stochastic.

Columns:

• positive_sentiment: Measure of positive sentiment, value between 0 and 1.

• neutral_sentiment: Measure of neutral sentiment, value between 0 and 1.

• negative_sentiment: Measure of positive sentiment, value between 0 and 1.

• ground_truth_sentiment: Ground truth sentiment category for the tweet.

• sentiment_latency_ms: Classifier inference time (in milliseconds).

Component: EntityExtractor

This component is responsible for extracting entities from the tweets. Given a tweet, it will output a list of entities. The EntityExtractor is owned by a separate team and may be stochastic.

Columns:

• entities: List of entities extracted from the tweet.

• entity_latency_ms: Time taken by the entity extractor to process the tweet.

Component: TradeRecommender

This component is responsible for recommending trades based on the sentiment and entities of the tweets. Given the sentiment vector and the list of entities from the SentimentClassifier and EntityExtractor, it will output a list of trades. The TradeRecommender is owned by a separate team, and while it is not stochastic in how it follows the trading strategy, it may be stochastic because its upstream components are stochastic.

Columns:

• buy_rec: List of company ticker symbols to buy.

• sell_rec: List of company ticker symbols to sell.

• trade_latency_ms: Time taken by the trading model to process the input and make a decision.

Component: Global

This component is a placeholder to represent the information that is shared (or derived) by more than one component. In this case, it represents the total time taken for the system to infer a trading recommendation from a tweet.

Columns:

• total_latency_ms: Total time taken for the system to infer a trading recommendation from a tweet.

The run configuration defines the columns of data that are associated with the components of the Trading Strategy (listed above), as well as the DAG of the components. The run_config.json file in this tutorial defines the run configuration for the Trading Strategy.

Columns

The columns are defined in the columns field of the run configuration. Below we show the columns associated to the EntityExtractor component. Notice how each column is associated with a specific component, and how we can support more than just numerical types, as depicted with the entities, which is of type list. Also, the the data we are submitting includes application data, such as entity_latency_ms.

import json

with open('run_config.json') as f:
    run_config_dict = json.load(f)

for c in run_config_dict['columns']:
    if c['component'] == 'EntityExtractor':
        print(json.dumps(c, indent=2))

{
  "component": "EntityExtractor",
  "name": "entities",
  "type": "list",
  "description": "List of entities extracted from the tweet."
}
{
  "component": "EntityExtractor",
  "name": "entity_latency_ms",
  "type": "float",
  "description": "Time taken by the entity extractor to process the tweet."
}

Components DAG

The components DAG is a dictionary which maps the component name to the list of the components that depend on it.

run_config_dict['components_dag']

{'TweetSource': ['SentimentClassifier', 'EntityExtractor'],
 'SentimentClassifier': ['TradeRecommender'],
 'EntityExtractor': ['TradeRecommender'],
 'TradeRecommender': [],
 'Global': []}

In this tutorial, we have prepared a number of parquet files with examples of the data that is generated by testing the Trading Strategy. In reality, the creation of these files would be done by the user, and we are going to cover how to do this now.

To wit, we will sketch out a hypothetical environment and how to pass the tweets through the system, and track each of the columns that are associated with the components of the system. We will then compile these columns into a parquet file.

In subsequent sections, we will show how to send these parquet files to dbnl and eventually run some tests.

Environment

We'll assume that we can access Snowflake to retrieve the tweets and that the SentimentClassifier and EntityExtractor are external services we call. We'll implement the TradeRecommender and orchestrate the whole process, logging important steps. Finally, we'll save the results as a parquet file.

Steps

Step 1: Simulate Tweets with Ground Truth Sentiment stored in Snowflake

We will simulate fetching tweets from Snowflake. The fetch_tweets function will return a list of tweets with the columns tweet_id, tweet_content, and ground_truth_sentiment.

import snowflake.connector

def get_snowflake_connection():
    """Simulate connecting to Snowflake."""
    return snowflake.connector.connect(
        user="user",
        password="password",
        account="account",
        warehouse="warehouse",
        database="database",
        schema="schema"
    )

def fetch_tweets(tweet_ids):
    """Simulate fetching tweets from Snowflake."""
    with get_snowflake_connection() as conn:
        cursor = conn.cursor()
        cursor.execute(
            f"""
            SELECT tweet_id, tweet_content, ground_truth_sentiment
            FROM tweets
            WHERE tweet_id IN ({', '.join(map(str, tweet_ids))})
            """
        )
        tweets = cursor.fetchall()
    return tweets

Step 2: Define Component Functions.

We'll define functions for each component in the process. Two of these functions will call the sentiment classifier and entity extractor services. The third function will implement the trade recommender. Note that in reality you would likely be calling all these services.

Component Function: Get Sentiment

import time
import requests

def call_sentiment_classifier(tweet):
    """Call an external SentimentClassifier API."""
    start_time = time.time()
    response = requests.post('http://example.com/sentiment', json={"text": tweet['tweet_content']})
    sentiment = response.json()
    latency = (time.time() - start_time) * 1000
    return {
        "positive_sentiment": sentiment['positive'],
        "neutral_sentiment": sentiment['neutral'],
        "negative_sentiment": sentiment['negative'],
        "sentiment_latency_ms": latency
    }

Component Function: Get Entities

import time
import requests

def call_entity_extractor(tweet):
    """Call an external EntityExtractor API."""
    start_time = time.time()
    response = requests.post('http://example.com/entity', json={"text": tweet['tweet_content']})
    entities = response.json()['entities']
    latency = (time.time() - start_time) * 1000
    return {
        "entities": entities,
        "entity_latency_ms": latency
    }

Component Function: Recommend Trades

def recommend_trade(sentiment, entities):
    """Recommend trades based on sentiment and entities."""
    start_time = time.time()
    buy_rec = []
    sell_rec = []
    for entity in entities:
        if sentiment['positive_sentiment'] >= 0.7:
            buy_rec.append(entity)
        else:
            sell_rec.append(entity)
    latency = (time.time() - start_time) * 1000
    return {
        "buy_rec": buy_rec,
        "sell_rec": sell_rec,
        "trade_latency_ms": latency
    }

Step 3: Implement the Orchestration

We'll implement the main function that calls each component, logs important information, and compiles the results.

import pandas as pd

def test_trading_strategy(tweet_ids):
    tweet_df = pd.DataFrame(
        fetch_tweets(tweet_ids), 
        columns=[
            "tweet_id", 
            "tweet_content", 
            "ground_truth_sentiment"
        ]
    )

    for _, tweet in tweet_df.iterrows():
        start_time = time.time()

        sentiment = call_sentiment_classifier(tweet)
        entity = call_entity_extractor(tweet)
        trade = recommend_trade(sentiment, entity['entities'])

        total_latency_ms = (time.time() - start_time) * 1000

        result = {
            "tweet_id": tweet["tweet_id"],
            "tweet_content": tweet["tweet_content"],
            **sentiment,
            **entity,
            **trade,
            "total_latency_ms": total_latency_ms
        }
        results.append(result)

    return pd.DataFrame(results)

# Run the trading strategy
results_df = test_trading_strategy(TWEET_IDS)

Step 4: Save Results as a Parquet File

We'll save the final results in a parquet file.

results_df.to_parquet('trading_strategy_results.parquet')

Test Results Creation

We have now shown how to simulate the trading environment by passing the tweets through the trading system, tracking each of the columns that are associated with the components of the system, and compiling the results into a parquet file.

There are 10 parquet files associated with this tutorial. The first file is the baseline run, which is the run that we will use to compare the other runs to.

In this section, we will show how to send the parquet file to dbnl. We will be using the dbnl SDK and Python 3.8 and above.

Create a Project, a Run Config, and Submit the Baseline Run

Create a Project

import os
from datetime import datetime

import dbnl

dbnl.login(api_token=os.environ['DBNL_API_TOKEN'])
PROJECT_NAME = 'Trading System Demo Project'
proj = dbnl.get_or_create_project(
    name=f"{PROJECT_NAME}_{datetime.now().strftime('%Y-%m-%d_%H-%M-%S')}",
    description="Trading System Demo Project Full"
)

View Project Trading System Demo Project_2024-05-21_08-05-55 at: https://app.dbnl.com/projects/proj_5BCxClOp6MujGKa6QQSmJt

Create a Run Configuration

import json

def read_json_file(file_path):
    with open(file_path, "r") as file:
        return json.load(file)

run_config = dbnl.create_run_config(
    project=proj, 
    **read_json_file("run_config.json")
)

Load the Baseline Run Results

We have organized the files such that the first alphabetically is the baseline.

import glob

filenames = iter(sorted(glob.glob('*.parquet')))
baseline_filename = next(filenames)

We can load this data into a pandas DataFrame.

import pandas as pd

test_data = pd.read_parquet(baseline_filename)
test_data.iloc[42]

tweet_content             Exciting times for $ENR in ENERGY, $UTIL in UT...
positive_sentiment                                                 0.279858
neutral_sentiment                                                   0.44201
negative_sentiment                                                 0.278132
ground_truth_sentiment                                   positive_sentiment
sentiment_latency_ms                                             144.190751
entities                                      [ENR, UTIL, COMS, MATR, FINS]
entity_latency_ms                                                  7.506861
buy_rec                                                                  []
sell_rec                                      [ENR, UTIL, COMS, MATR, FINS]
trade_latency_ms                                                  27.112794
total_latency_ms                                                 178.810406
Name: 42, dtype: object

Create a new run and submit the baseline test results.

def get_run_name(filename):
    run_name = filename.replace(".parquet", "")
    return run_name

run = dbnl.create_run(
    project=proj,
    display_name=get_run_name(baseline_filename),
    run_config=run_config,
    metadata=test_data.attrs,
)
dbnl.report_results(run=run, data=test_data.reset_index())
dbnl.close_run(run=run)

Run closed. View results at: https://app.dbnl.com/projects/proj_5BCxClOp6MujGKa6QQSmJt/runs/run_2tIHmtxDycYmkfzkYbbfZE

View the DAG of the components and the columns associated with each component

Consider spending some time within the Baseline Run. Notice when clicking on the components, only the columns associated with each component are displayed.

Set this Run as the Baseline

We can set this run we have just created as the baseline for all future tests. Now, when future runs are created, they will automatically be compared to this run whenever a test references the baseline.

dbnl.experimental.set_run_as_baseline(run=run)

This action can also be taken at the Test Configuration page.

In this section, we will create tests that compare the baseline run to the other runs. Rather than using the website to create the tests, we will use the dbnl sdk to do so; the web UI is quite useful for iterating on test design based on insights from dbnl runs. We have prepared a test configuration file that defines the tests that we will create.

Test Configuration

There are 14 tests in total and we show one below.

Example Test Payload (Soon to be deprecated schema)

Below is an example of a test payload which references the baseline run. This payload will eventually need to be hydrated with the proper project id as well as created tag ids. For now, we demonstrate the syntax of creating a test (to be described in the SentimentClassifier section right below).

test_payloads = read_json_file("test_payloads.json")
print(f"There are {len(test_payloads)} test payloads")
print(json.dumps(test_payloads[3], indent=2))

{
    "name": "scaled_ks_stat__negative_sentiment",
    "description": "This is a discrepancy test for negative_sentiment.",
    "tag_names": [
        "SentimentClassifier"
    ],
    "assertion": {
        "name": "less_than",
        "params": {
            "other": 0.1
        }
    },
    "statistic_name": "scaled_ks_stat",
    "statistic_params": {},
    "statistic_inputs": [
        {
            "select_query_template": {
                "select": "{EXPERIMENT}.negative_sentiment"
            }
        },
        {
            "select_query_template": {
                "select": "{BASELINE}.negative_sentiment"
            }
        }
    ]
}

What is being Tested?

These are all tests of consistency. We are testing that the system is behaving consistently across runs. We are actually not testing the system's performance against ground truth data. We are testing the system's performance against itself at a previous point in time. By holding the data constant, we can isolate the effects of the system's performance.

Sentiment Classifier

We write tests on consistency of the distributions of the probabilities of the sentiment classifier. These return a scaled version of the Kolmorogov-Smirnov statistic. We set a threshold about how large this statistic can be before we consider the test to have failed. These tests are all being compared to the corresponding distribution of the baseline run.

Entity Extractor

We write tests on the consistency of the distribution of the number of entities extracted. These return a scaled version of the chi-squared statistic. Similarly, we set a threshold about how large this statistic can be before we consider the test to have failed. These tests are all being compared to the corresponding distribution of the baseline run.

Purpose

By running these tests across every reasonable component of this system, we can isolate the components that are causing the system to behave differently in the event of a failure. By superimposing the results of these tests onto a DAG, we can see the path of the failure and the components that are causing the failure.

A Note On Thresholds

The thresholds are set by the user. This takes experience of seeing the system behave. Distributional has Forward-Deployed Engineers who are experienced in setting these thresholds; additionally we have plans to automate this process in our roadmap.

Submit Tests to DBNL

for test_payload in test_payloads:
    test_spec_dict = dbnl.experimental.prepare_incomplete_test_spec_payload(
        test_spec_dict=test_payload, 
        project_id=proj.id
    )
    dbnl.experimental.create_test(test_spec_dict=test_spec_dict)