1use std::{collections::HashMap, str::FromStr, sync::Arc};
19
20use arrow::{
21 array::{
22 Array, BinaryArray, BinaryBuilder, BooleanArray, BooleanBuilder, StringArray,
23 StringBuilder, UInt8Array, UInt64Array,
24 },
25 datatypes::{DataType, Field, Schema},
26 error::ArrowError,
27 record_batch::RecordBatch,
28};
29use nautilus_core::Params;
30use nautilus_model::{
31 enums::OptionKind,
32 identifiers::{InstrumentId, Symbol},
33 instruments::crypto_option::CryptoOption,
34 types::{money::Money, price::Price, quantity::Quantity},
35};
36#[allow(unused)]
37use rust_decimal::Decimal;
38#[allow(unused)]
39use serde_json::Value;
40
41use crate::arrow::{
42 ArrowSchemaProvider, EncodeToRecordBatch, EncodingError, KEY_INSTRUMENT_ID,
43 KEY_PRICE_PRECISION, KEY_SIZE_PRECISION, extract_column, extract_column_by_name_or_index,
44 extract_optional_string_column_by_name, optional_ustr_value,
45};
46
47fn option_kind_to_string(ok: OptionKind) -> String {
49 match ok {
50 OptionKind::Call => "Call".to_string(),
51 OptionKind::Put => "Put".to_string(),
52 }
53}
54
55fn option_kind_from_str(s: &str) -> Result<OptionKind, EncodingError> {
57 match s {
58 "Call" => Ok(OptionKind::Call),
59 "Put" => Ok(OptionKind::Put),
60 _ => Err(EncodingError::ParseError(
61 "option_kind",
62 format!("Unknown option kind: {s}"),
63 )),
64 }
65}
66
67impl ArrowSchemaProvider for CryptoOption {
68 fn get_schema(metadata: Option<HashMap<String, String>>) -> Schema {
69 let fields = vec![
70 Field::new("id", DataType::Utf8, false),
71 Field::new("raw_symbol", DataType::Utf8, false),
72 Field::new("underlying", DataType::Utf8, false),
73 Field::new("quote_currency", DataType::Utf8, false),
74 Field::new("settlement_currency", DataType::Utf8, false),
75 Field::new("is_inverse", DataType::Boolean, false),
76 Field::new("option_kind", DataType::Utf8, false),
77 Field::new("strike_price", DataType::Utf8, false),
78 Field::new("activation_ns", DataType::UInt64, false),
79 Field::new("expiration_ns", DataType::UInt64, false),
80 Field::new("price_precision", DataType::UInt8, false),
81 Field::new("size_precision", DataType::UInt8, false),
82 Field::new("price_increment", DataType::Utf8, false),
83 Field::new("size_increment", DataType::Utf8, false),
84 Field::new("multiplier", DataType::Utf8, false),
85 Field::new("lot_size", DataType::Utf8, true),
86 Field::new("max_quantity", DataType::Utf8, true), Field::new("min_quantity", DataType::Utf8, true), Field::new("max_notional", DataType::Utf8, true), Field::new("min_notional", DataType::Utf8, true), Field::new("max_price", DataType::Utf8, true), Field::new("min_price", DataType::Utf8, true), Field::new("margin_init", DataType::Utf8, false),
93 Field::new("margin_maint", DataType::Utf8, false),
94 Field::new("maker_fee", DataType::Utf8, false),
95 Field::new("taker_fee", DataType::Utf8, false),
96 Field::new("tick_scheme", DataType::Utf8, true),
97 Field::new("info", DataType::Binary, true), Field::new("ts_event", DataType::UInt64, false),
99 Field::new("ts_init", DataType::UInt64, false),
100 ];
101
102 let mut final_metadata = HashMap::new();
103 final_metadata.insert("class".to_string(), "CryptoOption".to_string());
104
105 if let Some(meta) = metadata {
106 final_metadata.extend(meta);
107 }
108
109 Schema::new_with_metadata(fields, final_metadata)
110 }
111}
112
113impl EncodeToRecordBatch for CryptoOption {
114 fn encode_batch(
115 #[allow(unused)] metadata: &HashMap<String, String>,
116 data: &[Self],
117 ) -> Result<RecordBatch, ArrowError> {
118 let mut id_builder = StringBuilder::new();
119 let mut raw_symbol_builder = StringBuilder::new();
120 let mut underlying_builder = StringBuilder::new();
121 let mut quote_currency_builder = StringBuilder::new();
122 let mut settlement_currency_builder = StringBuilder::new();
123 let mut is_inverse_builder = BooleanBuilder::new();
124 let mut option_kind_builder = StringBuilder::new();
125 let mut strike_price_builder = StringBuilder::new();
126 let mut activation_ns_builder = UInt64Array::builder(data.len());
127 let mut expiration_ns_builder = UInt64Array::builder(data.len());
128 let mut price_precision_builder = UInt8Array::builder(data.len());
129 let mut size_precision_builder = UInt8Array::builder(data.len());
130 let mut price_increment_builder = StringBuilder::new();
131 let mut size_increment_builder = StringBuilder::new();
132 let mut multiplier_builder = StringBuilder::new();
133 let mut lot_size_builder = StringBuilder::new();
134 let mut max_quantity_builder = StringBuilder::new();
135 let mut min_quantity_builder = StringBuilder::new();
136 let mut max_notional_builder = StringBuilder::new();
137 let mut min_notional_builder = StringBuilder::new();
138 let mut max_price_builder = StringBuilder::new();
139 let mut min_price_builder = StringBuilder::new();
140 let mut margin_init_builder = StringBuilder::new();
141 let mut margin_maint_builder = StringBuilder::new();
142 let mut maker_fee_builder = StringBuilder::new();
143 let mut taker_fee_builder = StringBuilder::new();
144 let mut tick_scheme_builder = StringBuilder::new();
145 let mut info_builder = BinaryBuilder::new();
146 let mut ts_event_builder = UInt64Array::builder(data.len());
147 let mut ts_init_builder = UInt64Array::builder(data.len());
148
149 for co in data {
150 id_builder.append_value(co.id.to_string());
151 raw_symbol_builder.append_value(co.raw_symbol);
152 underlying_builder.append_value(co.underlying.to_string());
153 quote_currency_builder.append_value(co.quote_currency.to_string());
154 settlement_currency_builder.append_value(co.settlement_currency.to_string());
155 is_inverse_builder.append_value(co.is_inverse);
156 option_kind_builder.append_value(option_kind_to_string(co.option_kind));
157 strike_price_builder.append_value(co.strike_price.to_string());
158 activation_ns_builder.append_value(co.activation_ns.as_u64());
159 expiration_ns_builder.append_value(co.expiration_ns.as_u64());
160 price_precision_builder.append_value(co.price_precision);
161 size_precision_builder.append_value(co.size_precision);
162 price_increment_builder.append_value(co.price_increment.to_string());
163 size_increment_builder.append_value(co.size_increment.to_string());
164 multiplier_builder.append_value(co.multiplier.to_string());
165 lot_size_builder.append_value(co.lot_size.to_string());
166
167 if let Some(max_qty) = co.max_quantity {
168 max_quantity_builder.append_value(max_qty.to_string());
169 } else {
170 max_quantity_builder.append_null();
171 }
172
173 if let Some(min_qty) = co.min_quantity {
174 min_quantity_builder.append_value(min_qty.to_string());
175 } else {
176 min_quantity_builder.append_null();
177 }
178
179 if let Some(max_not) = co.max_notional {
180 max_notional_builder.append_value(max_not.to_string());
181 } else {
182 max_notional_builder.append_null();
183 }
184
185 if let Some(min_not) = co.min_notional {
186 min_notional_builder.append_value(min_not.to_string());
187 } else {
188 min_notional_builder.append_null();
189 }
190
191 if let Some(max_p) = co.max_price {
192 max_price_builder.append_value(max_p.to_string());
193 } else {
194 max_price_builder.append_null();
195 }
196
197 if let Some(min_p) = co.min_price {
198 min_price_builder.append_value(min_p.to_string());
199 } else {
200 min_price_builder.append_null();
201 }
202
203 margin_init_builder.append_value(co.margin_init.to_string());
204 margin_maint_builder.append_value(co.margin_maint.to_string());
205 maker_fee_builder.append_value(co.maker_fee.to_string());
206 taker_fee_builder.append_value(co.taker_fee.to_string());
207
208 if let Some(tick_scheme) = co.tick_scheme {
209 tick_scheme_builder.append_value(tick_scheme);
210 } else {
211 tick_scheme_builder.append_null();
212 }
213
214 if let Some(ref info) = co.info {
216 match serde_json::to_vec(info) {
217 Ok(json_bytes) => {
218 info_builder.append_value(json_bytes);
219 }
220 Err(e) => {
221 return Err(ArrowError::InvalidArgumentError(format!(
222 "Failed to serialize info dict to JSON: {e}"
223 )));
224 }
225 }
226 } else {
227 info_builder.append_null();
228 }
229
230 ts_event_builder.append_value(co.ts_event.as_u64());
231 ts_init_builder.append_value(co.ts_init.as_u64());
232 }
233
234 let mut final_metadata = metadata.clone();
235 final_metadata.insert("class".to_string(), "CryptoOption".to_string());
236
237 RecordBatch::try_new(
238 Self::get_schema(Some(final_metadata)).into(),
239 vec![
240 Arc::new(id_builder.finish()),
241 Arc::new(raw_symbol_builder.finish()),
242 Arc::new(underlying_builder.finish()),
243 Arc::new(quote_currency_builder.finish()),
244 Arc::new(settlement_currency_builder.finish()),
245 Arc::new(is_inverse_builder.finish()),
246 Arc::new(option_kind_builder.finish()),
247 Arc::new(strike_price_builder.finish()),
248 Arc::new(activation_ns_builder.finish()),
249 Arc::new(expiration_ns_builder.finish()),
250 Arc::new(price_precision_builder.finish()),
251 Arc::new(size_precision_builder.finish()),
252 Arc::new(price_increment_builder.finish()),
253 Arc::new(size_increment_builder.finish()),
254 Arc::new(multiplier_builder.finish()),
255 Arc::new(lot_size_builder.finish()),
256 Arc::new(max_quantity_builder.finish()),
257 Arc::new(min_quantity_builder.finish()),
258 Arc::new(max_notional_builder.finish()),
259 Arc::new(min_notional_builder.finish()),
260 Arc::new(max_price_builder.finish()),
261 Arc::new(min_price_builder.finish()),
262 Arc::new(margin_init_builder.finish()),
263 Arc::new(margin_maint_builder.finish()),
264 Arc::new(maker_fee_builder.finish()),
265 Arc::new(taker_fee_builder.finish()),
266 Arc::new(tick_scheme_builder.finish()),
267 Arc::new(info_builder.finish()),
268 Arc::new(ts_event_builder.finish()),
269 Arc::new(ts_init_builder.finish()),
270 ],
271 )
272 }
273
274 fn metadata(&self) -> HashMap<String, String> {
275 let mut metadata = HashMap::new();
276 metadata.insert(KEY_INSTRUMENT_ID.to_string(), self.id.to_string());
277 metadata.insert(
278 KEY_PRICE_PRECISION.to_string(),
279 self.price_precision.to_string(),
280 );
281 metadata.insert(
282 KEY_SIZE_PRECISION.to_string(),
283 self.size_precision.to_string(),
284 );
285 metadata
286 }
287}
288
289pub fn decode_crypto_option_batch(
296 #[allow(unused)] metadata: &HashMap<String, String>,
297 record_batch: &RecordBatch,
298) -> Result<Vec<CryptoOption>, EncodingError> {
299 let cols = record_batch.columns();
300 let num_rows = record_batch.num_rows();
301
302 let id_values = extract_column::<StringArray>(cols, "id", 0, DataType::Utf8)?;
303 let raw_symbol_values = extract_column::<StringArray>(cols, "raw_symbol", 1, DataType::Utf8)?;
304 let underlying_values = extract_column::<StringArray>(cols, "underlying", 2, DataType::Utf8)?;
305 let quote_currency_values =
306 extract_column::<StringArray>(cols, "quote_currency", 3, DataType::Utf8)?;
307 let settlement_currency_values =
308 extract_column::<StringArray>(cols, "settlement_currency", 4, DataType::Utf8)?;
309 let is_inverse_values =
310 extract_column::<BooleanArray>(cols, "is_inverse", 5, DataType::Boolean)?;
311 let option_kind_values = extract_column::<StringArray>(cols, "option_kind", 6, DataType::Utf8)?;
312 let strike_price_values =
313 extract_column::<StringArray>(cols, "strike_price", 7, DataType::Utf8)?;
314 let activation_ns_values =
315 extract_column::<UInt64Array>(cols, "activation_ns", 8, DataType::UInt64)?;
316 let expiration_ns_values =
317 extract_column::<UInt64Array>(cols, "expiration_ns", 9, DataType::UInt64)?;
318 let price_precision_values =
319 extract_column::<UInt8Array>(cols, "price_precision", 10, DataType::UInt8)?;
320 let size_precision_values =
321 extract_column::<UInt8Array>(cols, "size_precision", 11, DataType::UInt8)?;
322 let price_increment_values =
323 extract_column::<StringArray>(cols, "price_increment", 12, DataType::Utf8)?;
324 let size_increment_values =
325 extract_column::<StringArray>(cols, "size_increment", 13, DataType::Utf8)?;
326 let multiplier_values = extract_column::<StringArray>(cols, "multiplier", 14, DataType::Utf8)?;
327 let lot_size_values = record_batch
328 .schema()
329 .index_of("lot_size")
330 .ok()
331 .map(|index| extract_column::<StringArray>(cols, "lot_size", index, DataType::Utf8))
332 .transpose()?;
333 let lot_size_offset = usize::from(lot_size_values.is_some());
334 let max_quantity_values = cols
335 .get(15 + lot_size_offset)
336 .ok_or_else(|| EncodingError::MissingColumn("max_quantity", 15 + lot_size_offset))?;
337 let min_quantity_values = cols
338 .get(16 + lot_size_offset)
339 .ok_or_else(|| EncodingError::MissingColumn("min_quantity", 16 + lot_size_offset))?;
340 let max_notional_values = cols
341 .get(17 + lot_size_offset)
342 .ok_or_else(|| EncodingError::MissingColumn("max_notional", 17 + lot_size_offset))?;
343 let min_notional_values = cols
344 .get(18 + lot_size_offset)
345 .ok_or_else(|| EncodingError::MissingColumn("min_notional", 18 + lot_size_offset))?;
346 let max_price_values = cols
347 .get(19 + lot_size_offset)
348 .ok_or_else(|| EncodingError::MissingColumn("max_price", 19 + lot_size_offset))?;
349 let min_price_values = cols
350 .get(20 + lot_size_offset)
351 .ok_or_else(|| EncodingError::MissingColumn("min_price", 20 + lot_size_offset))?;
352 let margin_init_values =
353 extract_column::<StringArray>(cols, "margin_init", 21 + lot_size_offset, DataType::Utf8)?;
354 let margin_maint_values =
355 extract_column::<StringArray>(cols, "margin_maint", 22 + lot_size_offset, DataType::Utf8)?;
356 let maker_fee_values =
357 extract_column::<StringArray>(cols, "maker_fee", 23 + lot_size_offset, DataType::Utf8)?;
358 let taker_fee_values =
359 extract_column::<StringArray>(cols, "taker_fee", 24 + lot_size_offset, DataType::Utf8)?;
360 let tick_scheme_values = extract_optional_string_column_by_name(record_batch, "tick_scheme")?;
361 let info_values = extract_column_by_name_or_index::<BinaryArray>(
362 record_batch,
363 "info",
364 25 + lot_size_offset,
365 DataType::Binary,
366 )?;
367 let ts_event_values = extract_column_by_name_or_index::<UInt64Array>(
368 record_batch,
369 "ts_event",
370 26 + lot_size_offset,
371 DataType::UInt64,
372 )?;
373 let ts_init_values = extract_column_by_name_or_index::<UInt64Array>(
374 record_batch,
375 "ts_init",
376 27 + lot_size_offset,
377 DataType::UInt64,
378 )?;
379
380 let mut result = Vec::with_capacity(num_rows);
381
382 for i in 0..num_rows {
383 let id = InstrumentId::from_str(id_values.value(i))
384 .map_err(|e| EncodingError::ParseError("id", format!("row {i}: {e}")))?;
385 let raw_symbol = Symbol::from(raw_symbol_values.value(i));
386 let underlying = super::decode_currency(
387 underlying_values.value(i),
388 "underlying",
389 "crypto_option.underlying",
390 i,
391 )?;
392 let quote_currency = super::decode_currency(
393 quote_currency_values.value(i),
394 "quote_currency",
395 "crypto_option.quote_currency",
396 i,
397 )?;
398 let settlement_currency = super::decode_currency(
399 settlement_currency_values.value(i),
400 "settlement_currency",
401 "crypto_option.settlement_currency",
402 i,
403 )?;
404 let is_inverse = is_inverse_values.value(i);
405 let option_kind = option_kind_from_str(option_kind_values.value(i))?;
406 let strike_price = Price::from_str(strike_price_values.value(i))
407 .map_err(|e| EncodingError::ParseError("strike_price", format!("row {i}: {e}")))?;
408 let activation_ns = nautilus_core::UnixNanos::from(activation_ns_values.value(i));
409 let expiration_ns = nautilus_core::UnixNanos::from(expiration_ns_values.value(i));
410 let price_prec = price_precision_values.value(i);
411 let size_prec = size_precision_values.value(i);
412
413 let price_increment = Price::from_str(price_increment_values.value(i))
414 .map_err(|e| EncodingError::ParseError("price_increment", format!("row {i}: {e}")))?;
415 let size_increment = Quantity::from_str(size_increment_values.value(i))
416 .map_err(|e| EncodingError::ParseError("size_increment", format!("row {i}: {e}")))?;
417 let multiplier = Quantity::from_str(multiplier_values.value(i))
418 .map_err(|e| EncodingError::ParseError("multiplier", format!("row {i}: {e}")))?;
419 let lot_size =
420 if let Some(values) = lot_size_values {
421 if values.is_null(i) {
422 None
423 } else {
424 Some(Quantity::from_str(values.value(i)).map_err(|e| {
425 EncodingError::ParseError("lot_size", format!("row {i}: {e}"))
426 })?)
427 }
428 } else {
429 None
430 };
431
432 let max_quantity =
433 if max_quantity_values.is_null(i) {
434 None
435 } else {
436 let max_qty_str = max_quantity_values
437 .as_any()
438 .downcast_ref::<StringArray>()
439 .ok_or_else(|| {
440 EncodingError::ParseError("max_quantity", format!("row {i}: invalid type"))
441 })?
442 .value(i);
443 Some(Quantity::from_str(max_qty_str).map_err(|e| {
444 EncodingError::ParseError("max_quantity", format!("row {i}: {e}"))
445 })?)
446 };
447
448 let min_quantity =
449 if min_quantity_values.is_null(i) {
450 None
451 } else {
452 let min_qty_str = min_quantity_values
453 .as_any()
454 .downcast_ref::<StringArray>()
455 .ok_or_else(|| {
456 EncodingError::ParseError("min_quantity", format!("row {i}: invalid type"))
457 })?
458 .value(i);
459 Some(Quantity::from_str(min_qty_str).map_err(|e| {
460 EncodingError::ParseError("min_quantity", format!("row {i}: {e}"))
461 })?)
462 };
463
464 let max_notional =
465 if max_notional_values.is_null(i) {
466 None
467 } else {
468 let max_not_str = max_notional_values
469 .as_any()
470 .downcast_ref::<StringArray>()
471 .ok_or_else(|| {
472 EncodingError::ParseError("max_notional", format!("row {i}: invalid type"))
473 })?
474 .value(i);
475 Some(Money::from_str(max_not_str).map_err(|e| {
476 EncodingError::ParseError("max_notional", format!("row {i}: {e}"))
477 })?)
478 };
479
480 let min_notional =
481 if min_notional_values.is_null(i) {
482 None
483 } else {
484 let min_not_str = min_notional_values
485 .as_any()
486 .downcast_ref::<StringArray>()
487 .ok_or_else(|| {
488 EncodingError::ParseError("min_notional", format!("row {i}: invalid type"))
489 })?
490 .value(i);
491 Some(Money::from_str(min_not_str).map_err(|e| {
492 EncodingError::ParseError("min_notional", format!("row {i}: {e}"))
493 })?)
494 };
495
496 let max_price = if max_price_values.is_null(i) {
497 None
498 } else {
499 let max_p_str = max_price_values
500 .as_any()
501 .downcast_ref::<StringArray>()
502 .ok_or_else(|| {
503 EncodingError::ParseError("max_price", format!("row {i}: invalid type"))
504 })?
505 .value(i);
506 Some(
507 Price::from_str(max_p_str)
508 .map_err(|e| EncodingError::ParseError("max_price", format!("row {i}: {e}")))?,
509 )
510 };
511
512 let min_price = if min_price_values.is_null(i) {
513 None
514 } else {
515 let min_p_str = min_price_values
516 .as_any()
517 .downcast_ref::<StringArray>()
518 .ok_or_else(|| {
519 EncodingError::ParseError("min_price", format!("row {i}: invalid type"))
520 })?
521 .value(i);
522 Some(
523 Price::from_str(min_p_str)
524 .map_err(|e| EncodingError::ParseError("min_price", format!("row {i}: {e}")))?,
525 )
526 };
527
528 let margin_init = Decimal::from_str(margin_init_values.value(i))
529 .map_err(|e| EncodingError::ParseError("margin_init", format!("row {i}: {e}")))?;
530 let margin_maint = Decimal::from_str(margin_maint_values.value(i))
531 .map_err(|e| EncodingError::ParseError("margin_maint", format!("row {i}: {e}")))?;
532 let maker_fee = Decimal::from_str(maker_fee_values.value(i))
533 .map_err(|e| EncodingError::ParseError("maker_fee", format!("row {i}: {e}")))?;
534 let taker_fee = Decimal::from_str(taker_fee_values.value(i))
535 .map_err(|e| EncodingError::ParseError("taker_fee", format!("row {i}: {e}")))?;
536
537 let info = if info_values.is_null(i) {
539 None
540 } else {
541 let info_bytes = info_values
542 .as_any()
543 .downcast_ref::<BinaryArray>()
544 .ok_or_else(|| EncodingError::ParseError("info", format!("row {i}: invalid type")))?
545 .value(i);
546
547 match serde_json::from_slice::<Params>(info_bytes) {
548 Ok(info_dict) => Some(info_dict),
549 Err(e) => {
550 return Err(EncodingError::ParseError(
551 "info",
552 format!("row {i}: failed to deserialize JSON: {e}"),
553 ));
554 }
555 }
556 };
557
558 let ts_event = nautilus_core::UnixNanos::from(ts_event_values.value(i));
559 let ts_init = nautilus_core::UnixNanos::from(ts_init_values.value(i));
560
561 let tick_scheme = optional_ustr_value(tick_scheme_values, i);
562
563 let crypto_option = CryptoOption::new_checked(
564 id,
565 raw_symbol,
566 underlying,
567 quote_currency,
568 settlement_currency,
569 is_inverse,
570 option_kind,
571 strike_price,
572 activation_ns,
573 expiration_ns,
574 price_prec,
575 size_prec,
576 price_increment,
577 size_increment,
578 Some(multiplier),
579 lot_size,
580 max_quantity,
581 min_quantity,
582 max_notional,
583 min_notional,
584 max_price,
585 min_price,
586 Some(margin_init),
587 Some(margin_maint),
588 Some(maker_fee),
589 Some(taker_fee),
590 tick_scheme,
591 info,
592 ts_event,
593 ts_init,
594 )
595 .map_err(|e| super::instrument_validation_error::<CryptoOption>(i, e))?;
596
597 result.push(crypto_option);
598 }
599
600 Ok(result)
601}