ITCHCPP 1.6.1
High-Performance NASDAQ TotalView-ITCH 5.0 Parser & Market-Data Platform
Loading...
Searching...
No Matches
arrow_export.cpp
Go to the documentation of this file.
1#include "itch/io/arrow_export.hpp"
2
3#ifdef ITCH_WITH_ARROW
4
5#include <arrow/api.h>
6#include <arrow/io/file.h>
7#include <parquet/arrow/writer.h>
8
9#include <string>
10#include <variant>
11
12#include "itch/price.hpp"
13
14namespace itch::io {
15
16struct ArrowExporter::Impl {
17 arrow::StringBuilder message_type;
18 arrow::UInt64Builder timestamp;
19 arrow::UInt16Builder stock_locate;
20 arrow::UInt16Builder tracking_number;
21 arrow::StringBuilder symbol;
22 arrow::UInt64Builder reference_number;
23 arrow::StringBuilder side;
24 arrow::UInt64Builder shares;
25 arrow::DoubleBuilder price;
26 arrow::UInt64Builder match_number;
27 arrow::StringBuilder printable;
28 arrow::StringBuilder extra;
29 std::uint64_t row_count {0};
30 std::string last_error;
31
32 // Builds the common header columns shared by every message type.
33 template <typename MsgType>
34 auto append_common(const MsgType& msg) -> void {
35 (void)message_type.Append(std::string {msg.message_type});
36 (void)timestamp.Append(msg.timestamp);
37 (void)stock_locate.Append(msg.stock_locate);
38 (void)tracking_number.Append(msg.tracking_number);
39 }
40
41 // Appends nulls/empties for the type-specific columns, overwritten as needed.
42 auto append_blank_specific() -> void {
43 (void)symbol.AppendEmptyValue();
44 (void)reference_number.AppendNull();
45 (void)side.AppendEmptyValue();
46 (void)shares.AppendNull();
47 (void)price.AppendNull();
48 (void)match_number.AppendNull();
49 (void)printable.AppendEmptyValue();
50 (void)extra.AppendEmptyValue();
51 }
52};
53
54ArrowExporter::ArrowExporter() : m_impl {std::make_unique<Impl>()} {}
55ArrowExporter::~ArrowExporter() = default;
56
57auto ArrowExporter::rows() const noexcept -> std::uint64_t { return m_impl->row_count; }
58auto ArrowExporter::error() const -> const std::string& { return m_impl->last_error; }
59
60auto ArrowExporter::append(const Message& message) -> void {
61 Impl& impl = *m_impl;
62 std::visit([&impl](const auto& msg) { impl.append_common(msg); }, message);
63
64 // Default every type-specific column to null/empty, then set what applies.
65 // Replacing the convenience of AppendEmptyValue with explicit per-type code
66 // keeps the column lengths in lockstep with the header columns.
67 std::string symbol_value;
68 bool has_reference = false;
69 std::uint64_t reference = 0;
70 std::string side_value;
71 bool has_shares = false;
72 std::uint64_t shares_value = 0;
73 bool has_price = false;
74 double price_value = 0.0;
75 bool has_match = false;
76 std::uint64_t match_value = 0;
77 std::string printable_value;
78 std::string extra_value;
79
80 if (const auto* add = std::get_if<AddOrderMessage>(&message)) {
81 symbol_value = to_string(add->stock, STOCK_LEN);
82 has_reference = true;
83 reference = add->order_reference_number;
84 side_value = std::string {add->buy_sell_indicator};
85 has_shares = true;
86 shares_value = add->shares;
87 has_price = true;
88 price_value = StandardPrice {add->price}.to_double();
89 } else if (const auto* add_mpid = std::get_if<AddOrderMPIDAttributionMessage>(&message)) {
90 symbol_value = to_string(add_mpid->stock, STOCK_LEN);
91 has_reference = true;
92 reference = add_mpid->order_reference_number;
93 side_value = std::string {add_mpid->buy_sell_indicator};
94 has_shares = true;
95 shares_value = add_mpid->shares;
96 has_price = true;
97 price_value = StandardPrice {add_mpid->price}.to_double();
98 extra_value = "mpid=" + to_string(add_mpid->attribution, 4);
99 } else if (const auto* exec = std::get_if<OrderExecutedMessage>(&message)) {
100 has_reference = true;
101 reference = exec->order_reference_number;
102 has_shares = true;
103 shares_value = exec->executed_shares;
104 has_match = true;
105 match_value = exec->match_number;
106 } else if (const auto* exec_px = std::get_if<OrderExecutedWithPriceMessage>(&message)) {
107 has_reference = true;
108 reference = exec_px->order_reference_number;
109 has_shares = true;
110 shares_value = exec_px->executed_shares;
111 has_price = true;
112 price_value = StandardPrice {exec_px->execution_price}.to_double();
113 has_match = true;
114 match_value = exec_px->match_number;
115 printable_value = std::string {exec_px->printable};
116 } else if (const auto* cancel = std::get_if<OrderCancelMessage>(&message)) {
117 has_reference = true;
118 reference = cancel->order_reference_number;
119 has_shares = true;
120 shares_value = cancel->cancelled_shares;
121 } else if (const auto* del = std::get_if<OrderDeleteMessage>(&message)) {
122 has_reference = true;
123 reference = del->order_reference_number;
124 } else if (const auto* replace = std::get_if<OrderReplaceMessage>(&message)) {
125 has_reference = true;
126 reference = replace->new_order_reference_number;
127 has_shares = true;
128 shares_value = replace->shares;
129 has_price = true;
130 price_value = StandardPrice {replace->price}.to_double();
131 extra_value = "orig=" + std::to_string(replace->original_order_reference_number);
132 } else if (const auto* trade = std::get_if<NonCrossTradeMessage>(&message)) {
133 symbol_value = to_string(trade->stock, STOCK_LEN);
134 has_reference = true;
135 reference = trade->order_reference_number;
136 side_value = std::string {trade->buy_sell_indicator};
137 has_shares = true;
138 shares_value = trade->shares;
139 has_price = true;
140 price_value = StandardPrice {trade->price}.to_double();
141 has_match = true;
142 match_value = trade->match_number;
143 } else if (const auto* cross = std::get_if<CrossTradeMessage>(&message)) {
144 symbol_value = to_string(cross->stock, STOCK_LEN);
145 has_shares = true;
146 shares_value = cross->shares;
147 has_price = true;
148 price_value = StandardPrice {cross->cross_price}.to_double();
149 has_match = true;
150 match_value = cross->match_number;
151 extra_value = std::string {"cross="} + cross->cross_type;
152 } else if (const auto* event = std::get_if<SystemEventMessage>(&message)) {
153 extra_value = std::string {"event="} + event->event_code;
154 } else if (const auto* directory = std::get_if<StockDirectoryMessage>(&message)) {
155 symbol_value = to_string(directory->stock, STOCK_LEN);
156 }
157
158 (void)impl.symbol.Append(symbol_value);
159 if (has_reference) {
160 (void)impl.reference_number.Append(reference);
161 } else {
162 (void)impl.reference_number.AppendNull();
163 }
164 (void)impl.side.Append(side_value);
165 if (has_shares) {
166 (void)impl.shares.Append(shares_value);
167 } else {
168 (void)impl.shares.AppendNull();
169 }
170 if (has_price) {
171 (void)impl.price.Append(price_value);
172 } else {
173 (void)impl.price.AppendNull();
174 }
175 if (has_match) {
176 (void)impl.match_number.Append(match_value);
177 } else {
178 (void)impl.match_number.AppendNull();
179 }
180 (void)impl.printable.Append(printable_value);
181 (void)impl.extra.Append(extra_value);
182 ++impl.row_count;
183}
184
185auto ArrowExporter::write_parquet(const std::string& path) -> bool {
186 Impl& impl = *m_impl;
187
188 auto finish =
189 [&impl](arrow::ArrayBuilder& builder, std::shared_ptr<arrow::Array>& out) -> bool {
190 const arrow::Status status = builder.Finish(&out);
191 if (!status.ok()) {
192 impl.last_error = status.ToString();
193 return false;
194 }
195 return true;
196 };
197
198 std::shared_ptr<arrow::Array> columns[12];
199 arrow::ArrayBuilder* builders[12] = {
200 &impl.message_type,
201 &impl.timestamp,
202 &impl.stock_locate,
203 &impl.tracking_number,
204 &impl.symbol,
205 &impl.reference_number,
206 &impl.side,
207 &impl.shares,
208 &impl.price,
209 &impl.match_number,
210 &impl.printable,
211 &impl.extra
212 };
213 for (int index = 0; index < 12; ++index) {
214 if (!finish(*builders[index], columns[index])) {
215 return false;
216 }
217 }
218
219 auto schema = arrow::schema({
220 arrow::field("message_type", arrow::utf8()),
221 arrow::field("timestamp", arrow::uint64()),
222 arrow::field("stock_locate", arrow::uint16()),
223 arrow::field("tracking_number", arrow::uint16()),
224 arrow::field("symbol", arrow::utf8()),
225 arrow::field("reference_number", arrow::uint64()),
226 arrow::field("side", arrow::utf8()),
227 arrow::field("shares", arrow::uint64()),
228 arrow::field("price", arrow::float64()),
229 arrow::field("match_number", arrow::uint64()),
230 arrow::field("printable", arrow::utf8()),
231 arrow::field("extra", arrow::utf8()),
232 });
233
234 std::vector<std::shared_ptr<arrow::Array>> column_vector {columns, columns + 12};
235 const auto table = arrow::Table::Make(schema, column_vector);
236
237 auto outfile_result = arrow::io::FileOutputStream::Open(path);
238 if (!outfile_result.ok()) {
239 impl.last_error = outfile_result.status().ToString();
240 return false;
241 }
242 const auto status = parquet::arrow::WriteTable(
243 *table,
244 arrow::default_memory_pool(),
245 *outfile_result,
246 impl.row_count > 0 ? impl.row_count : 1
247 );
248 if (!status.ok()) {
249 impl.last_error = status.ToString();
250 return false;
251 }
252 return true;
253}
254
255} // namespace itch::io
256
257#endif // ITCH_WITH_ARROW
arrow_export.hpp
Apache Arrow / Parquet columnar export.
itch::BasicPrice::to_double
constexpr auto to_double() const noexcept -> double
The price as a floating-point value (raw / 10^Decimals).
Definition price.hpp:72
printable
char printable
Definition csv_sink.cpp:25
reference_number
std::string reference_number
Definition csv_sink.cpp:20
tracking_number
std::uint16_t tracking_number
Definition csv_sink.cpp:18
symbol
std::string symbol
Definition csv_sink.cpp:19
price
std::string price
Definition csv_sink.cpp:23
timestamp
std::uint64_t timestamp
Definition csv_sink.cpp:16
stock_locate
std::uint16_t stock_locate
Definition csv_sink.cpp:17
message_type
char message_type
Definition csv_sink.cpp:15
extra
std::string extra
Definition csv_sink.cpp:26
match_number
std::string match_number
Definition csv_sink.cpp:24
side
char side
Definition csv_sink.cpp:21
shares
std::string shares
Definition csv_sink.cpp:22
itch::StandardPrice
BasicPrice< std::uint32_t, 4 > StandardPrice
Price scale for every ITCH price field except MWCB decline levels.
Definition price.hpp:102
itch::to_string
auto to_string(const char *source, size_t size) -> std::string
Converts a fixed-width character array to a string, trimming trailing spaces and NUL characters.
Definition messages.hpp:495
price.hpp
Strongly typed, fixed-point price representation for ITCH price fields.