digdigdig3_station/data/
balance_update.rs

1use digdigdig3::core::types::StreamEvent;
2use serde::{Deserialize, Serialize};
3
4use crate::series::DataPoint;
5
6/// On-disk record (48 bytes LE) for an account balance change event:
7///   u64  ts_ms       (8)
8///   u64  asset_hash  (8)  — fnv1a-64 of asset string (e.g. "USDT")
9///   f64  free        (8)  — available balance after change
10///   f64  locked      (8)  — locked/reserved balance after change
11///   f64  total       (8)  — free + locked
12///   f64  delta       (8)  — signed change (positive = credit, negative = debit, 0.0 if unknown)
13/// Total: 48 bytes
14#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
15pub struct BalanceUpdatePoint {
16    pub ts_ms: i64,
17    /// fnv1a-64 hash of the asset ticker string.  Use `asset` field for display;
18    /// `asset_hash` is the compact storage form.
19    pub asset_hash: u64,
20    pub free: f64,
21    pub locked: f64,
22    pub total: f64,
23    pub delta: f64,
24}
25
26const SIZE: usize = 48;
27
28impl DataPoint for BalanceUpdatePoint {
29    const RECORD_SIZE: usize = SIZE;
30
31    fn encode(&self, out: &mut [u8]) {
32        out[0..8].copy_from_slice(&(self.ts_ms as u64).to_le_bytes());
33        out[8..16].copy_from_slice(&self.asset_hash.to_le_bytes());
34        out[16..24].copy_from_slice(&self.free.to_le_bytes());
35        out[24..32].copy_from_slice(&self.locked.to_le_bytes());
36        out[32..40].copy_from_slice(&self.total.to_le_bytes());
37        out[40..48].copy_from_slice(&self.delta.to_le_bytes());
38    }
39
40    fn decode(bytes: &[u8]) -> Option<Self> {
41        if bytes.len() != SIZE {
42            return None;
43        }
44        let ts_ms = u64::from_le_bytes(bytes[0..8].try_into().ok()?) as i64;
45        let asset_hash = u64::from_le_bytes(bytes[8..16].try_into().ok()?);
46        let free = f64::from_le_bytes(bytes[16..24].try_into().ok()?);
47        let locked = f64::from_le_bytes(bytes[24..32].try_into().ok()?);
48        let total = f64::from_le_bytes(bytes[32..40].try_into().ok()?);
49        let delta = f64::from_le_bytes(bytes[40..48].try_into().ok()?);
50        Some(Self { ts_ms, asset_hash, free, locked, total, delta })
51    }
52
53    fn timestamp_ms(&self) -> i64 {
54        self.ts_ms
55    }
56
57    fn from_stream_event(ev: &StreamEvent) -> Option<Self> {
58        if let StreamEvent::BalanceUpdate(event) = ev {
59            Some(Self {
60                ts_ms: event.timestamp,
61                asset_hash: fnv1a_64(event.asset.as_bytes()),
62                free: event.free,
63                locked: event.locked,
64                total: event.total,
65                delta: event.delta.unwrap_or(0.0),
66            })
67        } else {
68            None
69        }
70    }
71}
72
73fn fnv1a_64(bytes: &[u8]) -> u64 {
74    let mut h: u64 = 0xcbf29ce484222325;
75    for b in bytes {
76        h ^= *b as u64;
77        h = h.wrapping_mul(0x100000001b3);
78    }
79    h
80}
digdigdig3_station/data/balance_update.rs

digdigdig3_station/data/
balance_update.rs
