From 46200fc391cf282bab1952c1304ad0958ed37953 Mon Sep 17 00:00:00 2001 From: Tyler Beason Date: Mon, 22 Jun 2026 14:20:55 -0500 Subject: [PATCH] Add Live symbol resolution (instrument_id -> symbol) Live data records carry only the opaque numeric hd.instrument_id; the gateway announces the human-readable symbol out-of-band via SymbolMappingMsg. The Live client now keeps a running instrument_id -> SymbolMappingMsg map, maintained by the reader as those control records arrive. - symbol_for(client, instrument_id_or_record; which=:in) -> resolve a record's symbol (which=:out for the output symbol); returns nothing if unseen. - symbol_map(client; which=:in) -> thread-safe snapshot Dict{UInt32,String}. - add_symbol_mapping_callback(client, cb) -> cb(instrument_id, msg) on each (re)mapping (lock-protected, errors logged-and-swallowed, mirrors add_reconnect_callback). symbol_for/symbol_map extend the DBN functions of the same name (the historical side keys on metadata + a timestamp; the live map is current-state, so no timestamp). Works in both typed and untyped mode and survives reconnects. In typed mode the mapping is captured before the control_channel overflow-drop, so it stays correct even when the control channel isn't drained. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 11 ++++ docs/src/api/live.md | 13 +++++ src/DatabentoAPI.jl | 8 ++- src/live/client.jl | 15 +++++ src/live/reader.jl | 7 +++ src/live/symbols.jl | 115 ++++++++++++++++++++++++++++++++++++++ test/runtests.jl | 1 + test/test_live_symbols.jl | 87 ++++++++++++++++++++++++++++ 8 files changed, 256 insertions(+), 1 deletion(-) create mode 100644 src/live/symbols.jl create mode 100644 test/test_live_symbols.jl diff --git a/CHANGELOG.md b/CHANGELOG.md index 222393a..637a64a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- **Live symbol resolution.** A `Live` client now keeps a running + `instrument_id → symbol` map, built by the reader from the gateway's + `SymbolMappingMsg` records. `symbol_for(client, instrument_id_or_record; + which=:in)` resolves the symbol for a data record (`which=:out` for the output + symbol), `symbol_map(client; which=:in)` returns a snapshot, and + `add_symbol_mapping_callback(client, cb)` fires `cb(instrument_id, msg)` on each + (re)mapping. Works in both typed and untyped mode and survives reconnects; in + typed mode the mapping is captured before any `control_channel` overflow drop, + so it's reliable even without draining the control channel. + ## [0.3.0] - 2026-06-22 ### Added diff --git a/docs/src/api/live.md b/docs/src/api/live.md index 3fa0be3..3fa7271 100644 --- a/docs/src/api/live.md +++ b/docs/src/api/live.md @@ -38,3 +38,16 @@ live_session ```@docs add_reconnect_callback ``` + +## Symbol resolution + +Live records carry only the numeric `instrument_id`; the reader maintains a +running `instrument_id → symbol` map from the gateway's `SymbolMappingMsg` +records (in both typed and untyped mode, without needing to drain +`control_channel`). + +```@docs +symbol_for(::Live, ::Integer) +symbol_map(::Live) +add_symbol_mapping_callback +``` diff --git a/src/DatabentoAPI.jl b/src/DatabentoAPI.jl index a3253d9..85a1853 100644 --- a/src/DatabentoAPI.jl +++ b/src/DatabentoAPI.jl @@ -52,7 +52,9 @@ using URIs import DatabentoBinaryEncoding as DBN using DatabentoBinaryEncoding: Schema, SType, Compression, Encoding, Action, Side, InstrumentClass -using DatabentoBinaryEncoding: symbol_map, symbol_for, add_symbol_column! +# `import` (not `using`) so the Live client can add `symbol_map`/`symbol_for` +# methods (extending the DBN functions) for live, current-state resolution. +import DatabentoBinaryEncoding: symbol_map, symbol_for, add_symbol_column! include("errors.jl") include("enums.jl") @@ -76,6 +78,7 @@ include("live/subscribe.jl") include("live/heartbeat.jl") include("live/streaming.jl") include("live/reconnect.jl") +include("live/symbols.jl") # Re-exported DatabentoBinaryEncoding enums (single import for users) export Schema, SType, Compression, Encoding, Action, Side, InstrumentClass @@ -121,4 +124,7 @@ export open_dbn_writer, write_record! # Live reconnect callback export add_reconnect_callback +# Live symbol resolution (symbol_map / symbol_for also re-exported above) +export add_symbol_mapping_callback + end # module DatabentoAPI diff --git a/src/live/client.jl b/src/live/client.jl index b399005..72e5475 100644 --- a/src/live/client.jl +++ b/src/live/client.jl @@ -98,6 +98,18 @@ mutable struct Live # gap_end in the reconnect callback. last_metadata_start_ns::Union{Nothing,Int64} + # Running instrument_id → SymbolMappingMsg map, built by the reader as + # SymbolMappingMsg records arrive. Captured in the reader *before* any + # control-channel drop, so it stays reliable even when control_channel + # overflows. Owned by the client, so it survives reconnects (the gateway + # re-sends mappings on re-subscribe). `symbols_lock` guards it: the reader + # task writes, consumer tasks read. The full record is stored so callers + # can resolve either the input or output symbol (and its [start,end] window). + symbols_by_id::Dict{UInt32,DBN.SymbolMappingMsg} + symbols_lock::ReentrantLock + # Callbacks fired on each SymbolMappingMsg: `cb(instrument_id, msg)`. + symbol_callbacks::Vector{Any} + # Reconnect policy + retry budget. reconnect_policy::ReconnectPolicy.T max_reconnect_attempts::Union{Int,Nothing} @@ -188,6 +200,9 @@ function Live(key::Union{Nothing,AbstractString} = nothing; # --- reconnect fields --- Dict{UInt32,Int64}(), Dict{UInt32,Int64}(), nothing, # last_metadata_start_ns + Dict{UInt32,DBN.SymbolMappingMsg}(), # symbols_by_id + ReentrantLock(), # symbols_lock + Any[], # symbol_callbacks rp, max_reconnect_attempts === nothing ? nothing : Int(max_reconnect_attempts), Int(immediate_reconnect_attempts), diff --git a/src/live/reader.jl b/src/live/reader.jl index 84c2a6e..e3a4a17 100644 --- a/src/live/reader.jl +++ b/src/live/reader.jl @@ -295,6 +295,10 @@ function _reader_loop_typed(c::Live) c.terminal_error = "gateway ErrorMsg" end end + # Update the symbol map before the drop decision below, so it + # stays correct even when control_channel overflows (a parent/ + # continuous subscription floods SymbolMappingMsg at start). + rec === nothing || _record_symbol!(c, rec) if rec !== nothing && ctrl_chan !== nothing && isopen(ctrl_chan) # Non-blocking put. This reader is the SOLE producer to # ctrl_chan (reconnect spawns readers sequentially, never @@ -442,6 +446,9 @@ function _reader_loop(c::Live) # reconnect supervisor sees timestamps from records that haven't # been consumed yet. _record_replay!(c, rec) + # Maintain the running instrument_id → symbol map (no-op unless the + # record is a SymbolMappingMsg). + _record_symbol!(c, rec) # ErrorMsg sets c.terminal_error *before* the put! so the # supervisor sees it once the reader exits and refuses to # reconnect — gateway-side errors are deterministic and diff --git a/src/live/symbols.jl b/src/live/symbols.jl new file mode 100644 index 0000000..b80e671 --- /dev/null +++ b/src/live/symbols.jl @@ -0,0 +1,115 @@ +# Live symbol resolution. +# +# Live data records carry only the opaque numeric `hd.instrument_id`. The gateway +# announces the human-readable symbol for each id out-of-band via +# `SymbolMappingMsg` records (a burst at subscription start, then occasional +# updates — e.g. a continuous/parent symbol rolling to a new contract). This +# layer keeps a running `instrument_id → SymbolMappingMsg` map on the `Live` +# client so consumers can resolve a record's symbol without manually tracking +# those control records. +# +# This is the live counterpart to the historical/DBN side, where the mapping is +# read once from `Metadata.mappings`. The difference: the historical map is +# interval-keyed (an id is reused across a date range, so `symbol_for` there +# takes a timestamp), whereas the live map is *current state* — it reflects the +# latest mapping seen, so the live `symbol_for` needs no timestamp. + +# Called by both reader loops on every control record; updates the map (and +# fires callbacks) only for SymbolMappingMsg. Invoked before the typed reader's +# control-channel drop logic, so the map is captured even if control_channel is +# full and the record itself is dropped. +@inline function _record_symbol!(c::Live, rec) + rec isa DBN.SymbolMappingMsg || return nothing + iid = rec.hd.instrument_id + @lock c.symbols_lock c.symbols_by_id[iid] = rec + _fire_symbol_callbacks(c, iid, rec) + return nothing +end + +# Snapshot callbacks under the lock, then fire outside it so a slow callback +# can't block the reader (mirrors _fire_reconnect_callbacks). +function _fire_symbol_callbacks(c::Live, iid::UInt32, rec::DBN.SymbolMappingMsg) + isempty(c.symbol_callbacks) && return nothing + cbs = @lock c.symbols_lock copy(c.symbol_callbacks) + for cb in cbs + try + cb(iid, rec) + catch e + @error "symbol-mapping callback raised" dataset = c.dataset exception = (e, catch_backtrace()) + end + end + return nothing +end + +# Pull the requested symbol string out of a SymbolMappingMsg. +@inline function _symbol_field(msg::DBN.SymbolMappingMsg, which::Symbol) + which === :in && return msg.stype_in_symbol + which === :out && return msg.stype_out_symbol + throw(ArgumentError("`which` must be :in or :out, got $(repr(which))")) +end + +""" + symbol_for(client::Live, instrument_id; which=:in) -> Union{String,Nothing} + symbol_for(client::Live, record; which=:in) -> Union{String,Nothing} + +Resolve the human-readable symbol the gateway last mapped to `instrument_id` +(or to `record.hd.instrument_id`), or `nothing` if no `SymbolMappingMsg` has been +seen for that id yet. The map is maintained automatically by the reader as +mappings stream in, so this works in both typed and untyped mode and without +draining `control_channel`. + +`which` selects which side of the mapping to return: +- `:in` (default) — the input symbol you subscribed with (e.g. `"AAPL"` for a + raw-symbol subscription, or the parent/continuous symbol such as `"ES.FUT"`). +- `:out` — the resolved output symbol (`stype_out_symbol`). + +See also [`symbol_map`](@ref) for a full snapshot and +[`add_symbol_mapping_callback`](@ref) for push-style updates. +""" +function symbol_for(c::Live, instrument_id::Integer; which::Symbol = :in) + msg = @lock c.symbols_lock get(c.symbols_by_id, UInt32(instrument_id), nothing) + return msg === nothing ? nothing : _symbol_field(msg, which) +end + +symbol_for(c::Live, rec; which::Symbol = :in) = + symbol_for(c, rec.hd.instrument_id; which = which) + +""" + symbol_map(client::Live; which=:in) -> Dict{UInt32,String} + +Return a thread-safe snapshot of the current `instrument_id → symbol` map for a +live `client`, built from the `SymbolMappingMsg` records seen so far. `which` +(`:in` default, or `:out`) selects which side of each mapping to use — see +[`symbol_for`](@ref). + +The returned `Dict` is a copy; mutating it does not affect the client, and the +client may add entries after the snapshot is taken. +""" +function symbol_map(c::Live; which::Symbol = :in) + out = Dict{UInt32,String}() + @lock c.symbols_lock begin + sizehint!(out, length(c.symbols_by_id)) + for (iid, msg) in c.symbols_by_id + out[iid] = _symbol_field(msg, which) + end + end + return out +end + +""" + add_symbol_mapping_callback(client::Live, cb) + +Register `cb(instrument_id::UInt32, msg::SymbolMappingMsg)` to be invoked each +time the gateway maps (or remaps) an instrument id — useful for reacting to +continuous/parent symbol rolls or persisting the mapping. Callbacks fire from +the reader task; they are snapshotted under a lock and run outside it, and an +exception in a callback is logged and swallowed (it won't disrupt the reader or +other callbacks). The map itself is updated before callbacks fire, so +[`symbol_for`](@ref) already reflects the new mapping inside the callback. +""" +function add_symbol_mapping_callback(c::Live, cb) + cb isa Function || cb isa Base.Callable || throw(ArgumentError( + "symbol-mapping callback must be callable, got $(typeof(cb))")) + @lock c.symbols_lock push!(c.symbol_callbacks, cb) + return nothing +end diff --git a/test/runtests.jl b/test/runtests.jl index c5c6921..f4e4595 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -26,6 +26,7 @@ try include("test_reconnect_backoff.jl") include("test_live_reconnect_supervisor.jl") include("test_typed_mode_errors.jl") + include("test_live_symbols.jl") end catch e global offline_failed = true diff --git a/test/test_live_symbols.jl b/test/test_live_symbols.jl new file mode 100644 index 0000000..3dfcaae --- /dev/null +++ b/test/test_live_symbols.jl @@ -0,0 +1,87 @@ +# Live symbol resolution: the running instrument_id → symbol map maintained by +# the reader, plus symbol_for / symbol_map / add_symbol_mapping_callback. +# +# These exercise the map directly via the reader hook `_record_symbol!` (no TCP +# needed) — the reader-loop integration is covered by the live reader tests. + +@testset "Live symbol resolution" begin + _smm(iid, insym, outsym) = DBN.SymbolMappingMsg( + DBN.RecordHeader(UInt8(0), DBN.RType.SYMBOL_MAPPING_MSG, UInt16(1), UInt32(iid), Int64(0)), + DBN.SType.PARENT, insym, DBN.SType.INSTRUMENT_ID, outsym, Int64(0), Int64(0)) + + _trade(iid) = DBN.TradeMsg( + DBN.RecordHeader(UInt8(0), DBN.RType.MBP_0_MSG, UInt16(1), UInt32(iid), Int64(0)), + Int64(1), UInt32(1), DBN.Action.TRADE, DBN.Side.BID, 0x00, 0x00, Int64(0), Int32(0), UInt32(1)) + + newclient() = Live("db-1234567890abcdef12345"; dataset = "GLBX.MDP3") + + @testset "empty map" begin + c = newclient() + @test symbol_for(c, 42) === nothing + @test isempty(symbol_map(c)) + end + + @testset "record + resolve (in/out)" begin + c = newclient() + DatabentoAPI._record_symbol!(c, _smm(42, "ES.FUT", "ESH4")) + DatabentoAPI._record_symbol!(c, _smm(99, "AAPL", "12345")) + # non-mapping records are a no-op + DatabentoAPI._record_symbol!(c, _trade(7)) + + @test symbol_for(c, 42) == "ES.FUT" # :in default + @test symbol_for(c, 42; which = :out) == "ESH4" + @test symbol_for(c, 99) == "AAPL" + @test symbol_for(c, 7) === nothing # data record didn't register + @test symbol_for(c, 123) === nothing # unknown id + + # record overload resolves via hd.instrument_id + @test symbol_for(c, _trade(42)) == "ES.FUT" + + @test symbol_map(c) == Dict(UInt32(42) => "ES.FUT", UInt32(99) => "AAPL") + @test symbol_map(c; which = :out) == Dict(UInt32(42) => "ESH4", UInt32(99) => "12345") + + @test_throws ArgumentError symbol_for(c, 42; which = :bogus) + end + + @testset "remap overwrites (e.g. continuous roll)" begin + c = newclient() + DatabentoAPI._record_symbol!(c, _smm(42, "ES.c.0", "ESH4")) + DatabentoAPI._record_symbol!(c, _smm(42, "ES.c.0", "ESM4")) # rolled + @test symbol_for(c, 42; which = :out) == "ESM4" + @test length(symbol_map(c)) == 1 + end + + @testset "snapshot is a copy" begin + c = newclient() + DatabentoAPI._record_symbol!(c, _smm(1, "A", "1")) + snap = symbol_map(c) + DatabentoAPI._record_symbol!(c, _smm(2, "B", "2")) + @test snap == Dict(UInt32(1) => "A") # unaffected by later mapping + @test length(symbol_map(c)) == 2 + end + + @testset "callbacks" begin + c = newclient() + seen = Tuple{UInt32,String}[] + add_symbol_mapping_callback(c, (iid, msg) -> push!(seen, (iid, msg.stype_in_symbol))) + # map is already updated when the callback fires + add_symbol_mapping_callback(c, (iid, _) -> push!(seen, (iid, something(symbol_for(c, iid), "?")))) + + DatabentoAPI._record_symbol!(c, _smm(42, "ES.FUT", "ESH4")) + @test (UInt32(42), "ES.FUT") in seen + @test count(==((UInt32(42), "ES.FUT")), seen) == 2 # both callbacks saw it + + @test_throws ArgumentError add_symbol_mapping_callback(c, 123) + end + + @testset "a raising callback is swallowed" begin + c = newclient() + add_symbol_mapping_callback(c, (_, _) -> error("boom")) + ok = Ref(false) + add_symbol_mapping_callback(c, (_, _) -> ok[] = true) + # should not throw, and the second callback still runs + @test (DatabentoAPI._record_symbol!(c, _smm(1, "A", "1")); true) + @test ok[] + @test symbol_for(c, 1) == "A" + end +end