data-pipelines/src/encoding.rs

use std::num::{NonZeroU64, NonZeroU8, NonZeroI32};
use std::mem::size_of;
use std::convert::TryFrom;
use serde::{Serialize, Deserialize};
use markets::crypto::{Exchange, Currency, Ticker, Side};

mod try_from_u8 {
    use std::convert::TryFrom;
    use std::fmt;
    use std::marker::PhantomData;
    use serde::{Serializer, Deserializer};
    use serde::de::Visitor;
    use serde::ser::Error as SerError;

    struct V<T>(PhantomData<T>);

    impl<'de, T> Visitor<'de> for V<T>
        where T: TryFrom<u8>
    {
        type Value = T;

        fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
            formatter.write_str("an integer code between 1-255")
        }

        fn visit_u8<E>(self, v: u8) -> Result<Self::Value, E>
            where E: serde::de::Error,
        {
            match T::try_from(v) {
                Ok(v) => Ok(v),
                Err(_) => {
                    Err(serde::de::Error::custom("Invalid code"))
                }
            }
        }

        fn visit_u64<E>(self, v: u64) -> Result<Self::Value, E>
            where E: serde::de::Error,
        {
            if v > 255 {
                return Err(serde::de::Error::custom("Value greater than 255"))
            }
            match T::try_from(v as u8) {
                Ok(v) => Ok(v),
                Err(_) => {
                    Err(serde::de::Error::custom("Invalid code"))
                }
            }
        }
    }


    pub fn deserialize<'de, D, T>(deserializer: D) -> Result<T, D::Error>
        where D: Deserializer<'de>,
              T: TryFrom<u8>
    {
        deserializer.deserialize_u8(V(PhantomData))
    }

    pub fn serialize<S, T>(item: &T, serializer: S) -> Result<S::Ok, S::Error>
        where S: Serializer,
              T: Copy,
              u8: From<T>
    {
        match u8::from(*item) {
            0 => Err(S::Error::custom("not implemented: no code for variant or value")),
            x => serializer.serialize_u8(x)
        }
    }
}

#[derive(Deserialize, Serialize, Debug, Clone)]
pub struct Serde32BytesTrade {
    pub time: u64,
    #[serde(with = "try_from_u8")]
    pub exch: Exchange,
    #[serde(with = "try_from_u8")]
    pub ticker: Ticker,
    pub price: f64,
    pub amount: f64,
    pub side: Option<Side>,
    pub server_time: Option<NonZeroI32>,
}

/// Represents the serialized form of a trades row
///
/// ```console,ignore
///                      1                   2                   3
///  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
/// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
/// |e|b|q|s| srvtm | time: u64     | price: f64    | amount: f64   |
/// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
///  | | | | |
///  | | | | |
///  | | | | |
///  | | | | -> server_time: Option<i32> - 0=None, other=nano offset from `time`
///  | | | |
///  | | | -> side: Option<Side> - 0=None, 1=Bid, 2=Ask
///  | | |  
///  | | -> quote: Currency - see markets::crypto for u8 <-> currency codes
///  | |    
///  | -> base: Currency - see markets::crypto for u8 <-> currency codes
///  |     
///  -> exch: Exchange - see markets::crypto for u8 <-> exchange codes
///
///  ```
///
#[derive(Debug, Clone)]
pub struct PackedTrade {
    pub exch: u8,
    pub base: u8,
    pub quote: u8,

    /// 0=None
    pub side: u8,

    /// relative offset from `time`; 0=None
    pub server_time: i32,

    pub time: u64,
    pub price: f64,
    pub amount: f64,
}

#[derive(Debug, Clone)]
pub struct ParseError(Box<String>);

/// Pull out individual fields on demand from the serialized bytes of a PackedTrade
#[repr(align(32))]
pub struct PackedTradeData<'a>(&'a [u8]);

impl<'a> PackedTradeData<'a> {

    const EXCH_OFFSET           : usize = 0;
    const BASE_OFFSET           : usize = 1;
    const QUOTE_OFFSET          : usize = 2;
    const SIDE_OFFSET           : usize = 3;
    const SERVER_TIME_OFFSET    : usize = 4;
    const TIME_OFFSET           : usize = 8;
    const PRICE_OFFSET          : usize = 16;
    const AMOUNT_OFFSET         : usize = 24;

    #[inline]
    pub fn exch(&self) -> Result<Exchange, markets::crypto::Error> {
        Exchange::try_from(self.0[Self::EXCH_OFFSET])
    }

    #[inline]
    pub fn base(&self) -> Result<Currency, markets::crypto::Error> {
        Currency::try_from(self.0[Self::BASE_OFFSET])
    }

    #[inline]
    pub fn quote(&self) -> Result<Currency, markets::crypto::Error> {
        Currency::try_from(self.0[Self::QUOTE_OFFSET])
    }

    #[inline]
    pub fn side(&self) -> Result<Option<Side>, markets::crypto::Error> {
        match self.0[Self::SIDE_OFFSET] {
            0 => Ok(None),
            other => Ok(Some(Side::try_from(other)?)),
        }
    }

    #[inline]
    pub fn time(&self) -> Result<u64, ParseError> {
        atoi::atoi(&self.0[Self::TIME_OFFSET..(Self::TIME_OFFSET + 8)])
            .ok_or_else(|| {
                ParseError(Box::new(format!("failed to parse integer: '{}'",
                    std::str::from_utf8(&self.0[Self::TIME_OFFSET..(Self::TIME_OFFSET + 8)]).unwrap_or("uft8 error")
                )))
            })
    }

    #[inline]
    pub fn price(&self) -> Result<f64, lexical::Error> {
        lexical::parse(&self.0[Self::PRICE_OFFSET..(Self::PRICE_OFFSET + 8)])
    }

    #[inline]
    pub fn amount(&self) -> Result<f64, lexical::Error> {
        lexical::parse(&self.0[Self::AMOUNT_OFFSET..(Self::AMOUNT_OFFSET + 8)])
    }

    /// `server_time` is stored in milliseconds, while `time` is nanoseconds.
    /// this is what you need to multiply the stored `server_time` data by to
    /// get it back to nanoseconds.
    const SERVER_TIME_DOWNSCALE_FACTOR: u64 = 1_000_000;

    #[inline]
    pub fn server_time(&self) -> Result<Option<u64>, ParseError> {
        let st: i32 =
            atoi::atoi(&self.0[Self::SERVER_TIME_OFFSET..(Self::SERVER_TIME_OFFSET + 4)])
                .ok_or_else(|| {
                    ParseError(Box::new(format!("failed to parse integer: '{}'",
                        std::str::from_utf8(&self.0[Self::SERVER_TIME_OFFSET..(Self::SERVER_TIME_OFFSET + 4)]).unwrap_or("uft8 error")
                    )))
                })?;

        // while the `server_time` delta is stored as a signed integer, to be able to express a
        // delta in both directions relative to `time`, we can't just add a negative `i64` to a
        // `u64`, it doesn't work like that. this match either subtracts the absolute value of a
        // negative delta, or adds a positive delta, to get around this conundrum.
        //
        // `SERVER_TIME_DOWNSCALE_FACTOR` is used to rescale the delta to nanoseconds prior to its
        // being applied to `time`.

        match st {
            0 => Ok(None),

            x @ std::i32::MIN .. 0  => Ok(Some(self.time()? - (x.abs() as u64 * Self::SERVER_TIME_DOWNSCALE_FACTOR))),

            x @ 1 ..= std::i32::MAX => Ok(Some(self.time()? + (x as u64 * Self::SERVER_TIME_DOWNSCALE_FACTOR))),
        }
    }
}


#[allow(unused)]
#[cfg(test)]
mod tests {
    use super::*;
    use markets::{e, t, c};

    #[test]
    fn verify_packed_trade_is_32_bytes() {
        assert_eq!(size_of::<PackedTrade>(), 32);
    }

    #[test]
    fn check_bincode_serialized_size() {
        let trade = Serde32BytesTrade {
            time: 1586996977191449698,
            exch: e!(bmex),
            ticker: t!(btc-usd),
            price: 1.234,
            amount: 4.567,
            side: None,
            server_time: NonZeroI32::new(1_000_000),
        };

        assert_eq!(size_of::<Serde32BytesTrade>(), 32);
        assert_eq!(bincode::serialized_size(&trade).unwrap(), 32);
    }

    #[test]
    fn example_of_36_byte_trades_struct_without_the_offset_i32() {
        #[repr(packed)]
        pub struct Trade36 {
            pub exch: Exchange,
            pub ticker: Ticker,
            pub side: Option<Side>,

            pub time: u64,
            pub price: f64,
            pub amount: f64,

            pub server_time: Option<NonZeroU64>,
        }

        assert_eq!(size_of::<Trade36>(), 36);
    }
}