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

polyester/codecs/decode/
market_data.rs
