|
|
@@ -1,8 +1,9 @@
|
|
|
-use tracing::warn;
|
|
|
-use crate::data_providers::deserialize_string_to_f32;
|
|
|
+use tracing::{info, warn};
|
|
|
+use crate::data_providers::{deserialize_string_to_f32, SpawnExecutor};
|
|
|
use crate::data_providers::deserialize_string_to_i64;
|
|
|
use std::collections::HashMap;
|
|
|
-
|
|
|
+use std::vec;
|
|
|
+use bytes::Bytes;
|
|
|
use iced::{
|
|
|
stream,
|
|
|
futures::{sink::SinkExt, Stream},
|
|
|
@@ -12,15 +13,17 @@ use regex::Regex;
|
|
|
use serde_json::json;
|
|
|
use serde_json::Value;
|
|
|
|
|
|
-use sonic_rs::{JsonValueTrait, Deserialize, Serialize};
|
|
|
+use sonic_rs::{Deserialize, JsonValueTrait, Serialize};
|
|
|
use sonic_rs::to_object_iter_unchecked;
|
|
|
|
|
|
use fastwebsockets::{Frame, FragmentCollector, OpCode};
|
|
|
+use http_body_util::Empty;
|
|
|
+use hyper::header::{CONNECTION, UPGRADE};
|
|
|
+use hyper::Request;
|
|
|
use hyper::upgrade::Upgraded;
|
|
|
use hyper_util::rt::TokioIo;
|
|
|
-
|
|
|
+use tokio::net::TcpStream;
|
|
|
use crate::data_providers::{
|
|
|
- setup_tcp_connection, setup_tls_connection, setup_websocket_connection,
|
|
|
Connection, Event, Kline, LocalDepthCache, MarketType, Order, State,
|
|
|
StreamError, TickerInfo, TickerStats, Trade, VecLocalDepthCache,
|
|
|
};
|
|
|
@@ -33,11 +36,11 @@ use tracing::{error};
|
|
|
|
|
|
#[derive(Serialize, Deserialize, Debug)]
|
|
|
struct SonicDepth {
|
|
|
- #[serde(rename = "u")]
|
|
|
+ #[serde(rename = "time")]
|
|
|
pub update_id: u64,
|
|
|
- #[serde(rename = "b")]
|
|
|
+ #[serde(rename = "bids")]
|
|
|
pub bids: Vec<BidAsk>,
|
|
|
- #[serde(rename = "a")]
|
|
|
+ #[serde(rename = "asks")]
|
|
|
pub asks: Vec<BidAsk>,
|
|
|
}
|
|
|
|
|
|
@@ -51,181 +54,128 @@ struct BidAsk {
|
|
|
|
|
|
#[derive(Serialize, Deserialize, Debug)]
|
|
|
struct SonicTrade {
|
|
|
- #[serde(rename = "T")]
|
|
|
+ #[serde(rename = "time")]
|
|
|
pub time: u64,
|
|
|
- #[serde(rename = "p")]
|
|
|
+ #[serde(rename = "last_price")]
|
|
|
pub price: String,
|
|
|
- #[serde(rename = "v")]
|
|
|
+ #[serde(rename = "last_qty")]
|
|
|
pub qty: String,
|
|
|
- #[serde(rename = "S")]
|
|
|
- pub is_sell: String,
|
|
|
+ #[serde(rename = "side")]
|
|
|
+ pub side: String,
|
|
|
}
|
|
|
|
|
|
-// #[derive(Deserialize, Debug, Clone)]
|
|
|
-// pub struct SonicKline {
|
|
|
-// #[serde(rename = "start")]
|
|
|
-// pub time: u64,
|
|
|
-// #[serde(rename = "open")]
|
|
|
-// pub open: String,
|
|
|
-// #[serde(rename = "high")]
|
|
|
-// pub high: String,
|
|
|
-// #[serde(rename = "low")]
|
|
|
-// pub low: String,
|
|
|
-// #[serde(rename = "close")]
|
|
|
-// pub close: String,
|
|
|
-// #[serde(rename = "volume")]
|
|
|
-// pub volume: String,
|
|
|
-// #[serde(rename = "interval")]
|
|
|
-// pub interval: String,
|
|
|
-// }
|
|
|
-
|
|
|
#[derive(Debug)]
|
|
|
enum StreamData {
|
|
|
Trade(Vec<SonicTrade>),
|
|
|
- Depth(SonicDepth, String, i64),
|
|
|
+ Depth(SonicDepth, i64),
|
|
|
// Kline(Ticker, Vec<SonicKline>),
|
|
|
}
|
|
|
|
|
|
-#[derive(Debug)]
|
|
|
-enum StreamName {
|
|
|
- Depth(Ticker),
|
|
|
- Trade(Ticker),
|
|
|
- Kline(Ticker),
|
|
|
- Unknown,
|
|
|
-}
|
|
|
-impl StreamName {
|
|
|
- fn from_topic(topic: &str, is_ticker: Option<Ticker>, market_type: MarketType) -> Self {
|
|
|
- let parts: Vec<&str> = topic.split('.').collect();
|
|
|
-
|
|
|
- if let Some(ticker_str) = parts.last() {
|
|
|
- let ticker = is_ticker.unwrap_or_else(|| Ticker::new(ticker_str, market_type));
|
|
|
-
|
|
|
- match parts.first() {
|
|
|
- Some(&"publicTrade") => StreamName::Trade(ticker),
|
|
|
- Some(&"orderbook") => StreamName::Depth(ticker),
|
|
|
- Some(&"kline") => StreamName::Kline(ticker),
|
|
|
- _ => StreamName::Unknown,
|
|
|
- }
|
|
|
- } else {
|
|
|
- StreamName::Unknown
|
|
|
- }
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-#[derive(Debug)]
|
|
|
-enum StreamWrapper {
|
|
|
- Trade,
|
|
|
- Depth,
|
|
|
- Kline,
|
|
|
-}
|
|
|
-
|
|
|
#[allow(unused_assignments)]
|
|
|
fn feed_de(
|
|
|
- slice: &[u8],
|
|
|
- ticker: Option<Ticker>,
|
|
|
- market_type: MarketType
|
|
|
-) -> Result<StreamData, StreamError> {
|
|
|
- let mut stream_type: Option<StreamWrapper> = None;
|
|
|
- let mut depth_wrap: Option<SonicDepth> = None;
|
|
|
-
|
|
|
- let mut data_type = String::new();
|
|
|
- // let mut topic_ticker = Ticker::default();
|
|
|
+ slice: &[u8]
|
|
|
+) -> Result<Vec<StreamData>, StreamError> {
|
|
|
+ // // 这里应该是做缓存，之前的bybit数据推送是增量的
|
|
|
+ // let mut depth_wrap: Option<SonicDepth> = None;
|
|
|
|
|
|
let iter: sonic_rs::ObjectJsonIter = unsafe { to_object_iter_unchecked(slice) };
|
|
|
+ let mut trade = SonicTrade {
|
|
|
+ time: 0,
|
|
|
+ price: "".to_string(),
|
|
|
+ qty: "".to_string(),
|
|
|
+ side: "".to_string(),
|
|
|
+ };
|
|
|
+ let mut depth = SonicDepth {
|
|
|
+ update_id: 0,
|
|
|
+ bids: vec![],
|
|
|
+ asks: vec![],
|
|
|
+ };
|
|
|
|
|
|
for elem in iter {
|
|
|
let (k, v) = elem.map_err(|e| StreamError::ParseError(e.to_string()))?;
|
|
|
-
|
|
|
- if k == "topic" {
|
|
|
- if let Some(val) = v.as_str() {
|
|
|
- let mut is_ticker = None;
|
|
|
-
|
|
|
- if let Some(ticker) = ticker {
|
|
|
- is_ticker = Some(ticker);
|
|
|
- }
|
|
|
+ // let v = &v.as_raw_faststr();
|
|
|
+ match k.as_str() {
|
|
|
+ "time" => {
|
|
|
+ let t: u64 = v.as_u64().unwrap();
|
|
|
|
|
|
- match StreamName::from_topic(val, is_ticker, market_type) {
|
|
|
- StreamName::Depth(_ticker) => {
|
|
|
- stream_type = Some(StreamWrapper::Depth);
|
|
|
-
|
|
|
- // topic_ticker = ticker;
|
|
|
- }
|
|
|
- StreamName::Trade(_ticker) => {
|
|
|
- stream_type = Some(StreamWrapper::Trade);
|
|
|
+ trade.time = t;
|
|
|
+ depth.update_id = t;
|
|
|
+ }
|
|
|
+ "last_price" => {
|
|
|
+ trade.price = v.to_string();
|
|
|
+ }
|
|
|
+ "last_qty" => {
|
|
|
+ trade.qty = v.to_string();
|
|
|
+ }
|
|
|
+ "side" => {
|
|
|
+ trade.side = v.to_string();
|
|
|
+ }
|
|
|
+ "asks" => {
|
|
|
+ let asks = serde_json::from_slice::<Value>(v.as_raw_str().as_bytes())
|
|
|
+ .unwrap();
|
|
|
|
|
|
- // topic_ticker = ticker;
|
|
|
- }
|
|
|
- StreamName::Kline(_ticker) => {
|
|
|
- stream_type = Some(StreamWrapper::Kline);
|
|
|
+ for ask in asks.as_array().unwrap() {
|
|
|
+ let order_book = BidAsk {
|
|
|
+ price: ask.as_array().unwrap()[0].to_string(),
|
|
|
+ qty: ask.as_array().unwrap()[1].to_string(),
|
|
|
+ };
|
|
|
|
|
|
- // topic_ticker = ticker;
|
|
|
- }
|
|
|
- _ => {
|
|
|
- error!("Unknown stream name");
|
|
|
- }
|
|
|
+ depth.asks.push(order_book);
|
|
|
}
|
|
|
}
|
|
|
- } else if k == "type" {
|
|
|
- v.as_str().unwrap().clone_into(&mut data_type);
|
|
|
- } else if k == "data" {
|
|
|
- match stream_type {
|
|
|
- Some(StreamWrapper::Trade) => {
|
|
|
- let trade_wrap: Vec<SonicTrade> = sonic_rs::from_str(&v.as_raw_faststr())
|
|
|
- .map_err(|e| StreamError::ParseError(e.to_string()))?;
|
|
|
-
|
|
|
- return Ok(StreamData::Trade(trade_wrap));
|
|
|
- }
|
|
|
- Some(StreamWrapper::Depth) => {
|
|
|
- if depth_wrap.is_none() {
|
|
|
- depth_wrap = Some(SonicDepth {
|
|
|
- update_id: 0,
|
|
|
- bids: Vec::new(),
|
|
|
- asks: Vec::new(),
|
|
|
- });
|
|
|
- }
|
|
|
- depth_wrap = Some(
|
|
|
- sonic_rs::from_str(&v.as_raw_faststr())
|
|
|
- .map_err(|e| StreamError::ParseError(e.to_string()))?,
|
|
|
- );
|
|
|
- }
|
|
|
- // Some(StreamWrapper::Kline) => {
|
|
|
- // let kline_wrap: Vec<SonicKline> = sonic_rs::from_str(&v.as_raw_faststr())
|
|
|
- // .map_err(|e| StreamError::ParseError(e.to_string()))?;
|
|
|
- //
|
|
|
- // return Ok(StreamData::Kline(topic_ticker, kline_wrap));
|
|
|
- // }
|
|
|
- _ => {
|
|
|
- error!("Unknown stream type");
|
|
|
- }
|
|
|
- }
|
|
|
- } else if k == "cts" {
|
|
|
- if let Some(dw) = depth_wrap {
|
|
|
- let time: u64 = v
|
|
|
- .as_u64()
|
|
|
- .ok_or_else(|| StreamError::ParseError("Failed to parse u64".to_string()))?;
|
|
|
+ "bids" => {
|
|
|
+ let bids = serde_json::from_slice::<Value>(v.as_raw_str().as_bytes())
|
|
|
+ .unwrap();
|
|
|
+
|
|
|
+ for bid in bids.as_array().unwrap() {
|
|
|
+ let order_book = BidAsk {
|
|
|
+ price: bid.as_array().unwrap()[0].to_string(),
|
|
|
+ qty: bid.as_array().unwrap()[1].to_string(),
|
|
|
+ };
|
|
|
|
|
|
- return Ok(StreamData::Depth(dw, data_type.to_string(), time as i64));
|
|
|
+ depth.bids.push(order_book);
|
|
|
+ }
|
|
|
}
|
|
|
+ &_ => {}
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- Err(StreamError::UnknownError("Unknown data".to_string()))
|
|
|
+ let update_id = depth.update_id;
|
|
|
+ Ok(vec![StreamData::Trade(vec![trade]), StreamData::Depth(depth, update_id as i64)])
|
|
|
}
|
|
|
|
|
|
-async fn connect(
|
|
|
- domain: &str,
|
|
|
-) -> Result<FragmentCollector<TokioIo<Upgraded>>, StreamError> {
|
|
|
- let tcp_stream = setup_tcp_connection(domain).await?;
|
|
|
- let tls_stream = setup_tls_connection(domain, tcp_stream).await?;
|
|
|
- let url = "ws://127.0.0.1:6789";
|
|
|
- setup_websocket_connection(domain, tls_stream, &url).await
|
|
|
+async fn connect() -> Result<FragmentCollector<TokioIo<Upgraded>>, StreamError> {
|
|
|
+ let url = "ws://localhost:6789";
|
|
|
+ let addr = "localhost:6789";
|
|
|
+ let stream = TcpStream::connect(&addr).await
|
|
|
+ .map_err(|e| StreamError::WebsocketError(e.to_string()))?;
|
|
|
+
|
|
|
+ // 2. 构建 WebSocket 握手请求
|
|
|
+ let req = Request::builder()
|
|
|
+ .method("GET")
|
|
|
+ .uri(url)
|
|
|
+ .header("Host", "localhost")
|
|
|
+ .header(UPGRADE, "websocket")
|
|
|
+ .header(CONNECTION, "upgrade")
|
|
|
+ .header("Sec-WebSocket-Key", fastwebsockets::handshake::generate_key())
|
|
|
+ .header("Sec-WebSocket-Version", "13")
|
|
|
+ .header("Sec-WebSocket-Protocol", "rust-websocket") // 可选协议
|
|
|
+ .header("User-Agent", "rust-client/1.0") // 添加 UA 头
|
|
|
+ .body(Empty::<Bytes>::new())
|
|
|
+ .map_err(|e| StreamError::WebsocketError(e.to_string()))?;
|
|
|
+
|
|
|
+ let (ws, _) = fastwebsockets::handshake::client(&SpawnExecutor, req, stream)
|
|
|
+ .await
|
|
|
+ .map_err(|e| StreamError::WebsocketError(e.to_string()))?;
|
|
|
+
|
|
|
+ Ok(FragmentCollector::new(ws))
|
|
|
}
|
|
|
|
|
|
async fn try_connect(
|
|
|
streams: &Value,
|
|
|
output: &mut futures::channel::mpsc::Sender<Event>,
|
|
|
) -> State {
|
|
|
- match connect("127.0.0.1:6789").await {
|
|
|
+ match connect().await {
|
|
|
Ok(mut websocket) => {
|
|
|
if let Err(e) = websocket
|
|
|
.write_frame(Frame::text(fastwebsockets::Payload::Borrowed(
|
|
|
@@ -288,61 +238,65 @@ pub fn connect_market_stream(ticker: Ticker) -> impl Stream<Item = Event> {
|
|
|
State::Connected(websocket) => match websocket.read_frame().await {
|
|
|
Ok(msg) => match msg.opcode {
|
|
|
OpCode::Text => {
|
|
|
- if let Ok(data) = feed_de(&msg.payload[..], Some(ticker), market_type) {
|
|
|
- match data {
|
|
|
- StreamData::Trade(de_trade_vec) => {
|
|
|
- for de_trade in &de_trade_vec {
|
|
|
- let trade = Trade {
|
|
|
- time: de_trade.time as i64,
|
|
|
- is_sell: de_trade.is_sell == "Sell",
|
|
|
- price: str_f32_parse(&de_trade.price),
|
|
|
- qty: str_f32_parse(&de_trade.qty),
|
|
|
- };
|
|
|
-
|
|
|
- trades_buffer.push(trade);
|
|
|
+ let result = feed_de(&msg.payload[..]);
|
|
|
+ match result {
|
|
|
+ Ok(data_vec) => {
|
|
|
+ let trade_handle_rst = &data_vec[0];
|
|
|
+ let depth_handle_rst = &data_vec[1];
|
|
|
+
|
|
|
+ match trade_handle_rst {
|
|
|
+ StreamData::Trade(de_trade_vec) => {
|
|
|
+ for de_trade in de_trade_vec {
|
|
|
+ let trade = Trade {
|
|
|
+ time: de_trade.time as i64,
|
|
|
+ is_sell: de_trade.side == "sell",
|
|
|
+ price: str_f32_parse(&de_trade.price),
|
|
|
+ qty: str_f32_parse(&de_trade.qty),
|
|
|
+ };
|
|
|
+
|
|
|
+ trades_buffer.push(trade);
|
|
|
+ }
|
|
|
}
|
|
|
+ StreamData::Depth(_, _) => {}
|
|
|
}
|
|
|
- StreamData::Depth(de_depth, data_type, time) => {
|
|
|
- let depth_update = VecLocalDepthCache {
|
|
|
- last_update_id: de_depth.update_id as i64,
|
|
|
- time,
|
|
|
- bids: de_depth
|
|
|
- .bids
|
|
|
- .iter()
|
|
|
- .map(|x| Order {
|
|
|
- price: str_f32_parse(&x.price),
|
|
|
- qty: str_f32_parse(&x.qty),
|
|
|
- })
|
|
|
- .collect(),
|
|
|
- asks: de_depth
|
|
|
- .asks
|
|
|
- .iter()
|
|
|
- .map(|x| Order {
|
|
|
- price: str_f32_parse(&x.price),
|
|
|
- qty: str_f32_parse(&x.qty),
|
|
|
- })
|
|
|
- .collect(),
|
|
|
- };
|
|
|
-
|
|
|
- if (data_type == "snapshot")
|
|
|
- || (depth_update.last_update_id == 1)
|
|
|
- {
|
|
|
+
|
|
|
+ match depth_handle_rst {
|
|
|
+ StreamData::Depth(de_depth, time) => {
|
|
|
+ let t = time.clone();
|
|
|
+
|
|
|
+ let depth_update = VecLocalDepthCache {
|
|
|
+ last_update_id: de_depth.update_id as i64,
|
|
|
+ time: t,
|
|
|
+ bids: de_depth
|
|
|
+ .bids
|
|
|
+ .iter()
|
|
|
+ .map(|x| Order {
|
|
|
+ price: str_f32_parse(&x.price),
|
|
|
+ qty: str_f32_parse(&x.qty),
|
|
|
+ })
|
|
|
+ .collect(),
|
|
|
+ asks: de_depth
|
|
|
+ .asks
|
|
|
+ .iter()
|
|
|
+ .map(|x| Order {
|
|
|
+ price: str_f32_parse(&x.price),
|
|
|
+ qty: str_f32_parse(&x.qty),
|
|
|
+ })
|
|
|
+ .collect(),
|
|
|
+ };
|
|
|
+
|
|
|
orderbook.fetched(&depth_update);
|
|
|
- } else if data_type == "delta" {
|
|
|
- orderbook.update_depth_cache(&depth_update);
|
|
|
-
|
|
|
- let _ = output
|
|
|
- .send(Event::DepthReceived(
|
|
|
- ticker,
|
|
|
- time,
|
|
|
- orderbook.get_depth(),
|
|
|
- std::mem::take(&mut trades_buffer).into_boxed_slice(),
|
|
|
- ))
|
|
|
- .await;
|
|
|
}
|
|
|
+
|
|
|
+ StreamData::Trade(_) => {}
|
|
|
}
|
|
|
}
|
|
|
+ Err(e) => {
|
|
|
+ // 处理错误
|
|
|
+ error!("处理数据失败: {}", e);
|
|
|
+ }
|
|
|
}
|
|
|
+
|
|
|
}
|
|
|
OpCode::Close => {
|
|
|
state = State::Disconnected;
|
