Add/defillama stablecoin streamline model (#130)

.github/workflows/dbt_run_streamline_daily.yml

@@ -4,7 +4,7 @@ run-name: dbt_run_streamline_daily
 on:
   workflow_dispatch:
   schedule:
-    # Runs "at 08:00 UTC" every day (see https://crontab.guru)
+    # Runs "at 07:30 UTC" every day (see https://crontab.guru)
     - cron: '30 7 * * *'
     
 env:
@@ -44,7 +44,7 @@ jobs:
           dbt deps
       - name: Run DBT Jobs
         run: |
-           dbt run --vars '{"STREAMLINE_INVOKE_STREAMS":True}' -m 1+models/bitquery/streamline/streamline__bitquery_realtime.sql 1+models/oklink/streamline/streamline__oklink_realtime.sql 1+models/artemis/streamline/streamline__artemis_realtime.sql
+           dbt run --vars '{"STREAMLINE_INVOKE_STREAMS":True}' -m tag:defillama_streamline 1+models/bitquery/streamline/streamline__bitquery_realtime.sql 1+models/oklink/streamline/streamline__oklink_realtime.sql 1+models/artemis/streamline/streamline__artemis_realtime.sql
 
   notify-failure:
     needs: [run_dbt_jobs]

macros/streamline/models.sql

@@ -2,7 +2,7 @@
         model,
         partition_function,
         partition_name,
-        other_cols
+        other_cols=''
 ) %}
 WITH meta AS (
     SELECT
@@ -17,7 +17,7 @@ WITH meta AS (
             ) A
 )
 SELECT
-    {{ other_cols }},
+    {% if other_cols != '' %}{{ other_cols }},{% endif %}
     DATA,
     _inserted_timestamp,
     s.{{ partition_name }},
@@ -44,7 +44,7 @@ WHERE
         model,
         partition_function,
         partition_name,
-        other_cols
+        other_cols=''
 ) %}
 WITH meta AS (
     SELECT
@@ -59,7 +59,7 @@ WITH meta AS (
         ) A
 )
 SELECT
-    {{ other_cols }},
+    {% if other_cols != '' %}{{ other_cols }},{% endif %}
     DATA,
     _inserted_timestamp,
     s.{{ partition_name }},

models/defillama/bronze/bronze__defillama_stablecoins.sql

@@ -1,6 +1,5 @@
 {{ config(
-    materialized = 'incremental',
+    materialized = 'table',
-    unique_key = 'stablecoin_id',
     tags = ['defillama']
 ) }}
 
@@ -28,39 +27,6 @@ FINAL AS (
         LATERAL FLATTEN (
             input => READ :data :peggedAssets
         )
-
-{% if is_incremental() %}
-WHERE
-    stablecoin_id NOT IN (
-        SELECT
-            DISTINCT stablecoin_id
-        FROM
-            {{ this }}
-    )
-)
-SELECT
-    stablecoin_id,
-    stablecoin,
-    symbol,
-    peg_type,
-    peg_mechanism,
-    price_source,
-    chains,
-    m.row_num + ROW_NUMBER() over (
-        ORDER BY
-            stablecoin
-    ) AS row_num,
-    _inserted_timestamp
-FROM
-    FINAL
-    JOIN (
-        SELECT
-            MAX(row_num) AS row_num
-        FROM
-            {{ this }}
-    ) m
-    ON 1 = 1
-{% else %}
 )
 SELECT
     stablecoin_id,
@@ -74,7 +40,9 @@ SELECT
         ORDER BY
             stablecoin
     ) AS row_num,
-    _inserted_timestamp
+    _inserted_timestamp,
+    sysdate() as inserted_timestamp,
+    sysdate() as modified_timestamp,
+    '{{ invocation_id }}' as _invocation_id
 FROM
-    FINAL
+    FINAL
-{% endif %}

models/defillama/gold/defillama__ez_stablecoin_metrics.sql

@@ -0,0 +1,58 @@
+{{ config(
+    materialized = 'incremental',
+    unique_key = 'defillama_ez_stablecoin_metrics_id',
+    cluster_by = ['date_day','chain'],
+    tags = ['defillama']
+) }}
+
+with base as (
+    select 
+    replace(lower(chain), ' ', '_') as chain,
+    date_day,
+    stablecoin_id,
+    stablecoin,
+    symbol,
+    peg_type,
+    peg_mechanism,
+    GET_PATH(total_circulating_usd, peg_type)::float as total_circulating_usd,
+    GET_PATH(total_minted_usd, peg_type)::float as total_minted_usd,
+    GET_PATH(total_circulating, peg_type)::float as total_circulating,
+    GET_PATH(total_bridged_to_usd, peg_type)::float as total_bridged_to_usd,
+    GET_PATH(total_unreleased, peg_type)::float as total_unreleased,
+    run_timestamp
+    from 
+    {{ ref('silver__defillama_stablecoin_metrics') }}
+    left join {{ ref('bronze__defillama_stablecoins') }} using (stablecoin_id)
+    {% if is_incremental() %}
+    where modified_timestamp > (
+        select coalesce(max(modified_timestamp), '2025-01-01') from {{ this }}
+    )
+    and date_day >= (
+        select coalesce(max(date_day), '1970-01-01') from {{ this }}
+    )
+    {% endif %}
+),
+latest_records as (
+    select *
+    from base
+    qualify row_number() over (partition by chain, stablecoin_id, date_day order by run_timestamp desc) = 1
+)
+select 
+    chain,
+    date_day,
+    stablecoin_id,
+    stablecoin,
+    symbol,
+    peg_type,
+    peg_mechanism,
+    total_circulating_usd,
+    total_minted_usd,
+    total_circulating,
+    total_bridged_to_usd,
+    total_unreleased,
+    {{ dbt_utils.generate_surrogate_key(
+        ['chain','date_day','stablecoin_id']
+    ) }} as defillama_ez_stablecoin_metrics_id,
+    sysdate() as inserted_timestamp,
+    sysdate() as modified_timestamp
+from latest_records

models/defillama/gold/defillama__ez_stablecoin_metrics.yml

@@ -0,0 +1,139 @@
+version: 2
+models:
+  - name: defillama__ez_stablecoin_metrics
+    description: |
+      This table provides comprehensive, easy-to-use stablecoin metrics aggregated from Defillama's stablecoin data. 
+      It combines stablecoin metadata (name, symbol, peg type, mechanism) with daily metrics (circulating supply, 
+      minted amount, bridged amounts) across different blockchain networks. The model deduplicates data by selecting 
+      the latest record for each stablecoin-chain-date combination and provides a clean interface for analyzing 
+      stablecoin performance and adoption across various networks.
+      
+      **Key Features:**
+      - Incremental processing for efficient data updates
+      - Deduplication of multiple API calls per day
+      - Cross-chain stablecoin analysis capabilities
+      - Peg type and mechanism classification
+      - USD-denominated metrics for easy comparison
+      
+      **Use Cases:**
+      - Stablecoin market analysis and trends
+      - Cross-chain stablecoin adoption tracking
+      - DeFi protocol stablecoin usage analysis
+      - Regulatory reporting on stablecoin circulation
+      - Risk assessment of different peg mechanisms
+
+    columns:
+      - name: chain
+        description: |
+          The blockchain network where the stablecoin metrics are measured. Values are normalized to lowercase 
+          with spaces replaced by underscores (e.g., 'ethereum', 'hyperliquid_l1', 'polygon'). This field 
+          enables cross-chain analysis of stablecoin adoption and usage patterns.
+        tests:
+          - not_null
+      
+      - name: date_day
+        description: |
+          The date for which the stablecoin metrics are reported. Stored as a DATE type, this field enables 
+          time-series analysis of stablecoin performance, including daily trends, seasonal patterns, and 
+          long-term growth analysis.
+        tests:
+          - not_null
+      
+      - name: stablecoin_id
+        description: |
+          Unique identifier for the stablecoin as defined by Defillama. This is the primary key used to link 
+          stablecoin metadata with daily metrics. Used for joining with stablecoin reference data to get 
+          additional attributes like name, symbol, and peg mechanism.
+        tests:
+          - not_null
+      
+      - name: stablecoin
+        description: |
+          The full name of the stablecoin (e.g., 'Tether USD', 'USD Coin', 'Dai'). This human-readable name 
+          is useful for reporting and analysis purposes, providing context for the metrics being analyzed.
+        tests:
+          - not_null
+      
+      - name: symbol
+        description: |
+          The ticker symbol for the stablecoin (e.g., 'USDT', 'USDC', 'DAI'). This short identifier is 
+          commonly used in trading pairs and financial reporting, making it easier to reference specific 
+          stablecoins in analysis.
+        tests:
+          - not_null
+      
+      - name: peg_type
+        description: |
+          The type of peg mechanism used by the stablecoin. Common values include 'peggedUSD', 'peggedEUR', 'peggedCNY', etc. 
+          This classification helps analyze the target currency of the stablecoin and its intended use case 
+          in global markets.
+        tests:
+          - not_null
+      
+      - name: peg_mechanism
+        description: |
+          The technical mechanism used to maintain the stablecoin's peg. Options include 'algorithmic', 
+          'crypto-backed', 'fiat-backed'. This field is crucial for risk assessment as different mechanisms 
+          have varying levels of stability and failure modes.
+        tests:
+          - not_null
+      
+      - name: total_circulating_usd
+        description: |
+          The total circulating supply of the stablecoin denominated in USD. This represents the actual 
+          amount of the stablecoin available in the market for trading and use in DeFi protocols. 
+      
+      - name: total_minted_usd
+        description: |
+          The total amount of the stablecoin that has been minted (created) denominated in USD. This 
+          represents the maximum supply that has been generated, which may be different from the circulating 
+          supply due to burning or other mechanisms.
+      
+      - name: total_circulating
+        description: |
+          The total circulating supply of the stablecoin in its native units (not USD denominated). This 
+          provides the raw token count that is available in the market, useful for understanding the 
+          actual token distribution and for calculations that require native units. See peg_type for the 
+          unit of the total_circulating.
+      
+      - name: total_bridged_to_usd
+        description: |
+          The total amount of the stablecoin that has been bridged to other networks, denominated in USD. 
+          This metric is particularly important for cross-chain analysis as it shows how much of the 
+          stablecoin supply is available on different blockchain networks through bridging protocols.
+      
+      - name: total_unreleased
+        description: |
+          The total amount of the stablecoin that has been minted but not yet released to the market, 
+          denominated in its native unit. This represents tokens that are held in reserve or locked in contracts, 
+          providing insight into the stablecoin issuer's reserve management strategy. See peg_type for the 
+          unit of the total_unreleased.
+      
+      - name: defillama_ez_stablecoin_metrics_id
+        description: |
+          A unique surrogate key generated from the combination of chain, date_day, and stablecoin_id. 
+          This serves as the primary key for the table and ensures uniqueness across the combination of 
+          these three critical dimensions. Generated using dbt_utils.generate_surrogate_key().
+        tests:
+          - unique
+          - not_null
+      
+      - name: inserted_timestamp
+        description: |
+          Timestamp when the record was inserted into this table. This is set to the current system time 
+          (SYSDATE()) when the record is created, providing an audit trail of when data was loaded.
+        tests:
+          - not_null
+          - dbt_expectations.expect_row_values_to_have_recent_data:
+              datepart: day
+              interval: 2
+      
+      - name: modified_timestamp
+        description: |
+          Timestamp when the record was last modified. Currently set to SYSDATE() on each run, this field 
+          can be used for incremental processing and to track when data was last updated.
+        tests:
+          - not_null
+          - dbt_expectations.expect_row_values_to_have_recent_data:
+              datepart: day
+              interval: 2

models/defillama/silver/silver__defillama_stablecoin_metrics.sql

@@ -0,0 +1,51 @@
+-- depends_on: {{ ref('bronze__defillama_stablecoin_metrics') }}
+{{ config(
+    materialized = 'incremental',
+    unique_key = 'defillama_stablecoin_metrics_id',
+    cluster_by = ['date_day','chain'],
+    tags = ['defillama']
+) }}
+
+with base as (
+    select 
+    value:"CHAIN"::string as chain,
+    value:"STABLECOIN_ID"::string as stablecoin_id,
+    to_timestamp(value:"RUN_TIMESTAMP"::string) as run_timestamp,
+    partition_key,
+    data:totalCirculatingUSD as total_circulating_usd,
+    data:totalMintedUSD as total_minted_usd,
+    data:totalCirculating as total_circulating,
+    data:totalBridgedToUSD as total_bridged_to_usd,
+    data:totalUnreleased as total_unreleased,
+    to_date(data:date::string) as date_day,
+    _inserted_timestamp
+    from 
+    {% if is_incremental() %}
+    {{ ref('bronze__defillama_stablecoin_metrics') }}
+    where _inserted_timestamp > (
+        select coalesce(max(_inserted_timestamp), '2025-01-01') from {{ this }}
+    )
+    {% else %}
+    {{ ref('bronze__defillama_stablecoin_metrics_fr') }}
+    {% endif %}
+)
+select 
+chain,
+date_day,
+stablecoin_id,
+total_circulating_usd,
+total_minted_usd,
+total_circulating,
+total_bridged_to_usd,
+total_unreleased,
+run_timestamp,
+partition_key,
+_inserted_timestamp,
+{{ dbt_utils.generate_surrogate_key(
+    ['chain','stablecoin_id','date_day', 'run_timestamp']
+) }} as defillama_stablecoin_metrics_id,
+sysdate() as inserted_timestamp,
+sysdate() as modified_timestamp,
+'{{ invocation_id }}' as _invocation_id
+from base 
+qualify row_number() over (partition by defillama_stablecoin_metrics_id order by _inserted_timestamp desc) = 1

models/defillama/silver/silver__defillama_stablecoin_metrics.yml

@@ -0,0 +1,30 @@
+version: 2
+models:
+  - name: silver__defillama_stablecoin_metrics
+    tests:
+      - dbt_utils.unique_combination_of_columns:
+            combination_of_columns:
+              - DEFILLAMA_STABLECOIN_METRICS_ID
+    
+    columns:
+      - name: _INSERTED_TIMESTAMP
+        tests:
+          - not_null
+          - dbt_expectations.expect_column_values_to_be_in_type_list:
+              column_type_list:
+                - TIMESTAMP_NTZ
+          - dbt_expectations.expect_row_values_to_have_recent_data:
+              datepart: day
+              interval: 2
+      - name: DATE_DAY
+        tests:
+          - not_null
+          - dbt_expectations.expect_column_values_to_be_in_type_list:
+              column_type_list:
+                - TIMESTAMP_NTZ
+          - dbt_expectations.expect_row_values_to_have_recent_data:
+              datepart: day
+              interval: 2
+      - name: DEFILLAMA_STABLECOIN_METRICS_ID
+        tests:
+          - not_null

models/defillama/streamline/bronze/bronze__defillama_stablecoin_metrics.sql

@@ -0,0 +1,9 @@
+{{ config (
+    materialized = 'view',
+    tags = ['defillama_streamline']
+) }}
+{{ streamline_external_table_query_v2(
+    model = 'defillama_stablecoin_metrics',
+    partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 4), '_', 1) AS INTEGER)",
+    partition_name = "partition_key"
+) }}

models/defillama/streamline/bronze/bronze__defillama_stablecoin_metrics_FR.sql

@@ -0,0 +1,9 @@
+{{ config (
+    materialized = 'view',
+    tags = ['defillama_streamline']
+) }}
+{{ streamline_external_table_FR_query_v2(
+    model = 'defillama_stablecoin_metrics',
+    partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 4), '_', 1) AS INTEGER)",
+    partition_name = "partition_key"
+) }}

models/defillama/streamline/streamline__defillama_stablecoin_metrics.sql

@@ -0,0 +1,53 @@
+{{ config (
+    materialized = "view",
+    post_hook = fsc_utils.if_data_call_function_v2(
+        func = 'streamline.udf_bulk_rest_api_v2',
+        target = "{{this.schema}}.{{this.identifier}}",
+        params ={ "external_table" :"defillama_stablecoin_metrics",
+        "sql_limit" :"10000",
+        "producer_batch_size" :"10",
+        "worker_batch_size" :"1",
+        "async_concurrent_requests" :"1",
+        "sql_source" :"{{this.identifier}}",
+        "exploded_key": tojson(['data'])
+        }
+    ),
+    tags = ['defillama_streamline']
+) }}
+
+WITH stablecoins as (
+
+    select 
+    stablecoin_id,
+    stablecoin,
+    symbol,
+    peg_type,
+    peg_mechanism,
+    price_source,
+    value::string as chain
+    from {{ ref('bronze__defillama_stablecoins') }},
+    lateral flatten (input => chains)
+)
+SELECT
+    chain,
+    stablecoin_id,
+    date_part('epoch_second', sysdate()) as run_timestamp,
+    date_part('epoch_second', sysdate()::DATE) AS partition_key,
+    {{ target.database }}.live.udf_api(
+        'GET',
+        'https://pro-api.llama.fi/{api_key}/stablecoins/stablecoincharts/' || chain || '?stablecoin=' || stablecoin_id,
+        OBJECT_CONSTRUCT(
+            'Content-Type', 'text/plain',
+            'Accept', 'text/plain',
+            'fsc-quantum-state', 'streamline'
+        ),
+        {},
+        'Vault/prod/external/defillama'
+    ) AS request
+FROM
+    stablecoins
+where chain is not null and stablecoin_id is not null
+
+order by chain
+
+limit 10000

models/sources.yml

@@ -10,6 +10,7 @@ sources:
       - name: artemis
       - name: defillama_protocol_historical
       - name: valuations_parquet
+      - name: defillama_stablecoin_metrics
   - name: tokenflow_eth
     database: flipside_prod_db
     schema: tokenflow_eth