Arbitrum RWA user balances

dbt_subprojects/tokens/macros/balances/balances_incremental_subset_daily.sql

@@ -0,0 +1,162 @@
+{#  @DEV here
+
+    @NOTICE this macro constructs the address level token balances table for given input table
+    @NOTICE aka, you give lists of tokens and/or address, it generates table with daily balances of the address-token pair
+
+    @PARAM blockchain               -- blockchain name
+    @PARAM address_list             -- must have an address column, can be none if only filtering on tokens
+    @PARAM token_list               -- must have a token_address column, can be none if only filtering on tokens
+    @PARAM address_token_list       -- for advanced usage, must have both (address, token_address) columns, can be none
+    @PARAM start_date               -- the start_date, used to generate the daily timeseries
+
+#}
+
+{%- macro balances_incremental_subset_daily(
+        blockchain,
+        start_date,
+        address_list = none,
+        token_list = none,
+        address_token_list = none
+    )
+%}
+
+WITH
+filtered_daily_agg_balances as (
+    select
+        b.blockchain,
+        b.day,
+        b.block_number,
+        b.block_time,
+        b.address,
+        b.token_address,
+        b.token_standard,
+        b.balance_raw,
+        CASE
+            WHEN b.token_standard = 'erc20' THEN b.balance_raw / power(10, erc20_tokens.decimals)
+            WHEN b.token_standard = 'native' THEN b.balance_raw / power(10, 18)
+            ELSE b.balance_raw
+        END as balance,
+        erc20_tokens.symbol as token_symbol,
+        token_id
+    from {{ref('tokens_'~blockchain~'_balances_daily_agg_base')}} b
+    {% if address_list is not none %}
+    inner join (select distinct address from {{address_list}}) f1
+    on f1.address = b.address
+    {% endif %}
+    {% if token_list is not none %}
+    inner join (select distinct token_address from {{token_list}}) f2
+    on f2.token_address = b.token_address
+    {% endif %}
+    {% if address_token_list is not none %}
+    inner join (select distinct address, token_address from {{address_token_list}}) f3
+    on f3.token_address = b.token_address
+    and f3.address = b.address
+    {% endif %}
+    left join {{ source('tokens', 'erc20') }} erc20_tokens on
+        erc20_tokens.blockchain = '{{blockchain}}'
+        AND erc20_tokens.contract_address = b.token_address
+        AND b.token_standard = 'erc20'
+    where day >= cast('{{start_date}}' as date)
+
+)
+,changed_balances as (
+    select *
+     , lead(cast(day as timestamp)) over (partition by token_address,address,token_id order by day asc) as next_update_day
+    from (
+    select * from (
+        select
+            blockchain
+            ,day
+            ,address
+            ,token_symbol
+            ,token_address
+            ,token_standard
+            ,token_id
+            ,balance
+        from filtered_daily_agg_balances
+        where day >= cast('{{start_date}}' as date)
+        {% if is_incremental() %}
+            and {{ incremental_predicate('day') }}
+        {% endif %}
+    )
+    -- if we're running incremental, we need to retrieve the last known balance updates from before the current window
+    -- so we can correctly populate the forward fill.
+    {% if is_incremental() %}
+    UNION ALL
+    select * from (
+        select
+            blockchain
+            ,max(day) as day
+            ,address
+            ,token_symbol
+            ,token_address
+            ,token_standard
+            ,token_id
+            ,max_by(balance, day) as balance
+        from filtered_daily_agg_balances
+        where day >= cast('{{start_date}}' as date)
+        and not {{ incremental_predicate('day') }}
+        group by 1,3,4,5,6,7
+        )
+    {% endif %}
+    )
+),
+
+days as (
+    select *
+    from unnest(
+         sequence(cast('{{start_date}}' as date)
+                , date(date_trunc('day',now()))
+                , interval '1' day
+                )
+         ) as foo(day)
+),
+forward_fill as (
+    select
+        blockchain,
+        cast(d.day as timestamp) as day,
+        address,
+        token_symbol,
+        token_address,
+        token_standard,
+        token_id,
+        balance,
+        b.day as last_updated,
+        b.next_update_day as next_update
+    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) -- perform forward fill
+)
+
+select
+    b.blockchain,
+    b.day,
+    b.address,
+    b.token_symbol,
+    b.token_address,
+    b.token_standard,
+    b.token_id,
+    b.balance,
+    b.balance * p.price as balance_usd,
+    b.last_updated,
+    b.next_update
+from(
+    select * from forward_fill
+    where balance > 0
+    {% if is_incremental() %}
+        and {{ incremental_predicate('day') }}
+    {% endif %}
+
+) b
+left join {{source('prices','usd')}} p
+    on (token_standard = 'erc20'
+        and p.blockchain = '{{blockchain}}'
+        and b.token_address = p.contract_address
+        and b.day = p.minute)
+    or (token_standard = 'native'
+        and p.blockchain is null
+        and p.contract_address is null
+        and p.symbol = (select native_token_symbol from {{source('evms','info')}} where blockchain = '{{blockchain}}')
+        and b.day = p.minute)
+{% endmacro %}

dbt_subprojects/tokens/models/rwa/arbitrum/rwa_arbitrum_balances.sql

@@ -0,0 +1,37 @@
+{{
+  config(
+    schema = 'rwa_arbitrum',
+    alias = 'balances',
+    materialized = 'incremental',
+    file_format = 'delta',
+    incremental_strategy = 'merge',
+    unique_key = ['day', 'address', 'token_address'],
+    incremental_predicates = [incremental_predicate('DBT_INTERNAL_DEST.day')]
+  )
+}}
+
+with
+rwa_tokens as (
+  select
+    project,
+    token_address
+  from {{ref('rwa_arbitrum_tokens')}}
+  where type = 'RWA'
+)
+
+,balances as (
+    {{
+      balances_incremental_subset_daily(
+            blockchain = 'arbitrum',
+            token_list = 'rwa_tokens',
+            start_date = '2023-11-17'
+      )
+    }}
+)
+
+select
+    t.project
+    ,b.*
+from balances b
+left join rwa_tokens t
+    on b.token_address = t.token_address

dbt_subprojects/tokens/models/rwa/arbitrum/rwa_arbitrum_dex_balances.sql

@@ -0,0 +1,41 @@
+{{
+  config(
+    schema = 'rwa_arbitrum',
+    alias = 'dex_balances',
+    materialized = 'incremental',
+    file_format = 'delta',
+    incremental_strategy = 'merge',
+    unique_key = ['day', 'address', 'token_address'],
+    incremental_predicates = [incremental_predicate('DBT_INTERNAL_DEST.day')]
+  )
+}}
+
+with dex_pools as (
+    select
+        project,
+        version,
+        pool as address,
+        token_address
+    from {{ref('rwa_arbitrum_dex_pools')}}
+)
+
+,balances as (
+    {{
+      balances_incremental_subset_daily(
+            blockchain = 'arbitrum',
+            address_token_list = 'dex_pools',
+            start_date = '2023-11-17',
+      )
+    }}
+)
+
+select
+    p.project
+    ,p.version
+    ,b.*
+from balances b
+left join dex_pools p
+    on b.address = p.address
+    and b.token_address = p.token_address
+
+

dbt_subprojects/tokens/models/rwa/arbitrum/rwa_arbitrum_dex_pools.sql

@@ -0,0 +1,39 @@
+
+{{
+    config(
+        schema = 'rwa_arbitrum'
+        ,alias = 'dex_pools'
+        ,materialized = 'table'
+        ,post_hook='{{ expose_spells(\'["arbitrum"]\',
+                        "sector",
+                        "rwa",
+                        \'["maybeYonas", "pyor_xyz"]\') }}'
+    )
+}}
+
+select
+    project,
+    version,
+    pool,
+    token_address,
+    token_is_rwa,
+    symbol,
+    protocol,
+    type
+from (values
+    ('curve', 'stableswap-ng', 0x4bD135524897333bec344e50ddD85126554E58B4, 0xaf88d065e77c8cC2239327C5EDb3A432268e5831, 0, 'USDC', 'Mountain Protocol', 'RWA'),
+    ('curve', 'stableswap-ng', 0x4bD135524897333bec344e50ddD85126554E58B4, 0x59D9356E565Ab3A36dD77763Fc0d87fEaf85508C, 0, 'USDM', 'Mountain Protocol', 'RWA'),
+    ('uniswap', '3', 0xfEAA137F43f88b7F767F5A67978FfF8EC11Cc6Ef, 0x82aF49447D8a07e3bd95BD0d56f35241523fBab1, 0, 'WETH', 'Frax Finance', 'Governance'),
+    ('uniswap', '3', 0xfEAA137F43f88b7F767F5A67978FfF8EC11Cc6Ef, 0x9d2F299715D94d8A7E6F5eaa8E654E8c74a988A7, 0, 'FXS', 'Frax Finance', 'Governance'),
+    ('ramses', '3', 0xa45Cbd521ED745F6C856C89F8dBB583bD1591fC2, 0x82aF49447D8a07e3bd95BD0d56f35241523fBab1, 0, 'WETH', 'Frax Finance', 'Governance'),
+    ('ramses', '3', 0xa45Cbd521ED745F6C856C89F8dBB583bD1591fC2, 0x9d2F299715D94d8A7E6F5eaa8E654E8c74a988A7, 0, 'FXS', 'Frax Finance', 'Governance')
+) as t(
+    project,
+    version,
+    pool,
+    token_address,
+    token_is_rwa,
+    symbol,
+    protocol,
+    type
+)