diff --git a/CHANGELOG.md b/CHANGELOG.md index 6ebe65e..8f557d5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Symbol resolution helpers that join the human-readable `raw_symbol` back onto + records using `Metadata.mappings`: `symbol_map(metadata)` builds an + `instrument_id -> [(start_date, end_date, raw_symbol)]` lookup, + `symbol_for(map_or_metadata, instrument_id, ts_event)` resolves the symbol + valid at a record's timestamp, `add_symbol_column!(df, metadata)` joins a + `:symbol` column onto a records DataFrame in place, and + `records_to_dataframe(records, metadata; symbols=true)` does the conversion + + join in one call. The join keys on `instrument_id`/`ts_event` (so it also works + for the row-expanded MBP-10 frame) and only applies when the query was resolved + with `stype_out = SType.INSTRUMENT_ID`. + ### Fixed - `read_stat_msg` now sizes the record body from `hd.length` instead of assuming diff --git a/Project.toml b/Project.toml index 011c0fe..5cd676e 100644 --- a/Project.toml +++ b/Project.toml @@ -1,7 +1,7 @@ name = "DatabentoBinaryEncoding" uuid = "90689371-c8cb-40d1-831f-18033db90f74" authors = ["Tyler Beason "] -version = "0.1.3" +version = "0.1.4" [deps] CSV = "336ed68f-0bac-5ca0-87d4-7b16caf5d00b" diff --git a/docs/src/api/conversion.md b/docs/src/api/conversion.md index 0de5acd..844c3fb 100644 --- a/docs/src/api/conversion.md +++ b/docs/src/api/conversion.md @@ -11,6 +11,17 @@ dbn_to_parquet records_to_dataframe ``` +## Symbol Resolution + +Join the human-readable `raw_symbol` back onto records (which carry only the +opaque numeric `instrument_id`) using `Metadata.mappings`. + +```@docs +symbol_map +symbol_for +add_symbol_column! +``` + ## Import (Other Formats to DBN) ```@docs diff --git a/src/DatabentoBinaryEncoding.jl b/src/DatabentoBinaryEncoding.jl index 7767e3b..df20e3d 100644 --- a/src/DatabentoBinaryEncoding.jl +++ b/src/DatabentoBinaryEncoding.jl @@ -99,6 +99,7 @@ include("decode.jl") include("encode.jl") include("streaming.jl") include("export.jl") +include("symbols.jl") include("import.jl") # Exports @@ -123,6 +124,7 @@ export DBN_VERSION, FIXED_PRICE_SCALE, UNDEF_PRICE, UNDEF_ORDER_SIZE, UNDEF_TIME export BidAskPair, VersionUpgradePolicy, DatasetCondition export write_header, read_header!, write_record, read_record, finalize_encoder export dbn_to_csv, dbn_to_json, dbn_to_parquet, records_to_dataframe +export symbol_map, symbol_for, add_symbol_column! export json_to_dbn, parquet_to_dbn, csv_to_dbn end # module DatabentoBinaryEncoding \ No newline at end of file diff --git a/src/symbols.jl b/src/symbols.jl new file mode 100644 index 0000000..7bcac71 --- /dev/null +++ b/src/symbols.jl @@ -0,0 +1,121 @@ +# Symbol resolution: join the human-readable symbol back onto records. +# +# Every record carries only a numeric `hd.instrument_id` — opaque on its own. +# The mapping from that id to a tradable symbol lives in the file/stream +# `Metadata.mappings`: a flat list of +# (raw_symbol, mapped_symbol, start_date, end_date) +# tuples, one per validity interval (`start_date`/`end_date` are raw `YYYYMMDD` +# integers; the same `raw_symbol` appears in several consecutive tuples when an +# instrument rolls or is renamed). When the query was resolved with +# `stype_out = INSTRUMENT_ID` (the historical default), `mapped_symbol` is the +# decimal instrument-id string, so the table is really an interval-keyed +# `instrument_id -> raw_symbol` map. These helpers invert it for joining. + +""" + symbol_map(metadata) -> Dict{UInt32,Vector{Tuple{Int64,Int64,String}}} + +Build an `instrument_id -> [(start_date, end_date, raw_symbol), …]` lookup from +`metadata.mappings`, suitable for repeated [`symbol_for`](@ref) calls in a hot +loop (build once, reuse). Each value is the list of validity intervals for that +instrument id, sorted by `start_date`; dates are raw `YYYYMMDD` integers and the +interval is half-open `[start_date, end_date)`. + +The join is only meaningful when the query was resolved with +`stype_out = SType.INSTRUMENT_ID` — that is what makes the mapping's +`mapped_symbol` field the decimal instrument-id string that record headers carry. +For any other `stype_out` (or empty mappings) the returned dict is empty. +""" +function symbol_map(metadata::Metadata) + smap = Dict{UInt32,Vector{Tuple{Int64,Int64,String}}}() + # Records key on instrument_id; only an instrument-id-resolved query has a + # mapped_symbol we can parse back into that key. + metadata.stype_out == SType.INSTRUMENT_ID || return smap + for (raw, mapped, sd, ed) in metadata.mappings + isempty(mapped) && continue + id = tryparse(UInt32, mapped) + id === nothing && continue # defensive: skip a non-numeric mapped_symbol + push!(get!(() -> Tuple{Int64,Int64,String}[], smap, id), (sd, ed, raw)) + end + for intervals in values(smap) + sort!(intervals, by = first) + end + return smap +end + +# Nanoseconds-since-epoch (UTC) -> YYYYMMDD integer, matching the date encoding +# used in Metadata.mappings. Pure integer arithmetic — no float round-trip. +function _ts_ns_to_yyyymmdd(ts_ns::Integer) + days = fld(Int64(ts_ns), 86_400_000_000_000) + d = Date(1970, 1, 1) + Day(days) + return year(d) * 10000 + month(d) * 100 + day(d) +end + +""" + symbol_for(smap::AbstractDict, instrument_id, ts_event) -> Union{String,Nothing} + symbol_for(metadata::Metadata, instrument_id, ts_event) -> Union{String,Nothing} + +Resolve the `raw_symbol` for `instrument_id` valid at `ts_event` (nanoseconds +since the Unix epoch, i.e. a record's `hd.ts_event`). Returns `nothing` when the +id is unknown or no interval covers the timestamp's date. + +The `AbstractDict` form takes a prebuilt [`symbol_map`](@ref) and is the one to +use in a loop; the `Metadata` form rebuilds the map on every call (convenient for +one-off lookups, wasteful otherwise). +""" +function symbol_for(smap::AbstractDict, instrument_id::Integer, ts_event::Integer) + intervals = get(smap, UInt32(instrument_id), nothing) + intervals === nothing && return nothing + d = _ts_ns_to_yyyymmdd(ts_event) + for (sd, ed, raw) in intervals + # Half-open [sd, ed); a degenerate single-day interval can arrive with + # sd == ed, so accept that as a point match too. + (sd <= d < ed || (sd == ed && d == sd)) && return raw + end + return nothing +end + +symbol_for(metadata::Metadata, instrument_id::Integer, ts_event::Integer) = + symbol_for(symbol_map(metadata), instrument_id, ts_event) + +""" + add_symbol_column!(df, metadata; column=:symbol) -> df + +Join the human-readable `raw_symbol` onto an already-built records DataFrame, +in place, as a new `column` (default `:symbol`). The join keys on the frame's +`instrument_id` and `ts_event` columns, so it works for both flat schemas and +the row-expanded MBP-10 frame; unmatched rows get `missing`. + +A no-op (returns `df` unchanged) when the frame lacks `instrument_id`/`ts_event` +(e.g. an empty frame). When `metadata` has mappings but was resolved with a +`stype_out` other than `INSTRUMENT_ID` — so nothing can be joined — the column is +filled with `missing` and a single warning is emitted. +""" +function add_symbol_column!(df::DataFrame, metadata::Metadata; column::Symbol = :symbol) + (hasproperty(df, :instrument_id) && hasproperty(df, :ts_event)) || return df + smap = symbol_map(metadata) + if isempty(smap) + if !isempty(metadata.mappings) && metadata.stype_out != SType.INSTRUMENT_ID + @warn "add_symbol_column!: cannot join symbols — metadata.stype_out is not \ + INSTRUMENT_ID, so records' instrument_id has no mapping" stype_out = metadata.stype_out + end + df[!, column] = Vector{Union{Missing,String}}(missing, nrow(df)) + return df + end + df[!, column] = [something(symbol_for(smap, id, ts), missing) + for (id, ts) in zip(df.instrument_id, df.ts_event)] + return df +end + +""" + records_to_dataframe(records, metadata; symbols=true) + +Like [`records_to_dataframe`](@ref), but joins the human-readable `raw_symbol` +onto the result as a `:symbol` column using `metadata.mappings` (see +[`add_symbol_column!`](@ref)). Pass `symbols=false` to skip the join and get the +same frame as the one-argument form. +""" +function records_to_dataframe(records::Vector, metadata::Metadata; symbols::Bool = true) + df = records_to_dataframe(records) + symbols && add_symbol_column!(df, metadata) + return df +end diff --git a/test/runtests.jl b/test/runtests.jl index 5a7741b..fcfa7cb 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -22,6 +22,7 @@ include("test_utils.jl") include("test_issue23_unset_stype.jl") # regression: 0xFF unset stype in v3 SymbolMappingMsg (issue #23) include("test_issues_32_35.jl") # regressions: pre-v3 StatMsg layout, stat_to_dataframe, mapping intervals, invalid enum bytes (issues #32-#35) include("test_show.jl") # compact one-line Base.show for record types + include("test_symbols.jl") # symbol_map / symbol_for / add_symbol_column! / records_to_dataframe(records, metadata) # Run compatibility tests if the Rust CLI is available dbn_cli_path = if Sys.iswindows() diff --git a/test/test_symbols.jl b/test/test_symbols.jl new file mode 100644 index 0000000..a495283 --- /dev/null +++ b/test/test_symbols.jl @@ -0,0 +1,80 @@ +# Symbol resolution: symbol_map / symbol_for / add_symbol_column! / +# records_to_dataframe(records, metadata). + +@testset "Symbol resolution" begin + ts(d::Date) = DBN.datetime_to_ts(DateTime(d)) + + # AAPL rolls instrument id mid-range (101 -> 111); MSFT is stable (102). + mappings = [ + ("AAPL", "101", 20240102, 20240104), + ("AAPL", "111", 20240104, 20240106), + ("MSFT", "102", 20240102, 20240106), + ] + md = DBN.Metadata(3, "XNAS.ITCH", DBN.Schema.TRADES, 0, nothing, nothing, + DBN.SType.RAW_SYMBOL, DBN.SType.INSTRUMENT_ID, false, + ["AAPL", "MSFT"], String[], String[], mappings) + + @testset "symbol_map" begin + smap = symbol_map(md) + @test length(smap) == 3 + @test smap[UInt32(101)] == [(20240102, 20240104, "AAPL")] + @test smap[UInt32(111)] == [(20240104, 20240106, "AAPL")] + @test smap[UInt32(102)] == [(20240102, 20240106, "MSFT")] + end + + @testset "symbol_for (map and metadata forms)" begin + smap = symbol_map(md) + @test symbol_for(smap, 101, ts(Date(2024, 1, 3))) == "AAPL" + @test symbol_for(smap, 111, ts(Date(2024, 1, 5))) == "AAPL" # rolled id + @test symbol_for(smap, 102, ts(Date(2024, 1, 2))) == "MSFT" # inclusive start + @test symbol_for(smap, 101, ts(Date(2024, 1, 4))) === nothing # exclusive end + @test symbol_for(smap, 101, ts(Date(2024, 1, 5))) === nothing # out of interval + @test symbol_for(smap, 999, ts(Date(2024, 1, 3))) === nothing # unknown id + # metadata convenience form rebuilds the map internally + @test symbol_for(md, 102, ts(Date(2024, 1, 3))) == "MSFT" + end + + @testset "degenerate single-day interval (sd == ed)" begin + md_pt = DBN.Metadata(3, "X", DBN.Schema.TRADES, 0, nothing, nothing, + DBN.SType.RAW_SYMBOL, DBN.SType.INSTRUMENT_ID, false, + ["ZZ"], String[], String[], + [("ZZ", "7", 20240103, 20240103)]) + @test symbol_for(md_pt, 7, ts(Date(2024, 1, 3))) == "ZZ" + @test symbol_for(md_pt, 7, ts(Date(2024, 1, 4))) === nothing + end + + hdr(id, d) = DBN.RecordHeader(0, DBN.RType.MBP_0_MSG, 1, UInt32(id), ts(d)) + recs = [ + DBN.TradeMsg(hdr(101, Date(2024, 1, 3)), 100, 5, DBN.Action.TRADE, DBN.Side.BID, 0x00, 0, ts(Date(2024, 1, 3)), 0, 1), + DBN.TradeMsg(hdr(102, Date(2024, 1, 5)), 200, 7, DBN.Action.TRADE, DBN.Side.ASK, 0x00, 0, ts(Date(2024, 1, 5)), 0, 2), + DBN.TradeMsg(hdr(999, Date(2024, 1, 3)), 300, 9, DBN.Action.TRADE, DBN.Side.BID, 0x00, 0, ts(Date(2024, 1, 3)), 0, 3), + ] + + @testset "records_to_dataframe with symbols" begin + df = records_to_dataframe(recs, md) + @test isequal(df.symbol, ["AAPL", "MSFT", missing]) + # symbols=false matches the one-argument frame exactly (no :symbol col) + @test !hasproperty(records_to_dataframe(recs, md; symbols = false), :symbol) + end + + @testset "add_symbol_column! custom column name" begin + df = records_to_dataframe(recs) + add_symbol_column!(df, md; column = :ticker) + @test isequal(df.ticker, ["AAPL", "MSFT", missing]) + end + + @testset "non-INSTRUMENT_ID stype_out fills missing + warns" begin + md_raw = DBN.Metadata(3, "X", DBN.Schema.TRADES, 0, nothing, nothing, + DBN.SType.RAW_SYMBOL, DBN.SType.RAW_SYMBOL, false, + ["AAPL"], String[], String[], mappings) + @test isempty(symbol_map(md_raw)) + local df + @test_logs (:warn,) (df = records_to_dataframe(recs, md_raw)) + @test all(ismissing, df.symbol) + end + + @testset "empty frame is a no-op" begin + df = records_to_dataframe(DBN.TradeMsg[], md) + @test !hasproperty(df, :symbol) + end +end