From 3e46332aa68ce945c557d3c53ba4451c5db4e8d5 Mon Sep 17 00:00:00 2001 From: Simon Chow Date: Tue, 23 Jan 2024 13:40:59 -0500 Subject: [PATCH 1/3] Pass asset info in trade agg --- .../intermediate/trades/int_trade_agg_day.sql | 54 ++++++++ .../trades/int_trade_agg_month.sql | 54 ++++++++ .../trades/int_trade_agg_week.sql | 48 +++++++ .../trades/int_trade_agg_year.sql | 48 +++++++ models/marts/trade_agg.sql | 127 ++---------------- 5 files changed, 216 insertions(+), 115 deletions(-) diff --git a/models/intermediate/trades/int_trade_agg_day.sql b/models/intermediate/trades/int_trade_agg_day.sql index bdcb189..f2535ef 100644 --- a/models/intermediate/trades/int_trade_agg_day.sql +++ b/models/intermediate/trades/int_trade_agg_day.sql @@ -10,7 +10,13 @@ with date('{{ dbt_airflow_macros.ds() }}') as day_agg , ledger_closed_at , selling_asset_id + , selling_asset_code + , selling_asset_issuer + , selling_asset_type , buying_asset_id + , buying_asset_code + , buying_asset_issuer + , buying_asset_type , concat(history_operation_id, `order`) as trade_key , price_n , price_d @@ -29,7 +35,13 @@ with day_agg , ledger_closed_at , selling_asset_id as asset_a + , selling_asset_code as asset_a_code + , selling_asset_issuer as asset_a_issuer + , selling_asset_type as asset_a_type , buying_asset_id as asset_b + , buying_asset_code as asset_b_code + , buying_asset_issuer as asset_b_issuer + , buying_asset_type as asset_b_type , trade_key , price_n , price_d @@ -41,7 +53,13 @@ with day_agg , ledger_closed_at , selling_asset_id as asset_b + , selling_asset_code as asset_b_code + , selling_asset_issuer as asset_b_issuer + , selling_asset_type as asset_b_type , buying_asset_id as asset_a + , buying_asset_code as asset_a_code + , buying_asset_issuer as asset_a_issuer + , buying_asset_type as asset_a_type , trade_key , price_n , price_d @@ -56,7 +74,13 @@ with day_agg , ledger_closed_at , asset_a + , asset_a_code + , asset_a_issuer + , asset_a_type , asset_b + , asset_b_code + , asset_b_issuer + , asset_b_type , trade_key , price_n , price_d @@ -76,7 +100,13 @@ with day_agg , ledger_closed_at , asset_a + , asset_a_code + , asset_a_issuer + , asset_a_type , asset_b + , asset_b_code + , asset_b_issuer + , asset_b_type , trade_key , price_n , price_d @@ -91,7 +121,13 @@ with select day_agg , asset_a + , asset_a_code + , asset_a_issuer + , asset_a_type , asset_b + , asset_b_code + , asset_b_issuer + , asset_b_type , count(trade_key) as trade_count_daily , sum(asset_a_amount) as asset_a_volume_daily , sum(asset_b_amount) as asset_b_volume_daily @@ -102,7 +138,13 @@ with group by day_agg , asset_a + , asset_a_code + , asset_a_issuer + , asset_a_type , asset_b + , asset_b_code + , asset_b_issuer + , asset_b_type ) /* obtain window function metrics for the asset pair */ @@ -110,7 +152,13 @@ with select day_agg , asset_a + , asset_a_code + , asset_a_issuer + , asset_a_type , asset_b + , asset_b_code + , asset_b_issuer + , asset_b_type , ledger_closed_at , first_value(price_n) over ( partition by @@ -155,7 +203,13 @@ with select trade_day_agg_group.day_agg , trade_day_agg_group.asset_a + , trade_day_agg_group.asset_a_code + , trade_day_agg_group.asset_a_issuer + , trade_day_agg_group.asset_a_type , trade_day_agg_group.asset_b + , trade_day_agg_group.asset_b_code + , trade_day_agg_group.asset_b_issuer + , trade_day_agg_group.asset_b_type , trade_day_agg_group.trade_count_daily , trade_day_agg_group.asset_a_volume_daily , trade_day_agg_group.asset_b_volume_daily diff --git a/models/intermediate/trades/int_trade_agg_month.sql b/models/intermediate/trades/int_trade_agg_month.sql index 2b99d93..05a7e82 100644 --- a/models/intermediate/trades/int_trade_agg_month.sql +++ b/models/intermediate/trades/int_trade_agg_month.sql @@ -10,7 +10,13 @@ with ledger_closed_at , cast(ledger_closed_at as date) as day_agg , selling_asset_id + , selling_asset_code + , selling_asset_issuer + , selling_asset_type , buying_asset_id + , buying_asset_code + , buying_asset_issuer + , buying_asset_type , concat(history_operation_id, `order`) as trade_key , price_n , price_d @@ -29,7 +35,13 @@ with day_agg , ledger_closed_at , selling_asset_id as asset_a + , selling_asset_code as asset_a_code + , selling_asset_issuer as asset_a_issuer + , selling_asset_type as asset_a_type , buying_asset_id as asset_b + , buying_asset_code as asset_b_code + , buying_asset_issuer as asset_b_issuer + , buying_asset_type as asset_b_type , trade_key , price_n , price_d @@ -41,7 +53,13 @@ with day_agg , ledger_closed_at , selling_asset_id as asset_b + , selling_asset_code as asset_b_code + , selling_asset_issuer as asset_b_issuer + , selling_asset_type as asset_b_type , buying_asset_id as asset_a + , buying_asset_code as asset_a_code + , buying_asset_issuer as asset_a_issuer + , buying_asset_type as asset_a_type , trade_key , price_n , price_d @@ -56,7 +74,13 @@ with day_agg , ledger_closed_at , asset_a + , asset_a_code + , asset_a_issuer + , asset_a_type , asset_b + , asset_b_code + , asset_b_issuer + , asset_b_type , trade_key , price_n , price_d @@ -76,7 +100,13 @@ with day_agg , ledger_closed_at , asset_a + , asset_a_code + , asset_a_issuer + , asset_a_type , asset_b + , asset_b_code + , asset_b_issuer + , asset_b_type , trade_key , price_n , price_d @@ -91,7 +121,13 @@ with select date('{{ dbt_airflow_macros.ds() }}') as day_agg , asset_a + , asset_a_code + , asset_a_issuer + , asset_a_type , asset_b + , asset_b_code + , asset_b_issuer + , asset_b_type , count(trade_key) as trade_count_monthly , sum(asset_a_amount) as asset_a_volume_monthly , sum(asset_b_amount) as asset_b_volume_monthly @@ -102,7 +138,13 @@ with where cast(ledger_closed_at as date) >= date_sub(date('{{ dbt_airflow_macros.ds() }}'), interval 30 day) group by asset_a + , asset_a_code + , asset_a_issuer + , asset_a_type , asset_b + , asset_b_code + , asset_b_issuer + , asset_b_type ) /* obtain window function metrics for the asset pair */ @@ -110,7 +152,13 @@ with select day_agg , asset_a + , asset_a_code + , asset_a_issuer + , asset_a_type , asset_b + , asset_b_code + , asset_b_issuer + , asset_b_type , ledger_closed_at , first_value(price_n) over ( partition by @@ -156,7 +204,13 @@ with select trade_day_agg_group.day_agg , trade_day_agg_group.asset_a + , trade_day_agg_group.asset_a_code + , trade_day_agg_group.asset_a_issuer + , trade_day_agg_group.asset_a_type , trade_day_agg_group.asset_b + , trade_day_agg_group.asset_b_code + , trade_day_agg_group.asset_b_issuer + , trade_day_agg_group.asset_b_type , trade_day_agg_group.trade_count_monthly , trade_day_agg_group.asset_a_volume_monthly , trade_day_agg_group.asset_b_volume_monthly diff --git a/models/intermediate/trades/int_trade_agg_week.sql b/models/intermediate/trades/int_trade_agg_week.sql index f534a2d..3fb6f67 100644 --- a/models/intermediate/trades/int_trade_agg_week.sql +++ b/models/intermediate/trades/int_trade_agg_week.sql @@ -10,7 +10,13 @@ with ledger_closed_at , cast(ledger_closed_at as date) as day_agg , selling_asset_id + , selling_asset_code + , selling_asset_issuer + , selling_asset_type , buying_asset_id + , buying_asset_code + , buying_asset_issuer + , buying_asset_type , concat(history_operation_id, `order`) as trade_key , price_n , price_d @@ -29,7 +35,13 @@ with day_agg , ledger_closed_at , selling_asset_id as asset_a + , selling_asset_code as asset_a_code + , selling_asset_issuer as asset_a_issuer + , selling_asset_type as asset_a_type , buying_asset_id as asset_b + , buying_asset_code as asset_b_code + , buying_asset_issuer as asset_b_issuer + , buying_asset_type as asset_b_type , trade_key , price_n , price_d @@ -41,7 +53,13 @@ with day_agg , ledger_closed_at , selling_asset_id as asset_b + , selling_asset_code as asset_b_code + , selling_asset_issuer as asset_b_issuer + , selling_asset_type as asset_b_type , buying_asset_id as asset_a + , buying_asset_code as asset_a_code + , buying_asset_issuer as asset_a_issuer + , buying_asset_type as asset_a_type , trade_key , price_n , price_d @@ -56,7 +74,13 @@ with day_agg , ledger_closed_at , asset_a + , asset_a_code + , asset_a_issuer + , asset_a_type , asset_b + , asset_b_code + , asset_b_issuer + , asset_b_type , trade_key , price_n , price_d @@ -76,7 +100,13 @@ with day_agg , ledger_closed_at , asset_a + , asset_a_code + , asset_a_issuer + , asset_a_type , asset_b + , asset_b_code + , asset_b_issuer + , asset_b_type , trade_key , price_n , price_d @@ -91,7 +121,13 @@ with select date('{{ dbt_airflow_macros.ds() }}') as day_agg , asset_a + , asset_a_code + , asset_a_issuer + , asset_a_type , asset_b + , asset_b_code + , asset_b_issuer + , asset_b_type , count(trade_key) as trade_count_weekly , sum(asset_a_amount) as asset_a_volume_weekly , sum(asset_b_amount) as asset_b_volume_weekly @@ -110,7 +146,13 @@ with select day_agg , asset_a + , asset_a_code + , asset_a_issuer + , asset_a_type , asset_b + , asset_b_code + , asset_b_issuer + , asset_b_type , ledger_closed_at , first_value(price_n) over ( partition by @@ -156,7 +198,13 @@ with select trade_day_agg_group.day_agg , trade_day_agg_group.asset_a + , trade_day_agg_group.asset_a_code + , trade_day_agg_group.asset_a_issuer + , trade_day_agg_group.asset_a_type , trade_day_agg_group.asset_b + , trade_day_agg_group.asset_b_code + , trade_day_agg_group.asset_b_issuer + , trade_day_agg_group.asset_b_type , trade_day_agg_group.trade_count_weekly , trade_day_agg_group.asset_a_volume_weekly , trade_day_agg_group.asset_b_volume_weekly diff --git a/models/intermediate/trades/int_trade_agg_year.sql b/models/intermediate/trades/int_trade_agg_year.sql index 6cd7b81..fa6afb4 100644 --- a/models/intermediate/trades/int_trade_agg_year.sql +++ b/models/intermediate/trades/int_trade_agg_year.sql @@ -16,7 +16,13 @@ with ledger_closed_at , cast(ledger_closed_at as date) as day_agg , selling_asset_id + , selling_asset_code + , selling_asset_issuer + , selling_asset_type , buying_asset_id + , buying_asset_code + , buying_asset_issuer + , buying_asset_type , concat(history_operation_id, `order`) as trade_key , price_n , price_d @@ -36,7 +42,13 @@ with day_agg , ledger_closed_at , selling_asset_id as asset_a + , selling_asset_code as asset_a_code + , selling_asset_issuer as asset_a_issuer + , selling_asset_type as asset_a_type , buying_asset_id as asset_b + , buying_asset_code as asset_b_code + , buying_asset_issuer as asset_b_issuer + , buying_asset_type as asset_b_type , trade_key , price_n , price_d @@ -48,7 +60,13 @@ with day_agg , ledger_closed_at , selling_asset_id as asset_b + , selling_asset_code as asset_b_code + , selling_asset_issuer as asset_b_issuer + , selling_asset_type as asset_b_type , buying_asset_id as asset_a + , buying_asset_code as asset_a_code + , buying_asset_issuer as asset_a_issuer + , buying_asset_type as asset_a_type , trade_key , price_n , price_d @@ -63,7 +81,13 @@ with day_agg , ledger_closed_at , asset_a + , asset_a_code + , asset_a_issuer + , asset_a_type , asset_b + , asset_b_code + , asset_b_issuer + , asset_b_type , trade_key , price_n , price_d @@ -83,7 +107,13 @@ with day_agg , ledger_closed_at , asset_a + , asset_a_code + , asset_a_issuer + , asset_a_type , asset_b + , asset_b_code + , asset_b_issuer + , asset_b_type , trade_key , price_n , price_d @@ -98,7 +128,13 @@ with select date('{{ dbt_airflow_macros.ds() }}') as day_agg , asset_a + , asset_a_code + , asset_a_issuer + , asset_a_type , asset_b + , asset_b_code + , asset_b_issuer + , asset_b_type , count(trade_key) as trade_count_yearly , sum(asset_a_amount) as asset_a_volume_yearly , sum(asset_b_amount) as asset_b_volume_yearly @@ -117,7 +153,13 @@ with select day_agg , asset_a + , asset_a_code + , asset_a_issuer + , asset_a_type , asset_b + , asset_b_code + , asset_b_issuer + , asset_b_type , ledger_closed_at , first_value(price_n) over ( partition by @@ -163,7 +205,13 @@ with select trade_day_agg_group.day_agg , trade_day_agg_group.asset_a + , trade_day_agg_group.asset_a_code + , trade_day_agg_group.asset_a_issuer + , trade_day_agg_group.asset_a_type , trade_day_agg_group.asset_b + , trade_day_agg_group.asset_b_code + , trade_day_agg_group.asset_b_issuer + , trade_day_agg_group.asset_b_type , trade_day_agg_group.trade_count_yearly , trade_day_agg_group.asset_a_volume_yearly , trade_day_agg_group.asset_b_volume_yearly diff --git a/models/marts/trade_agg.sql b/models/marts/trade_agg.sql index b32f549..54f5004 100644 --- a/models/marts/trade_agg.sql +++ b/models/marts/trade_agg.sql @@ -49,9 +49,15 @@ with , join_trades as ( select - join_table_daily.day_agg - , join_table_daily.asset_a - , join_table_daily.asset_b + join_table_yearly.day_agg + , join_table_yearly.asset_a + , join_table_yearly.asset_a_code + , join_table_yearly.asset_a_issuer + , join_table_yearly.asset_a_type + , join_table_yearly.asset_b + , join_table_yearly.asset_b_code + , join_table_yearly.asset_b_issuer + , join_table_yearly.asset_b_type , join_table_daily.trade_count_daily , join_table_daily.asset_a_volume_daily , join_table_daily.asset_b_volume_daily @@ -92,7 +98,7 @@ with , join_table_yearly.open_d_yearly , join_table_yearly.close_n_yearly , join_table_yearly.close_d_yearly - from trade_daily as join_table_daily + from trade_yearly as join_table_yearly left join trade_weekly as join_table_weekly on join_table_daily.day_agg = join_table_weekly.day_agg @@ -104,120 +110,11 @@ with and join_table_daily.asset_a = join_table_monthly.asset_a and join_table_daily.asset_b = join_table_monthly.asset_b left join - trade_yearly as join_table_yearly + trade_daily as join_table_daily on join_table_daily.day_agg = join_table_yearly.day_agg and join_table_daily.asset_a = join_table_yearly.asset_a and join_table_daily.asset_b = join_table_yearly.asset_b ) - , join_asset_a as ( - select - join_trades.day_agg - , join_trades.asset_a - , join_trades.asset_b - , join_trades.trade_count_daily - , join_trades.asset_a_volume_daily - , join_trades.asset_b_volume_daily - , join_trades.avg_price_daily - , join_trades.high_price_daily - , join_trades.low_price_daily - , join_trades.open_n_daily - , join_trades.open_d_daily - , join_trades.close_n_daily - , join_trades.close_d_daily - , join_trades.trade_count_weekly - , join_trades.asset_a_volume_weekly - , join_trades.asset_b_volume_weekly - , join_trades.avg_price_weekly - , join_trades.high_price_weekly - , join_trades.low_price_weekly - , join_trades.open_n_weekly - , join_trades.open_d_weekly - , join_trades.close_n_weekly - , join_trades.close_d_weekly - , join_trades.trade_count_monthly - , join_trades.asset_a_volume_monthly - , join_trades.asset_b_volume_monthly - , join_trades.avg_price_monthly - , join_trades.high_price_monthly - , join_trades.low_price_monthly - , join_trades.open_n_monthly - , join_trades.open_d_monthly - , join_trades.close_n_monthly - , join_trades.close_d_monthly - , join_trades.trade_count_yearly - , join_trades.asset_a_volume_yearly - , join_trades.asset_b_volume_yearly - , join_trades.avg_price_yearly - , join_trades.high_price_yearly - , join_trades.low_price_yearly - , join_trades.open_n_yearly - , join_trades.open_d_yearly - , join_trades.close_n_yearly - , join_trades.close_d_yearly - , ha.asset_code as asset_a_code - , ha.asset_type as asset_a_type - , ha.asset_issuer as asset_a_issuer - from join_trades - left join history_assets as ha - on join_trades.asset_a = ha.asset_id - ) - - , join_asset_b as ( - select - join_asset_a.day_agg - , join_asset_a.asset_a_type - , join_asset_a.asset_a_code - , join_asset_a.asset_a_issuer - , join_asset_a.asset_a - , ha.asset_code as asset_b_code - , ha.asset_type as asset_b_type - , ha.asset_issuer as asset_b_issuer - , join_asset_a.asset_b - , join_asset_a.trade_count_daily - , join_asset_a.asset_a_volume_daily - , join_asset_a.asset_b_volume_daily - , join_asset_a.avg_price_daily - , join_asset_a.high_price_daily - , join_asset_a.low_price_daily - , join_asset_a.open_n_daily - , join_asset_a.open_d_daily - , join_asset_a.close_n_daily - , join_asset_a.close_d_daily - , join_asset_a.trade_count_weekly - , join_asset_a.asset_a_volume_weekly - , join_asset_a.asset_b_volume_weekly - , join_asset_a.avg_price_weekly - , join_asset_a.high_price_weekly - , join_asset_a.low_price_weekly - , join_asset_a.open_n_weekly - , join_asset_a.open_d_weekly - , join_asset_a.close_n_weekly - , join_asset_a.close_d_weekly - , join_asset_a.trade_count_monthly - , join_asset_a.asset_a_volume_monthly - , join_asset_a.asset_b_volume_monthly - , join_asset_a.avg_price_monthly - , join_asset_a.high_price_monthly - , join_asset_a.low_price_monthly - , join_asset_a.open_n_monthly - , join_asset_a.open_d_monthly - , join_asset_a.close_n_monthly - , join_asset_a.close_d_monthly - , join_asset_a.trade_count_yearly - , join_asset_a.asset_a_volume_yearly - , join_asset_a.asset_b_volume_yearly - , join_asset_a.avg_price_yearly - , join_asset_a.high_price_yearly - , join_asset_a.low_price_yearly - , join_asset_a.open_n_yearly - , join_asset_a.open_d_yearly - , join_asset_a.close_n_yearly - , join_asset_a.close_d_yearly - from join_asset_a - left join history_assets as ha - on join_asset_a.asset_b = ha.asset_id - ) - select * -from join_asset_b \ No newline at end of file +from join_trades \ No newline at end of file From 673900249977250b1af917c5a6edeb085f873fb8 Mon Sep 17 00:00:00 2001 From: Simon Chow Date: Tue, 23 Jan 2024 13:48:32 -0500 Subject: [PATCH 2/3] Add correct group by --- models/intermediate/trades/int_trade_agg_week.sql | 6 ++++++ models/intermediate/trades/int_trade_agg_year.sql | 6 ++++++ 2 files changed, 12 insertions(+) diff --git a/models/intermediate/trades/int_trade_agg_week.sql b/models/intermediate/trades/int_trade_agg_week.sql index 3fb6f67..cfa7d79 100644 --- a/models/intermediate/trades/int_trade_agg_week.sql +++ b/models/intermediate/trades/int_trade_agg_week.sql @@ -138,7 +138,13 @@ with where cast(ledger_closed_at as date) >= date_sub(date('{{ dbt_airflow_macros.ds() }}'), interval 7 day) group by asset_a + , asset_a_code + , asset_a_issuer + , asset_a_type , asset_b + , asset_b_code + , asset_b_issuer + , asset_b_type ) /* obtain window function metrics for the asset pair */ diff --git a/models/intermediate/trades/int_trade_agg_year.sql b/models/intermediate/trades/int_trade_agg_year.sql index fa6afb4..b5db74c 100644 --- a/models/intermediate/trades/int_trade_agg_year.sql +++ b/models/intermediate/trades/int_trade_agg_year.sql @@ -145,7 +145,13 @@ with where cast(ledger_closed_at as date) >= date_sub(date('{{ dbt_airflow_macros.ds() }}'), interval 365 day) group by asset_a + , asset_a_code + , asset_a_issuer + , asset_a_type , asset_b + , asset_b_code + , asset_b_issuer + , asset_b_type ) /* obtain window function metrics for the asset pair */ From 479fea7da6011cbb17ec0a21138fa7318a46ee4c Mon Sep 17 00:00:00 2001 From: Simon Chow Date: Tue, 23 Jan 2024 13:51:20 -0500 Subject: [PATCH 3/3] Update joins --- models/marts/trade_agg.sql | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/models/marts/trade_agg.sql b/models/marts/trade_agg.sql index 54f5004..f1c6929 100644 --- a/models/marts/trade_agg.sql +++ b/models/marts/trade_agg.sql @@ -101,19 +101,19 @@ with from trade_yearly as join_table_yearly left join trade_weekly as join_table_weekly - on join_table_daily.day_agg = join_table_weekly.day_agg - and join_table_daily.asset_a = join_table_weekly.asset_a - and join_table_daily.asset_b = join_table_weekly.asset_b + on join_table_yearly.day_agg = join_table_weekly.day_agg + and join_table_yearly.asset_a = join_table_weekly.asset_a + and join_table_yearly.asset_b = join_table_weekly.asset_b left join trade_monthly as join_table_monthly - on join_table_daily.day_agg = join_table_monthly.day_agg - and join_table_daily.asset_a = join_table_monthly.asset_a - and join_table_daily.asset_b = join_table_monthly.asset_b + on join_table_yearly.day_agg = join_table_monthly.day_agg + and join_table_yearly.asset_a = join_table_monthly.asset_a + and join_table_yearly.asset_b = join_table_monthly.asset_b left join trade_daily as join_table_daily - on join_table_daily.day_agg = join_table_yearly.day_agg - and join_table_daily.asset_a = join_table_yearly.asset_a - and join_table_daily.asset_b = join_table_yearly.asset_b + on join_table_yearly.day_agg = join_table_daily.day_agg + and join_table_yearly.asset_a = join_table_daily.asset_a + and join_table_yearly.asset_b = join_table_daily.asset_b ) select *