CREATE MATERIALIZED VIEW insights.flow_values_mv
WITH (
backfill_order=FIXED(olap.transaction_types_dm -> olap.transactions_dm, asset_service.assets_dm -> olap.transactions_dm, asset_service.foreign_exchange_rates_eod_ft -> olap.transactions_dm, olap.fee_transactions_dm -> olap.transactions_dm, olap.income_transactions_dm -> olap.transactions_dm)
) AS
WITH mapped_sources AS (
SELECT
transaction_id,
source_asset_id
FROM olap.fee_transactions_dm
WHERE
NOT source_account_id IS NULL AND NOT source_asset_id IS NULL
UNION ALL
SELECT
transaction_id,
source_asset_id
FROM olap.income_transactions_dm
WHERE
NOT source_account_id IS NULL AND NOT source_asset_id IS NULL
), legs AS (
SELECT
t.transaction_id,
t.account_id,
t.asset_id,
t.asset_id AS classification_asset_id,
t.transaction_valuation_date,
t.currency_code,
t.net_value,
tt.type AS transaction_type
FROM olap.transactions_dm AS t
JOIN olap.transaction_types_dm AS tt
ON tt.transaction_type_id = t.transaction_type_id
WHERE
tt.type IN ('TRADE', 'TRANSFER')
UNION ALL
SELECT
t.transaction_id,
t.account_id,
t.asset_id,
t.asset_id AS classification_asset_id,
t.transaction_valuation_date,
t.currency_code,
t.net_value,
tt.type AS transaction_type
FROM olap.transactions_dm AS t
JOIN olap.transaction_types_dm AS tt
ON tt.transaction_type_id = t.transaction_type_id AND tt.type IN ('INCOME', 'FEE')
JOIN mapped_sources AS ms
ON ms.transaction_id = t.transaction_id
UNION ALL
SELECT
t.transaction_id,
t.account_id AS account_id,
ms.source_asset_id AS asset_id,
t.asset_id AS classification_asset_id,
t.transaction_valuation_date,
t.currency_code,
-t.net_value AS net_value,
tt.type AS transaction_type
FROM olap.transactions_dm AS t
JOIN olap.transaction_types_dm AS tt
ON tt.transaction_type_id = t.transaction_type_id AND tt.type IN ('INCOME', 'FEE')
JOIN mapped_sources AS ms
ON ms.transaction_id = t.transaction_id
), legs_enriched AS (
SELECT
l.transaction_id,
l.account_id,
l.asset_id,
l.transaction_valuation_date,
l.currency_code,
l.net_value,
l.transaction_type,
COALESCE(a.type, 'UNSPECIFIED') AS asset_type,
a.issue_currency_code AS asset_currency,
hv.type AS position_type,
CAST(DATE_TRUNC('MONTH', l.transaction_valuation_date) AS DATE) AS dim_transaction_month
FROM legs AS l
LEFT JOIN asset_service.assets_dm AS a
ON a.id = l.asset_id
LEFT JOIN olap.holding_values_raw_ft AS hv
ON hv.account_id = l.account_id
AND hv.asset_id = l.classification_asset_id
AND hv.dim_value_date = l.transaction_valuation_date
)
SELECT
l.transaction_id,
l.account_id,
l.asset_id,
atag.account_group_id,
l.transaction_valuation_date AS dim_transaction_date,
l.transaction_type,
l.currency_code AS transaction_currency,
l.position_type,
atag.base_currency AS group_currency,
l.asset_type,
l.asset_currency,
l.net_value,
l.net_value * COALESCE(fx_group.rate, CASE WHEN l.currency_code = atag.base_currency THEN 1 ELSE NULL END) AS net_value_group,
l.net_value * COALESCE(fx_system.rate, CASE WHEN l.currency_code = 'SAR' THEN 1 ELSE NULL END) AS net_value_system
FROM legs_enriched AS l
JOIN insights.account_to_account_groups_txn_binned_mv AS atag
ON atag.account_id = l.account_id
AND atag.dim_transaction_month = l.dim_transaction_month
AND l.transaction_valuation_date >= atag.effective_start_date
AND (
atag.effective_end_date IS NULL
OR l.transaction_valuation_date < atag.effective_end_date
)
AND (
atag.opening_date IS NULL OR l.transaction_valuation_date >= atag.opening_date
)
AND atag.source_entity_type <> 'user'
LEFT JOIN asset_service.foreign_exchange_rates_eod_ft AS fx_group
ON fx_group.source_currency_code = l.currency_code
AND fx_group.target_currency_code = atag.base_currency
AND fx_group.date = l.transaction_valuation_date
LEFT JOIN asset_service.foreign_exchange_rates_eod_ft AS fx_system
ON fx_system.source_currency_code = l.currency_code
AND fx_system.target_currency_code = 'SAR'
AND fx_system.date = l.transaction_valuation_date
WHERE
(
NOT fx_group.rate IS NULL OR l.currency_code = atag.base_currency
)
AND (
NOT fx_system.rate IS NULL OR l.currency_code = 'SAR'
)