Fix records_to_dataframe FieldError on MBP-1/MBP-10/BBO (issue #40)

CHANGELOG.md

@@ -22,6 +22,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- `records_to_dataframe` (and therefore `to_dataframe` / `dbn_to_parquet`) no
+  longer throws `FieldError` on **TBBO / MBP-1** and **MBP-10**. The converters
+  read non-existent flat `bid_px_00…` fields; the bid/ask data actually lives in
+  the nested `levels::BidAskPair` (MBP-1) / `levels::NTuple{10,BidAskPair}`
+  (MBP-10). MBP-1 now reads through `levels`, MBP-10 indexes the level tuple, and
+  the consolidated/BBO family (`CMBP1Msg`, `TCBBOMsg`, `CBBO1sMsg`, `CBBO1mMsg`,
+  `BBO1sMsg`, `BBO1mMsg`) — which share the MBP-1 layout — now route to the same
+  converter instead of falling through to the generic mixed-record path ([#40]).
 - `read_stat_msg` now sizes the record body from `hd.length` instead of assuming
   the 80-byte DBN v3 layout. Pre-v3 64-byte `StatMsg` records (32-bit `quantity`,
   as currently served by the historical gateway for STATISTICS) decode correctly
@@ -107,3 +115,4 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 [#33]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/issues/33
 [#34]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/issues/34
 [#35]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/issues/35
+[#40]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/issues/40

src/export.jl

@@ -111,8 +111,11 @@ function records_to_dataframe(records::Vector)
             return trades_to_dataframe(convert(Vector{TradeMsg}, records))
         elseif record_type <: MBOMsg
             return mbo_to_dataframe(convert(Vector{MBOMsg}, records))
-        elseif record_type <: MBP1Msg
-            return mbp1_to_dataframe(convert(Vector{MBP1Msg}, records))
+        elseif record_type <: TopOfBookMsg
+            # MBP-1 and the consolidated/BBO family (CMBP-1, TBBO, CBBO, BBO) all
+            # share the MBP-1 record layout: scalar fields plus a single
+            # top-of-book `levels::BidAskPair`. One converter handles them all.
+            return mbp1_to_dataframe(convert(Vector{record_type}, records))
         elseif record_type <: MBP10Msg
             return mbp10_to_dataframe(convert(Vector{MBP10Msg}, records))
         elseif record_type <: OHLCVMsg
@@ -170,18 +173,24 @@ function mbo_to_dataframe(records::Vector{MBOMsg})
     )
 end
 
-function mbp1_to_dataframe(records::Vector{MBP1Msg})
+# MBP-1 and the consolidated/BBO family share an identical record layout: the
+# scalar trade fields plus a single top-of-book `levels::BidAskPair`. They differ
+# only in which fields the gateway populates, so one DataFrame converter serves
+# all of them.
+const TopOfBookMsg = Union{MBP1Msg,CMBP1Msg,TCBBOMsg,CBBO1sMsg,CBBO1mMsg,BBO1sMsg,BBO1mMsg}
+
+function mbp1_to_dataframe(records::Vector{<:TopOfBookMsg})
     DataFrame(
         ts_event = [r.hd.ts_event for r in records],
         ts_recv = [r.ts_recv for r in records],
         instrument_id = [r.hd.instrument_id for r in records],
         publisher_id = [r.hd.publisher_id for r in records],
-        bid_price = [price_to_float(r.bid_px_00) for r in records],
-        ask_price = [price_to_float(r.ask_px_00) for r in records],
-        bid_size = [r.bid_sz_00 for r in records],
-        ask_size = [r.ask_sz_00 for r in records],
-        bid_ct = [r.bid_ct_00 for r in records],
-        ask_ct = [r.ask_ct_00 for r in records],
+        bid_price = [price_to_float(r.levels.bid_px) for r in records],
+        ask_price = [price_to_float(r.levels.ask_px) for r in records],
+        bid_size = [r.levels.bid_sz for r in records],
+        ask_size = [r.levels.ask_sz for r in records],
+        bid_ct = [r.levels.bid_ct for r in records],
+        ask_ct = [r.levels.ask_ct for r in records],
         flags = [r.flags for r in records],
         ts_in_delta = [r.ts_in_delta for r in records],
         sequence = [r.sequence for r in records],
@@ -191,45 +200,32 @@ function mbp1_to_dataframe(records::Vector{MBP1Msg})
 end
 
 function mbp10_to_dataframe(records::Vector{MBP10Msg})
-    # For MBP10, we need to expand the levels
+    # MBP-10 carries 10 price levels in `levels::NTuple{10,BidAskPair}`; expand
+    # each record into one row per level (0-indexed in the output).
     rows = []
     for record in records
-        # Add each level as a separate row
-        levels = [
-            (record.bid_px_00, record.ask_px_00, record.bid_sz_00, record.ask_sz_00, record.bid_ct_00, record.ask_ct_00),
-            (record.bid_px_01, record.ask_px_01, record.bid_sz_01, record.ask_sz_01, record.bid_ct_01, record.ask_ct_01),
-            (record.bid_px_02, record.ask_px_02, record.bid_sz_02, record.ask_sz_02, record.bid_ct_02, record.ask_ct_02),
-            (record.bid_px_03, record.ask_px_03, record.bid_sz_03, record.ask_sz_03, record.bid_ct_03, record.ask_ct_03),
-            (record.bid_px_04, record.ask_px_04, record.bid_sz_04, record.ask_sz_04, record.bid_ct_04, record.ask_ct_04),
-            (record.bid_px_05, record.ask_px_05, record.bid_sz_05, record.ask_sz_05, record.bid_ct_05, record.ask_ct_05),
-            (record.bid_px_06, record.ask_px_06, record.bid_sz_06, record.ask_sz_06, record.bid_ct_06, record.ask_ct_06),
-            (record.bid_px_07, record.ask_px_07, record.bid_sz_07, record.ask_sz_07, record.bid_ct_07, record.ask_ct_07),
-            (record.bid_px_08, record.ask_px_08, record.bid_sz_08, record.ask_sz_08, record.bid_ct_08, record.ask_ct_08),
-            (record.bid_px_09, record.ask_px_09, record.bid_sz_09, record.ask_sz_09, record.bid_ct_09, record.ask_ct_09)
-        ]
-
-        for (level, (bid_px, ask_px, bid_sz, ask_sz, bid_ct, ask_ct)) in enumerate(levels)
+        for (level, pair) in enumerate(record.levels)
             push!(rows, (
                 ts_event = record.hd.ts_event,
                 ts_recv = record.ts_recv,
                 instrument_id = record.hd.instrument_id,
                 publisher_id = record.hd.publisher_id,
                 level = level - 1,  # 0-indexed
-                bid_price = price_to_float(bid_px),
-                ask_price = price_to_float(ask_px),
-                bid_size = bid_sz,
-                ask_size = ask_sz,
-                bid_ct = bid_ct,
-                ask_ct = ask_ct,
+                bid_price = price_to_float(pair.bid_px),
+                ask_price = price_to_float(pair.ask_px),
+                bid_size = pair.bid_sz,
+                ask_size = pair.ask_sz,
+                bid_ct = pair.bid_ct,
+                ask_ct = pair.ask_ct,
                 flags = record.flags,
                 ts_in_delta = record.ts_in_delta,
                 sequence = record.sequence,
-                action = string(Action.T(record.action)),
+                action = string(record.action),
                 side = string(record.side)
             ))
         end
     end
-    
+
     return DataFrame(rows)
 end
 

test/runtests.jl

@@ -23,6 +23,7 @@ include("test_utils.jl")
     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)
+    include("test_issue40_mbp_dataframe.jl")  # regression: MBP-1/MBP-10/BBO records_to_dataframe via nested levels (issue #40)
 
     # Run compatibility tests if the Rust CLI is available
     dbn_cli_path = if Sys.iswindows()

test/test_issue40_mbp_dataframe.jl

@@ -0,0 +1,59 @@
+# Regression for issue #40: records_to_dataframe threw FieldError on TBBO/MBP-1
+# and MBP-10 because the converters read flat `bid_px_00…` fields that don't
+# exist — the bid/ask data lives in the nested `levels::BidAskPair` (MBP-1 and
+# the consolidated/BBO family) or `levels::NTuple{10,BidAskPair}` (MBP-10).
+
+@testset "Issue #40: MBP/BBO records_to_dataframe" begin
+    ts = Int64(1_700_000_000_000_000_000)
+    hd(rt, id) = DBN.RecordHeader(UInt8(0), rt, UInt16(1), UInt32(id), ts)
+    # Distinct values per level so a wrong index is caught.
+    bap(o) = DBN.BidAskPair(Int64(1000 + o), Int64(2000 + o),
+                            UInt32(10 + o), UInt32(20 + o), UInt32(o), UInt32(100 + o))
+
+    @testset "MBP-1 reads through levels" begin
+        recs = [DBN.MBP1Msg(hd(DBN.RType.MBP_1_MSG, 42), Int64(1000), UInt32(5),
+                            DBN.Action.MODIFY, DBN.Side.BID, 0x00, 0x00,
+                            ts, Int32(3), UInt32(7), bap(0))]
+        df = records_to_dataframe(recs)   # must not throw
+        @test df.bid_price == [price_to_float(Int64(1000))]
+        @test df.ask_price == [price_to_float(Int64(2000))]
+        @test df.bid_size == UInt32[10]
+        @test df.ask_size == UInt32[20]
+        @test df.bid_ct == UInt32[0]
+        @test df.ask_ct == UInt32[100]
+        @test df.action == ["MODIFY"]
+        @test df.instrument_id == UInt32[42]
+    end
+
+    @testset "MBP-10 expands all 10 nested levels" begin
+        levels = ntuple(i -> bap(i - 1), 10)
+        recs = [DBN.MBP10Msg(hd(DBN.RType.MBP_10_MSG, 42), Int64(1000), UInt32(5),
+                             DBN.Action.MODIFY, DBN.Side.BID, 0x00, 0x00,
+                             ts, Int32(3), UInt32(7), levels)]
+        df = records_to_dataframe(recs)   # must not throw
+        @test nrow(df) == 10
+        @test df.level == collect(0:9)
+        @test df.bid_price == [price_to_float(Int64(1000 + o)) for o in 0:9]
+        @test df.ask_price == [price_to_float(Int64(2000 + o)) for o in 0:9]
+        @test df.bid_size == UInt32[10 + o for o in 0:9]
+    end
+
+    @testset "consolidated/BBO family route to the bid/ask converter" begin
+        # Each shares the MBP-1 layout; previously these fell through to
+        # mixed_records_to_dataframe and produced no bid_price column.
+        for (T, rt) in ((DBN.TCBBOMsg, DBN.RType.TCBBO_MSG),
+                        (DBN.CMBP1Msg, DBN.RType.CMBP_1_MSG),
+                        (DBN.CBBO1sMsg, DBN.RType.CBBO_1S_MSG),
+                        (DBN.CBBO1mMsg, DBN.RType.CBBO_1M_MSG),
+                        (DBN.BBO1sMsg, DBN.RType.BBO_1S_MSG),
+                        (DBN.BBO1mMsg, DBN.RType.BBO_1M_MSG))
+            recs = [T(hd(rt, 42), Int64(1000), UInt32(5),
+                      DBN.Action.NONE, DBN.Side.NONE, 0x00, 0x00,
+                      ts, Int32(0), UInt32(7), bap(0))]
+            df = records_to_dataframe(recs)
+            @test hasproperty(df, :bid_price)
+            @test df.bid_price == [price_to_float(Int64(1000))]
+            @test df.ask_price == [price_to_float(Int64(2000))]
+        end
+    end
+end