binary_option_tools_core/pocketoption/
pocket_client.rs

1use std::{
2    collections::HashMap,
3    time::{Duration, Instant},
4};
5
6use chrono::{DateTime, Utc};
7use tracing::{debug, info, warn};
8use uuid::Uuid;
9
10use crate::{
11    contstants::TIMOUT_TIME,
12    error::{BinaryOptionsResult, BinaryOptionsToolsError},
13    general::{client::WebSocketClient, types::Data},
14    pocketoption::{
15        parser::basic::LoadHistoryPeriod,
16        types::order::SuccessCloseOrder,
17        validators::{candle_validator, order_result_validator},
18        ws::ssid::Ssid,
19    },
20};
21
22use super::{
23    error::PocketOptionError,
24    parser::message::WebSocketMessage,
25    types::{
26        base::ChangeSymbol,
27        callback::PocketCallback,
28        data_v2::PocketData,
29        info::MessageInfo,
30        order::{Action, Deal, OpenOrder},
31        update::{DataCandle, UpdateBalance},
32    },
33    validators::{history_validator, order_validator},
34    ws::{connect::PocketConnect, listener::Handler, stream::StreamAsset},
35};
36
37/// Class to connect automatically to Pocket Option's quick trade passing a valid SSID
38pub type PocketOption =
39    WebSocketClient<WebSocketMessage, Handler, PocketConnect, Ssid, PocketData, PocketCallback>;
40
41impl PocketOption {
42    pub async fn new(ssid: impl ToString) -> BinaryOptionsResult<Self> {
43        let ssid = Ssid::parse(ssid)?;
44        let data = Data::new(PocketData::default());
45        let handler = Handler::new(ssid.clone());
46        let timeout = Duration::from_millis(500);
47        let callback = PocketCallback;
48        let client = WebSocketClient::init(
49            ssid,
50            PocketConnect {},
51            data,
52            handler,
53            timeout,
54            Some(callback),
55        )
56        .await?;
57        // println!("Initialized!");
58        Ok(client)
59    }
60
61    pub async fn trade(
62        &self,
63        asset: impl ToString,
64        action: Action,
65        amount: f64,
66        time: u32,
67    ) -> BinaryOptionsResult<(Uuid, Deal)> {
68        let order = OpenOrder::new(
69            amount,
70            asset.to_string(),
71            action,
72            time,
73            self.credentials.demo() as u32,
74        )?;
75        let request_id = order.request_id;
76        let res = self
77            .send_message_with_timout(
78                Duration::from_secs(TIMOUT_TIME),
79                "Trade",
80                WebSocketMessage::OpenOrder(order),
81                MessageInfo::SuccessopenOrder,
82                order_validator(request_id),
83            )
84            .await?;
85        if let WebSocketMessage::SuccessopenOrder(order) = res {
86            debug!("Successfully opened buy trade!");
87            return Ok((order.id, order));
88        }
89        Err(PocketOptionError::UnexpectedIncorrectWebSocketMessage(res.info()).into())
90    }
91
92    pub async fn buy(
93        &self,
94        asset: impl ToString,
95        amount: f64,
96        time: u32,
97    ) -> BinaryOptionsResult<(Uuid, Deal)> {
98        info!(target: "Buy", "Placing a buy trade for asset '{}', with amount '{}' and time '{}'", asset.to_string(), amount, time);
99        self.trade(asset, Action::Call, amount, time).await
100    }
101
102    pub async fn sell(
103        &self,
104        asset: impl ToString,
105        amount: f64,
106        time: u32,
107    ) -> BinaryOptionsResult<(Uuid, Deal)> {
108        info!(target: "Sell", "Placing a sell trade for asset '{}', with amount '{}' and time '{}'", asset.to_string(), amount, time);
109        self.trade(asset, Action::Put, amount, time).await
110    }
111
112    pub async fn get_deal_end_time(&self, id: Uuid) -> Option<DateTime<Utc>> {
113        if let Some(trade) = self
114            .data
115            .get_opened_deals()
116            .await
117            .iter()
118            .find(|d| *d == &id)
119        {
120            return Some(trade.close_timestamp - Duration::from_secs(2 * 3600)); // Pocket Option server seems 2 hours advanced
121        }
122
123        if let Some(trade) = self
124            .data
125            .get_opened_deals()
126            .await
127            .iter()
128            .find(|d| *d == &id)
129        {
130            return Some(trade.close_timestamp - Duration::from_secs(2 * 3600)); // Pocket Option server seems 2 hours advanced
131        }
132        None
133    }
134
135    pub async fn check_results(&self, trade_id: Uuid) -> BinaryOptionsResult<Deal> {
136        // TODO: Add verification so it doesn't try to wait if no trade has been made with that id
137
138        info!(target: "CheckResults", "Checking results for trade of id {}", trade_id);
139        if let Some(trade) = self
140            .data
141            .get_closed_deals()
142            .await
143            .iter()
144            .find(|d| d.id == trade_id)
145        {
146            return Ok(trade.clone());
147        }
148        debug!("Trade result not found in closed deals list, waiting for closing order to check.");
149        if let Some(timestamp) = self.get_deal_end_time(trade_id).await {
150            let exp = timestamp
151                .signed_duration_since(Utc::now()) // TODO: Change this since the current time depends on the timezone.
152                .to_std()?;
153            debug!(target: "CheckResult", "Expiration time in {exp:?} seconds.");
154            let start = Instant::now();
155            // println!("Expiration time in {exp:?} seconds.");
156            let res: WebSocketMessage = match self
157                .send_message_with_timeout_and_retry(
158                    exp + Duration::from_secs(TIMOUT_TIME),
159                    "CheckResult",
160                    WebSocketMessage::None,
161                    MessageInfo::SuccesscloseOrder,
162                    order_result_validator(trade_id),
163                )
164                .await
165            {
166                Ok(msg) => msg,
167                Err(e) => {
168                    info!(target: "CheckResults", "Time elapsed, {:?}, checking closed deals one last time.", start.elapsed());
169                    if let Some(deal) = self
170                        .get_closed_deals()
171                        .await
172                        .iter()
173                        .find(|d| d.id == trade_id)
174                    {
175                        WebSocketMessage::SuccesscloseOrder(SuccessCloseOrder {
176                            profit: 0.0,
177                            deals: vec![deal.to_owned()],
178                        })
179                    } else {
180                        return Err(e);
181                    }
182                }
183            };
184
185            if let WebSocketMessage::SuccesscloseOrder(order) = res {
186                return order
187                    .deals
188                    .iter()
189                    .find(|d| d.id == trade_id)
190                    .cloned()
191                    .ok_or(
192                        PocketOptionError::UnreachableError("Error finding correct trade".into())
193                            .into(),
194                    );
195            }
196            return Err(PocketOptionError::UnexpectedIncorrectWebSocketMessage(res.info()).into());
197        }
198        warn!("No opened trade with the given uuid please check if you are passing the correct id");
199        Err(BinaryOptionsToolsError::Unallowed("Couldn't check result for a deal that is not in the list of opened trades nor closed trades.".into()))
200    }
201
202    pub async fn get_candles(
203        &self,
204        asset: impl ToString,
205        period: i64,
206        offset: i64,
207    ) -> BinaryOptionsResult<Vec<DataCandle>> {
208        info!(target: "GetCandles", "Retrieving candles for asset '{}' with period of '{}' and offset of '{}'", asset.to_string(), period, offset);
209        let time = self.data.get_server_time().await.div_euclid(period) * period;
210        if time == 0 {
211            return Err(BinaryOptionsToolsError::GeneralParsingError(
212                "Server time is invalid.".to_string(),
213            ));
214        }
215        let request = LoadHistoryPeriod::new(asset.to_string(), time, period, offset)?;
216        let index = request.index;
217        debug!(
218            "Sent get candles message, message: {:?}",
219            WebSocketMessage::GetCandles(request).to_string()
220        );
221        let request = LoadHistoryPeriod::new(asset.to_string(), time, period, offset)?;
222        let res = self
223            .send_message_with_timeout_and_retry(
224                Duration::from_secs(TIMOUT_TIME),
225                "GetCandles",
226                WebSocketMessage::GetCandles(request),
227                MessageInfo::LoadHistoryPeriod,
228                candle_validator(index),
229            )
230            .await?;
231        if let WebSocketMessage::LoadHistoryPeriod(history) = res {
232            return Ok(history.candle_data());
233        }
234        Err(PocketOptionError::UnexpectedIncorrectWebSocketMessage(res.info()).into())
235    }
236
237    pub async fn history(
238        &self,
239        asset: impl ToString,
240        period: i64,
241    ) -> BinaryOptionsResult<Vec<DataCandle>> {
242        info!(target: "History", "Retrieving candles for asset '{}' with period of '{}'", asset.to_string(), period);
243
244        let request = ChangeSymbol::new(asset.to_string(), period);
245        let res = self
246            .send_message_with_timeout_and_retry(
247                Duration::from_secs(TIMOUT_TIME),
248                "History",
249                WebSocketMessage::ChangeSymbol(request),
250                MessageInfo::UpdateHistoryNew,
251                history_validator(asset.to_string(), period),
252            )
253            .await?;
254        if let WebSocketMessage::UpdateHistoryNew(history) = res {
255            return Ok(history.candle_data());
256        }
257        Err(PocketOptionError::UnexpectedIncorrectWebSocketMessage(res.info()).into())
258    }
259
260    pub async fn get_closed_deals(&self) -> Vec<Deal> {
261        info!(target: "GetClosedDeals", "Retrieving list of closed deals");
262        self.data.get_closed_deals().await
263    }
264
265    pub async fn clear_closed_deals(&self) {
266        info!(target: "ClearClosedDeals", "Clearing list of closed deals");
267        self.data.clean_closed_deals().await
268    }
269
270    pub async fn get_opened_deals(&self) -> Vec<Deal> {
271        info!(target: "GetOpenDeals", "Retrieving list of open deals");
272        self.data.get_opened_deals().await
273    }
274
275    pub async fn get_balance(&self) -> UpdateBalance {
276        info!(target: "GetBalance", "Retrieving account balance");
277        self.data.get_balance().await
278    }
279
280    pub async fn get_payout(&self) -> HashMap<String, i32> {
281        info!(target: "GetPayout", "Retrieving payout for all the assets");
282        self.data.get_full_payout().await
283    }
284
285    pub async fn subscribe_symbol(&self, asset: impl ToString) -> BinaryOptionsResult<StreamAsset> {
286        info!(target: "SubscribeSymbol", "Subscribing to asset '{}'", asset.to_string());
287        let _ = self.history(asset.to_string(), 1).await?;
288        debug!("Created StreamAsset instance.");
289        Ok(self.data.add_stream(asset.to_string()).await)
290    }
291
292    pub async fn subscribe_symbol_chuncked(
293        &self,
294        asset: impl ToString,
295        chunck_size: impl Into<usize>,
296    ) -> BinaryOptionsResult<StreamAsset> {
297        info!(target: "SubscribeSymbolChuncked", "Subscribing to asset '{}'", asset.to_string());
298        let _ = self.history(asset.to_string(), 1).await?;
299        debug!("Created StreamAsset instance.");
300        Ok(self
301            .data
302            .add_stream_chuncked(asset.to_string(), chunck_size.into())
303            .await)
304    }
305
306    pub fn kill(self) {
307        drop(self)
308    }
309}
310
311#[cfg(test)]
312mod tests {
313    use std::time::Instant;
314
315    use futures_util::{
316        future::{try_join3, try_join_all},
317        StreamExt,
318    };
319    use rand::{random, seq::SliceRandom, thread_rng};
320    use tokio::{task::JoinHandle, time::sleep};
321
322    use crate::utils::{time::timeout, tracing::start_tracing};
323
324    use super::*;
325
326    #[tokio::test]
327    #[should_panic(expected = "MaxDemoTrades")]
328    async fn test_pocket_option() {
329        // start_tracing()?;
330        let ssid = r#"42["auth",{"session":"looc69ct294h546o368s0lct7d","isDemo":1,"uid":87742848,"platform":2}]	"#;
331        let api = PocketOption::new(ssid).await.unwrap();
332        // let mut loops = 0;
333        // while loops < 100 {
334        //     loops += 1;
335        //     sleep(Duration::from_millis(100)).await;
336        // }
337        for i in 0..100 {
338            let now = Instant::now();
339            let _ = api.buy("EURUSD_otc", 1.0, 60).await.expect("MaxDemoTrades");
340            println!("Loop n°{i}, Elapsed time: {:.8?} ms", now.elapsed());
341        }
342    }
343
344    #[tokio::test]
345    async fn test_subscribe_symbol_v2() -> anyhow::Result<()> {
346        start_tracing(true)?;
347        fn to_future(stream: StreamAsset, id: i32) -> JoinHandle<anyhow::Result<()>> {
348            tokio::spawn(async move {
349                while let Some(item) = stream.to_stream().next().await {
350                    info!("StreamAsset n°{}, price: {}", id, item?.close);
351                }
352                Ok(())
353            })
354        }
355        // start_tracing()?;
356        let ssid = r#"42["auth",{"session":"looc69ct294h546o368s0lct7d","isDemo":1,"uid":87742848,"platform":2}]	"#;
357        let client = PocketOption::new(ssid).await?;
358        let stream_asset1 = client.subscribe_symbol("EURUSD_otc").await?;
359        let stream_asset2 = client.subscribe_symbol("#FB_otc").await?;
360        let stream_asset3 = client.subscribe_symbol("YERUSD_otc").await?;
361
362        let f1 = to_future(stream_asset1, 1);
363        let f2 = to_future(stream_asset2, 2);
364        let f3 = to_future(stream_asset3, 3);
365        let _ = try_join3(f1, f2, f3).await?;
366        Ok(())
367    }
368
369    #[tokio::test]
370    async fn test_get_payout() -> anyhow::Result<()> {
371        let ssid = r#"42["auth",{"session":"looc69ct294h546o368s0lct7d","isDemo":1,"uid":87742848,"platform":2}]	"#;
372        let api = PocketOption::new(ssid).await?;
373        tokio::time::sleep(Duration::from_secs(5)).await;
374        dbg!(api.get_payout().await);
375        Ok(())
376    }
377
378    #[tokio::test]
379    async fn test_check_win_v1() -> anyhow::Result<()> {
380        start_tracing(true)?;
381        let ssid = r#"42["auth",{"session":"t0mc6nefcv7ncr21g4fmtioidb","isDemo":1,"uid":90000798,"platform":2}]	"#;
382        let client = PocketOption::new(ssid).await.unwrap();
383        let mut test = 0;
384        let mut checks = Vec::new();
385        while test < 1000 {
386            test += 1;
387            if test % 100 == 0 {
388                let res = client.sell("EURUSD_otc", 1.0, 15).await?;
389                dbg!("Trade id: {}", res.0);
390                let m_client = client.clone();
391                let res: tokio::task::JoinHandle<Result<(), BinaryOptionsToolsError>> =
392                    tokio::spawn(async move {
393                        let result = m_client.check_results(res.0).await?;
394                        dbg!("Trade result: {}", result.profit);
395                        Ok(())
396                    });
397                checks.push(res);
398            } else if test % 100 == 50 {
399                let res = &client.buy("#AAPL_otc", 1.0, 5).await?;
400                dbg!(res);
401            }
402            sleep(Duration::from_millis(100)).await;
403        }
404        try_join_all(checks).await?;
405        Ok(())
406    }
407
408    #[tokio::test]
409    async fn test_check_win_v2() -> anyhow::Result<()> {
410        start_tracing(true)?;
411        let ssid = r#"42["auth",{"session":"t0mc6nefcv7ncr21g4fmtioidb","isDemo":1,"uid":90000798,"platform":2}]	"#;
412        let client = PocketOption::new(ssid).await.unwrap();
413        let times = [5, 15, 30, 60, 300];
414        for time in times {
415            info!("Checkind for an expiration of '{time}' seconds!");
416            let res: Result<(), BinaryOptionsToolsError> =
417                tokio::time::timeout(Duration::from_secs(time as u64 + 30), async {
418                    let (id1, _) = client.buy("EURUSD_otc", 1.5, time).await?;
419                    let (id2, _) = client.sell("EURUSD_otc", 4.2, time).await?;
420                    let r1 = client.check_results(id1).await?;
421                    let r2 = client.check_results(id2).await?;
422                    assert_eq!(r1.id, id1);
423                    assert_eq!(r2.id, id2);
424                    Ok(())
425                })
426                .await?;
427            res?;
428        }
429
430        Ok(())
431    }
432
433    #[tokio::test]
434    async fn test_check_win_v3() -> anyhow::Result<()> {
435        let ssid = r#"42["auth",{"session":"t0mc6nefcv7ncr21g4fmtioidb","isDemo":1,"uid":90000798,"platform":2}]	"#;
436        let client = PocketOption::new(ssid).await.unwrap();
437        let times = [5, 15, 30, 60, 300];
438        let assets = ["#AAPL_otc", "#MSFT_otc", "EURUSD_otc", "YERUSD_otc"];
439        for asset in assets {
440            for time in times {
441                println!("Checkind for an expiration of '{time}' seconds!");
442                let at = tokio::time::Instant::now() + Duration::from_secs(time as u64 + 5);
443                let res: Result<Duration, BinaryOptionsToolsError> =
444                    tokio::time::timeout_at(at, async {
445                        let start = tokio::time::Instant::now();
446                        let (id1, _) = client.buy(asset, 1.5, time).await?;
447                        let (id2, _) = client.sell(asset, 4.2, time).await?;
448                        let r1 = client.check_results(id1).await?;
449                        let r2 = client.check_results(id2).await?;
450                        assert_eq!(r1.id, id1);
451                        assert_eq!(r2.id, id2);
452                        let elapsed = start.elapsed();
453                        Ok(elapsed)
454                    })
455                    .await?;
456                let duration = res?;
457                println!(
458                    "Test passed for expiration of '{time}' seconds in '{:#?}'!",
459                    duration
460                );
461            }
462        }
463
464        Ok(())
465    }
466
467    #[tokio::test]
468    #[should_panic(expected = "CheckResults")]
469    async fn test_timeout() {
470        let ssid = r#"42["auth",{"session":"t0mc6nefcv7ncr21g4fmtioidb","isDemo":1,"uid":90000798,"platform":2}]	"#;
471        let client = PocketOption::new(ssid).await.unwrap();
472        let (id, _) = client.buy("EURUSD_otc", 1.5, 60).await.unwrap();
473        dbg!(&id);
474        let check = client.check_results(id);
475        let res = timeout(Duration::from_secs(30), check, "CheckResults".into())
476            .await
477            .expect("CheckResults");
478        dbg!(res);
479    }
480
481    #[tokio::test]
482    async fn test_buy_check() -> anyhow::Result<()> {
483        start_tracing(false)?;
484        let ssid = r#"42["auth",{"session":"t0mc6nefcv7ncr21g4fmtioidb","isDemo":1,"uid":90000798,"platform":2}]	"#;
485        let client = PocketOption::new(ssid).await.unwrap();
486        let time_frames = [5, 15, 30, 60, 300];
487        let assets = ["EURUSD_otc"];
488        let mut rng = thread_rng();
489        loop {
490            let amount = (random::<f64>() * 10.0).max(1.0);
491            let asset = assets.choose(&mut rng).ok_or(anyhow::anyhow!("Error"))?;
492            let timeframe = time_frames
493                .choose(&mut rng)
494                .ok_or(anyhow::anyhow!("Error"))?;
495            let direction = if random() { Action::Call } else { Action::Put };
496            println!("Placing '{direction:?}' trade on asset '{asset}', amount '{amount}' usd and expiration of '{timeframe}'s.");
497            let (id, _) = client
498                .trade(asset, direction, amount, timeframe.to_owned())
499                .await?;
500            match client.check_results(id).await {
501                Ok(res) => println!("Result for trade: {}", res.profit),
502                Err(e) => eprintln!("Error, {e}\nTime: {}", Utc::now()),
503            }
504        }
505    }
506
507    #[tokio::test]
508    async fn test_server_time() -> anyhow::Result<()> {
509        // start_tracing(true)?;
510        // start_tracing()?;
511        let ssid = r#"42["auth",{"session":"looc69ct294h546o368s0lct7d","isDemo":1,"uid":87742848,"platform":2}]	"#;
512        let client = PocketOption::new(ssid).await?;
513        let stream = client.subscribe_symbol("EURUSD_otc").await?;
514        while let Some(item) = stream.to_stream().next().await {
515            let time = item?.time;
516            let now_test = Utc::now() + Duration::from_secs(2 * 3600);
517            let dif = time - now_test;
518            println!("Difference: {:?}", dif);
519        }
520        Ok(())
521    }
522
523    #[tokio::test]
524    async fn test_get_candles() -> anyhow::Result<()> {
525        let ssid = r#"42["auth",{"session":"t0mc6nefcv7ncr21g4fmtioidb","isDemo":1,"uid":90000798,"platform":2}]	"#;
526        // time: 1733040000, offset: 540000, period: 3600
527        let client = PocketOption::new(ssid).await.unwrap();
528        for i in 0..1000 {
529            let candles = client.get_candles("EURUSD_otc", 60, 6000).await?;
530            println!("Candles n°{} len: {}, ", i + 1, candles.len());
531        }
532        Ok(())
533    }
534
535    #[tokio::test]
536    async fn test_history() -> anyhow::Result<()> {
537        let ssid = r#"42["auth",{"session":"t0mc6nefcv7ncr21g4fmtioidb","isDemo":1,"uid":90000798,"platform":2}]	"#;
538        // time: 1733040000, offset: 540000, period: 3600
539        let client = PocketOption::new(ssid).await.unwrap();
540        for i in 0..1000 {
541            let candles = client.history("EURUSD_otc", 6000).await?;
542            println!("Candles n°{} len: {}, ", i + 1, candles.len());
543        }
544        Ok(())
545    }
546
547}
binary_option_tools_core/pocketoption/pocket_client.rs

binary_option_tools_core/pocketoption/
pocket_client.rs
