fix: on chain comparison with Defillama

src/constants.rs

@@ -81,10 +81,10 @@ lazy_static! {
         &["network", "pair"]
     )
     .unwrap();
-    pub static ref API_ON_OFF_PRICE_DEVIATION: GaugeVec = register_gauge_vec!(
+    pub static ref ON_OFF_PRICE_DEVIATION: GaugeVec = register_gauge_vec!(
         opts!(
-            "api_on_off_price_deviation",
-            "Price deviation betqeen the on-chain price and the off-chain price."
+            "on_off_price_deviation",
+            "On chain price deviation from the reference price"
         ),
         &["network", "pair"]
     )

src/monitoring/mod.rs

@@ -4,6 +4,7 @@ pub mod publisher_balance;
 pub mod source_deviation;
 pub mod time_since_last_update;
 
+pub use on_off_deviation::on_off_price_deviation;
 pub use price_deviation::price_deviation;
 pub use publisher_balance::publisher_balance;
 pub use source_deviation::source_deviation;

src/monitoring/on_off_deviation.rs

@@ -8,19 +8,22 @@ use starknet::{
     providers::Provider,
 };
 
+use crate::monitoring::price_deviation::CoinPricesDTO;
 use crate::{
     config::{get_config, DataType},
+    constants::COINGECKO_IDS,
     error::MonitoringError,
+    types::Entry,
 };
 
-pub async fn raw_on_off_price_deviation(
-    pair_id: &String,
-    off_chain_price: f64,
+pub async fn on_off_price_deviation<T: Entry>(
+    pair_id: String,
+    timestamp: u64,
 ) -> Result<(f64, u32), MonitoringError> {
+    let ids = &COINGECKO_IDS;
     let config = get_config(None).await;
-
     let client = &config.network().provider;
-    let field_pair = cairo_short_string_to_felt(pair_id).expect("failed to convert pair id");
+    let field_pair = cairo_short_string_to_felt(&pair_id).expect("failed to convert pair id");
 
     let data = client
         .call(
@@ -34,13 +37,14 @@ pub async fn raw_on_off_price_deviation(
         .await
         .map_err(|e| MonitoringError::OnChain(e.to_string()))?;
 
-    let decimals = config
-        .decimals(DataType::Spot)
-        .get(pair_id)
-        .ok_or(MonitoringError::OnChain(format!(
-            "Failed to get decimals for pair {:?}",
-            pair_id
-        )))?;
+    let decimals =
+        config
+            .decimals(DataType::Spot)
+            .get(&pair_id)
+            .ok_or(MonitoringError::OnChain(format!(
+                "Failed to get decimals for pair {:?}",
+                pair_id
+            )))?;
 
     let on_chain_price = data
         .first()
@@ -51,7 +55,48 @@ pub async fn raw_on_off_price_deviation(
             "Failed to convert to f64".to_string(),
         ))?;
 
-    let deviation = (off_chain_price - on_chain_price) / on_chain_price;
+    let coingecko_id = *ids.get(&pair_id).expect("Failed to get coingecko id");
+
+    let api_key = std::env::var("DEFILLAMA_API_KEY");
+
+    let request_url = if let Ok(api_key) = api_key {
+        format!(
+            "https://coins.llama.fi/prices/historical/{timestamp}/coingecko:{id}?apikey={apikey}",
+            timestamp = timestamp,
+            id = coingecko_id,
+            apikey = api_key
+        )
+    } else {
+        format!(
+            "https://coins.llama.fi/prices/historical/{timestamp}/coingecko:{id}",
+            timestamp = timestamp,
+            id = coingecko_id,
+        )
+    };
+
+    let response = reqwest::get(&request_url)
+        .await
+        .map_err(|e| MonitoringError::Api(e.to_string()))?;
+
+    let coins_prices: CoinPricesDTO = response.json().await.map_err(|e| {
+        MonitoringError::Api(format!(
+            "Failed to convert to DTO object, got error {:?}",
+            e.to_string()
+        ))
+    })?;
+
+    let api_id = format!("coingecko:{}", coingecko_id);
+
+    let reference_price = coins_prices
+        .get_coins()
+        .get(&api_id)
+        .ok_or(MonitoringError::Api(format!(
+            "Failed to get coingecko price for id {:?}",
+            coingecko_id
+        )))?
+        .get_price();
+
+    let deviation = (reference_price - on_chain_price) / on_chain_price;
     let num_sources_aggregated = (*data.get(3).unwrap()).try_into().map_err(|e| {
         MonitoringError::Conversion(format!("Failed to convert num sources {:?}", e))
     })?;

src/monitoring/price_deviation.rs

@@ -15,19 +15,30 @@ use crate::{constants::COINGECKO_IDS, error::MonitoringError, types::Entry};
 //   }
 // }
 #[derive(serde::Deserialize, Debug)]
-struct CoinPricesDTO {
+pub struct CoinPricesDTO {
     coins: HashMap<String, CoinPriceDTO>,
 }
 
 #[allow(unused)]
 #[derive(serde::Deserialize, Debug)]
-struct CoinPriceDTO {
+pub struct CoinPriceDTO {
     price: f64,
     symbol: String,
     timestamp: u64,
     confidence: f64,
 }
 
+impl CoinPricesDTO {
+    pub fn get_coins(&self) -> &HashMap<String, CoinPriceDTO> {
+        &self.coins
+    }
+}
+impl CoinPriceDTO {
+    pub fn get_price(&self) -> f64 {
+        self.price.clone()
+    }
+}
+
 /// Calculates the deviation of the price from a trusted API (DefiLLama)
 pub async fn price_deviation<T: Entry>(
     query: &T,

src/processing/api.rs

@@ -8,13 +8,11 @@ use starknet::{
 use crate::{
     config::get_config,
     constants::{
-        API_NUM_SOURCES, API_ON_OFF_PRICE_DEVIATION, API_PRICE_DEVIATION, API_SEQUENCER_DEVIATION,
-        API_TIME_SINCE_LAST_UPDATE,
+        API_NUM_SOURCES, API_PRICE_DEVIATION, API_SEQUENCER_DEVIATION, API_TIME_SINCE_LAST_UPDATE,
     },
     error::MonitoringError,
     monitoring::{
-        on_off_deviation::raw_on_off_price_deviation, price_deviation::raw_price_deviation,
-        time_since_last_update::raw_time_since_last_update,
+        price_deviation::raw_price_deviation, time_since_last_update::raw_time_since_last_update,
     },
     processing::common::query_pragma_api,
 };
@@ -37,7 +35,6 @@ pub async fn process_data_by_pair(pair: String) -> Result<(), MonitoringError> {
 
     let price_deviation = raw_price_deviation(&pair, normalized_price).await?;
     let time_since_last_update = raw_time_since_last_update(result.timestamp)?;
-    let (on_off_price_deviation, _) = raw_on_off_price_deviation(&pair, normalized_price).await?;
 
     API_PRICE_DEVIATION
         .with_label_values(&[network_env, &pair])
@@ -48,9 +45,7 @@ pub async fn process_data_by_pair(pair: String) -> Result<(), MonitoringError> {
     API_NUM_SOURCES
         .with_label_values(&[network_env, &pair])
         .set(result.num_sources_aggregated as i64);
-    API_ON_OFF_PRICE_DEVIATION
-        .with_label_values(&[network_env, &pair])
-        .set(on_off_price_deviation);
+
     Ok(())
 }
 

src/processing/future.rs

@@ -5,6 +5,7 @@ use crate::config::get_config;
 use crate::config::DataType;
 use crate::config::NetworkName;
 use crate::constants::NUM_SOURCES;
+use crate::constants::ON_OFF_PRICE_DEVIATION;
 use crate::constants::PAIR_PRICE;
 use crate::constants::PRICE_DEVIATION;
 use crate::constants::PRICE_DEVIATION_SOURCE;
@@ -13,7 +14,9 @@ use crate::constants::TIME_SINCE_LAST_UPDATE_PUBLISHER;
 use crate::diesel::QueryDsl;
 use crate::error::MonitoringError;
 use crate::models::FutureEntry;
-use crate::monitoring::{price_deviation, source_deviation, time_since_last_update};
+use crate::monitoring::{
+    on_off_price_deviation, price_deviation, source_deviation, time_since_last_update,
+};
 
 use crate::schema::future_entry::dsl as testnet_dsl;
 use crate::schema::mainnet_future_entry::dsl as mainnet_dsl;
@@ -79,6 +82,8 @@ pub async fn process_data_by_pair_and_sources(
     let mut timestamps = Vec::new();
 
     let config = get_config(None).await;
+    let network_env = &config.network_str();
+    let data_type = "future";
 
     let decimals = *config
         .decimals(DataType::Future)
@@ -91,6 +96,11 @@ pub async fn process_data_by_pair_and_sources(
         timestamps.push(res);
     }
 
+    let (on_off_deviation, _) =
+        on_off_price_deviation::<FutureEntry>(pair.clone(), *timestamps.last().unwrap()).await?;
+    ON_OFF_PRICE_DEVIATION
+        .with_label_values(&[network_env, &pair.clone(), data_type])
+        .set(on_off_deviation);
     Ok(*timestamps.last().unwrap())
 }
 
@@ -143,7 +153,6 @@ pub async fn process_data_by_pair_and_source(
             let source_deviation_labels =
                 PRICE_DEVIATION_SOURCE.with_label_values(&[network_env, pair, src, data_type]);
             let num_sources_labels = NUM_SOURCES.with_label_values(&[network_env, pair, data_type]);
-
             // Compute metrics
             let time = time_since_last_update(&data);
             let price_as_f64 = data.price.to_f64().ok_or(MonitoringError::Price(

src/processing/spot.rs

@@ -5,6 +5,7 @@ use crate::config::get_config;
 use crate::config::DataType;
 use crate::config::NetworkName;
 use crate::constants::NUM_SOURCES;
+use crate::constants::ON_OFF_PRICE_DEVIATION;
 use crate::constants::PAIR_PRICE;
 use crate::constants::PRICE_DEVIATION;
 use crate::constants::PRICE_DEVIATION_SOURCE;
@@ -13,7 +14,9 @@ use crate::constants::TIME_SINCE_LAST_UPDATE_PUBLISHER;
 use crate::diesel::QueryDsl;
 use crate::error::MonitoringError;
 use crate::models::SpotEntry;
-use crate::monitoring::{price_deviation, source_deviation, time_since_last_update};
+use crate::monitoring::{
+    on_off_price_deviation, price_deviation, source_deviation, time_since_last_update,
+};
 
 use crate::schema::mainnet_spot_entry::dsl as mainnet_dsl;
 use crate::schema::spot_entry::dsl as testnet_dsl;
@@ -77,8 +80,9 @@ pub async fn process_data_by_pair_and_sources(
     sources: Vec<String>,
 ) -> Result<u64, MonitoringError> {
     let mut timestamps = Vec::new();
-
     let config = get_config(None).await;
+    let data_type: &str = "spot";
+    let network_env = &config.network_str();
 
     let decimals = *config.decimals(DataType::Spot).get(&pair.clone()).unwrap();
 
@@ -87,7 +91,11 @@ pub async fn process_data_by_pair_and_sources(
         let res = process_data_by_pair_and_source(pool.clone(), &pair, &src, decimals).await?;
         timestamps.push(res);
     }
-
+    let (on_off_deviation, _) =
+        on_off_price_deviation::<SpotEntry>(pair.clone(), *timestamps.last().unwrap()).await?;
+    ON_OFF_PRICE_DEVIATION
+        .with_label_values(&[network_env, &pair.clone(), data_type])
+        .set(on_off_deviation);
     Ok(*timestamps.last().unwrap())
 }
 
@@ -126,7 +134,7 @@ pub async fn process_data_by_pair_and_source(
     match filtered_by_source_result {
         Ok(data) => {
             let network_env = &config.network_str();
-            let data_type = "spot";
+            let data_type: &str = "spot";
 
             // Get the labels
             let time_labels = TIME_SINCE_LAST_UPDATE_PUBLISHER.with_label_values(&[