feat: funding rates table ingested from kafka (#226)

* feat: refacto pragma-ingestor

* feat: funding rates table migratin

* feat: funding rates ingestor tasks

* fix: clippy

Author: EvolveArt

Cargo.lock

@@ -98,8 +98,9 @@ nonzero_ext = { version = "0.3.0" }
 serde_json = { version = "1.0.122", features = ["arbitrary_precision"] }
 starknet = "0.14.0"
 starknet-crypto = "0.7.4"
+clap = { version = "4.4", features = ["derive", "env"] }
 reqwest = { version = "0.12.5", features = ["blocking"] }
-rdkafka = "0.36.2"
+rdkafka = "0.37.0"
 thiserror = "2.0.12"
 strum = { version = "0.26", features = ["derive"] }
 url = "2.5.0"

Cargo.toml

@@ -0,0 +1,10 @@
+-- This file should undo anything in `up.sql`
+
+-- Remove compression policy first
+SELECT remove_compression_policy('funding_rates');
+
+-- Disable compression
+ALTER TABLE funding_rates SET (timescaledb.compress = false);
+
+-- Drop the hypertable
+DROP TABLE funding_rates CASCADE;

pragma-entities/migrations/2025-04-20-145559_create-funding-rates-table/down.sql

@@ -0,0 +1,26 @@
+-- Your SQL goes here
+
+CREATE TABLE funding_rates (
+    id uuid DEFAULT uuid_generate_v4(),
+    source VARCHAR NOT NULL,
+    pair VARCHAR NOT NULL,
+    annualized_rate DOUBLE PRECISION NOT NULL,
+    timestamp TIMESTAMPTZ NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    PRIMARY KEY (id, timestamp)
+);
+
+-- Convert the table to a hypertable
+SELECT create_hypertable('funding_rates', 'timestamp', chunk_time_interval => INTERVAL '1 day');
+
+-- Create an index for efficient querying by pair
+CREATE INDEX idx_funding_rates_pair ON funding_rates(pair);
+
+-- Enable compression
+ALTER TABLE funding_rates SET (
+    timescaledb.compress,
+    timescaledb.compress_segmentby = 'source,pair'
+);
+
+-- Add compression policy to compress chunks older than 7 days
+SELECT add_compression_policy('funding_rates', INTERVAL '7 days');

pragma-entities/migrations/2025-04-20-145559_create-funding-rates-table/up.sql

@@ -27,7 +27,7 @@ impl From<crate::FutureEntry> for FutureEntry {
             source: future_entry.source,
             timestamp: future_entry.timestamp.and_utc().timestamp_millis() as u64,
             expiration_timestamp,
-            publisher_signature: future_entry.publisher_signature,
+            publisher_signature: future_entry.publisher_signature.unwrap_or_default(),
             price: future_entry.price.to_u128().unwrap_or(0), // change default value ?
         }
     }

pragma-entities/src/dto/future_entry.rs

@@ -15,6 +15,7 @@ pub use error::InfraError;
 pub use models::{
     checkpoint_error::CheckpointError,
     entry::{Entry, NewEntry},
+    funding_rate::{FundingRate, NewFundingRate},
     future_entry::{FutureEntry, NewFutureEntry},
     publisher::{NewPublisher, Publishers},
     publisher_error::PublisherError,

pragma-entities/src/lib.rs

@@ -25,7 +25,7 @@ pub struct FutureEntry {
     // If expiration_timestamp is None, it means the entry is a perpetual future
     // else it is a regular future entry that will expire at the expiration_timestamp.
     pub expiration_timestamp: Option<NaiveDateTime>,
-    pub publisher_signature: String,
+    pub publisher_signature: Option<String>,
     pub price: BigDecimal,
 }
 

pragma-entities/src/models/entries/future_entry.rs

@@ -0,0 +1,38 @@
+use chrono::NaiveDateTime;
+use diesel::prelude::*;
+use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+use crate::schema::funding_rates;
+
+#[derive(Debug, Clone, Queryable, Serialize, Deserialize)]
+#[diesel(table_name = funding_rates)]
+#[diesel(check_for_backend(diesel::pg::Pg))]
+pub struct FundingRate {
+    pub id: Uuid,
+    pub source: String,
+    pub pair: String,
+    pub annualized_rate: f64,
+    pub timestamp: NaiveDateTime,
+    pub created_at: NaiveDateTime,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, Insertable, AsChangeset)]
+#[diesel(table_name = funding_rates)]
+pub struct NewFundingRate {
+    pub source: String,
+    pub pair: String,
+    pub annualized_rate: f64,
+    pub timestamp: NaiveDateTime,
+}
+
+impl FundingRate {
+    pub fn create_many(
+        conn: &mut PgConnection,
+        new_entries: Vec<NewFundingRate>,
+    ) -> Result<Vec<Self>, diesel::result::Error> {
+        diesel::insert_into(funding_rates::table)
+            .values(&new_entries)
+            .get_results(conn)
+    }
+}

pragma-entities/src/models/funding_rate.rs

@@ -1,5 +1,6 @@
 pub mod checkpoint_error;
 pub mod entries;
+pub mod funding_rate;
 pub mod publisher;
 pub mod publisher_error;
 

pragma-entities/src/models/mod.rs

@@ -4,11 +4,22 @@ diesel::table! {
     entries (id, timestamp) {
         id -> Uuid,
         pair_id -> Varchar,
-        publisher -> Text,
-        timestamp -> Timestamptz,
         price -> Numeric,
+        timestamp -> Timestamptz,
+        publisher -> Text,
+        publisher_signature -> Nullable<Text>,
         source -> Varchar,
-        publisher_signature -> Nullable<Varchar>,
+    }
+}
+
+diesel::table! {
+    funding_rates (id, timestamp) {
+        id -> Uuid,
+        source -> Varchar,
+        pair -> Varchar,
+        annualized_rate -> Float8,
+        timestamp -> Timestamptz,
+        created_at -> Timestamptz,
     }
 }
 
@@ -20,7 +31,7 @@ diesel::table! {
         timestamp -> Timestamptz,
         expiration_timestamp -> Nullable<Timestamptz>,
         publisher -> Text,
-        publisher_signature -> Text,
+        publisher_signature -> Nullable<Text>,
         source -> Varchar,
     }
 }
@@ -31,9 +42,9 @@ diesel::table! {
         name -> Varchar,
         master_key -> Varchar,
         active_key -> Varchar,
-        active -> Bool,
         account_address -> Varchar,
+        active -> Bool,
     }
 }
 
-diesel::allow_tables_to_appear_in_same_query!(entries, future_entries, publishers,);
+diesel::allow_tables_to_appear_in_same_query!(entries, funding_rates, future_entries, publishers,);

pragma-entities/src/schema.rs

@@ -23,3 +23,4 @@ serde_json = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
 tracing = { workspace = true }
+clap = { workspace = true, features = ["derive"] }

pragma-ingestor/Cargo.toml

@@ -1,21 +1,16 @@
-use serde::{Deserialize, Serialize};
+use clap::Parser;
 use std::sync::LazyLock;
 
-use crate::error::PragmaConsumerError;
+pub(crate) static CONFIG: LazyLock<Ingestor> = LazyLock::new(load_configuration);
 
-pub static CONFIG: LazyLock<Ingestor> = LazyLock::new(load_configuration);
-
-#[derive(Debug, Serialize, Deserialize)]
-pub struct Ingestor {
-    pub num_consumers: usize,
-}
-
-impl Ingestor {
-    pub fn from_env() -> Result<Self, PragmaConsumerError> {
-        envy::from_env::<Self>().map_err(PragmaConsumerError::LoadConfig)
-    }
+#[derive(Parser, Debug)]
+#[command(author, version, about, long_about = None)]
+pub(crate) struct Ingestor {
+    /// Number of consumers to run
+    #[arg(long, env = "NUM_CONSUMERS", default_value = "10")]
+    pub(crate) num_consumers: usize,
 }
 
-pub fn load_configuration() -> Ingestor {
-    Ingestor::from_env().expect("cannot load configuration env")
+pub(crate) fn load_configuration() -> Ingestor {
+    Ingestor::parse()
 }