Databento OHLCV-1d (Rust)

Cargo.toml

[package]
name = "databento-pg"
version = "0.1.0"
edition = "2021"

[dependencies]
databento = "0.12.0"
dotenv = "0.15.0"
futures = "0.3.30"
serde = { version = "1.0.205", features = ["derive"] }
serde_json = "1.0.122"
time = { version = "0.3.36", features = ["macros"] }
tokio = { version = "1.39.2", features = ["full"] }

clock.rs

//! Wanted to call this module `time` but that's already taken by the `time` crate. This module
//! will contain all the time-related utilities and functions.

use time::{macros::datetime, OffsetDateTime};

// Get the time::Date from an OffsetDateTime with nanosecond level precision
pub fn get_date_from_datetime(datetime: OffsetDateTime) -> time::Date {
    time::Date::from_ordinal_date(datetime.year(), datetime.ordinal()).unwrap()
}

main.rs

use std::str::FromStr;

use databento::{
    dbn::{OhlcvMsg, Record, SType, Schema},
    historical::{symbology::ResolveParams, timeseries::GetRangeParams},
    HistoricalClient,
};
use dotenv::dotenv;
use time::macros::{date, datetime};

mod clock;

pub const NASDAQ_DATASET_ID: &str = "XNAS.ITCH";
pub const OHLCV1D_SCHEMA_ID: &str = "ohlcv-1d";

#[tokio::main]
async fn main() {
    dotenv().ok();

    // let mut symbol_hashmap: HashMap<String, u32> = HashMap::new();

    let schema = Schema::from_str(OHLCV1D_SCHEMA_ID).expect("Failed to parse schema");

    let symbols = vec!["TSLA"];
    let datetime_range = (
        datetime!(2021-11-03 14:00 -4),
        datetime!(2021-12-03 16:00 -4),
    );
    let date_range = (date!(2021 - 11 - 03), date!(2021 - 12 - 03));

    // Symbol <-> InstrumentId resolution params
    let resolve_params = ResolveParams::builder()
        .dataset(NASDAQ_DATASET_ID)
        .symbols(symbols.clone())
        .stype_in(SType::RawSymbol)
        .stype_out(SType::InstrumentId)
        .date_range(date_range)
        .build();

    // Symbol -> TimeSeries obj params
    let get_range_params = GetRangeParams::builder()
        .dataset(NASDAQ_DATASET_ID)
        .date_time_range(datetime_range)
        .symbols(symbols.clone())
        .schema(schema)
        .stype_in(SType::RawSymbol)
        .stype_out(SType::InstrumentId)
        .limit(None)
        .build();

    let mut client = HistoricalClient::builder()
        .key_from_env()
        .expect("DATABENTO_API_KEY must be set")
        .build()
        .expect("Failed to build databento HTTP client");

    let mut sym_client = client.symbology();
    let resolution = sym_client.resolve(&resolve_params).await.unwrap();
    let ts_symbol_map = resolution.symbol_map().unwrap();

    let mut response = client
        .timeseries()
        .get_range(&get_range_params)
        .await
        .unwrap();

    while let Some(msg) = response.decode_record::<OhlcvMsg>().await.unwrap() {
        let header = msg.header();
        let timestamp = header.ts_event().expect("No timestamp found");

        let date_from_ts = clock::get_date_from_datetime(timestamp);
        let instrument_id = header.instrument_id;

        let symbol = ts_symbol_map.get(date_from_ts, instrument_id);

        if symbol.is_none() {
            println!("No symbol found for instrument_id: {}", instrument_id);
            println!("Date: {}", date_from_ts);
            continue;
        }

        let symbol = symbol.unwrap();
        let closing_price = msg.close;

        println!(
            "Symbol: {},\tDate: {},\tClose: {}",
            symbol, date_from_ts, closing_price
        );
    }
}