avin_trader 0.4.0

Open source cross-platform trading system
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111

/*****************************************************************************
 * URL:         http://avin.info
 * AUTHOR:      Alex Avin
 * E-MAIL:      mr.alexavin@gmail.com
 * LICENSE:     MIT
 ****************************************************************************/

use std::collections::HashMap;

use avin_connect::Tinkoff;
use avin_core::{Action, Asset, Event, TimeFrame, TradeList};
use avin_strategy::{BuySell, Strategy};

use super::work::Work;

pub struct Trader {
    action_tx: tokio::sync::mpsc::UnboundedSender<Action>,
    action_rx: tokio::sync::mpsc::UnboundedReceiver<Action>,
    event_tx: tokio::sync::mpsc::UnboundedSender<Event>,
    event_rx: tokio::sync::mpsc::UnboundedReceiver<Event>,

    works: HashMap<String, tokio::sync::mpsc::UnboundedSender<Event>>,
    trades: TradeList,
}
impl Trader {
    pub fn new() -> Self {
        let (action_tx, action_rx) = tokio::sync::mpsc::unbounded_channel();
        let (event_tx, event_rx) = tokio::sync::mpsc::unbounded_channel();

        Self {
            action_tx,
            action_rx,
            event_tx,
            event_rx,
            works: HashMap::new(),
            trades: TradeList::new("Trader_unittest"),
        }
    }

    pub async fn start(&mut self) {
        log::info!(":: Trader load broker");
        let sender = self.event_tx.clone();
        let mut broker = Tinkoff::new(sender);
        broker.connect().await.unwrap();
        broker.create_marketdata_stream().await.unwrap();
        broker.create_transactions_stream().await.unwrap();
        let broker_tx = broker.get_sender();

        log::info!(":: Trader load assets");
        let mut asset = Asset::new("moex_share_vtbr").unwrap();
        self.load_charts(&mut asset);
        broker
            .subscribe_bar(asset.iid(), &TimeFrame::M1)
            .await
            .unwrap();

        log::info!(":: Trader load strategys");
        let sender = self.action_tx.clone();
        let account = broker.get_account("Agni").await.unwrap();
        let mut strategy = BuySell::default();
        strategy.init(sender, account, &mut asset);

        log::info!(":: Trader start work");
        let mut work = Work::new(asset);
        work.add_strategy(strategy);
        self.works.insert(work.figi().clone(), work.get_sender());
        tokio::spawn(async move { work.start().await });

        log::info!(":: Trader start broker");
        tokio::spawn(async move { start_broker(broker).await });

        log::info!(":: Trader start main loop");
        loop {
            // await events from broker -> send to work (asset & strategy)
            if let Some(e) = self.event_rx.recv().await {
                log::debug!("Trader receive {e}");
                let work = self.works.get(e.figi()).unwrap();
                work.send(e).unwrap();
            };

            // process actions from strategys
            while let Ok(a) = self.action_rx.try_recv() {
                log::debug!("Trader get {a}");
                match a {
                    Action::TradeClosed(trade) => {
                        self.trades.add(trade);
                    }
                    other => broker_tx.send(other).unwrap(),
                }
            }
        }
    }

    // private
    fn load_charts(&mut self, asset: &mut Asset) {
        log::info!(":: Trader load charts {asset}");

        for tf in TimeFrame::all() {
            asset.load_chart_empty(tf);
        }
    }
}
impl Default for Trader {
    fn default() -> Self {
        Trader::new()
    }
}

async fn start_broker(mut broker: Tinkoff) {
    broker.start().await
}