add etf update thresholds (#642)

models/staging/bitcoin/__bitcoin__sources.yml

@@ -14,6 +14,7 @@ sources:
       - name: raw_bitcoin_etf_addresses
       - name: raw_bitcoin_etf_metadata
       - name: raw_bitcoin_fidelity_outflows
+      - name: raw_etf_update_thresholds
 
   - name: BITCOIN_FLIPSIDE
     schema: core

models/staging/bitcoin/fact_bitcoin_etf_flows.sql

@@ -37,6 +37,7 @@ WITH unaggregated AS (
     INNER JOIN {{ ref('fact_bitcoin_etf_metadata') }} eat ON a.issuer=eat.issuer
     WHERE i.value > 0
     AND i.block_timestamp > '2019-07-24'::date
+    AND i.block_timestamp < (select bitcoin_update_threshold from {{ ref('fact_bitcoin_etf_update_threshold') }})
 
     UNION ALL
 

models/staging/bitcoin/fact_bitcoin_etf_update_threshold.sql

@@ -0,0 +1,20 @@
+{{
+    config(
+        materialized='table',
+        snowflake_warehouse='BITCOIN'
+    )
+}}
+
+-- Credit to @hildobby for the original version of this dataset: https://dune.com/data/dune.hildobby.dataset_etf_update_thresholds
+
+with
+    max_extraction as (
+        select max(extraction_date) as max_date
+        from {{ source("PROD_LANDING", "raw_etf_update_thresholds") }}
+    )
+select
+    left(value:bitcoin, 10)::date as bitcoin_update_threshold
+from
+    {{ source("PROD_LANDING", "raw_etf_update_thresholds") }},
+    lateral flatten(input => parse_json(source_json))
+where extraction_date = (select max_date from max_extraction)

models/staging/ethereum/fact_ethereum_etf_flows.sql

@@ -35,6 +35,7 @@ WITH time_series AS (
     INNER JOIN {{ ref('fact_ethereum_etf_addresses') }} a ON a.address=t.from_address
         AND a.track_outflow = true
     WHERE t.block_timestamp >= date('2017-10-16')
+    AND t.block_timestamp < (select ethereum_update_threshold from {{ ref('fact_ethereum_etf_update_threshold') }})
     GROUP BY 1, 2
 
     UNION ALL

models/staging/ethereum/fact_ethereum_etf_update_threshold.sql

@@ -0,0 +1,20 @@
+{{
+    config(
+        materialized='table',
+        snowflake_warehouse='ETHEREUM'
+    )
+}}
+
+-- Credit to @hildobby for the original version of this dataset: https://dune.com/data/dune.hildobby.dataset_etf_update_thresholds
+
+with
+    max_extraction as (
+        select max(extraction_date) as max_date
+        from {{ source("PROD_LANDING", "raw_etf_update_thresholds") }}
+    )
+select
+    left(value:ethereum, 10)::date as ethereum_update_threshold
+from
+    {{ source("PROD_LANDING", "raw_etf_update_thresholds") }},
+    lateral flatten(input => parse_json(source_json))
+where extraction_date = (select max_date from max_extraction)