DUX-823 Prices v2 rework

dbt_subprojects/tokens/models/prices_v2/_schema.yml

@@ -0,0 +1,38 @@
+version: 2
+
+models:
+  - name: prices_v2_dex_minute_raw
+    meta:
+      sector: prices
+      contributors: 0xRob
+    description: "sparse minute-level prices sourced from dex.trades"
+    data_tests:
+      - dbt_utils.unique_combination_of_columns:
+          combination_of_columns:
+            - blockchain
+            - contract_address
+            - timestamp
+
+  - name: prices_v2_coinpaprika_minute
+    meta:
+      sector: prices
+      contributors: 0xRob
+    description: "sparse minute-level prices from coinpaprika (only trusted tokens)"
+    data_tests:
+      - dbt_utils.unique_combination_of_columns:
+          combination_of_columns:
+            - blockchain
+            - contract_address
+            - timestamp
+
+  - name: prices_v2_minute_raw
+    meta:
+      sector: prices
+      contributors: 0xRob
+    description: "sparse minute-level prices from all sources"
+    data_tests:
+      - dbt_utils.unique_combination_of_columns:
+          combination_of_columns:
+            - blockchain
+            - contract_address
+            - timestamp

dbt_subprojects/tokens/models/prices_v2/minute/prices_v2_coinpaprika_minute.sql

@@ -0,0 +1,29 @@
+{{ config(
+        schema = 'prices_v2'
+        , alias = 'coinpaprika_minute'
+        , materialized = 'incremental'
+        , file_format = 'delta'
+        , partition_by = ['date']
+        , incremental_strategy = 'merge'
+        , unique_key = ['blockchain', 'contract_address', 'timestamp']
+        , incremental_predicates = [incremental_predicate('DBT_INTERNAL_DEST.timestamp')]
+        )
+}}
+
+select
+    ptt.blockchain
+    , ptt.contract_address
+    , p.minute as timestamp
+    , p.price
+    , cast(null as double) as volume
+    , 'coinpaprika' as source
+    , date_trunc('day', p.minute) as date --partition
+from
+    {{ source('prices','usd_0003') }} as p  -- todo: fix this source
+inner join
+    {{ ref('prices_trusted_tokens') }} as ptt
+    on p.token_id = ptt.token_id
+{% if is_incremental() %}
+where
+    {{ incremental_predicate('p.minute') }}
+{% endif %}

dbt_subprojects/tokens/models/prices_v2/minute/prices_v2_dex_filter.sql

@@ -0,0 +1,32 @@
+{{ config(
+    schema='prices_v2'
+    , alias = 'dex_filter'
+    , materialized = 'view'
+    )
+}}
+
+WITH dex_volume_over_10k as (
+    select
+        blockchain
+        ,contract_address
+    from(
+        SELECT
+            d.blockchain,
+            d.token_bought_address as contract_address,
+            sum(d.amount_usd) as volume -- in USD
+        FROM {{ source('dex','trades') }} d
+        group by 1,2
+        UNION ALL
+        SELECT
+            d.blockchain,
+            d.token_sold_address as contract_address,
+            sum(d.amount_usd) as volume -- in USD
+        FROM {{ source('dex','trades') }} d
+        group by 1,2
+    )
+    group by 1,2
+    having sum(volume) >= 10000
+)
+
+select *
+from dex_volume_over_10k

dbt_subprojects/tokens/models/prices_v2/minute/prices_v2_dex_minute.sql

@@ -0,0 +1,11 @@
+{{ config(
+    schema='prices_v2'
+    , alias = 'dex_minute'
+    , materialized = 'view'
+    )
+}}
+
+SELECT
+*
+FROM {{ ref('prices_v2_dex_minute_raw') }}
+INNER JOIN {{ ref('prices_v2_dex_filter') }} using (blockchain, contract_address)

dbt_subprojects/tokens/models/prices_v2/minute/prices_v2_dex_minute_raw.sql

@@ -0,0 +1,65 @@
+{{ config(
+    schema='prices_v2'
+    , alias = 'dex_minute_raw'
+    , materialized = 'incremental'
+    , file_format = 'delta'
+    , partition_by = ['date']
+    , incremental_strategy = 'merge'
+    , unique_key = ['blockchain', 'contract_address', 'timestamp']
+    , incremental_predicates = [incremental_predicate('DBT_INTERNAL_DEST.timestamp')]
+    )
+}}
+
+WITH dex_trades_filter_and_unnest as (
+    SELECT
+        d.blockchain,
+        d.token_bought_address as contract_address,
+        d.block_time as timestamp,
+        d.amount_usd/d.token_bought_amount as price,
+        d.amount_usd as volume -- in USD
+    FROM {{ source('dex','trades') }} d
+    INNER JOIN {{ref('prices_trusted_tokens')}} t
+        on t.blockchain = d.blockchain
+        and t.contract_address = d.token_sold_address -- the token traded against is trusted
+    LEFT JOIN {{ref('prices_trusted_tokens')}} anti_t
+        on anti_t.blockchain = d.blockchain
+        and anti_t.contract_address = d.token_bought_address -- the subjected token is already in trusted
+    WHERE d.amount_usd > 0 and token_bought_amount > 0 and token_bought_address is not null
+    and anti_t.contract_address is null
+    {% if is_incremental() %}
+    AND {{ incremental_predicate('d.block_time') }}
+    {% endif %}
+
+    UNION ALL
+
+    SELECT
+        d.blockchain,
+        d.token_sold_address as contract_address,
+        d.block_time as timestamp,
+        d.amount_usd/d.token_sold_amount as price,
+        d.amount_usd as volume -- in USD
+    FROM {{ source('dex','trades') }} d
+    INNER JOIN {{ref('prices_trusted_tokens')}} t
+        on t.blockchain = d.blockchain
+        and t.contract_address = d.token_bought_address -- the token traded against is trusted
+    LEFT JOIN {{ref('prices_trusted_tokens')}} anti_t
+        on anti_t.blockchain = d.blockchain
+        and anti_t.contract_address = d.token_sold_address -- the subjected token is already in trusted
+    WHERE d.amount_usd > 0 and token_sold_amount > 0 and token_sold_address is not null
+    and anti_t.contract_address is null
+    {% if is_incremental() %}
+    AND {{ incremental_predicate('d.block_time') }}
+    {% endif %}
+)
+
+
+SELECT
+    blockchain,
+    contract_address,
+    date_trunc('minute',timestamp) as timestamp,
+    sum(price*volume)/sum(volume) as price, -- vwap
+    sum(volume) as volume,
+    'dex.trades' as source,
+    date_trunc('day',timestamp) as date -- partition
+FROM dex_trades_filter_and_unnest
+group by 1,2,3,7

dbt_subprojects/tokens/models/prices_v2/prices_v2_latest.sql

@@ -0,0 +1,23 @@
+{{ config(
+        schema='prices_v2',
+        alias = 'latest',
+        materialized = 'incremental',
+        file_format = 'delta',
+        incremental_strategy = 'merge',
+        unique_key = ['blockchain', 'contract_address']
+        )
+}}
+
+
+SELECT
+    blockchain
+    , contract_address
+    , max(timestamp) as timestamp
+    , max_by(price,timestamp) as price
+    , max_by(volume,timestamp) as volume
+    , max_by(source,timestamp) as source
+FROM {{ ref('prices_v2_minute_raw') }}
+{% if is_incremental() %}
+WHERE {{ incremental_predicate('timestamp') }}
+{% endif %}
+GROUP BY 1,2

dbt_subprojects/tokens/models/prices_v2/prices_v2_minute_raw.sql

@@ -0,0 +1,41 @@
+{{ config(
+        schema='prices_v2',
+        alias = 'minute_raw',
+        file_format = 'delta',
+        materialized = 'incremental',
+        partition_by = ['date'],
+        incremental_strategy = 'merge',
+        unique_key = ['blockchain', 'contract_address', 'timestamp'],
+        incremental_predicates = [incremental_predicate('DBT_INTERNAL_DEST.timestamp')]
+        )
+}}
+
+-- this model feeds into sqlmesh which performs a forward fill and aggregates upto higher timeframes
+
+{% set prices_models = [
+    ref('prices_v2_dex_minute')
+    ,ref('prices_v2_coinpaprika_minute')
+] %}
+
+
+SELECT *
+FROM
+(
+    {% for model in prices_models %}
+    SELECT
+        blockchain
+        , contract_address
+        , timestamp
+        , price
+        , volume    -- can be null
+        , source    -- dex.trades/coinpaprika/..
+        , date      -- partition
+    FROM {{ model }}
+    {% if is_incremental() %}
+    WHERE {{ incremental_predicate('timestamp') }}
+    {% endif %}
+    {% if not loop.last %}
+    UNION ALL
+    {% endif %}
+    {% endfor %}
+)