T 12801 add uniswap abstraction (#1111)

This PR adds a canonical example of Uniswap v2 and v3 trades, migrated from Postgres to Spark.
It build two incremental tables: uniswap_ethereum_v2.trades and uniswap_ethereum_v3.trades, that are then UNIONed to create the uniswap_ethereum.trades view exposed in the Dune App.

It also includes tests against Postgres data from dex.trades

I've checked that:

* [x] I tested the query on dune.com after compiling the model with dbt compile (compiled queries are written to the target directory)
* [x] I used "refs" to reference other models in this repo and "sources" to reference raw or decoded tables 
* [x] the directory tree matches the pattern /sector/blockchain/ e.g. /tokens/ethereum
* [x] if adding a new model, I added a test
* [x] the filename is unique and ends with .sql
* [x] each file has only one view, table or function defined  
* [x] column names are `lowercase_snake_cased`

spellbook/dbt_project.yml

@@ -29,6 +29,12 @@ models:
     nft:
       +schema: nft
       +materialized: view
+    uniswap:
+      +schema: uniswap
+      +materialized: view
+      ethereum:
+        +schema: uniswap_ethereum
+        +materialized: view
     opensea:
       +schema: opensea
       +materialized: view

spellbook/macros/alter_table_properties.sql

@@ -109,6 +109,15 @@ ALTER VIEW tokens_ethereum.nft SET TBLPROPERTIES('dune.public'='true',
                                                     'dune.data_explorer.contributors'='["dot2dotseurat","soispoke"]');
 {% endset %}
 
+{% set uniswap_ethereum_trades %}
+ALTER VIEW uniswap_ethereum.trades SET TBLPROPERTIES('dune.public'='true',
+                                                    'dune.data_explorer.blockchains'='["ethereum"]',
+                                                    'dune.data_explorer.category'='abstraction',
+                                                    'dune.data_explorer.abstraction.type'='project',
+                                                    'dune.data_explorer.abstraction.name'='uniswap',
+                                                    'dune.data_explorer.contributors'='["soispoke"]');
+{% endset %}
+
 {% do run_query(balances_ethereum_erc20_hour) %}
 {% do run_query(balances_ethereum_erc20_latest) %}
 {% do run_query(magiceden_trades) %}
@@ -120,6 +129,7 @@ ALTER VIEW tokens_ethereum.nft SET TBLPROPERTIES('dune.public'='true',
 {% do run_query(tokens_ethereum_erc20) %}
 {% do run_query(transfers_ethereum_erc20) %}
 {% do run_query(tokens_ethereum_nft) %}
+{% do run_query(uniswap_ethereum_trades) %}
 
 {% do log("Tables generated", info=True) %}
 {%- else -%}

spellbook/macros/optimize_tables.sql

@@ -20,11 +20,21 @@ OPTIMIZE opensea_solana.trades;
 OPTIMIZE magiceden_solana.trades;
 {% endset %}
 
+{% set uniswap_v2_ethereum_trades %}
+OPTIMIZE uniswap_v2_ethereum.trades;
+{% endset %}
+
+{% set uniswap_v3_ethereum_trades %}
+OPTIMIZE uniswap_v3_ethereum.trades;
+{% endset %}
+
 {% do run_query(transfers_ethereum_erc20_agg_hour) %}
 {% do run_query(transfers_ethereum_erc20_agg_day) %}
 {% do run_query(opensea_ethereum_trades) %}
 {% do run_query(opensea_solana_trades) %}
 {% do run_query(magiceden_solana_trades) %}
+{% do run_query(uniswap_v2_ethereum_trades) %}
+{% do run_query(uniswap_v3_ethereum_trades) %}
 
 {% do log("Tables Optimized", info=True) %}
 {%- else -%}

spellbook/models/uniswap/ethereum/uniswap_ethereum_schema.yml

@@ -0,0 +1,131 @@
+version: 2
+
+models:
+  - name: uniswap_ethereum_trades
+    meta:
+      blockchain: ethereum
+      project: uniswap
+      contributors: masquot, soispoke
+    config:
+      tags: ['ethereum','uniswap','trades']
+    description: >
+        Uniswap V1 trades on Ethereum
+    columns:
+      - &blockchain
+        name: blockchain
+        description: "Blockchain"
+      - &project 
+        name: project
+        description:  "Project name"  
+      - &version
+        name: version
+        description:  "Version"  
+      - &block_time
+        name: block_time
+        description: "UTC event block time"
+      - &token_a_symbol
+        name: token_a_symbol
+        description: "Symbol for first token traded"
+      - &token_b_symbol
+        name: token_b_symbol
+        description: "Symbol for second token traded"
+      - &token_a_amount
+        name: token_a_amount
+        description: "Value of the trade at time of execution in the original currency for the first token"
+      - &token_b_amount
+        name: token_b_amount
+        description: "Value of the trade at time of execution in the original currency for the second token"
+      - &trader_a
+        name: trader_a
+        description: "Address of trader A"
+      - &trader_b
+        name: trader_b
+        description: "Address of trader B"
+      - &amount_usd
+        name: amount_usd
+        description:  "USD value of the trade at time of execution"
+      - &token_a_address
+        name: token_a_address
+        description: "Contract address of first token traded"
+      - &token_b_address
+        name: token_b_address
+        description: "Contract address of second token traded"
+      - &exchange_contract_address
+        name: exchange_contract_address
+        description:  "Exchange contract address"
+      - &tx_hash
+        name: tx_hash
+        description:  "Transaction hash"
+      - &tx_from
+        name: tx_from
+        description:  "Initiated this transaction"
+      - &tx_to
+        name: tx_to
+        description:  "Receive this transaction"   
+      - &trade_id
+        name: trade_id
+        description:  "Trade ID (derived from evt_index for ethereum dex trades)"      
+
+  - name: uniswap_v2_ethereum_trades
+    meta:
+      blockchain: ethereum
+      project: uniswap_v2
+      contributors: masquot, soispoke
+    config:
+      tags: ['ethereum','uniswap_v2','trades']
+    description: >
+        Uniswap V2 trades on Ethereum
+    columns:
+      - &unique_id
+        name: unique_id
+        description:  "Unique id, combination of tx_hash and trade_id"
+        tests:
+          - unique
+      - *blockchain
+      - *project
+      - *version  
+      - *block_time
+      - *token_a_symbol
+      - *token_b_symbol
+      - *token_a_amount
+      - *token_b_amount
+      - *trader_a
+      - *trader_b
+      - *amount_usd
+      - *token_a_address
+      - *token_b_address
+      - *exchange_contract_address
+      - *tx_hash
+      - *tx_from
+      - *tx_to
+      - *trade_id      
+
+  - name: uniswap_v3_ethereum_trades
+    meta:
+      blockchain: ethereum
+      project: uniswap_v3
+      contributors: masquot, soispoke
+    config:
+      tags: ['ethereum','uniswap_v3','trades']
+    description: >
+        Uniswap V3 trades on Ethereum
+    columns:
+      - *unique_id
+      - *blockchain
+      - *project
+      - *version  
+      - *block_time
+      - *token_a_symbol
+      - *token_b_symbol
+      - *token_a_amount
+      - *token_b_amount
+      - *trader_a
+      - *trader_b
+      - *amount_usd
+      - *token_a_address
+      - *token_b_address
+      - *exchange_contract_address
+      - *tx_hash
+      - *tx_from
+      - *tx_to
+      - *trade_id    

spellbook/models/uniswap/ethereum/uniswap_ethereum_sources.yml

@@ -0,0 +1,22 @@
+version: 2
+
+sources: 
+  - name: uniswap_v2_ethereum
+    freshness:
+      warn_after: { count: 12, period: hour }
+      error_after: { count: 24, period: hour }
+    tables:
+      - name: pair_evt_swap
+        loaded_at_field: evt_block_time
+      - name: factory_evt_paircreated
+        loaded_at_field: evt_block_time
+
+  - name: uniswap_v3_ethereum
+    freshness:
+      warn_after: { count: 12, period: hour }
+      error_after: { count: 24, period: hour }
+    tables:
+      - name: pair_evt_swap
+        loaded_at_field: evt_block_time
+      - name: factory_evt_poolcreated
+        loaded_at_field: evt_block_time

spellbook/models/uniswap/ethereum/uniswap_ethereum_trades.sql

@@ -0,0 +1,12 @@
+ {{
+  config(
+        alias='trades'
+  )
+}}
+
+SELECT blockchain, project, version, block_time, token_a_symbol, token_b_symbol, 
+       token_a_amount, token_b_amount, trader_a, trader_b, usd_amount, token_a_address, 
+       token_b_address, exchange_contract_address, tx_hash, tx_from, tx_to, trade_id 
+FROM (SELECT * FROM {{ ref('uniswap_v2_ethereum_trades') }} 
+UNION ALL
+SELECT * FROM {{ ref('uniswap_v3_ethereum_trades') }}) 

spellbook/models/uniswap/ethereum/uniswap_v2_ethereum_trades.sql

@@ -0,0 +1,71 @@
+ {{
+  config(
+        schema = 'uniswap_v2', 
+        alias='trades',
+        materialized ='incremental',
+        file_format ='delta',
+        incremental_strategy='merge',
+        unique_key='unique_id'
+  )
+}}
+
+SELECT
+    tx_hash || evt_index::string as unique_id,
+    'ethereum' as blockchain,
+    'uniswap' as project, 
+    'v2' as version,
+    dex.block_time,
+    erc20a.symbol AS token_a_symbol,
+    erc20b.symbol AS token_b_symbol,
+    token_a_amount_raw / power(10, erc20a.decimals) AS token_a_amount,
+    token_b_amount_raw / power(10, erc20b.decimals) AS token_b_amount,
+    coalesce(trader_a, tx.from) as trader_a, -- subqueries rely on this COALESCE to avoid redundant joins with the transactions table
+    trader_b,
+    coalesce(
+        usd_amount,
+        token_a_amount_raw / power(10, erc20a.decimals) * pa.price,
+        token_b_amount_raw / power(10, erc20b.decimals) * pb.price
+    ) as usd_amount,
+    token_a_address,
+    token_b_address,
+    exchange_contract_address,
+    tx_hash,
+    tx.from as tx_from,
+    tx.to as tx_to,
+    evt_index as trade_id
+FROM (
+    --Uniswap v2
+    SELECT
+        t.evt_block_time AS block_time,
+        t.to AS trader_a,
+        cast(NULL as string) AS trader_b,
+        -- when amount0 is negative it means trader_a is buying token0 from the pool
+        CASE WHEN amount0Out = 0 THEN amount1Out ELSE amount0Out END AS token_a_amount_raw,
+        CASE WHEN amount0In = 0 OR amount1Out = 0 THEN amount1In ELSE amount0In END AS token_b_amount_raw,
+        NULL::double AS usd_amount,
+        CASE WHEN amount0Out = 0 THEN f.token1 ELSE f.token0 END AS token_a_address,
+        CASE WHEN amount0In = 0 OR amount1Out = 0 THEN f.token1 ELSE f.token0 END AS token_b_address,
+        t.contract_address as exchange_contract_address,
+        t.evt_tx_hash AS tx_hash,
+        t.evt_index
+    FROM {{ source('uniswap_v2_ethereum', 'pair_evt_swap') }} t
+    INNER JOIN {{ source('uniswap_v2_ethereum', 'factory_evt_paircreated') }} f ON f.pair = t.contract_address
+    WHERE  t.contract_address NOT IN (
+            '0xed9c854cb02de75ce4c9bba992828d6cb7fd5c71', -- remove WETH-UBOMB wash trading pair
+            '0xf9c1fA7d41bf44ADe1dd08D37CC68f67Ae75bF92', -- remove WETH-WETH wash trading pair 
+            '0x854373387e41371ac6e307a1f29603c6fa10d872' ) -- remove FEG/ETH token pair
+) dex
+INNER JOIN {{ source('ethereum', 'transactions') }} tx
+    ON dex.tx_hash = tx.hash
+LEFT JOIN {{ ref('tokens_ethereum_erc20') }} erc20a ON erc20a.contract_address = dex.token_a_address
+LEFT JOIN {{ ref('tokens_ethereum_erc20') }} erc20b ON erc20b.contract_address = dex.token_b_address
+LEFT JOIN {{ source('prices', 'usd') }} pa ON pa.minute = date_trunc('minute', dex.block_time)
+    AND pa.contract_address = dex.token_a_address
+    AND pa.blockchain = 'ethereum'
+LEFT JOIN {{ source('prices', 'usd') }} pb ON pb.minute = date_trunc('minute', dex.block_time)
+    AND pb.contract_address = dex.token_b_address
+    AND pb.blockchain = 'ethereum'
+{% if is_incremental() %}
+-- this filter will only be applied on an incremental run
+WHERE dex.block_time > now() - interval 2 days
+{% endif %} 

spellbook/models/uniswap/ethereum/uniswap_v3_ethereum_trades.sql

@@ -0,0 +1,63 @@
+{{config(schema = 'uniswap_v3', 
+        alias='trades',
+        materialized ='incremental',
+        file_format ='delta',
+        incremental_strategy='merge',
+        unique_key='unique_id')
+}}
+
+SELECT
+    tx_hash || evt_index::string as unique_id,
+    'ethereum' as blockchain,
+    'uniswap' as project, 
+    'v3' as version,
+    dex.block_time,
+    erc20a.symbol AS token_a_symbol,
+    erc20b.symbol AS token_b_symbol,
+    token_a_amount_raw / power(10, erc20a.decimals) AS token_a_amount,
+    token_b_amount_raw / power(10, erc20b.decimals) AS token_b_amount,
+    coalesce(trader_a, tx.from) as trader_a, -- subqueries rely on this COALESCE to avoid redundant joins with the transactions table
+    trader_b,
+    coalesce(
+        usd_amount,
+        token_a_amount_raw / power(10, erc20a.decimals) * pa.price,
+        token_b_amount_raw / power(10, erc20b.decimals) * pb.price
+        ) as usd_amount,
+    token_a_address,
+    token_b_address,
+    exchange_contract_address,
+    tx_hash,
+    tx.from as tx_from,
+    tx.to as tx_to,
+    evt_index as trade_id
+    FROM (--Uniswap v3
+    SELECT
+    t.evt_block_time AS block_time,
+    t.recipient AS trader_a,
+    cast(NULL as string) AS trader_b,
+        -- when amount0 is negative it means trader_a is buying token0 from the pool
+        CASE WHEN amount0 < 0 THEN abs(amount0) ELSE abs(amount1) END AS token_a_amount_raw,
+        CASE WHEN amount0 < 0 THEN abs(amount1) ELSE abs(amount0) END AS token_b_amount_raw,
+        NULL::double AS usd_amount,
+        CASE WHEN amount0 < 0 THEN f.token0 ELSE f.token1 END AS token_a_address,
+        CASE WHEN amount0 < 0 THEN f.token1 ELSE f.token0 END AS token_b_address,
+        t.contract_address as exchange_contract_address,
+        t.evt_tx_hash AS tx_hash,
+        t.evt_index
+        FROM {{ source('uniswap_v3_ethereum', 'pair_evt_swap') }} t
+        INNER JOIN {{ source('uniswap_v3_ethereum', 'factory_evt_poolcreated') }} f ON t.evt_tx_hash = f.evt_tx_hash
+        ) dex
+    INNER JOIN {{ source('ethereum', 'transactions') }} tx
+    ON dex.tx_hash = tx.hash
+    LEFT JOIN {{ ref('tokens_ethereum_erc20') }} erc20a ON erc20a.contract_address = dex.token_a_address
+    LEFT JOIN {{ ref('tokens_ethereum_erc20') }} erc20b ON erc20b.contract_address = dex.token_b_address
+    LEFT JOIN {{ source('prices', 'usd') }} pa ON pa.minute = date_trunc('minute', dex.block_time)
+        AND pa.contract_address = dex.token_a_address
+        AND pa.blockchain = 'ethereum'
+    LEFT JOIN {{ source('prices', 'usd') }} pb ON pb.minute = date_trunc('minute', dex.block_time)
+        AND pb.contract_address = dex.token_b_address
+        AND pb.blockchain = 'ethereum'
+{% if is_incremental() %}
+-- this filter will only be applied on an incremental run
+WHERE dex.block_time > now() - interval 2 days
+{% endif %}