scid_intraday_record_struct.py

from struct import calcsize
import polars as pl

def scid_parse_trades(filepath: str) -> pl.DataFrame:
    """Sierrachart's SCID format. Must be configured to record single trade ticks."""
    intraday_record_struct = np.dtype(
        [
            ("timestamp", "<q"),
            ("open", "<f"),
            ("ask_price", "<f"),
            ("bid_price", "<f"),
            ("trade_price", "<f"),
            ("num_trades", "<I"),
            ("total_vol", "<I"),
            ("bid_vol", "<I"),
            ("ask_vol", "<I"),
        ]
    )
    
    """Constants from Taylor O'Brien's tick_db (https://github.com/toobrien/tick_db) """
    SC_EPOCH = np.datetime64("1899-12-30T00:00:00.000000") # datetime64[us]
    INTRADAY_HEADER_FMT  = "4cIIHHI36c"
    INTRADAY_HEADER_LEN  = calcsize(INTRADAY_HEADER_FMT)
    
    intraday_records = np.memmap(
        filepath, 
        offset=INTRADAY_HEADER_LEN,
        dtype=intraday_record_struct, 
        mode="readonly"
    )
    
    
    df = pl.DataFrame(
        {
            field_name: intraday_records[field_name]
            for field_name in intraday_records.dtype.fields
        }
    ).with_columns(
        timestamp=(pl.col("timestamp") + SC_EPOCH).cast(pl.Datetime(time_unit="us")),
        side=pl.col("bid_vol")==0
    ).drop(
        ["open"]
    )
    
    return df

filepath = "/data/CME-HO/HOF21_FUT_CME.scid"
scid_parse_trades(filepath)