Revamp NFT to new sector spell design structure

.github/workflows/dbt_slim_ci.yml

@@ -36,7 +36,7 @@ jobs:
 
       - name: dbt dependencies
         run: "dbt deps"
-      
+
       - name: Activate DuneSQL Cluster
         run: "./scripts/ensure_cluster.sh"
 
@@ -49,7 +49,7 @@ jobs:
           [[ -z "$test" ]] && { echo "Success: All models have a custom schema"; exit 0; } || { echo "Found models without custom schema:"; echo "$test"; exit 1; }
 
       - name: dbt seed
-        run: "dbt seed $PROFILE --select state:modified --exclude tag:prod_exclude tag:remove --state ."
+        run: "dbt seed $PROFILE --select @state:modified --exclude tag:prod_exclude tag:remove --state ."
 
       - name: dbt run initial model(s)
         run: "dbt -x run $PROFILE --select state:modified --exclude tag:prod_exclude tag:remove --defer --state ."

macros/models/_sector/nft/add_nft_tx_data.sql

@@ -0,0 +1,19 @@
+{% macro add_nft_tx_data(
+    model_cte
+    ,blockchain
+    )
+%}
+
+select
+    model.*
+    , tx."from" as tx_from
+    , tx."to" as tx_to
+    , bytearray_reverse(bytearray_substring(bytearray_reverse(tx.data),1,32))  as tx_data_marker
+from {{model_cte}} model
+inner join {{source(blockchain, 'transactions')}} tx
+    on model.block_number = tx.block_number
+    and model.tx_hash = tx.hash
+    {% if is_incremental() %}
+    where {{incremental_predicate('tx.block_time')}}
+    {% endif %}
+{% endmacro %}

macros/models/_sector/nft/enrich_nft_trades.sql

@@ -0,0 +1,112 @@
+{% macro enrich_nft_trades(base_trades) %}
+-- Macro to apply the NFT trades enrichment(s) to base models
+-- 1. add NFT token information
+-- 2. add ERC20 token information + handle ERC20 decimals
+-- 3. handle USD columns
+-- 4. add aggregator columns
+-- 5. fix buyer or seller for aggregator txs
+-- 6. calculate platform and royalty rates
+
+-- TODO: We should remove this CTE and include ETH into the general prices table once everything is migrated
+WITH prices_patch as (
+    SELECT
+        contract_address
+        ,blockchain
+        ,decimals
+        ,minute
+        ,price
+        ,symbol
+    FROM {{ ref('prices_usd_forward_fill') }}
+    WHERE blockchain = 'ethereum'
+    {% if is_incremental() %}
+    AND {{incremental_predicate('minute')}}
+    {% endif %}
+    UNION ALL
+    SELECT
+        {{ var("ETH_ERC20_ADDRESS") }} as contract_address
+        ,'ethereum' as blockchain
+        ,18 as decimals
+        ,minute
+        ,price
+        ,'ETH' as symbol
+    FROM {{ ref('prices_usd_forward_fill') }}
+    WHERE blockchain is null AND symbol = 'ETH'
+    {% if is_incremental() %}
+    AND {{incremental_predicate('minute')}}
+    {% endif %}
+),
+enrichments as (
+SELECT
+    base.blockchain,
+    base.project,
+    base.project_version,
+    base.block_date,
+    base.block_month,
+    base.block_time,
+    base.block_number,
+    base.tx_hash,
+    base.sub_tx_trade_id,
+    base.project_contract_address,
+    base.trade_category,
+    base.trade_type,
+    case when base.buyer = coalesce(agg1.contract_address,agg2.contract_address) then base.tx_from else base.buyer end as buyer,
+    case when base.seller = coalesce(agg1.contract_address,agg2.contract_address) then base.tx_from else base.seller end as seller,
+    base.nft_contract_address,
+    base.nft_token_id,
+    base.nft_amount,
+    base.price_raw,
+    base.currency_contract,
+    base.platform_fee_amount_raw,
+    base.royalty_fee_amount_raw,
+    base.platform_fee_address,
+    base.royalty_fee_address,
+    base.tx_from as tx_from,
+    base.tx_to as tx_to,
+    nft.name as nft_collection,
+    nft.standard as nft_standard,
+    coalesce(erc20.symbol,p.symbol) as currency_symbol,
+    base.price_raw/pow(10,coalesce(erc20.decimals,p.decimals,18)) as price,
+    base.platform_fee_amount_raw/pow(10,coalesce(erc20.decimals,p.decimals,18)) as platform_fee_amount,
+    base.royalty_fee_amount_raw/pow(10,coalesce(erc20.decimals,p.decimals,18)) as royalty_fee_amount,
+    base.price_raw/pow(10,coalesce(erc20.decimals,p.decimals,18))*p.price as price_usd,
+    base.platform_fee_amount_raw/pow(10,coalesce(erc20.decimals,p.decimals,18))*p.price as platform_fee_amount_usd,
+    base.royalty_fee_amount_raw/pow(10,coalesce(erc20.decimals,p.decimals,18))*p.price as royalty_fee_amount_usd,
+    case when base.price_raw > uint256 '0' then cast(100*base.platform_fee_amount_raw/base.price_raw as double) else double '0' end as platform_fee_percentage,
+    case when base.price_raw > uint256 '0' then cast(100*base.royalty_fee_amount_raw/base.price_raw as double) else double '0' end as royalty_fee_percentage,
+    coalesce(agg1.contract_address,agg2.contract_address) as aggregator_address,
+    {% if aggregator_markers != null %}
+    CASE WHEN coalesce(agg_mark.aggregator_name, agg1.name, agg2.name)='Gem' AND base.block_number >= 16971894 THEN 'OpenSea Pro' -- 16971894 is the first block of 2023-04-04 which is when Gem rebranded to OpenSea Pro
+        ELSE coalesce(agg_mark.aggregator_name, agg1.name, agg2.name)
+        END as aggregator_name
+    {% else %}
+    coalesce(agg1.name,agg2.name) as aggregator_name
+    {% endif %}
+FROM {{base_trades}} base
+LEFT JOIN {{ref('tokens_nft')}} nft
+    ON nft.blockchain = base.blockchain
+    AND nft.contract_address = base.nft_contract_address
+LEFT JOIN {{ ref('tokens_erc20') }} erc20
+    ON erc20.blockchain = base.blockchain
+    AND erc20.contract_address = base.currency_contract
+LEFT JOIN prices_patch p
+    ON p.blockchain = base.blockchain
+    AND p.contract_address = base.currency_contract
+    AND p.minute = date_trunc('minute',base.block_time)
+LEFT JOIN {{ ref('nft_aggregators') }} agg1
+    ON agg1.blockchain = base.blockchain
+    AND (base.buyer = agg1.contract_address
+        OR base.seller = agg1.contract_address)
+LEFT JOIN {{ ref('nft_aggregators') }} agg2
+    ON agg1.contract_address is null    -- only match if agg1 produces no matches, this prevents duplicates
+    AND agg2.blockchain = base.blockchain
+    AND tx_to = agg2.contract_address
+LEFT JOIN {{ ref('nft_ethereum_aggregators_markers') }} agg_mark
+    ON bytearray_starts_with(bytearray_reverse(base.tx_data_marker), bytearray_reverse(agg_mark.hash_marker)) -- eq to end_with()
+
+{% if is_incremental() %}
+WHERE {{incremental_predicate('base.block_time')}}
+{% endif %}
+)
+
+select * from enrichments
+{% endmacro %}

macros/models/_sector/nft/platforms/element_v1_base_trades.sql

@@ -1,8 +1,11 @@
 -- Element NFT trades (re-usable macro for all chains)
-{% macro element_v1_base_trades(erc721_sell_order_filled, erc721_buy_order_filled, erc1155_sell_order_filled, erc1155_buy_order_filled) %}
+{% macro element_v1_base_trades(blockchain, erc721_sell_order_filled, erc721_buy_order_filled, erc1155_sell_order_filled, erc1155_buy_order_filled) %}
 
 
 SELECT
+  '{{blockchain}}' as blockchain,
+  'element' as project,
+  'v1' as project_version,
   evt_block_time AS block_time,
   evt_block_number AS block_number,
   'Buy' AS trade_category,
@@ -26,12 +29,15 @@ SELECT
   evt_index AS sub_tx_trade_id
 FROM {{ erc721_sell_order_filled }}
 {% if is_incremental() %}
-WHERE evt_block_time >= date_trunc('day', now() - interval '7' day)
+WHERE {{incremental_predicate('evt_block_time')}}
 {% endif %}
 
 UNION ALL
 
 SELECT
+  '{{blockchain}}' as blockchain,
+  'element' as project,
+  'v1' as project_version,
   evt_block_time AS block_time,
   evt_block_number AS block_number,
   'Sell' AS trade_category,
@@ -55,12 +61,15 @@ SELECT
   evt_index AS sub_tx_trade_id
 FROM {{ erc721_buy_order_filled }}
 {% if is_incremental() %}
-WHERE evt_block_time >= date_trunc('day', now() - interval '7' day)
+WHERE {{incremental_predicate('evt_block_time')}}
 {% endif %}
 
 UNION ALL
 
 SELECT
+  '{{blockchain}}' as blockchain,
+  'element' as project,
+  'v1' as project_version,
   evt_block_time AS block_time,
   evt_block_number AS block_number,
   'Buy' AS trade_category,
@@ -84,12 +93,15 @@ SELECT
   evt_index AS sub_tx_trade_id
 FROM {{ erc1155_buy_order_filled }}
 {% if is_incremental() %}
-WHERE evt_block_time >= date_trunc('day', now() - interval '7' day)
+WHERE {{incremental_predicate('evt_block_time')}}
 {% endif %}
 
 UNION ALL
 
 SELECT
+  '{{blockchain}}' as blockchain,
+  'element' as project,
+  'v1' as project_version,
   evt_block_time AS block_time,
   evt_block_number AS block_number,
   'Buy' AS trade_category,
@@ -113,7 +125,7 @@ SELECT
   evt_index AS sub_tx_trade_id
 FROM {{ erc1155_sell_order_filled }}
 {% if is_incremental() %}
-WHERE evt_block_time >= date_trunc('day', now() - interval '7' day)
+WHERE {{incremental_predicate('evt_block_time')}}
 {% endif %}
 
 {% endmacro %}

macros/models/_sector/nft/port_to_old_schema.sql

@@ -39,7 +39,7 @@ SELECT
     CAST(royalty_fee_amount as double) as royalty_fee_amount,
     royalty_fee_amount_usd,
     royalty_fee_percentage,
-    concat(project,'-',cast(block_number as varchar),'-',cast(tx_hash as varchar),'-',cast(sub_tx_trade_id as varchar)) as unique_trade_id
+    concat(project,'-',project_version,'-',cast(block_number as varchar),'-',cast(tx_hash as varchar),'-',cast(sub_tx_trade_id as varchar)) as unique_trade_id
 FROM {{ model }}
 
 {% endmacro %}

models/_sector/nft/aggregators/chains/nft_ethereum_aggregators_markers.sql

@@ -1,5 +1,5 @@
 {{ config(
-        
+
         schema = 'nft_ethereum',
         alias = 'aggregators_markers',
 		materialized = 'table',
@@ -105,7 +105,10 @@
     ) AS temp_table (hash_marker ,aggregator_name, router_name)
   )
 
-  SELECT *
-    ,length(hash_marker) as hash_marker_size
+  SELECT
+    bytearray_reverse(bytearray_substring(bytearray_reverse(hash_marker),1,32)) as hash_marker  -- limit to 32 bytes, there should not be any collisions.
+    , aggregator_name
+    , router_name
+    , length(hash_marker) as hash_marker_size
   FROM all_markers
 

models/_sector/nft/trades/compatibility_views/nft_events.sql

@@ -0,0 +1,10 @@
+{{ config(
+        schema = 'nft',
+        alias = 'events',
+        materialized = 'view'
+        )
+}}
+
+-- kept for backward compatibility
+SELECT *
+FROM {{ ref('nft_trades') }}