feat: candlestick views

EvolveArt · EvolveArt · commit e3450100391d · 2024-01-12T15:56:10.000+01:00
diff --git a/pragma-entities/migrations/2024-01-12-135355_add_candlestick_views/down.sql b/pragma-entities/migrations/2024-01-12-135355_add_candlestick_views/down.sql
@@ -0,0 +1,6 @@
+-- This file should undo anything in `up.sql`
+DROP MATERIALIZED VIEW IF EXISTS one_day_candle;
+DROP MATERIALIZED VIEW IF EXISTS one_hour_candle;
+DROP MATERIALIZED VIEW IF EXISTS fifteen_minute_candle;
+DROP MATERIALIZED VIEW IF EXISTS five_minute_candle;
+DROP MATERIALIZED VIEW IF EXISTS one_minute_candle;
diff --git a/pragma-entities/migrations/2024-01-12-135355_add_candlestick_views/up.sql b/pragma-entities/migrations/2024-01-12-135355_add_candlestick_views/up.sql
@@ -0,0 +1,95 @@
+-- 1 day candle
+CREATE MATERIALIZED VIEW one_day_candle
+WITH (timescaledb.continuous) AS
+    SELECT
+        time_bucket('1 day', timestamp) AS bucket,
+        pair_id,
+        FIRST(price, timestamp) AS "open",
+        MAX(price) AS high,
+        MIN(price) AS low,
+        LAST(price, timestamp) AS "close"
+    FROM entries
+    GROUP BY bucket, pair_id
+    WITH NO DATA;
+
+
+SELECT add_continuous_aggregate_policy('one_day_candle',
+    start_offset => INTERVAL '3 days',
+    end_offset => INTERVAL '1 day',
+    schedule_interval => INTERVAL '1 day');
+
+-- 1 hour candle
+CREATE MATERIALIZED VIEW one_hour_candle
+WITH (timescaledb.continuous) AS
+    SELECT
+        time_bucket('1 hour', timestamp) AS bucket,
+        pair_id,
+        FIRST(price, timestamp) AS "open",
+        MAX(price) AS high,
+        MIN(price) AS low,
+        LAST(price, timestamp) AS "close"
+    FROM entries
+    GROUP BY bucket, pair_id
+    WITH NO DATA;
+
+SELECT add_continuous_aggregate_policy('one_hour_candle',
+    start_offset => INTERVAL '3 hours',
+    end_offset => INTERVAL '1 hour',
+    schedule_interval => INTERVAL '1 hour');
+
+-- 15 minute candle
+CREATE MATERIALIZED VIEW fifteen_minute_candle
+WITH (timescaledb.continuous) AS
+    SELECT
+        time_bucket('15 minutes', timestamp) AS bucket,
+        pair_id,
+        FIRST(price, timestamp)::numeric AS "open",
+        MAX(price)::numeric AS high,
+        MIN(price)::numeric AS low,
+        LAST(price, timestamp)::numeric AS "close"
+    FROM entries
+    GROUP BY bucket, pair_id
+    WITH NO DATA;
+
+SELECT add_continuous_aggregate_policy('fifteen_minute_candle',
+    start_offset => INTERVAL '45 minutes',
+    end_offset => INTERVAL '15 minutes',
+    schedule_interval => INTERVAL '15 minutes');
+
+-- 5 minute candle
+CREATE MATERIALIZED VIEW five_minute_candle
+WITH (timescaledb.continuous) AS
+    SELECT
+        time_bucket('5 minutes', timestamp) AS bucket,
+        pair_id,
+        FIRST(price, timestamp) AS "open",
+        MAX(price) AS high,
+        MIN(price) AS low,
+        LAST(price, timestamp) AS "close"
+    FROM entries
+    GROUP BY bucket, pair_id
+    WITH NO DATA;
+
+SELECT add_continuous_aggregate_policy('five_minute_candle',
+    start_offset => INTERVAL '15 minutes',
+    end_offset => INTERVAL '5 minutes',
+    schedule_interval => INTERVAL '5 minutes');
+
+-- 1 minute candle
+CREATE MATERIALIZED VIEW one_minute_candle
+WITH (timescaledb.continuous) AS
+    SELECT
+        time_bucket('1 minute', timestamp) AS bucket,
+        pair_id,
+        FIRST(price, timestamp) AS "open",
+        MAX(price) AS high,
+        MIN(price) AS low,
+        LAST(price, timestamp) AS "close"
+    FROM entries
+    GROUP BY bucket, pair_id
+    WITH NO DATA;
+
+SELECT add_continuous_aggregate_policy('one_minute_candle',
+    start_offset => INTERVAL '3 minutes',
+    end_offset => INTERVAL '1 minute',
+    schedule_interval => INTERVAL '1 minute');
diff --git a/pragma-node/src/handlers/entries/get_entry.rs b/pragma-node/src/handlers/entries/get_entry.rs
@@ -19,7 +19,8 @@ use super::GetEntryParams;
         ),
         params(
             ("quote" = String, Path, description = "Quote Asset"),
-            ("base" = String, Path, description = "Base Asset")
+            ("base" = String, Path, description = "Base Asset"),
+            GetEntryParams,
         ),
     )]
 pub async fn get_entry(
diff --git a/pragma-node/src/handlers/entries/get_ohlc.rs b/pragma-node/src/handlers/entries/get_ohlc.rs
@@ -0,0 +1,73 @@
+use axum::extract::{Query, State};
+use axum::Json;
+
+use crate::handlers::entries::{GetOHLCResponse, Interval};
+use crate::infra::repositories::entry_repository::{self, OHLCEntry};
+use crate::utils::PathExtractor;
+use crate::AppState;
+use pragma_entities::{error::InfraError, EntryError};
+
+use super::utils::currency_pair_to_pair_id;
+use super::GetEntryParams;
+
+#[utoipa::path(
+        get,
+        path = "/node/v1/aggregation/candlestick/{base}/{quote}",
+        responses(
+            (status = 200, description = "Get OHLC data successfuly", body = [GetOHLCResponse])
+        ),
+        params(
+            ("quote" = String, Path, description = "Quote Asset"),
+            ("base" = String, Path, description = "Base Asset"),
+            GetEntryParams,
+        ),
+    )]
+pub async fn get_ohlc(
+    State(state): State<AppState>,
+    PathExtractor(pair): PathExtractor<(String, String)>,
+    Query(params): Query<GetEntryParams>,
+) -> Result<Json<GetOHLCResponse>, EntryError> {
+    tracing::info!("Received get entry request for pair {:?}", pair);
+    // Construct pair id
+    let pair_id = currency_pair_to_pair_id(&pair.1, &pair.0);
+
+    let now = chrono::Utc::now().naive_utc().timestamp_millis() as u64;
+
+    let timestamp = if let Some(timestamp) = params.timestamp {
+        timestamp
+    } else {
+        now
+    };
+
+    let interval = if let Some(interval) = params.interval {
+        interval
+    } else {
+        Interval::OneMinute
+    };
+
+    // Validate given timestamp
+    if timestamp > now {
+        return Err(EntryError::InvalidTimestamp);
+    }
+
+    let entries = entry_repository::get_ohlc(&state.pool, pair_id.clone(), interval, timestamp)
+        .await
+        .map_err(|db_error| to_entry_error(db_error, &pair_id))?;
+
+    Ok(Json(adapt_entry_to_entry_response(pair_id, &entries)))
+}
+
+fn adapt_entry_to_entry_response(pair_id: String, entries: &[OHLCEntry]) -> GetOHLCResponse {
+    GetOHLCResponse {
+        pair_id,
+        data: entries.to_vec(),
+    }
+}
+
+fn to_entry_error(error: InfraError, pair_id: &String) -> EntryError {
+    match error {
+        InfraError::InternalServerError => EntryError::InternalServerError,
+        InfraError::NotFound => EntryError::NotFound(pair_id.to_string()),
+        InfraError::InvalidTimeStamp => EntryError::InvalidTimestamp,
+    }
+}
diff --git a/pragma-node/src/handlers/entries/mod.rs b/pragma-node/src/handlers/entries/mod.rs
@@ -1,14 +1,18 @@
 use serde::{Deserialize, Serialize};
 
 use starknet::core::types::FieldElement;
-use utoipa::ToSchema;
+use utoipa::{IntoParams, ToSchema};
 
 pub use create_entry::create_entries;
 pub use get_entry::get_entry;
+pub use get_ohlc::get_ohlc;
 pub use get_volatility::get_volatility;
 
+use crate::infra::repositories::entry_repository::OHLCEntry;
+
 pub mod create_entry;
 pub mod get_entry;
+pub mod get_ohlc;
 pub mod get_volatility;
 
 pub mod utils;
@@ -48,6 +52,12 @@ pub struct GetEntryResponse {
     decimals: u32,
 }
 
+#[derive(Debug, Serialize, Deserialize, ToSchema)]
+pub struct GetOHLCResponse {
+    pair_id: String,
+    data: Vec<OHLCEntry>,
+}
+
 #[derive(Debug, Serialize, Deserialize, ToSchema)]
 pub struct GetVolatilityResponse {
     pair_id: String,
@@ -58,7 +68,7 @@ pub struct GetVolatilityResponse {
 /// Query parameters structs
 
 // Define an enum for the allowed intervals
-#[derive(Default, Debug, Deserialize)]
+#[derive(Default, Debug, Deserialize, ToSchema)]
 pub enum Interval {
     #[serde(rename = "1min")]
     #[default]
@@ -69,7 +79,7 @@ pub enum Interval {
     OneHour,
 }
 
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Deserialize, IntoParams, ToSchema)]
 pub struct GetEntryParams {
     pub timestamp: Option<u64>,
     pub interval: Option<Interval>,
diff --git a/pragma-node/src/infra/repositories/entry_repository.rs b/pragma-node/src/infra/repositories/entry_repository.rs
@@ -243,3 +243,125 @@ pub async fn get_decimals(
 
     Ok(decimals.to_u32().unwrap())
 }
+
+#[derive(Debug, Clone, Serialize, Deserialize, Queryable)]
+pub struct OHLCEntry {
+    pub time: NaiveDateTime,
+    pub open: BigDecimal,
+    pub low: BigDecimal,
+    pub high: BigDecimal,
+    pub close: BigDecimal,
+}
+
+#[derive(Serialize, QueryableByName, Clone, Debug)]
+pub struct OHLCEntryRaw {
+    #[diesel(sql_type = diesel::sql_types::Timestamptz)]
+    pub time: NaiveDateTime,
+    #[diesel(sql_type = diesel::sql_types::Numeric)]
+    pub open: BigDecimal,
+    #[diesel(sql_type = diesel::sql_types::Numeric)]
+    pub high: BigDecimal,
+    #[diesel(sql_type = diesel::sql_types::Numeric)]
+    pub low: BigDecimal,
+    #[diesel(sql_type = diesel::sql_types::Numeric)]
+    pub close: BigDecimal,
+}
+
+pub async fn get_ohlc(
+    pool: &deadpool_diesel::postgres::Pool,
+    pair_id: String,
+    interval: Interval,
+    time: u64,
+) -> Result<Vec<OHLCEntry>, InfraError> {
+    let conn = pool.get().await.map_err(adapt_infra_error)?;
+
+    let raw_sql = match interval {
+        Interval::OneMinute => {
+            r#"
+        -- query the materialized realtime view
+        SELECT
+            bucket AS time,
+            open,
+            high,
+            low,
+            close
+        FROM
+            one_minute_candle
+        WHERE
+            pair_id = $1
+            AND
+            bucket <= $2
+        ORDER BY
+            time DESC
+        LIMIT 10000;
+    "#
+        }
+        Interval::FifteenMinutes => {
+            r#"
+        -- query the materialized realtime view
+        SELECT
+            bucket AS time,
+            open,
+            high,
+            low,
+            close
+        FROM
+            fifteen_minute_candle
+        WHERE
+            pair_id = $1
+            AND
+            bucket <= $2
+        ORDER BY
+            time DESC
+        LIMIT 10000;
+    "#
+        }
+        Interval::OneHour => {
+            r#"
+        -- query the materialized realtime view
+        SELECT
+            bucket AS time,
+            open,
+            high,
+            low,
+            close
+        FROM
+            one_hour_candle
+        WHERE
+            pair_id = $1
+            AND
+            bucket <= $2
+        ORDER BY
+            time DESC
+        LIMIT 10000;
+    "#
+        }
+    };
+
+    let date_time =
+        NaiveDateTime::from_timestamp_millis(time as i64).ok_or(InfraError::InvalidTimeStamp)?;
+
+    let raw_entries = conn
+        .interact(move |conn| {
+            diesel::sql_query(raw_sql)
+                .bind::<diesel::sql_types::Text, _>(pair_id)
+                .bind::<diesel::sql_types::Timestamptz, _>(date_time)
+                .load::<OHLCEntryRaw>(conn)
+        })
+        .await
+        .map_err(adapt_infra_error)?
+        .map_err(adapt_infra_error)?;
+
+    let entries: Vec<OHLCEntry> = raw_entries
+        .into_iter()
+        .map(|raw_entry| OHLCEntry {
+            time: raw_entry.time,
+            open: raw_entry.open,
+            high: raw_entry.high,
+            low: raw_entry.low,
+            close: raw_entry.close,
+        })
+        .collect();
+
+    Ok(entries)
+}
diff --git a/pragma-node/src/main.rs b/pragma-node/src/main.rs
@@ -30,12 +30,14 @@ async fn main() {
         paths(
             handlers::entries::create_entry::create_entries,
             handlers::entries::get_entry::get_entry,
+            handlers::entries::get_ohlc::get_ohlc,
             handlers::entries::get_volatility::get_volatility,
         ),
         components(
             schemas(pragma_entities::dto::Entry, pragma_entities::EntryError),
             schemas(pragma_entities::dto::Publisher, pragma_entities::PublisherError),
-            schemas(handlers::entries::CreateEntryRequest, handlers::entries::CreateEntryResponse, handlers::entries::GetEntryResponse, handlers::entries::GetVolatilityResponse),
+            schemas(handlers::entries::CreateEntryRequest, handlers::entries::CreateEntryResponse, handlers::entries::GetEntryResponse, handlers::entries::GetVolatilityResponse, handlers::entries::GetOHLCResponse),
+            schemas(handlers::entries::GetEntryParams, handlers::entries::Interval),
             schemas(handlers::entries::Entry, handlers::entries::BaseEntry),
             schemas(pragma_entities::error::InfraError),
         ),
diff --git a/pragma-node/src/routes.rs b/pragma-node/src/routes.rs
