dbt for Healthcare Data Engineering

{{ config(materialized='view', schema='staging', tags=['staging','claims','phi']) }}

with source as (
    select * from {{ source('claims_raw', 'claim_header') }}
),

staged as (
    select
        -- Identifiers — ISO-11179 naming
        clm_nbr::varchar(20)            as clm_nbr,
        mbr_id::varchar(20)             as mbr_id,
        billng_prvdr_npi::varchar(10)   as billng_prvdr_npi,
        rndrng_prvdr_npi::varchar(10)   as rndrng_prvdr_npi,
        payer_id::varchar(20)           as payer_id,

        -- Dates
        svc_dt_from::date               as svc_dt_from,
        svc_dt_to::date                 as svc_dt_to,
        paid_dt::date                   as paid_dt,
        submsn_dt::date                 as submsn_dt,

        -- Diagnosis codes — always VARCHAR(7), never INTEGER
        prim_diag_cd::varchar(7)        as prim_diag_cd,
        sec_diag_cd::varchar(7)         as sec_diag_cd,
        icd_vrsn::varchar(5)            as icd_vrsn,

        -- Financial measures
        bld_amt::decimal(12,2)          as bld_amt,
        alwd_amt::decimal(12,2)         as alwd_amt,
        paid_amt::decimal(12,2)         as paid_amt,
        ded_amt::decimal(10,2)          as ded_amt,
        copay_amt::decimal(10,2)        as copay_amt,
        coins_amt::decimal(10,2)        as coins_amt,

        -- Status codes
        clm_sts_cd::varchar(20)         as clm_sts_cd,
        clm_type_cd::varchar(20)        as clm_type_cd,

        -- PHI fields — tagged for masking policy
        -- meta: {phi: true}
        mbr_first_nm::varchar(100)      as mbr_first_nm,
        mbr_last_nm::varchar(100)       as mbr_last_nm,
        mbr_dob::date                   as mbr_dob,

        -- Metadata
        rec_creat_ts::timestamp         as rec_creat_ts,
        rec_updt_ts::timestamp          as rec_updt_ts

    from source
    where clm_nbr is not null
)

select * from staged

version: 2
models:
  - name: stg_claims__header
    description: "Standardized claims header — ISO-11179 naming, typed columns, PHI tagged"
    columns:
      - name: clm_nbr
        description: "Unique claim identifier"
        tests: [not_null, unique]
      - name: billng_prvdr_npi
        description: "Billing provider NPI"
        tests:
          - not_null
          - dbt_utils.expression_is_true:
              expression: "length(billng_prvdr_npi) = 10"
      - name: prim_diag_cd
        description: "Primary ICD-10-CM diagnosis code"
        meta:
          phi: true
        tests:
          - dbt_utils.expression_is_true:
              expression: "length(prim_diag_cd) between 3 and 7"
      - name: icd_vrsn
        tests:
          - accepted_values:
              values: ['ICD9', 'ICD10']
      - name: paid_amt
        tests:
          - dbt_utils.expression_is_true:
              expression: "paid_amt >= 0"
      - name: mbr_first_nm
        meta:
          phi: true
          phi_type: direct
      - name: mbr_last_nm
        meta:
          phi: true
          phi_type: direct
      - name: mbr_dob
        meta:
          phi: true
          phi_type: direct

{{ config(
    materialized='table',
    schema='quality',
    tags=['quality', 'hedis', 'cbp']
) }}

-- HEDIS Controlling High Blood Pressure (CBP)
-- Denominator: Members 18-85 with hypertension diagnosis
-- Numerator: Most recent BP reading < 140/90

with measurement_period as (
    select
        '{{ var("hedis_start_date") }}'::date   as start_dt,
        '{{ var("hedis_end_date") }}'::date     as end_dt,
        {{ var("measurement_year") }}            as msrmt_yr
),

-- Denominator: active members 18-85 with HTN
denominator as (
    select distinct
        m.mbr_id,
        m.mbr_key,
        m.plan_cd,
        m.mbr_dob,
        datediff('year', m.mbr_dob, p.end_dt) as age_at_end
    from {{ ref('dim_member') }} m
    cross join measurement_period p
    -- Age 18-85 at end of measurement year
    where datediff('year', m.mbr_dob, p.end_dt) between 18 and 85
    -- Active enrollment during measurement year
      and m.eff_dt <= p.end_dt
      and coalesce(m.exp_dt, '9999-12-31') >= p.start_dt
      and m.curr_rec_flg = true
    -- Hypertension diagnosis in measurement or prior year
      and exists (
          select 1
          from {{ ref('fact_claim_header') }} c
          join {{ ref('dim_icd10') }} d
              on d.icd10_key = c.prim_diag_key
          where c.mbr_key = m.mbr_key
            and d.icd10_cd like 'I1%'    -- hypertension codes
            and c.svc_dt_from between
                dateadd('year', -1, p.start_dt) and p.end_dt
      )
),

-- Numerator: most recent BP < 140/90
numerator as (
    select
        d.mbr_id,
        max(r.result_dt)            as last_bp_dt,
        max(r.systlc_bp_val)
            keep (dense_rank last
                  order by r.result_dt) as last_systolic,
        max(r.diastlc_bp_val)
            keep (dense_rank last
                  order by r.result_dt) as last_diastolic
    from denominator d
    join {{ ref('fct_lab_result') }} r
        on r.mbr_id = d.mbr_id
    cross join measurement_period p
    where r.result_dt between p.start_dt and p.end_dt
      and r.loinc_cd in ('8480-6', '8462-4')   -- BP LOINC codes
    group by d.mbr_id
),

final as (
    select
        p.msrmt_yr,
        d.mbr_id,
        d.mbr_key,
        d.plan_cd,
        d.age_at_end,
        true                                    as in_denominator,
        n.last_bp_dt,
        n.last_systolic,
        n.last_diastolic,
        case
            when n.last_systolic < 140
             and n.last_diastolic < 90
            then true
            else false
        end                                     as in_numerator,
        case
            when n.mbr_id is null then 'NO_BP_READING'
            when n.last_systolic >= 140 then 'HIGH_SYSTOLIC'
            when n.last_diastolic >= 90 then 'HIGH_DIASTOLIC'
            else 'CONTROLLED'
        end                                     as bp_status_desc,
        current_timestamp                       as rec_creat_ts
    from denominator d
    cross join measurement_period p
    left join numerator n on n.mbr_id = d.mbr_id
)

select * from final

{% test npi_format(model, column_name) %}
-- Validates NPI is exactly 10 numeric digits
-- Usage: - dbt_utils.expression_is_true OR this custom test
select count(*) as failures
from {{ model }}
where {{ column_name }} is not null
  and (
    length({{ column_name }}) != 10
    or {{ column_name }} ~ '[^0-9]'
  )
{% endtest %}

{% macro apply_phi_masking(table_ref, phi_columns) %}
-- Apply Snowflake dynamic data masking to PHI columns
-- Call in post-hook after model creation
{% for col in phi_columns %}
    alter table {{ table_ref }}
    modify column {{ col }}
    set masking policy phi_name_mask;
{% endfor %}
{% endmacro %}

-- Usage in model config:
-- +post-hook: "{{ apply_phi_masking(this, ['mbr_first_nm','mbr_last_nm']) }}"

{% macro hedis_measurement_period(year_offset=0) %}
-- Returns standard HEDIS measurement period dates
{% set yr = var('measurement_year') + year_offset %}
{
    'start_dt': "'" ~ yr ~ "-01-01'::date",
    'end_dt':   "'" ~ yr ~ "-12-31'::date"
}
{% endmacro %}