Repository: yasstake/rbot (language: python)

Issue:
Make global access toboard


<!-- This is an auto-generated comment: release notes by coderabbit.ai -->

## Summary by CodeRabbit

- **新機能**
  - Bybitでユーザーストリームを開始する機能が追加されました。
  - アーカイブと最新のデータをダウンロードする新しいメソッドが追加されました。
  - RESTサーバーが追加され、指定された取引所、カテゴリ、シンボルからオーダーブックデータを取得する機能が提供されます。

- **バグ修正**
  - オーダーブックリストのパス作成メソッドが交換名を別のパラメータとして取るように変更されました。

- **リファクタ**
  - コード内の複数のTODOコメントとコメントアウトされたコードが削除されました。
  - `MarketConfig`構造体がデフォルトトレイトを実装するようになりました。
  - 非同期機能が複数のコンポーネントに導入され、コードの効率性が向上しました。

- **スタイル**
  - テストモジュールでの`message`定数の扱いが変更されました。

- **テスト**
  - Bybitと関連するテストが更新され、新しい機能のテストが追加されました。

- **ドキュメント**
  - コード内の簡潔な要約が提供され、機能の目的と影響が明確に説明されています。

- **その他**
  - `rbot`、`rbot_lib`、`rbot_market`、`rbot_server`、`rbot_session`における内部構造と機能の重要な更新が行われました。

<!-- end of auto-generated comment: release notes by coderabbit.ai -->


Discussion / hints:

Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index 1b79b8ab..771d3cd8 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -5,10 +5,12 @@ members = [
     "rbot_lib",
     "rbot_market",
     "rbot_session",
+    "rbot_server",
+    "rbot_blockon",
     "exchanges/binance",
     "exchanges/bitbank",
     "exchanges/bitflyer",
-    "exchanges/bybit",
+    "exchanges/bybit", "rbot_server", "rbot_blockon",
 ]
 
 resolver = "2"
@@ -16,8 +18,10 @@ resolver = "2"
 [workspace.dependencies]
 rbot = { path = "./rbot" }
 rbot_lib = { path = "./rbot_lib" }
+rbot_blockon = { path = "./rbot_blockon" }
 rbot_session = { path = "./rbot_session" }
 rbot_market = { path = "./rbot_market" }
+rbot_server = { path = "./rbot_server" }
 binance = { path = "./exchanges/binance" }
 bitbank = { path = "./exchanges/bitbank" }
 bitflyer = { path = "./exchanges/bitflyer" }
@@ -25,6 +29,8 @@ bybit = { path = "./exchanges/bybit" }
 
 anyhow = { version = "1.0.79" }
 
+features = {version = "0.10.0"}
+
 chrono = { version = "0.4.33" }
 serde = { version = "1.0" }
 serde_json = { version = "1.0" }
@@ -101,4 +107,7 @@ socket2 = { version = "0.5.5" }
 rmp-serde = { version = "1.1.2" }
 once_cell = { version = "1.19.0" }
 
+actix-web={version="4.5"}
+actix-rt={version="2.9"}
+
 #pyo3 = {version="0.20.2"}
diff --git a/MEMO/note.txt b/MEMO/note.txt
index 247db996..d1cb5541 100644
--- a/MEMO/note.txt
+++ b/MEMO/note.txt
@@ -20,3 +20,12 @@
 
     Pyo3オブジェクト直前インターフェースはpyo3::基本はAnyHowを使う。    
 
+＜時刻はi64型を使う＞
+    専用クラスへ変更しようとしたがすでに遅し。
+    最初からきちんと設計しないと途中からの変更は難しい。
+
+<Asyncとblocking境界＞
+    PythonInterfaceはBlocking
+
+    その直前までasync。
+    
\ No newline at end of file
diff --git a/exchanges/binance/src/market.rs b/exchanges/binance/src/market.rs
index eccfb291..07303ca5 100644
--- a/exchanges/binance/src/market.rs
+++ b/exchanges/binance/src/market.rs
@@ -589,13 +589,9 @@ impl BinanceMarket {
         );
     }
 
-    // TODO: implment retry logic
+
     pub fn start_market_stream(&mut self) {
         let endpoint = &self.config.public_ws_endpoint;
-        /*
-        let mut subscribe_message = BinanceWsOpMessage::new();
-        subscribe_message.add_params(&self.config.public_subscribe_channel);
-        */
 
         let mut agent_channel = self.channel.clone();
 
@@ -617,7 +613,6 @@ impl BinanceMarket {
         let db_channel = self.db.start_thread();
         let board = self.board.clone();
 
-        // TODO: change to tokio
         let handler = tokio::task::spawn(async move {
                 loop {
                     let message = websocket.receive_text().await;
@@ -687,17 +682,6 @@ impl BinanceMarket {
         log::info!("start_market_stream");
     }
 
-    /*
-    // TODO: 単に待っているだけなので、終了処理を実装する。
-    pub fn stop_market_stream(&mut self) {
-        match self.public_handler.take() {
-            Some(h) => {
-                h.join().unwrap();
-            }
-            None => {}
-        }
-    }
-    */
 
     pub fn start_user_stream(&mut self) {
         let mut agent_channel = self.channel.clone();
@@ -750,7 +734,7 @@ impl BinanceMarket {
     pub fn get_channel(&mut self) -> MarketStream {
 
         // TODO: confirm buffer size
-        let ch = self.channel.lock().unwrap().open_channel(0);
+        let ch = self.channel.lock().unwrap().open_channel(1000);
 
         return MarketStream { reciver: ch };
     }
diff --git a/exchanges/bybit/Cargo.toml b/exchanges/bybit/Cargo.toml
index b57a040e..332c59c0 100644
--- a/exchanges/bybit/Cargo.toml
+++ b/exchanges/bybit/Cargo.toml
@@ -5,8 +5,10 @@ edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
+
 [dependencies]
 rbot_lib = {workspace = true}
+rbot_blockon = {workspace = true}
 rbot_market = {workspace = true}
 
 anyhow = {workspace = true}
@@ -14,6 +16,8 @@ anyhow = {workspace = true}
 chrono = {workspace = true}
 log = {workspace = true}
 
+features = {workspace = true}
+
 rust_decimal = {workspace = true}
 rust_decimal_macros = {workspace = true}
 serde = {workspace = true}
@@ -31,6 +35,7 @@ async-stream = {workspace = true}
 
 
 
+
 # https://pyo3.rs/v0.13.2/faq
 [dependencies.pyo3]
 version = "0.20"
diff --git a/exchanges/bybit/src/config.rs b/exchanges/bybit/src/config.rs
index b6a34ead..48583d33 100644
--- a/exchanges/bybit/src/config.rs
+++ b/exchanges/bybit/src/config.rs
@@ -72,6 +72,10 @@ impl BybitServerConfig {
 }
 
 impl ServerConfig for BybitServerConfig {
+    fn get_exchange_name(&self) -> String {
+        self.exchange_name.clone()
+    }
+
     fn get_public_ws_server(&self) -> String {
         self.public_ws.clone()
     }
@@ -95,6 +99,7 @@ impl ServerConfig for BybitServerConfig {
     fn get_historical_web_base(&self) -> String {
         self.history_web_base.clone()
     }
+
 }
 
 #[derive(Debug, Clone, Serialize)]
@@ -164,57 +169,22 @@ impl BybitConfig {
     }
 }
 
-/*
-#[derive(Debug, Clone)]
-#[pyclass]
-pub struct BybitConfig {
-    pub server_config: BybitServerConfig,
-    pub market_config: MarketConfig,
-    pub db_base_dir: String,
-}
-
-#[pymethods]
-impl BybitConfig {
-    pub fn get_db_path(&self) -> String {
-        let mut exchange_name = self.exchange_name.clone();
-
-        if self.testnet {
-            exchange_name = format!("{}-TESTNET", exchange_name);
-        }
 
-        let db_path = db_full_path(&exchange_name, &self.trade_category, &self.trade_symbol, &self.db_base_dir);
+#[cfg(test)]
+mod test_bybit_config {
+    use super::*;
 
-        return db_path.to_str().unwrap().to_string();
+    #[test]
+    fn test_bybit_server_config() {
+        let config = BybitServerConfig::new(true);
+        println!("{:?}", config);
     }
 
-    #[classattr]
-    pub fn SPOT_BTCUSDT() -> Self {
-        return BybitConfig {
-            exchange_name: "BYBIT".to_string(),
-            testnet: false,
-            rest_endpoint: "https://api.bybit.com".to_string(),
-            history_web_base: "https://public.bybit.com".to_string(),
-            db_base_dir: "".to_string(),
-            public_stream_endpoint: "wss://stream.bybit.com/v5/public/spot".to_string(),
-            private_stream_endpoint: "wss://stream.bybit.com/v5/private".to_string(),
-            market_config: MarketConfig {
-                price_unit:dec![0.05],
-                price_scale:3,
-                size_unit:dec![0.001],
-                size_scale:4,
-                maker_fee:dec![0.00_01],
-                taker_fee:dec![0.00_01],
-                price_type:PriceType::Home,
-                fee_type:FeeType::Home,
-                home_currency:"USDT".to_string(),
-                foreign_currency:"BTC".to_string(),
-                market_order_price_slip:dec![0.01],
-                board_depth:250,
-                trade_category: "spot".to_string(),
-                trade_symbol: "BTCUSDT".to_string(),
-            }
-            }
-        };
+
+    #[test]
+    fn test_bybit_config() {
+        let config = BybitConfig::new();
+        println!("{:?}", config);
     }
 }
-*/
+
diff --git a/exchanges/bybit/src/market.rs b/exchanges/bybit/src/market.rs
index c8a6407f..0222e762 100644
--- a/exchanges/bybit/src/market.rs
+++ b/exchanges/bybit/src/market.rs
@@ -10,29 +10,31 @@ use csv::StringRecord;
 
 use futures::StreamExt;
 use polars_core::export::num::FromPrimitive;
+use pyo3::ffi::getter;
 
 use std::sync::{Arc, Mutex, RwLock};
 
-use std::thread::{sleep, JoinHandle};
+use std::thread::sleep;
 use std::time::Duration;
 
 use rbot_lib::common::{
-    convert_klines_to_trades, flush_log, time_string, to_naive_datetime, AccountStatus, BoardItem, BoardTransfer, LogStatus, MarketConfig, MarketMessage, MarketStream, MicroSec, MultiMarketMessage, Order, OrderBook, OrderBookRaw, OrderSide, OrderStatus, OrderType, Trade, DAYS, FLOOR_DAY, HHMM, NOW, RUNTIME, SEC
+    convert_klines_to_trades, flush_log, time_string, to_naive_datetime, AccountCoins, AccountPair,
+    BoardItem, BoardTransfer, LogStatus, MarketConfig, MarketMessage, MarketStream, MicroSec,
+    MultiMarketMessage, Order, OrderBook, OrderBookRaw, OrderSide, OrderStatus, OrderType,
+    ServerConfig, Trade, DAYS, FLOOR_DAY, HHMM, MARKET_HUB, NOW, SEC,
 };
 
-use rbot_lib::common::BLOCK_ON;
-
 use rbot_lib::db::{db_full_path, TradeTable, TradeTableDb, KEY};
-
-use rbot_lib::net::{latest_archive_date, RestApi, UdpSender};
+use rbot_lib::net::{latest_archive_date, BroadcastMessage, RestApi, UdpSender};
 
 use rbot_market::MarketImpl;
 use rbot_market::{MarketInterface, OrderInterface, OrderInterfaceImpl};
 
+use crate::market;
 use crate::message::BybitUserWsMessage;
 
 use crate::rest::BybitRestApi;
-use crate::ws::{BybitPublicWsClient, BybitWsOpMessage};
+use crate::ws::{BybitPrivateWsClient, BybitPublicWsClient, BybitWsOpMessage};
 
 use pyo3::prelude::*;
 use pyo3_polars::PyDataFrame;
@@ -46,73 +48,16 @@ use super::message::{BybitAccountInformation, BybitPublicWsMessage};
 
 use anyhow::Context;
 
-#[derive(Debug)]
-pub struct BybitOrderBook {
-    board: OrderBook,
-}
-
-impl BybitOrderBook {
-    pub fn new(config: &MarketConfig) -> Self {
-        return BybitOrderBook {
-            board: OrderBook::new(&config),
-        };
-    }
-
-    fn get_board_vec(&self) -> anyhow::Result<(Vec<BoardItem>, Vec<BoardItem>)> {
-        let (bids, asks) = self.board.get_board_vec().unwrap();
-
-        Ok((bids, asks))
-    }
+use rbot_blockon::BLOCK_ON;
 
-    fn get_board_json(&self, size: usize) -> anyhow::Result<String> {
-        let json = self.board.get_json(size).unwrap();
-
-        Ok(json)
-    }
-
-    fn get_board(&mut self) -> anyhow::Result<(PyDataFrame, PyDataFrame)> {
-        let (mut bids, mut asks) = self.board.get_board()?;
-
-        if bids.is_empty() || asks.is_empty() {
-            return Ok((PyDataFrame(bids), PyDataFrame(asks)));
-        }
-
-        let bids_edge: f64 = bids.column(KEY::price).unwrap().max().unwrap();
-        let asks_edge: f64 = asks.column(KEY::price).unwrap().min().unwrap();
-
-        if asks_edge < bids_edge {
-            log::warn!("bids_edge({}) < asks_edge({})", bids_edge, asks_edge);
-
-            self.reflesh_board();
-
-            (bids, asks) = self.board.get_board().unwrap();
-        }
-
-        return Ok((PyDataFrame(bids), PyDataFrame(asks)));
-    }
-
-    fn get_edge_price(&self) -> anyhow::Result<(Decimal, Decimal)> {
-        Ok(self.board.get_edge_price())
-    }
-
-    pub fn update(&mut self, board: &OrderBookRaw) {
-        let bids = board.get_bids();
-        let asks = board.get_asks();
-        self.board.update(&bids, &asks, board.snapshot);
-    }
-
-    fn reflesh_board(&mut self) {
-        // TODO: implement
-        println!("reflesh board :NOT IMPLEMENTED");
-        // TODO: reflesh board from rest api
-    }
-}
+use tokio::task::JoinHandle;
 
 #[pyclass]
 #[derive(Debug)]
 pub struct Bybit {
     enable_order: bool,
     server_config: BybitServerConfig,
+    user_handler: Option<JoinHandle<()>>,
 }
 
 #[pymethods]
@@ -125,9 +70,15 @@ impl Bybit {
         return Bybit {
             enable_order: false,
             server_config: server_config.clone(),
+            user_handler: None,
         };
     }
 
+    #[getter]
+    pub fn get_exchange_name(&self) -> String {
+        self.server_config.exchange_name.clone()
+    }
+
     pub fn open_market(&self, config: &MarketConfig) -> BybitMarket {
         return BybitMarket::new(&self.server_config, config);
     }
@@ -142,7 +93,6 @@ impl Bybit {
         OrderInterfaceImpl::get_enable_order_feature(self)
     }
 
-    #[pyo3(signature = (market_config, *, side, price, size, client_order_id=None))]
     pub fn limit_order(
         &self,
         market_config: &MarketConfig,
@@ -151,10 +101,12 @@ impl Bybit {
         size: Decimal,
         client_order_id: Option<&str>,
     ) -> anyhow::Result<Vec<Order>> {
-        OrderInterfaceImpl::limit_order(self, market_config, side, price, size, client_order_id)
+        BLOCK_ON(async {
+            OrderInterfaceImpl::limit_order(self, market_config, side, price, size, client_order_id)
+                .await
+        })
     }
 
-    #[pyo3(signature = (market_config, *, side, size, client_order_id=None))]
     pub fn market_order(
         &self,
         market_config: &MarketConfig,
@@ -162,27 +114,30 @@ impl Bybit {
         size: Decimal,
         client_order_id: Option<&str>,
     ) -> anyhow::Result<Vec<Order>> {
-        OrderInterfaceImpl::market_order(self, market_config, side, size, client_order_id)
+        BLOCK_ON(async {
+            OrderInterfaceImpl::market_order(self, market_config, side, size, client_order_id).await
+        })
     }
 
-    #[pyo3(signature = (market_config, *, order_id))]
     pub fn cancel_order(
         &self,
         market_config: &MarketConfig,
         order_id: &str,
     ) -> anyhow::Result<Order> {
-        OrderInterfaceImpl::cancel_order(self, market_config, order_id)
+        BLOCK_ON(async { OrderInterfaceImpl::cancel_order(self, market_config, order_id).await })
     }
 
-    #[pyo3(signature = (market_config))]
     pub fn get_open_orders(&self, market_config: &MarketConfig) -> anyhow::Result<Vec<Order>> {
-        OrderInterfaceImpl::get_open_orders(self, market_config)
+        BLOCK_ON(async { OrderInterfaceImpl::get_open_orders(self, market_config).await })
     }
 
-    #[pyo3(signature = (market_config))]
-    // TODO: implement and test
-    pub fn get_account(&self, market_config: &MarketConfig) -> anyhow::Result<AccountStatus> {
-        OrderInterfaceImpl::get_account(self, market_config)
+    #[getter]
+    pub fn get_account(&self) -> anyhow::Result<AccountCoins> {
+        BLOCK_ON(async { OrderInterfaceImpl::get_account(self).await })
+    }
+
+    pub fn start_user_stream(&mut self) -> anyhow::Result<()> {
+        BLOCK_ON(async { OrderInterfaceImpl::async_start_user_stream(self).await })
     }
 }
 
@@ -199,8 +154,52 @@ impl OrderInterfaceImpl<BybitRestApi, BybitServerConfig> for Bybit {
         &self.server_config
     }
 
-    fn start_user_stream(&mut self) {
-        todo!()
+    async fn async_start_user_stream(&mut self) -> anyhow::Result<()> {
+        let exchange_name = self.server_config.exchange_name.clone();
+        let server_config = self.server_config.clone();
+
+        self.user_handler = Some(tokio::task::spawn(async move {
+            let mut ws = BybitPrivateWsClient::new(&server_config).await;
+            ws.connect().await;
+
+            let mut market_channel = MARKET_HUB.open_channel();
+            let mut ws_stream = Box::pin(ws.open_stream().await);
+
+            while let Some(message) = ws_stream.next().await {
+                if message.is_err() {
+                    log::error!("Error in ws_stream.recv: {:?}", message);
+                    continue;
+                }
+
+                let message = message.unwrap();
+                match message {
+                    MultiMarketMessage::Order(order) => {
+                        for o in order {
+                            market_channel.send(BroadcastMessage {
+                                exchange: exchange_name.clone(),
+                                category: o.category.clone(),
+                                symbol: o.symbol.clone(),
+                                msg: MarketMessage::Order(o.clone()),
+                            });
+                            log::debug!("Order: {:?}", o);
+                        }
+                    }
+                    MultiMarketMessage::Account(account) => {
+                        market_channel.send(BroadcastMessage {
+                            exchange: exchange_name.clone(),
+                            category: "".to_string(),
+                            symbol: "".to_string(),
+                            msg: MarketMessage::Account(account.clone()),
+                        });
+                    }
+                    _ => {
+                        log::info!("User stream message: {:?}", message);
+                    }
+                }
+            }
+        }));
+
+        Ok(())
     }
 }
 
@@ -226,19 +225,20 @@ impl BybitMarket {
         log::debug!("open market BybitMarket::new");
         BLOCK_ON(async { Self::async_new(server_config, config).await })
     }
-
+    #[getter]
     fn get_config(&self) -> MarketConfig {
         MarketImpl::get_config(self)
     }
-
+    #[getter]
     fn get_exchange_name(&self) -> String {
         MarketImpl::get_exchange_name(self)
     }
-
+    #[getter]
     fn get_trade_category(&self) -> String {
         MarketImpl::get_trade_category(self)
     }
 
+    #[getter]
     fn get_trade_symbol(&self) -> String {
         MarketImpl::get_trade_symbol(self)
     }
@@ -247,6 +247,7 @@ impl BybitMarket {
         MarketImpl::drop_table(self)
     }
 
+    #[getter]
     fn get_cache_duration(&self) -> MicroSec {
         MarketImpl::get_cache_duration(self)
     }
@@ -306,26 +307,32 @@ impl BybitMarket {
         MarketImpl::get_board_json(self, size)
     }
 
+    #[getter]
     fn get_board(&mut self) -> anyhow::Result<(PyDataFrame, PyDataFrame)> {
         MarketImpl::get_board(self)
     }
 
+    #[getter]
     fn get_board_vec(&self) -> anyhow::Result<(Vec<BoardItem>, Vec<BoardItem>)> {
         MarketImpl::get_board_vec(self)
     }
 
+    #[getter]
     fn get_edge_price(&self) -> anyhow::Result<(Decimal, Decimal)> {
         MarketImpl::get_edge_price(self)
     }
 
+    #[getter]
     fn get_file_name(&self) -> String {
         MarketImpl::get_file_name(self)
     }
 
+    #[getter]
     fn get_market_config(&self) -> MarketConfig {
         MarketImpl::get_market_config(self)
     }
 
+    #[getter]
     fn get_running(&self) -> bool {
         MarketImpl::get_running(self)
     }
@@ -338,6 +345,7 @@ impl BybitMarket {
         MarketImpl::_repr_html_(self)
     }
 
+    #[pyo3(signature = (ndays, force=false, verbose=true, low_priority=true))]
     fn download_archive(
         &mut self,
         ndays: i64,
@@ -345,7 +353,14 @@ impl BybitMarket {
         verbose: bool,
         low_priority: bool,
     ) -> anyhow::Result<i64> {
-        MarketImpl::download_archives(self, ndays, force, verbose, low_priority)
+        BLOCK_ON(async {
+            MarketImpl::async_download_archives(self, ndays, force, verbose, low_priority).await
+        })
+    }
+
+    #[pyo3(signature = (verbose=true))]
+    fn download_latest(&mut self, verbose: bool) -> anyhow::Result<i64> {
+        BLOCK_ON(async { self.async_download_lastest(verbose).await })
     }
 
     fn expire_unfix_data(&mut self) -> anyhow::Result<()> {
@@ -356,10 +371,6 @@ impl BybitMarket {
         MarketImpl::find_latest_gap(self)
     }
 
-    fn download_latest(&mut self, verbose: bool) -> anyhow::Result<i64> {
-        MarketImpl::download_latest(self, verbose)
-    }
-
     fn download_gap(&mut self, verbose: bool) -> anyhow::Result<i64> {
         MarketImpl::download_gap(self, verbose)
     }
@@ -368,21 +379,6 @@ impl BybitMarket {
         MarketImpl::start_market_stream(self)
     }
 
-    fn get_trade_list(&self) -> anyhow::Result<Vec<OrderStatus>> {
-        todo!()
-        //MarketImpl::get_trade_list(self)
-    }
-
-    fn get_account(&self) -> anyhow::Result<AccountStatus> {
-        todo!()
-        //MarketImpl::get_account(self)
-    }
-
-    fn get_recent_trades(&self) -> anyhow::Result<Vec<Trade>> {
-        todo!()
-        //MarketImpl::get_recent_trades(self)
-    }
-
     fn open_realtime_channel(&mut self) -> anyhow::Result<MarketStream> {
         MarketImpl::open_realtime_channel(self)
     }
@@ -415,7 +411,7 @@ impl BybitMarket {
             server_config: server_config.clone(),
             config: config.clone(),
             db: Arc::new(Mutex::new(db.unwrap())),
-            board: Arc::new(RwLock::new(OrderBook::new(&config))),
+            board: Arc::new(RwLock::new(OrderBook::new(server_config, &config))),
             public_handler: None,
         };
     }
@@ -439,12 +435,19 @@ impl MarketImpl<BybitRestApi, BybitServerConfig> for BybitMarket {
     }
 
     fn download_latest(&mut self, verbose: bool) -> anyhow::Result<i64> {
-        BLOCK_ON(async { self.async_download_lastest().await })
+        BLOCK_ON(async { self.async_download_lastest(verbose).await })
     }
 
     fn download_gap(&mut self, verbose: bool) -> anyhow::Result<i64> {
         log::debug!("[start] download_gap ");
-        let (unfix_start, unfix_end) = self.find_latest_gap()?;
+        let gap_result = self.find_latest_gap();
+        if gap_result.is_err() {
+            log::warn!("no gap found: {:?}", gap_result);
+            return Ok(0);
+        }
+
+        let (unfix_start, unfix_end) = gap_result.unwrap();
+        //let (unfix_start, unfix_end) = self.find_latest_gap();
         log::debug!("unfix_start: {:?}, unfix_end: {:?}", unfix_start, unfix_end);
 
         if unfix_end - unfix_start <= HHMM(0, 1) {
@@ -464,9 +467,10 @@ impl MarketImpl<BybitRestApi, BybitServerConfig> for BybitMarket {
 
         let expire_message = TradeTableDb::expire_control_message(unfix_start, unfix_end);
 
-        let mut lock = self.db.lock().unwrap();
-        let tx = lock.start_thread();
-        drop(lock);
+        let tx = {
+            let mut lock = self.db.lock().unwrap();
+            lock.start_thread()
+        };
 
         tx.send(expire_message)?;
         tx.send(trades)?;
@@ -517,15 +521,42 @@ impl MarketImpl<BybitRestApi, BybitServerConfig> for BybitMarket {
     fn start_market_stream(&mut self) -> anyhow::Result<()> {
         BLOCK_ON(async { self.async_start_market_stream().await })
     }
+
+    fn get_latest_archive_date(&self) -> anyhow::Result<MicroSec> {
+        todo!()
+    }
 }
 
 impl BybitMarket {
-    async fn async_download_lastest(&mut self) -> anyhow::Result<i64> {
+    async fn async_download_lastest(&mut self, verbose: bool) -> anyhow::Result<i64> {
+        if verbose {
+            print!("async_download_lastest");
+            flush_log();
+        }
+
         let trades = BybitRestApi::get_recent_trades(&self.server_config, &self.config).await?;
         let rec = trades.len() as i64;
 
-        let tx = self.db.lock().unwrap().start_thread();
-        tx.send(trades)?;
+        if verbose {
+            println!("rec: {}", rec);
+            flush_log();
+        }
+
+        {
+            log::debug!("get db");
+            flush_log();
+            let tx = self.db.lock();
+
+            if tx.is_err() {
+                log::error!("Error in self.db.lock: {:?}", tx);
+                return Err(anyhow::anyhow!("Error in self.db.lock: {:?}", tx));
+            }
+            let mut lock = tx.unwrap();
+            let tx = lock.start_thread();
+
+            log::debug!("start thread done");
+            tx.send(trades)?;
+        }
 
         Ok(rec)
     }
@@ -536,20 +567,27 @@ impl BybitMarket {
             return Ok(());
         }
 
-        let mut lock = self.db.lock().unwrap();
-        let db_channel = lock.start_thread();
-        drop(lock);
+        let db_channel = {
+            let mut lock = self.db.lock().unwrap();
+            lock.start_thread()
+        };
 
         let mut orderbook = self.board.clone();
 
         let server_config = self.server_config.clone();
         let config = self.config.clone();
 
+        let hub_channel = MARKET_HUB.open_channel();
+
         self.public_handler = Some(tokio::task::spawn(async move {
             let mut public_ws = BybitPublicWsClient::new(&server_config, &config).await;
 
+            let exchange_name = server_config.exchange_name.clone();
+            let trade_category = config.trade_category.clone();
+            let trade_symbol = config.trade_symbol.clone();
+
             public_ws.connect().await;
-            let mut ws_stream = public_ws.open_stream().await;    
+            let mut ws_stream = public_ws.open_stream().await;
             let mut ws_stream = Box::pin(ws_stream);
 
             loop {
@@ -571,27 +609,45 @@ impl BybitMarket {
                 match messages {
                     MultiMarketMessage::Trade(trade) => {
                         log::debug!("Trade: {:?}", trade);
-                        let r = db_channel.send(trade);
+                        let r = db_channel.send(trade.clone());
 
                         if r.is_err() {
                             log::error!("Error in db_channel.send: {:?}", r);
                         }
+
+                        for message in trade {
+                            let r = hub_channel.send(BroadcastMessage {
+                                exchange: exchange_name.clone(),
+                                category: trade_category.clone(),
+                                symbol: trade_symbol.clone(),
+                                msg: MarketMessage::Trade(message),
+                            });
+                            if r.is_err() {
+                                log::error!("Error in hub_channel.send: {:?}", r);
+                            }
+                        }
                     }
                     MultiMarketMessage::Orderbook(ob) => {
                         let board = BoardTransfer::from_orderbook(&ob);
                         let snapshot = ob.snapshot;
 
-                        let mut b = orderbook.write().unwrap();
-                        b.update(&board.bids, &board.asks, snapshot);
-                        drop(b);
+                        {
+                            let mut b = orderbook.write().unwrap();
+                            b.update(&board.bids, &board.asks, snapshot);
+                        }
+                    }
+                    MultiMarketMessage::Control(control) => {
+                        // TODO: alert or recovery.
+                        if control.status == false {
+                            log::error!("Control message: {:?}", control);
+                        }
                     }
                     _ => {
-                        log::warn!("Not implemented/unexpected message: {:?}", messages);
+                        log::info!("Market stream message: {:?}", messages);
                     } /*
                       MultiMarketMessage::Order(_) => todo!(),
                       MultiMarketMessage::Account(_) => todo!(),
                       MultiMarketMessage::Message(_) => todo!(),
-                      MultiMarketMessage::Control(_) => todo!(),
                       */
                 }
             }
@@ -601,6 +657,133 @@ impl BybitMarket {
     }
 }
 
+#[cfg(test)]
+mod bybit_test {
+    use rust_decimal_macros::dec;
+
+    use crate::{Bybit, BybitConfig};
+
+    #[test]
+    fn test_create() {
+        let bybit = Bybit::new(false);
+
+        assert!(bybit.enable_order == false);
+    }
+
+    #[test]
+    fn test_open_market() {
+        let market = Bybit::new(false).open_market(&BybitConfig::BTCUSDT());
+        assert!(market.get_config().trade_symbol == "BTCUSDT");
+    }
+
+    #[test]
+    fn test_limit_order() {
+        let mut bybit = Bybit::new(false);
+        let config = BybitConfig::BTCUSDT();
+
+        let rec = bybit.limit_order(&config, "Buy", dec![45000.0], dec![0.001], None);
+        println!("{:?}", rec);
+        assert!(rec.is_err()); // first enable flag.
+
+        bybit.set_enable_order_with_my_own_risk(true);
+        let rec = bybit.limit_order(&config, "Buy", dec![45000.0], dec![0.001], None);
+        println!("{:?}", rec);
+        assert!(rec.is_ok()); // first enable flag.
+    }
+
+    #[test]
+    fn test_market_order() {
+        let mut bybit = Bybit::new(false);
+        let config = BybitConfig::BTCUSDT();
+
+        let rec = bybit.market_order(&config, "Buy", dec![0.001], None);
+        println!("{:?}", rec);
+        assert!(rec.is_err()); // first enable flag.
+
+        bybit.set_enable_order_with_my_own_risk(true);
+        let rec = bybit.market_order(&config, "Buy", dec![0.001], None);
+        println!("{:?}", rec);
+        assert!(rec.is_ok()); // first enable flag.
+    }
+
+    #[test]
+    fn test_cancel_order() -> anyhow::Result<()> {
+        let mut bybit = Bybit::new(false);
+        let config = BybitConfig::BTCUSDT();
+
+        bybit.set_enable_order_with_my_own_risk(true);
+        let rec = bybit.limit_order(&config, "Buy", dec![45000.0], dec![0.001], None)?;
+
+        let order_id = rec[0].order_id.clone();
+
+        let rec = bybit.cancel_order(&config, &order_id)?;
+        println!("{:?}", rec);
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_get_open_orders() -> anyhow::Result<()> {
+        let mut bybit = Bybit::new(false);
+        let config = BybitConfig::BTCUSDT();
+
+        let rec = bybit.get_open_orders(&config)?;
+        println!("{:?}", rec);
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_get_account() {
+        let mut bybit = Bybit::new(false);
+        let config = BybitConfig::BTCUSDT();
+
+        let rec = bybit.get_account();
+        println!("{:?}", rec);
+        assert!(rec.is_ok());
+    }
+}
+
+#[cfg(test)]
+mod market_test {
+    use crate::BybitConfig;
+
+    #[test]
+    fn test_create() {
+        use super::*;
+        let server_config = BybitServerConfig::new(false);
+        let market_config = BybitConfig::BTCUSDT();
+
+        let market = BybitMarket::new(&server_config, &market_config);
+    }
+
+    #[ignore]
+    #[tokio::test]
+    async fn test_download_archive() {
+        use super::*;
+        let server_config = BybitServerConfig::new(false);
+        let market_config = BybitConfig::BTCUSDT();
+
+        let mut market = BybitMarket::new(&server_config, &market_config);
+
+        let rec = market.async_download_archives(1, true, true, false).await;
+        assert!(rec.is_ok());
+    }
+
+    #[ignore]
+    #[test]
+    fn test_download_latest() {
+        use super::*;
+        let server_config = BybitServerConfig::new(false);
+        let market_config = BybitConfig::BTCUSDT();
+
+        let mut market = BybitMarket::new(&server_config, &market_config);
+
+        let rec = market.download_latest(true);
+        assert!(rec.is_ok());
+    }
+}
+
 /*
 //----------- FOR BACKUP --------------------------------------------
 //#[derive(Debug)]
diff --git a/exchanges/bybit/src/message.rs b/exchanges/bybit/src/message.rs
index dd43e331..8f3eac37 100644
--- a/exchanges/bybit/src/message.rs
+++ b/exchanges/bybit/src/message.rs
@@ -10,11 +10,14 @@ use std::str::FromStr;
 use pyo3::pyclass;
 use rust_decimal::Decimal;
 use rust_decimal_macros::dec;
+use serde::Deserializer;
 use serde_derive::{Deserialize, Serialize};
 use serde_json::Value;
 
 use rbot_lib::common::{
-    msec_to_microsec, string_to_decimal, string_to_i64, time_string, AccountStatus, Board, BoardTransfer, ControlMessage, Kline, LogStatus, MarketConfig, MarketMessage, MicroSec, MultiMarketMessage, Order, OrderBookRaw, OrderSide, OrderStatus, OrderType, Trade
+    msec_to_microsec, string_to_decimal, string_to_i64, time_string, AccountCoins, AccountPair,
+    Board, BoardTransfer, Coin, ControlMessage, Kline, LogStatus, MarketConfig, MarketMessage,
+    MicroSec, MultiMarketMessage, Order, OrderBookRaw, OrderSide, OrderStatus, OrderType, Trade,
 };
 
 use crate::Bybit;
@@ -46,10 +49,24 @@ pub fn bybit_order_status(status: &str) -> OrderStatus {
     }
 }
 
+fn deserialize_ret_code<'de, D>(deserializer: D) -> Result<i64, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let value = <Value as serde::Deserialize>::deserialize(deserializer)?;
+    if let Some(code) = value.get("retCode").or_else(|| value.get("return_code")).and_then(Value::as_i64) {
+        Ok(code as i64)
+    } else {
+        Err(serde::de::Error::custom("retCode/ret_code field not found"))
+    }
+}
+
+
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[pyclass]
 pub struct BybitRestResponse {
-    #[serde(rename = "retCode")]
+    #[serde(rename="retCode")]
     pub return_code: i64,
     #[serde(rename = "retMsg")]
     pub return_message: String,
@@ -61,6 +78,12 @@ pub struct BybitRestResponse {
     pub body: Value,
 }
 
+impl BybitRestResponse {
+    pub fn is_success(&self) -> bool {
+        self.return_code == 0
+    }
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[pyclass]
 pub struct BybitRestBoard {
@@ -494,11 +517,11 @@ pub struct BybitOrderStatus {
 }
 
 impl Into<Order> for &BybitOrderStatus {
-    
     fn into(self) -> Order {
         let order_type = OrderType::from(&self.orderType);
 
         Order {
+            category: "linear".to_string(),     // for default
             symbol: self.symbol.clone(),
             create_time: self.createdTime,
             status: bybit_order_status(&self.orderStatus),
@@ -517,8 +540,8 @@ impl Into<Order> for &BybitOrderStatus {
             commission: self.cumExecFee,
             commission_asset: "".to_string(),
             is_maker: order_type.is_maker(),
-            message: "".to_string(),                // DUMMY value
-            commission_home: dec![0.0],             // DUMMY value
+            message: "".to_string(),    // DUMMY value
+            commission_home: dec![0.0], // DUMMY value
             commission_foreign: dec![0.0],
             home_change: dec![0.0],
             foreign_change: dec![0.0],
@@ -541,10 +564,12 @@ pub struct BybitMultiOrderStatus {
 
 impl Into<Vec<Order>> for BybitMultiOrderStatus {
     fn into(self) -> Vec<Order> {
+        let category = self.category.clone();
         let mut orders: Vec<Order> = vec![];
 
         for order in self.list.iter() {
-            let o: Order = order.into();
+            let mut o: Order = order.into();
+            o.category = category.clone();
             orders.push(o);
         }
 
@@ -562,7 +587,7 @@ pub struct BybitAccountInformation {}
 #[serde(untagged)]
 pub enum BybitPublicWsMessage {
     Status(BybitWsStatus),
-    Pong(BybitWsPongReply), 
+    Pong(BybitWsPongReply),
     Trade(BybitWsTradeMessage),
     Orderbook(BybitWsOrderbookMessage),
 }
@@ -624,7 +649,7 @@ impl Into<MultiMarketMessage> for BybitPublicWsMessage {
                 })
             }
             BybitPublicWsMessage::Pong(pong) => {
-                return MultiMarketMessage::Control(ControlMessage{
+                return MultiMarketMessage::Control(ControlMessage {
                     status: true,
                     operation: pong.op,
                     message: pong.conn_id,
@@ -655,6 +680,15 @@ pub struct BybitWsStatus {
     pub args: Option<Vec<String>>,
 }
 
+impl Into<ControlMessage> for BybitWsStatus {
+    fn into(self) -> ControlMessage {
+        ControlMessage {
+            status: self.success,
+            operation: self.op.clone(),
+            message: format!("{:?} args:[{:?}]", self.ret_msg, self.args),
+        }
+    }
+}
 
 /*
 "{\"op\":\"pong\",\"args\":[\"1707031861056\"],\"conn_id\":\"cmsuqo1qo29shn0o3qb0-44fy\"}"
@@ -678,8 +712,6 @@ pub struct BybitWsTradeMessage {
     pub timestamp: BybitTimestamp,
 }
 
-
-
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct BybitWsTrade {
     #[serde(rename = "i")]
@@ -908,47 +940,61 @@ pub fn merge_order_and_execution(
 */
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct BybitAccountStatus {
-    accountType: String,
+    pub accountType: String,
     #[serde(deserialize_with = "string_to_decimal")]
-    accountLTV: Decimal, // account total borrowed size / (account total equity + account total borrowed size). In non-unified mode & unified (inverse) & unified (isolated_margin), the field will be returned as an empty string.
+    pub accountLTV: Decimal, // account total borrowed size / (account total equity + account total borrowed size). In non-unified mode & unified (inverse) & unified (isolated_margin), the field will be returned as an empty string.
     #[serde(deserialize_with = "string_to_decimal")]
-    accountIMRate: Decimal, // account initial margin rate. In non-unified mode & unified (inverse) & unified (isolated_margin), the field will be returned as an empty string.
+    pub accountIMRate: Decimal, // account initial margin rate. In non-unified mode & unified (inverse) & unified (isolated_margin), the field will be returned as an empty string.
     #[serde(deserialize_with = "string_to_decimal")]
-    accountMMRate: Decimal, // account maintenance margin rate. In non-unified mode & unified (inverse) & unified (isolated_margin), the field will be returned as an empty string.
+    pub accountMMRate: Decimal, // account maintenance margin rate. In non-unified mode & unified (inverse) & unified (isolated_margin), the field will be returned as an empty string.
     #[serde(deserialize_with = "string_to_decimal")]
-    totalEquity: Decimal, // account total equity. In non-unified mode & unified (inverse) & unified (isolated_margin), the field will be returned as an empty string.
+    pub totalEquity: Decimal, // account total equity. In non-unified mode & unified (inverse) & unified (isolated_margin), the field will be returned as an empty string.
     #[serde(deserialize_with = "string_to_decimal")]
-    totalWalletBalance: Decimal, // account total wallet balance. In non-unified mode & unified (inverse) & unified (isolated_margin), the field will be returned as an empty string.
+    pub totalWalletBalance: Decimal, // account total wallet balance. In non-unified mode & unified (inverse) & unified (isolated_margin), the field will be returned as an empty string.
     #[serde(deserialize_with = "string_to_decimal")]
-    totalMarginBalance: Decimal, // account total margin balance. In non-unified mode & unified (inverse) & unified (isolated_margin), the field will be returned as an empty string.
+    pub totalMarginBalance: Decimal, // account total margin balance. In non-unified mode & unified (inverse) & unified (isolated_margin), the field will be returned as an empty string.
     #[serde(deserialize_with = "string_to_decimal")]
-    totalAvailableBalance: Decimal, // account total available balance. In non-unified mode & unified (inverse) & unified (isolated_margin), the field will be returned as an empty string.
+    pub totalAvailableBalance: Decimal, // account total available balance. In non-unified mode & unified (inverse) & unified (isolated_margin), the field will be returned as an empty string.
     #[serde(deserialize_with = "string_to_decimal")]
-    totalPerpUPL: Decimal, // account total unrealized PnL. In non-unified mode & unified (inverse) & unified (isolated_margin), the field will be returned as an empty string.
+    pub totalPerpUPL: Decimal, // account total unrealized PnL. In non-unified mode & unified (inverse) & unified (isolated_margin), the field will be returned as an empty string.
     #[serde(deserialize_with = "string_to_decimal")]
-    totalInitialMargin: Decimal, // account total initial margin. In non-unified mode & unified (inverse) & unified (isolated_margin), the field will be returned as an empty string.
+    pub totalInitialMargin: Decimal, // account total initial margin. In non-unified mode & unified (inverse) & unified (isolated_margin), the field will be returned as an empty string.
     #[serde(deserialize_with = "string_to_decimal")]
-    totalMaintenanceMargin: Decimal, // account total maintenance margin. In non-unified mode & unified (inverse) & unified (isolated_margin), the field will be returned as an empty string.
-    coin: Vec<BybitAccountCoin>,
+    pub totalMaintenanceMargin: Decimal, // account total maintenance margin. In non-unified mode & unified (inverse) & unified (isolated_margin), the field will be returned as an empty string.
+    pub coin: Vec<BybitAccountCoin>,
+}
+
+impl Into<AccountCoins> for BybitAccountStatus {
+    fn into(self) -> AccountCoins {
+        let mut coins: Vec<Coin> = vec![];
+        for coin in self.coin.iter() {
+            let c: Coin = coin.into();
+            coins.push(c);
+        }
+
+        AccountCoins { coins: coins }
+    }
 }
 
-// TODO: implment
-// TODO: MarketConfigを引き継ぐように変更する。
-// Margin取引のときは、homeだけ有効
-// home_freeは有効マージン相当として扱う
-impl Into<AccountStatus> for &BybitAccountStatus {
-    fn into(self) -> AccountStatus {
-        let account = AccountStatus {
-            home: dec![0.0],
-            home_free: dec![0.0],
-            home_locked: dec![0.0],
-            foreign: dec![0.0],
-            foreign_free: dec![0.0],
-            foreign_locked: dec![0.0],
-        };
-
-        account
+pub fn convert_coin_to_account_status(
+    config: &MarketConfig,
+    coins: Vec<BybitAccountCoin>,
+) -> AccountPair {
+    let mut account = AccountPair::default();
+
+    for coin in coins.iter() {
+        if coin.coin == config.home_currency {
+            account.home.volume = coin.equity;
+            account.home.free = coin.availableToWithdraw;
+            account.home.locked = coin.locked;
+        } else if coin.coin == config.foreign_currency {
+            account.foreign.volume = coin.equity;
+            account.foreign.free = coin.availableToWithdraw;
+            account.foreign.locked = coin.locked;
+        }
     }
+
+    account
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -987,6 +1033,47 @@ pub struct BybitAccountCoin {
     marginCollateral: bool,
 }
 
+impl Into<Coin> for &BybitAccountCoin {
+    fn into(self) -> Coin {
+        Coin {
+            symbol: self.coin.clone(),
+            volume: self.equity,
+            free: self.availableToWithdraw,
+            locked: self.locked,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct BybitAccountResponse {
+    list: Vec<BybitAccountResponseList>
+
+ //   coin: Vec<BybitAccountCoin>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct BybitAccountResponseList {
+    coin: Vec<BybitAccountCoin>,
+}
+
+
+
+// TODO: check if flatten is OK.
+impl Into<AccountCoins> for BybitAccountResponse {
+    fn into(self) -> AccountCoins {
+        let mut coins: Vec<Coin> = vec![];
+        for l in self.list {
+            for coin in l.coin.iter() {
+            let c: Coin = coin.into();
+            coins.push(c);
+            }
+        }
+
+        AccountCoins { coins: coins }
+    }
+}
+
+
 #[cfg(test)]
 #[allow(non_snake_case)]
 #[allow(unused_imports)]
@@ -994,6 +1081,7 @@ pub struct BybitAccountCoin {
 mod bybit_message_test {
     use super::*;
     use polars_core::utils::rayon::result;
+    use rbot_lib::common::init_debug_log;
 
     use crate::message::{
         BybitAccountStatus, BybitExecution, BybitMultiOrderStatus, BybitRestResponse,
@@ -1010,10 +1098,8 @@ mod bybit_message_test {
         "#;
 
         let result = serde_json::from_str::<BybitRestResponse>(&message);
-
-        assert!(result.is_ok());
-
         println!("{:?}", result);
+        assert!(result.is_ok());
     }
 
     #[test]
@@ -1260,39 +1346,89 @@ mod bybit_message_test {
         let order = serde_json::from_str::<BybitUserMessage>(message);
         println!("{:?}", order);
 
-
         let message = r#"{"topic":"execution","id":"100467532_BTCUSDT_8883610598","creationTime":1705761437507,"data":[{"category":"linear","symbol":"BTCUSDT","closedSize":"0","execFee":"0.02285465","execId":"2800474f-1e3d-571e-9cc8-46e3bcb82699","execPrice":"41553.9","execQty":"0.001","execType":"Trade","execValue":"41.5539","feeRate":"0.00055","tradeIv":"","markIv":"","blockTradeId":"","markPrice":"41547.63","indexPrice":"","underlyingPrice":"","leavesQty":"0","orderId":"e4385ca4-59cf-4ef8-aa34-61b7ad99ae84","orderLinkId":"SkeltonAgentlp9qlB-0001","orderPrice":"43607.8","orderQty":"0.001","orderType":"Market","stopOrderType":"UNKNOWN","side":"Buy","execTime":"1705761437503","isLeverage":"0","isMaker":false,"seq":8883610598,"marketUnit":"","createType":"CreateByUser"}]}"#;
         let execution = serde_json::from_str::<BybitUserMessage>(message);
         println!("{:?}", execution);
-
     }
 
-
-
     #[test]
     fn test_bybit_order_and_execution() {
         let message = r#"
         [{"orderId":"43fac7fc-e2ae-4e80-bb50-0f2ff171fcc0","orderLinkId":"","blockTradeId":"","symbol":"BTCUSDT","price":"40643.7","qty":"0.001","side":"Sell","isLeverage":"","positionIdx":0,"orderStatus":"Filled","cancelType":"UNKNOWN","rejectReason":"EC_NoError","avgPrice":"42782","leavesQty":"0","leavesValue":"0","cumExecQty":"0.001","cumExecValue":"42.782","cumExecFee":"0.0235301","timeInForce":"IOC","orderType":"Market","stopOrderType":"","orderIv":"","triggerPrice":"0","takeProfit":"0","stopLoss":"0","tpTriggerBy":"","slTriggerBy":"","triggerDirection":0,"triggerBy":"","lastPriceOnCreated":"42782.8","reduceOnly":true,"closeOnTrigger":true,"smpType":"None","smpGroup":0,"smpOrderId":"","tpslMode":"UNKNOWN","tpLimitPrice":"0","slLimitPrice":"0","placeType":"","createdTime":1707036474698,"updatedTime":1707036474702}]
         "#;
-        let order = serde_json::from_str::<Vec<BybitOrderStatus>>(message);        
+        let order = serde_json::from_str::<Vec<BybitOrderStatus>>(message);
         println!("{:?}", order);
-        
+
         let message = r#"        
         [{"category":"linear","symbol":"BTCUSDT","orderId":"43fac7fc-e2ae-4e80-bb50-0f2ff171fcc0","orderLinkId":"","side":"Sell","orderPrice":"40643.7","orderQty":"0.001","leavesQty":"0","orderType":"Market","execFee":"0.0235301","execId":"be10c344-e1d3-5949-8f56-d45dd192acca","execPrice":"42782","execQty":"0.001","execValue":"42.782","execTime":1707036474699,"isMaker":false,"feeRate":"0.00055","seq":8909836703}]
         "#;
 
-
-        let execution = serde_json::from_str::<Vec<BybitExecution>>(message);        
+        let execution = serde_json::from_str::<Vec<BybitExecution>>(message);
         println!("{:?}", execution);
+    }
+
+    #[test]
+    fn test_account_status_message() -> anyhow::Result<()> {
+        init_debug_log();
 
+        let message: &str = r#"
+        {"retCode":0,"retMsg":"OK","result":{"list":[{"totalEquity":"11671.04063119","accountIMRate":"0.0784","totalMarginBalance":"11671.04063119","totalInitialMargin":"915.40917399","accountType":"UNIFIED","totalAvailableBalance":"10755.6314572","accountMMRate":"0.0042","totalPerpUPL":"1471.35122086","totalWalletBalance":"10199.68941033","accountLTV":"0","totalMaintenanceMargin":"50.11637468","coin":[{"availableToBorrow":"","bonus":"0","accruedInterest":"0","availableToWithdraw":"10196.98720872","totalOrderIM":"12.1254","equity":"11667.94862481","totalPositionMM":"49.37769736","usdValue":"11671.04063119","unrealisedPnl":"1470.96141609","collateralSwitch":true,"spotHedgingQty":"0","borrowAmount":"0.000000000000000000","totalPositionIM":"903.04125483","walletBalance":"10196.98720872","cumRealisedPnl":"196.98720872","locked":"0","marginCollateral":true,"coin":"USDT"},{"availableToBorrow":"","bonus":"","accruedInterest":"","availableToWithdraw":"","totalOrderIM":"","equity":"","totalPositionMM":"","usdValue":"","unrealisedPnl":"","collateralSwitch":false,"spotHedgingQty":"0","borrowAmount":"","totalPositionIM":"","walletBalance":"","cumRealisedPnl":"","locked":"","marginCollateral":true,"coin":"BTC"}]}]},"retExtInfo":{},"time":1707918827165}
+        "#;
+
+        let response = serde_json::from_str::<BybitRestResponse>(message);
+
+        println!("{:?}", response);
+
+        let response = response.unwrap().body;
+
+        println!("R>>>{:?}", response);
+        let account_response = serde_json::from_value::<BybitAccountResponse>(response)?;
+        println!("AR>>{:?}", account_response);
+
+        //let ac = (account_response.list);
+
+        //let ac: AccountCoins = account_response.into();
+
+        Ok(())
     }
 
-    
+    #[test]
+    fn test_parse_account_coin() {
+        let MESSAGE: &str = r#"
+        {
+            "list":
+            [
+                {
+                    "totalEquity":"11671.04063119","accountIMRate":"0.0784","totalMarginBalance":"11671.04063119","totalInitialMargin":"915.40917399","accountType":"UNIFIED","totalAvailableBalance":"10755.6314572","accountMMRate":"0.0042","totalPerpUPL":"1471.35122086","totalWalletBalance":"10199.68941033","accountLTV":"0","totalMaintenanceMargin":"50.11637468",
+                    "coin":[
+                    {"availableToBorrow":"","bonus":"0","accruedInterest":"0","availableToWithdraw":"10196.98720872","totalOrderIM":"12.1254","equity":"11667.94862481","totalPositionMM":"49.37769736","usdValue":"11671.04063119","unrealisedPnl":"1470.96141609","collateralSwitch":true,"spotHedgingQty":"0","borrowAmount":"0.000000000000000000","totalPositionIM":"903.04125483","walletBalance":"10196.98720872","cumRealisedPnl":"196.98720872","locked":"0","marginCollateral":true,"coin":"USDT"},
+                    {"availableToBorrow":"","bonus":"","accruedInterest":"","availableToWithdraw":"","totalOrderIM":"","equity":"","totalPositionMM":"","usdValue":"","unrealisedPnl":"","collateralSwitch":false,"spotHedgingQty":"0","borrowAmount":"","totalPositionIM":"","walletBalance":"","cumRealisedPnl":"","locked":"","marginCollateral":true,"coin":"BTC"}
+                    ]
+                }
+            ]
+        }
+        "#;
 
+        let response = serde_json::from_str::<BybitAccountResponse>(MESSAGE);
+        println!("{:?}", response);
+        assert!(response.is_ok());
+    }
 
+    #[test]
+    fn test_parse_coin() {
+        const M: &str = r#"{"availableToBorrow":"","bonus":"0","accruedInterest":"0","availableToWithdraw":"10196.98720872","totalOrderIM":"12.1254","equity":"11667.94862481","totalPositionMM":"49.37769736","usdValue":"11671.04063119","unrealisedPnl":"1470.96141609","collateralSwitch":true,"spotHedgingQty":"0","borrowAmount":"0.000000000000000000","totalPositionIM":"903.04125483","walletBalance":"10196.98720872","cumRealisedPnl":"196.98720872","locked":"0","marginCollateral":true,"coin":"USDT"}"#;
 
+        let coins = serde_json::from_str::<BybitAccountCoin>(M);
 
+        println!("{:?}", coins);
+        assert!(coins.is_ok());
+    }
 
+    fn test_parse() {
+        let message: &str = r#"
+        {"retCode":0,"retMsg":"OK","result":{"list":[{"totalEquity":"11745.04972951","accountIMRate":"0.0779","totalMarginBalance":"11745.04972951","totalInitialMargin":"916.00037165","accountType":"UNIFIED","totalAvailableBalance":"10829.04935785","accountMMRate":"0.0042","totalPerpUPL":"1543.8483218","totalWalletBalance":"10201.2014077","accountLTV":"0","totalMaintenanceMargin":"50.14874128","coin":[{"availableToBorrow":"","bonus":"0","accruedInterest":"0","availableToWithdraw":"10191.91657171","totalOrderIM":"12.1254","equity":"11734.3597278","totalPositionMM":"49.37769736","usdValue":"11745.04972951","unrealisedPnl":"1542.44315609","collateralSwitch":true,"spotHedgingQty":"0","borrowAmount":"0.000000000000000000","totalPositionIM":"903.04125483","walletBalance":"10191.91657171","cumRealisedPnl":"191.91657171","locked":"0","marginCollateral":true,"coin":"USDT"},{"availableToBorrow":"","bonus":"","accruedInterest":"","availableToWithdraw":"","totalOrderIM":"","equity":"","totalPositionMM":"","usdValue":"","unrealisedPnl":"","collateralSwitch":false,"spotHedgingQty":"0","borrowAmount":"","totalPositionIM":"","walletBalance":"","cumRealisedPnl":"","locked":"","marginCollateral":true,"coin":"BTC"}]}]},"retExtInfo":{},"time":1708051591009}        
+        "#;
+    }
 }
 
 /*
diff --git a/exchanges/bybit/src/rest.rs b/exchanges/bybit/src/rest.rs
index 6d71a3e2..b7cdb698 100644
--- a/exchanges/bybit/src/rest.rs
+++ b/exchanges/bybit/src/rest.rs
@@ -1,12 +1,14 @@
 #![allow(unused)]
 // Copyright(c) 2022-2024. yasstake. All rights reserved.
 
+use std::convert;
 use std::fmt::format;
 
 use csv::StringRecord;
 use polars_core::export::num::FromPrimitive;
 use rbot_lib::common::time_string;
-use rbot_lib::common::AccountStatus;
+use rbot_lib::common::AccountCoins;
+use rbot_lib::common::AccountPair;
 use rbot_lib::common::BoardTransfer;
 use rbot_lib::common::Kline;
 use rbot_lib::common::LogStatus;
@@ -32,7 +34,11 @@ use rbot_lib::common::{
 
 use rbot_lib::net::{rest_get, rest_post, RestApi};
 
+use crate::message::convert_coin_to_account_status;
 use crate::message::microsec_to_bybit_timestamp;
+use crate::message::BybitAccountCoin;
+use crate::message::BybitAccountResponse;
+use crate::message::BybitAccountStatus;
 
 use super::config::BybitServerConfig;
 use super::message::BybitKlinesResponse;
@@ -122,6 +128,7 @@ impl RestApi<BybitServerConfig> for BybitRestApi {
             1000 // max records.
         );
 
+
         let r = Self::rest_get(server, path, &params)
             .await
             .with_context(|| {
@@ -135,6 +142,9 @@ impl RestApi<BybitServerConfig> for BybitRestApi {
             .with_context(|| format!("parse error in get_recent_trades"))?;
 
         Ok(result.into())
+
+
+//        Err(anyhow!("not implemented"))
     }
 
     async fn get_trade_klines(
@@ -210,7 +220,7 @@ impl RestApi<BybitServerConfig> for BybitRestApi {
         };
 
         let order = BybitOrderRequest {
-            category: category,
+            category: category.clone(),
             symbol: symbol.clone(),
             side: side.to_string(),
             order_type: order_type.to_string(),
@@ -240,6 +250,7 @@ impl RestApi<BybitServerConfig> for BybitRestApi {
 
         // in bybit only order id is valid when order is created.
         let mut order = Order {
+            category: category,
             symbol: symbol,
             create_time: msec_to_microsec(result.time),
             status: OrderStatus::New,
@@ -284,8 +295,9 @@ impl RestApi<BybitServerConfig> for BybitRestApi {
         config: &MarketConfig,
         order_id: &str,
     ) -> anyhow::Result<Order> {
+        let category = config.trade_category.clone();
         let message = CancelOrderMessage {
-            category: config.trade_category.clone(),
+            category: category.clone(),
             symbol: config.trade_symbol.clone(),
             order_id: order_id.to_string(),
         };
@@ -304,6 +316,7 @@ impl RestApi<BybitServerConfig> for BybitRestApi {
         let r = serde_json::from_value::<BybitOrderRestResponse>(result.body)?;
 
         let mut order = Order {
+            category,
             symbol: "".to_string(),
             create_time: msec_to_microsec(result.time),
             status: OrderStatus::Canceled,
@@ -342,7 +355,7 @@ impl RestApi<BybitServerConfig> for BybitRestApi {
     // TODO: implement paging.
     async fn open_orders(
         server: &BybitServerConfig,
-        config: &MarketConfig,
+config: &MarketConfig,
     ) -> anyhow::Result<Vec<Order>> {
         let query_string = format!(
             "category={}&symbol={}&limit=50",
@@ -378,10 +391,24 @@ impl RestApi<BybitServerConfig> for BybitRestApi {
     }
 
     async fn get_account(
-        _server: &BybitServerConfig,
-        _config: &MarketConfig,
-    ) -> anyhow::Result<AccountStatus> {
-        todo!()
+        server: &BybitServerConfig
+    ) -> anyhow::Result<AccountCoins> {
+
+        let path = "/v5/account/wallet-balance";
+        // TODO: implement otherthn accountType=UNIFIED(e.g.inverse)
+        let query_string = format!("accountType=UNIFIED");
+        //let query_string = format!("accountType=UNIFIED");
+
+        let response = Self::get_sign(&server, path, &query_string)
+            .await
+            .with_context(|| format!("get_account error: {}/{}/{}", &server.get_rest_server(), path, &query_string))?;
+
+        ensure!(response.is_success(), format!("return_code = {}, msg={}", response.is_success(), response.return_message));
+
+        let account_status = serde_json::from_value::<BybitAccountResponse>(response.body)?;
+        let coins: AccountCoins = account_status.into();
+
+        Ok(coins)
     }
 
     fn format_historical_data_url(
@@ -473,14 +500,14 @@ impl BybitRestApi {
         query_string: &str,
     ) -> anyhow::Result<BybitRestResponse> {
         let timestamp = format!("{}", NOW() / 1_000);
-        let api_key = server.get_api_key();
+        let api_key = server.get_api_key().extract();
+        let api_secret = server.get_api_secret().extract();
         let recv_window = "5000";
 
-        let param_to_sign = format!("{}{}{}{}", timestamp, api_key, recv_window, query_string);
-        let sign = hmac_sign(&server.get_api_secret(), &param_to_sign);
+        let param_to_sign = format!("{}{}{}{}", timestamp, api_key.clone(), recv_window, query_string);
+        let sign = hmac_sign(&api_secret, &param_to_sign);
 
         let mut headers: Vec<(&str, &str)> = vec![];
-        let api_key = server.get_api_key();
 
         headers.push(("X-BAPI-SIGN", &sign));
         headers.push(("X-BAPI-API-KEY", &api_key));
@@ -489,7 +516,7 @@ impl BybitRestApi {
 
         let result = rest_get(&server.rest_server, path, headers, Some(query_string), None)
             .await
-            .with_context(|| format!("get_sign error: {}/{}", &server.rest_server, path))?;
+            .with_context(|| format!("get_sign error: {}/{}/{}", &server.rest_server, path, query_string))?;
 
         Self::parse_rest_response(result)
     }
@@ -500,14 +527,14 @@ impl BybitRestApi {
         body: &str,
     ) -> anyhow::Result<BybitRestResponse> {
         let timestamp = format!("{}", NOW() / 1_000);
-        let api_key = server.get_api_key();
+        let api_key = server.get_api_key().extract();
+        let api_secret = server.get_api_secret().extract();
         let recv_window = "5000";
 
-        let param_to_sign = format!("{}{}{}{}", timestamp, api_key, recv_window, body);
-        let sign = hmac_sign(&server.get_api_secret(), &param_to_sign);
+        let param_to_sign = format!("{}{}{}{}", timestamp, api_key.clone(), recv_window, body);
+        let sign = hmac_sign(&api_secret, &param_to_sign);
 
         let mut headers: Vec<(&str, &str)> = vec![];
-        let api_key = server.get_api_key();
 
         headers.push(("X-BAPI-SIGN", &sign));
         headers.push(("X-BAPI-API-KEY", &api_key));
@@ -541,7 +568,7 @@ impl BybitRestApi {
             .with_context(|| format!("parse error in parse_rest_response: {:?}", response))?;
 
         ensure!(
-            result.return_code == 0,
+            result.is_success(),
             format!("parse rest response error = {}", result.return_message)
         );
 
@@ -590,347 +617,6 @@ impl BybitRestApi {
     }
 }
 
-/*
-
-
-pub fn new_limit_order(
-    server: &BybitServerConfig,
-    config: &MarketConfig,
-    side: OrderSide,
-    price: Decimal,
-    size: Decimal,
-    client_order_id: Option<&str>,
-) -> Result<Order, String> {
-    new_order(
-        server,
-        config,
-        side,
-        price,
-        size,
-        OrderType::Limit,
-        client_order_id,
-    )
-}
-
-pub fn new_market_order(
-    server: &BybitServerConfig,
-    config: &MarketConfig,
-    side: OrderSide,
-    size: Decimal,
-    client_order_id: Option<&str>,
-) -> Result<Order, String> {
-    new_order(
-        server,
-        config,
-        side,
-        dec![0.0],
-        size,
-        OrderType::Market,
-        client_order_id,
-    )
-}
-
-/// create new limit order
-/// https://bybit-exchange.github.io/docs/v5/order/create-order
-pub fn new_order(
-    server: &BybitServerConfig,
-    config: &MarketConfig,
-    side: OrderSide,
-    price: Decimal, // when order_type is Market, this value is ignored.
-    size: Decimal,
-    order_type: OrderType,
-    client_order_id: Option<&str>,
-) -> Result<Order, String> {
-    let category = config.trade_category.clone();
-    let symbol = config.trade_symbol.clone();
-
-    let price = if order_type == OrderType::Market {
-        None
-    } else {
-        Some(price)
-    };
-
-    let order = BybitOrderRequest {
-        category: category,
-        symbol: config.trade_symbol.clone(),
-        side: side.to_string(),
-        order_type: order_type.to_string(),
-        qty: size,
-        order_link_id: client_order_id,
-        price: price,
-    };
-
-    log::debug!("order={:?}", order);
-
-    let order_json = serde_json::to_string(&order).unwrap();
-    log::debug!("order_json={}", order_json);
-
-    let path = "/v5/order/create";
-
-    let result = bybit_post_sign(&server, path, &order_json);
-
-    if result.is_err() {
-        let result = result.unwrap_err();
-        return Err(result);
-    }
-
-    let result = result.unwrap();
-
-    let response = serde_json::from_value::<BybitOrderRestResponse>(result.body);
-    if response.is_err() {
-        let response = response.unwrap_err();
-        return Err(response.to_string());
-    }
-    let r = response.unwrap();
-
-    let is_maker = if order_type == OrderType::Limit {
-        true
-    } else {
-        false
-    };
-
-    let order = Order {
-        symbol: symbol,
-        create_time: msec_to_microsec(result.time),
-        status: OrderStatus::New,
-        order_id: r.order_id,
-        client_order_id: r.order_link_id,
-        order_side: side,
-        order_type: order_type,
-        order_price: if order_type == OrderType::Market {
-            dec![0.0]
-        } else {
-            price.unwrap()
-        },
-        order_size: size,
-        remain_size: size,
-        transaction_id: "".to_string(),
-        update_time: msec_to_microsec(result.time),
-        execute_price: dec![0.0],
-        execute_size: dec![0.0],
-        quote_vol: dec![0.0],
-        commission: dec![0.0],
-        commission_asset: "".to_string(),
-        is_maker: is_maker,
-        message: "".to_string(),
-        commission_home: dec![0.0],
-        commission_foreign: dec![0.0],
-        home_change: dec![0.0],
-        foreign_change: dec![0.0],
-        free_home_change: dec![0.0],
-        free_foreign_change: dec![0.0],
-        lock_home_change: dec![0.0],
-        lock_foreign_change: dec![0.0],
-        log_id: 0,
-    };
-
-    return Ok(order);
-}
-
-
-pub fn cancel_order(
-    server: &BybitServerConfig,
-    config: &MarketConfig,
-    order_id: &str,
-) -> Result<Order, String> {
-    let message = CancelOrderMessage {
-        category: config.trade_category.clone(),
-        symbol: config.trade_symbol.clone(),
-        order_id: order_id.to_string(),
-    };
-
-    let message_json = serde_json::to_string(&message).unwrap();
-    let path = "/v5/order/cancel";
-    let result = bybit_post_sign(&server, path, &message_json);
-
-    if result.is_err() {
-        let result = result.unwrap_err();
-        return Err(result);
-    }
-
-    let result = result.unwrap();
-
-    let response = serde_json::from_value::<BybitOrderRestResponse>(result.body);
-    if response.is_err() {
-        let response = response.unwrap_err();
-        return Err(response.to_string());
-    }
-    let r = response.unwrap();
-
-    let order = Order {
-        symbol: "".to_string(),
-        create_time: msec_to_microsec(result.time),
-        status: OrderStatus::Canceled,
-        order_id: r.order_id,
-        client_order_id: r.order_link_id,
-        order_side: OrderSide::Unknown,
-        order_type: OrderType::Limit,
-        order_price: dec![0.0],
-        order_size: dec![0.0],
-        remain_size: dec![0.0],
-        transaction_id: "".to_string(),
-        update_time: msec_to_microsec(result.time),
-        execute_price: dec![0.0],
-        execute_size: dec![0.0],
-        quote_vol: dec![0.0],
-        commission: dec![0.0],
-        commission_asset: "".to_string(),
-        is_maker: true,
-        message: "".to_string(),
-        commission_home: dec![0.0],
-        commission_foreign: dec![0.0],
-        home_change: dec![0.0],
-        foreign_change: dec![0.0],
-        free_home_change: dec![0.0],
-        free_foreign_change: dec![0.0],
-        lock_home_change: dec![0.0],
-        lock_foreign_change: dec![0.0],
-        log_id: 0,
-    };
-
-    return Ok(order);
-}
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-struct CancelAllMessage {
-    category: String,
-    symbol: String,
-}
-
-pub fn cancell_all_orders(
-    server: &BybitServerConfig,
-    config: &MarketConfig,
-) -> Result<Vec<Order>, String> {
-    let message = CancelAllMessage {
-        category: config.trade_category.clone(),
-        symbol: config.trade_symbol.clone(),
-    };
-
-    let message_json = serde_json::to_string(&message).unwrap();
-    let path = "/v5/order/cancel-all";
-    let result = bybit_post_sign(&server, path, &message_json);
-
-    if result.is_err() {
-        let result = result.unwrap_err();
-        return Err(result);
-    }
-
-    let result = result.unwrap();
-
-    let response = serde_json::from_value::<BybitMultiOrderRestResponse>(result.body);
-
-    if response.is_err() {
-        let response = response.unwrap_err();
-        return Err(response.to_string());
-    }
-
-    let response = response.unwrap();
-
-    let mut orders: Vec<Order> = vec![];
-
-    for r in response.list {
-        println!("r={:?}", r);
-
-        let order = Order {
-            symbol: "".to_string(),
-            create_time: msec_to_microsec(result.time),
-            status: OrderStatus::Canceled,
-            order_id: r.order_id,
-            client_order_id: r.order_link_id,
-            order_side: OrderSide::Unknown,
-            order_type: OrderType::Limit,
-            order_price: dec![0.0],
-            order_size: dec![0.0],
-            remain_size: dec![0.0],
-            transaction_id: "".to_string(),
-            update_time: msec_to_microsec(result.time),
-            execute_price: dec![0.0],
-            execute_size: dec![0.0],
-            quote_vol: dec![0.0],
-            commission: dec![0.0],
-            commission_asset: "".to_string(),
-            is_maker: true,
-            message: "".to_string(),
-            commission_home: dec![0.0],
-            commission_foreign: dec![0.0],
-            home_change: dec![0.0],
-            foreign_change: dec![0.0],
-            free_home_change: dec![0.0],
-            free_foreign_change: dec![0.0],
-            lock_home_change: dec![0.0],
-            lock_foreign_change: dec![0.0],
-            log_id: 0,
-        };
-        orders.push(order);
-    }
-
-    return Ok(orders);
-}
-
-pub fn get_balance(server: &str, config: &MarketConfig) -> Result<BybitAccountInformation, String> {
-    return Err("Not implemented".to_string());
-}
-
-pub fn order_status(server: &str, config: &MarketConfig) -> Result<Vec<BybitOrderStatus>, String> {
-    return Err("Not implemented".to_string());
-}
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-struct OpenOrderRequest {
-    category: String,
-    symbol: String,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    #[serde(rename = "orderId")]
-    order_id: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    cursor: Option<String>,
-}
-
-pub fn open_orders(
-    server: &BybitServerConfig,
-    config: &MarketConfig,
-) -> Result<Vec<Order>, String> {
-    let query_string = format!(
-        "category={}&symbol={}&limit=50",
-        config.trade_category, config.trade_symbol
-    );
-
-    let path = "/v5/order/realtime";
-
-    let result = bybit_get_sign(&server, path, &query_string);
-
-    if result.is_err() {
-        let result = result.unwrap_err();
-        return Err(result);
-    }
-
-    let result = result.unwrap();
-
-    if result.body.is_null() {
-        return Ok(vec![]);
-    }
-
-    let orders: Vec<Order> = vec![];
-
-    println!("result.body={:?}", result.body);
-
-    let response = serde_json::from_value::<BybitMultiOrderStatus>(result.body);
-
-    if response.is_err() {
-        let response = response.unwrap_err();
-        return Err(response.to_string());
-    }
-
-    let orders: Vec<Order> = response.unwrap().into();
-
-    Ok(orders)
-}
-
-pub fn trade_list(server: &str, config: &MarketConfig) -> Result<Vec<BybitOrderStatus>, String> {
-    return Err("Not implemented".to_string());
-}
-
-*/
 
 #[cfg(test)]
 #[allow(non_snake_case)]
@@ -940,40 +626,23 @@ pub fn trade_list(server: &str, config: &MarketConfig) -> Result<Vec<BybitOrderS
 mod bybit_rest_test {
     use super::*;
     use std::{any, thread::sleep, time::Duration};
-    use rbot_lib::common::init_log;
+    use rbot_lib::common::{init_log};
 
     use crate::config::BybitConfig;
     use pyo3::ffi::Py_Initialize;
     use rbot_lib::common::{init_debug_log, time_string, HHMM};
     use rust_decimal_macros::dec;
 
-    #[tokio::test]
-    async fn test_rest_get() -> anyhow::Result<()> {
-        let server_config = BybitServerConfig::new(false);
-        let config = BybitConfig::BTCUSDT();
-
-        BybitRestApi::rest_get(&server_config, "/v2/public/time", "").await?;
-        Ok(())
-    }
 
     #[tokio::test]
-    async fn test_get_sign() {
+    async fn get_board_snapshot_test()  -> anyhow::Result<()> {
         let server_config = BybitServerConfig::new(false);
         let config = BybitConfig::BTCUSDT();
 
-        let r = BybitRestApi::get_sign(&server_config, "/v2/public/time", "").await;
-        assert!(r.is_ok());
+        let r = BybitRestApi::get_board_snapshot(&server_config, &config).await?;
         println!("{:?}", r);
-    }
-
-    #[tokio::test]
-    async fn get_board_snapshot_test_trait() {
-        let server_config = BybitServerConfig::new(false);
-        let config = BybitConfig::BTCUSDT();
 
-        let r = BybitRestApi::get_board_snapshot(&server_config, &config).await;
-        assert!(r.is_ok());
-        println!("{:?}", r);
+        Ok(())
     }
 
     #[tokio::test]
@@ -1010,7 +679,7 @@ mod bybit_rest_test {
 
     #[tokio::test]
     async fn test_new_limit_order() {
-        let server_config = BybitServerConfig::new(true);
+        let server_config = BybitServerConfig::new(false);
         let config = BybitConfig::BTCUSDT();
 
         let r = BybitRestApi::new_order(
@@ -1030,7 +699,7 @@ mod bybit_rest_test {
 
     #[tokio::test]
     async fn test_new_market_order() {
-        let server_config = BybitServerConfig::new(true);
+        let server_config = BybitServerConfig::new(false);
         let config = BybitConfig::BTCUSDT();
 
         let r = BybitRestApi::new_order(
@@ -1050,7 +719,7 @@ mod bybit_rest_test {
 
     #[tokio::test]
     async fn test_cancel_order() {
-        let server_config = BybitServerConfig::new(true);
+        let server_config = BybitServerConfig::new(false);
         let config = BybitConfig::BTCUSDT();
 
         let r = BybitRestApi::new_order(
@@ -1077,7 +746,7 @@ mod bybit_rest_test {
     async fn test_open_orders() {
         init_debug_log();
 
-        let server_config = BybitServerConfig::new(true);
+        let server_config = BybitServerConfig::new(false);
         let config = BybitConfig::BTCUSDT();
 
         let r = BybitRestApi::open_orders(&server_config, &config).await;
@@ -1085,4 +754,15 @@ mod bybit_rest_test {
         assert!(r.is_ok());
         println!("{:?}", r);
     }
+
+    #[tokio::test]
+    async fn test_get_account() {
+        init_debug_log();
+        let server_config = BybitServerConfig::new(false);
+        let config = BybitConfig::BTCUSDT();
+
+        let r = BybitRestApi::get_account(&server_config).await;
+        println!("{:?}", r);
+        assert!(r.is_ok());
+    }
 }
diff --git a/exchanges/bybit/src/ws.rs b/exchanges/bybit/src/ws.rs
index 11b8ee1c..246abeb6 100644
--- a/exchanges/bybit/src/ws.rs
+++ b/exchanges/bybit/src/ws.rs
@@ -1,18 +1,25 @@
 #![allow(unused)]
+use std::pin::Pin;
+
 use async_stream::stream;
 use futures::Stream;
 use futures::StreamExt;
-use rbot_lib::common::AccountStatus;
+use rbot_lib::common::AccountCoins;
+use rbot_lib::common::AccountPair;
+use rbot_lib::common::ControlMessage;
 use rbot_lib::common::MarketMessage;
 use rbot_lib::common::Order;
+use rbot_lib::common::MARKET_HUB;
+use rbot_lib::net::BroadcastMessage;
 use serde_derive::Deserialize;
 use serde_derive::Serialize;
 
 use rbot_lib::common::{hmac_sign, MarketConfig, MultiMarketMessage, ServerConfig, NOW};
 
 use rbot_lib::net::{AutoConnectClient, WsOpMessage};
+use tokio::task::JoinHandle;
 
-
+use crate::message::convert_coin_to_account_status;
 use crate::message::merge_order_and_execution;
 use crate::message::BybitExecution;
 use crate::message::BybitOrderStatus;
@@ -33,7 +40,6 @@ pub struct BybitWsOpMessage {
     pub id: i64,
 }
 
-
 impl WsOpMessage for BybitWsOpMessage {
     fn new() -> Self {
         BybitWsOpMessage {
@@ -71,9 +77,9 @@ impl WsOpMessage for BybitWsOpMessage {
     }
 }
 
-#[derive(Debug)]
 pub struct BybitPublicWsClient {
     ws: AutoConnectClient<BybitServerConfig, BybitWsOpMessage>,
+    handler: Option<tokio::task::JoinHandle<()>>,
 }
 
 impl BybitPublicWsClient {
@@ -99,13 +105,101 @@ impl BybitPublicWsClient {
 
         public_ws.subscribe(&config.public_subscribe_channel).await;
 
-        Self { ws: public_ws }
+        Self {
+            ws: public_ws,
+            handler: None,
+        }
     }
 
     pub async fn connect(&mut self) {
         self.ws.connect().await
     }
 
+    /*
+        pub async fn start_thread(&mut self) -> anyhow::Result<()> {
+            let exchange_name = self.ws.get_server().exchange_name.clone();
+            let category = self.ws.get_config().trade_category.clone();
+            let symbol = self.ws.get_config().trade_symbol.clone();
+
+            let hub_channel = MARKET_HUB.open_channel();
+
+            let mut stream = self.ws.open_stream().await;
+            //let mut s = tokio::pin!(stream);
+
+            //let mut s = Box::pin(self.ws.open_stream().await);
+            let mut s = Box::pin(stream);
+    //        let mut s = Box::pin(stream) as Pin<&mut dyn Stream<Item = Result<Bytes, WsError>> + Send>;
+
+            let handle = tokio::task::spawn_local(async move {
+
+                while let Some(message) = s.next().await {
+                    match message {
+                        Ok(m) => match Self::parse_message(m) {
+                            Err(e) => {
+                                println!("Parse Error: {:?}", e);
+                                continue;
+                            }
+                            Ok(m) => {
+                                let market_message = Self::convert_ws_message(m);
+                                if market_message.is_err() {
+                                    log::error!("Convert Error: {:?}", market_message);
+                                    continue;
+                                }
+
+                                let market_message = market_message.unwrap();
+                                hub_channel.send(HubMessage {
+                                    exchange: exchange_name.clone(),
+                                    category: category.clone(),
+                                    symbol: symbol.clone(),
+                                    msg: market_message,
+                                });
+                            }
+                        },
+                        Err(e) => {
+                            println!("Receive Error: {:?}", e);
+                        }
+                    }
+                }
+            });
+
+            self.handler = Some(handle);
+
+            Ok(())
+        }
+    */
+    /*
+        pub async fn start_thread<'a>(&'a mut self) -> anyhow::Result<JoinHandle<()>> {
+            let exchange_name = self.ws.get_server().exchange_name.clone();
+            let category = self.ws.get_config().trade_category.clone();
+            let symbol = self.ws.get_config().trade_symbol.clone();
+            let hub_channel = MARKET_HUB.open_channel();
+
+            let mut stream = self.open_stream().await;
+
+            let handle = tokio::task::spawn(async move {
+                let mut s = Box::pin(stream);
+                while let Some(message) = s.next().await {
+                    match message {
+                        Ok(m) => {
+                            let market_message = m;
+                            hub_channel.send(HubMessage {
+                                exchange: exchange_name.clone(),
+                                category: category.clone(),
+                                symbol: symbol.clone(),
+                                msg: market_message,
+                            });
+                        }
+                        Err(e) => {
+                            println!("Receive Error: {:?}", e);
+                        }
+                    }
+                }
+            });
+
+            Ok(handle)
+        }
+    */
+
     pub async fn open_stream<'a>(
         &'a mut self,
     ) -> impl Stream<Item = Result<MultiMarketMessage, String>> + 'a {
@@ -115,21 +209,22 @@ impl BybitPublicWsClient {
             while let Some(message) = s.next().await {
                 match message {
                     Ok(m) => {
-                        let m = Self::parse_message(m);
-
-                        match m {
+                        match Self::parse_message(m) {
                             Err(e) => {
                                 println!("Parse Error: {:?}", e);
                                 continue;
                             }
                             Ok(m) => {
-                                match m {
-                                    BybitPublicWsMessage::Status(s) => {
-                                        log::debug!("status message: {:?}", s);
+                                let market_message = Self::convert_ws_message(m);
+
+                                match market_message
+                                {
+                                    Err(e) => {
+                                        println!("Convert Error: {:?}", e);
+                                        continue;
                                     }
-                                    _ => {
-                                        let m = Self::convert_ws_message(m);
-                                        yield m;
+                                    Ok(m) => {
+                                        yield Ok(m);
                                     }
                                 }
                             }
@@ -164,10 +259,12 @@ pub struct BybitPrivateWsClient {
 }
 
 impl BybitPrivateWsClient {
-    pub async fn new(server: &BybitServerConfig, config: &MarketConfig) -> Self {
+    pub async fn new(server: &BybitServerConfig) -> Self {
+        let dummy_config = MarketConfig::new("dummy", "dummy", "dummy", 0, 0, 0);
+
         let mut private_ws = AutoConnectClient::new(
             server,
-            config,
+            &dummy_config,
             &server.private_ws,
             PING_INTERVAL_SEC,
             SWITCH_INTERVAL_SEC,
@@ -184,15 +281,12 @@ impl BybitPrivateWsClient {
             ])
             .await;
 
-        Self {
-            ws: private_ws,
-        }
+        Self { ws: private_ws }
     }
 
-    
     fn make_auth_message(server: &BybitServerConfig) -> String {
-        let api_key = server.get_api_key();
-        let secret_key = server.get_api_secret();
+        let api_key = server.get_api_key().extract();
+        let secret_key = server.get_api_secret().extract();
         let time_stamp = (NOW() / 1_000) + 1_000 * 10; // 10 seconds in the future
         let sign = hmac_sign(&secret_key, &format!("GET/realtime{}", time_stamp));
 
@@ -212,7 +306,6 @@ impl BybitPrivateWsClient {
     pub async fn open_stream<'a>(
         &'a mut self,
     ) -> impl Stream<Item = Result<MultiMarketMessage, String>> + 'a {
-        let market_config = self.ws.get_config();
         let mut s = Box::pin(self.ws.open_stream().await);
 
         stream! {
@@ -233,7 +326,9 @@ impl BybitPrivateWsClient {
                             Ok(m) => {
                                 match m {
                                     BybitUserWsMessage::status(s) => {
-                                        log::debug!("status message: {:?}", s);
+                                        let control: ControlMessage = s.into();
+                                        log::debug!("status message: {:?}", control);
+                                        yield Ok(MultiMarketMessage::Control(control));
                                     }
                                     BybitUserWsMessage::pong(p) => {
                                         log::debug!("pong message: {:?}", p);
@@ -253,11 +348,7 @@ impl BybitPrivateWsClient {
                                                     let order = merge_order_and_execution(&last_orders, &last_executions);
                                                     last_orders.clear();
                                                     last_executions.clear();
-                                                    
-                                                    for o in order.iter() {
-                                                        let mut o = o.clone();
-                                                        o.update_balance(&market_config);
-                                                    }
+
                                                     yield Ok(MultiMarketMessage::Order(order));
                                                 }
                                             }
@@ -275,10 +366,6 @@ impl BybitPrivateWsClient {
                                                     last_orders.clear();
                                                     last_executions.clear();
 
-                                                    for o in order.iter() {
-                                                        let mut o = o.clone();
-                                                        o.update_balance(&market_config);
-                                                    }
                                                     yield Ok(MultiMarketMessage::Order(order));
                                                 }
                                             }
@@ -287,12 +374,12 @@ impl BybitPrivateWsClient {
                                                 creationTime,
                                                 data,
                                             } => {
-                                                let mut account_status: Vec<AccountStatus> = vec![];
-                                                
-                                                for account in data.iter() {
-                                                    let a: AccountStatus = account.into();
+                                                let mut coins = AccountCoins::new();
+                                                for account in data {
+                                                    let mut account_coins: AccountCoins = account.into();
+                                                    coins.append(account_coins);
                                                 }
-                                                yield Ok(MultiMarketMessage::Account(account_status));
+                                                yield Ok(MultiMarketMessage::Account(coins));
                                             }
                                         }
                                     }
@@ -326,7 +413,6 @@ impl BybitPrivateWsClient {
     ) -> Vec<Order> {
         merge_order_and_execution(orders, executions)
     }
-    
 }
 
 #[cfg(test)]
@@ -367,7 +453,7 @@ mod bybit_ws_test {
         let server = BybitServerConfig::new(true);
         let config = BybitConfig::BTCUSDT();
 
-        let mut ws = BybitPrivateWsClient::new(&server, &config).await;
+        let mut ws = BybitPrivateWsClient::new(&server).await;
 
         ws.connect().await;
 
@@ -383,123 +469,3 @@ mod bybit_ws_test {
         }
     }
 }
-
-/*
-    pub fn connect(&mut self) {
-        BLOCK_ON(async {
-        BLOCK_ON(async {
-            self.ws.connect(|message|{
-                let m = serde_json::from_str::<BybitWsMessage>(&message);
-
-                if m.is_err() {
-                    log::warn!("Error in serde_json::from_str: {:?}", message);
-                    println!("ERR: {:?}", message);
-                }
-
-                let m = m.unwrap();
-
-                return m.into();
-            }).await
-        })
-    }
-
-    pub fn open_channel(&mut self) -> Receiver<MultiMarketMessage> {
-        BLOCK_ON(async {
-            self.ws.open_channel().await
-        })
-    }
-}
-*/
-
-/*
-pub async fn listen_userdata_stream<F>(
-    config: &BybitServerConfig,
-    mut f: F,
-) -> tokio::task::JoinHandle<()>
-where
-    F: FnMut(BybitUserStreamMessage) + Send + 'static,
-{
-    let url = config.private_ws.clone();
-
-    let mut websocket: AutoConnectClient<BybitServerConfig, BybitWsOpMessage> =
-        AutoConnectClient::new(
-
-            config,
-            &url,
-            // Arc::new(RwLock::new(message)),
-            PING_INTERVAL_SEC,
-            SWITCH_INTERVAL_SEC,
-            SYNC_RECORDS,
-            Some(make_auth_message),
-        );
-
-    websocket
-        .subscribe(&vec![
-            "execution".to_string(),
-            "order".to_string(),
-            "wallet".to_string(),
-        ])
-        .await;
-
-    websocket.connect().await;
-
-    let handle = tokio::task::spawn(async move {
-        loop {
-            let msg = websocket.receive_text().await;
-            if msg.is_err() {
-                log::warn!("Error in websocket.receive_message: {:?}", msg);
-                continue;
-            }
-            let msg = msg.unwrap();
-            log::debug!("raw msg: {}", msg);
-            let msg = serde_json::from_str::<BybitUserStreamMessage>(msg.as_str());
-            if msg.is_err() {
-                log::warn!("Error in serde_json::from_str: {:?}", msg);
-                continue;
-            }
-            let msg = msg.unwrap();
-            f(msg);
-        }
-    });
-
-    handle
-}
-
-*/
-
-/*
-#[cfg(test)]
-mod test_ws {
-    use crate::common::init_debug_log;
-
-    #[test]
-    fn test_make_auth_message() {
-        use crate::exchange::bybit::config::BybitServerConfig;
-        let config = BybitServerConfig::new(false);
-        let msg = super::make_auth_message(&config);
-        println!("{}", msg);
-    }
-
-    #[test]
-    fn test_connect() {
-        use crate::exchange::bybit::config::BybitServerConfig;
-        use crate::exchange::bybit::ws::listen_userdata_stream;
-        let config = BybitServerConfig::new(true);
-
-        init_debug_log();
-
-        listen_userdata_stream(&config, |msg| {
-            println!("{:?}", msg);
-        });
-
-        std::thread::sleep(std::time::Duration::from_secs(5));
-    }
-}
-
-/*
-TODO: reconnet ws when disconnected.
-[2024-01-20T04:55:43Z WARN  rbot::exchange::bybit::ws] Error in websocket.receive_message: Err("Disconnected wss://stream-testnet.bybit.com/v5/private: Trying to work with closed connection")
-[2024-01-20T04:55:43Z ERROR rbot::exchange::ws] Disconnected from server
-*/
-
-*/
diff --git a/rbot/Cargo.toml b/rbot/Cargo.toml
index b0944de9..f2f51902 100644
--- a/rbot/Cargo.toml
+++ b/rbot/Cargo.toml
@@ -3,6 +3,9 @@ name = "rbot"
 version = "0.3.1"
 edition = "2021"
 
+[buildflags]
+rustflags = ["--cfg", "tokio_unstable"]
+
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 [lib]
 name = "rbot"
@@ -16,6 +19,8 @@ rbot_session = {path="../rbot_session"}
 rbot_market = {path="../rbot_market"}
 once_cell = {workspace = true}
 
+console-subscriber = {version="0.2.0"}
+
 
 # https://pyo3.rs/v0.13.2/faq
 [dependencies.pyo3]
diff --git a/rbot/src/lib.rs b/rbot/src/lib.rs
index 9f1d30a1..604c7b77 100644
--- a/rbot/src/lib.rs
+++ b/rbot/src/lib.rs
@@ -4,12 +4,9 @@
 
 use pyo3::{pymodule, types::PyModule, wrap_pyfunction, PyResult, Python};
 use rbot_lib::common::{
-    init_debug_log, init_log,
-    Order, OrderSide,
-    time_string,
-    NOW, DAYS_BEFORE, DAYS, HHMM, MIN, SEC, FLOOR_SEC,
-    MarketConfig, OrderStatus, AccountStatus,
-    OrderType, Trade, BoardItem,
+    get_orderbook_list,
+    init_debug_log, init_log, time_string, 
+    AccountPair, BoardItem, MarketConfig, Order, OrderSide, OrderStatus, OrderType, Trade, DAYS, DAYS_BEFORE, FLOOR_SEC, HHMM, MIN, NOW, SEC
 };
 
 
@@ -17,6 +14,8 @@ use rbot_session::{Logger, Session, Runner, ExecuteMode};
 
 use bybit::{Bybit, BybitConfig};
 
+use console_subscriber;
+
 /*
 use exchange::BoardItem;
 use exchange::bybit::{BybitMarket, Bybit};
@@ -39,11 +38,15 @@ use session::Runner;
 /// A Python module implemented in Rust.
 #[pymodule]
 fn rbot(_py: Python, m: &PyModule) -> PyResult<()> {
+    // console_subscriber::init();
+
     m.add("__version__", env!("CARGO_PKG_VERSION"))?;
 
     m.add_function(wrap_pyfunction!(init_log, m)?)?;
     m.add_function(wrap_pyfunction!(init_debug_log, m)?)?;
 
+    m.add_function(wrap_pyfunction!(get_orderbook_list, m)?)?;
+
     // time util
     m.add_function(wrap_pyfunction!(time_string, m)?)?;
     m.add_function(wrap_pyfunction!(NOW, m)?)?;
@@ -58,7 +61,7 @@ fn rbot(_py: Python, m: &PyModule) -> PyResult<()> {
     // classes
     m.add_class::<MarketConfig>()?;
     m.add_class::<OrderStatus>()?;
-    m.add_class::<AccountStatus>()?;
+    m.add_class::<AccountPair>()?;
     
     m.add_class::<Logger>()?;
 
diff --git a/rbot_blockon/Cargo.toml b/rbot_blockon/Cargo.toml
new file mode 100644
index 00000000..ff20518c
--- /dev/null
+++ b/rbot_blockon/Cargo.toml
@@ -0,0 +1,12 @@
+[package]
+name = "rbot_blockon"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+log= {workspace=true}
+features = {workspace=true}
+tokio = { workspace=true}
+once_cell = {workspace=true}
diff --git a/rbot_blockon/src/lib.rs b/rbot_blockon/src/lib.rs
new file mode 100644
index 00000000..20a99f99
--- /dev/null
+++ b/rbot_blockon/src/lib.rs
@@ -0,0 +1,35 @@
+
+use std::future::Future;
+use once_cell::sync::Lazy;
+use tokio::time::{timeout, Duration};
+
+
+pub static RUNTIME: Lazy<tokio::runtime::Runtime> =
+    Lazy::new(|| tokio::runtime::Runtime::new().unwrap());
+
+
+#[allow(non_snake_case)]
+pub fn BLOCK_ON<F: Future>(f: F) -> F::Output {
+    let result = RUNTIME.block_on(f);
+    result
+}
+
+#[allow(non_snake_case)]
+pub fn BLOCK_ON_TIMEOUT<F>(timeout_sec: u64, f: F) -> F::Output
+where
+    F: Future,
+{
+    log::debug!("BLOCK_ON_TIMEOUT: (timeout={})", timeout_sec);
+
+    let result = RUNTIME.block_on(async {
+        let duration = Duration::from_secs(timeout_sec);
+
+        match timeout(duration, f).await {
+            Ok(result) => result,
+            Err(_) => panic!("Timeout"),
+        }        
+    });
+
+    result
+}
+
diff --git a/rbot_lib/src/common/ch.rs b/rbot_lib/src/common/ch.rs
index 857b64dc..aa4d59cc 100644
--- a/rbot_lib/src/common/ch.rs
+++ b/rbot_lib/src/common/ch.rs
@@ -10,7 +10,8 @@ use serde_derive::Serialize;
 
 use super::order::Order;
 use super::order::Trade;
-use super::AccountStatus;
+use super::AccountCoins;
+use super::AccountPair;
 use super::OrderBookRaw;
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -24,7 +25,7 @@ pub struct ControlMessage {
 pub enum MarketMessage {
     Trade(Trade),
     Order(Order),
-    Account(AccountStatus),
+    Account(AccountCoins),
     Orderbook(OrderBookRaw),
     Control(ControlMessage),
     Message(String),
@@ -40,7 +41,7 @@ impl MarketMessage {
         MarketMessage::Order(order)
     }
 
-    pub fn from_account(account: AccountStatus) -> Self {
+    pub fn from_account(account: AccountCoins) -> Self {
         MarketMessage::Account(account)
     }
 
@@ -71,7 +72,7 @@ impl MarketMessage {
 pub enum MultiMarketMessage {
     Trade(Vec<Trade>),
     Order(Vec<Order>),
-    Account(Vec<AccountStatus>),
+    Account(AccountCoins),
     Orderbook(OrderBookRaw),
     Message(String),
     Control(ControlMessage),
diff --git a/rbot_lib/src/common/config.rs b/rbot_lib/src/common/config.rs
index fa3f9910..fce429f3 100644
--- a/rbot_lib/src/common/config.rs
+++ b/rbot_lib/src/common/config.rs
@@ -127,13 +127,77 @@ impl MarketConfig {
             public_subscribe_channel: vec![],
         }
     }
+}
 
+impl Default for MarketConfig {
+    fn default() -> Self {
+        Self {
+            trade_category: "DUMMY".to_string(),
+            trade_symbol: "DUMMY".to_string(),
+            price_unit: dec![0.01],
+            price_scale: 2,
+            size_unit: dec![1000.0],
+            size_scale: 0,
+            maker_fee: dec![0.0], // dec![0.00_015],  // 0.015%
+            taker_fee: dec![0.0], // dec![0.00_015],  // 0.015%
+            price_type: PriceType::Foreign,
+            fee_type: FeeType::Home,
+            home_currency: "DUMMY".to_string(),
+            foreign_currency: "DUMMY".to_string(),
+            market_order_price_slip: dec![0.0],
+            board_depth: 1000,
+            public_subscribe_channel: vec![],
+        }
+    }
 
+}
 
+/*
+pub struct ServerConfigStruct {
+    exchange_name: String, 
+    trade_category: String,
+    home_currency: String,
+    foreign_currency: String,
+    price_scale: u32,
+    size_scale: u32,
+    board_depth: u32,
 }
 
+impl ServerConfig for ServerConfigStruct {
+    fn get_exchange_name(&self) -> String {
+        self.exchange_name.clone()
+    }
+
+    fn get_historical_web_base(&self) -> String {
+        todo!()
+    }
 
-pub trait ServerConfig {
+    fn get_public_ws_server(&self) -> String {
+        todo!()
+    }
+
+    fn get_user_ws_server(&self) -> String {
+        todo!()
+    }
+
+    fn get_rest_server(&self) -> String {
+        todo!()
+    }
+
+    fn get_api_key(&self) -> SecretString {
+        todo!()
+    }
+
+    fn get_api_secret(&self) -> SecretString {
+        todo!()
+    }
+}
+*/
+
+
+
+pub trait ServerConfig : Send + Sync {
+    fn get_exchange_name(&self) -> String;
     fn get_historical_web_base(&self) -> String;
     fn get_public_ws_server(&self) -> String;
     fn get_user_ws_server(&self) -> String;
@@ -141,3 +205,4 @@ pub trait ServerConfig {
     fn get_api_key(&self) -> SecretString;
     fn get_api_secret(&self) -> SecretString;
 }
+
diff --git a/rbot_lib/src/common/hub.rs b/rbot_lib/src/common/hub.rs
index 714459c8..ac411e0c 100644
--- a/rbot_lib/src/common/hub.rs
+++ b/rbot_lib/src/common/hub.rs
@@ -1,11 +1,20 @@
+use futures::Stream;
+use tokio::runtime::Runtime;
 use tokio::sync::broadcast;
 use tokio::sync::broadcast::Receiver;
 use tokio::sync::broadcast::Sender;
+use tokio::task::JoinHandle;
 
 use crate::net::BroadcastMessage;
 
+use async_stream::stream;
 use once_cell::sync::Lazy;
 
+use super::MarketMessage;
+
+use futures::StreamExt;
+use tokio::task::spawn;
+
 pub struct MarketHub {
     tx: Sender<BroadcastMessage>,
     _rx: Receiver<BroadcastMessage>,
@@ -13,24 +22,132 @@ pub struct MarketHub {
 
 const CHANNEL_SIZE: usize = 1024;
 
-pub static MARKET_HUB: Lazy<MarketHub> = Lazy::new(|| MarketHub::open());
+pub static MARKET_HUB: Lazy<MarketHub> = Lazy::new(|| MarketHub::new());
+
+pub fn stream_receiver(
+    stream: impl Stream<Item = anyhow::Result<MarketMessage>> + Send + 'static,
+) -> (crossbeam_channel::Receiver<MarketMessage>, JoinHandle<()>) {
+    let (tx, rx) = crossbeam_channel::unbounded();
+
+    let handle = tokio::spawn(async move {
+        let mut stream = Box::pin(stream);
+
+        while let Some(message) = stream.next().await {
+            match message {
+                Ok(msg) => {
+                    let r = tx.send(msg);
+                    if r.is_err() {
+                        log::error!("send message error: {:?}", r);
+                    }
+                }
+                Err(e) => {
+                    log::error!("receive message error: {:?}", e);
+                }
+            }
+        }
+    });
+
+    (rx, handle)
+}
 
 impl MarketHub {
-    pub fn open() -> Self {
+    pub fn new() -> Self {
         let (tx, _rx) = broadcast::channel(CHANNEL_SIZE);
         Self { tx, _rx }
     }
 
-    pub fn subscribe(&self) -> Receiver<BroadcastMessage> {
-        return self.tx.subscribe();
+    pub fn subscribe(&self,
+        exchange: &str,
+        category: &str,
+        symbol: &str,
+        agent_id: &str,
+    ) -> anyhow::Result<crossbeam_channel::Receiver<MarketMessage>> {
+        let exchange = exchange.to_string();
+        let category = category.to_string();
+        let symbol = symbol.to_string();
+        let agent_id = agent_id.to_string();
+
+        let (tx, rx) = crossbeam_channel::unbounded();
+        let mut ch = self.tx.subscribe();
+
+        std::thread::spawn(move ||{
+            let runtime = Runtime::new().unwrap();
+
+            runtime.block_on(async move {
+                loop {
+                    let msg = ch.recv().await;
+    
+                    if msg.is_err() {
+                        break;
+                    }
+    
+                    let msg = msg.unwrap();
+    
+                    if msg.filter(&exchange, &category, &symbol) {
+                        let market_message = msg.msg.clone();
+    
+                        match market_message {
+                            MarketMessage::Order(ref order) => {
+                                if order.is_my_order(&agent_id) {
+                                    let r = tx.send(market_message.clone());
+                                    if r.is_err() {
+                                        log::error!("open_channel: {}/{:?}", r.err().unwrap(), msg);
+                                        break;
+                                    }
+                                }
+                            }
+                            _ => {
+                                let r = tx.send(market_message.clone());
+                                if r.is_err() {
+                                    log::error!("open_channel: {}/{:?}", r.err().unwrap(), msg);
+                                    break;
+                                }
+                            }
+                        }
+                    }
+                }
+            });
+        });
+
+        Ok(rx)
     }
 
+
+
+    pub async fn subscribe_stream<'a>(
+        &self,
+        exchange: &'a str,
+        category: &'a str,
+        symbol: &'a str,
+        agent_id: &'a str,
+    ) -> impl Stream<Item = anyhow::Result<MarketMessage>> + 'a {
+        let mut ch = self.tx.subscribe();
+
+        stream! {
+            loop {
+                let msg = ch.recv().await?;
+
+                if msg.filter(exchange, category, symbol) {
+                    match msg.msg {
+                        MarketMessage::Order(ref order) => {
+                            if order.is_my_order(agent_id) {
+                                yield Ok(msg.msg);
+                            }
+                        }
+                        _ => {
+                            yield Ok(msg.msg);
+                        }
+                    }
+                }
+            }
+        }
+    }
     pub fn publish(&self, message: BroadcastMessage) -> anyhow::Result<()> {
         self.tx.send(message)?;
         Ok(())
     }
 
-    pub fn clone_channel(&self) -> Sender<BroadcastMessage> {
+    pub fn open_channel(&self) -> Sender<BroadcastMessage> {
         self.tx.clone()
     }
 }
@@ -42,22 +159,55 @@ mod test_market_hub {
     use crate::common::init_debug_log;
     use crate::common::MarketMessage;
     use crate::common::Trade;
+    use futures::StreamExt;
     use tokio::sync::broadcast::Receiver;
     use tokio::sync::broadcast::Sender;
+    use tokio::time::sleep;
     use tokio::time::Duration;
 
     #[tokio::test]
-    async fn test_market_hub() {
+    async fn test_receive_channel() {
         init_debug_log();
-        let tx = MARKET_HUB.clone_channel();
+        let tx = MARKET_HUB.open_channel();
 
-        for i in 0..CHANNEL_SIZE * 2 {
+        let rx = MARKET_HUB.subscribe_stream("a", "b", "c", "").await;
+        let mut rx = Box::pin(rx);
+
+        for i in 0..CHANNEL_SIZE {
+            let msg = BroadcastMessage {
+                exchange: "a".to_string(),
+                category: "b".to_string(),
+                symbol: "c".to_string(),
+                msg: MarketMessage::make_message(&format!("message {}", i)),
+            };
+            let r = tx.send(msg.clone());
+            if r.is_err() {
+                println!("error: {:?}", r);
+            }
+        }
+
+        log::debug!("send message done");
 
-            let msg = BroadcastMessage{
+        for _i in 0..10 {
+            if let Some(r) = rx.next().await {
+                println!("ch: m.essage: {:?}", r);
+            }
+        }
+    }
+
+    #[test]
+    fn test_market_hub() {
+        init_debug_log();
+        let tx = MARKET_HUB.open_channel();
+        let mut rx = MARKET_HUB.subscribe("a", "b", "c", "").unwrap();
+        let mut rx2 = MARKET_HUB.subscribe("a", "b", "c", "").unwrap();
+
+        for i in 0..CHANNEL_SIZE * 2 {
+            let msg = BroadcastMessage {
                 exchange: "a".to_string(),
                 category: "b".to_string(),
                 symbol: "c".to_string(),
-                msg: MarketMessage::make_message(&format!("message {}", i))
+                msg: MarketMessage::make_message(&format!("message {}", i)),
             };
             let r = tx.send(msg.clone());
             if r.is_err() {
@@ -65,15 +215,13 @@ mod test_market_hub {
             }
         }
 
-        let mut rx = MARKET_HUB.subscribe();
-        let mut rx2 = MARKET_HUB.subscribe();        
 
         for i in 10..20 {
-            let msg = BroadcastMessage{
+            let msg = BroadcastMessage {
                 exchange: "a".to_string(),
                 category: "b".to_string(),
                 symbol: "c".to_string(),
-                msg: MarketMessage::make_message(&format!("message {}", i))
+                msg: MarketMessage::make_message(&format!("message {}", i)),
             };
 
             let r = tx.send(msg.clone());
@@ -81,9 +229,23 @@ mod test_market_hub {
                 println!("error: {:?}", r);
             }
         }
-        
+
+        log::debug!("-------rx1--------");
+
+        for _i in 10..20 {
+            let r = rx.recv();
+            if r.is_err() {
+                println!("error: {:?}", r);
+            } else {
+                let r = r.unwrap();
+                println!("message: {:?}", r);
+            }
+        }
+
+        log::debug!("-------rx2--------");
+
         for _i in 10..20 {
-            let r = rx.recv().await;
+            let r = rx2.recv();
             if r.is_err() {
                 println!("error: {:?}", r);
             } else {
@@ -91,7 +253,75 @@ mod test_market_hub {
                 println!("message: {:?}", r);
             }
         }
+    }
+
+    #[tokio::test]
+    async fn test_stream_receiver() {
+        let tx = MARKET_HUB.open_channel();
+
+        let rx = MARKET_HUB.subscribe_stream("a", "b", "c", "").await;
+
+        let (rx2, handle2) = stream_receiver(rx);
+
+        for i in 0..10 {
+            let msg = BroadcastMessage {
+                exchange: "a".to_string(),
+                category: "b".to_string(),
+                symbol: "c".to_string(),
+                msg: MarketMessage::make_message(&format!("message {}", i)),
+            };
+            let r = tx.send(msg.clone());
+            if r.is_err() {
+                println!("error: {:?}", r);
+            }
+        }
+
+        for _i in 0..10 {
+            let r = rx2.recv().unwrap();
+            println!("ch: message: {:?}", r);
+        }
+
+        handle2.await.unwrap();
+
+        sleep(Duration::from_secs(5));
+    }
+
+    /*
+    #[tokio::test]
+
+
+    async fn test_receiver() {
+
+        init_debug_log();
+        let tx = MARKET_HUB.open_channel();
+        let mut rx2 = MARKET_HUB.subscribe();
+
+        let handle = spawn(async {
+            let rx = MARKET_HUB.subscribe_stream("a", "b", "c", "").await;
+            let (rx, handle) = rx.unwrap();
+
+            for _i in 0..10 {
+                let r = rx.recv().unwrap();
+                println!("ch: message: {:?}", r);
+            }
+        });
+        let rx = MARKET_HUB.subscribe_stream("a", "b", "c", "").await;
+        let (rx, handle) = rx.unwrap();
+
+        for i in 0..CHANNEL_SIZE {
+            let msg = BroadcastMessage {
+                exchange: "a".to_string(),
+                category: "b".to_string(),
+                symbol: "c".to_string(),
+                msg: MarketMessage::make_message(&format!("message {}", i)),
+            };
+            let r = tx.send(msg.clone());
+            if r.is_err() {
+                println!("error: {:?}", r);
+            }
+        }
 
+        log::debug!("send message done");
         for _i in 10..20 {
             let r = rx2.recv().await;
             if r.is_err() {
@@ -102,5 +332,12 @@ mod test_market_hub {
             }
         }
 
+        for _i in 0..10 {
+            let r = rx.recv().unwrap();
+            println!("ch: message: {:?}", r);
+        }
+
+        sleep(Duration::from_secs(5)).await;
     }
+    */
 }
diff --git a/rbot_lib/src/common/order.rs b/rbot_lib/src/common/order.rs
index 06e8b94d..25f29d3b 100644
--- a/rbot_lib/src/common/order.rs
+++ b/rbot_lib/src/common/order.rs
@@ -380,21 +380,73 @@ impl Default for AccountChange {
     }
 }
 
+
 #[pyclass]
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
-pub struct AccountStatus {
-    pub home: Decimal,
-    pub home_free: Decimal,
-    pub home_locked: Decimal,
-
-    pub foreign: Decimal,
-    pub foreign_free: Decimal,
-    pub foreign_locked: Decimal,
+pub struct Coin {
+    pub symbol: String,
+    pub volume: Decimal,
+    pub free: Decimal,
+    pub locked: Decimal,
 }
 
-impl Default for AccountStatus {
+impl Default for Coin {
     fn default() -> Self {
-        return AccountStatus {
+        return Coin {
+            symbol: "".to_string(),
+            volume: dec![0.0],
+            free: dec![0.0],
+            locked: dec![0.0],
+        };
+    }
+}
+
+#[pyclass]
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct AccountCoins {
+    pub coins: Vec<Coin>,
+}
+
+impl AccountCoins {
+    pub fn new() -> Self {
+        return AccountCoins { coins: Vec::new() };
+    }
+
+    pub fn append(&mut self, mut coins: AccountCoins) {
+        self.coins.append(&mut coins.coins);
+    }
+
+    pub fn extract_pair(&self, config: &MarketConfig) -> AccountPair {
+        let mut home = Coin::default();
+        let mut foreign = Coin::default();
+
+        for coin in self.coins.iter() {
+            if coin.symbol == config.home_currency {
+                home = coin.clone();
+            } else if coin.symbol == config.foreign_currency {
+                foreign = coin.clone();
+            }
+        }
+
+        return AccountPair { home, foreign };
+    }
+}
+
+
+#[pyclass]
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct AccountPair {
+    pub home: Coin,
+    pub foreign: Coin,
+}
+
+impl Default for AccountPair {
+    fn default() -> Self {
+        AccountPair {
+            home: Coin::default(),
+            foreign: Coin::default(),
+        }
+            /*
             home: dec![0.0],
             home_free: dec![0.0],
             home_locked: dec![0.0],
@@ -402,20 +454,20 @@ impl Default for AccountStatus {
             foreign: dec![0.0],
             foreign_free: dec![0.0],
             foreign_locked: dec![0.0],
-        };
+            */
     }
 }
 
 #[pymethods]
-impl AccountStatus {
+impl AccountPair {
     pub fn apply_order(&mut self, order: &Order) {
-        self.foreign += order.foreign_change;
-        self.foreign_free += order.free_foreign_change;
-        self.foreign_locked += order.lock_foreign_change;
+        self.foreign.volume += order.foreign_change;
+        self.foreign.free += order.free_foreign_change;
+        self.foreign.locked += order.lock_foreign_change;
 
-        self.home += order.home_change;
-        self.home_free += order.free_home_change;
-        self.home_locked += order.lock_home_change;
+        self.home.volume += order.home_change;
+        self.home.free += order.free_home_change;
+        self.home.locked += order.lock_home_change;
     }
 
     pub fn __str__(&self) -> String {
@@ -427,27 +479,27 @@ impl AccountStatus {
     }
     #[getter]
     pub fn get_home(&self) -> f64 {
-        return self.home.to_f64().unwrap();
+        return self.home.volume.to_f64().unwrap();
     }
     #[getter]
     pub fn get_home_free(&self) -> f64 {
-        return self.home_free.to_f64().unwrap();
+        return self.home.free.to_f64().unwrap();
     }
     #[getter]
     pub fn get_home_locked(&self) -> f64 {
-        return self.home_locked.to_f64().unwrap();
+        return self.home.locked.to_f64().unwrap();
     }
     #[getter]
     pub fn get_foreign(&self) -> f64 {
-        return self.foreign.to_f64().unwrap();
+        return self.foreign.volume.to_f64().unwrap();
     }
     #[getter]
     pub fn get_foreign_free(&self) -> f64 {
-        return self.foreign_free.to_f64().unwrap();
+        return self.foreign.free.to_f64().unwrap();
     }
     #[getter]
     pub fn get_foreign_locked(&self) -> f64 {
-        return self.foreign_locked.to_f64().unwrap();
+        return self.foreign.locked.to_f64().unwrap();
     }
 }
 
@@ -456,6 +508,8 @@ impl AccountStatus {
 pub struct Order {
     // オーダー作成時に必須のデータ。以後変化しない。
     #[pyo3(get)]
+    pub category: String,
+    #[pyo3(get)]
     pub symbol: String,
     #[pyo3(get)]
     pub create_time: MicroSec, // in us
@@ -510,6 +564,7 @@ pub struct Order {
 impl Order {
     #[new]
     pub fn new(
+        category: &str,
         symbol: &str,
         create_time: MicroSec,
         order_id: &str,
@@ -521,6 +576,7 @@ impl Order {
         size: Decimal,
     ) -> Self {
         return Order {
+            category: category.to_string(),
             symbol: symbol.to_string(),
             create_time,
             status: order_status,
@@ -1109,6 +1165,7 @@ mod order_tests {
 
     fn create_order() -> Order {
         let mut order = Order::new(
+            "",
             "",
             0,
             "",
diff --git a/rbot_lib/src/common/orderbook.rs b/rbot_lib/src/common/orderbook.rs
index 4466be00..4ae82e34 100644
--- a/rbot_lib/src/common/orderbook.rs
+++ b/rbot_lib/src/common/orderbook.rs
@@ -8,7 +8,7 @@ use polars_core::{
     prelude::{DataFrame, NamedFrom},
     series::Series,
 };
-use pyo3::pyclass;
+use pyo3::{pyclass, pyfunction};
 use rust_decimal::prelude::*;
 use rust_decimal_macros::dec;
 use serde::Deserialize;
@@ -18,34 +18,91 @@ use crate::common::MarketConfig;
 
 use rmp_serde::to_vec;
 
-use super::string_to_decimal;
+use super::{order, string_to_decimal, MicroSec, Order, OrderSide, OrderStatus, OrderType, ServerConfig};
 
-pub static ALL_BOARD: Lazy<OrderBookList> = Lazy::new(||OrderBookList::new());
+static ALL_BOARD: Lazy<Mutex<OrderBookList>> = Lazy::new(||Mutex::new(OrderBookList::new()));
 
-struct OrderBookList<'a> {
-    books: HashMap<&'a str, Arc<Mutex<OrderBook>>>,
+pub struct OrderBookList {
+    books: HashMap<String, Arc<Mutex<OrderBookRaw>>>,
 }
 
-impl<'a> OrderBookList<'a> {
+impl OrderBookList {
+    pub fn make_path(exchange_name: &str, config: &MarketConfig) -> String {
+        Self::make_path_from_str(exchange_name, &config.trade_category, &config.trade_symbol)
+    }
+
+    pub fn make_path_from_str(exchange_name: &str, trade_category: &str, trade_symbol: &str) -> String {
+        format!("{}/{}/{}", exchange_name, trade_category, trade_symbol)        
+    }
+
+    pub fn parse_path(path: &str) -> (String, String, String) {
+        let v: Vec<&str> = path.split('/').collect();
+        (v[0].to_string(), v[1].to_string(), v[2].to_string())
+    }
+
     pub fn new() -> Self {
         OrderBookList {
             books: HashMap::new(),
         }
     }
 
-    pub fn get(&self, path: &str) -> Option<&Arc<Mutex<OrderBook>>> {
-        self.books.get(path)
+    pub fn get(&self, key: &String) -> Option<OrderBook> {
+        let board = self.books.get(key);
+
+        if board.is_none() {
+            return None;
+        }
+
+        let (exchange, category, symbol) = Self::parse_path(key);
+        
+        Some(OrderBook {
+            exchage: exchange,
+            category: category,
+            symbol: symbol,
+            board: board.unwrap().clone()
+        })
     }
 
-    pub fn register(&mut self, path: &'a str, book: Arc<Mutex<OrderBook>>) {
-        self.books.insert(path, book);
+    pub fn register(&mut self, path: &String, book: Arc<Mutex<OrderBookRaw>>) {
+        self.books.insert(path.clone(), book);
     }
 
-    pub fn unregister(&mut self, key: &str) {
+    pub fn unregister(&mut self, key: &String) {
         self.books.remove(key);
     }
+
+    pub fn list(&self) -> Vec<String> {
+        self.books.keys().map(|s| s.to_string()).collect()
+    }
 }
 
+#[pyfunction]
+pub fn get_orderbook_list() -> Vec<String> {
+    ALL_BOARD.lock().unwrap().list()
+}
+
+#[pyfunction]
+pub fn get_order_book_json(path: &str, size: usize) -> anyhow::Result<String> {
+    let board = ALL_BOARD.lock().unwrap().get(&path.to_string())
+        .ok_or_else(|| anyhow::anyhow!("orderbook path=({})not found", path))?;
+
+    board.get_json(size)
+}
+
+#[pyfunction]
+pub fn get_orderbook_vec(path: &str) -> anyhow::Result<(Vec<BoardItem>, Vec<BoardItem>)> {
+    let board = ALL_BOARD.lock().unwrap().get(&path.to_string())
+        .ok_or_else(|| anyhow::anyhow!("orderbook path=({})not found", path))?;
+
+    board.get_board_vec()
+}
+
+pub fn get_orderbook_df(path: &str) -> anyhow::Result<(DataFrame, DataFrame)> {
+    let board = ALL_BOARD.lock().unwrap().get(&path.to_string())
+        .ok_or_else(|| anyhow::anyhow!("orderbook path=({})not found", path))?;
+
+    board.get_board()
+}
 
 
 #[pyclass]
@@ -303,13 +360,28 @@ impl OrderBookRaw {
 #[pyclass]
 #[derive(Debug)]
 pub struct OrderBook {
-    board: Mutex<OrderBookRaw>,
+    exchage: String,
+    category: String,
+    symbol: String,
+    board: Arc<Mutex<OrderBookRaw>>,
 }
 
 impl OrderBook {
-    pub fn new(config: &MarketConfig) -> Self {
+    pub fn new(server: &dyn ServerConfig, config: &MarketConfig) -> Self {
+        let exchange_name = server.get_exchange_name();
+        let category = config.trade_category.clone();
+        let symbol = config.trade_symbol.clone();
+
+        let path = OrderBookList::make_path(&exchange_name, config);
+        let board = Arc::new(Mutex::new(OrderBookRaw::new(config.board_depth)));
+        
+        ALL_BOARD.lock().unwrap().register(&path,board.clone());
+        
         OrderBook {
-            board: Mutex::new(OrderBookRaw::new(config.board_depth)),
+            exchage: exchange_name,
+            category: category,
+            symbol: symbol,
+            board: board,
         }
     }
 
@@ -333,11 +405,13 @@ impl OrderBook {
 
     pub fn get_json(&self, size: usize) -> anyhow::Result<String> {
         let board = self.board.lock().unwrap();
-        let bids = board.bids.get();
-        let bids = bids[0..size.min(bids.len())].to_vec();
+        let mut bids = board.bids.get();
+        let mut asks = board.asks.get();
 
-        let asks = board.asks.get();
-        let asks = asks[0..size.min(asks.len())].to_vec();
+        if size != 0 {  // if there is a size limit, cut off the data.
+            bids = bids[0..size.min(bids.len())].to_vec();
+            asks = asks[0..size.min(asks.len())].to_vec();
+        }
 
         let json = serde_json::json!({
             "bids": bids,
@@ -358,6 +432,92 @@ impl OrderBook {
             .update(bids_diff, asks_diff, force);
     }
 
+    pub fn dry_market_order(
+        &mut self, 
+        create_time: MicroSec,
+        order_id: &str,
+        client_order_id: &str,
+        symbol: &str,
+        side: OrderSide,
+        size: Decimal,
+        transaction_id: &str,
+    ) -> anyhow::Result<Vec<Order>> {
+        let board = self.board.lock().unwrap();
+
+        let board = if side == OrderSide::Buy {
+            board.asks.get()
+        } else {
+            board.bids.get()
+        };
+
+        let mut orders: Vec<Order> = vec![];
+        let mut split_index = 0;
+
+        let mut remain_size = size;
+
+        // TODO: consume boards
+        for item in board {
+            if remain_size <= dec![0.0] {
+                break;
+            }
+
+            let execute_size;
+            let order_status;
+            split_index += 1;
+
+            if remain_size <= item.size {
+                order_status = OrderStatus::Filled;
+                execute_size = remain_size;
+                remain_size = dec![0.0];
+            } else {
+                order_status = OrderStatus::PartiallyFilled;
+                execute_size = item.size;
+                remain_size -= item.size;
+            }
+
+            let mut order = Order::new(
+                &self.category,
+                symbol,
+                create_time,
+                &order_id,
+                &client_order_id.to_string(),
+                side,
+                OrderType::Market,
+                order_status,
+                dec![0.0],
+                size,
+            );
+
+            order.transaction_id = format!("{}-{}", transaction_id, split_index);
+            order.update_time = create_time;
+            order.is_maker = false;
+            order.execute_price = item.price;
+            order.execute_size = execute_size;
+            order.remain_size = remain_size;
+            order.quote_vol = order.execute_price * order.execute_size;
+
+            orders.push(order);
+        }
+
+        if remain_size > dec![0.0] {
+            log::error!("remain_size > 0.0: {:?}", remain_size);
+        }
+
+        Ok(orders)
+    }
+}
+
+impl Drop for OrderBook {
+    fn drop(&mut self) {
+        let count = Arc::strong_count(&self.board);
+        log::debug!("drop orderbook: {}", count);
+
+        if count <= 1 {
+            ALL_BOARD.lock().unwrap().unregister(
+                &OrderBookList::make_path_from_str(
+                    &self.exchage, &self.category, &self.symbol));
+        }
+    }
 }
 
 #[cfg(test)]
diff --git a/rbot_lib/src/common/util.rs b/rbot_lib/src/common/util.rs
index 31d4ade8..cdcf7fa4 100644
--- a/rbot_lib/src/common/util.rs
+++ b/rbot_lib/src/common/util.rs
@@ -3,8 +3,7 @@
 
 #![allow(dead_code)]
 
-use std::{fmt, io::Write, ops::Deref};
-use futures::Future;
+
 use hmac::{Hmac, Mac};
 use once_cell::sync::Lazy;
 use polars_core::export::num::FromPrimitive;
@@ -12,61 +11,57 @@ use rust_decimal::Decimal;
 use serde::{de, Deserialize, Deserializer, Serialize, Serializer};
 use serde_json::Value;
 use sha2::Sha256;
+use std::{fmt, io::Write, ops::Deref};
+use anyhow::anyhow;
 
 use super::env_rbot_db_root;
 
-pub static RUNTIME: Lazy<tokio::runtime::Runtime> = Lazy::new(|| tokio::runtime::Runtime::new().unwrap());
 pub static DB_ROOT: Lazy<String> = Lazy::new(|| {
     if let Ok(path) = env_rbot_db_root() {
         return path;
-    }
-    else{
+    } else {
         "".to_string()
     }
 });
 
 #[derive(Clone, Deserialize)]
-pub struct SecretString{
-    str: String
+pub struct SecretString {
+    secret: String,
 }
 
-impl SecretString{
-    pub fn new(s: &str) -> SecretString{
-        SecretString{
-            str: s.to_string()
+impl SecretString {
+    pub fn new(s: &str) -> SecretString {
+        SecretString {
+            secret: s.to_string(),
         }
     }
 
-    fn as_str(&self) -> &str{
-        &self.str
-    } 
-
-    fn to_string(&self) -> String{
-        self.str.clone()
-    }   
+    pub fn extract(&self) -> String {
+        self.secret.clone()
+    }
 }
 
-impl Deref for SecretString{
+impl Deref for SecretString {
     type Target = String;
 
-    fn deref(&self) -> &Self::Target{
-        &self.str
+    fn deref(&self) -> &Self::Target {
+        &self.secret
     }
 }
 
-impl fmt::Display for SecretString{
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result{
+impl fmt::Display for SecretString {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         write!(f, "********")
     }
 }
 
-impl fmt::Debug for SecretString{
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result{
+impl fmt::Debug for SecretString {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         write!(f, "********")
     }
 }
 
-impl Serialize for SecretString{
+impl Serialize for SecretString {
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
     where
         S: Serializer,
@@ -76,13 +71,6 @@ impl Serialize for SecretString{
 }
 
 
-#[allow(non_snake_case)]
-pub fn BLOCK_ON<F: Future>(f: F) -> F::Output {
-    let r = RUNTIME.block_on(f);
-    return r;
-}
-
-
 pub fn string_to_f64<'de, D>(deserializer: D) -> Result<f64, D::Error>
 where
     D: Deserializer<'de>,
@@ -174,4 +162,3 @@ pub fn hmac_sign(secret_key: &String, message: &String) -> String {
 
     hex::encode(mac.into_bytes())
 }
-
diff --git a/rbot_lib/src/db/sqlite.rs b/rbot_lib/src/db/sqlite.rs
index f2e0c1e9..bfd3c86c 100644
--- a/rbot_lib/src/db/sqlite.rs
+++ b/rbot_lib/src/db/sqlite.rs
@@ -608,7 +608,8 @@ impl TradeTable {
     pub fn start_thread(&mut self) -> Sender<Vec<Trade>> {
         // check if the thread is already started
         // check self.tx is valid and return clone of self.tx
-        if self.tx.is_some() && self.is_running() {
+        log::debug!("start_thread");
+        if self.is_running() {
             log::info!("DB Thread is already started, reuse tx");
             return self.tx.clone().unwrap();
         }
diff --git a/rbot_lib/src/net/rest.rs b/rbot_lib/src/net/rest.rs
index 8b4692b8..9844384e 100644
--- a/rbot_lib/src/net/rest.rs
+++ b/rbot_lib/src/net/rest.rs
@@ -1,4 +1,4 @@
-// Copyright(c) 2023. yasstake. All rights reserved.
+// Copyright(c) 2023-4. yasstake. All rights reserved.
 // Abloultely no warranty.
 
 use anyhow::Context;
@@ -24,9 +24,10 @@ use tempfile::tempdir;
 //use tokio::spawn;
 use zip::ZipArchive;
 
+use crate::common::AccountCoins;
 use crate::common::{
-    flush_log, to_naive_datetime, AccountStatus, BoardTransfer, Kline, LogStatus, MarketConfig,
-    MicroSec, Order, OrderSide, OrderType, ServerConfig, Trade, BLOCK_ON, DAYS, FLOOR_DAY, TODAY,
+    flush_log, to_naive_datetime, BoardTransfer, Kline, LogStatus, MarketConfig,
+    MicroSec, Order, OrderSide, OrderType, ServerConfig, Trade, DAYS, FLOOR_DAY, TODAY,
 };
 //use crate::db::KEY::low;
 
@@ -65,13 +66,12 @@ where
     ) -> impl std::future::Future<Output = anyhow::Result<Order>> + Send;
     fn open_orders(
         server: &T,
-        config: &MarketConfig,
+config: &MarketConfig,
     ) -> impl std::future::Future<Output = anyhow::Result<Vec<Order>>> + Send;
 
     fn get_account(
         server: &T,
-        config: &MarketConfig,
-    ) -> impl std::future::Future<Output = anyhow::Result<AccountStatus>> + Send;
+    ) -> impl std::future::Future<Output = anyhow::Result<AccountCoins>> + Send;
 
     fn history_web_url(server: &T, config: &MarketConfig, date: MicroSec) -> String {
         let history_web_base = server.get_historical_web_base();
@@ -102,8 +102,8 @@ where
     fn rec_to_trade(rec: &StringRecord) -> Trade;
     fn archive_has_header() -> bool;
 
-    fn latest_archive_date(server: &T, config: &MarketConfig) -> anyhow::Result<MicroSec> {
-        BLOCK_ON(async {
+    fn latest_archive_date(server: &T, config: &MarketConfig) -> impl std::future::Future<Output = anyhow::Result<MicroSec>> + Send {
+        async move {
             let f = |date: MicroSec| -> String { Self::history_web_url(server, config, date) };
             let mut latest = TODAY();
             let mut i = 0;
@@ -122,7 +122,7 @@ where
                     return Err(anyhow!("Find archive retry over {}/{}/{}", i, latest, f(latest)));
                 }
             }
-        })
+        }
     }
 
     fn download_archive(
@@ -465,7 +465,7 @@ where
     urls
 }
 
-const MAX_BUFFER_SIZE: usize = 2000;
+const MAX_BUFFER_SIZE: usize = 4096;
 
 const LOW_QUEUE_SIZE: usize = 5;
 
diff --git a/rbot_lib/src/net/udp.rs b/rbot_lib/src/net/udp.rs
index b15e0779..29cc4d33 100644
--- a/rbot_lib/src/net/udp.rs
+++ b/rbot_lib/src/net/udp.rs
@@ -2,6 +2,11 @@ use std::mem::MaybeUninit;
 use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
 use std::str::FromStr;
 
+use futures::{stream, Stream};
+use polars_core::utils::arrow::legacy::kernels::concatenate;
+use pyo3::ffi::symtable;
+use reqwest::header::AGE;
+use tokio::runtime::Runtime;
 use tokio::task::spawn;
 
 use crossbeam_channel::Receiver;
@@ -13,6 +18,9 @@ use pyo3::pyclass;
 use serde_derive::Deserialize;
 use serde_derive::Serialize;
 
+use async_stream::stream;
+use futures::stream::StreamExt;
+
 use crate::common::{env_rbot_multicast_addr, env_rbot_multicast_port, MarketMessage};
 //use crate::common::{AccountStatus, Order, Trade};
 
@@ -28,6 +36,14 @@ pub struct BroadcastMessage {
     pub msg: MarketMessage,
 }
 
+impl BroadcastMessage {
+    pub fn filter(&self, exchange: &str, category: &str, symbol: &str) -> bool {
+        (self.exchange == exchange || exchange == "")
+            && (self.category == category || category == "")
+            && (self.symbol == symbol || symbol == "")
+    }
+}
+
 impl Into<MarketMessage> for BroadcastMessage {
     fn into(self) -> MarketMessage {
         self.msg
@@ -101,16 +117,13 @@ const UDP_SIZE: usize = 4096;
 
 #[derive(Debug)]
 pub struct UdpReceiver {
-    market_name: String,
-    market_category: String,
-    symbol: String,
     socket: Socket,
     buf: [MaybeUninit<u8>; UDP_SIZE],
 }
 
 impl UdpReceiver {
     // TODO: remove aget_id from param
-    pub fn open(market_name: &str, market_category: &str, symbol: &str, _agent_id: &str) -> Self {
+    pub fn open() -> Self {
         let multicast_addr = Ipv4Addr::from_str(&env_rbot_multicast_addr());
         if multicast_addr.is_err() {
             log::error!("multicast_addr error {:?}", multicast_addr);
@@ -137,29 +150,14 @@ impl UdpReceiver {
         let buf = [MaybeUninit::uninit(); UDP_SIZE]; // Initialize the buffer with a properly sized array
 
         Self {
-            market_name: market_name.to_string(),
-            market_category: market_category.to_string(),
-            symbol: symbol.to_string(),
             socket: socket,
             buf: buf,
         }
     }
 
-    // TODO: check address for security reson.
     pub fn receive(&mut self) -> Result<String, std::io::Error> {
         let (amt, _addr) = self.socket.recv_from(&mut self.buf)?;
 
-        /* TODO: implment
-        if let Some(sendr_ip) = addr.as_socket_ipv4() {
-            if *(sendr_ip.ip()) != IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)) {
-                return Err(std::io::Error::new(
-                    std::io::ErrorKind::Other,
-                    format!("invalid address 1 {:?}/{:?}", addr, self.local_addr),
-                ));
-            }
-        }
-        */
-
         let msg = &self.buf[..amt];
         let m = unsafe { std::mem::transmute::<_, &[u8]>(msg) };
 
@@ -173,47 +171,107 @@ impl UdpReceiver {
         Ok(msg)
     }
 
-    pub fn receive_market_message(&mut self) -> Result<MarketMessage, std::io::Error> {
-        let mut msg: BroadcastMessage;
+    pub async fn async_receive(&mut self) -> Result<String, std::io::Error> {
+        let (amt, _addr) = self.socket.recv_from(&mut self.buf)?;
+
+        let msg = &self.buf[..amt];
+        let m = unsafe { std::mem::transmute::<_, &[u8]>(msg) };
 
-        loop {
-            msg = self.receive_message()?;
+        let msg = std::str::from_utf8(m).unwrap();
+        Ok(msg.to_string())
+    }
 
-            log::debug!("receive_market_message raw: {:?}", msg);
+    pub async fn async_receive_message(&mut self) -> Result<BroadcastMessage, std::io::Error> {
+        let msg = self.async_receive().await?;
+        let msg = serde_json::from_str::<BroadcastMessage>(&msg)?;
+        Ok(msg)
+    }
 
-            if (msg.exchange == self.market_name || self.market_name == "")
-                && (msg.category == self.market_category || self.market_category == "")
-                && (msg.symbol == self.symbol || self.symbol == "")
-            {
-                break;
+    pub async fn receive_stream<'a>(
+        &'a mut self,
+        exchange: &'a str,
+        category: &'a str,
+        symbol: &'a str,
+        agent_id: &'a str,
+    ) -> impl Stream<Item = anyhow::Result<MarketMessage>> + 'a {
+        stream! {
+            loop {
+                let msg = self.async_receive_message().await?;
+
+                if msg.filter(exchange, category, symbol) {
+                    match msg.msg {
+                        MarketMessage::Order(ref order) => {
+                            if order.is_my_order(agent_id) {
+                                yield Ok(msg.msg);
+                            }
+                        }
+                        _ => {
+                            yield Ok(msg.msg);
+                        }
+                    }
+                }
             }
         }
+    }
+
+    pub fn receive_market_message(&mut self) -> Result<MarketMessage, std::io::Error> {
+        let msg = self.receive_message()?;
 
         let market_message: MarketMessage = msg.into();
         Ok(market_message)
     }
 
     pub fn open_channel(
-        market_name: &str,
-        market_category: &str,
+        exchange: &str,
+        category: &str,
         symbol: &str,
         agent_id: &str,
-    ) -> Result<Receiver<MarketMessage>, std::io::Error> {
-        let mut udp = Self::open(market_name, market_category, symbol, agent_id);
-        let (tx, rx) = crossbeam_channel::unbounded();
+    ) -> anyhow::Result<Receiver<MarketMessage>> {
+        let exchange = exchange.to_string();
+        let category = category.to_string();
+        let symbol = symbol.to_string();
+        let agent_id = agent_id.to_string();
+
+        let mut udp = Self::open();
+        let (tx, rx) = crossbeam_channel::unbounded::<MarketMessage>();
 
         // TOD: change to async
-        spawn(async move {
-            loop {
-                let msg = udp.receive_market_message().unwrap();
-                let r = tx.send(msg.clone());
+        std::thread::spawn(move || {
+            let runtime = Runtime::new().unwrap();
+            runtime.block_on(async move {
+                loop {
+                    let msg = udp.async_receive_message().await;
+
+                    if msg.is_err() {
+                        break;
+                    }
+
+                    let msg = msg.unwrap();
 
-                if r.is_err() {
-                    log::error!("open_channel: {}/{:?}", r.err().unwrap(), msg);
-                    break;
+                    if msg.filter(&exchange, &category, &symbol) {
+                        let market_message = msg.msg.clone();
+
+                        match market_message {
+                            MarketMessage::Order(ref order) => {
+                                if order.is_my_order(&agent_id) {
+                                    let r = tx.send(market_message.clone());
+                                    if r.is_err() {
+                                        log::error!("open_channel: {}/{:?}", r.err().unwrap(), msg);
+                                        break;
+                                    }
+                                }
+                            }
+                            _ => {
+                                let r = tx.send(market_message.clone());
+                                if r.is_err() {
+                                    log::error!("open_channel: {}/{:?}", r.err().unwrap(), msg);
+                                    break;
+                                }
+                            }
+                        }
+                    }
                 }
-                tokio::task::yield_now().await;
-            }
+            });
         });
 
         Ok(rx)
@@ -222,9 +280,9 @@ impl UdpReceiver {
 
 #[cfg(test)]
 mod test_udp {
-    use async_std::task::spawn;
-
     use crate::common::init_debug_log;
+    use futures::stream::StreamExt;
+    use tokio::task::spawn;
 
     #[test]
     fn send_test2() {
@@ -234,13 +292,12 @@ mod test_udp {
         assert!(r.is_ok());
     }
 
-
-    #[test]
-    fn receive_test3() {
+    #[tokio::test]
+    async fn receive_test3() {
         init_debug_log();
 
         // receive message
-        let mut receiver = super::UdpReceiver::open("EXA", "linear", "BTCUSDT", "b");
+        let mut receiver = super::UdpReceiver::open();
         let mut count = 0;
         spawn({
             async move {
@@ -263,7 +320,51 @@ mod test_udp {
             let r = sender.send(&format!("hello world {}", i));
             assert!(r.is_ok());
         }
+    }
+
+    #[tokio::test]
+    async fn test_open_channel() -> anyhow::Result<()> {
+        init_debug_log();
+
+        let receiver = super::UdpReceiver::open_channel("EXA", "linear", "BCTUSD", "AGENTID")?;
+
+        spawn(async move {
+            for i in 0..10 {
+                let msg = receiver.recv().unwrap();
+                println!("{}:{:?}", i, msg);
+            }
+        });
+
+        let sender = super::UdpSender::open("EXA", "linear", "BCTUSD");
+        for i in 0..10 {
+            let r = sender.send(&format!("hello world {}", i));
+            assert!(r.is_ok());
+        }
+
+        Ok(())
+    }
 
+    #[tokio::test]
+    async fn test_receive_stream() -> anyhow::Result<()> {
+        init_debug_log();
+
+        let mut udp = super::UdpReceiver::open();
+
+        spawn(async move {
+            let receiver = udp.receive_stream("", "", "", "").await;
+            let mut r = Box::pin(receiver);
+            while let Some(msg) = r.next().await {
+                let msg = msg.unwrap();
+                println!("{:?}", msg);
+            }
+        });
+
+        let sender = super::UdpSender::open("EXA", "linear", "BCTUSD");
+        for i in 0..10 {
+            let r = sender.send(&format!("hello world {}", i));
+            assert!(r.is_ok());
+        }
 
+        Ok(())
     }
 }
diff --git a/rbot_lib/src/net/ws.rs b/rbot_lib/src/net/ws.rs
index 132f5337..89b8c7fb 100644
--- a/rbot_lib/src/net/ws.rs
+++ b/rbot_lib/src/net/ws.rs
@@ -75,222 +75,6 @@ impl WsOpMessage for BinanceWsOpMessage {
 
 
 
-/*
-
-
-#[derive(Debug)]
-/// Tは、WsMessageを実装した型。Subscribeメッセージの取引所の差分を実装する。
-/// WebSocketClientは、文字列のメッセージのやり取りを行う。
-pub struct WebSocketClient<T, U>
-where
-    T: Send + Clone + 'static,
-    U: WsOpMessage + Sync + Send + Sync + Clone + 'static,
-{
-    // url: String,
-    //handle: Option<tokio::task::JoinHandle<()>>,
-    connection: Arc<Mutex<AutoConnectClient<T, U>>>,
-    message: Arc<Mutex<BroadcastChannel<MultiMarketMessage>>>,
-    handle: Option<JoinHandle<()>>, //  control_ch: MultiChannel<String>, // send sbscribe message
-}
-
-impl<T, U> WebSocketClient<T, U>
-where
-    T: Send + Sync + Clone,
-    U: WsOpMessage + Send + Sync + Clone + 'static,
-{
-    pub async fn new(
-        config: &T,
-        market_config: &MarketConfig,
-        url: &str,
-        subscribe: Vec<String>,
-        ping_interval_sec: i64,
-        switch_interval_sec: i64,
-        sync_wait_records: i64,
-        init_fn: Option<fn(&T) -> String>,
-        url_generator: Option<fn(&T, &MarketConfig) -> String>,
-    ) -> Self {
-        let mut client: AutoConnectClient<T, U> = AutoConnectClient::new(
-            &config,
-            market_config,
-            url,
-            ping_interval_sec,
-            switch_interval_sec,
-            sync_wait_records,
-            init_fn,
-            url_generator,
-        );
-
-        client.subscribe(&subscribe).await;
-
-        Self {
-            // handle: None,
-            connection: Arc::new(Mutex::new(client)),
-            message: Arc::new(Mutex::new(BroadcastChannel::new())),
-            handle: None,
-        }
-    }
-
-    pub async fn set_init_function(&mut self, init_fn: fn(&T) -> String) {
-        let mut lock = self.connection.as_ref().lock().await;
-        lock.init_fn = Some(init_fn);
-        drop(lock);
-    }
-
-    /*
-    pub fn connect<F>(&mut self, convert: F)
-    where
-        F: Fn(String) -> MultiMarketMessage + Send + Sync + Clone + 'static,
-    {
-        log::debug!("blocking connect start");
-
-        RUNTIME.block_on(async {
-            self._connect(convert).await;
-        });
-    }
-
-    pub fn connect_websocket(&mut self) {
-        log::debug!("blocking connect start");
-
-        RUNTIME.block_on(async {
-            self._connect_websocket().await;
-        });
-    }
-    */
-
-    pub async fn connect_websocket(&mut self) {
-        log::debug!("connect start");
-
-        let websocket = self.connection.clone();
-        let mut websocket = websocket.lock().await;
-        log::debug!("connect start2. get lock.");
-        websocket.connect().await;
-        log::debug!("connected");
-        drop(websocket);
-    }
-
-    /*
-    pub fn receive_text(&mut self) -> Result<String, String> {
-        let result = RUNTIME.block_on(async { Self::_receive_text(&self.connection).await });
-
-        result
-    }
-    */
-
-    pub async fn receive_text(
-        connection: &Arc<Mutex<AutoConnectClient<T, U>>>,
-    ) -> Result<String, String> {
-        let mut lock = connection.as_ref().lock().await;
-        let message = lock.receive_text().await;
-        drop(lock);
-
-        if message.is_err() {
-            log::error!("No message");
-            return Err("No message".to_string());
-        }
-
-        let message = message.unwrap();
-
-        return Ok(message.to_string());
-    }
-    
-
-    /// Receive message from websocket, and process one message
-    pub async fn process_event() {
-        log::debug!("start_event_loop");
-
-    }
-
-
-    /// connect to websocket server
-    /// start listening thread
-    pub async fn connect<F>(&mut self, convert: F)
-    where
-        F: Fn(String) -> MultiMarketMessage + Send + Sync + Clone + 'static,
-    {
-        self.connect_websocket().await;
-
-        let websocket = self.connection.clone();
-        let message_ch = self.message.clone();
-
-        let handle = tokio::spawn(async move {
-            loop {
-                let message = Self::receive_text(&websocket).await;
-                if message.is_err() {
-                    log::warn!("Error in websocket.receive_message: {:?}", message);
-                    continue;
-                }
-                let m = message.unwrap();
-
-                let m = convert(m);
-                let result = Self::send_message_channel(&message_ch, m).await;
-                if result.is_err() {
-                    log::warn!("Error in websocket.receive_message: {:?}", result);
-                    continue;
-                }
-            }
-        });
-
-        self.handle = Some(handle)
-    }
-
-    pub async fn send_message_channel(
-        ch: &Arc<Mutex<BroadcastChannel<MultiMarketMessage>>>,
-        message: MultiMarketMessage,
-    ) -> Result<(), anyhow::Error> {
-        //    log::debug!("send_message_channel: {:?}", message);
-        let mut lock = ch.as_ref().lock().await;
-        //        log::debug!("send_message_channel: lock ok");
-        lock.send(message)
-    }
-
-    /*
-    /// Append subscribe list and send subscribe message to send queue
-    pub async fn subscribe(&mut self, message: &Vec<String>) {
-        log::debug!("subscribe: {:?}", message);
-
-        let mut lock = self.subscribe_list.as_ref().write().unwrap();
-        lock.add_params(message);
-        drop(lock);
-
-        let message = self.subscribe_list.as_ref().read().unwrap().make_message();
-
-        log::debug!("call subscribe: {:?}", message);
-
-        let websocket = self.connection.clone();
-        let mut lock = websocket.lock().await;
-
-        for m in &message {
-            lock.send_text(m);
-
-            // TODO: receive status
-        }
-
-        drop(lock);
-    }
-    */
-
-    pub async fn switch(&mut self) {
-        let websocket = self.connection.clone();
-        let mut lock = websocket.lock().await;
-        lock.switch().await;
-        drop(lock);
-    }
-    /*
-    pub fn open_channel(&mut self) -> Receiver<MultiMarketMessage> {
-        RUNTIME.block_on(async { self._open_channel().await })
-    }
-    */
-
-    /// get receive queue
-    pub async fn open_channel(&mut self) -> Receiver<MultiMarketMessage> {
-        log::debug!("open_channel");
-        let mut lock = self.message.as_ref().lock().await;
-        log::debug!("open_channel lock ok");
-        lock.open_channel(0)
-    }
-}
-*/
-
 #[derive(Debug)]
 pub struct SimpleWebsocket<T, U> {
     server: T,
@@ -598,6 +382,10 @@ where
         }
     }
 
+    pub fn get_server(&self) -> T {
+        self.server.clone()
+    }
+
     pub fn get_config(&self) -> MarketConfig {
         self.config.clone()
     }
@@ -885,6 +673,9 @@ mod test_exchange_ws {
     }
 
     impl ServerConfig for TestServerConfig {
+        fn get_exchange_name(&self) -> String {
+            self.exchange_name.clone()
+        }
         fn get_public_ws_server(&self) -> String {
             self.public_ws.clone()
         }
diff --git a/rbot_market/src/market.rs b/rbot_market/src/market.rs
index bf65963e..d9de5cb7 100644
--- a/rbot_market/src/market.rs
+++ b/rbot_market/src/market.rs
@@ -1,5 +1,5 @@
 use crossbeam_channel::Sender;
-use rbot_lib::common::BLOCK_ON;
+use rbot_lib::common::AccountCoins;
 use rbot_lib::db::db_full_path;
 use rust_decimal_macros::dec;
 use std::sync::{Arc, Mutex, RwLock};
@@ -17,7 +17,7 @@ use anyhow::Context;
 
 use rbot_lib::{
     common::{
-        flush_log, time_string, AccountStatus, MarketConfig, MarketStream, MicroSec, Order,
+        flush_log, time_string, AccountPair, MarketConfig, MarketStream, MicroSec, Order,
         OrderSide, OrderType, ServerConfig, Trade, DAYS, NOW,
     },
     db::{df::KEY, sqlite::TradeTable},
@@ -66,7 +66,7 @@ pub trait OrderInterface {
     ) -> Vec<Order>;
     fn cancel_order(&self, market_config: &MarketConfig, order_id: &str) -> anyhow::Result<Order>;
     fn get_open_orders(&self, market_config: &MarketConfig) -> anyhow::Result<Vec<Order>>;
-    fn get_account(&self, market_config: &MarketConfig) -> anyhow::Result<AccountStatus>;
+    fn get_account(&self, market_config: &MarketConfig) -> anyhow::Result<AccountPair>;
 }
 
 pub trait OrderInterfaceImpl<T, U>
@@ -95,7 +95,7 @@ where
         OrderSide::from(side)
     }
 
-    fn make_order(
+    async fn make_order(
         &self,
         market_config: &MarketConfig,
         side: &str,
@@ -108,28 +108,26 @@ where
         let size = self.size_dp(&market_config, size);
         let order_side = Self::order_side(side);
 
-        BLOCK_ON(async {
-            T::new_order(
-                &self.get_server_config(),
-                &market_config,
-                order_side,
-                price,
-                size,
-                order_type,
-                client_order_id,
+        T::new_order(
+            &self.get_server_config(),
+            &market_config,
+            order_side,
+            price,
+            size,
+            order_type,
+            client_order_id,
+        )
+        .await
+        .with_context(|| {
+            format!(
+                "Error in make_order: {:?} {:?} {:?} {:?} {:?} {:?}",
+                &market_config, &side, &price, &size, &order_type, &client_order_id
             )
-            .await
-            .with_context(|| {
-                format!(
-                    "Error in make_order: {:?} {:?} {:?} {:?} {:?} {:?}",
-                    &market_config, &side, &price, &size, &order_type, &client_order_id
-                )
-            })
         })
     }
 
     //------ REST API ----
-    fn limit_order(
+    async fn limit_order(
         &self,
         market_config: &MarketConfig,
         side: &str,
@@ -146,9 +144,10 @@ where
             OrderType::Limit,
             client_order_id,
         )
+        .await
     }
 
-    fn market_order(
+    async fn market_order(
         &self,
         market_config: &MarketConfig,
         side: &str,
@@ -164,39 +163,39 @@ where
             OrderType::Market,
             client_order_id,
         )
+        .await
     }
 
-    fn cancel_order(&self, market_config: &MarketConfig, order_id: &str) -> anyhow::Result<Order> {
+    async fn cancel_order(
+        &self,
+        market_config: &MarketConfig,
+        order_id: &str,
+    ) -> anyhow::Result<Order> {
         check_if_enable_order!(self);
-        BLOCK_ON(async {
-            T::cancel_order(&self.get_server_config(), &market_config, order_id)
-                .await
-                .with_context(|| {
-                    format!(
-                        "Error in cancel_order: {:?} {:?}",
-                        &market_config, &order_id
-                    )
-                })
-        })
+
+        T::cancel_order(&self.get_server_config(), &market_config, order_id)
+            .await
+            .with_context(|| {
+                format!(
+                    "Error in cancel_order: {:?} {:?}",
+                    &market_config, &order_id
+                )
+            })
     }
 
-    fn get_open_orders(&self, market_config: &MarketConfig) -> anyhow::Result<Vec<Order>> {
-        BLOCK_ON(async {
-            T::open_orders(&self.get_server_config(), market_config)
-                .await
-                .with_context(|| format!("Error in get_open_orders: {:?}", &market_config))
-        })
+    async fn get_open_orders(&self, market_config: &MarketConfig) -> anyhow::Result<Vec<Order>> {
+        T::open_orders(&self.get_server_config(), market_config)
+            .await
+            .with_context(|| format!("Error in get_open_orders: {:?}", &market_config))
     }
 
-    fn get_account(&self, market_config: &MarketConfig) -> anyhow::Result<AccountStatus> {
-        BLOCK_ON(async {
-            T::get_account(&self.get_server_config(), market_config)
-                .await
-                .with_context(|| format!("Error in get_account: {:?}", &market_config))
-        })
+    async fn get_account(&self) -> anyhow::Result<AccountCoins> {
+        T::get_account(&self.get_server_config())
+            .await
+            .with_context(|| format!("Error in get_account"))
     }
 
-    fn start_user_stream(&mut self);
+    async fn async_start_user_stream(&mut self) -> anyhow::Result<()>;
 }
 
 pub trait MarketInterface {
@@ -315,10 +314,9 @@ where
 
         return db_path.to_str().unwrap().to_string();
     }
+    
+    fn get_latest_archive_date(&self) -> anyhow::Result<MicroSec>;
 
-    fn get_latest_archive_date(&self) -> anyhow::Result<MicroSec> {
-        T::latest_archive_date(&self.get_server_config(), &self.get_config())
-    }
 
     /// Check if database is valid at the date
     fn validate_db_by_date(&mut self, date: MicroSec) -> anyhow::Result<bool> {
@@ -333,19 +331,18 @@ where
         let start_time = NOW() - DAYS(2);
 
         let db = self.get_db();
-        let mut lock = db.lock().unwrap();
 
-        let fix_time = lock.latest_fix_time(start_time)?;
-        let unfix_time = lock.first_unfix_time(fix_time)?;
-        drop(lock);
+        let (fix_time, unfix_time) = {
+            let mut lock = db.lock().unwrap();
+            let fix_time = lock.latest_fix_time(start_time)?;
+            if fix_time == 0 {
+                return Err(anyhow!("No data found"));
+            }
 
-        if fix_time == 0 {
-            return Err(anyhow!("No data found"));
-        }
-        log::debug!(
-            "latest FIX time: {} / first unfix time: {}",
-            time_string(fix_time),
-            time_string(unfix_time));
+            let unfix_time = lock.first_unfix_time(fix_time)?;
+
+            (fix_time, unfix_time)
+        };
 
         Ok((fix_time, unfix_time))
     }
@@ -591,9 +588,11 @@ where
 
     fn get_board(&mut self) -> anyhow::Result<(PyDataFrame, PyDataFrame)> {
         let orderbook = self.get_order_book();
-        let lock = orderbook.read().unwrap();
 
-        let (mut bids, mut asks) = lock.get_board()?;
+        let (mut bids, mut asks) = {
+            let lock = orderbook.read().unwrap();
+            lock.get_board()?
+        };
 
         if bids.shape().0 == 0 || asks.shape().0 == 0 {
             return Ok((PyDataFrame(bids), PyDataFrame(asks)));
@@ -608,12 +607,13 @@ where
             self.reflesh_order_book();
 
             let orderbook = self.get_order_book();
-            let lock = orderbook.read().unwrap();
 
-            let r = lock.get_board();
+            (bids, asks) = {
+                let lock = orderbook.read().unwrap();
 
-            (bids, asks) = r.unwrap();
-            drop(lock)
+                lock.get_board()
+            }
+            .with_context(|| "Error in get_board")?;
         }
 
         return Ok((PyDataFrame(bids), PyDataFrame(asks)));
@@ -621,27 +621,35 @@ where
 
     fn get_board_json(&self, size: usize) -> anyhow::Result<String> {
         let orderbook = self.get_order_book();
-        let lock = orderbook.read().unwrap();
 
-        let json = lock.get_json(size)?;
-        drop(lock);
+        let json = {
+            let lock = orderbook.read().unwrap();
+            lock.get_json(size)?            
+        };
 
         Ok(json)
     }
 
     fn get_board_vec(&self) -> anyhow::Result<(Vec<BoardItem>, Vec<BoardItem>)> {
         let orderbook = self.get_order_book();
-        let lock = orderbook.read().unwrap();
-        let (bids, asks) = lock.get_board_vec()?;
+        
+        let (bids, asks) = {
+            let lock = orderbook.read().unwrap();            
+            lock.get_board_vec()?
+        };
 
         Ok((bids, asks))
     }
 
     fn get_edge_price(&self) -> anyhow::Result<(Decimal, Decimal)> {
         let orderbook = self.get_order_book();
-        let lock = orderbook.read().unwrap();
 
-        Ok(lock.get_edge_price())
+        let edge_price = {
+            let lock = orderbook.read().unwrap();
+            lock.get_edge_price()
+        };
+
+        Ok(edge_price)
     }
 
     fn get_market_config(&self) -> MarketConfig;
@@ -649,100 +657,28 @@ where
     fn start_db_thread(&mut self) -> Sender<Vec<Trade>> {
         let db = self.get_db();
         let mut lock = db.lock().unwrap();
-        BLOCK_ON(async { lock.start_thread() })
-    }
 
-    /// Download historical data archive and store to database.
-    fn download_archives(
-        &mut self,
-        ndays: i64,
-        force: bool,
-        verbose: bool,
-        low_priority: bool,
-    ) -> anyhow::Result<i64> {
-        log::info!("log download: {} days", ndays);
-        if verbose {
-            println!("log download: {} days", ndays);
-            flush_log();
-        }
-
-        let latest_date = match self.get_latest_archive_date() {
-            Ok(timestamp) => timestamp,
-            Err(_) => NOW() - DAYS(2),
-        };
-
-        log::info!("archive latest_date: {}", time_string(latest_date));
-        if verbose {
-            println!("archive latest_date: {}", time_string(latest_date));
-            flush_log();
-        }
-
-        let mut download_rec: i64 = 0;
-
-        let tx = self.start_db_thread();
-
-        for i in 0..ndays {
-            let date = latest_date - i * DAYS(1);
-
-            if !force && self.validate_db_by_date(date)? {
-                log::info!("{} is valid", time_string(date));
-
-                if verbose {
-                    println!("{} skip download", time_string(date));
-                    flush_log();
-                }
-                continue;
-            }
-
-            match self.download_archive(&tx, date, low_priority, verbose) {
-                Ok(rec) => {
-                    log::info!("downloaded: {}", download_rec);
-                    download_rec += rec;
-                }
-                Err(e) => {
-                    log::error!("Error in download_log: {:?}", e);
-                    if verbose {
-                        println!("Error in download_log: {:?}", e);
-                    }
-                }
-            }
-        }
-
-        Ok(download_rec)
+        lock.start_thread()
     }
 
-    fn download_archive(
+    fn download_recent_trades(
         &self,
-        tx: &Sender<Vec<Trade>>,
-        date: MicroSec,
-        low_priority: bool,
-        verbose: bool,
-    ) -> anyhow::Result<i64> {
-        BLOCK_ON(async {
-            T::download_archive(
-                &self.get_server_config(),
-                &self.get_config(),
-                tx,
-                date,
-                low_priority,
-                verbose,
-            )
-            .await
-            .with_context(|| format!("Error in download_archive: {:?}", date))
-        })
-    }
+        market_config: &MarketConfig,
+    ) -> impl std::future::Future<Output = anyhow::Result<Vec<Trade>>> + Send where Self: Sync {async {
+        T::get_recent_trades(&self.get_server_config(), market_config).await
+    } }
+
+
 
-    fn download_recent_trades(&self, market_config: &MarketConfig) -> anyhow::Result<Vec<Trade>> {
-        BLOCK_ON(async { T::get_recent_trades(&self.get_server_config(), market_config).await })
-    }
 
     fn expire_unfix_data(&mut self) -> anyhow::Result<()> {
         let db = self.get_db();
         let now = NOW();
 
-        let mut lock = db.lock().unwrap();
-        let expire_message = lock.make_expire_control_message(now)?;
-        drop(lock);
+        let expire_message = {
+            let mut lock = db.lock().unwrap();
+            lock.make_expire_control_message(now)?
+        };
 
         let tx = self.start_db_thread();
         let r = tx.send(expire_message);
@@ -764,4 +700,96 @@ where
         time_from: MicroSec,
         time_to: MicroSec,
     ) -> anyhow::Result<MarketStream>;
+
+
+    /*------------   async ----------------*/
+    async fn async_get_latest_archive_date(&self) -> anyhow::Result<MicroSec> {
+        let server_config = self.get_server_config();
+        let config = self.get_config().clone();
+
+        let r = T::latest_archive_date(&server_config, &config).await;
+        r
+    }
+
+    async fn async_download_archive(
+        &self,
+        tx: &Sender<Vec<Trade>>,
+        date: MicroSec,
+        low_priority: bool,
+        verbose: bool,
+    ) -> anyhow::Result<i64> {
+        T::download_archive(
+            &self.get_server_config(),
+            &self.get_config(),
+            tx,
+            date,
+            low_priority,
+            verbose,
+        )
+        .await
+        .with_context(|| format!("Error in download_archive: {:?}", date))
+    }
+
+        /// Download historical data archive and store to database.
+        async fn async_download_archives(
+            &mut self,
+            ndays: i64,
+            force: bool,
+            verbose: bool,
+            low_priority: bool,
+        ) -> anyhow::Result<i64> {
+            log::info!("log download: {} days", ndays);
+            if verbose {
+                println!("log download: {} days", ndays);
+                flush_log();
+            }
+    
+            let latest_date = match self.async_get_latest_archive_date().await {
+                Ok(timestamp) => timestamp,
+                Err(_) => NOW() - DAYS(2),
+            };
+    
+            log::info!("archive latest_date: {}", time_string(latest_date));
+            if verbose {
+                println!("archive latest_date: {}", time_string(latest_date));
+                flush_log();
+            }
+    
+            let mut download_rec: i64 = 0;
+    
+            let tx = self.start_db_thread();
+    
+            for i in 0..ndays {
+                let date = latest_date - i * DAYS(1);
+    
+                if !force && self.validate_db_by_date(date)? {
+                    log::info!("{} is valid", time_string(date));
+    
+                    if verbose {
+                        println!("{} skip download", time_string(date));
+                        flush_log();
+                    }
+                    continue;
+                }
+    
+                match self
+                    .async_download_archive(&tx, date, low_priority, verbose)
+                    .await
+                {
+                    Ok(rec) => {
+                        log::info!("downloaded: {}", download_rec);
+                        download_rec += rec;
+                    }
+                    Err(e) => {
+                        log::error!("Error in download_log: {:?}", e);
+                        if verbose {
+                            println!("Error in download_log: {:?}", e);
+                        }
+                    }
+                }
+            }
+    
+            Ok(download_rec)
+        }
+    
 }
diff --git a/rbot_server/Cargo.toml b/rbot_server/Cargo.toml
new file mode 100644
index 00000000..48f58a72
--- /dev/null
+++ b/rbot_server/Cargo.toml
@@ -0,0 +1,16 @@
+[package]
+name = "rbot_server"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+rbot_lib = { path = "../rbot_lib" }
+
+anyhow={workspace=true}
+tokio={workspace=true}
+actix-rt={workspace=true}
+actix-web = {workspace=true}
+serde = {workspace=true}
+
diff --git a/rbot_server/src/lib.rs b/rbot_server/src/lib.rs
new file mode 100644
index 00000000..6f1ad8b0
--- /dev/null
+++ b/rbot_server/src/lib.rs
@@ -0,0 +1,3 @@
+
+mod server;
+
diff --git a/rbot_server/src/server.rs b/rbot_server/src/server.rs
new file mode 100644
index 00000000..c99fb046
--- /dev/null
+++ b/rbot_server/src/server.rs
@@ -0,0 +1,79 @@
+use actix_web::{web, App, HttpResponse, HttpServer, Responder, get};
+
+use rbot_lib::common::{get_order_book_json, OrderBookList};
+use serde::{ser, Deserialize};
+use actix_rt::System;
+
+#[get("/")]
+async fn greet() -> impl Responder {
+    HttpResponse::Ok().body("Hello, World!")
+}
+
+#[derive(Deserialize)]
+struct PathInfo {
+    exchange: String,
+    category: String,
+    symbol: String,
+}
+
+
+#[get("/board/json/{exchange}/{category}/{symbol}")]
+async fn get_board_json(path: web::Path<PathInfo>) -> impl Responder {
+    let path = OrderBookList::make_path_from_str(&path.exchange, &path.category, &path.symbol);
+
+    let json = get_order_book_json(&path, 0);
+    
+    if json.is_err() {
+        return HttpResponse::NotFound().body(format!("Not Found {}", path));
+    }
+
+    HttpResponse::Ok().body(json.unwrap())
+}
+
+fn start_server() -> anyhow::Result<()> {
+    let sys = actix_rt::System::new();
+
+    let server = HttpServer::new(|| {
+        App::new()
+        .service(greet)
+    })
+    .bind("127.0.0.1:8080").expect("Failed to bind server")
+    .run();
+
+    sys.block_on(server).unwrap();
+
+    Ok(())
+}
+
+fn start_rest_server() -> anyhow::Result<()> {
+    tokio::task::spawn_blocking(move || {
+        start_server()
+    });
+
+    Ok(())
+}
+
+
+#[cfg(test)]
+mod test_rest_server {
+    use super::*;
+    use std::thread::sleep;
+    use std::time::Duration;
+    
+
+    #[test]
+    fn test_start_server_block(){
+        let server = start_server();
+        assert!(server.is_ok());
+    }
+
+
+    #[test]
+    fn test_start_server() {
+        let server = start_rest_server();
+        assert!(server.is_ok());
+
+
+        sleep(Duration::from_secs(120));
+    }
+}
\ No newline at end of file
diff --git a/rbot_session/Cargo.toml b/rbot_session/Cargo.toml
index 57ff1ae1..994574c0 100644
--- a/rbot_session/Cargo.toml
+++ b/rbot_session/Cargo.toml
@@ -8,7 +8,9 @@ edition = "2021"
 [dependencies]
 rbot_lib = {workspace = true}
 rbot_market = {workspace = true}
+rbot_blockon = {workspace = true}
 
+anyhow = {workspace = true}
 
 rust_decimal = {workspace = true}
 rust_decimal_macros = {workspace = true}
@@ -25,11 +27,17 @@ serde_derive = {workspace = true}
 serde_with = {workspace = true}
 serde_json = {workspace = true}
 
+futures = {workspace = true}
+
 log = {workspace = true}
 env_logger = {workspace = true}
 
 crossbeam-channel = {workspace = true}
 
+async-stream = {workspace = true}
+
+tokio={workspace = true}
+
 # https://pyo3.rs/v0.13.2/faq
 [dependencies.pyo3]
 version = "0.20"
diff --git a/rbot_session/src/logger.rs b/rbot_session/src/logger.rs
index d92ce46f..46cf646b 100644
--- a/rbot_session/src/logger.rs
+++ b/rbot_session/src/logger.rs
@@ -11,7 +11,7 @@ use pyo3_polars::PyDataFrame;
 use serde_derive::{Deserialize, Serialize};
 use serde_with::skip_serializing_none;
 
-use rbot_lib::common::{ordervec_to_dataframe, AccountStatus, MicroSec, Order};
+use rbot_lib::common::{ordervec_to_dataframe, AccountPair, MicroSec, Order};
 
 #[skip_serializing_none]
 #[derive(Debug, Serialize, Deserialize, PartialEq, Clone)]
@@ -78,7 +78,7 @@ pub enum LogMessage {
     #[serde(rename = "O")]
     Order(Order),
     #[serde(rename = "A")]
-    Account(AccountStatus),
+    Account(AccountPair),
     #[serde(rename = "i")]
     UserIndicator(Indicator),
     #[serde(rename = "I")]
@@ -167,12 +167,12 @@ pub fn account_logrec_to_df(accounts: Vec<SingleLogRecord>) -> DataFrame {
         match rec.data {
             LogMessage::Account(account) => {
                 timestamp.push(rec.timestamp);                            
-                home.push(account.home.to_f64().unwrap());
-                home_free.push(account.home_free.to_f64().unwrap());
-                home_locked.push(account.home_locked.to_f64().unwrap());
-                foreign.push(account.foreign.to_f64().unwrap());
-                foreign_free.push(account.foreign_free.to_f64().unwrap());
-                foreign_locked.push(account.foreign_locked.to_f64().unwrap());
+                home.push(account.home.volume.to_f64().unwrap());
+                home_free.push(account.home.free.to_f64().unwrap());
+                home_locked.push(account.home.locked.to_f64().unwrap());
+                foreign.push(account.foreign.volume.to_f64().unwrap());
+                foreign_free.push(account.foreign.free.to_f64().unwrap());
+                foreign_locked.push(account.foreign.locked.to_f64().unwrap());
             }
             _ => {
                 panic!("not supported message type");
@@ -358,7 +358,7 @@ impl Logger {
     pub fn log_account(
         &mut self,
         timestamp: MicroSec,
-        status: &AccountStatus,
+        status: &AccountPair,
     ) -> Result<(), std::io::Error> {
         self.log_message(timestamp, &LogMessage::Account(status.clone()))
     }
@@ -841,6 +841,7 @@ mod tests {
     #[test]
     fn test_log_record_order() {
         let order = Order::new(
+            "linear",
             "BTCUSD",
             1,
             "order-1",
diff --git a/rbot_session/src/runner.rs b/rbot_session/src/runner.rs
index a9d5706f..7540c0aa 100644
--- a/rbot_session/src/runner.rs
+++ b/rbot_session/src/runner.rs
@@ -4,14 +4,17 @@ use crossbeam_channel::Receiver;
 use pyo3::{pyclass, pymethods, types::IntoPyDict, Py, PyAny, PyErr, PyObject, Python};
 use rbot_lib::{
     common::{
-        flush_log, time_string, AccountStatus, LogStatus, MarketMessage, MarketStream, MicroSec,
-        Order, Trade, FLOOR_SEC, NOW, SEC,
+        flush_log, time_string, AccountPair, LogStatus, MarketConfig, MarketMessage, MarketStream,
+        MicroSec, Order, Trade, FLOOR_SEC, MARKET_HUB, NOW, SEC,
     },
     net::UdpReceiver,
 };
 use rust_decimal::prelude::ToPrimitive;
 
 use super::{has_method, ExecuteMode, Session};
+use async_stream::stream;
+use futures::{Stream, StreamExt};
+use rbot_blockon::BLOCK_ON;
 
 #[pyclass]
 #[derive(Debug, Clone)]
@@ -44,6 +47,7 @@ pub struct Runner {
     execute_mode: ExecuteMode,
     agent_id: String,
 
+    config: MarketConfig,
     exchange_name: String,
     category: String,
     symbol: String,
@@ -76,7 +80,7 @@ impl Runner {
             last_print_real_time: 0,
 
             agent_id: "".to_string(),
-
+            config: MarketConfig::default(),
             exchange_name: "".to_string(),
             category: "".to_string(),
             symbol: "".to_string(),
@@ -99,9 +103,10 @@ impl Runner {
         self.last_print_real_time = 0;
     }
 
-    #[pyo3(signature = (*, market, agent, start_time=0, end_time=0, execute_time=0, verbose=false, log_file=None))]
+    #[pyo3(signature = (*, exchange, market, agent, start_time=0, end_time=0, execute_time=0, verbose=false, log_file=None))]
     pub fn back_test(
         &mut self,
+        exchange: PyObject,
         market: PyObject,
         agent: &PyAny,
         start_time: MicroSec,
@@ -109,23 +114,26 @@ impl Runner {
         execute_time: i64,
         verbose: bool,
         log_file: Option<String>,
-    ) -> Result<Py<Session>, PyErr> {
+    ) -> anyhow::Result<Py<Session>> {
         self.update_market_info(&market)?;
         self.update_agent_info(agent)?;
 
-        let stream = Self::open_backtest_stream(&market, start_time, end_time);
-        let reciever = stream.reciver;
         self.execute_time = execute_time;
         self.print_interval = SEC(60 * 60);
         self.verbose = verbose;
         self.execute_mode = ExecuteMode::BackTest;
 
-        self.run(market, &reciever, agent, true, log_file)
+        BLOCK_ON(async {
+            let receiver = Self::open_backtest_stream(&market, start_time, end_time).await;
+            self.async_run(exchange, market, receiver, agent, true, log_file)
+                .await
+        })
     }
 
-    #[pyo3(signature = (*, market, agent, log_memory=false, execute_time=0, verbose=false, log_file=None, client=false))]
+    #[pyo3(signature = (*, exchange, market, agent, log_memory=false, execute_time=0, verbose=false, log_file=None, client=false))]
     pub fn dry_run(
         &mut self,
+        exchange: PyObject,
         market: PyObject,
         agent: &PyAny,
         log_memory: bool,
@@ -133,7 +141,7 @@ impl Runner {
         verbose: bool,
         log_file: Option<String>,
         client: bool,
-    ) -> Result<Py<Session>, PyErr> {
+    ) -> anyhow::Result<Py<Session>> {
         self.execute_time = execute_time;
         self.verbose = verbose;
         self.execute_mode = ExecuteMode::Dry;
@@ -141,33 +149,101 @@ impl Runner {
         self.update_market_info(&market)?;
         self.update_agent_info(agent)?;
 
-        let receiver = if client {
-            UdpReceiver::open_channel(
-                &self.exchange_name,
-                &self.category,
-                &self.symbol.as_str(),
-                &self.agent_id.as_str(),
-            )
-            .unwrap()
+        let exchange_name = self.exchange_name.clone();
+        let category = self.category.clone();
+        let symbol = self.symbol.clone();
+        let agent_id = self.agent_id.clone();
+
+        if client {
+            BLOCK_ON(async {
+                let mut udp = UdpReceiver::open();
+                let receiver = udp
+                    .receive_stream(&exchange_name, &category, &symbol, &agent_id)
+                    .await;
+
+                self.async_run(exchange, market, receiver, agent, log_memory, log_file)
+                    .await
+            })
         } else {
-            let stream = Self::get_market_stream(&market);
-            let reciever = stream.reciver;
+            self.prepare_data(&exchange, &market)?;
 
-            let r = self.prepare_data(&market);
+            BLOCK_ON(async {
+                let receiver = MARKET_HUB
+                    .subscribe_stream(&exchange_name, &category, &symbol, &agent_id)
+                    .await;
 
-            if r.is_err() {
-                return Err(r.unwrap_err());
-            }
+                self.async_run(exchange, market, receiver, agent, log_memory, log_file)
+                    .await
+            })
+        }
+    }
 
-            reciever
-        };
+    #[pyo3(signature = (*,exchange,  market, agent, log_memory=false, execute_time=0, verbose=false, log_file=None, client=false))]
+    pub fn real_run_bak(
+        &mut self,
+        exchange: PyObject,
+        market: PyObject,
+        agent: &PyAny,
+        log_memory: bool,
+        execute_time: i64,
+        verbose: bool,
+        log_file: Option<String>,
+        client: bool,
+    ) -> anyhow::Result<Py<Session>> {
+        self.update_market_info(&market)?;
+        self.update_agent_info(agent)?;
+
+        self.execute_time = execute_time;
+        self.verbose = verbose;
+        self.execute_mode = ExecuteMode::Real;
+
+        let exchange_name = self.exchange_name.clone();
+        let category = self.category.clone();
+        let symbol = self.symbol.clone();
+        let agent_id = self.agent_id.clone();
+
+        if client {
+            //            let rt = tokio::runtime::Runtime::new().unwrap();
+            let rt = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .unwrap();
+
+            rt.block_on(async {
+                let mut udp = UdpReceiver::open();
+
+                let receiver = udp
+                    .receive_stream(&exchange_name, &category, &symbol, &agent_id)
+                    .await;
 
-        self.run(market, &receiver, agent, log_memory, log_file)
+                self.async_run(exchange, market, receiver, agent, log_memory, log_file)
+                    .await
+            })
+        } else {
+            self.prepare_data(&exchange, &market)?;
+
+            //            let rt = tokio::runtime::Runtime::new().unwrap();
+            let rt = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .unwrap();
+
+            rt.block_on(async {
+                let receiver = MARKET_HUB
+                    .subscribe_stream(&exchange_name, &category, &symbol, &agent_id)
+                    .await;
+
+                self.async_run(exchange, market, receiver, agent, log_memory, log_file)
+                    .await
+            })
+        }
     }
 
-    #[pyo3(signature = (*, market, agent, log_memory=false, execute_time=0, verbose=false, log_file=None, client=false))]
+
+    #[pyo3(signature = (*,exchange,  market, agent, log_memory=false, execute_time=0, verbose=false, log_file=None, client=false))]
     pub fn real_run(
         &mut self,
+        exchange: PyObject,
         market: PyObject,
         agent: &PyAny,
         log_memory: bool,
@@ -175,7 +251,7 @@ impl Runner {
         verbose: bool,
         log_file: Option<String>,
         client: bool,
-    ) -> Result<Py<Session>, PyErr> {
+    ) -> anyhow::Result<Py<Session>> {
         self.update_market_info(&market)?;
         self.update_agent_info(agent)?;
 
@@ -183,30 +259,26 @@ impl Runner {
         self.verbose = verbose;
         self.execute_mode = ExecuteMode::Real;
 
-        let receiver = if client {
-            UdpReceiver::open_channel(
-                &self.exchange_name,
-                &self.category,
-                &self.symbol.as_str(),
-                &self.agent_id.as_str(),
-            )
-            .unwrap()
-        } else {
-            let stream = Self::get_market_stream(&market);
-            let reciever = stream.reciver;
+        let exchange_name = self.exchange_name.clone();
+        let category = self.category.clone();
+        let symbol = self.symbol.clone();
+        let agent_id = self.agent_id.clone();
 
-            let r = self.prepare_data(&market);
+        if client {
+                let receiver = UdpReceiver::
+                    open_channel(&exchange_name, &category, &symbol, &agent_id)?;
 
-            if r.is_err() {
-                return Err(r.unwrap_err());
-            }
-
-            reciever
-        };
+                self.run(exchange, market, &receiver, agent, log_memory, log_file)
+        } else {
+            self.prepare_data(&exchange, &market)?;
+                let receiver = MARKET_HUB
+                    .subscribe(&exchange_name, &category, &symbol, &agent_id)?;
 
-        self.run(market, &receiver, agent, log_memory, log_file)
+                self.run(exchange, market, &receiver, agent, log_memory, log_file)
+        }
     }
 
+
     /*
     pub fn start_proxy(&mut self, market: PyObject, port: i64) -> Result<(), PyErr> {
         let sender = UdpSender::open_with_port(port, port + 1);
@@ -302,16 +374,13 @@ impl Runner {
         "".to_string()
     }
 
-    pub fn prepare_data(&self, market: &PyObject) -> Result<(), PyErr> {
+    pub fn prepare_data(&self, exchange: &PyObject, market: &PyObject) -> anyhow::Result<()> {
         // prepare market data
         // 1. start market & user stream
         // 2. download market data
-        let r = Python::with_gil(|py| {
+        Python::with_gil(|py| {
             if self.execute_mode == ExecuteMode::Real || self.execute_mode == ExecuteMode::Dry {
-                let r = market.call_method0(py, "start_market_stream");
-                if r.is_err() {
-                    return Err(r.unwrap_err());
-                }
+                market.call_method0(py, "start_market_stream")?;
 
                 if self.verbose {
                     println!("--- start market stream ---");
@@ -320,10 +389,7 @@ impl Runner {
             }
 
             if self.execute_mode == ExecuteMode::Real {
-                let r = market.call_method0(py, "start_user_stream");
-                if r.is_err() {
-                    return Err(r.unwrap_err());
-                }
+                exchange.call_method0(py, "start_user_stream")?;
 
                 if self.verbose {
                     println!("--- start user stream ---");
@@ -337,29 +403,34 @@ impl Runner {
                     flush_log();
                 }
 
-                let kwargs = vec![("ndays", 1)];
+                let kwargs = if self.verbose {vec![("verbose", true)]} else {vec![("verbose", false)]};
+                let kwargs = kwargs.into_py_dict(py);
 
-                // let r = market.call()
+                market.call_method(py, "download_latest", (), Some(kwargs))?;
+                log::debug!("download_latest is done");
+                market.call_method(py, "download_gap", (), Some(kwargs))?;
+                log::debug!("download_gap is done");
 
-                //let r = market.call_method1(py, "download", kwargs);
-                let r = market.call_method(py, "download", (), Some(kwargs.into_py_dict(py)));
-                if r.is_err() {
-                    return Err(r.unwrap_err());
-                }
+                let kwargs = vec![("ndays", 1)];
+                market.call_method(py, "download_archive", (), Some(kwargs.into_py_dict(py)))?;
+                log::debug!("download_archive is done");
             }
 
-            Ok(())
-        });
+            Ok::<(), anyhow::Error>(()) // Add type annotation for Result
+        })?;
 
-        r
+        Ok(())
     }
 
     pub fn update_market_info(&mut self, market: &PyObject) -> Result<(), PyErr> {
         let r = Python::with_gil(|py| {
+            let config = market.getattr(py, "market_config").unwrap();
+
             let exchange_name = market.getattr(py, "exchange_name").unwrap();
             let category = market.getattr(py, "trade_category").unwrap();
             let symbol = market.getattr(py, "trade_symbol").unwrap();
 
+            self.config = config.extract::<MarketConfig>(py).unwrap();
             self.exchange_name = exchange_name.extract::<String>(py).unwrap();
             self.category = category.extract::<String>(py).unwrap();
             self.symbol = symbol.extract::<String>(py).unwrap();
@@ -429,36 +500,22 @@ impl Runner {
         Ok(())
     }
 
-    pub fn run(
-        &mut self,
+    fn create_session(
+        &self,
+        exchange: PyObject,
         market: PyObject,
-        receiver: &Receiver<MarketMessage>,
-        agent: &PyAny,
         log_memory: bool,
         log_file: Option<String>,
-    ) -> Result<Py<Session>, PyErr> {
-        if self.verbose {
-            print!("--- run {:?} mode --- ", self.execute_mode);
-            print!("market: {}, ", self.exchange_name);
-            print!("agent_id: {}, ", self.agent_id);
-            print!("log_memory: {}, ", log_memory);
-            println!("duration: {}[sec], ", self.execute_time);
-
-            flush_log();
-        }
-
-        let result = Python::with_gil(|py| {
-            if self.verbose {
-                println!("--- run loop ---");
-                flush_log();
-            }
-
+    ) -> Py<Session> {
+        Python::with_gil(|py| {
             let session_name = self.agent_id.clone();
+
             let mut session = Session::new(
+                exchange,
                 market,
                 self.execute_mode.clone(),
                 Some(&session_name),
-                log_memory,
+                false,
             );
 
             if log_file.is_some() {
@@ -475,124 +532,458 @@ impl Runner {
                 println!("WARNING: log_memory and log_file are both false. Please set one of them to true.");
             }
 
-            let py_session = Py::new(py, session).unwrap();
-            let mut warm_up_loop: i64 = WARMUP_STEPS;
+            Py::new(py, session).unwrap()
+        })
+    }
+
+    pub fn execute_message_update_session(
+        &mut self,
+        py_session: &Py<Session>,
+        message: &MarketMessage,
+    ) -> anyhow::Result<()> {
+        Python::with_gil(|py| {
+            let mut session = py_session.borrow_mut(py);
+            session.on_message(&message);
+            self.last_timestamp = session.get_timestamp();
+
+            Ok(())
+        })
+    }
 
-            if self.has_on_init {
-                self.call_agent_on_init(&py, agent, &py_session)?;
+    pub fn execute_message(
+        &mut self,
+        py_session: &Py<Session>,
+        agent: &PyAny,
+        message: &MarketMessage,
+        interval_sec: i64,
+    ) -> anyhow::Result<()> {
+        Python::with_gil(|py| {
+            let result = self.on_message(&py, agent, &py_session, &message, interval_sec);
+            if result.is_err() {
+                return Err(result.unwrap_err());
             }
+            self.loop_count += 1;
+
+            Ok(())
+        })?;
+
+        Ok(())
+    }
+
+    pub async fn async_run(
+        &mut self,
+        exchange: PyObject,
+        market: PyObject,
+        receiver: impl Stream<Item = anyhow::Result<MarketMessage>>,
+        // receiver: &Receiver<MarketMessage>,
+        agent: &PyAny,
+        log_memory: bool,
+        log_file: Option<String>,
+    ) -> anyhow::Result<Py<Session>> {
+        if self.verbose {
+            print!("--- run {:?} mode --- ", self.execute_mode);
+            print!("market: {}, ", self.exchange_name);
+            print!("agent_id: {}, ", self.agent_id);
+            print!("log_memory: {}, ", log_memory);
+            println!("duration: {}[sec], ", self.execute_time);
 
-            let interval_sec = self.get_clock_interval(&py, &py_session)?;
+            flush_log();
+        }
 
-            if self.has_on_clock && interval_sec == 0 {
-                print!("*** WARNING: on_clock is set but clock_interval is 0. (on_clock will not be called ***");
+        let py_session = self.create_session(exchange, market, log_memory, log_file);
+
+        self.call_agent_on_init(&agent, &py_session);
+        let interval_sec = self.get_clock_interval(&py_session)?;
+
+        let mut rec = Box::pin(receiver);
+
+        // warm up loop
+        let mut warm_up_loop: i64 = WARMUP_STEPS;
+        while let Some(message) = rec.next().await {
+            let message = message?;
+            self.execute_message_update_session(&py_session, &message)?;
+
+            if let MarketMessage::Trade(_trade) = &message {
+                warm_up_loop = warm_up_loop - 1;
+                if warm_up_loop <= 0 {
+                    break;
+                }
             }
+        }
 
-            self.reset_count();
-            let loop_start_time = NOW();
+        let mut remain_time: i64 = 0;
 
-            let mut remain_time: i64 = 0;
+        let loop_start_time = NOW();
+        while let Some(message) = rec.next().await {
+            let message = message?;
+            self.execute_message(&py_session, agent, &message, interval_sec)?;
+            self.loop_count += 1;
 
-            loop {
-                let message = receiver.recv();
+            // break if the running time exceeceds the loop_duration
+            if self.start_timestamp == 0 {
+                self.start_timestamp = self.last_timestamp;
+                if self.verbose {
+                    println!("start_timestamp: {}", time_string(self.start_timestamp));
+                }
+            } else if 0 < self.execute_time {
+                remain_time = self.start_timestamp + SEC(self.execute_time) - self.last_timestamp;
 
-                if message.is_err() {
-                    log::info!("Data stream is closed {:?}", message);
+                if remain_time < 0 {
+                    log::debug!("remain_time out break");
                     break;
                 }
+            }
 
-                let message = message.unwrap();
-                if self.execute_mode == ExecuteMode::BackTest {
-                    if let MarketMessage::Trade(trade) = &message {
-                        if trade.status == LogStatus::UnFix {
-                            log::info!("UnFix trade is found. Stop running.");
-                            break;
-                        }
-                    }
+            // print progress if verbose flag is set.
+            if self.verbose {
+                self.print_progress(remain_time);
+            }
+        }
+        self.print_run_result(loop_start_time);
+
+        Ok(py_session)
+    }
+
+
+    pub fn run(
+        &mut self,
+        exchange: PyObject,
+        market: PyObject,
+//        receiver: impl Stream<Item = anyhow::Result<MarketMessage>>,
+        receiver: &Receiver<MarketMessage>,
+        agent: &PyAny,
+        log_memory: bool,
+        log_file: Option<String>,
+    ) -> anyhow::Result<Py<Session>> {
+        if self.verbose {
+            print!("--- run {:?} mode --- ", self.execute_mode);
+            print!("market: {}, ", self.exchange_name);
+            print!("agent_id: {}, ", self.agent_id);
+            print!("log_memory: {}, ", log_memory);
+            println!("duration: {}[sec], ", self.execute_time);
+
+            flush_log();
+        }
+
+        let py_session = self.create_session(exchange, market, log_memory, log_file);
+
+        self.call_agent_on_init(&agent, &py_session);
+        let interval_sec = self.get_clock_interval(&py_session)?;
+
+        // warm up loop
+        let mut warm_up_loop: i64 = WARMUP_STEPS;
+        while let Ok(message) = receiver.recv() {
+            self.execute_message_update_session(&py_session, &message)?;
+
+            if let MarketMessage::Trade(_trade) = &message {
+                warm_up_loop = warm_up_loop - 1;
+                if warm_up_loop <= 0 {
+                    break;
                 }
+            }
+        }
 
-                if 0 < warm_up_loop {
-                    // in warm up loop, we don't call agent, just update session
-                    let mut session = py_session.borrow_mut(py);
-                    session.on_message(&message);
-                    self.last_timestamp = session.get_timestamp();
+        let mut remain_time: i64 = 0;
 
-                    if let MarketMessage::Trade(_trade) = &message {
-                        warm_up_loop = warm_up_loop - 1;
-                    }
+        let loop_start_time = NOW();
+        while let Ok(message) = receiver.recv() {
+            self.execute_message(&py_session, agent, &message, interval_sec)?;
+            self.loop_count += 1;
 
-                    if self.verbose {
-                        print!("\rwarm up loop: {}", warm_up_loop)
-                    }
+            // break if the running time exceeceds the loop_duration
+            if self.start_timestamp == 0 {
+                self.start_timestamp = self.last_timestamp;
+                if self.verbose {
+                    println!("start_timestamp: {}", time_string(self.start_timestamp));
+                }
+            } else if 0 < self.execute_time {
+                remain_time = self.start_timestamp + SEC(self.execute_time) - self.last_timestamp;
+
+                if remain_time < 0 {
+                    log::debug!("remain_time out break");
+                    break;
+                }
+            }
 
-                    drop(session);
+            // print progress if verbose flag is set.
+            if self.verbose {
+                self.print_progress(remain_time);
+            }
+        }
+        self.print_run_result(loop_start_time);
+
+        Ok(py_session)
+    }
+
+
+    fn print_run_result(&self, loop_start_time: i64) {
+        if self.verbose {
+            println!(
+                "\n--- end run --- {} ({}[rec])",
+                time_string(self.last_timestamp),
+                self.loop_count
+            );
+
+            println!("Done running agent");
+
+            let process_duration = self.last_timestamp - self.start_timestamp;
+            println!(
+                "Process duration: {} [Days] ({} sec)",
+                process_duration / SEC(60 * 60 * 24),
+                process_duration / 1_000_000
+            );
 
-                    continue;
+            let elapsed = (NOW() - loop_start_time) / 1_000_000;
+            println!("Elapsed time: {} sec", elapsed);
+            println!("Loop count: {}", self.loop_count);
+            if 0 < elapsed {
+                println!("Loop per sec: {}", self.loop_count / elapsed);
+            }
+            println!("on_tick count: {}", self.on_tick_count);
+            println!("on_clock count: {}", self.on_clock_count);
+            println!("on_update count: {}", self.on_update_count);
+            println!("on_account_update count: {}", self.on_account_update_count);
+        }
+    }
+    /*
+        pub fn run(
+            &mut self,
+            exchange: PyObject,
+            market: PyObject,
+            receiver: &impl Stream<Item = anyhow::Result<MarketMessage>>,
+            // receiver: &Receiver<MarketMessage>,
+            agent: &PyAny,
+            log_memory: bool,
+            log_file: Option<String>,
+        ) -> Result<Py<Session>, PyErr> {
+            if self.verbose {
+                print!("--- run {:?} mode --- ", self.execute_mode);
+                print!("market: {}, ", self.exchange_name);
+                print!("agent_id: {}, ", self.agent_id);
+                print!("log_memory: {}, ", log_memory);
+                println!("duration: {}[sec], ", self.execute_time);
+
+                flush_log();
+            }
+
+            let result = Python::with_gil(|py| {
+                if self.verbose {
+                    println!("--- run loop ---");
+                    flush_log();
                 }
 
-                // after warm up loop, we call agent(in the method session is updated)
-                self.loop_count += 1;
+                let session_name = self.agent_id.clone();
+                let mut session = Session::new(
+                    exchange,
+                    market,
+                    self.execute_mode.clone(),
+                    Some(&session_name),
+                    log_memory,
+                );
 
-                let r = self.on_message(&py, agent, &py_session, &message, interval_sec);
-                if r.is_err() {
-                    return Err(r.unwrap_err());
+                if log_file.is_some() {
+                    let log_file = log_file.unwrap();
+
+                    if session.open_log(&log_file).is_err() {
+                        log::error!("Failed to open log file: {}", &log_file);
+                        println!("Failed to open log file: {}", &log_file);
+                    };
+                } else if !log_memory {
+                    log::error!(
+                        "log_memory and log_file are both false. Please set one of them to true."
+                    );
+                    println!("WARNING: log_memory and log_file are both false. Please set one of them to true.");
                 }
 
-                // break if the running time exceeceds the loop_duration
-                if self.start_timestamp == 0 {
-                    self.start_timestamp = self.last_timestamp;
-                    if self.verbose {
-                        println!("start_timestamp: {}", time_string(self.start_timestamp));
-                    }
-                } else if 0 < self.execute_time {
-                    remain_time =
-                        self.start_timestamp + SEC(self.execute_time) - self.last_timestamp;
+                let py_session = Py::new(py, session).unwrap();
+                let mut warm_up_loop: i64 = WARMUP_STEPS;
+
+                if self.has_on_init {
+                    self.call_agent_on_init_py(&py, agent, &py_session)?;
+                }
+
+                let interval_sec = self.get_clock_interval(&py, &py_session)?;
+
+                if self.has_on_clock && interval_sec == 0 {
+                    print!("*** WARNING: on_clock is set but clock_interval is 0. (on_clock will not be called ***");
+                }
+
+                self.reset_count();
+                let loop_start_time = NOW();
+
+                let mut remain_time: i64 = 0;
+
+                loop {
+                    let message = receiver.recv();
 
-                    if remain_time < 0 {
-                        log::debug!("remain_time out break");
+                    if message.is_err() {
+                        log::info!("Data stream is closed {:?}", message);
                         break;
                     }
+
+                    let message = message.unwrap();
+                    if self.execute_mode == ExecuteMode::BackTest {
+                        if let MarketMessage::Trade(trade) = &message {
+                            if trade.status == LogStatus::UnFix {
+                                log::info!("UnFix trade is found. Stop running.");
+                                break;
+                            }
+                        }
+                    }
+
+                    if 0 < warm_up_loop {
+                        // in warm up loop, we don't call agent, just update session
+                        let mut session = py_session.borrow_mut(py);
+                        session.on_message(&message);
+                        self.last_timestamp = session.get_timestamp();
+
+                        if let MarketMessage::Trade(_trade) = &message {
+                            warm_up_loop = warm_up_loop - 1;
+                        }
+
+                        if self.verbose {
+                            print!("\rwarm up loop: {}", warm_up_loop)
+                        }
+
+                        drop(session);
+
+                        continue;
+                    }
+
+                    // after warm up loop, we call agent(in the method session is updated)
+                    self.loop_count += 1;
+
+                    let r = self.on_message(&py, agent, &py_session, &message, interval_sec);
+                    if r.is_err() {
+                        return Err(r.unwrap_err());
+                    }
+
+                    // break if the running time exceeceds the loop_duration
+                    if self.start_timestamp == 0 {
+                        self.start_timestamp = self.last_timestamp;
+                        if self.verbose {
+                            println!("start_timestamp: {}", time_string(self.start_timestamp));
+                        }
+                    } else if 0 < self.execute_time {
+                        remain_time =
+                            self.start_timestamp + SEC(self.execute_time) - self.last_timestamp;
+
+                        if remain_time < 0 {
+                            log::debug!("remain_time out break");
+                            break;
+                        }
+                    }
+
+                    // print progress if verbose flag is set.
+                    if self.verbose {
+                        self.print_progress(remain_time);
+                    }
                 }
 
-                // print progress if verbose flag is set.
                 if self.verbose {
-                    self.print_progress(remain_time);
+                    println!(
+                        "\n--- end run --- {} ({}[rec])",
+                        time_string(self.last_timestamp),
+                        self.loop_count
+                    );
+
+                    println!("Done running agent");
+
+                    let process_duration = self.last_timestamp - self.start_timestamp;
+                    println!(
+                        "Process duration: {} [Days] ({} sec)",
+                        process_duration / SEC(60 * 60 * 24),
+                        process_duration / 1_000_000
+                    );
+
+                    let elapsed = (NOW() - loop_start_time) / 1_000_000;
+                    println!("Elapsed time: {} sec", elapsed);
+                    println!("Loop count: {}", self.loop_count);
+                    if 0 < elapsed {
+                        println!("Loop per sec: {}", self.loop_count / elapsed);
+                    }
+                    println!("on_tick count: {}", self.on_tick_count);
+                    println!("on_clock count: {}", self.on_clock_count);
+                    println!("on_update count: {}", self.on_update_count);
+                    println!("on_account_update count: {}", self.on_account_update_count);
+                }
+
+                Ok(py_session)
+            });
+
+            result
+        }
+    */
+    /*
+    pub fn execute_message(
+        &mut self,
+        py  : &Python,
+        agent: &PyAny,
+        py_session: &Py<Session>,
+        interval_sec: i64,
+        message: MarketMessage,
+        warm_up: bool,
+    ) -> anyhow::Result<()> {
+        if self.execute_mode == ExecuteMode::BackTest {
+            if let MarketMessage::Trade(trade) = &message {
+                if trade.status == LogStatus::UnFix {
+                    log::info!("UnFix trade is found. Stop running.");
+                    return Err(anyhow::anyhow!("UnFix trade is found. Stop running."));
                 }
             }
+        }
+
+        if warm_up {
+            // in warm up loop, we don't call agent, just update session
+            let mut session = py_session.borrow_mut(py);
+            session.on_message(&message);
+            self.last_timestamp = session.get_timestamp();
+
+            if let MarketMessage::Trade(_trade) = &message {
+                warm_up_loop = warm_up_loop - 1;
+            }
 
             if self.verbose {
-                println!(
-                    "\n--- end run --- {} ({}[rec])",
-                    time_string(self.last_timestamp),
-                    self.loop_count
-                );
+                print!("\rwarm up loop: {}", warm_up_loop)
+            }
 
-                println!("Done running agent");
+            drop(session);
 
-                let process_duration = self.last_timestamp - self.start_timestamp;
-                println!(
-                    "Process duration: {} [Days] ({} sec)",
-                    process_duration / SEC(60 * 60 * 24),
-                    process_duration / 1_000_000
-                );
+            continue;
+        }
 
-                let elapsed = (NOW() - loop_start_time) / 1_000_000;
-                println!("Elapsed time: {} sec", elapsed);
-                println!("Loop count: {}", self.loop_count);
-                if 0 < elapsed {
-                    println!("Loop per sec: {}", self.loop_count / elapsed);
-                }
-                println!("on_tick count: {}", self.on_tick_count);
-                println!("on_clock count: {}", self.on_clock_count);
-                println!("on_update count: {}", self.on_update_count);
-                println!("on_account_update count: {}", self.on_account_update_count);
+        // after warm up loop, we call agent(in the method session is updated)
+        self.loop_count += 1;
+
+        let r = self.on_message(&py, agent, &py_session, &message, interval_sec);
+        if r.is_err() {
+            return Err(r.unwrap_err());
+        }
+
+        // break if the running time exceeceds the loop_duration
+        if self.start_timestamp == 0 {
+            self.start_timestamp = self.last_timestamp;
+            if self.verbose {
+                println!("start_timestamp: {}", time_string(self.start_timestamp));
             }
+        } else if 0 < self.execute_time {
+            remain_time =
+                self.start_timestamp + SEC(self.execute_time) - self.last_timestamp;
+
+            if remain_time < 0 {
+                log::debug!("remain_time out break");
+                break;
+            }
+        }
+
+        // print progress if verbose flag is set.
+        if self.verbose {
+            self.print_progress(remain_time);
+        }
 
-            Ok(py_session)
-        });
 
-        result
     }
+    */
 
     pub fn print_progress(&mut self, remain_time: i64) -> bool {
         if self.last_timestamp - self.last_print_tick_time < self.print_interval
@@ -661,7 +1052,9 @@ impl Runner {
         py_session: &Py<Session>,
         message: &MarketMessage,
         interval_sec: i64,
-    ) -> Result<(), PyErr> {
+    ) -> anyhow::Result<()> {
+        let config = self.config.clone();
+
         // on clockはSession更新前に呼ぶ
         // こうすることでsession.curent_timestampより先の値でon_clockが呼ばれる.
         // これは、on_clockが呼ばれた時点で、ohlcの更新が終わっていることを保証するため.
@@ -699,7 +1092,9 @@ impl Runner {
                 // IN Real run, account message is from user stream.
                 // AccountUpdateはFilledかPartiallyFilledのみ発生。
                 if self.has_account_update {
-                    self.call_agent_on_account_update(py, agent, py_session, account)?;
+                    let account_pair = account.extract_pair(&config);
+
+                    self.call_agent_on_account_update(py, agent, py_session, &account_pair)?;
                 }
             }
             _ => {
@@ -742,7 +1137,7 @@ impl Runner {
         session.update_psudo_account_by_order(order)
     }
 
-    fn get_clock_interval(
+    fn get_clock_interval_py(
         self: &mut Self,
         py: &Python,
         py_session: &Py<Session>,
@@ -753,7 +1148,16 @@ impl Runner {
         Ok(interval_sec)
     }
 
-    fn call_agent_on_init(
+    fn get_clock_interval(self: &mut Self, py_session: &Py<Session>) -> Result<i64, PyErr> {
+        Python::with_gil(|py| {
+            let session = py_session.borrow_mut(py);
+            let interval_sec = session.get_clock_interval_sec();
+
+            Ok(interval_sec)
+        })
+    }
+
+    fn call_agent_on_init_py(
         self: &mut Self,
         py: &Python,
         agent: &PyAny,
@@ -766,6 +1170,23 @@ impl Runner {
         Ok(())
     }
 
+    fn call_agent_on_init(
+        self: &mut Self,
+        agent: &PyAny,
+        py_session: &Py<Session>,
+    ) -> anyhow::Result<()> {
+        if !self.has_on_init {
+            return Ok(());
+        }
+
+        Python::with_gil(|py| {
+            let session = py_session.borrow_mut(py);
+            agent.call_method1("on_init", (session,))
+        })?;
+
+        Ok(())
+    }
+
     fn call_agent_on_tick(
         self: &mut Self,
         py: &Python,
@@ -820,7 +1241,7 @@ impl Runner {
         py: &Python,
         agent: &PyAny,
         py_session: &Py<Session>,
-        account: &AccountStatus,
+        account: &AccountPair,
     ) -> Result<(), PyErr> {
         let mut session = py_session.borrow_mut(*py);
         session.set_real_account(account);
@@ -871,13 +1292,12 @@ impl Runner {
         })
     }
 
-    /// TODO: return error
-    pub fn open_backtest_stream(
+    pub async fn open_backtest_stream(
         market: &PyObject,
         time_from: MicroSec,
         time_to: MicroSec,
-    ) -> MarketStream {
-        Python::with_gil(|py| {
+    ) -> impl Stream<Item = anyhow::Result<MarketMessage>> {
+        let market_stream = Python::with_gil(|py| {
             let stream = market.call_method1(py, "open_backtest_channel", (time_from, time_to));
 
             if stream.is_ok() {
@@ -889,7 +1309,17 @@ impl Runner {
                 log::error!("Error in open_backtest_channel: {:?}", err);
                 panic!("Error in open_backtest_channel: {:?}", err);
             }
-        })
+        });
+
+        let market_stream = market_stream.reciver;
+
+        stream! {
+            loop {
+                let message = market_stream.recv()?;
+
+                yield Ok(message);
+            }
+        }
     }
 
     // base 64 generated by co pilot!!
diff --git a/rbot_session/src/session.rs b/rbot_session/src/session.rs
index 27e6c0ff..a69068fc 100644
--- a/rbot_session/src/session.rs
+++ b/rbot_session/src/session.rs
@@ -5,16 +5,18 @@ use std::sync::Mutex;
 
 use pyo3::{pyclass, pymethods, PyAny, PyObject, Python};
 
+use pyo3_polars::PyDataFrame;
 use rust_decimal::{prelude::ToPrimitive, Decimal};
 use rust_decimal_macros::dec;
 
 use super::{Logger, OrderList};
 use rbot_lib::common::{
-    date_string, hour_string, min_string, time_string, AccountStatus, MarketConfig, MicroSec,
-    OrderSide, OrderStatus, NOW,Trade, MarketMessage, SEC,Order,OrderType
+    date_string, get_orderbook_df, hour_string, min_string, time_string, AccountPair, MarketConfig, MarketMessage, MicroSec, Order, OrderBookList, OrderSide, OrderStatus, OrderType, Trade, NOW, SEC
 };
 use pyo3::prelude::*;
 
+use anyhow;
+
 #[derive(Debug, Clone, PartialEq)]
 #[pyclass]
 pub enum ExecuteMode {
@@ -53,8 +55,9 @@ pub struct Session {
     execute_mode: ExecuteMode,
     buy_orders: OrderList,
     sell_orders: OrderList,
-    real_account: AccountStatus,
-    psudo_account: AccountStatus,
+    real_account: AccountPair,
+    psudo_account: AccountPair,
+    exchange: PyObject,
     market: PyObject,
     current_timestamp: MicroSec,
     current_clock_time: MicroSec,
@@ -82,6 +85,8 @@ pub struct Session {
     asks_edge: Decimal,
     bids_edge: Decimal,
 
+    exchange_name: String,
+    trade_category: String,
     market_config: MarketConfig,
 
     dummy_q: Mutex<VecDeque<Vec<Order>>>,
@@ -92,8 +97,9 @@ pub struct Session {
 #[pymethods]
 impl Session {
     #[new]
-    #[pyo3(signature = (market, execute_mode, session_name=None, log_memory=true))]
+    #[pyo3(signature = (exchange, market, execute_mode, session_name=None, log_memory=true))]
     pub fn new(
+        exchange: PyObject,
         market: PyObject,
         execute_mode: ExecuteMode,
         session_name: Option<&str>,
@@ -120,12 +126,22 @@ impl Session {
             config
         });
 
+        let exchange_name = Python::with_gil(|py| {
+            let exchange_name = exchange.getattr(py, "exchange_name").unwrap();
+            let exchange_name: String = exchange_name.extract(py).unwrap();
+
+            exchange_name
+        });
+
+        let category = config.trade_category.clone();
+
         let mut session = Self {
             execute_mode: execute_mode,
             buy_orders: OrderList::new(OrderSide::Buy),
             sell_orders: OrderList::new(OrderSide::Sell),
-            real_account: AccountStatus::default(),
-            psudo_account: AccountStatus::default(),
+            real_account: AccountPair::default(),
+            psudo_account: AccountPair::default(),
+            exchange,
             market,
             current_timestamp: 0,
             current_clock_time: 0,
@@ -153,6 +169,8 @@ impl Session {
             asks_edge: dec![0.0],
             bids_edge: dec![0.0],
 
+            exchange_name,
+            trade_category: category,
             market_config: config,
 
             dummy_q: Mutex::new(VecDeque::new()),
@@ -201,13 +219,11 @@ impl Session {
 
     #[getter]
     // TODO: リモート動作時にRESTでコールする。
-    pub fn get_board(&self) -> Result<Py<PyAny>, PyErr> {
-        if self.execute_mode == ExecuteMode::BackTest {
-            return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(
-                "get_board is not supported in backtest mode",
-            ));
-        }
-        Python::with_gil(|py| self.market.getattr(py, "board"))
+    pub fn get_board(&self) -> anyhow::Result<(PyDataFrame, PyDataFrame)> {
+        let path = OrderBookList::make_path(&self.exchange_name, &self.market_config);
+        let (bids, asks) = get_orderbook_df(&path)?;
+
+        Ok((PyDataFrame(bids), PyDataFrame(asks)))
     }
 
     #[getter]
@@ -289,17 +305,17 @@ impl Session {
     }
 
     #[getter]
-    pub fn get_psudo_account(&self) -> AccountStatus {
+    pub fn get_psudo_account(&self) -> AccountPair {
         self.psudo_account.clone()
     }
 
     #[getter]
-    pub fn get_real_account(&self) -> AccountStatus {
+    pub fn get_real_account(&self) -> AccountPair {
         self.real_account.clone()
     }
 
     #[getter]
-    pub fn get_account(&self) -> AccountStatus {
+    pub fn get_account(&self) -> AccountPair {
         match self.execute_mode {
             ExecuteMode::Real => self.real_account.clone(),
             ExecuteMode::BackTest => self.psudo_account.clone(),
@@ -365,7 +381,8 @@ impl Session {
     /// if fail return None
     pub fn real_cancel_order(&mut self, order_id: &str) -> PyResult<Py<PyAny>> {
         Python::with_gil(|py| {
-            let r = self.market.call_method1(py, "cancel_order", (order_id,));
+            let r = 
+                self.exchange.call_method1(py, "cancel_order", (self.market_config.clone(), order_id,));
 
             if r.is_err() {
                 let none = Python::None(py);
@@ -413,8 +430,8 @@ impl Session {
         let local_id = self.new_order_id();
 
         Python::with_gil(|py| {
-            self.market
-                .call_method1(py, "market_order", (side, size, local_id))
+            self.exchange
+                .call_method1(py, "market_order", (self.market_config.clone(), side, size, local_id))
         })
     }
 
@@ -473,6 +490,7 @@ impl Session {
         let execute_price = self.calc_dummy_execute_price_by_slip(order_side);
 
         let mut order = Order::new(
+            &self.trade_category,
             &self.market_config.trade_symbol,
             self.calc_log_timestamp(),
             &local_id,
@@ -539,8 +557,8 @@ impl Session {
         // then call market.limit_order
         let r = Python::with_gil(|py| {
             let result =
-                self.market
-                    .call_method1(py, "limit_order", (side, pricedp, sizedp, local_id));
+                self.exchange
+                    .call_method1(py, "limit_order", (self.market_config.clone(), side, pricedp, sizedp, local_id));
 
             match result {
                 // if success update order list
@@ -599,6 +617,7 @@ impl Session {
         );
 
         let mut order = Order::new(
+            &self.trade_category,
             &self.market_config.trade_symbol,
             self.calc_log_timestamp(),
             &local_id,
@@ -680,6 +699,7 @@ impl Session {
 
 impl Session {
     pub fn on_message(&mut self, message: &MarketMessage) -> Vec<Order> {
+        let config = self.market_config.clone();
         let mut new_orders = vec![];
 
         match message {
@@ -705,9 +725,11 @@ impl Session {
                 log::debug!("on_message: order={:?}", order);
                 self.on_order_update(&mut order);
             }
-            MarketMessage::Account(account) => {
-                log::debug!("on_message: account={:?}", account);
-                self.on_account_update(account);
+            MarketMessage::Account(coins) => {
+                log::debug!("on_message: account={:?}", coins);
+
+                let mut account: AccountPair = coins.extract_pair(&config);
+                self.on_account_update(&account);
             }
             MarketMessage::Orderbook(orderbook) => {
                 log::warn!("IGNORED MESSAGE: on_message: orderbook={:?}", orderbook);
@@ -761,7 +783,7 @@ impl Session {
         )
     }
 
-    pub fn log_account(&mut self, account: &AccountStatus) -> Result<(), std::io::Error> {
+    pub fn log_account(&mut self, account: &AccountPair) -> Result<(), std::io::Error> {
         let time = self.calc_log_timestamp();
 
         self.log.log_account(time, account)
@@ -800,7 +822,7 @@ impl Session {
         }
     }
 
-    pub fn on_account_update(&mut self, account: &AccountStatus) {
+    pub fn on_account_update(&mut self, account: &AccountPair) {
         self.real_account = account.clone();
 
         if self.log_account(account).is_err() {
@@ -848,8 +870,10 @@ impl Session {
             return Ok(());
         }
 
+        let config = self.market_config.clone();
+
         let r = Python::with_gil(|py| {
-            let result = self.market.getattr(py, "open_orders");
+            let result = self.exchange.call_method1(py, "get_open_orders", (config.clone(),));
 
             match result {
                 // if success update order list
@@ -858,6 +882,10 @@ impl Session {
                     log::debug!("OpenOrders {:?}", orderlist);
 
                     for order in orders {
+                        if order.symbol != config.trade_symbol {
+                            continue;
+                        }
+
                         log::debug!("OpenOrder {:?}", order);
                         if order.order_side == OrderSide::Buy {
                             self.buy_orders.update_or_insert(&order);
@@ -1059,7 +1087,7 @@ impl Session {
         }
     }
 
-    pub fn set_real_account(&mut self, account: &AccountStatus) {
+    pub fn set_real_account(&mut self, account: &AccountPair) {
         self.real_account = account.clone();
     }
 }


Test patch:
diff --git a/rbot/test/bybit_test/bybit.ipynb b/rbot/test/bybit_test/bybit.ipynb
index 1819d99a..9787e684 100644
--- a/rbot/test/bybit_test/bybit.ipynb
+++ b/rbot/test/bybit_test/bybit.ipynb
@@ -8,9 +8,12 @@
    "source": [
     "from json2html import *\n",
     "from IPython.display import HTML\n",
+    "from rbot import init_debug_log\n",
+    "\n",
+    "init_debug_log()\n",
     "\n",
     "from rbot import Bybit\n",
-    "bybit = Bybit(True)\n"
+    "bybit = Bybit(False)\n"
    ]
   },
   {
@@ -26,6 +29,18 @@
     "orders = bybit.get_open_orders(config)"
    ]
   },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from rbot import init_debug_log\n",
+    "init_debug_log()\n",
+    "\n",
+    "bybit.start_user_stream()"
+   ]
+  },
   {
    "cell_type": "code",
    "execution_count": null,
@@ -60,7 +75,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "order = bybit.limit_order(config, 'Buy', 40000, 0.001)"
+    "order = bybit.limit_order(config, side='Buy', price=40000, size=0.001)"
    ]
   },
   {
@@ -78,7 +93,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "r = bybit.cancel_order(config, order[0].order_id)"
+    "r = bybit.cancel_order(config, order_id=order[0].order_id)"
    ]
   },
   {
@@ -86,9 +101,7 @@
    "execution_count": null,
    "metadata": {},
    "outputs": [],
-   "source": [
-    "r"
-   ]
+   "source": []
   },
   {
    "cell_type": "code",
@@ -138,7 +151,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "market.download(ndays=4, force=False, verbose=True, archive_only=True, low_priority=False)"
+    "market.download_archive(ndays=4, force=False, verbose=True, low_priority=False)"
    ]
   },
   {
@@ -155,7 +168,9 @@
    "execution_count": null,
    "metadata": {},
    "outputs": [],
-   "source": []
+   "source": [
+    "market.download_gap(verbose=True)"
+   ]
   },
   {
    "cell_type": "code",
diff --git a/rbot/test/bybit_test/market.ipynb b/rbot/test/bybit_test/market.ipynb
index b1164d21..62ccee66 100644
--- a/rbot/test/bybit_test/market.ipynb
+++ b/rbot/test/bybit_test/market.ipynb
@@ -48,6 +48,28 @@
     "\n"
    ]
   },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from rbot import get_orderbook_list\n",
+    "\n",
+    "get_orderbook_list()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from rbot import get_orderbook\n",
+    "\n",
+    "get_orderbook(\"BYBIT/linear/BTCUSDT\")"
+   ]
+  },
   {
    "cell_type": "code",
    "execution_count": null,
diff --git a/rbot/test/bybit_test/nullagent.py b/rbot/test/bybit_test/nullagent.py
index c9a45846..aaecc21c 100644
--- a/rbot/test/bybit_test/nullagent.py
+++ b/rbot/test/bybit_test/nullagent.py
@@ -59,7 +59,8 @@ def on_update(self, session, updated_order):
 
 
 
-bybit  = Bybit(testnet=True)
+bybit  = Bybit(production=False)
+bybit.enable_order_with_my_own_risk = True
 
 market = bybit.open_market(BybitConfig.BTCUSDT)
 
@@ -73,12 +74,13 @@ def on_update(self, session, updated_order):
 init_debug_log()
 
 session = runner.real_run(
+                exchange=bybit,
                 market=market,
                 agent=agent, 
 #                execute_time = 10,
                 verbose=True,
                 log_file="skelton_bot.log",
-                client= True
+                client= False
             )
 
     
diff --git a/rbot/test/manual/session.ipynb b/rbot/test/manual/session.ipynb
index 3190fd2a..efc60a02 100644
--- a/rbot/test/manual/session.ipynb
+++ b/rbot/test/manual/session.ipynb
@@ -157,17 +157,11 @@
    "source": [
     "from rbot import Session\n",
     "from rbot import Runner\n",
-    "from rbot import BinanceMarket\n",
-    "from rbot import BinanceConfig\n",
+    "from rbot import Bybit\n",
+    "from rbot import BybitConfig\n",
     "from rbot import ExecuteMode\n",
     "from rbot import NOW, HHMM, SEC, time_string\n",
     "\n",
-    "\n",
-    "#from rbot import MarketMessage\n",
-    "\n",
-    "#from rbot import Trade\n",
-    "#from rbot import OrderSide\n",
-    "\n",
     "from rbot import init_log"
    ]
   },
@@ -189,12 +183,98 @@
    "outputs": [],
    "source": [
     "\n",
-    "config = BinanceConfig.BTCUSDT\n",
-    "market = BinanceMarket(config)\n",
+    "\n",
+    "config = BybitConfig.BTCUSDT\n",
+    "\n",
+    "bybit = Bybit(False)\n",
+    "bybit.enable_order_with_my_own_risk = True\n",
+    "\n",
+    "market = bybit.open_market(config)\n",
     "\n",
     "\n"
    ]
   },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "market.start_market_stream()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "session = Session(bybit, market, ExecuteMode.Real)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "session.board"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "session.market_order(\"buy\", 0.001)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "session.limit_order(\"Buy\", 45000, 0.001)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "session.buy_orders"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "session.sell_orders"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "session.cancel_order(\"669545c3-3e60-4c29-9360-4e4484cd27b3\")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "session.psudo_account"
+   ]
+  },
   {
    "cell_type": "code",
    "execution_count": null,
diff --git a/rbot_session/src/mod_test.rs b/rbot_session/src/mod_test.rs
index 883656c0..97491b8f 100644
--- a/rbot_session/src/mod_test.rs
+++ b/rbot_session/src/mod_test.rs
@@ -14,6 +14,7 @@ mod tests {
         let now = NOW();
 
         let order1 = Order::new(
+            "linear",
             "BTCUSDT",
             now,
             "MYORDER-1",
@@ -107,6 +108,7 @@ mod tests {
         let now = NOW();
 
         let order1 = Order::new(
+            "linear",
             "BTCUSDT",
             now,
             "MYORDER-1",