Truncate timestamp to shift windowing in trade agg

stellar · Sep 6, 2024 · 4f97540 · 4f97540
1 parent 9e39777
commit 4f97540
Show file tree

Hide file tree

Showing 6 changed files with 27 additions and 27 deletions.
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
@@ -20,14 +20,14 @@ repos:
         language_version: 14.21.3
 
   - repo: https://github.com/sqlfluff/sqlfluff
-    rev: 3.0.5
+    rev: 3.0.7
     hooks:
       - id: sqlfluff-lint
         args: ["lint", "--processes", "0", "--config=./.sqlfluff"]
         additional_dependencies:
           [
-            "sqlfluff-templater-dbt==3.0.5",
-            "dbt-bigquery==1.3.0",
+            "sqlfluff-templater-dbt==3.0.7",
+            "dbt-bigquery==1.8.2",
             "python-dotenv",
           ]
         entry: ./pre-commit/for_pre_commit.sh
@@ -36,8 +36,8 @@ repos:
         args: ["fix", "--force", "--processes", "0", "--config=./.sqlfluff"]
         additional_dependencies:
           [
-            "sqlfluff-templater-dbt==3.0.5",
-            "dbt-bigquery==1.3.0",
+            "sqlfluff-templater-dbt==3.0.7",
+            "dbt-bigquery==1.8.2",
             "python-dotenv",
           ]
         entry: ./pre-commit/for_pre_commit.sh

diff --git a/models/intermediate/trades/int_trade_agg_day.sql b/models/intermediate/trades/int_trade_agg_day.sql
@@ -8,7 +8,7 @@
 with
     base_trades as (
         select
-            date('{{ dbt_airflow_macros.ts(timezone=none) }}') as day_agg
+            date(timestamp_trunc('{{ dbt_airflow_macros.ts() }}', day)) as day_agg
             , ledger_closed_at
             , selling_asset_id
             , selling_asset_code
@@ -25,8 +25,8 @@ with
             , buying_amount
         from {{ ref('stg_history_trades') }}
         where
-            ledger_closed_at < TIMESTAMP_ADD('{{ dbt_airflow_macros.ts(timezone=none) }}', INTERVAL 1 DAY )
-            and ledger_closed_at >= '{{ dbt_airflow_macros.ts(timezone=none) }}'
+            ledger_closed_at < timestamp_add(timestamp_trunc('{{ dbt_airflow_macros.ts() }}', day), interval 1 day )
+            and ledger_closed_at >= timestamp_trunc('{{ dbt_airflow_macros.ts() }}', day)
     )
 
     /* duplicates trades in order to obtain all trades between an asset pair, regardless

diff --git a/models/intermediate/trades/int_trade_agg_month.sql b/models/intermediate/trades/int_trade_agg_month.sql
@@ -25,8 +25,8 @@ with
             , buying_amount
         from {{ ref('stg_history_trades') }}
         where
-            ledger_closed_at < timestamp_add('{{ dbt_airflow_macros.ts(timezone=none) }}', interval 1 day)
-            and ledger_closed_at >= timestamp_sub('{{ dbt_airflow_macros.ts(timezone=none) }}', interval 31 day)
+            ledger_closed_at < timestamp_add(timestamp_trunc('{{ dbt_airflow_macros.ts() }}', day), interval 1 day)
+            and ledger_closed_at >= timestamp_sub(timestamp_trunc('{{ dbt_airflow_macros.ts() }}', day), interval 31 day)
     )
 
     /* duplicates trades in order to obtain all trades between an asset pair, regardless
@@ -120,7 +120,7 @@ with
     /* obtain aggregate function metrics for the asset pair */
     , trade_day_agg_group as (
         select
-            date('{{ dbt_airflow_macros.ts(timezone=none) }}') as day_agg
+            date('{{ dbt_airflow_macros.ts() }}') as day_agg
             , asset_a
             , asset_a_code
             , asset_a_issuer
@@ -136,7 +136,7 @@ with
             , max(price_n / price_d) as high_price_monthly
             , min(price_n / price_d) as low_price_monthly
         from dedup_asset_pair
-        where cast(ledger_closed_at as date) >= date_sub(date('{{ dbt_airflow_macros.ts(timezone=none) }}'), interval 30 day)
+        where cast(ledger_closed_at as date) >= date_sub(date('{{ dbt_airflow_macros.ts() }}'), interval 30 day)
         group by
             asset_a
             , asset_a_code
@@ -197,7 +197,7 @@ with
                 order by ledger_closed_at desc
             ) as dedup_rows
         from dedup_asset_pair
-        where cast(ledger_closed_at as date) >= date_sub(date('{{ dbt_airflow_macros.ts(timezone=none) }}'), interval 30 day)
+        where cast(ledger_closed_at as date) >= date_sub(date('{{ dbt_airflow_macros.ts() }}'), interval 30 day)
     )
 
     /* joins all metrics related to the asset pair while deduplicating the window functions */

diff --git a/models/intermediate/trades/int_trade_agg_week.sql b/models/intermediate/trades/int_trade_agg_week.sql
@@ -25,8 +25,8 @@ with
             , buying_amount
         from {{ ref('stg_history_trades') }}
         where
-            ledger_closed_at < timestamp_add('{{ dbt_airflow_macros.ts(timezone=none) }}', interval 1 day)
-            and ledger_closed_at >= timestamp_sub('{{ dbt_airflow_macros.ts(timezone=none) }}', interval 8 day)
+            ledger_closed_at < timestamp_add(timestamp_trunc('{{ dbt_airflow_macros.ts() }}', day), interval 1 day)
+            and ledger_closed_at >= timestamp_sub(timestamp_trunc('{{ dbt_airflow_macros.ts() }}', day), interval 8 day)
     )
 
     /* duplicates trades in order to obtain all trades between an asset pair, regardless
@@ -120,7 +120,7 @@ with
     /* obtain aggregate function metrics for the asset pair */
     , trade_day_agg_group as (
         select
-            date('{{ dbt_airflow_macros.ts(timezone=none) }}') as day_agg
+            date('{{ dbt_airflow_macros.ts() }}') as day_agg
             , asset_a
             , asset_a_code
             , asset_a_issuer
@@ -136,7 +136,7 @@ with
             , max(price_n / price_d) as high_price_weekly
             , min(price_n / price_d) as low_price_weekly
         from dedup_asset_pair
-        where cast(ledger_closed_at as date) >= date_sub(date('{{ dbt_airflow_macros.ts(timezone=none) }}'), interval 7 day)
+        where cast(ledger_closed_at as date) >= date_sub(date('{{ dbt_airflow_macros.ts() }}'), interval 7 day)
         group by
             asset_a
             , asset_a_code
@@ -197,7 +197,7 @@ with
                 order by ledger_closed_at desc
             ) as dedup_rows
         from dedup_asset_pair
-        where cast(ledger_closed_at as date) >= date_sub(date('{{ dbt_airflow_macros.ts(timezone=none) }}'), interval 7 day)
+        where cast(ledger_closed_at as date) >= date_sub(date('{{ dbt_airflow_macros.ts() }}'), interval 7 day)
     )
 
     /* joins all metrics related to the asset pair while deduplicating the window functions */

diff --git a/models/intermediate/trades/int_trade_agg_year.sql b/models/intermediate/trades/int_trade_agg_year.sql
@@ -31,8 +31,8 @@ with
             , buying_amount
         from {{ ref('stg_history_trades') }}
         where
-            ledger_closed_at < timestamp_add('{{ dbt_airflow_macros.ts(timezone=none) }}', interval 1 day)
-            and ledger_closed_at >= timestamp_sub('{{ dbt_airflow_macros.ts(timezone=none) }}', interval 366 day)
+            ledger_closed_at < timestamp_add(timestamp_trunc('{{ dbt_airflow_macros.ts() }}', day), interval 1 day)
+            and ledger_closed_at >= timestamp_sub(timestamp_trunc('{{ dbt_airflow_macros.ts() }}', day), interval 366 day)
     )
 
     /* duplicates trades in order to obtain all trades between an asset pair, regardless
@@ -126,7 +126,7 @@ with
     /* obtain aggregate function metrics for the asset pair */
     , trade_day_agg_group as (
         select
-            date('{{ dbt_airflow_macros.ts(timezone=none) }}') as day_agg
+            date('{{ dbt_airflow_macros.ts() }}') as day_agg
             , asset_a
             , asset_a_code
             , asset_a_issuer
@@ -142,7 +142,7 @@ with
             , max(price_n / price_d) as high_price_yearly
             , min(price_n / price_d) as low_price_yearly
         from dedup_asset_pair
-        where cast(ledger_closed_at as date) >= date_sub(date('{{ dbt_airflow_macros.ts(timezone=none) }}'), interval 365 day)
+        where cast(ledger_closed_at as date) >= date_sub(date('{{ dbt_airflow_macros.ts() }}'), interval 365 day)
         group by
             asset_a
             , asset_a_code
@@ -203,7 +203,7 @@ with
                 order by ledger_closed_at desc
             ) as dedup_rows
         from dedup_asset_pair
-        where cast(ledger_closed_at as date) >= date_sub(date('{{ dbt_airflow_macros.ts(timezone=none) }}'), interval 365 day)
+        where cast(ledger_closed_at as date) >= date_sub(date('{{ dbt_airflow_macros.ts() }}'), interval 365 day)
     )
 
     /* joins all metrics related to the asset pair while deduplicating the window functions */

diff --git a/models/marts/trade_agg.sql b/models/marts/trade_agg.sql
@@ -15,31 +15,31 @@ with
         select *
         from {{ ref('int_trade_agg_day') }}
         {% if is_incremental() %}
-            where day_agg = date('{{ dbt_airflow_macros.ts(timezone=none) }}')
+            where day_agg = date('{{ dbt_airflow_macros.ts() }}')
         {% endif %}
     )
 
     , trade_weekly as (
         select *
         from {{ ref('int_trade_agg_week') }}
         {% if is_incremental() %}
-            where day_agg = date('{{ dbt_airflow_macros.ts(timezone=none) }}')
+            where day_agg = date('{{ dbt_airflow_macros.ts() }}')
         {% endif %}
     )
 
     , trade_monthly as (
         select *
         from {{ ref('int_trade_agg_month') }}
         {% if is_incremental() %}
-            where day_agg = date('{{ dbt_airflow_macros.ts(timezone=none) }}')
+            where day_agg = date('{{ dbt_airflow_macros.ts() }}')
         {% endif %}
     )
 
     , trade_yearly as (
         select *
         from {{ ref('int_trade_agg_year') }}
         {% if is_incremental() %}
-            where day_agg = date('{{ dbt_airflow_macros.ts(timezone=none) }}')
+            where day_agg = date('{{ dbt_airflow_macros.ts() }}')
         {% endif %}
     )