Capture — API Reference

stream_to_file(; schema, symbols, dataset, ...) -> String

Capture a single Live schema to a single (optionally rotating) DBN file. Returns the path to the most recently opened output file.

The default compress_level = 1 favours throughput; bench measurements show L1 is ~35% faster on the write boundary than L3 in exchange for ~15% larger files (see benchmark/PERF_REPORT.md). Pass a higher level for archival.

reconnect = true (default) re-establishes the TCP session after a drop using full-jitter exponential backoff (1s base, 60s cap). After max_reconnect_attempts retries (default 10) the loop gives up and returns; pass nothing for unlimited.

frame_seconds = 60.0 (default) closes the active zstd frame and starts a new one every minute, so a hard kill (SIGKILL, OOM, power loss) loses at most one frame of records instead of corrupting the whole file. Pass nothing to write a single frame. The output is a standards-compliant multi-frame .dbn.zst and reads back as one continuous record stream via any zstd reader.

Press Ctrl-C for a clean shutdown — the active zstd frame footer is flushed before the function returns.

stream_multi_to_files(; schemas, symbols, dataset, ...) -> Dict{Schema.T,String}

Capture N Live schemas concurrently, one rotating-DBN file per schema. Returns a dict mapping each schema to its most recently opened output path. Press Ctrl-C to stop early; honours duration_s if provided.

reconnect/max_reconnect_attempts work as in stream_to_file: full-jitter exponential backoff between retries (1s base, 60s cap), default cap of 10 attempts, pass nothing for unlimited.

frame_seconds = 60.0 (default) writes a fresh zstd frame every minute on each per-schema file for crash safety (multi-frame .dbn.zst; reads back as one continuous stream). Pass nothing to write a single frame.

Example:

paths = stream_multi_to_files(
    dataset  = "OPRA.PILLAR",
    schemas  = [Schema.TCBBO, Schema.CBBO_1S, Schema.STATUS],
    symbols  = ["SPXW.OPT"],
    stype_in = SType.PARENT,
    duration_s = 60,
)

open_dbn_writer(f::Function;
                base_dir, dataset, schema, symbols, stype_in,
                compress = true, compress_level = 1,
                rotate_seconds = nothing,
                frame_seconds = 60.0,
                explicit_path = nothing)

Open a DBN file writer, run f(writer), and guarantee the file is flushed and closed in finally — so on InterruptException (Ctrl-C), an exception, or normal exit the zstd frame footer is emitted and current_path is a well-formed .dbn.zst.

Use this when iterating records from a Live client (or any other source) and writing them to a Databento Binary Encoding file with the same durability features stream_to_file uses internally:

• Optional time-based file rotation via rotate_seconds (close the current file, open the next one with a fresh timestamp).
• In-file zstd frame rotation via frame_seconds (default 60s) for crash safety. A hard kill loses at most one frame of records instead of corrupting the whole file. Pass nothing to disable.

Inside the do-block, call write_record!(writer, rec) for each record. The writer also exposes writer.current_path if you need it.

Live(dataset = "GLBX.MDP3") do client
    connect!(client)
    subscribe!(client; schema = Schema.TRADES, symbols = ["ES.FUT"], stype_in = SType.PARENT)
    start!(client)
    open_dbn_writer(; base_dir = "./capture", dataset = "GLBX.MDP3",
                      schema = Schema.TRADES, symbols = ["ES.FUT"],
                      stype_in = SType.PARENT) do writer
        for rec in client
            write_record!(writer, rec)
        end
    end
end

Returns whatever f(writer) returns.

write_record!(writer, rec)

Write one DBN record to a writer opened via open_dbn_writer. Thread-safe: holds the writer's internal lock for the duration of the write so concurrent calls (e.g. from per-schema consumer tasks) cannot interleave bytes.

default_live_path(; dataset, schema, base_dir = joinpath(pwd(), "live"), compress = true)

Compute the default output path for a live capture session. Format: {base_dir}/{dataset}/{schema_str(schema)}/{utc-yyyymmddTHHMMSSZ}.dbn[.zst].

read_capture(path) -> (DBN.Metadata, Vector{DBN.DBNRecord})

Decode a .dbn or .dbn.zst file written by stream_to_file / stream_multi_to_files (or by a Databento batch download). Thin wrapper around DBN.read_dbn_with_metadata — kept as a convenience export so callers don't have to import DBN themselves.