✨ 2 hour aggregates

pragma-entities/migrations/2024-02-14-160804_add_2hours_interval/down.sql

@@ -0,0 +1,4 @@
+-- This file should undo anything in `up.sql`
+DROP MATERIALIZED VIEW IF EXISTS twap_2_hours_agg;
+DROP MATERIALIZED VIEW IF EXISTS price_2_h_agg;
+DROP MATERIALIZED VIEW IF EXISTS two_hour_candle;

pragma-entities/migrations/2024-02-14-160804_add_2hours_interval/up.sql

@@ -0,0 +1,54 @@
+-- Your SQL goes here
+
+-- aggregate
+CREATE MATERIALIZED VIEW price_2_h_agg
+WITH (timescaledb.continuous, timescaledb.materialized_only = false)
+AS SELECT 
+    pair_id,
+    time_bucket('2 hours'::interval, timestamp) as bucket,
+    approx_percentile(0.5, percentile_agg(price))::numeric AS median_price,
+    COUNT(DISTINCT source) as num_sources
+FROM entries
+GROUP BY bucket, pair_id
+WITH NO DATA;
+
+SELECT add_continuous_aggregate_policy('price_2_h_agg',
+  start_offset => NULL,
+  end_offset => INTERVAL '2 hours',
+  schedule_interval => INTERVAL '2 hours');
+
+-- twap
+CREATE MATERIALIZED VIEW twap_2_hours_agg
+WITH (timescaledb.continuous, timescaledb.materialized_only = false)
+AS SELECT 
+    pair_id,
+    time_bucket('2 hours'::interval, timestamp) as bucket,
+    average(time_weight('Linear', timestamp, price))::numeric as price_twap,
+    COUNT(DISTINCT source) as num_sources
+FROM entries
+GROUP BY bucket, pair_id
+WITH NO DATA;
+
+SELECT add_continuous_aggregate_policy('twap_2_hours_agg',
+  start_offset => NULL,
+  end_offset => INTERVAL '2 hours',
+  schedule_interval => INTERVAL '2 hours');
+
+-- ohlc
+CREATE MATERIALIZED VIEW two_hour_candle
+WITH (timescaledb.continuous) AS
+    SELECT
+        time_bucket('2 hours', 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('two_hour_candle',
+    start_offset => INTERVAL '6 hours',
+    end_offset => INTERVAL '2 hours',
+    schedule_interval => INTERVAL '2 hours');

pragma-entities/migrations/2024-02-14-161427_set_twap_realtime/down.sql

@@ -0,0 +1,4 @@
+-- This file should undo anything in `up.sql`
+ALTER MATERIALIZED VIEW twap_1_min_agg set (timescaledb.materialized_only = true);
+ALTER MATERIALIZED VIEW twap_15_min_agg set (timescaledb.materialized_only = true);
+ALTER MATERIALIZED VIEW twap_1_hour_agg set (timescaledb.materialized_only = true);

pragma-entities/migrations/2024-02-14-161427_set_twap_realtime/up.sql

@@ -0,0 +1,4 @@
+-- Your SQL goes here
+ALTER MATERIALIZED VIEW twap_1_min_agg set (timescaledb.materialized_only = false);
+ALTER MATERIALIZED VIEW twap_15_min_agg set (timescaledb.materialized_only = false);
+ALTER MATERIALIZED VIEW twap_1_hour_agg set (timescaledb.materialized_only = false);

pragma-node/src/handlers/entries/mod.rs

@@ -77,6 +77,8 @@ pub enum Interval {
     FifteenMinutes,
     #[serde(rename = "1h")]
     OneHour,
+    #[serde(rename = "2h")]
+    TwoHours,
 }
 
 // Supported Aggregation Modes

pragma-node/src/infra/repositories/entry_repository.rs

@@ -284,6 +284,24 @@ pub async fn get_twap_price(
         ORDER BY
             time DESC
         LIMIT 1;
+    "#
+        }
+        Interval::TwoHours => {
+            r#"
+            -- query the materialized realtime view
+        SELECT
+            bucket AS time,
+            price_twap AS median_price,
+            num_sources
+        FROM
+            twap_2_hours_agg
+        WHERE
+            pair_id = $1
+            AND
+            bucket <= $2
+        ORDER BY
+            time DESC
+        LIMIT 1;
     "#
         }
     };
@@ -373,6 +391,24 @@ pub async fn get_median_price(
         ORDER BY
             time DESC
         LIMIT 1;
+    "#
+        }
+        Interval::TwoHours => {
+            r#"
+            -- query the materialized realtime view
+        SELECT
+            bucket AS time,
+            median_price,
+            num_sources
+        FROM
+            price_2_h_agg
+        WHERE
+            pair_id = $1
+            AND
+            bucket <= $2
+        ORDER BY
+            time DESC
+        LIMIT 1;
     "#
         }
     };
@@ -582,6 +618,26 @@ pub async fn get_ohlc(
         ORDER BY
             time DESC
         LIMIT 10000;
+    "#
+        }
+        Interval::TwoHours => {
+            r#"
+            -- query the materialized realtime view
+        SELECT
+            bucket AS time,
+            open,
+            high,
+            low,
+            close
+        FROM
+            two_hour_candle
+        WHERE
+            pair_id = $1
+            AND
+            bucket <= $2
+        ORDER BY
+            time DESC
+        LIMIT 10000;
     "#
         }
     };