polyester/codecs/decode/
market_data.rs

1//! Public market-data decoders (trades / candles).
2
3use serde_json::Value;
4
5use super::common::api_data_from_proto;
6use super::money::{decode_price_ticks, decode_qty_scaled};
7use crate::codecs::scalars::{format_price_ticks, format_qty_scaled};
8use crate::errors::{Error, Result};
9use crate::models::{Candle, CandlesResult, MarketTrade, MarketTradesResult, SpotConfig};
10use crate::proto::marketdata::v1::{
11    CandlePoint, GetCandlesColumnsResponse, GetCandlesResponse, GetSpotConfigResponse,
12    GetTradesResponse, MarketTrade as ProtoMarketTrade, Timeframe,
13};
14
15pub fn spot_config_from_proto(msg: &GetSpotConfigResponse) -> SpotConfig {
16    let mut raw = api_data_from_proto(msg).raw;
17    // proto3 omits scalar zeroes during proto-JSON conversion. Quantity scale
18    // zero is nevertheless valid, so restore the typed wire value rather than
19    // treating an omitted JSON key as an unknown scale.
20    if let Some(pairs) = raw.get_mut("pairs").and_then(Value::as_array_mut) {
21        for (pair, typed) in pairs.iter_mut().zip(&msg.pairs) {
22            if let Some(object) = pair.as_object_mut() {
23                // Proto-JSON uses camelCase. Writing the snake_case alias as a
24                // second key makes serde reject the object as a duplicate field
25                // when consumers re-deserialize into generated PairConfig types.
26                object.remove("base_quantity_scale");
27                object.insert(
28                    "baseQuantityScale".to_owned(),
29                    Value::from(typed.base_quantity_scale),
30                );
31                object.remove("quote_quantity_scale");
32                object.insert(
33                    "quoteQuantityScale".to_owned(),
34                    Value::from(typed.quote_quantity_scale),
35                );
36            }
37        }
38    }
39    SpotConfig { raw }
40}
41
42pub fn timeframe_label(tf: Timeframe) -> &'static str {
43    match tf {
44        Timeframe::Sec1 => "1s",
45        Timeframe::Min1 => "1m",
46        Timeframe::Min5 => "5m",
47        Timeframe::Min15 => "15m",
48        Timeframe::Min30 => "30m",
49        Timeframe::Hour1 => "1h",
50        Timeframe::Hour4 => "4h",
51        Timeframe::Hour12 => "12h",
52        Timeframe::Day1 => "1d",
53        Timeframe::Week1 => "1w",
54        Timeframe::Month1 => "1mo",
55        Timeframe::TimeframeUnspecified => "",
56    }
57}
58
59fn enum_value_timeframe(value: buffa::EnumValue<Timeframe>) -> String {
60    value
61        .as_known()
62        .map(|known| timeframe_label(known).to_owned())
63        .unwrap_or_else(|| format!("UNKNOWN({})", value.to_i32()))
64}
65
66pub fn market_trade_from_proto(msg: &ProtoMarketTrade, quantity_scale: u32) -> MarketTrade {
67    let symbol_id = msg.symbol_id;
68    let symbol_id_opt = if symbol_id == 0 {
69        None
70    } else {
71        Some(symbol_id)
72    };
73    MarketTrade {
74        symbol_id,
75        match_id: if msg.match_id == 0 {
76            String::new()
77        } else {
78            msg.match_id.to_string()
79        },
80        price: decode_price_ticks(msg.price_ticks, None),
81        qty: decode_qty_scaled(msg.qty_scaled, Some(quantity_scale), None, symbol_id_opt),
82        ts_ns: if msg.ts_ns == 0 {
83            String::new()
84        } else {
85            msg.ts_ns.to_string()
86        },
87        side: if msg.is_buy {
88            "buy".to_owned()
89        } else {
90            "sell".to_owned()
91        },
92    }
93}
94
95pub fn market_trades_from_proto(
96    msg: &GetTradesResponse,
97    quantity_scale: u32,
98) -> MarketTradesResult {
99    MarketTradesResult {
100        trades: msg
101            .trades
102            .iter()
103            .map(|trade| market_trade_from_proto(trade, quantity_scale))
104            .collect(),
105        next_page_token: msg.next_page_token.clone(),
106    }
107}
108
109pub fn candle_point_from_proto(
110    msg: &CandlePoint,
111    volume_scale: u32,
112    symbol_id: u32,
113    timeframe: &str,
114) -> Result<Candle> {
115    Ok(Candle {
116        ts_sec: msg.ts_sec as i64,
117        open: format_price_ticks(msg.open),
118        high: format_price_ticks(msg.high),
119        low: format_price_ticks(msg.low),
120        close: format_price_ticks(msg.close),
121        volume: format_qty_scaled(msg.volume, volume_scale)
122            .map_err(|e| Error::validation(format!("candle volume scale invalid: {e}")))?,
123        symbol_id,
124        timeframe: timeframe.to_owned(),
125    })
126}
127
128pub fn candles_from_proto(msg: &GetCandlesResponse, volume_scale: u32) -> Result<CandlesResult> {
129    let timeframe = enum_value_timeframe(msg.timeframe);
130    let symbol_id = msg.symbol_id;
131    let mut candles = Vec::with_capacity(msg.candles.len());
132    for c in &msg.candles {
133        candles.push(candle_point_from_proto(
134            c,
135            volume_scale,
136            symbol_id,
137            &timeframe,
138        )?);
139    }
140    Ok(CandlesResult {
141        symbol_id,
142        timeframe,
143        candles,
144        next_page_token: msg.next_page_token.clone(),
145    })
146}
147
148/// Decode columnar OHLCV into row-oriented [`CandlesResult`] (Go `CandlesColumnsFromProto`).
149pub fn candles_columns_from_proto(
150    msg: &GetCandlesColumnsResponse,
151    volume_scale: u32,
152) -> Result<CandlesResult> {
153    let rows = msg.ts_sec.len();
154    let lengths = [
155        ("open", msg.open.len()),
156        ("high", msg.high.len()),
157        ("low", msg.low.len()),
158        ("close", msg.close.len()),
159        ("volume", msg.volume.len()),
160    ];
161    if lengths.iter().any(|(_, len)| *len != rows) {
162        return Err(Error::transport(format!(
163            "invalid GetCandlesColumns response lengths: ts_sec={rows}, open={}, high={}, low={}, close={}, volume={}",
164            msg.open.len(),
165            msg.high.len(),
166            msg.low.len(),
167            msg.close.len(),
168            msg.volume.len()
169        )));
170    }
171
172    let timeframe = enum_value_timeframe(msg.timeframe);
173    let symbol_id = msg.symbol_id;
174    let mut candles = Vec::with_capacity(rows);
175    for (i, &ts) in msg.ts_sec.iter().enumerate() {
176        let volume = format_qty_scaled(msg.volume[i], volume_scale)
177            .map_err(|e| Error::validation(format!("candle volume scale invalid: {e}")))?;
178        candles.push(Candle {
179            ts_sec: ts as i64,
180            open: format_price_ticks(msg.open[i]),
181            high: format_price_ticks(msg.high[i]),
182            low: format_price_ticks(msg.low[i]),
183            close: format_price_ticks(msg.close[i]),
184            volume,
185            symbol_id,
186            timeframe: timeframe.clone(),
187        });
188    }
189    Ok(CandlesResult {
190        symbol_id,
191        timeframe,
192        candles,
193        next_page_token: msg.next_page_token.clone(),
194    })
195}
196
197#[cfg(test)]
198mod tests {
199    use super::*;
200
201    #[test]
202    fn market_trades_maps_side_and_ids() {
203        let msg = GetTradesResponse {
204            trades: vec![ProtoMarketTrade {
205                symbol_id: 3,
206                match_id: 99,
207                is_buy: true,
208                price_ticks: 1_500_000,
209                qty_scaled: 100,
210                ts_ns: 42,
211                ..Default::default()
212            }],
213            next_page_token: "page-2".into(),
214            ..Default::default()
215        };
216        let list = market_trades_from_proto(&msg, 6);
217        assert_eq!(list.trades.len(), 1);
218        assert_eq!(list.next_page_token, "page-2");
219        let t = &list.trades[0];
220        assert_eq!(t.symbol_id, 3);
221        assert_eq!(t.match_id, "99");
222        assert_eq!(t.side, "buy");
223        assert_eq!(t.price.as_ref().unwrap().as_ticks(), 1_500_000);
224        assert_eq!(t.qty.as_ref().unwrap().as_scaled(), 100);
225        assert_eq!(t.qty.as_ref().unwrap().format(None).unwrap(), "0.0001");
226        assert_eq!(t.ts_ns, "42");
227    }
228
229    #[test]
230    fn spot_config_preserves_valid_zero_quantity_scale() {
231        let msg = GetSpotConfigResponse {
232            pairs: vec![crate::proto::marketdata::v1::PairConfig {
233                symbol: "WHOLE-USDT".into(),
234                symbol_id: 9,
235                base_quantity_scale: 0,
236                quote_quantity_scale: 0,
237                ..Default::default()
238            }],
239            ..Default::default()
240        };
241        let spot = spot_config_from_proto(&msg);
242        assert_eq!(spot.raw["pairs"][0]["baseQuantityScale"], 0);
243        assert_eq!(spot.raw["pairs"][0]["quoteQuantityScale"], 0);
244        assert!(spot.raw["pairs"][0].get("base_quantity_scale").is_none());
245        assert!(spot.raw["pairs"][0].get("quote_quantity_scale").is_none());
246        let round_trip: GetSpotConfigResponse =
247            serde_json::from_value(spot.raw.clone()).expect("spot config round-trip");
248        assert_eq!(round_trip.pairs[0].base_quantity_scale, 0);
249        assert_eq!(round_trip.pairs[0].quote_quantity_scale, 0);
250    }
251
252    #[test]
253    fn unknown_timeframe_preserves_numeric_value() {
254        let msg = GetCandlesResponse {
255            timeframe: buffa::EnumValue::from(77),
256            ..Default::default()
257        };
258        assert_eq!(
259            candles_from_proto(&msg, 8).unwrap().timeframe,
260            "UNKNOWN(77)"
261        );
262    }
263
264    #[test]
265    fn candles_format_ohlcv() {
266        let msg = GetCandlesResponse {
267            symbol_id: 1,
268            timeframe: Timeframe::Min1.into(),
269            candles: vec![CandlePoint {
270                ts_sec: 10,
271                open: 1_000_000,
272                high: 2_000_000,
273                low: 500_000,
274                close: 1_500_000,
275                volume: 100_000_000,
276                ..Default::default()
277            }],
278            ..Default::default()
279        };
280        let result = candles_from_proto(&msg, 8).expect("candles");
281        assert_eq!(result.symbol_id, 1);
282        assert_eq!(result.timeframe, "1m");
283        assert_eq!(result.candles.len(), 1);
284        let c = &result.candles[0];
285        assert_eq!(c.ts_sec, 10);
286        assert_eq!(c.open, "1");
287        assert_eq!(c.high, "2");
288        assert_eq!(c.low, "0.5");
289        assert_eq!(c.close, "1.5");
290        assert_eq!(c.volume, "1");
291    }
292
293    #[test]
294    fn candle_decode_rejects_invalid_volume_scale() {
295        let msg = GetCandlesResponse {
296            symbol_id: 1,
297            timeframe: Timeframe::Min1.into(),
298            candles: vec![CandlePoint {
299                ts_sec: 10,
300                volume: 1,
301                ..Default::default()
302            }],
303            ..Default::default()
304        };
305        let err = candles_from_proto(&msg, 65535).expect_err("invalid scale");
306        assert!(err.to_string().to_ascii_lowercase().contains("scale"));
307    }
308
309    #[test]
310    fn candles_columns_decode_rejects_invalid_volume_scale() {
311        let msg = GetCandlesColumnsResponse {
312            symbol_id: 1,
313            timeframe: Timeframe::Min1.into(),
314            ts_sec: vec![10],
315            open: vec![1],
316            high: vec![1],
317            low: vec![1],
318            close: vec![1],
319            volume: vec![1],
320            ..Default::default()
321        };
322        let err = candles_columns_from_proto(&msg, 65535).expect_err("invalid scale");
323        assert!(err.to_string().to_ascii_lowercase().contains("scale"));
324    }
325
326    #[test]
327    fn candles_columns_rejects_short_parallel_arrays() {
328        let msg = GetCandlesColumnsResponse {
329            symbol_id: 1,
330            timeframe: Timeframe::Min1.into(),
331            ts_sec: vec![10, 20],
332            open: vec![1, 2],
333            high: vec![1],
334            low: vec![1, 2],
335            close: vec![1, 2],
336            volume: vec![1, 2],
337            ..Default::default()
338        };
339        let err = candles_columns_from_proto(&msg, 8)
340            .expect_err("misaligned columnar response must fail closed");
341        assert!(err.to_string().contains("response lengths"));
342        assert!(err.to_string().contains("high=1"));
343    }
344}
polyester/codecs/decode/market_data.rs

polyester/codecs/decode/
market_data.rs
