Introduce dex_aggregator_base_trades

dbt_subprojects/dex/models/aggregator_trades/_schema.yml

@@ -92,4 +92,47 @@ models:
         name: trace_address
       - &evt_index
         name: evt_index
-        description: "Index of the corresponding trade event"
+        description: "Index of the corresponding trade event"
+
+  - name: dex_aggregator_base_trades
+    meta:
+      blockchain: ethereum, gnosis, avalanche_c, fantom, optimism, arbitrum, bnb
+      sector: dex_aggregator
+      contributors: hosuke
+    config:
+      tags: ['ethereum', 'gnosis', 'avalanche_c', 'fantom', 'aggregator', 'dex', 'trades', 'cross-chain']
+    description: >
+      Base trades data without amount_usd column
+    tests:
+      - dbt_utils.unique_combination_of_columns:
+          combination_of_columns:
+            - block_date
+            - blockchain
+            - project
+            - version
+            - tx_hash
+            - evt_index
+            - trace_address
+    columns:
+      - *blockchain
+      - *project
+      - *version
+      - *block_date
+      - *block_time
+      - *token_bought_symbol
+      - *token_sold_symbol
+      - *token_pair
+      - *token_bought_amount
+      - *token_sold_amount
+      - *token_bought_amount_raw
+      - *token_sold_amount_raw
+      - *token_bought_address
+      - *token_sold_address
+      - *taker
+      - *maker
+      - *project_contract_address
+      - *tx_hash
+      - *tx_from
+      - *tx_to
+      - *trace_address
+      - *evt_index

dbt_subprojects/dex/models/aggregator_trades/dex_aggregator_base_trades.sql

@@ -0,0 +1,61 @@
+{{ config(
+    schema ='dex_aggregator',
+    , alias = 'base_trades'
+    , partition_by = ['block_month', 'blockchain', 'project']
+    , materialized = 'incremental'
+    , file_format = 'delta'
+    , incremental_strategy = 'merge'
+    , unique_key = ['blockchain', 'project', 'version', 'tx_hash', 'evt_index', 'trace_address']
+    , incremental_predicates = [incremental_predicate('DBT_INTERNAL_DEST.block_time')]
+    )
+}}
+
+{% set base_trade_models = [
+] %}
+
+with base_union as (
+    SELECT *
+    FROM
+    (
+        {% for model in base_trade_models %}
+        SELECT
+            blockchain
+            , project
+            , version
+            , block_date
+            , block_month
+            , block_time
+            , token_bought_symbol
+            , token_sold_symbol
+            , token_pair
+            , token_bought_amount
+            , token_sold_amount
+            , try_cast(token_bought_amount_raw as uint256) as token_bought_amount_raw
+            , try_cast(token_sold_amount_raw as uint256) as token_sold_amount_raw
+            , token_bought_address
+            , token_sold_address
+            , taker
+            , maker
+            , project_contract_address
+            , tx_hash
+            , tx_from
+            , tx_to
+            , trace_address
+            , evt_index
+        FROM
+            {{ model }}
+        WHERE
+           token_sold_amount_raw >= 0 and token_bought_amount_raw >= 0
+        {% if is_incremental() %}
+            AND {{ incremental_predicate('block_time') }}
+        {% endif %}
+        {% if not loop.last %}
+        UNION ALL
+        {% endif %}
+        {% endfor %}
+    )
+)
+select
+    *
+from
+    base_union

dbt_subprojects/dex/models/aggregator_trades/dex_aggregator_trades.sql

@@ -10,11 +10,11 @@
         post_hook='{{ expose_spells(\'["ethereum", "gnosis", "avalanche_c", "fantom", "bnb", "optimism", "arbitrum"]\',
                                 "sector",
                                 "dex_aggregator",
-                                \'["bh2smith", "Henrystats", "jeff-dude", "rantum" ]\') }}'
+                                \'["bh2smith", "Henrystats", "jeff-dude", "rantum", "hosuke"]\') }}'
         )
 }}
 
-{% set dex_aggregator_models = [
+{% set as_is_models = [
     ref('cow_protocol_trades')
     ,ref('paraswap_trades')
     ,ref('lifi_trades')
@@ -30,37 +30,96 @@
     ,ref('odos_trades')
 ] %}
 
-{% for aggregator_model in dex_aggregator_models %}
-SELECT
-    blockchain
-    , project
-    , version
-    , block_date
-    , block_month
-    , block_time
-    , token_bought_symbol
-    , token_sold_symbol
-    , token_pair
-    , token_bought_amount
-    , token_sold_amount
-    , try_cast(token_bought_amount_raw as uint256) as token_bought_amount_raw
-    , try_cast(token_sold_amount_raw as uint256) as token_sold_amount_raw
-    , amount_usd
-    , token_bought_address
-    , token_sold_address
-    , taker
-    , maker
-    , project_contract_address
-    , tx_hash
-    , tx_from
-    , tx_to
-    , trace_address
-    , evt_index
-FROM {{ aggregator_model }}
-{% if is_incremental() %}
-WHERE block_date >= date_trunc('day', now() - interval '7' day)
-{% endif %}
-{% if not loop.last %}
-UNION ALL
-{% endif %}
+WITH aggregator_base_trades AS (
+    SELECT *
+    FROM ref('dex_aggregator_base_trades')
+    {% if is_incremental() %}
+    WHERE {{ incremental_predicate('block_time') }}
+    {% endif %
+)
+
+, enrriched_aggregator_base_trades AS (
+    {{
+        add_amount_usd(
+            trades_cte = 'aggregator_base_trades'
+        )
+    }}
+)
+
+, as_is_dexs AS (
+    {% for model in as_is_models %}
+    SELECT
+        blockchain
+        , project
+        , version
+        , block_date
+        , block_month
+        , block_time
+        , token_bought_symbol
+        , token_sold_symbol
+        , token_pair
+        , token_bought_amount
+        , token_sold_amount
+        , try_cast(token_bought_amount_raw as uint256) as token_bought_amount_raw
+        , try_cast(token_sold_amount_raw as uint256) as token_sold_amount_raw
+        , amount_usd
+        , token_bought_address
+        , token_sold_address
+        , taker
+        , maker
+        , project_contract_address
+        , tx_hash
+        , tx_from
+        , tx_to
+        , trace_address
+        , evt_index
+    FROM
+        {{ model }}
+    {% if is_incremental() %}
+    WHERE {{ incremental_predicate('block_time') }}
+    {% endif %}
+    {% if not loop.last %}
+    UNION ALL
+    {% endif %}
+    {% endfor %}
+)
+                                                                                    en
+{% set cte_to_union = [
+    'enrriched_aggregator_base_trades'
+    , 'as_is_dexs'
+    , 'dexs'
+    ]
+%}
+
+{% for cte in cte_to_union %}
+    SELECT
+        blockchain
+        , project
+        , version
+        , block_date
+        , block_month
+        , block_time
+        , token_bought_symbol
+        , token_sold_symbol
+        , token_pair
+        , token_bought_amount
+        , token_sold_amount
+        , token_bought_amount_raw
+        , token_sold_amount_raw
+        , amount_usd
+        , token_bought_address
+        , token_sold_address
+        , taker
+        , maker
+        , project_contract_address
+        , tx_hash
+        , tx_from
+        , tx_to
+        , trace_address
+        , evt_index
+    FROM
+        {{ cte }}
+    {% if not loop.last %}
+    UNION ALL
+    {% endif %}
 {% endfor %}