🔧 dbt Integration with Microsoft Fabric¶

-- models/staging/casino/stg_slot_telemetry.sql
{{ config(materialized='view') }}

with source as (
    select * from {{ source('bronze_casino', 'bronze_slot_telemetry') }}
),

renamed as (
    select
        cast(machine_id as varchar(20))       as machine_id,
        cast(event_type as varchar(20))       as event_type,
        cast(wager_amount as decimal(12, 2))  as wager_amount,
        cast(payout_amount as decimal(12, 2)) as payout_amount,
        cast(event_timestamp as datetime2)    as event_at,
        cast(floor_location as varchar(50))   as floor_location,
        cast(denomination as decimal(6, 2))   as denomination,
        cast(game_title as varchar(100))      as game_title,
        cast(session_id as varchar(40))       as session_id,
        -- Derived columns
        wager_amount - payout_amount          as hold_amount,
        cast(ingestion_timestamp as datetime2) as _loaded_at
    from source
    where machine_id is not null
      and event_timestamp is not null
)

select * from renamed

-- models/intermediate/casino/int_slot_metrics.sql
{{ config(materialized='table') }}

with telemetry as (
    select * from {{ ref('stg_slot_telemetry') }}
    where event_type = 'spin'
),

daily_metrics as (
    select
        machine_id,
        floor_location,
        denomination,
        game_title,
        cast(event_at as date)              as gaming_date,
        count(*)                             as spin_count,
        sum(wager_amount)                    as total_coin_in,
        sum(payout_amount)                   as total_coin_out,
        sum(hold_amount)                     as total_hold,
        case
            when sum(wager_amount) > 0
            then round(sum(hold_amount) / sum(wager_amount) * 100, 2)
            else 0
        end                                  as hold_pct,
        min(event_at)                        as first_spin_at,
        max(event_at)                        as last_spin_at,
        datediff(minute, min(event_at), max(event_at)) as active_minutes
    from telemetry
    group by
        machine_id, floor_location, denomination,
        game_title, cast(event_at as date)
),

validated as (
    select
        *,
        case
            when hold_pct < 0 or hold_pct > 25 then 'ANOMALY'
            when hold_pct < 2 then 'LOW_HOLD'
            when hold_pct > 15 then 'HIGH_HOLD'
            else 'NORMAL'
        end as hold_status
    from daily_metrics
    where spin_count > 0
)

select * from validated

-- models/marts/casino/fct_slot_performance.sql
{{ config(
    materialized='table',
    description='Daily slot machine performance fact table for Power BI Direct Lake consumption'
) }}

with metrics as (
    select * from {{ ref('int_slot_metrics') }}
),

machines as (
    select * from {{ ref('dim_machine') }}
),

dates as (
    select * from {{ ref('dim_date') }}
)

select
    -- Keys
    {{ dbt_utils.generate_surrogate_key(['m.machine_id', 'm.gaming_date']) }} as performance_key,
    d.date_key,
    mc.machine_key,
    -- Measures
    m.spin_count,
    m.total_coin_in,
    m.total_coin_out,
    m.total_hold,
    m.hold_pct,
    m.active_minutes,
    m.hold_status,
    -- Calculated measures
    case
        when m.active_minutes > 0
        then round(m.total_coin_in / (m.active_minutes / 60.0), 2)
        else 0
    end as coin_in_per_hour,
    m.first_spin_at,
    m.last_spin_at,
    current_timestamp as _dbt_loaded_at
from metrics m
left join machines mc on m.machine_id = mc.machine_id
left join dates d on m.gaming_date = d.calendar_date

-- models/intermediate/casino/int_player_sessions.sql
{{ config(
    materialized='incremental',
    unique_key='session_id',
    incremental_strategy='merge',
    on_schema_change='sync_all_columns'
) }}

with sessions as (
    select * from {{ ref('stg_player_sessions') }}
    {% if is_incremental() %}
    where _loaded_at > (select max(_loaded_at) from {{ this }})
    {% endif %}
),

enriched as (
    select
        session_id,
        player_id,
        machine_id,
        card_in_at,
        card_out_at,
        datediff(minute, card_in_at, card_out_at) as session_duration_minutes,
        total_wagered,
        total_won,
        total_wagered - total_won as net_loss,
        _loaded_at
    from sessions
    where card_in_at is not null
      and card_out_at is not null
      and card_out_at > card_in_at
)

select * from enriched

-- tests/assert_ctr_threshold.sql
-- Validate that all CTR compliance events are for transactions >= $10,000
-- Failure indicates a data quality issue in the compliance pipeline

select
    compliance_event_id,
    player_id,
    transaction_amount,
    filing_type
from {{ ref('fct_compliance_events') }}
where filing_type = 'CTR'
  and transaction_amount < 10000

-- tests/assert_sar_structuring.sql
-- Validate SAR filings: structuring pattern = 3+ transactions of $8K-$9.9K in 24h

with sar_filings as (
    select
        player_id,
        filing_date,
        pattern_transaction_count,
        pattern_total_amount
    from {{ ref('fct_compliance_events') }}
    where filing_type = 'SAR'
      and pattern_type = 'STRUCTURING'
)

select *
from sar_filings
where pattern_transaction_count < 3
   or pattern_total_amount < 24000  -- minimum 3 x $8,000

-- tests/assert_crop_yield_positive.sql
-- USDA crop yield values must be positive where production is reported

select
    state_name,
    commodity_name,
    crop_year,
    yield_per_acre
from {{ ref('fct_usda_crop_rankings') }}
where production_value > 0
  and (yield_per_acre is null or yield_per_acre <= 0)

-- models/staging/casino/stg_slot_telemetry.sql
{{ config(materialized='view') }}

select
    machine_id,
    event_type,
    cast(wager_amount as decimal(12, 2)) as wager_amount,
    cast(payout_amount as decimal(12, 2)) as payout_amount,
    wager_amount - payout_amount as hold_amount,
    event_timestamp as event_at,
    floor_location,
    denomination,
    game_title,
    session_id,
    ingestion_timestamp as _loaded_at
from {{ source('bronze_casino', 'bronze_slot_telemetry') }}
where machine_id is not null

-- models/intermediate/casino/int_slot_metrics.sql (see Model Patterns section above)

-- models/marts/casino/fct_slot_performance.sql (see Model Patterns section above)

-- models/intermediate/casino/int_compliance_ctr.sql
{{ config(materialized='table') }}

with transactions as (
    select * from {{ ref('stg_player_transactions') }}
),

ctr_candidates as (
    select
        player_id,
        transaction_id,
        transaction_amount,
        transaction_at,
        transaction_type,
        -- CTR threshold: $10,000
        case
            when transaction_amount >= {{ var('ctr_threshold', 10000) }}
            then true
            else false
        end as is_ctr_reportable
    from transactions
    where transaction_amount >= {{ var('ctr_threshold', 10000) }}
)

select
    transaction_id,
    player_id,
    transaction_amount,
    transaction_at,
    transaction_type,
    'CTR' as filing_type,
    'PENDING' as filing_status,
    current_timestamp as flagged_at
from ctr_candidates
where is_ctr_reportable = true

-- models/intermediate/casino/int_compliance_sar.sql
{{ config(materialized='table') }}

with transactions as (
    select * from {{ ref('stg_player_transactions') }}
    where transaction_amount between {{ var('sar_lower', 8000) }}
                                 and {{ var('sar_upper', 9999.99) }}
),

structuring_patterns as (
    select
        player_id,
        cast(transaction_at as date) as txn_date,
        count(*) as txn_count,
        sum(transaction_amount) as total_amount,
        min(transaction_at) as first_txn_at,
        max(transaction_at) as last_txn_at
    from transactions
    group by player_id, cast(transaction_at as date)
    having count(*) >= 3  -- 3+ transactions in a single day
)

select
    {{ dbt_utils.generate_surrogate_key(['player_id', 'txn_date']) }} as sar_pattern_id,
    player_id,
    txn_date as pattern_date,
    txn_count as pattern_transaction_count,
    total_amount as pattern_total_amount,
    'SAR' as filing_type,
    'STRUCTURING' as pattern_type,
    'PENDING' as filing_status,
    current_timestamp as flagged_at
from structuring_patterns

-- models/staging/federal/stg_usda_crop_production.sql
{{ config(materialized='view') }}

with source as (
    select * from {{ source('bronze_federal', 'bronze_usda_crop_production') }}
),

renamed as (
    select
        cast(state_fips_code as varchar(5))   as state_fips,
        cast(state_name as varchar(50))        as state_name,
        cast(commodity_desc as varchar(100))   as commodity_name,
        cast(statisticcat_desc as varchar(50)) as statistic_category,
        cast(year as int)                      as crop_year,
        cast(value as decimal(18, 2))          as value,
        cast(unit_desc as varchar(30))         as unit,
        cast(source_desc as varchar(50))       as data_source,
        cast(ingestion_timestamp as datetime2) as _loaded_at
    from source
    where value is not null
      and state_fips_code is not null
)

select * from renamed

-- models/intermediate/federal/int_usda_crop_analysis.sql
{{ config(materialized='table') }}

with production as (
    select * from {{ ref('stg_usda_crop_production') }}
    where statistic_category = 'PRODUCTION'
),

acreage as (
    select * from {{ ref('stg_usda_crop_production') }}
    where statistic_category = 'AREA HARVESTED'
),

combined as (
    select
        p.state_fips,
        p.state_name,
        p.commodity_name,
        p.crop_year,
        p.value as production_value,
        p.unit as production_unit,
        a.value as acreage_harvested,
        case
            when a.value > 0
            then round(p.value / a.value, 2)
            else null
        end as yield_per_acre,
        p._loaded_at
    from production p
    left join acreage a
        on p.state_fips = a.state_fips
        and p.commodity_name = a.commodity_name
        and p.crop_year = a.crop_year
)

select * from combined

-- models/intermediate/federal/int_epa_release_trends.sql
{{ config(materialized='table') }}

with releases as (
    select * from {{ ref('stg_epa_tri_releases') }}
),

annual_trends as (
    select
        facility_name,
        facility_state,
        chemical_name,
        release_medium,
        reporting_year,
        sum(release_amount_lbs) as total_release_lbs,
        count(distinct facility_id) as facility_count,
        -- Year-over-year change
        lag(sum(release_amount_lbs)) over (
            partition by facility_state, chemical_name, release_medium
            order by reporting_year
        ) as prior_year_release_lbs
    from releases
    group by
        facility_name, facility_state, chemical_name,
        release_medium, reporting_year
)

select
    *,
    case
        when prior_year_release_lbs > 0
        then round(
            (total_release_lbs - prior_year_release_lbs) / prior_year_release_lbs * 100, 2
        )
        else null
    end as yoy_change_pct
from annual_trends

-- models/marts/federal/dim_state.sql
{{ config(materialized='table') }}

with state_data as (
    select * from {{ ref('seed_state_codes') }}
)

select
    {{ dbt_utils.generate_surrogate_key(['state_fips']) }} as state_key,
    state_fips,
    state_abbreviation,
    state_name,
    census_region,
    census_division,
    current_timestamp as _dbt_loaded_at
from state_data