use digdigdig3::core::types::{StreamEvent, Ticker};
use serde::{Deserialize, Serialize};
use crate::series::DataPoint;
#[inline]
fn opt_f64(v: Option<f64>) -> f64 {
v.unwrap_or(f64::NAN)
}
#[inline]
fn opt_i64_enc(v: Option<i64>) -> u64 {
match v {
Some(x) => x as u64,
None => i64::MIN as u64,
}
}
#[inline]
fn opt_i64_dec(raw: u64) -> Option<i64> {
let v = raw as i64;
if v == i64::MIN { None } else { Some(v) }
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TickerIndicatorsPoint {
pub ts_ms: i64,
pub last: f64,
pub bid: f64,
pub ask: f64,
pub bid_qty: f64,
pub ask_qty: f64,
pub high_24h: f64,
pub low_24h: f64,
pub open_price: f64,
pub prev_close_price: f64,
pub vol_24h: f64,
pub quote_vol_24h: f64,
pub change_pct_24h: f64,
pub weighted_avg_price: f64,
pub last_qty: f64,
pub mark_price: f64,
pub index_price: f64,
pub open_interest: f64,
pub funding_rate: f64,
pub count: Option<i64>,
}
const INDICATORS_SIZE: usize = 160;
impl TickerIndicatorsPoint {
pub fn from_ticker(t: &Ticker) -> Self {
Self {
ts_ms: t.timestamp,
last: t.last_price,
bid: opt_f64(t.bid_price),
ask: opt_f64(t.ask_price),
bid_qty: opt_f64(t.bid_qty),
ask_qty: opt_f64(t.ask_qty),
high_24h: opt_f64(t.high_24h),
low_24h: opt_f64(t.low_24h),
open_price: opt_f64(t.open_price),
prev_close_price: opt_f64(t.prev_close_price),
vol_24h: opt_f64(t.volume_24h),
quote_vol_24h: opt_f64(t.quote_volume_24h),
change_pct_24h: opt_f64(t.price_change_percent_24h),
weighted_avg_price: opt_f64(t.weighted_avg_price),
last_qty: opt_f64(t.last_qty),
mark_price: opt_f64(t.mark_price),
index_price: opt_f64(t.index_price),
open_interest: opt_f64(t.open_interest),
funding_rate: opt_f64(t.funding_rate),
count: t.count,
}
}
}
impl DataPoint for TickerIndicatorsPoint {
const RECORD_SIZE: usize = INDICATORS_SIZE;
fn encode(&self, out: &mut [u8]) {
out[0..8].copy_from_slice(&(self.ts_ms as u64).to_le_bytes());
out[8..16].copy_from_slice(&self.last.to_le_bytes());
out[16..24].copy_from_slice(&self.bid.to_le_bytes());
out[24..32].copy_from_slice(&self.ask.to_le_bytes());
out[32..40].copy_from_slice(&self.bid_qty.to_le_bytes());
out[40..48].copy_from_slice(&self.ask_qty.to_le_bytes());
out[48..56].copy_from_slice(&self.high_24h.to_le_bytes());
out[56..64].copy_from_slice(&self.low_24h.to_le_bytes());
out[64..72].copy_from_slice(&self.open_price.to_le_bytes());
out[72..80].copy_from_slice(&self.prev_close_price.to_le_bytes());
out[80..88].copy_from_slice(&self.vol_24h.to_le_bytes());
out[88..96].copy_from_slice(&self.quote_vol_24h.to_le_bytes());
out[96..104].copy_from_slice(&self.change_pct_24h.to_le_bytes());
out[104..112].copy_from_slice(&self.weighted_avg_price.to_le_bytes());
out[112..120].copy_from_slice(&self.last_qty.to_le_bytes());
out[120..128].copy_from_slice(&self.mark_price.to_le_bytes());
out[128..136].copy_from_slice(&self.index_price.to_le_bytes());
out[136..144].copy_from_slice(&self.open_interest.to_le_bytes());
out[144..152].copy_from_slice(&self.funding_rate.to_le_bytes());
out[152..160].copy_from_slice(&opt_i64_enc(self.count).to_le_bytes());
}
fn decode(bytes: &[u8]) -> Option<Self> {
if bytes.len() != INDICATORS_SIZE {
return None;
}
Some(Self {
ts_ms: u64::from_le_bytes(bytes[0..8].try_into().ok()?) as i64,
last: f64::from_le_bytes(bytes[8..16].try_into().ok()?),
bid: f64::from_le_bytes(bytes[16..24].try_into().ok()?),
ask: f64::from_le_bytes(bytes[24..32].try_into().ok()?),
bid_qty: f64::from_le_bytes(bytes[32..40].try_into().ok()?),
ask_qty: f64::from_le_bytes(bytes[40..48].try_into().ok()?),
high_24h: f64::from_le_bytes(bytes[48..56].try_into().ok()?),
low_24h: f64::from_le_bytes(bytes[56..64].try_into().ok()?),
open_price: f64::from_le_bytes(bytes[64..72].try_into().ok()?),
prev_close_price: f64::from_le_bytes(bytes[72..80].try_into().ok()?),
vol_24h: f64::from_le_bytes(bytes[80..88].try_into().ok()?),
quote_vol_24h: f64::from_le_bytes(bytes[88..96].try_into().ok()?),
change_pct_24h: f64::from_le_bytes(bytes[96..104].try_into().ok()?),
weighted_avg_price: f64::from_le_bytes(bytes[104..112].try_into().ok()?),
last_qty: f64::from_le_bytes(bytes[112..120].try_into().ok()?),
mark_price: f64::from_le_bytes(bytes[120..128].try_into().ok()?),
index_price: f64::from_le_bytes(bytes[128..136].try_into().ok()?),
open_interest: f64::from_le_bytes(bytes[136..144].try_into().ok()?),
funding_rate: f64::from_le_bytes(bytes[144..152].try_into().ok()?),
count: opt_i64_dec(u64::from_le_bytes(bytes[152..160].try_into().ok()?)),
})
}
fn timestamp_ms(&self) -> i64 { self.ts_ms }
fn from_stream_event(ev: &StreamEvent) -> Option<Self> {
if let StreamEvent::Ticker { ticker, .. } = ev {
Some(Self::from_ticker(ticker))
} else {
None
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TickerFullPoint {
pub ts_ms: i64,
pub last: f64,
pub bid: f64,
pub ask: f64,
pub bid_qty: f64,
pub ask_qty: f64,
pub high_24h: f64,
pub low_24h: f64,
pub vol_24h: f64,
pub quote_vol_24h: f64,
pub price_change_24h: f64,
pub price_change_pct_24h: f64,
pub open_price: f64,
pub prev_close_price: f64,
pub prev_price_24h: f64,
pub prev_price_1h: f64,
pub weighted_avg_price: f64,
pub open_utc: f64,
pub turnover_24h: f64,
pub first_id: Option<i64>,
pub last_id: Option<i64>,
pub count: Option<i64>,
pub mark_price: f64,
pub index_price: f64,
pub open_interest: f64,
pub open_interest_value: f64,
pub single_open_interest: f64,
pub funding_rate: f64,
pub next_funding_time: Option<i64>,
pub funding_interval_hour: f64,
pub funding_cap: f64,
pub basis: f64,
pub basis_rate: f64,
pub predicted_delivery_price: f64,
pub delivery_time: Option<i64>,
pub settlement_price: f64,
pub funding_8h: f64,
pub min_price: f64,
pub max_price: f64,
pub volume_notional: f64,
pub last_qty: f64,
pub interest_value: f64,
pub last_trade_time: Option<i64>,
pub open_time: Option<i64>,
pub update_id: Option<i64>,
pub state: Option<String>,
}
const FULL_BLOB_OFFSET: usize = 364;
const FULL_SIZE: usize = 376;
fn encode_ticker_full_numeric(p: &TickerFullPoint, out: &mut [u8]) {
let mut o = 0usize;
macro_rules! w8 {
($v:expr) => { out[o..o+8].copy_from_slice(&$v.to_le_bytes()); o += 8; };
}
w8!((p.ts_ms as u64));
w8!(p.last);
w8!(p.bid);
w8!(p.ask);
w8!(p.bid_qty);
w8!(p.ask_qty);
w8!(p.high_24h);
w8!(p.low_24h);
w8!(p.vol_24h);
w8!(p.quote_vol_24h);
w8!(p.price_change_24h);
w8!(p.price_change_pct_24h);
w8!(p.open_price);
w8!(p.prev_close_price);
w8!(p.prev_price_24h);
w8!(p.prev_price_1h);
w8!(p.weighted_avg_price);
w8!(p.open_utc);
w8!(p.turnover_24h);
w8!(opt_i64_enc(p.first_id));
w8!(opt_i64_enc(p.last_id));
w8!(opt_i64_enc(p.count));
w8!(p.mark_price);
w8!(p.index_price);
w8!(p.open_interest);
w8!(p.open_interest_value);
w8!(p.single_open_interest);
w8!(p.funding_rate);
w8!(opt_i64_enc(p.next_funding_time));
w8!(p.funding_interval_hour);
w8!(p.funding_cap);
w8!(p.basis);
w8!(p.basis_rate);
w8!(p.predicted_delivery_price);
w8!(opt_i64_enc(p.delivery_time));
w8!(p.settlement_price);
w8!(p.funding_8h);
w8!(p.min_price);
w8!(p.max_price);
w8!(p.volume_notional);
w8!(p.last_qty);
w8!(p.interest_value);
w8!(opt_i64_enc(p.last_trade_time));
w8!(opt_i64_enc(p.open_time));
w8!(opt_i64_enc(p.update_id));
debug_assert_eq!(o, FULL_BLOB_OFFSET);
}
fn decode_ticker_full_numeric(bytes: &[u8]) -> Option<TickerFullPoint> {
if bytes.len() < FULL_BLOB_OFFSET {
return None;
}
let mut o = 0usize;
macro_rules! rf64 {
() => {{ let v = f64::from_le_bytes(bytes[o..o+8].try_into().ok()?); o += 8; v }};
}
macro_rules! ru64 {
() => {{ let v = u64::from_le_bytes(bytes[o..o+8].try_into().ok()?); o += 8; v }};
}
let result = Some(TickerFullPoint {
ts_ms: ru64!() as i64,
last: rf64!(),
bid: rf64!(),
ask: rf64!(),
bid_qty: rf64!(),
ask_qty: rf64!(),
high_24h: rf64!(),
low_24h: rf64!(),
vol_24h: rf64!(),
quote_vol_24h: rf64!(),
price_change_24h: rf64!(),
price_change_pct_24h: rf64!(),
open_price: rf64!(),
prev_close_price: rf64!(),
prev_price_24h: rf64!(),
prev_price_1h: rf64!(),
weighted_avg_price: rf64!(),
open_utc: rf64!(),
turnover_24h: rf64!(),
first_id: opt_i64_dec(ru64!()),
last_id: opt_i64_dec(ru64!()),
count: opt_i64_dec(ru64!()),
mark_price: rf64!(),
index_price: rf64!(),
open_interest: rf64!(),
open_interest_value: rf64!(),
single_open_interest: rf64!(),
funding_rate: rf64!(),
next_funding_time: opt_i64_dec(ru64!()),
funding_interval_hour: rf64!(),
funding_cap: rf64!(),
basis: rf64!(),
basis_rate: rf64!(),
predicted_delivery_price: rf64!(),
delivery_time: opt_i64_dec(ru64!()),
settlement_price: rf64!(),
funding_8h: rf64!(),
min_price: rf64!(),
max_price: rf64!(),
volume_notional: rf64!(),
last_qty: rf64!(),
interest_value: rf64!(),
last_trade_time: opt_i64_dec(ru64!()),
open_time: opt_i64_dec(ru64!()),
update_id: opt_i64_dec(ru64!()),
state: None, });
debug_assert_eq!(o, FULL_BLOB_OFFSET, "ticker full decode offset mismatch");
result
}
impl TickerFullPoint {
pub fn from_ticker(t: &Ticker) -> Self {
Self {
ts_ms: t.timestamp,
last: t.last_price,
bid: opt_f64(t.bid_price),
ask: opt_f64(t.ask_price),
bid_qty: opt_f64(t.bid_qty),
ask_qty: opt_f64(t.ask_qty),
high_24h: opt_f64(t.high_24h),
low_24h: opt_f64(t.low_24h),
vol_24h: opt_f64(t.volume_24h),
quote_vol_24h: opt_f64(t.quote_volume_24h),
price_change_24h: opt_f64(t.price_change_24h),
price_change_pct_24h: opt_f64(t.price_change_percent_24h),
open_price: opt_f64(t.open_price),
prev_close_price: opt_f64(t.prev_close_price),
prev_price_24h: opt_f64(t.prev_price_24h),
prev_price_1h: opt_f64(t.prev_price_1h),
weighted_avg_price: opt_f64(t.weighted_avg_price),
open_utc: opt_f64(t.open_utc),
turnover_24h: opt_f64(t.turnover_24h),
first_id: t.first_id,
last_id: t.last_id,
count: t.count,
mark_price: opt_f64(t.mark_price),
index_price: opt_f64(t.index_price),
open_interest: opt_f64(t.open_interest),
open_interest_value: opt_f64(t.open_interest_value),
single_open_interest: opt_f64(t.single_open_interest),
funding_rate: opt_f64(t.funding_rate),
next_funding_time: t.next_funding_time,
funding_interval_hour: opt_f64(t.funding_interval_hour),
funding_cap: opt_f64(t.funding_cap),
basis: opt_f64(t.basis),
basis_rate: opt_f64(t.basis_rate),
predicted_delivery_price: opt_f64(t.predicted_delivery_price),
delivery_time: t.delivery_time,
settlement_price: opt_f64(t.settlement_price),
funding_8h: opt_f64(t.funding_8h),
min_price: opt_f64(t.min_price),
max_price: opt_f64(t.max_price),
volume_notional: opt_f64(t.volume_notional),
last_qty: opt_f64(t.last_qty),
interest_value: opt_f64(t.interest_value),
last_trade_time: t.last_trade_time,
open_time: t.open_time,
update_id: t.update_id,
state: t.state.clone(),
}
}
}
impl DataPoint for TickerFullPoint {
const RECORD_SIZE: usize = FULL_SIZE;
fn encode(&self, out: &mut [u8]) {
encode_ticker_full_numeric(self, out);
}
fn decode(bytes: &[u8]) -> Option<Self> {
if bytes.len() != FULL_SIZE {
return None;
}
decode_ticker_full_numeric(bytes)
}
fn timestamp_ms(&self) -> i64 { self.ts_ms }
fn from_stream_event(ev: &StreamEvent) -> Option<Self> {
if let StreamEvent::Ticker { ticker, .. } = ev {
Some(Self::from_ticker(ticker))
} else {
None
}
}
fn encode_blob(&self) -> Option<Vec<u8>> {
let s = self.state.as_deref().unwrap_or("");
let bytes = s.as_bytes();
let len = bytes.len().min(u16::MAX as usize);
let mut blob = Vec::with_capacity(2 + len);
blob.extend_from_slice(&(len as u16).to_le_bytes());
blob.extend_from_slice(&bytes[..len]);
Some(blob)
}
fn decode_blob(header: &[u8], blob: &[u8]) -> Option<Self> {
let mut p = decode_ticker_full_numeric(header)?;
if blob.len() >= 2 {
let slen = u16::from_le_bytes(blob[0..2].try_into().ok()?) as usize;
if blob.len() >= 2 + slen {
let s = std::str::from_utf8(&blob[2..2 + slen]).ok()?;
if !s.is_empty() {
p.state = Some(s.to_owned());
}
}
}
Some(p)
}
fn blob_pointer_offset() -> Option<usize> { Some(FULL_BLOB_OFFSET) }
}