C911 [TON]: updating ton stablecoin pipelines part 1 (#606)

macros/stablecoins/agg_chain_stablecoin_transfers.sql

@@ -46,6 +46,45 @@
             )
     -- TODO: Refactor to support native currencies. Currently assumes everything is $1
     -- b/c of perf issues when joining
+    {% elif chain in ("ton") %}
+        select
+            block_timestamp
+            , trunc(block_timestamp, 'day') as date
+            , null as block_number
+            , event_index as index
+            , trace_id
+            , tx_hash
+            , from_address
+            , to_address
+            , type = 'mint' or lower(from_address) in (
+                select distinct (lower(premint_address))
+                from {{ ref("fact_ton_stablecoin_premint_addresses") }}
+            ) as is_mint
+            , type = 'burn' or lower(to_address) in (
+                select distinct (lower(premint_address))
+                from {{ ref("fact_ton_stablecoin_premint_addresses") }}
+            ) as is_burn
+            , coalesce(amount / POWER(10, num_decimals), 0) as amount
+            , case
+                when is_mint then amount / POWER(10, num_decimals) when is_burn then -1 * amount / POWER(10, num_decimals) else 0
+            end as inflow
+            , case
+                when
+                    not is_mint
+                    and not is_burn
+                then amount / POWER(10, num_decimals)
+                else 0
+            end as transfer_volume
+            , contracts.symbol
+            , transfers.contract_address
+        from {{ ref('fact_ton_token_transfers') }} transfers
+        left join {{ ref('fact_ton_stablecoin_contracts') }} contracts
+            on lower(transfers.contract_address) = lower(contracts.contract_address)
+        where lower(transfers.contract_address) in (
+                select lower(contract_address)
+                from {{ ref('fact_ton_stablecoin_contracts') }}
+            )
+            and tx_status = 'TRUE'
     {% elif chain in ("solana") %}
     -- CASE 1: Mints into non-premint addresses
         select

models/dimensions/current_balances/dim_ton_current_balances.sql

@@ -0,0 +1,5 @@
+-- depends_on: {{ ref("fact_ton_address_balances_by_token") }}
+{{ config(materialized="incremental", unique_key=["address", "contract_address"]) }}
+
+
+{{ current_balances("ton") }}

models/metrics/stablecoins/contracts/fact_ton_stablecoin_contracts.sql

@@ -0,0 +1,7 @@
+{{ config(materialized="table") }}
+select symbol, contract_address, num_decimals, coingecko_id, initial_supply
+from
+    (
+        values
+            ('USDT', '0:B113A994B5024A16719F69139328EB759596C38A25F59028B146FECDC3621DFE', 6, 'tether', 0)
+    ) as results(symbol, contract_address, num_decimals, coingecko_id, initial_supply)

models/metrics/stablecoins/contracts/fact_ton_stablecoin_premint_addresses.sql

@@ -0,0 +1,12 @@
+{{ config(materialized="table") }}
+-- Premint address can be an account owner or a token account
+select contract_address, premint_address
+from
+    (
+        values
+            -- USDT
+            (
+                '0:B113A994B5024A16719F69139328EB759596C38A25F59028B146FECDC3621DFE',
+                '0:23FA979918F1FE702DB9100BF843E87C7015ECCD39A4721E9B6BAC170BC04CE3'
+            )
+    ) as results(contract_address, premint_address)

models/projects/ton/core/ez_ton_address_balances_by_token.sql

@@ -1,14 +1,12 @@
--- depends_on: {{ ref("fact_ton_address_credit_by_token") }}
--- depends_on: {{ ref("fact_ton_address_debit_by_token") }}
--- depends_on: {{ source("BALANCES", "ez_ton_current_balances") }}
 {{
     config(
-        materialized="table",
+        materialized="view",
         database="ton",
         schema="core",
         name="ez_address_balances_by_token",
         snowflake_warehouse="TON_MD",
     )
 }}
 
-{{ address_balances("ton") }}
+select *
+from {{ ref("fact_ton_address_balances_by_token") }}

models/projects/ton/core/ez_ton_current_balances.sql

@@ -1,12 +1,12 @@
--- depends_on: {{ ref("ez_ton_address_balances_by_token") }}
 {{
     config(
-        materialized="table",
+        materialized="view",
         database="ton",
         schema="core",
         name="ez_current_balances",
         snowflake_warehouse="TON_MD",
     )
 }}
 
-{{ current_balances("ton") }}
+select *
+from {{ ref("dim_ton_current_balances") }}

models/projects/ton/raw/ez_ton_stablecoin_transfers.sql

@@ -1,62 +1,11 @@
 {{
     config(
-        materialized="incremental",
-        unique_key="tx_hash",
+        materialized="view",
         snowflake_warehouse="TON",
         database="ton",
         schema="raw",
         alias="ez_stablecoin_transfers",
     )
 }}
-
-select
-    block_timestamp,
-    trunc(block_timestamp, 'day') as date,
-    null as block_number,
-    trace_id as index,
-    trace_id,
-    tx_hash,
-    from_address,
-    from_type,
-    to_address,
-    to_type,
-    -- Mint: From: Premint, To: Contract
-    from_address in (
-        select distinct (premint_address)
-            from pc_dbt_db.prod.fact_ton_stablecoin_contracts
-        ) 
-        and to_address not in (
-        select distinct (premint_address)
-            from pc_dbt_db.prod.fact_ton_stablecoin_contracts
-        )
-    as is_mint,
-    -- BURN: From: Contract, To: Premint
-    from_address not in (
-        select distinct (premint_address)
-            from pc_dbt_db.prod.fact_ton_stablecoin_contracts
-        ) 
-        and to_address in (
-        select distinct (premint_address)
-            from pc_dbt_db.prod.fact_ton_stablecoin_contracts
-        )
-    as is_burn,
-    coalesce(amount / POWER(10, decimal), 0) as amount,
-    case
-        when is_mint then amount / POWER(10, decimal) when is_burn then -1 * amount / POWER(10, decimal) else 0
-    end as inflow,
-    case
-        when
-            not is_mint
-            and not is_burn
-        then amount / POWER(10, decimal)
-        else 0
-    end as transfer_volume,
-    fact_ton_stablecoin_contracts.symbol,
-    fact_ton_stablecoin_contracts.contract_address
-from 
-    {{ ref('fact_ton_stablecoin_transfers') }} as transfers
-left join
-    pc_dbt_db.prod.fact_ton_stablecoin_contracts
-    on lower(transfers.symbol)
-    = lower(fact_ton_stablecoin_contracts.symbol)
-where dest_verified and source_verified and account_verified
+select *
+from {{ ref("fact_ton_stablecoin_transfers") }}

models/staging/ton/__ton__sources.yml

@@ -14,3 +14,9 @@ sources:
       - name: raw_ton_app_txns_fees
       - name: raw_ton_transactions
       - name: raw_ton_transactions_avro
+      - name: raw_ton_jetton_events_avro
+  - name: BALANCES
+    schema: PROD
+    database: PC_DBT_DB
+    tables:
+      - name: dim_ton_current_balances

models/staging/ton/fact_ton_address_balances_by_token.sql

@@ -0,0 +1,11 @@
+-- depends_on: {{ ref("fact_ton_address_credit_by_token") }}
+-- depends_on: {{ ref("fact_ton_address_debit_by_token") }}
+-- depends_on: {{ source("BALANCES", "dim_ton_current_balances") }}
+{{
+    config(
+        materialized="table",
+        snowflake_warehouse="TON_MD",
+    )
+}}
+
+{{ address_balances("ton") }}

models/staging/ton/fact_ton_address_credit_by_token.sql

@@ -5,26 +5,20 @@
         snowflake_warehouse="TON_MD",
     )
 }}
-
-with ton_transfers as (
-    {{
-        dbt_utils.union_relations(
-            relations=[
-                ref("ez_ton_stablecoin_transfers"),
-            ]
-        )
-    }}
-)
-select 
+select
     to_address as address,
     contract_address,
     block_timestamp,
     cast(amount as float) as credit,
-    tx_hash
-from ton_transfers
-where 
-    block_timestamp < to_date(sysdate())
+    null as credit_usd,
+    tx_hash,
+    null as trace_index,
+    event_index
+from {{ ref("fact_ton_token_transfers") }}
+where
+    to_address is not null
+    and to_date(block_timestamp) < to_date(sysdate())
     {% if is_incremental() %}
         and block_timestamp
         >= (select dateadd('day', -3, max(block_timestamp)) from {{ this }})
-    {% endif %}
+    {% endif %}

models/staging/ton/fact_ton_address_debit_by_token.sql

@@ -6,25 +6,20 @@
     )
 }}
 
-with ton_transfers as (
-    {{
-        dbt_utils.union_relations(
-            relations=[
-                ref("ez_ton_stablecoin_transfers"),
-            ]
-        )
-    }}
-)
-select 
+select
     from_address as address,
     contract_address,
     block_timestamp,
     cast(amount * -1 as float) as debit,
-    tx_hash
-from ton_transfers
-where 
-    block_timestamp < to_date(sysdate())
+    null as debit_usd,
+    tx_hash,
+    null as trace_index,
+    event_index
+from {{ ref("fact_ton_token_transfers")}}
+where
+    to_date(block_timestamp) < to_date(sysdate())
+    and from_address is not null
     {% if is_incremental() %}
         and block_timestamp
         >= (select dateadd('day', -3, max(block_timestamp)) from {{ this }})
-    {% endif %}
+    {% endif %}

models/staging/ton/fact_ton_stablecoin_transfers.sql

@@ -4,51 +4,4 @@
         unique_key="tx_hash",
     )
 }}
-with raw_data as (
-    select
-        extraction_date
-        , source_json
-        , source_json:"amount"::bigint as amount
-        , source_json:"date_timestamp"::timestamp as timestamp
-        , source_json:"decoded_hash"::string as tx_hash
-        , PARSE_JSON(source_json:"decoded_value"::string) as payload
-        , source_json:"from_address"::string as from_address
-        , source_json:"from_type"::string as from_type
-        , source_json:"to_address"::string as to_address
-        , source_json:"to_type"::string as to_type
-        , source_json:"url"::string as url
-        , source_json:"decimals"::int as decimal
-        , source_json:"symbol"::string as symbol
-        , source_json:"trace_id"::string as trace_id
-        , source_json:"dest_verified"::boolean as dest_verified
-        , source_json:"source_verified"::boolean as source_verified
-        , source_json:"account_verified"::boolean as account_verified
-    from
-        {{ source("PROD_LANDING", "raw_ton_stablecoin_transfers") }}
-    {% if is_incremental() %}
-        where source_json:"date_timestamp"::timestamp > (select dateadd('day', -3, max(block_timestamp)) from {{ this }})
-    {% endif %}
-)
-select 
-    tx_hash
-    , coalesce(max_by(amount, extraction_date), 0) as amount
-    , max_by(timestamp, extraction_date) as block_timestamp
-    , max_by(payload, extraction_date) as payload
-    , max_by(from_address, extraction_date) as from_address
-    , max_by(from_type, extraction_date) as from_type
-    , max_by(to_address, extraction_date) as to_address
-    , max_by(to_type, extraction_date) as to_type
-    , max_by(trace_id, extraction_date) as trace_id
-    , max_by(decimal, extraction_date) as decimal
-    , case when (max_by(symbol, extraction_date) = 'USD₮' or max_by(url, extraction_date) = 'https://tether.to/usdt-ton.json')  then 'USDT' else max_by(symbol, extraction_date) end as symbol
-    , max_by(dest_verified, extraction_date) as dest_verified
-    , max_by(source_verified, extraction_date) as source_verified
-    , max_by(account_verified, extraction_date) as account_verified
-from raw_data
-where 
-    from_address is not null and 
-    to_address is not null and 
-    timestamp is not null and
-    -- Black list transaction: https://tonviewer.com/transaction/513bacb858a3148367643ec9da96ba9e8af8bdf02c1950b106e9a88fe3e94935
-    tx_hash not in ('513bacb858a3148367643ec9da96ba9e8af8bdf02c1950b106e9a88fe3e94935')
-group by tx_hash
+{{agg_chain_stablecoin_transfers("ton")}}