Alter Swap model to make Candlestick views with timescaledb.continuous

src/demo_uniswap/models/__init__.py

@@ -46,7 +46,7 @@ class Token(CachedModel):
     # volume in token units
     volume = fields.DecimalField(decimal_places=18, max_digits=96, default=0)
     # volume in derived USD
-    volume_usd = fields.DecimalField(decimal_places=18, max_digits=96, default=0)
+    volume_usd = fields.DecimalField(decimal_places=18, max_digits=96, default=0, index=True)
     # volume in USD even on pools with less reliable USD values
     untracked_volume_usd = fields.DecimalField(decimal_places=18, max_digits=96, default=0)
     # fees in USD
@@ -336,7 +336,7 @@ class Swap(Model):
     # pointer to transaction
     transaction_hash = fields.TextField()
     # timestamp of transaction
-    timestamp = fields.BigIntField()
+    timestamp = fields.DatetimeField(index=True)
     # pool swap occured within
     pool: fields.ForeignKeyRelation[Pool] = fields.ForeignKeyField('models.Pool', related_name='swaps')
     # allow indexing by tokens
@@ -410,4 +410,4 @@ class Flash(Model):
     # amount token1 paid for flash
     amount1_paid = fields.DecimalField(decimal_places=18, max_digits=96, default=0)
     # index within the txn
-    log_index = fields.BigIntField()
+    log_index = fields.BigIntField()

src/demo_uniswap/sql/on_reindex/20_create_ca_quotes_1m.sql

@@ -1,25 +1,36 @@
--- CREATE MATERIALIZED VIEW
---     candlestick_1m
--- WITH (timescaledb.continuous) AS
---
--- SELECT
---     time_bucket('1 minute'::INTERVAL, to_timestamp(timestamp)) AS bucket,
---     token0_id as token_id,
---     candlestick_agg(
---         to_timestamp(timestamp),
---         abs(amount_usd/amount0),
---         amount_usd
---     ) as candlestick
--- FROM swap
---     WHERE
---         amount_usd!=0
---     AND
---         amount0!=0
---
--- GROUP BY
---     bucket,
---     token0_id
--- ORDER BY
---     bucket,
---     token0_id
--- WITH NO DATA;
+CREATE MATERIALIZED VIEW
+    candlestick_1m
+WITH (timescaledb.continuous) AS
+
+SELECT
+    time_bucket('1 minute'::INTERVAL, timestamp) AS bucket,
+    token0_id as token_id,
+    candlestick_agg(
+        timestamp,
+        abs(amount_usd/amount0),
+        amount_usd
+    ) as candlestick
+FROM swap
+    WHERE
+        amount_usd!=0
+    AND
+        amount0!=0
+
+GROUP BY
+    bucket,
+    token0_id
+ORDER BY
+    bucket,
+    token0_id
+WITH NO DATA;
+
+CREATE INDEX candlestick_1m_bucket ON candlestick_1m(bucket);
+CREATE INDEX candlestick_1m_token_id ON candlestick_1m(token_id);
+
+SELECT add_continuous_aggregate_policy(
+    'candlestick_1m',
+    start_offset => INTERVAL '1 hour',
+    end_offset => INTERVAL '0 minutes',
+    schedule_interval => INTERVAL '1 minute',
+    initial_start := '2018-07-01'
+);

src/demo_uniswap/sql/on_reindex/21_create_ca_quotes_1h.sql

@@ -0,0 +1,36 @@
+CREATE MATERIALIZED VIEW
+    candlestick_1h
+WITH (timescaledb.continuous) AS
+
+SELECT
+    time_bucket('1 hour'::INTERVAL, timestamp) AS bucket,
+    token0_id as token_id,
+    candlestick_agg(
+        timestamp,
+        abs(amount_usd/amount0),
+        amount_usd
+    ) as candlestick
+FROM swap
+    WHERE
+        amount_usd!=0
+    AND
+        amount0!=0
+
+GROUP BY
+    bucket,
+    token0_id
+ORDER BY
+    bucket,
+    token0_id
+WITH NO DATA;
+
+CREATE INDEX candlestick_1h_bucket ON candlestick_1h(bucket);
+CREATE INDEX candlestick_1h_token_id ON candlestick_1h(token_id);
+
+SELECT add_continuous_aggregate_policy(
+    'candlestick_1h',
+    start_offset => INTERVAL '2 hour',
+    end_offset => INTERVAL '0 minutes',
+    schedule_interval => INTERVAL '1 hour',
+    initial_start := '2018-07-01'
+);

src/demo_uniswap/sql/on_reindex/22_create_ca_quotes_1d.sql

@@ -0,0 +1,36 @@
+CREATE MATERIALIZED VIEW
+    candlestick_1d
+WITH (timescaledb.continuous) AS
+
+SELECT
+    time_bucket('1 day'::INTERVAL, timestamp) AS bucket,
+    token0_id as token_id,
+    candlestick_agg(
+        timestamp,
+        abs(amount_usd/amount0),
+        amount_usd
+    ) as candlestick
+FROM swap
+    WHERE
+        amount_usd!=0
+    AND
+        amount0!=0
+
+GROUP BY
+    bucket,
+    token0_id
+ORDER BY
+    bucket,
+    token0_id
+WITH NO DATA;
+
+CREATE INDEX candlestick_1d_bucket ON candlestick_1d(bucket);
+CREATE INDEX candlestick_1d_token_id ON candlestick_1d(token_id);
+
+SELECT add_continuous_aggregate_policy(
+    'candlestick_1d',
+    start_offset => INTERVAL '2 days',
+    end_offset => INTERVAL '0 minutes',
+    schedule_interval => INTERVAL '1 hour',
+    initial_start := '2018-07-01'
+);