Merge branch 'main' of gitlab.com:cief-data/dbt_cloud into shipping-fix-order-created-datetime

models/marts/reporting/rep_exchange__daily_orders.sql

@@ -6,9 +6,9 @@
 
 -- working days mapping for each service type
 {% set service_type_working_day_mapping = {
-    '1 DAY TRANSFER': 1,
+    'BANK TRANSFER (EXPRESS)': 1,
     '1688 PAYMENT': 1,
-    '3 DAYS TRANSFER': 3,
+    'BANK TRANSFER (SAVER)': 3,
     'ENTERPRISE TO ENTERPRISE 公打公': 7
 } %}
 
@@ -209,8 +209,9 @@ fct_and_dim_joins AS (
         -- On time delivery boolean (CIEF internal SLA)
         CASE
             -- On-time boolean is null when order is not complete
-            WHEN service_type IN ('BA') THEN NULL
+            WHEN service_type IN ('BA', 'RECEIVED-ON-BEHALF') THEN NULL
             WHEN transaction_status NOT IN ('UPLOADED_BANK_SLIP') THEN NULL
+            WHEN order_created_datetime < '2023-01-01' THEN NULL
             WHEN estimated_order_delivery_datetime_website_sla > orders.operation_uploaded_bank_slip_datetime THEN 1
             WHEN estimated_order_delivery_datetime_website_sla < orders.operation_uploaded_bank_slip_datetime THEN 0
             ELSE -999
@@ -219,8 +220,9 @@ fct_and_dim_joins AS (
         -- On time delivery boolean (Customer Expectation)
         CASE
             -- On-time boolean is null when order is not complete
-            WHEN service_type IN ('BA') THEN NULL
+            WHEN service_type IN ('BA', 'RECEIVED-ON-BEHALF') THEN NULL
             WHEN transaction_status NOT IN ('UPLOADED_BANK_SLIP') THEN NULL
+            WHEN order_created_datetime < '2023-01-01' THEN NULL
             WHEN estimated_order_delivery_datetime_customer_expectation > orders.operation_uploaded_bank_slip_datetime THEN 1
             WHEN estimated_order_delivery_datetime_customer_expectation < orders.operation_uploaded_bank_slip_datetime THEN 0
             ELSE -999

models/marts/reporting/rep_exchange__exchange_rate_analysis.sql

@@ -5,7 +5,7 @@ WITH currency_exchange_rate_compilation AS (
 
 orders AS (
     SELECT * FROM {{ ref('fct_exchange__transaction_orders') }}
-    WHERE service_type NOT IN ('BA')
+    WHERE service_type NOT IN ('BA', 'RECEIVED-ON-BEHALF')
 ),
 
 
@@ -59,7 +59,7 @@ restructure_and_expand_table_cief_vs_world_rate AS (
 
 ),
 
-restructure_and_expand_table_cief_1688_vs_c2m_altr AS (
+restructure_and_expand_table_cief_1688_vs_c2m_tbdf AS (
     
     SELECT
         t1.provider,
@@ -83,7 +83,7 @@ restructure_and_expand_table_cief_1688_vs_c2m_altr AS (
     LEFT JOIN currency_exchange_rate_compilation t2
         ON ( t1.exchange_rate_date = t2.exchange_rate_date )
         AND t2.provider = 'C2M'
-        AND t2.service_type = 'ALTR'
+        AND t2.service_type = 'TBDF'
 
     WHERE 
         t1.provider = 'CIEF'
@@ -94,7 +94,7 @@ restructure_and_expand_table_cief_1688_vs_c2m_altr AS (
 
 ),
 
-restructure_and_expand_table_cief_1_day_transfer_vs_c2m_tbdf AS (
+restructure_and_expand_table_cief_bank_transfer_express_vs_c2m_altr AS (
     
     SELECT
         t1.provider,
@@ -118,7 +118,7 @@ restructure_and_expand_table_cief_1_day_transfer_vs_c2m_tbdf AS (
     LEFT JOIN currency_exchange_rate_compilation t2
         ON ( t1.exchange_rate_date = t2.exchange_rate_date )
         AND t2.provider = 'C2M'
-        AND t2.service_type = 'TBDF'
+        AND t2.service_type = 'ALTR'
 
     CROSS JOIN get_base_myr_value_for_comparison
 
@@ -127,11 +127,11 @@ restructure_and_expand_table_cief_1_day_transfer_vs_c2m_tbdf AS (
         AND
         competitor_name IN ('C2M')
         AND 
-        t1.service_type IN ('1 DAY TRANSFER')
+        t1.service_type IN ('BANK TRANSFER (EXPRESS)')
 
 ),
 
-restructure_and_expand_table_cief_3_days_vs_sd AS (
+restructure_and_expand_table_cief_bank_transfer_saver_vs_sd AS (
     
     SELECT
         t1.provider,
@@ -163,7 +163,7 @@ restructure_and_expand_table_cief_3_days_vs_sd AS (
         AND
         competitor_name IN ('SD')
         AND 
-        t1.service_type IN ('3 DAYS TRANSFER')
+        t1.service_type IN ('BANK TRANSFER (SAVER)')
 
 ),
 
@@ -171,21 +171,23 @@ union_expanded_tables AS (
     
     SELECT * FROM restructure_and_expand_table_cief_vs_world_rate
     UNION ALL
-    SELECT * FROM restructure_and_expand_table_cief_1688_vs_c2m_altr
+    SELECT * FROM restructure_and_expand_table_cief_1688_vs_c2m_tbdf
     UNION ALL
-    SELECT * FROM restructure_and_expand_table_cief_1_day_transfer_vs_c2m_tbdf
+    SELECT * FROM restructure_and_expand_table_cief_bank_transfer_express_vs_c2m_altr
     UNION ALL
-    SELECT * FROM restructure_and_expand_table_cief_3_days_vs_sd
+    SELECT * FROM restructure_and_expand_table_cief_bank_transfer_saver_vs_sd
 
 ),
 
-rate_comparison AS (
+exchange_amount_compute AS (
     
     SELECT
         *,
         
+        -- CIEF exchanged CNY after deducting service charge
         ( (comparison_base_value_myr / 1.02) * myr_to_cny_currency_exchange_rate ) AS cief_amount_cny,
 
+        -- Competitor exchanged CNY after deducting service charge
         CASE
             WHEN competitor_name = 'WORLD RATE'
                 THEN comparison_base_value_myr * competitor_myr_to_cny_currency_exchange_rate
@@ -197,6 +199,25 @@ rate_comparison AS (
                 THEN ( comparison_base_value_myr * competitor_myr_to_cny_currency_exchange_rate ) - handling_fee_cny
         END AS competitor_amount_cny,
 
+        -- Median and Average difference in CNY amount exchanged comparing CIEF to competitor
+        MEDIAN(1 - DIV0(competitor_amount_cny, cief_amount_cny)) OVER (
+            PARTITION BY service_type, exchange_rate_date, competitor_name, competitor_service_type, competitor_payment_method
+            )   AS median_amount_percent_difference,
+
+        median_amount_percent_difference - 0.5 * median_amount_percent_difference AS upper_bound_median_amount_percent_difference,
+        median_amount_percent_difference + 0.5 * median_amount_percent_difference AS lower_bound_median_amount_percent_difference,
+
+        AVG(1 - DIV0(competitor_amount_cny, cief_amount_cny)) OVER (
+            PARTITION BY service_type, exchange_rate_date, competitor_name, competitor_service_type, competitor_payment_method
+            )   AS average_amount_percent_difference,
+
+        average_amount_percent_difference - 0.5 * average_amount_percent_difference AS upper_bound_average_amount_percent_difference,
+        average_amount_percent_difference + 0.5 * average_amount_percent_difference AS lower_bound_average_amount_percent_difference,
+
+        AVG(1 - DIV0(COMPETITOR_AMOUNT_CNY, CIEF_AMOUNT_CNY)) OVER (
+            PARTITION BY service_type, exchange_rate_date, payment_method, competitor_name, competitor_service_type, competitor_payment_method
+            )   AS comparison_average_percent_difference,
+
         MAX(exchange_rate_date) OVER ( PARTITION BY competitor_name, payment_method, competitor_payment_method)     AS latest_exchange_rate_date
 
     FROM union_expanded_tables
@@ -227,23 +248,31 @@ order_metrics AS (
 exchange_rate_join_order_metrics AS (
 
     SELECT
-        rate_comparison.provider,
+        exchange_amount_compute.provider,
-        rate_comparison.payment_method,
+        exchange_amount_compute.payment_method,
-        rate_comparison.service_type,
+        exchange_amount_compute.service_type,
-        rate_comparison.myr_to_cny_currency_exchange_rate,
+        exchange_amount_compute.myr_to_cny_currency_exchange_rate,
-        rate_comparison.exchange_rate_date,
+        exchange_amount_compute.exchange_rate_date,
-        rate_comparison.competitor_name,
+        exchange_amount_compute.competitor_name,
-        rate_comparison.start_range_myr,
+        exchange_amount_compute.start_range_myr,
-        rate_comparison.end_range_myr,
+        exchange_amount_compute.end_range_myr,
-        rate_comparison.competitor_payment_method,
+        exchange_amount_compute.competitor_payment_method,
-        rate_comparison.competitor_service_type,
+        exchange_amount_compute.competitor_service_type,
-        rate_comparison.handling_fee_myr,
+        exchange_amount_compute.handling_fee_myr,
-        rate_comparison.handling_fee_cny,
+        exchange_amount_compute.handling_fee_cny,
-        rate_comparison.competitor_myr_to_cny_currency_exchange_rate,
+        exchange_amount_compute.competitor_myr_to_cny_currency_exchange_rate,
-        rate_comparison.comparison_base_value_myr,
+        exchange_amount_compute.comparison_base_value_myr,
-        rate_comparison.cief_amount_cny,
+        exchange_amount_compute.cief_amount_cny,
-        rate_comparison.competitor_amount_cny,
+        exchange_amount_compute.competitor_amount_cny,
-        rate_comparison.latest_exchange_rate_date,
+        exchange_amount_compute.latest_exchange_rate_date,
+        exchange_amount_compute.median_amount_percent_difference,
+        exchange_amount_compute.upper_bound_median_amount_percent_difference,
+        exchange_amount_compute.lower_bound_median_amount_percent_difference,
+        exchange_amount_compute.average_amount_percent_difference,
+        exchange_amount_compute.upper_bound_average_amount_percent_difference,
+        exchange_amount_compute.lower_bound_average_amount_percent_difference,
+        exchange_amount_compute.comparison_average_percent_difference,
+
         order_metrics.daily_sales_cny,
         order_metrics.total_daily_sales_cny,
         order_metrics.daily_sales_myr,
@@ -253,14 +282,14 @@ exchange_rate_join_order_metrics AS (
 
         '{{ modules.datetime.datetime.now(modules.pytz.timezone("Asia/Kuala_Lumpur")) }}'                       AS _dbt_ran_datetime 
 
-    FROM rate_comparison
+    FROM exchange_amount_compute
 
     LEFT JOIN order_metrics
         ON 
         ( 
-            rate_comparison.exchange_rate_date  = order_metrics.order_date 
+            exchange_amount_compute.exchange_rate_date  = order_metrics.order_date 
-            AND rate_comparison.service_type    = order_metrics.service_type
+            AND exchange_amount_compute.service_type    = order_metrics.service_type
-            AND rate_comparison.payment_method  = order_metrics.order_payment_method
+            AND exchange_amount_compute.payment_method  = order_metrics.order_payment_method
         )       
 
 ),
@@ -296,6 +325,13 @@ final__rep_exchange__exchange_rate_analysis AS (
         total_daily_sales_cny,
         daily_sales_myr,
         total_daily_sales_myr,
+        median_amount_percent_difference,
+        upper_bound_median_amount_percent_difference,
+        lower_bound_median_amount_percent_difference,
+        average_amount_percent_difference,
+        upper_bound_average_amount_percent_difference,
+        lower_bound_average_amount_percent_difference,
+        comparison_average_percent_difference,
 
         -- date/times
         exchange_rate_date,

models/marts/reporting/rep_exchange__wallet_cash_flow_analysis.sql

@@ -0,0 +1,85 @@
+-- IMPORT
+WITH daily_wallet_transactions AS (
+    SELECT * FROM {{ ref('fct_exchange__wallet_daily_transactions') }}
+),
+
+companies AS (
+    SELECT * FROM {{ ref('dim_exchange__companies') }}
+),
+
+
+-- LOGIC
+daily_wallet_transaction_enrich AS (
+
+    SELECT 
+        daily_wallet_transactions.* EXCLUDE _dbt_ran_datetime,
+
+        companies.company_marking_id,
+        companies.company_type,
+        companies.business_type,
+        companies.exchange_rate_segment,
+        companies.is_migrated_company,
+        companies.company_created_datetime,
+
+        MD5_NUMBER_LOWER64(CONCAT(daily_wallet_transactions.wallet_id, transaction_date))::STRING       AS wallet_cash_flow_transaction_id,
+
+        '{{ modules.datetime.datetime.now(modules.pytz.timezone("Asia/Kuala_Lumpur")) }}'               AS _dbt_ran_datetime
+
+    FROM daily_wallet_transactions
+
+    LEFT JOIN companies
+        ON ( daily_wallet_transactions.company_id = companies.company_id)
+
+),
+
+
+-- FINAL
+final__rep_exchange__wallet_cash_flow_analysis AS (
+
+    SELECT
+        --ids
+        wallet_cash_flow_transaction_id,
+        wallet_id,
+        company_id,
+        company_marking_id,
+        currency_id,
+        first_wallet_log_id,
+        last_wallet_log_id,
+
+        --dimensions
+        company_type,
+        business_type,
+        exchange_rate_segment,
+        is_migrated_company,
+        is_latest_wallet_balance,
+
+        --measures
+        count_daily_wallet_activity,
+        count_total_daily_transaction_credit,
+        count_total_daily_transaction_debit,
+        total_daily_transaction_credit,
+        total_daily_transaction_debit,
+        original_daily_first_wallet_balance,
+        original_daily_last_wallet_balance,
+        daily_0_origin_first_wallet_balance,
+        daily_0_origin_last_wallet_balance,
+        daily_calculated_first_wallet_balance,
+        daily_calculated_last_wallet_balance,
+
+        --date/time
+        transaction_date,
+        company_created_datetime,
+        wallet_registration_datetime,
+        first_daily_transaction_start_datetime,
+        first_daily_transaction_completed_datetime,
+        last_daily_transaction_start_datetime,
+        last_daily_transaction_completed_datetime,
+
+        --metadata
+        _dbt_ran_datetime
+
+    FROM daily_wallet_transaction_enrich
+
+)
+
+SELECT * FROM final__rep_exchange__wallet_cash_flow_analysis

models/marts/warehouse/fct_exchange__transaction_orders.sql

@@ -159,35 +159,35 @@ transaction_order_datetime_imputation AS (
                 THEN COALESCE(order_completed_datetime, DATEADD(minute, 433, new_order_approved_datetime))
 
             /*
-            Imputation for order_approved_datetime for 1 day transfer
+            Imputation for order_approved_datetime for BANK TRANSFER (EXPRESS)
                 - If payment_method is 'cash', add median of 59 minute to order_approved datetime
                 - If payment_method is 'wallet', add median of 206 minute to order_approved datetime
                 - If payment_method is 'cheque', reuse order_completed datetime, all orders have completed datetime
                 - If payment_method is 'payment_gateway', add median of 214.5 minute to order_approved datetime
             */
-            WHEN order_created_datetime >= '2023-01-01' AND service_type = '1 DAY TRANSFER' AND payment_method = 'CASH'  
+            WHEN order_created_datetime >= '2023-01-01' AND service_type = 'BANK TRANSFER (EXPRESS)' AND payment_method = 'CASH'  
                 THEN COALESCE(order_completed_datetime, DATEADD(minute, 59, new_order_approved_datetime))
-            WHEN order_created_datetime >= '2023-01-01' AND service_type = '1 DAY TRANSFER' AND payment_method = 'WALLET'  
+            WHEN order_created_datetime >= '2023-01-01' AND service_type = 'BANK TRANSFER (EXPRESS)' AND payment_method = 'WALLET'  
                 THEN COALESCE(order_completed_datetime, DATEADD(minute, 206, new_order_approved_datetime))
-            WHEN order_created_datetime >= '2023-01-01' AND service_type = '1 DAY TRANSFER' AND payment_method = 'CHEQUE'  
+            WHEN order_created_datetime >= '2023-01-01' AND service_type = 'BANK TRANSFER (EXPRESS)' AND payment_method = 'CHEQUE'  
                 THEN order_completed_datetime
-            WHEN order_created_datetime >= '2023-01-01' AND service_type = '1 DAY TRANSFER' AND payment_method = 'PAYMENT_GATEWAY'  
+            WHEN order_created_datetime >= '2023-01-01' AND service_type = 'BANK TRANSFER (EXPRESS)' AND payment_method = 'PAYMENT_GATEWAY'  
                 THEN COALESCE(order_completed_datetime, DATEADD(minute, 214.5, new_order_approved_datetime))
 
             /*
-            Imputation for order_approved_datetime for 3 days transfer
+            Imputation for order_approved_datetime for BANK TRANSFER (SAVER)
                 - If payment_method is 'cash', add median of 2702 minute to order_approved datetime
                 - If payment_method is 'wallet', add median of 4502.92 minute to order_approved datetime
                 - If payment_method is 'cheque', reuse order_completed datetime, all orders have completed datetime
                 - If payment_method is 'payment_gateway', add median of 5253 minute to order_approved datetime
             */
-            WHEN order_created_datetime >= '2023-01-01' AND service_type = '3 DAYS TRANSFER' AND payment_method = 'CASH'  
+            WHEN order_created_datetime >= '2023-01-01' AND service_type = 'BANK TRANSFER (SAVER)' AND payment_method = 'CASH'  
                 THEN COALESCE(order_completed_datetime, DATEADD(minute, 2702, new_order_approved_datetime))
-            WHEN order_created_datetime >= '2023-01-01' AND service_type = '3 DAYS TRANSFER' AND payment_method = 'WALLET'  
+            WHEN order_created_datetime >= '2023-01-01' AND service_type = 'BANK TRANSFER (SAVER)' AND payment_method = 'WALLET'  
                 THEN COALESCE(order_completed_datetime, DATEADD(minute, 4502.92, new_order_approved_datetime))
-            WHEN order_created_datetime >= '2023-01-01' AND service_type = '3 DAYS TRANSFER' AND payment_method = 'CHEQUE'  
+            WHEN order_created_datetime >= '2023-01-01' AND service_type = 'BANK TRANSFER (SAVER)' AND payment_method = 'CHEQUE'  
                 THEN order_completed_datetime
-            WHEN order_created_datetime >= '2023-01-01' AND service_type = '3 DAYS TRANSFER' AND payment_method = 'PAYMENT_GATEWAY'  
+            WHEN order_created_datetime >= '2023-01-01' AND service_type = 'BANK TRANSFER (SAVER)' AND payment_method = 'PAYMENT_GATEWAY'  
                 THEN COALESCE(order_completed_datetime, DATEADD(minute, 5253, new_order_approved_datetime))
                 
         END     AS new_order_completed_datetime
@@ -536,6 +536,8 @@ purchase_order_invoice_status_redefined AS (
         CASE 
             WHEN order_status IN ('REFUNDED')
                 THEN 'ORDER_REFUNDED'
+            WHEN order_status IN ('EXPIRED')
+                THEN 'ORDER_EXPIRED'
             WHEN order_status IN ('COMPLETED', 'APPROVED') AND cost_status IS NULL
                 THEN 'PENDING_GENERATE_WHITE_FORM'
             WHEN order_status IN ('COMPLETED', 'APPROVED') AND cost_status IN ('PENDING_VERIFICATION')

models/marts/warehouse/fct_exchange__wallet_daily_transactions.sql

@@ -0,0 +1,370 @@
+-- IMPORT
+WITH wallet_balance_logs AS (
+    SELECT * FROM {{ ref('stg_exchange__wallet_balance_logs') }}
+),
+
+latest_wallet_balances AS (
+    SELECT * FROM {{ ref('stg_exchange__wallet_balances') }}
+),
+
+wallet_transaction_logs AS (
+    SELECT * FROM {{ ref('stg_exchange__transaction_wallet_logs') }}
+    WHERE status IN ('APPROVED') 
+),
+
+
+-- LOGIC
+-- wallet transactions
+daily_transaction_log AS (
+
+    SELECT
+        wallet_id,
+        company_id,
+        base_currency_id,
+        quote_currency_id,
+        DATE(updated_datetime)      AS transaction_date,
+        
+        MIN(created_datetime)       AS first_daily_transaction_start_datetime,
+        MAX(created_datetime)       AS last_daily_transaction_start_datetime,
+        
+        MIN(updated_datetime)       AS first_daily_transaction_completed_datetime,
+        MAX(updated_datetime)       AS last_daily_transaction_completed_datetime,
+        
+        SUM(CASE 
+                WHEN transaction_type IN ('PAYMENT', 'DEBIT_NOTE')
+                    THEN -( base_value )
+                    ELSE 0
+            END) AS total_daily_transaction_debit,
+        
+        SUM(CASE
+                WHEN transaction_type IN ('TOP_UP', 'CREDIT_NOTE')
+                    THEN base_value
+                    ELSE 0
+            END) AS total_daily_transaction_credit,
+
+        COUNT(CASE
+                WHEN transaction_type IN ('TOP_UP', 'CREDIT_NOTE')
+                    THEN wallet_id
+            END) AS count_total_daily_transaction_credit,
+
+        COUNT(CASE 
+                WHEN transaction_type IN ('PAYMENT', 'DEBIT_NOTE')
+                    THEN wallet_id
+            END) AS count_total_daily_transaction_debit
+        
+    FROM wallet_transaction_logs
+
+    GROUP BY
+        wallet_id,
+        company_id,
+        base_currency_id,
+        quote_currency_id,
+        transaction_date
+    
+),
+
+-- wallet balances
+wallet_balance_list AS (
+
+    SELECT
+        wallet_log_id,
+        wallet_id,
+        company_id,
+        currency_id,
+        wallet_balance,
+        created_datetime    AS wallet_registration_datetime,
+        updated_datetime
+    FROM wallet_balance_logs
+    
+    UNION
+    
+    SELECT
+        MD5_NUMBER_LOWER64(CONCAT(wallet_id, wallet_balance, updated_datetime)) AS wallet_log_id,
+        wallet_id,
+        company_id,
+        currency_id,
+        wallet_balance,
+        created_datetime    AS wallet_registration_datetime,
+        updated_datetime
+    FROM latest_wallet_balances
+
+),
+
+wallet_balance_list_enrich AS (
+
+    SELECT
+        wallet_log_id,
+        wallet_id,
+        company_id,
+        currency_id,
+        wallet_balance,
+        wallet_registration_datetime,
+        updated_datetime,
+        DATE(updated_datetime)  AS updated_date,        
+        LAG(wallet_balance) OVER (PARTITION BY wallet_id ORDER BY updated_datetime)         AS prev_wallet_balance,
+        LEAD(wallet_balance) OVER (PARTITION BY wallet_id ORDER BY updated_datetime)        AS next_wallet_balance
+        
+    FROM wallet_balance_list
+
+),
+
+wallet_balance_daily_aggregate AS (
+
+    SELECT
+        DISTINCT
+        wallet_id,
+        company_id,
+        currency_id,
+        wallet_registration_datetime,
+        updated_date,
+        COUNT(wallet_id) OVER (PARTITION BY wallet_id, updated_date)                                        AS count_daily_wallet_activity,
+        FIRST_VALUE(wallet_log_id) OVER (PARTITION BY wallet_id, updated_date ORDER BY updated_datetime)    AS first_wallet_log_id,
+        LAST_VALUE(wallet_log_id) OVER (PARTITION BY wallet_id, updated_date ORDER BY updated_datetime)     AS last_wallet_log_id,
+        
+        FIRST_VALUE(wallet_balance) OVER (PARTITION BY wallet_id, updated_date ORDER BY updated_datetime)   AS daily_first_wallet_balance,
+        LAST_VALUE(wallet_balance) OVER (PARTITION BY wallet_id, updated_date ORDER BY updated_datetime)    AS daily_last_wallet_balance,
+        
+    FROM wallet_balance_list_enrich
+        
+),
+
+wallet_balance_list_join_wallet_transaction AS ( 
+
+    SELECT
+        COALESCE(wallet_balance_daily_aggregate.updated_date, daily_transaction_log.transaction_date)   AS transaction_date,
+        COALESCE(wallet_balance_daily_aggregate.wallet_id, daily_transaction_log.wallet_id)             AS wallet_id,
+        COALESCE(wallet_balance_daily_aggregate.company_id, daily_transaction_log.company_id)           AS company_id,
+        COALESCE(wallet_balance_daily_aggregate.currency_id, daily_transaction_log.base_currency_id)    AS currency_id,
+        
+        -- Wallet data
+        COALESCE(wallet_balance_daily_aggregate.count_daily_wallet_activity, 0)                         AS count_daily_wallet_activity,
+        COALESCE(wallet_balance_daily_aggregate.first_wallet_log_id, NULL)                              AS first_wallet_log_id,
+        COALESCE(wallet_balance_daily_aggregate.last_wallet_log_id, NULL)                               AS last_wallet_log_id,
+        COALESCE(wallet_balance_daily_aggregate.daily_first_wallet_balance, NULL)                       AS daily_first_wallet_balance,
+        COALESCE(wallet_balance_daily_aggregate.daily_last_wallet_balance, NULL)                        AS daily_last_wallet_balance,
+        COALESCE(wallet_balance_daily_aggregate.wallet_registration_datetime, NULL)                     AS wallet_registration_datetime,
+
+        -- Transaction data
+        COALESCE(daily_transaction_log.first_daily_transaction_start_datetime, NULL)                    AS first_daily_transaction_start_datetime,
+        COALESCE(daily_transaction_log.last_daily_transaction_start_datetime, NULL)                     AS last_daily_transaction_start_datetime,
+        COALESCE(daily_transaction_log.first_daily_transaction_completed_datetime, NULL)                AS first_daily_transaction_completed_datetime,
+        COALESCE(daily_transaction_log.last_daily_transaction_completed_datetime, NULL)                 AS last_daily_transaction_completed_datetime,
+        COALESCE(daily_transaction_log.total_daily_transaction_debit, NULL)                             AS total_daily_transaction_debit,
+        COALESCE(daily_transaction_log.total_daily_transaction_credit, NULL)                            AS total_daily_transaction_credit,
+        COALESCE(daily_transaction_log.count_total_daily_transaction_credit, 0)                         AS count_total_daily_transaction_credit,
+        COALESCE(daily_transaction_log.count_total_daily_transaction_debit, 0)                          AS count_total_daily_transaction_debit,
+
+    FROM wallet_balance_daily_aggregate
+
+    FULL OUTER JOIN daily_transaction_log
+        ON (wallet_balance_daily_aggregate.wallet_id = daily_transaction_log.wallet_id)
+        AND (wallet_balance_daily_aggregate.updated_date = daily_transaction_log.transaction_date)
+        
+),
+
+minor_data_imputation AS (
+
+    SELECT
+        wallet_id,
+        company_id,
+        currency_id,
+        first_wallet_log_id,
+        last_wallet_log_id,
+        count_daily_wallet_activity,
+        count_total_daily_transaction_credit,
+        count_total_daily_transaction_debit,
+        daily_last_wallet_balance,
+
+        CASE
+            WHEN ROW_NUMBER() OVER (PARTITION BY wallet_id ORDER BY transaction_date DESC) = 1
+            THEN 1
+            ELSE 0
+        END AS is_latest_wallet_balance,
+
+        CASE
+            WHEN ROW_NUMBER() OVER (PARTITION BY wallet_id ORDER BY transaction_date) = 1
+            THEN 1
+            ELSE 0
+        END AS is_earliest_wallet_balance,
+
+        /*
+        Initialize for older records without wallet balance. (2 scenario)
+        1. Set closing balance as 0
+        2. Set closing balance as difference between credit and debit
+        */
+        CASE 
+            WHEN is_earliest_wallet_balance = 1 
+                THEN COALESCE(daily_first_wallet_balance, 0) 
+            ELSE daily_first_wallet_balance
+        END daily_first_wallet_balance,
+
+        CASE 
+            WHEN is_earliest_wallet_balance = 1
+                THEN COALESCE(daily_last_wallet_balance, 0)
+            ELSE daily_last_wallet_balance
+        END daily_0_origin_last_wallet_balance,
+
+        CASE 
+            WHEN is_earliest_wallet_balance = 1
+                THEN COALESCE(daily_last_wallet_balance, total_daily_transaction_credit + total_daily_transaction_debit)
+            ELSE daily_last_wallet_balance
+        END daily_calculated_last_wallet_balance,
+
+        
+        CASE
+            WHEN wallet_registration_datetime IS NULL
+                THEN (
+                    SELECT MIN(wallet_registration_datetime)
+                    FROM wallet_balance_list_join_wallet_transaction t1
+                    WHERE t1.wallet_id = wallet_balance_list_join_wallet_transaction.wallet_id
+                    )
+        END AS new_wallet_registration_datetime,
+
+        COALESCE(total_daily_transaction_credit, 0)     AS total_daily_transaction_credit,
+        COALESCE(total_daily_transaction_debit, 0)      AS total_daily_transaction_debit,
+        transaction_date,
+        LEAD(transaction_date) OVER (PARTITION BY wallet_id ORDER BY transaction_date)      AS next_transaction_date,
+        LAG(transaction_date) OVER (PARTITION BY wallet_id ORDER BY transaction_date)       AS prev_transaction_date,
+        wallet_registration_datetime,
+        first_daily_transaction_start_datetime,
+        first_daily_transaction_completed_datetime,
+        last_daily_transaction_start_datetime,
+        last_daily_transaction_completed_datetime
+
+    FROM wallet_balance_list_join_wallet_transaction
+
+),
+
+recursive_imputation AS (
+
+    SELECT 
+        *,
+        
+        daily_first_wallet_balance              AS new_daily_0_origin_first_wallet_balance,
+        daily_0_origin_last_wallet_balance      AS new_daily_0_origin_last_wallet_balance,
+        
+        daily_first_wallet_balance              AS new_daily_calculated_first_wallet_balance,
+        daily_calculated_last_wallet_balance    AS new_daily_calculated_last_wallet_balance,
+        
+    FROM minor_data_imputation
+    
+    WHERE is_earliest_wallet_balance = 1
+
+    UNION ALL
+
+    SELECT
+        t1.*,
+        
+        t2.new_daily_0_origin_last_wallet_balance AS new_daily_0_origin_first_wallet_balance,
+        t2.new_daily_0_origin_last_wallet_balance + t1.total_daily_transaction_credit + t1.total_daily_transaction_debit  AS new_daily_0_origin_last_wallet_balance,
+        
+        t2.new_daily_calculated_last_wallet_balance AS new_daily_calculated_first_wallet_balance,
+        t2.new_daily_calculated_last_wallet_balance + t1.total_daily_transaction_credit + t1.total_daily_transaction_debit  AS new_daily_calculated_last_wallet_balance
+        
+        
+    FROM minor_data_imputation t1
+
+    INNER JOIN recursive_imputation t2
+    
+    ON  (
+        t1.wallet_id = t2.wallet_id
+        AND 
+        t2.transaction_date = t1.prev_transaction_date
+        )
+        
+),
+
+casting_and_organize AS (
+
+    SELECT
+        wallet_id,
+        company_id,
+        currency_id,
+        first_wallet_log_id::STRING     AS first_wallet_log_id,
+        last_wallet_log_id::STRING      AS last_wallet_log_id,
+        is_earliest_wallet_balance,
+        is_latest_wallet_balance,
+        count_daily_wallet_activity,
+        count_total_daily_transaction_credit,
+        count_total_daily_transaction_debit,
+        total_daily_transaction_credit,
+        total_daily_transaction_debit,
+        LAG(daily_last_wallet_balance) OVER (PARTITION BY wallet_id ORDER BY transaction_date) AS prev_daily_last_wallet_balance,
+        
+        COALESCE(CASE
+                WHEN prev_daily_last_wallet_balance IS NULL
+                    THEN NULL
+                WHEN daily_first_wallet_balance != prev_daily_last_wallet_balance
+                    THEN prev_daily_last_wallet_balance
+                ELSE daily_first_wallet_balance
+            END, 0)     AS original_daily_first_wallet_balance,
+        
+        COALESCE(daily_last_wallet_balance, 0)                   AS original_daily_last_wallet_balance,
+
+        COALESCE(new_daily_0_origin_first_wallet_balance, 0)     AS daily_0_origin_first_wallet_balance,
+        COALESCE(new_daily_0_origin_last_wallet_balance, 0)      AS daily_0_origin_last_wallet_balance,
+        
+        COALESCE(new_daily_calculated_first_wallet_balance, 0)   AS daily_calculated_first_wallet_balance,
+        COALESCE(new_daily_calculated_last_wallet_balance, 0)    AS daily_calculated_last_wallet_balance,
+        
+        prev_transaction_date,
+        transaction_date,
+        next_transaction_date,
+        first_daily_transaction_start_datetime,
+        first_daily_transaction_completed_datetime,
+        last_daily_transaction_start_datetime,
+        last_daily_transaction_completed_datetime,
+        
+        COALESCE(wallet_registration_datetime, new_wallet_registration_datetime)            AS wallet_registration_datetime,
+        
+        '{{ modules.datetime.datetime.now(modules.pytz.timezone("Asia/Kuala_Lumpur")) }}'   AS _dbt_ran_datetime
+
+    FROM recursive_imputation
+    
+),
+
+
+-- FINAL
+final__fct_exchange__daily_wallet_transactions AS (
+
+    SELECT
+        --ids
+        wallet_id,
+        company_id,
+        currency_id,
+        first_wallet_log_id,
+        last_wallet_log_id,
+        
+        --dimensions
+        is_earliest_wallet_balance,
+        is_latest_wallet_balance,
+
+        --measures
+        count_daily_wallet_activity,
+        count_total_daily_transaction_credit,
+        count_total_daily_transaction_debit,
+        total_daily_transaction_credit,
+        total_daily_transaction_debit,
+        original_daily_first_wallet_balance,
+        original_daily_last_wallet_balance,
+        daily_0_origin_first_wallet_balance,
+        daily_0_origin_last_wallet_balance,
+        daily_calculated_first_wallet_balance,
+        daily_calculated_last_wallet_balance,
+        
+        --date/times
+        transaction_date,
+        first_daily_transaction_start_datetime,
+        first_daily_transaction_completed_datetime,
+        last_daily_transaction_start_datetime,
+        last_daily_transaction_completed_datetime,
+        wallet_registration_datetime,
+
+        --metadata
+        _dbt_ran_datetime
+
+    FROM casting_and_organize
+
+)
+
+SELECT * FROM final__fct_exchange__daily_wallet_transactions

models/staging/exchange/base/base_exchange__wallet_logs.sql

@@ -0,0 +1,58 @@
+-- IMPORT
+WITH source AS (
+    SELECT * FROM {{ source('src_exchange_mysql', 'wallet_logs') }}
+),
+
+
+-- LOGIC
+renamed AS (
+
+    SELECT
+        id              AS wallet_log_id,
+        wallet_id,
+        owner_type,
+        owner_id,
+        code,
+        currency_id,
+        amount,
+        deleted_at      AS deleted_datetime,
+        created_at      AS created_datetime,
+        updated_at      AS updated_datetime,
+
+        '{{ modules.datetime.datetime.now(modules.pytz.timezone("Asia/Kuala_Lumpur")) }}' AS _dbt_ran_datetime
+
+    FROM source
+
+),
+
+
+-- FINAL
+final__base_exchange__wallet_logs AS (
+
+    SELECT
+        -- ids
+        wallet_log_id,
+        wallet_id,
+        owner_id,
+        currency_id,
+
+        -- dimensions
+        code,
+        owner_type,
+
+        -- measures
+        amount,
+
+        -- date/time
+        created_datetime,
+        updated_datetime,
+        deleted_datetime,
+
+        -- metadata
+        _dbt_ran_datetime
+
+    FROM renamed
+    
+)
+
+SELECT * FROM final__base_exchange__wallet_logs

models/staging/exchange/base/base_exchange__wallets.sql

@@ -0,0 +1,56 @@
+-- IMPORT
+WITH source AS (
+    SELECT * FROM {{ source('src_exchange_mysql', 'wallets') }}
+),
+
+
+-- LOGIC
+renamed AS (
+
+    SELECT
+        id              AS wallet_id,
+        owner_type,
+        owner_id,
+        code,
+        currency_id,
+        amount,
+        deleted_at      AS deleted_datetime,
+        created_at      AS created_datetime,
+        updated_at      AS updated_datetime,
+
+        '{{ modules.datetime.datetime.now(modules.pytz.timezone("Asia/Kuala_Lumpur")) }}' AS _dbt_ran_datetime
+
+    FROM source
+
+),
+
+
+-- FINAL
+final__base_exchange__wallets AS (
+
+    SELECT
+        -- ids
+        wallet_id,
+        owner_id,
+        currency_id,
+
+        -- dimensions
+        code,
+        owner_type,
+
+        -- measures
+        amount,
+
+        -- date/time
+        created_datetime,
+        updated_datetime,
+        deleted_datetime,
+
+        -- metadata
+        _dbt_ran_datetime
+
+    FROM renamed
+    
+)
+
+SELECT * FROM final__base_exchange__wallets

models/staging/exchange/stg_exchange__transaction_wallets.sql

@@ -22,13 +22,13 @@ transactions_join_transaction_types_payment_methods_currencies_status AS (
 
     SELECT
         transactions.transaction_id AS transaction_wallet_id,
-        IFF(transactions.owner_type = 'App\\Models\\Wallet', transactions.owner_id, null)  AS wallet_id,
+        IFF(transactions.owner_type = 'App\\Models\\Wallet', transactions.owner_id, null)   AS wallet_id,
 
-        COALESCE(transaction_types.name, transactions.transaction_type::string)   AS transaction_type,
+        COALESCE(transaction_types.name, transactions.transaction_type::string)             AS transaction_type,
 
-        transactions.receiver_user_id           AS company_id,
+        transactions.receiver_user_id                                           AS company_id,
 
-        COALESCE(payment_methods.name, transactions.payment_method::string)   AS payment_method,
+        COALESCE(payment_methods.name, transactions.payment_method::string)     AS payment_method,
 
         transactions.payment_reference,
         transactions.bill_number,
@@ -50,6 +50,7 @@ transactions_join_transaction_types_payment_methods_currencies_status AS (
         transactions.deleted_datetime,
         transactions.created_datetime,
         transactions.updated_datetime,
+
         '{{ modules.datetime.datetime.now(modules.pytz.timezone("Asia/Kuala_Lumpur")) }}' AS _dbt_ran_datetime
 
     FROM transactions
@@ -63,9 +64,9 @@ transactions_join_transaction_types_payment_methods_currencies_status AS (
     LEFT JOIN status
         ON (transactions.status = status.id)
 
-    
     WHERE 
         transactions.owner_type = 'App\\Models\\Wallet' 
+
 ),
 
 
@@ -105,6 +106,6 @@ final__stg_exchange__transaction_wallets AS (
 
     FROM transactions_join_transaction_types_payment_methods_currencies_status
 
-) 
+)
 
 SELECT * FROM final__stg_exchange__transaction_wallets

models/staging/exchange/stg_exchange__wallet_balance_logs.sql

@@ -0,0 +1,58 @@
+-- IMPORT
+WITH wallet_logs AS (
+    SELECT * FROM {{ ref('base_exchange__wallet_logs') }}
+),
+
+
+-- LOGIC
+renamed AS (
+
+    SELECT
+        wallet_log_id,
+        wallet_id,
+        owner_id        AS company_id,
+        currency_id,
+        code,
+        amount          AS wallet_balance,
+        created_datetime,
+        updated_datetime,
+        deleted_datetime,
+
+        '{{ modules.datetime.datetime.now(modules.pytz.timezone("Asia/Kuala_Lumpur")) }}' AS _dbt_ran_datetime
+
+    FROM wallet_logs
+
+    WHERE owner_type = 'App\\Models\\Company'
+
+),
+
+
+-- FINAL
+final__base_exchange__wallet_logs AS (
+
+    SELECT
+        --ids
+        wallet_log_id,
+        wallet_id,
+        company_id,
+        currency_id,
+
+        --dimensions
+        code,
+        
+        --measures
+        wallet_balance,
+
+        --date/times
+        created_datetime,
+        updated_datetime,
+        deleted_datetime,
+
+        --metadata
+        _dbt_ran_datetime
+
+    FROM renamed
+
+)
+
+SELECT * FROM final__base_exchange__wallet_logs

models/staging/exchange/stg_exchange__wallet_balances.sql

@@ -0,0 +1,56 @@
+-- IMPORT
+WITH wallets AS (
+    SELECT * FROM {{ ref('base_exchange__wallets') }}
+),
+
+
+-- LOGIC
+renamed AS (
+
+    SELECT
+        wallet_id,
+        owner_id        AS company_id,
+        currency_id,
+        code,
+        amount          AS wallet_balance,
+        created_datetime,
+        updated_datetime,
+        deleted_datetime,
+
+        '{{ modules.datetime.datetime.now(modules.pytz.timezone("Asia/Kuala_Lumpur")) }}' AS _dbt_ran_datetime
+
+    FROM wallets
+
+    WHERE owner_type = 'App\\Models\\Company'
+
+),
+
+
+-- FINAL
+final__stg__exchange__wallets AS (
+
+    SELECT
+        --ids
+        wallet_id,
+        company_id,
+        currency_id,
+
+        --dimensions
+        code,
+        
+        --measures
+        wallet_balance,
+
+        --date/times
+        created_datetime,
+        updated_datetime,
+        deleted_datetime,
+
+        --metadata
+        _dbt_ran_datetime
+
+    FROM renamed
+
+)
+
+SELECT * FROM final__stg__exchange__wallets