[Feat] add forex

Cargo.toml

@@ -31,8 +31,6 @@ bothan-htx = { path = "bothan-htx", version = "0.1.0" }
 bothan-kraken = { path = "bothan-kraken", version = "0.1.0" }
 bothan-okx = { path = "bothan-okx", version = "0.1.0" }
 bothan-band = { path = "bothan-band", version = "0.1.0" }
-
-anyhow = "1.0.86"
 async-trait = "0.1.77"
 bincode = "2.0.1"
 chrono = "0.4.39"
@@ -46,8 +44,6 @@ mockito = "1.4.0"
 num-traits = "0.2.19"
 opentelemetry = { version = "0.28.0", features = ["metrics"] }
 prost = "0.13.1"
-protoc-gen-prost = "0.4.0"
-protoc-gen-tonic = "0.4.1"
 rand = "0.8.5"
 reqwest = { version = "0.12.3", features = ["json"] }
 rust_decimal = "1.10.2"

README.md

@@ -23,7 +23,7 @@ This project comprises primarily of 6 main components:
 - `bothan-{exchange}` - Exchange-specific implementations
 - [`proto`](proto/) - Protocol buffer definitions
 
-## Supported Data Sources
+## Supported Crypto Data Sources
 
 - [Binance](bothan-binance)
 - [Bitfinex](bothan-bitfinex)
@@ -38,6 +38,12 @@ This project comprises primarily of 6 main components:
 - [Band/kiwi](bothan-band)
 - [Band/macaw](bothan-band)
 
+## Supported Forex Data Sources
+
+- [Band/owlet](bothan-band)
+- [Band/fieldfare](bothan-band)
+- [Band/xenops](bothan-band)
+
 ## Features
 
 - **Unified API**: Consistent interface across all supported exchanges

bothan-api/server-cli/config.example.toml

@@ -41,6 +41,11 @@ enabled = false
 # The threshold in seconds after which data is considered stale.
 stale_threshold = 300
 
+# Manager configuration for handling forex data sources
+[manager.forex]
+# The threshold in seconds after which data is considered stale.
+stale_threshold = 7200
+
 # Configuration for the data sources that the manager will use.
 # If any of these [manager.crypto.source] sections (e.g., section manager.crypto.source.binance) are removed,
 # that specific source will not be used in bothan.
@@ -114,6 +119,24 @@ url = "https://macaw.bandchain.org"
 # Update interval for Band Macaw source
 update_interval = "1m"
 
+[manager.forex.source.band_owlet]
+# URL for Band Owlet source
+url = "https://owlet.bandchain.org"
+# Update interval for Band Owlet source
+update_interval = "1m"
+
+[manager.forex.source.band_fieldfare]
+# URL for Band Fieldfare source
+url = "https://fieldfare.bandchain.org"
+# Update interval for Band Fieldfare source
+update_interval = "1m"
+
+[manager.forex.source.band_xenops]
+# URL for Band Xenops source
+url = "https://xenops.bandchain.org"
+# Update interval for Band Xenops source
+update_interval = "1m"
+
 # Telemetry configuration
 [telemetry]
 # Enable or disable telemetry.

bothan-api/server-cli/src/commands/query.rs

@@ -6,7 +6,7 @@
 //!
 //! ## Features
 //!
-//! - Query prices from Binance, Bitfinex, Bybit, Coinbase, CoinGecko, CoinMarketCap, HTX, Kraken, OKX
+//! - Query prices from Binance, Bitfinex, Bybit, Coinbase, CoinGecko, CoinMarketCap, HTX, Kraken, OKX, Band (Kiwi, Macaw, Owlet, Fieldfare, Xenops)
 //! - Customizable timeout and query IDs
 //! - Pretty-printed table output
 //!
@@ -124,55 +124,86 @@ pub enum QuerySubCommand {
         #[clap(flatten)]
         args: QueryArgs,
     },
+    /// Query Band/owlet prices
+    #[clap(name = "band/owlet")]
+    BandOwlet {
+        #[clap(flatten)]
+        args: QueryArgs,
+    },
+    /// Query Band/fieldfare prices
+    #[clap(name = "band/fieldfare")]
+    BandFieldfare {
+        #[clap(flatten)]
+        args: QueryArgs,
+    },
+    /// Query Band/xenops prices
+    #[clap(name = "band/xenops")]
+    BandXenops {
+        #[clap(flatten)]
+        args: QueryArgs,
+    },
 }
 
 impl QueryCli {
     pub async fn run(&self, app_config: AppConfig) -> anyhow::Result<()> {
-        let source_config = app_config.manager.crypto.source;
+        let crypto_config = app_config.manager.crypto.source;
+        let forex_config = app_config.manager.forex.source;
         let config_err = anyhow!("Config is missing. Please check your config.toml.");
         match &self.subcommand {
             QuerySubCommand::Binance { args } => {
-                let opts = source_config.binance.ok_or(config_err)?;
+                let opts = crypto_config.binance.ok_or(config_err)?;
                 query_binance(opts, &args.query_ids, args.timeout).await?;
             }
             QuerySubCommand::Bitfinex { args } => {
-                let opts = source_config.bitfinex.ok_or(config_err)?;
+                let opts = crypto_config.bitfinex.ok_or(config_err)?;
                 query_bitfinex(opts, &args.query_ids, args.timeout).await?;
             }
             QuerySubCommand::Bybit { args } => {
-                let opts = source_config.bybit.ok_or(config_err)?;
+                let opts = crypto_config.bybit.ok_or(config_err)?;
                 query_bybit(opts, &args.query_ids, args.timeout).await?;
             }
             QuerySubCommand::Coinbase { args } => {
-                let opts = source_config.coinbase.ok_or(config_err)?;
+                let opts = crypto_config.coinbase.ok_or(config_err)?;
                 query_coinbase(opts, &args.query_ids, args.timeout).await?;
             }
             QuerySubCommand::CoinGecko { args } => {
-                let opts = source_config.coingecko.ok_or(config_err)?;
+                let opts = crypto_config.coingecko.ok_or(config_err)?;
                 query_coingecko(opts, &args.query_ids, args.timeout).await?;
             }
             QuerySubCommand::CoinMarketCap { args } => {
-                let opts = source_config.coinmarketcap.ok_or(config_err)?;
+                let opts = crypto_config.coinmarketcap.ok_or(config_err)?;
                 query_coinmarketcap(opts, &args.query_ids, args.timeout).await?;
             }
             QuerySubCommand::Htx { args } => {
-                let opts = source_config.htx.ok_or(config_err)?;
+                let opts = crypto_config.htx.ok_or(config_err)?;
                 query_htx(opts, &args.query_ids, args.timeout).await?;
             }
             QuerySubCommand::Kraken { args } => {
-                let opts = source_config.kraken.ok_or(config_err)?;
+                let opts = crypto_config.kraken.ok_or(config_err)?;
                 query_kraken(opts, &args.query_ids, args.timeout).await?;
             }
             QuerySubCommand::Okx { args } => {
-                let opts = source_config.okx.ok_or(config_err)?;
+                let opts = crypto_config.okx.ok_or(config_err)?;
                 query_okx(opts, &args.query_ids, args.timeout).await?;
             }
             QuerySubCommand::BandKiwi { args } => {
-                let opts = source_config.band_kiwi.ok_or(config_err)?;
+                let opts = crypto_config.band_kiwi.ok_or(config_err)?;
                 query_band(opts, &args.query_ids, args.timeout).await?;
             }
             QuerySubCommand::BandMacaw { args } => {
-                let opts = source_config.band_macaw.ok_or(config_err)?;
+                let opts = crypto_config.band_macaw.ok_or(config_err)?;
+                query_band(opts, &args.query_ids, args.timeout).await?;
+            }
+            QuerySubCommand::BandOwlet { args } => {
+                let opts = forex_config.band_owlet.ok_or(config_err)?;
+                query_band(opts, &args.query_ids, args.timeout).await?;
+            }
+            QuerySubCommand::BandFieldfare { args } => {
+                let opts = forex_config.band_fieldfare.ok_or(config_err)?;
+                query_band(opts, &args.query_ids, args.timeout).await?;
+            }
+            QuerySubCommand::BandXenops { args } => {
+                let opts = forex_config.band_xenops.ok_or(config_err)?;
                 query_band(opts, &args.query_ids, args.timeout).await?;
             }
         }

bothan-api/server-cli/src/commands/start.rs

@@ -15,11 +15,12 @@ use bothan_api::api::BothanServer;
 use bothan_api::config::AppConfig;
 use bothan_api::config::ipfs::IpfsAuthentication;
 use bothan_api::config::manager::crypto_info::sources::CryptoSourceConfigs;
+use bothan_api::config::manager::forex_info::sources::ForexSourceConfigs;
 use bothan_api::proto::bothan::v1::{BothanServiceServer, FILE_DESCRIPTOR_SET};
 use bothan_api::{REGISTRY_REQUIREMENT, VERSION};
 use bothan_core::ipfs::{IpfsClient, IpfsClientBuilder};
-use bothan_core::manager::CryptoAssetInfoManager;
-use bothan_core::manager::crypto_asset_info::CryptoAssetWorkerOpts;
+use bothan_core::manager::AssetInfoManager;
+use bothan_core::manager::asset_info::AssetWorkerOpts;
 use bothan_core::monitoring::{Client as MonitoringClient, Signer};
 use bothan_core::store::rocksdb::RocksDbStore;
 use bothan_core::telemetry;
@@ -187,23 +188,36 @@ async fn init_bothan_server<S: Store + 'static>(
     ipfs_client: IpfsClient,
     monitoring_client: Option<Arc<MonitoringClient>>,
 ) -> anyhow::Result<Arc<BothanServer<S>>> {
-    let stale_threshold = config.manager.crypto.stale_threshold;
+    let prefix_stale_thresholds = init_prefix_stale_thresholds(
+        config.manager.crypto.stale_threshold,
+        config.manager.forex.stale_threshold,
+    );
     let bothan_version =
         Version::from_str(VERSION).with_context(|| "Failed to parse bothan version")?;
     let registry_version_requirement = VersionReq::from_str(REGISTRY_REQUIREMENT)
         .with_context(|| "Failed to parse registry version requirement")?;
 
-    let opts = match init_crypto_opts(&config.manager.crypto.source).await {
+    let crypto_opts = match init_crypto_opts(&config.manager.crypto.source).await {
         Ok(workers) => workers,
         Err(e) => {
             bail!("failed to initialize workers: {:?}", e);
         }
     };
-    let manager = match CryptoAssetInfoManager::build(
+
+    let forex_opts = match init_forex_opts(&config.manager.forex.source).await {
+        Ok(workers) => workers,
+        Err(e) => {
+            bail!("failed to initialize workers: {:?}", e);
+        }
+    };
+
+    let worker_opts = crypto_opts.into_iter().chain(forex_opts).collect();
+
+    let manager = match AssetInfoManager::build(
         store,
-        opts,
+        worker_opts,
         ipfs_client,
-        stale_threshold,
+        prefix_stale_thresholds,
         bothan_version,
         registry_version_requirement,
         monitoring_client,
@@ -237,9 +251,19 @@ async fn init_bothan_server<S: Store + 'static>(
     Ok(Arc::new(BothanServer::new(manager, metrics)))
 }
 
+fn init_prefix_stale_thresholds(
+    crypto_stale_threshold: i64,
+    forex_stale_threshold: i64,
+) -> HashMap<String, i64> {
+    HashMap::from([
+        ("CS".to_string(), crypto_stale_threshold),
+        ("FS".to_string(), forex_stale_threshold),
+    ])
+}
+
 async fn init_crypto_opts(
     source: &CryptoSourceConfigs,
-) -> Result<HashMap<String, CryptoAssetWorkerOpts>, AssetWorkerError> {
+) -> Result<HashMap<String, AssetWorkerOpts>, AssetWorkerError> {
     let mut worker_opts = HashMap::new();
 
     add_worker_opts(&mut worker_opts, &source.binance).await?;
@@ -257,8 +281,20 @@ async fn init_crypto_opts(
     Ok(worker_opts)
 }
 
-async fn add_worker_opts<O: Clone + Into<CryptoAssetWorkerOpts>>(
-    workers_opts: &mut HashMap<String, CryptoAssetWorkerOpts>,
+async fn init_forex_opts(
+    source: &ForexSourceConfigs,
+) -> Result<HashMap<String, AssetWorkerOpts>, AssetWorkerError> {
+    let mut worker_opts = HashMap::new();
+
+    add_worker_opts(&mut worker_opts, &source.band_owlet).await?;
+    add_worker_opts(&mut worker_opts, &source.band_fieldfare).await?;
+    add_worker_opts(&mut worker_opts, &source.band_xenops).await?;
+
+    Ok(worker_opts)
+}
+
+async fn add_worker_opts<O: Clone + Into<AssetWorkerOpts>>(
+    workers_opts: &mut HashMap<String, AssetWorkerOpts>,
     opts: &Option<O>,
 ) -> Result<(), AssetWorkerError> {
     if let Some(opts) = opts {

bothan-api/server/src/api/server.rs

@@ -16,8 +16,8 @@
 use std::sync::Arc;
 use std::time::Instant;
 
-use bothan_core::manager::CryptoAssetInfoManager;
-use bothan_core::manager::crypto_asset_info::error::{PushMonitoringRecordError, SetRegistryError};
+use bothan_core::manager::AssetInfoManager;
+use bothan_core::manager::asset_info::error::{PushMonitoringRecordError, SetRegistryError};
 use bothan_lib::metrics::server::{Metrics, ServiceName};
 use bothan_lib::store::Store;
 use semver::Version;
@@ -35,13 +35,13 @@ pub const PRECISION: u32 = 9;
 
 /// The `BothanServer` struct represents a server that implements the `BothanService` trait.
 pub struct BothanServer<S: Store + 'static> {
-    manager: Arc<CryptoAssetInfoManager<S>>,
+    manager: Arc<AssetInfoManager<S>>,
     metrics: Metrics,
 }
 
 impl<S: Store> BothanServer<S> {
     /// Creates a new `BothanServer` instance.
-    pub fn new(manager: Arc<CryptoAssetInfoManager<S>>, metrics: Metrics) -> Self {
+    pub fn new(manager: Arc<AssetInfoManager<S>>, metrics: Metrics) -> Self {
         BothanServer { manager, metrics }
     }
 }

bothan-api/server/src/api/utils.rs

@@ -7,7 +7,7 @@
 //!
 //! - `parse_price_state`: Converts a `PriceState` to a `Price` API response.
 
-use bothan_core::manager::crypto_asset_info::types::PriceState;
+use bothan_core::manager::asset_info::types::PriceState;
 use rust_decimal::prelude::Zero;
 use tracing::error;
 

bothan-api/server/src/config/manager.rs

@@ -10,14 +10,23 @@
 //! ```
 
 use crypto_info::CryptoInfoManagerConfig;
+use forex_info::ForexInfoManagerConfig;
 use serde::{Deserialize, Serialize};
 
+/// Shared Band worker serde helpers.
+pub(crate) mod band_serde;
 /// Crypto info manager configuration module.
 pub mod crypto_info;
+/// Forex info manager configuration module.
+pub mod forex_info;
 
 /// The configuration for all bothan-api's manager.
 #[derive(Clone, Debug, Default, Serialize, Deserialize)]
 pub struct ManagerConfig {
     /// The configuration for the crypto info manager.
     pub crypto: CryptoInfoManagerConfig,
+    /// The configuration for the forex info manager.
+    /// Note: This field is optional and will be set to default if not provided.
+    #[serde(default)]
+    pub forex: ForexInfoManagerConfig,
 }

bothan-api/server/src/config/manager/band_serde.rs

@@ -0,0 +1,19 @@
+// Macro to generate deserialization functions for Band workers with preset names.
+// This macro defines a function that:
+// - Deserializes an Option<WorkerOpts>,
+// - If present, creates a new WorkerOpts with the given name and original URL/update_interval.
+macro_rules! de_band_named {
+    ($fn_name:ident, $name:expr) => {
+        fn $fn_name<'de, D>(d: D) -> Result<Option<bothan_band::WorkerOpts>, D::Error>
+        where
+            D: serde::Deserializer<'de>,
+        {
+            use serde::Deserialize;
+            let v = Option::<bothan_band::WorkerOpts>::deserialize(d)?;
+            let v = v.map(|w| bothan_band::WorkerOpts::new($name, &w.url, Some(w.update_interval)));
+            Ok(v)
+        }
+    };
+}
+
+pub(crate) use de_band_named;