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 Trade {
    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)])
    }

    #[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")
                    )))
                })?;
        match st {
            0 => Ok(None),

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

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


#[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 = Trade {
            time: 1586996977191449698,
            exch: e!(bmex),
            ticker: t!(btc-usd),
            price: 1.234,
            amount: 4.567,
            side: None,
            //server_time: NonZeroU64::new(1586996977191449698 + 1_000_000),
            server_time: NonZeroI32::new(1_000_000),
            //server_time: Some(1586996977191449698 + 1_000_000),
        };

        /*
        let packed = PackedTrade {
            time: trade.time,
            exch: u8::from(trade.exch),
            base: u8::from(trade.ticker.base),
            quote: u8::from(trade.ticker.quote),
            amount: trade.amount,
            price: trade.price,
            //side: trade.side.and_then(|s| NonZeroU8::new(u8::from(s))),
            //server_time: NonZeroI32::new(1_000_000),
            side: trade.side.map(|s| u8::from(s)).unwrap_or(0),
            server_time: 1_000_000,

        };
        */

        assert_eq!(size_of::<Trade>(), 32);

        //assert_eq!(serde_json::to_string(&trade).unwrap().len(), 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);
    }
}