Crypto made easy: import live trade data

The fascination with cryptocurrency is going strong, especially after Elon Musk fanned the flames by voicing his preference for Dogecoin over Bitcoin. Cryptocurrency dominated trending Google searches in 2021, with Dogecoin and Ethereum making the top 10 list of most popular searches. If you have more than a casual interest, Deephaven provides essential tools to get you started. We'll show you how to use crypto data to make informed choices on entering and leaving the crypto market.

Earlier this year, we wrote a program to pull live and historical crypto prices. We stream data from CoinGecko and use that data to know when the current price is high or lower than a value based on historical data.

In this iteration, we compute exponential moving averages (EMAs), standard deviations, and averages on the data to help you choose when to buy and sell your favorite crypto.

To get started, launch Deephaven (see our quick start) and run our original code, included below. You'll need the appropriate import statements for working with real-time data.

Python code for our Crypto program

To use CoinGecko, we need to install the PyCoinGecko library. See How to install Python packages for the different ways to do this. In this case, we install it for a one-time use.

import os

os.system("pip install pycoingecko")

from deephaven import DynamicTableWriter
from deephaven import pandas as dhpd
from deephaven import dtypes as dht
from deephaven import time as dhtu
from deephaven import merge

from pycoingecko import CoinGeckoAPI
from time import sleep, time
import pandas as pd
import threading

# seconds_to_sleep should be 10 or higher. If too fast, will hit request limit.
seconds_to_sleep = 60

# if get_history = true, the days to pull
days_history = 2

# coins to get data
ids = [
    "bitcoin",
    "ethereum",
    "litecoin",
    "dogecoin",
    "tether",
    "binancecoin",
    "cardano",
    "ripple",
    "polkadot",
]

# Set one to True to get historical or live data
get_history = False
get_live = False


########## The code below need not be changed ##########
def dt_from_ms(milliseconds):
    return dhtu.millis_to_datetime(milliseconds)


def get_coingecko_table_historical(days_history=90):
    table_array = []
    cg = CoinGeckoAPI()

    coingecko_coldefs = {
        "Timestamp": dht.DateTime,
        "Coin": dht.string,
        "Price": dht.double,
        "MarketCap": dht.double,
        "TotalVolume": dht.double,
    }
    dtw = DynamicTableWriter(coingecko_coldefs)
    coin_hist_table = dtw.table

    # get historical data
    for id in ids:
        coin_data_hist = cg.get_coin_market_chart_by_id(
            id, vs_currency="usd", days=days_history
        )
        n_hist = len(coin_data_hist["prices"])
        for idx in range(n_hist):
            dt = dt_from_ms(coin_data_hist["prices"][idx][0])
            price = coin_data_hist["prices"][idx][1]
            market_cap = coin_data_hist["market_caps"][idx][1]
            volume = coin_data_hist["total_volumes"][idx][1]
            dtw.write_row(dt, id, price, market_cap, volume)
        table_array.append(coin_hist_table)
    return (
        merge(table_array)
        .select_distinct(["Timestamp", "Coin", "Price", "MarketCap", "TotalVolume"])
        .sort_descending(["Timestamp", "Coin"])
        .format_columns("Price = Decimal(`#,###.############`)")
    )


def get_coingecko_table_live(seconds_to_sleep=60):
    # array to store tables for current and previous data
    table_array = []

    dtw_col_defs = {
        "Coin": dht.string,
        "Timestamp": dht.DateTime,
        "Price": dht.double,
        "MarketCap": dht.double,
        "TotalVolume": dht.double,
    }
    table_writer = DynamicTableWriter(dtw_col_defs)
    live_crypto = table_writer.table

    def thread_func():
        cg = CoinGeckoAPI()

        while True:
            cg.get_coins_markets(vs_currency="usd")
            coin_data = cg.get_price(
                ids,
                vs_currencies="usd",
                include_market_cap=True,
                include_24hr_vol=True,
                include_24hr_change=False,
                include_last_updated_at=True,
            )
            for id in ids:
                # Write a row of data to the table
                table_writer.write_row(
                    id,
                    dhtu.millis_to_datetime(
                        int(coin_data[id]["last_updated_at"]) * 1000
                    ),
                    float(coin_data[id]["usd"]),
                    coin_data[id]["usd_market_cap"],
                    coin_data[id]["usd_24h_vol"],
                )

            sleep(seconds_to_sleep)

    thread = threading.Thread(target=thread_func)
    thread.start()

    table_array.append(live_crypto)
    result = (
        merge(table_array)
        .select_distinct(["Timestamp", "Coin", "Price", "MarketCap", "TotalVolume"])
        .sort_descending(["Timestamp", "Coin"])
        .format_columns("Price = Decimal(`#,###.############`)")
    )

    return result


historical_crypto_table = None
live_crypto_table = None

if get_history:
    historical_crypto_table = get_coingecko_table_historical(days_history)

if get_live:
    live_crypto_table = get_coingecko_table_live(seconds_to_sleep)

Now the fun part. Let's do some analysis on the coins. First, let's see what assets we can pull. This code gets a list of all IDs currently in CoinGecko and save them as a list to filter.

import json
import requests

coins_available = []

headers = {
    "accept": "application/json",
}
response = requests.get("https://api.coingecko.com/api/v3/coins/list", headers=headers)
resp = json.loads(response.content)

for d in resp:
    for key in d.keys():
        if key == "id":
            coins_available.append(d[key])

print(f"CoinGecko has data on {len(coins_available)} coins.")

That is A LOT of assets so it's to limit them to what we actually want; in this case, we'll filter the assets to only those that contain the string doge.

import re

r = re.compile("^.*doge.*$")
ids = list(filter(r.match, coins_available))[:25]

We end up with over 200 assets without trimming. For ease, I'm going to create a targeted list of IDs to work with - the first 25 values. The code below recreates live_crypto_table and historical_crypto_table. Remember there is a pull limit, so only run historical_crypto_table once - if you set get_history to True previously, set it to False in the code below.

get_history = True

get_live = True

if get_history:
    historical_crypto_table = get_coingecko_table_historical(days_history)

if get_live:
    live_crypto_table = get_coingecko_table_live(seconds_to_sleep)

We can even merge the data for a seamless table:

combined_crypto = merge([live_crypto_table, historical_crypto_table])

The steps above collect and organize our data. Now, we can perform EMAs.

Here we add a column that contains our EMA value and highlights if the current price is above that value. EMAs are familiar tools to stock investors and allow us to see how the asset has changed with time. EMAs keep track of the average with the newest values having more weight than older values. This rolling window of averages can give us insight into what the crypto asset might do next. In this EMA, we set the window to be just 10 minutes - like everything in this code example, you can fine-tune this value to fit your data needs.

from deephaven.experimental.ema import ByEmaSimple

ema1sec = ByEmaSimple("BD_SKIP", "BD_SKIP", "TIME", 10, "MINUTES", type="LEVEL")

ema_data = combined_crypto.update(
    formulas=["EMA_data = ema1sec.update(Timestamp, Price)"]
).last_by(by=["Coin"])
ema_data_grouped = (
    combined_crypto.update(
        formulas=["EMA_data = ema1sec.update(Timestamp, Price, Coin)"]
    )
    .last_by(by=["Coin"])
    .sort(order_by=["Coin"])
    .format_columns(
        [
            "Price = Decimal(`#,###.############`)",
            "EMA_data = Decimal(`#,###.############`)",
        ]
    )
    .format_column_where("Coin", "EMA_data < Price", "IVORY")
)

You can easily create different calculations, such as standard deviations and averages, by modifying the code:

from deephaven import agg as agg

agg_list = [agg.std(cols=["Std_price = Price"]), agg.avg(cols=["Avg_price = Price"])]

std_data = (
    ema_data_grouped.join(
        table=combined_crypto.agg_by(agg_list, by=["Coin"]), on=["Coin"]
    )
    .format_columns(
        [
            "Std_price = Decimal(`#,###.############`)",
            "Avg_price = Decimal(`#,###.############`)",
        ]
    )
    .format_column_where("Coin", " Price  < Avg_price - Std_price", "IVORY")
)

This code provides a basic starter. Change the crypto assets to monitor your personal favorites, or switch out the calculations to suit your needs. Deephaven enables you to make smart choices about entering and leaving the market. With the speed of data changes in the crypto world, it's important to use the fastest real-time data engine possible: Deephaven.

We hope this program inspires you. If you make something of your own or have an idea to share, we'd love to hear about it on Slack!

Are you crypto-curious?