Create Safe on Ethereum Balances

dbt_subprojects/hourly_spellbook/models/_project/safe/ethereum/safe_ethereum_schema.yml → dbt_subprojects/hourly_spellbook/models/_project/safe/ethereum/_schema.yml

@@ -6,9 +6,6 @@ models:
       blockchain: ethereum
       project: safe
       contributors: scherbovich, tschubotz, frankmaseo, danielpartida
-    freshness:
-      warn_after: { count: 12, period: hour }
-      error_after: { count: 24, period: hour }
     config:
       tags: ['safe', 'ethereum']
     description: "Safe addresses"
@@ -37,9 +34,6 @@ models:
       blockchain: ethereum
       project: safe
       contributors: scherbovich, tschubotz
-    freshness:
-      warn_after: { count: 12, period: hour }
-      error_after: { count: 24, period: hour }
     config:
       tags: ['safe', 'transfers', 'ethereum']
     description: "Eth transfers for safes"
@@ -75,9 +69,6 @@ models:
       blockchain: ethereum
       project: safe
       contributors: tschubotz
-    freshness:
-      warn_after: { count: 12, period: hour }
-      error_after: { count: 24, period: hour }
     config:
       tags: ['safe', 'singletons', 'ethereum']
     description: "Singletons addresses used with Safes"
@@ -93,9 +84,6 @@ models:
       blockchain: ethereum
       project: safe
       contributors: tschubotz, danielpartida
-    freshness:
-      warn_after: { count: 12, period: hour }
-      error_after: { count: 24, period: hour }
     config:
       tags: ['safe', 'ethereum']
     description: "Safe transactions"
@@ -157,9 +145,6 @@ models:
       blockchain: ethereum
       project: safe
       contributors: gentrexha
-    freshness:
-      warn_after: { count: 12, period: hour }
-      error_after: { count: 24, period: hour }
     config:
       tags: ['safe', 'ethereum']
     description: "Safe signer thresholds"
@@ -169,3 +154,29 @@ models:
       - &threshold
         name: threshold
         description: "Number of signer threshold"
+
+  - name: safe_ethereum_balances
+    tests:
+      - dbt_utils.unique_combination_of_columns:
+          combination_of_columns:
+            - day
+            - blockchain
+            - address
+            - token_address
+    meta:
+      blockchain: ethereum
+      project: safe
+      contributors: safeintern
+    config:
+      tags: ['safe', 'ethereum']
+    description: “Safe addresses balances”
+    columns:
+      - name: day
+      - name: blockchain
+      - name: address
+      - name: token_address
+      - name: token_standard
+      - name: token_id
+      - name: token_symbol
+      - name: token_balance
+      - name: balance_usd

dbt_subprojects/hourly_spellbook/models/_project/safe/ethereum/safe_ethereum_balances.sql

@@ -0,0 +1,95 @@
+{{
+    config(
+        schema = 'safe_ethereum',
+        alias = 'balances',
+        partition_by = ['day'],
+        materialized = 'incremental',
+        incremental_strategy = 'merge',
+        file_format = 'delta',
+        unique_key = ['day', 'blockchain', 'address', 'token_address'],
+        post_hook = '{{ expose_spells(\'["ethereum"]\',
+                                    "project",
+                                    "safe",
+                                    \'["safeintern"]\') }}'
+    )
+}}
+
+with changed_balances as (
+    select
+        a.blockchain,
+        day,
+        a.address,
+        token_symbol,
+        token_address,
+        token_standard,
+        token_id,
+        balance,
+        lead(cast(day as date)) over (partition by token_address, a.address, token_id order by day asc) as next_update_day
+    from {{ source('tokens_ethereum', 'balances_daily_agg') }} a
+    join (
+        select
+            address
+            , blockchain
+        from {{ ref('safe_ethereum_safes') }} s
+        where blockchain = 'ethereum'
+    ) q on q.address = a.address
+    where day >= date('2021-07-01')
+        and token_standard in ('native', 'erc20')
+        {% if is_incremental() %}
+        and {{ incremental_predicate('day') }}
+        {% endif %}
+),
+days as (
+    select *
+    from unnest(
+        sequence(cast('2021-07-01' as date), date(date_trunc('day', now())), interval '1' day)
+    ) as foo(day)
+),
+forward_fill as (
+    select
+        blockchain,
+        cast(d.day as date) as day,
+        address,
+        token_symbol,
+        token_address,
+        token_standard,
+        token_id,
+        balance
+    from days d
+    left join changed_balances b
+        on d.day >= b.day
+        and (b.next_update_day is null OR d.day < b.next_update_day) 
+    where d.day >= cast('2021-07-01' as date)
+        {% if is_incremental() %}
+        and {{ incremental_predicate('d.day') }}
+        {% endif %}
+)
+select
+    b.day,
+    b.blockchain,
+    b.address,
+    b.token_address,
+    b.token_standard,
+    b.token_id,
+    b.token_symbol,
+    sum(b.balance) as token_balance,
+    sum(b.balance * p.price) as balance_usd
+from (
+    select * from forward_fill
+    where balance > 0
+) b
+left join {{ ref('prices_usd_daily') }} p
+    on (
+        token_standard = 'erc20'
+        and b.blockchain = p.blockchain
+        and b.token_address = p.contract_address
+        and b.day = p.day
+    )
+    or (
+        token_standard = 'native'
+        and p.blockchain is null
+        and p.contract_address is null
+        and p.symbol = 'ETH'
+        and b.day = p.day
+    )
+group by 1, 2, 3, 4, 5, 6, 7

sources/_subprojects_outputs/tokens/balances.yml

@@ -0,0 +1,5 @@
+version: 2
+sources:
+  - name: tokens_ethereum
+    tables:
+      - name: balances_daily_agg