DAT-2194; Substreams SQL; Uniswap v2; Add DBT models

.gitignore

@@ -9,6 +9,8 @@ Reference/
 **/target/
 **/build/
 **/data/
+**/logs/dbt.log
+**/logs/
 **/*.spkg
 **/buf.gen.yaml
 replay.log

Cargo.toml

@@ -15,6 +15,7 @@ members = [
     "eth-supply",
     "synthetix",
     "aave-v2",
+    "sql/uniswap_v2",
 ]
 exclude = ["messari-cli"]
 

sql/dbt/.gitignore

@@ -0,0 +1,4 @@
+
+target/
+dbt_packages/
+logs/

sql/dbt/README.md

@@ -0,0 +1,15 @@
+Welcome to your new dbt project!
+
+### Using the starter project
+
+Try running the following commands:
+- dbt run
+- dbt test
+
+
+### Resources:
+- Learn more about dbt [in the docs](https://docs.getdbt.com/docs/introduction)
+- Check out [Discourse](https://discourse.getdbt.com/) for commonly asked questions and answers
+- Join the [chat](https://community.getdbt.com/) on Slack for live discussions and support
+- Find [dbt events](https://events.getdbt.com) near you
+- Check out [the blog](https://blog.getdbt.com/) for the latest news on dbt's development and best practices

sql/dbt/dbt_project.yml

@@ -0,0 +1,37 @@
+
+# Name your project! Project names should contain only lowercase characters
+# and underscores. A good package name should reflect your organization's
+# name or the intended use of these models
+name: 'messari_substreams'
+version: '1.0.0'
+config-version: 2
+
+# This setting configures which "profile" dbt uses for this project.
+profile: 'messari_substreams'
+
+# These configurations specify where dbt should look for different types of files.
+# The `model-paths` config, for example, states that models in this project can be
+# found in the "models/" directory. You probably won't need to change these!
+model-paths: ["models"]
+analysis-paths: ["analyses"]
+test-paths: ["tests"]
+seed-paths: ["seeds"]
+macro-paths: ["macros"]
+snapshot-paths: ["snapshots"]
+
+clean-targets:         # directories to be removed by `dbt clean`
+  - "target"
+  - "dbt_packages"
+
+
+# Configuring models
+# Full documentation: https://docs.getdbt.com/docs/configuring-models
+
+# In this example config, we tell dbt to build all models in the example/
+# directory as views. These settings can be overridden in the individual model
+# files using the `{{ config(...) }}` macro.
+models:
+  messari_substreams:
+    # Config indicated by + and applies to all files under models/example/
+    example:
+      +materialized: view

sql/dbt/models/protocol_uniswap_v2_ethereum_mainnet/analytics__protocol_uniswap_v2_ethereum_mainnet__compiled_metrics_deltas_liquidity_pool_input_token_balances.sql

@@ -0,0 +1,48 @@
+{{
+    config(
+        materialized = 'view',
+        alias = 'COMPILED_METRICS_DELTAS_LIQUIDITY_POOL_INPUT_TOKEN_BALANCES',
+        unique_key='id',
+    )
+}}
+
+{% set deltas_input_token_balances = ref('analytics__protocol_uniswap_v2_ethereum_mainnet__entity_modifications_deltas_input_token_balances') %}
+
+WITH
+    incremental_deltas AS (SELECT * FROM {{ deltas_input_token_balances }})
+
+    , final AS (
+        SELECT
+            id.id
+            , id.pool__id
+            , id.token__id
+            , DATE_TRUNC('hour', id.block_timestamp) AS hour
+
+            -- Delta Calculations
+            , id.delta
+            , COALESCE(id.delta / POWER(10, token.decimals) * token.price_usd, 0) AS delta_usd
+
+            -- Cumulative Sum Delta Calculations
+            , SUM(id.delta)  OVER (PARTITION BY id.pool__id, id.token__id ORDER BY id.block_number, id.transaction_index, id.log_index) AS cumulative_sum
+
+            -- Cumulative Sum Delta Calculations (Current Timestamp Price)
+            , COALESCE(cumulative_sum / POWER(10, token.decimals) * token.price_usd, 0) AS cumulative_sum_usd_current_timestamp_price
+
+            , id.transaction_hash
+            , id.block_number
+            , id.block_timestamp
+            , id.transaction_index
+            , id.log_index
+        FROM incremental_deltas id
+        LEFT JOIN {{ token_prices_usd_hour }} token 
+            ON id.token__id = token.contract_address 
+                AND DATE_TRUNC('hour', id.block_timestamp) = token.time
+        ORDER BY 
+            pool__id
+            , token__id
+            , block_number
+            , transaction_index
+            , log_index
+    )
+
+SELECT * FROM final