kabu_defi_price/
price_actor.rs

1use eyre::eyre;
2use kabu_core_components::Component;
3use std::collections::HashMap;
4use std::marker::PhantomData;
5use std::ops::{Div, Mul};
6use std::sync::Arc;
7use std::time::Duration;
8use tokio::sync::RwLock;
9
10use alloy_network::Network;
11use alloy_primitives::{Address, U256, address};
12use alloy_provider::Provider;
13
14use kabu_core_blockchain::Blockchain;
15use kabu_defi_address_book::TokenAddressEth;
16use kabu_defi_pools::CurvePool;
17use kabu_defi_pools::protocols::CurveProtocol;
18use kabu_types_market::{Market, Pool};
19use tracing::{debug, error, info};
20
21async fn price_worker<N: Network, P: Provider<N> + Clone + 'static>(
22    client: P,
23    market: Arc<RwLock<Market>>,
24    once: bool,
25) -> eyre::Result<()> {
26    let curve_tricrypto_usdc = CurveProtocol::new_u256_3_eth_to(client.clone(), address!("7F86Bf177Dd4F3494b841a37e810A34dD56c829B"));
27    let curve_tricrypto_usdt = CurveProtocol::new_u256_3_eth_to(client.clone(), address!("f5f5b97624542d72a9e06f04804bf81baa15e2b4"));
28
29    let mut coins_hash_map: HashMap<Address, CurvePool<P, N>> = HashMap::new();
30
31    let curve_tricrypto_usdc_pool = CurvePool::fetch_pool_data(client.clone(), curve_tricrypto_usdc).await?;
32
33    let curve_tricrypto_usdt_pool = CurvePool::fetch_pool_data(client.clone(), curve_tricrypto_usdt).await?;
34
35    coins_hash_map.insert(TokenAddressEth::USDC, curve_tricrypto_usdc_pool.clone());
36    coins_hash_map.insert(TokenAddressEth::WBTC, curve_tricrypto_usdc_pool.clone());
37    coins_hash_map.insert(TokenAddressEth::USDT, curve_tricrypto_usdt_pool.clone());
38
39    let one_ether = U256::from(10).pow(U256::from(18));
40    let weth_amount = one_ether.mul(U256::from(5));
41
42    match market.read().await.get_token(&TokenAddressEth::WETH) {
43        Some(token) => {
44            token.set_eth_price(Some(one_ether));
45        }
46        _ => {
47            error!("WETH_NOT_FOUND")
48        }
49    }
50
51    loop {
52        for (token_address, curve_pool) in coins_hash_map.iter() {
53            debug!("Fetching price of {} at {}", token_address, curve_pool.get_address());
54
55            match curve_pool.fetch_out_amount(TokenAddressEth::WETH, *token_address, weth_amount).await {
56                Ok(out_amount) => {
57                    let price = out_amount.mul(one_ether).div(weth_amount);
58                    info!("Price of ETH in {token_address:#20x} is {price}");
59                    match market.read().await.get_token(token_address) {
60                        Some(tkn) => {
61                            tkn.set_eth_price(Some(price));
62                            debug!("Price is set");
63                        }
64                        _ => {
65                            error!(address=%token_address, "Token not found");
66                        }
67                    }
68                }
69                Err(error) => {
70                    error!(%error, "fetch_out_amount")
71                }
72            }
73        }
74
75        let usdt_price = market.read().await.get_token_or_default(&TokenAddressEth::USDT).get_eth_price();
76        let usdc_price = market.read().await.get_token_or_default(&TokenAddressEth::USDC).get_eth_price();
77
78        let mut usd_price: Option<U256> = None;
79        if let Some(usdc_price) = usdc_price
80            && let Some(usdt_price) = usdt_price
81        {
82            usd_price = Some((usdc_price + usdt_price) >> 1);
83        }
84
85        if let Some(usd_price) = usd_price {
86            match market.read().await.get_token(&TokenAddressEth::DAI) {
87                Some(tkn) => {
88                    tkn.set_eth_price(Some(U256::from(10).pow(U256::from(12)).mul(usd_price)));
89                }
90                _ => {
91                    error!("Token {:#20x} not found", TokenAddressEth::DAI);
92                }
93            }
94        }
95        if once {
96            break;
97        }
98
99        let _ = tokio::time::sleep(Duration::new(60, 0)).await;
100    }
101    Ok(())
102}
103
104#[derive(Clone)]
105pub struct PriceComponent<P, N> {
106    client: P,
107    only_once: bool,
108
109    market: Option<Arc<RwLock<Market>>>,
110    _n: PhantomData<N>,
111}
112
113impl<P, N> PriceComponent<P, N>
114where
115    N: Network,
116    P: Provider<N> + Send + Sync + Clone + 'static,
117{
118    pub fn new(client: P) -> Self {
119        Self { client, only_once: false, market: None, _n: PhantomData }
120    }
121
122    pub fn only_once(self) -> Self {
123        Self { only_once: true, ..self }
124    }
125
126    pub fn on_bc(self, bc: &Blockchain) -> Self {
127        Self { market: Some(bc.market()), ..self }
128    }
129
130    pub fn with_market(self, market: Arc<RwLock<Market>>) -> Self {
131        Self { market: Some(market), ..self }
132    }
133}
134
135impl<P, N> Component for PriceComponent<P, N>
136where
137    N: Network,
138    P: Provider<N> + Send + Sync + Clone + 'static,
139{
140    fn spawn(self, executor: reth_tasks::TaskExecutor) -> eyre::Result<()> {
141        let name = self.name();
142        let market = self.market.ok_or_else(|| eyre!("Market is not set"))?;
143
144        executor.spawn_critical(name, async move {
145            if let Err(e) = price_worker(self.client, market, self.only_once).await {
146                tracing::error!("Price worker failed: {}", e);
147            }
148        });
149        Ok(())
150    }
151
152    fn name(&self) -> &'static str {
153        "PriceComponent"
154    }
155}
kabu_defi_price/price_actor.rs

kabu_defi_price/
price_actor.rs
