Claims¶

The claims module provides lower-level access to claims data for custom analysis beyond what the cohort module offers.

When to Use¶

Use the claims module when you need to:

Fetch raw claims data with custom filters
Build custom SQL queries with configuration-driven column names
Get index dates from diagnosis claims
Add demographics to an existing patient list
Work with claims data outside the cohort workflow

Use Cohort Module First

For most use cases, start with get_cohort() which handles claims, demographics, and enrollment in one call. Use this module for custom analysis pipelines.

Quick Example¶

from alx_heor.database import RedshiftConnection
from alx_heor.claims import get_claims, get_index_dates, add_demographics

conn = RedshiftConnection().connect()

# Get all gMG diagnosis claims
df_claims = get_claims(
    conn,
    source="iqvia",
    schema="iqvia_pharmetrics_2024q3",
    diagnosis_codes=["G700", "G7000", "G7001"],
    start_year=2015,
    end_year=2024,
)

# Get first diagnosis date per patient
df_index = get_index_dates(
    df_claims,
    source="iqvia",
    min_count=2,
    days_apart=30,
)

# Add demographics
df_with_demo = add_demographics(
    conn,
    df_index,
    source="iqvia",
    schema="iqvia_pharmetrics_2024q3",
)

Common Patterns¶

Custom Column Selection¶

df = get_claims(
    conn,
    source="iqvia",
    schema="iqvia_pharmetrics_2024q3",
    diagnosis_codes=["G700"],
    columns=["pat_id", "from_dt", "diag1", "pos", "prov_spec"],
    start_year=2020,
    end_year=2024,
)

Building SQL Manually¶

from alx_heor.claims import build_claims_sql

sql = build_claims_sql(
    source="iqvia",
    schema="iqvia_pharmetrics_2024q3",
    diagnosis_codes=["G700", "G7000"],
    columns=["pat_id", "from_dt"],
    start_year=2020,
    end_year=2024,
)

# Modify SQL as needed
sql += " AND pos IN ('21', '22')"

df = conn.query(sql)

cohort - High-level cohort building (uses claims internally)
config - Column name mappings
database - Database connection

claims ¶

Claims data processing utilities for Real-World Evidence (RWE) studies.

This module provides the foundation for cohort identification in retrospective healthcare database studies. Claims data represents healthcare encounters (diagnoses, procedures, prescriptions) captured during routine clinical care, making it essential for identifying patients with specific conditions.

Core Functions:

get_claims: Retrieve claims filtered by diagnosis codes (ICD-9/ICD-10)
build_claims_sql: Preview the SQL query before execution
get_index_dates: Identify index dates (study entry) for a patient cohort
add_demographics: Add patient age and sex from claims or enrollment data
filter_demographics: Apply age/sex inclusion criteria

Typical Workflow:

get_claims() - Query claims matching your target diagnosis codes
get_index_dates() - Determine each patient's index date (first/second Dx)
add_demographics() - Add age at index and sex
filter_demographics() - Apply age/sex inclusion criteria
Continue to enrollment module for continuous enrollment requirements

For a fully automated workflow, use cohort.get_cohort() which combines all these steps with comprehensive attrition tracking.

Example

Basic workflow for identifying gMG (generalized Myasthenia Gravis) patients:

from alx_heor import RedshiftConnection from alx_heor.claims import get_claims, get_index_dates, add_demographics

conn = RedshiftConnection().connect() df_claims = get_claims( ... conn, ... source='iqvia', ... schema='iqvia_pharmetrics_2024q3', ... diagnosis_codes=['G700', 'G7000', 'G7001'], # gMG ICD-10 codes ... start_year=2015, ... end_year=2024, ... ) print(f"Total claims: {len(df_claims):,}") Total claims: 1,234,567 df_index = get_index_dates( ... df_claims, ... source='iqvia', ... min_diagnosis_count=2, ... days_apart=30, ... ) print(f"Patients with 2+ Dx: {len(df_index):,}") Patients with 2+ Dx: 45,678 df_cohort = add_demographics(df_index, df_claims=df_claims, source='iqvia') df_adults = df_cohort[df_cohort['age_at_index'] >= 18] print(f"Adult patients: {len(df_adults):,}") Adult patients: 42,103

See Also

cohort.get_cohort : High-level function that automates this entire workflow enrollment.get_enrollment : Next step after claims-based identification config.get_source_config : View column mappings for each data source

Notes

ICD-9 codes (e.g., '3410') were used before Oct 2015; ICD-10 (e.g., 'G700') after
Always include BOTH ICD-9 and ICD-10 codes for studies spanning the transition
The 2+ diagnoses 30 days apart criterion reduces false positives from rule-out testing
IQVIA uses pat_id, Optum uses patid - the source parameter handles this

get_claims ¶

get_claims(conn: RedshiftConnection, source: str, schema: str, diagnosis_codes: list[str], start_year: int, end_year: int, table_pattern: str | None = None, columns: list[str] | None = None, diagnosis_columns: list[str] | None = None, include_demographics: bool = False) -> pd.DataFrame

Retrieve claims data filtered by diagnosis codes from yearly claims tables.

This is typically the first step in cohort identification. Claims are stored in yearly tables (e.g., claims_2020, claims_2021) and this function builds a UNION ALL query across all specified years, filtering to records where any diagnosis column contains one of the target ICD codes.

Workflow:

get_claims() <-- you are here
get_index_dates() - Identify index dates from the claims
add_demographics() - Add age/sex to the cohort
filter_demographics() - Apply inclusion criteria

Parameters:

Name	Type	Description	Default
`conn`	`RedshiftConnection`	Active database connection. Must be connected before calling.	required
`source`	`str`	Data source name: 'iqvia', 'optum', 'komodo'. Determines default column names and table patterns via config.	required
`schema`	`str`	Database schema name (e.g., 'iqvia_pharmetrics_2024q3'). Use `conn.get_schemas('iqvia')` to list available schemas.	required
`diagnosis_codes`	`list[str]`	ICD-9 or ICD-10 codes to filter claims. Include both versions for studies spanning Oct 2015 (ICD-10 transition date in US). Examples: ['G700', 'G7000', 'G7001'] for gMG, ['G35'] for MS.	required
`start_year`	`int`	First year of claims data to include (e.g., 2015).	required
`end_year`	`int`	Last year of claims data to include (e.g., 2024).	required
`table_pattern`	`str`	Override table name pattern (e.g., 'claims_{year}'). Uses source default if not specified. The `{year}` placeholder is required.	`None`
`columns`	`list[str]`	Override columns to select. Uses source default if not specified. Diagnosis columns are automatically added if not included.	`None`
`diagnosis_columns`	`list[str]`	Override which diagnosis columns to check (e.g., ['diag1', 'diag2']). Uses source default if not specified. IQVIA has 12 diagnosis columns plus diag_admit; Optum has 5.	`None`
`include_demographics`	`bool`	If True, include demographics (year_of_birth, sex) by querying the enrollment table. For IQVIA, demographics are in `enroll`, not claims. This adds columns needed by add_demographics().	`False`

Returns:

Type	Description
`DataFrame`	Claims data filtered by diagnosis codes. One row per claim, not per patient. Contains all columns from the SELECT clause.

See Also

build_claims_sql : Preview the generated SQL without executing it get_index_dates : Next step - identify index dates from claims cohort.get_cohort : High-level function that automates the full workflow config.get_source_config : View column mappings for each data source

Notes

An empty diagnosis_codes list will return ALL claims (expensive!)
Claims can be very large (millions of rows). Consider memory limits.
IQVIA claims tables: claims_2006 through claims_2025
The function checks ALL diagnosis columns (diag1-diag12, diag_admit for IQVIA)
Codes are matched exactly - 'G70' will NOT match 'G700'

Examples:

Identify patients with gMG (both ICD-9 and ICD-10):

>>> conn = RedshiftConnection().connect()
>>> df_claims = get_claims(
...     conn,
...     source='iqvia',
...     schema='iqvia_pharmetrics_2024q3',
...     diagnosis_codes=['3589', 'G700', 'G7000', 'G7001'],  # MG codes
...     start_year=2015,
...     end_year=2024,
... )
>>> print(f"Claims: {len(df_claims):,}, Patients: {df_claims['pat_id'].nunique():,}")
Claims: 2,456,789, Patients: 89,123

Preview the SQL before running (useful for debugging):

>>> sql = build_claims_sql(
...     source='iqvia',
...     schema='iqvia_pharmetrics_2024q3',
...     diagnosis_codes=['G700'],
...     start_year=2023, end_year=2024,
... )
>>> print(sql[:200])  # Preview first 200 chars

Source code in alx_heor\claims\__init__.py

def get_claims(
    conn: RedshiftConnection,
    source: str,
    schema: str,
    diagnosis_codes: list[str],
    start_year: int,
    end_year: int,
    table_pattern: str | None = None,
    columns: list[str] | None = None,
    diagnosis_columns: list[str] | None = None,
    include_demographics: bool = False,
) -> pd.DataFrame:
    """Retrieve claims data filtered by diagnosis codes from yearly claims tables.

    This is typically the first step in cohort identification. Claims are stored
    in yearly tables (e.g., claims_2020, claims_2021) and this function builds
    a UNION ALL query across all specified years, filtering to records where
    any diagnosis column contains one of the target ICD codes.

    **Workflow:**

    1. **get_claims()** <-- you are here
    2. get_index_dates() - Identify index dates from the claims
    3. add_demographics() - Add age/sex to the cohort
    4. filter_demographics() - Apply inclusion criteria

    Parameters
    ----------
    conn : RedshiftConnection
        Active database connection. Must be connected before calling.
    source : str
        Data source name: 'iqvia', 'optum', 'komodo'.
        Determines default column names and table patterns via config.
    schema : str
        Database schema name (e.g., 'iqvia_pharmetrics_2024q3').
        Use `conn.get_schemas('iqvia')` to list available schemas.
    diagnosis_codes : list[str]
        ICD-9 or ICD-10 codes to filter claims. Include both versions for
        studies spanning Oct 2015 (ICD-10 transition date in US).
        Examples: ['G700', 'G7000', 'G7001'] for gMG, ['G35'] for MS.
    start_year : int
        First year of claims data to include (e.g., 2015).
    end_year : int
        Last year of claims data to include (e.g., 2024).
    table_pattern : str, optional
        Override table name pattern (e.g., 'claims_{year}'). Uses source
        default if not specified. The `{year}` placeholder is required.
    columns : list[str], optional
        Override columns to select. Uses source default if not specified.
        Diagnosis columns are automatically added if not included.
    diagnosis_columns : list[str], optional
        Override which diagnosis columns to check (e.g., ['diag1', 'diag2']).
        Uses source default if not specified. IQVIA has 12 diagnosis columns
        plus diag_admit; Optum has 5.
    include_demographics : bool, default=False
        If True, include demographics (year_of_birth, sex) by querying
        the enrollment table. For IQVIA, demographics are in `enroll`,
        not claims. This adds columns needed by add_demographics().

    Returns
    -------
    pd.DataFrame
        Claims data filtered by diagnosis codes. One row per claim, not per
        patient. Contains all columns from the SELECT clause.

    See Also
    --------
    build_claims_sql : Preview the generated SQL without executing it
    get_index_dates : Next step - identify index dates from claims
    cohort.get_cohort : High-level function that automates the full workflow
    config.get_source_config : View column mappings for each data source

    Notes
    -----
    - An empty `diagnosis_codes` list will return ALL claims (expensive!)
    - Claims can be very large (millions of rows). Consider memory limits.
    - IQVIA claims tables: claims_2006 through claims_2025
    - The function checks ALL diagnosis columns (diag1-diag12, diag_admit for IQVIA)
    - Codes are matched exactly - 'G70' will NOT match 'G700'

    Examples
    --------
    Identify patients with gMG (both ICD-9 and ICD-10):

    >>> conn = RedshiftConnection().connect()
    >>> df_claims = get_claims(
    ...     conn,
    ...     source='iqvia',
    ...     schema='iqvia_pharmetrics_2024q3',
    ...     diagnosis_codes=['3589', 'G700', 'G7000', 'G7001'],  # MG codes
    ...     start_year=2015,
    ...     end_year=2024,
    ... )
    >>> print(f"Claims: {len(df_claims):,}, Patients: {df_claims['pat_id'].nunique():,}")
    Claims: 2,456,789, Patients: 89,123

    Preview the SQL before running (useful for debugging):

    >>> sql = build_claims_sql(
    ...     source='iqvia',
    ...     schema='iqvia_pharmetrics_2024q3',
    ...     diagnosis_codes=['G700'],
    ...     start_year=2023, end_year=2024,
    ... )
    >>> print(sql[:200])  # Preview first 200 chars
    """
    sql = build_claims_sql(
        source=source,
        schema=schema,
        diagnosis_codes=diagnosis_codes,
        start_year=start_year,
        end_year=end_year,
        table_pattern=table_pattern,
        columns=columns,
        diagnosis_columns=diagnosis_columns,
    )
    df = conn.query(sql)

    # Include demographics if requested
    if include_demographics:
        df = _add_demographics_to_claims(conn, schema, source, df)

    return df

build_claims_sql ¶

build_claims_sql(source: str, schema: str, diagnosis_codes: list[str], start_year: int, end_year: int, table_pattern: str | None = None, columns: list[str] | None = None, diagnosis_columns: list[str] | None = None) -> str

Build SQL query for claims data without executing it.

This function generates the UNION ALL SQL query that get_claims() would execute. Use this to inspect, debug, or modify the query before running it. Particularly useful for understanding the query structure or for running modified versions directly via conn.query().

Workflow:

build_claims_sql() - Preview SQL (optional, for debugging)
get_claims() - Execute the query
get_index_dates() - Identify index dates from claims

Parameters:

Name	Type	Description	Default
`source`	`str`	Data source name: 'iqvia', 'optum', 'komodo'.	required
`schema`	`str`	Database schema name.	required
`diagnosis_codes`	`list[str]`	ICD codes to filter claims.	required
`start_year`	`int`	Start year for claims data.	required
`end_year`	`int`	End year for claims data.	required
`table_pattern`	`str`	Override table name pattern.	`None`
`columns`	`list[str]`	Override columns to select.	`None`
`diagnosis_columns`	`list[str]`	Override diagnosis columns to check.	`None`

Returns:

Type	Description
`str`	SQL query string ready for execution.

See Also

get_claims : Execute the query and return results config.get_source_config : View column mappings and table patterns

Notes

The generated SQL uses UNION ALL across yearly tables (claims_2020, etc.)
Diagnosis filtering uses OR across all diagnosis columns
SELECT clause excludes SELECT * to avoid pulling unnecessary data

Examples:

Preview SQL for debugging:

>>> sql = build_claims_sql(
...     source='iqvia',
...     schema='iqvia_pharmetrics_2024q3',
...     diagnosis_codes=['G700', 'G7000'],
...     start_year=2023,
...     end_year=2024,
... )
>>> print(sql)
SELECT pat_id, from_dt, to_dt, ...
FROM (
    SELECT ... FROM iqvia_pharmetrics_2024q3.claims_2024
    UNION ALL
    SELECT ... FROM iqvia_pharmetrics_2024q3.claims_2023
) claims
WHERE diag1 IN ('G700', 'G7000') OR diag2 IN ('G700', 'G7000') ...

Modify and execute directly:

>>> sql = build_claims_sql(source='iqvia', ...)
>>> sql_with_limit = sql + " LIMIT 1000"  # Add limit for testing
>>> df_sample = conn.query(sql_with_limit)

Source code in alx_heor\claims\__init__.py

def build_claims_sql(
    source: str,
    schema: str,
    diagnosis_codes: list[str],
    start_year: int,
    end_year: int,
    table_pattern: str | None = None,
    columns: list[str] | None = None,
    diagnosis_columns: list[str] | None = None,
) -> str:
    """Build SQL query for claims data without executing it.

    This function generates the UNION ALL SQL query that `get_claims()` would
    execute. Use this to inspect, debug, or modify the query before running it.
    Particularly useful for understanding the query structure or for running
    modified versions directly via `conn.query()`.

    **Workflow:**

    1. **build_claims_sql()** - Preview SQL (optional, for debugging)
    2. get_claims() - Execute the query
    3. get_index_dates() - Identify index dates from claims

    Parameters
    ----------
    source : str
        Data source name: 'iqvia', 'optum', 'komodo'.
    schema : str
        Database schema name.
    diagnosis_codes : list[str]
        ICD codes to filter claims.
    start_year : int
        Start year for claims data.
    end_year : int
        End year for claims data.
    table_pattern : str, optional
        Override table name pattern.
    columns : list[str], optional
        Override columns to select.
    diagnosis_columns : list[str], optional
        Override diagnosis columns to check.

    Returns
    -------
    str
        SQL query string ready for execution.

    See Also
    --------
    get_claims : Execute the query and return results
    config.get_source_config : View column mappings and table patterns

    Notes
    -----
    - The generated SQL uses UNION ALL across yearly tables (claims_2020, etc.)
    - Diagnosis filtering uses OR across all diagnosis columns
    - SELECT clause excludes SELECT * to avoid pulling unnecessary data

    Examples
    --------
    Preview SQL for debugging:

    >>> sql = build_claims_sql(
    ...     source='iqvia',
    ...     schema='iqvia_pharmetrics_2024q3',
    ...     diagnosis_codes=['G700', 'G7000'],
    ...     start_year=2023,
    ...     end_year=2024,
    ... )
    >>> print(sql)
    SELECT pat_id, from_dt, to_dt, ...
    FROM (
        SELECT ... FROM iqvia_pharmetrics_2024q3.claims_2024
        UNION ALL
        SELECT ... FROM iqvia_pharmetrics_2024q3.claims_2023
    ) claims
    WHERE diag1 IN ('G700', 'G7000') OR diag2 IN ('G700', 'G7000') ...

    Modify and execute directly:

    >>> sql = build_claims_sql(source='iqvia', ...)
    >>> sql_with_limit = sql + " LIMIT 1000"  # Add limit for testing
    >>> df_sample = conn.query(sql_with_limit)
    """
    # Get source configuration
    config = get_source_config(source)

    # Use provided values or fall back to source defaults
    pattern = table_pattern or config["claims_table_pattern"]
    select_columns = columns or config["default_claims_columns"]
    diag_columns = diagnosis_columns or config["columns"]["diagnosis"]

    # Ensure all diagnosis columns are included in SELECT
    # (WHERE clause references them, so they must be selected)
    all_columns = list(select_columns)
    for col in diag_columns:
        if col not in all_columns:
            all_columns.append(col)

    # Build UNION ALL for yearly tables
    union_parts = []
    for year in range(start_year, end_year + 1):
        table_name = pattern.format(year=year)
        cols = ", ".join(all_columns)
        union_parts.append(f"SELECT {cols} FROM {schema}.{table_name}")

    union_sql = " UNION ALL ".join(union_parts)

    # Build WHERE clause for diagnosis codes
    quoted_codes = ", ".join(f"'{code}'" for code in diagnosis_codes)
    diag_conditions = " OR ".join(
        f"{col} IN ({quoted_codes})" for col in diag_columns
    )

    # Use specific columns in outer SELECT (not SELECT *)
    cols_outer = ", ".join(all_columns)

    sql = f"""
        SELECT {cols_outer}
        FROM ({union_sql}) claims
        WHERE {diag_conditions}
    """

    return sql.strip()

get_index_dates ¶

get_index_dates(df_claims: DataFrame, source: str | None = None, patient_id_col: str | None = None, date_col: str | None = None, min_diagnosis_count: int = 1, days_apart: int = 30) -> pd.DataFrame

Identify index dates for patients based on diagnosis claims.

The index date is the anchor point for a retrospective cohort study. It defines when each patient "enters" the study, and all baseline/follow-up periods are calculated relative to this date. In RWE studies, the index date is typically the first (or second) qualifying diagnosis date.

The "2+ diagnoses 30 days apart" criterion is a standard RWE approach to reduce false positives. A single diagnosis may represent rule-out testing (e.g., patient presents with symptoms, gets tested, but doesn't have the condition). Requiring two diagnoses separated by time increases confidence that the patient truly has the condition.

Workflow:

get_claims() - Query raw claims data
get_index_dates() <-- you are here
add_demographics() - Add age/sex to cohort
filter_demographics() - Apply inclusion criteria

Parameters:

Name	Type	Description	Default
`df_claims`	`DataFrame`	Claims data from get_claims(). Must contain patient ID and date columns.	required
`source`	`str`	Data source name: 'iqvia', 'optum', 'komodo'. If provided, uses source-specific column defaults.	`None`
`patient_id_col`	`str`	Name of the patient ID column. If not provided, uses source default or falls back to 'pat_id'.	`None`
`date_col`	`str`	Name of the service date column. If not provided, uses source default or falls back to 'from_dt'.	`None`
`min_diagnosis_count`	`int`	Minimum number of diagnoses required: - 1: Any single diagnosis qualifies (more inclusive, more false positives) - 2: Requires 2 diagnoses (standard RWE criterion, fewer false positives) - 3+: Very restrictive (use for conditions with high testing frequency)	`1`
`days_apart`	`int`	Minimum days between first and last diagnosis when min_diagnosis_count >= 2. Common values: 30 (standard), 60 (more restrictive), 7 (less restrictive).	`30`

Returns:

Type	Description
`DataFrame`	DataFrame with columns: - patient_id: Patient identifier (renamed from source-specific column) - index_date: First qualifying diagnosis date (datetime64)

See Also

get_claims : Previous step - retrieve raw claims add_demographics : Next step - add age/sex cohort.get_cohort : High-level function that handles all steps cohort.DiagnosisCriteria : More advanced diagnosis criteria options

Notes

The index date is always the FIRST diagnosis date, even with min_count=2
Dates are converted to datetime64 format automatically
Patients not meeting criteria are excluded from output
For second diagnosis as index date, use cohort.CohortCriteria(index_date_method="second_dx")
Consider your study's sensitivity/specificity tradeoff when choosing min_count

Examples:

Standard RWE approach: 2+ diagnoses 30 days apart for gMG:

>>> df_index = get_index_dates(
...     df_claims,
...     source='iqvia',
...     min_diagnosis_count=2,
...     days_apart=30,
... )
>>> print(f"Patients meeting 2+ Dx criterion: {len(df_index):,}")
Patients meeting 2+ Dx criterion: 12,847

Any single diagnosis (more inclusive, use for rare conditions):

>>> df_index_any = get_index_dates(
...     df_claims,
...     source='iqvia',
...     min_diagnosis_count=1,
... )
>>> print(f"Patients with any Dx: {len(df_index_any):,}")
Patients with any Dx: 25,123

Compare sensitivity vs specificity:

>>> # More specific (fewer false positives)
>>> df_strict = get_index_dates(df_claims, source='iqvia', min_diagnosis_count=2, days_apart=60)
>>> # More sensitive (fewer false negatives)
>>> df_lenient = get_index_dates(df_claims, source='iqvia', min_diagnosis_count=1)
>>> print(f"Strict: {len(df_strict):,}, Lenient: {len(df_lenient):,}")

Source code in alx_heor\claims\__init__.py

def get_index_dates(
    df_claims: pd.DataFrame,
    source: str | None = None,
    patient_id_col: str | None = None,
    date_col: str | None = None,
    min_diagnosis_count: int = 1,
    days_apart: int = 30,
) -> pd.DataFrame:
    """Identify index dates for patients based on diagnosis claims.

    The index date is the anchor point for a retrospective cohort study. It defines
    when each patient "enters" the study, and all baseline/follow-up periods are
    calculated relative to this date. In RWE studies, the index date is typically
    the first (or second) qualifying diagnosis date.

    The "2+ diagnoses 30 days apart" criterion is a standard RWE approach to reduce
    false positives. A single diagnosis may represent rule-out testing (e.g., patient
    presents with symptoms, gets tested, but doesn't have the condition). Requiring
    two diagnoses separated by time increases confidence that the patient truly has
    the condition.

    **Workflow:**

    1. get_claims() - Query raw claims data
    2. **get_index_dates()** <-- you are here
    3. add_demographics() - Add age/sex to cohort
    4. filter_demographics() - Apply inclusion criteria

    Parameters
    ----------
    df_claims : pd.DataFrame
        Claims data from get_claims(). Must contain patient ID and date columns.
    source : str, optional
        Data source name: 'iqvia', 'optum', 'komodo'.
        If provided, uses source-specific column defaults.
    patient_id_col : str, optional
        Name of the patient ID column. If not provided, uses source default
        or falls back to 'pat_id'.
    date_col : str, optional
        Name of the service date column. If not provided, uses source default
        or falls back to 'from_dt'.
    min_diagnosis_count : int, default=1
        Minimum number of diagnoses required:
        - 1: Any single diagnosis qualifies (more inclusive, more false positives)
        - 2: Requires 2 diagnoses (standard RWE criterion, fewer false positives)
        - 3+: Very restrictive (use for conditions with high testing frequency)
    days_apart : int, default=30
        Minimum days between first and last diagnosis when min_diagnosis_count >= 2.
        Common values: 30 (standard), 60 (more restrictive), 7 (less restrictive).

    Returns
    -------
    pd.DataFrame
        DataFrame with columns:
        - patient_id: Patient identifier (renamed from source-specific column)
        - index_date: First qualifying diagnosis date (datetime64)

    See Also
    --------
    get_claims : Previous step - retrieve raw claims
    add_demographics : Next step - add age/sex
    cohort.get_cohort : High-level function that handles all steps
    cohort.DiagnosisCriteria : More advanced diagnosis criteria options

    Notes
    -----
    - The index date is always the FIRST diagnosis date, even with min_count=2
    - Dates are converted to datetime64 format automatically
    - Patients not meeting criteria are excluded from output
    - For second diagnosis as index date, use `cohort.CohortCriteria(index_date_method="second_dx")`
    - Consider your study's sensitivity/specificity tradeoff when choosing min_count

    Examples
    --------
    Standard RWE approach: 2+ diagnoses 30 days apart for gMG:

    >>> df_index = get_index_dates(
    ...     df_claims,
    ...     source='iqvia',
    ...     min_diagnosis_count=2,
    ...     days_apart=30,
    ... )
    >>> print(f"Patients meeting 2+ Dx criterion: {len(df_index):,}")
    Patients meeting 2+ Dx criterion: 12,847

    Any single diagnosis (more inclusive, use for rare conditions):

    >>> df_index_any = get_index_dates(
    ...     df_claims,
    ...     source='iqvia',
    ...     min_diagnosis_count=1,
    ... )
    >>> print(f"Patients with any Dx: {len(df_index_any):,}")
    Patients with any Dx: 25,123

    Compare sensitivity vs specificity:

    >>> # More specific (fewer false positives)
    >>> df_strict = get_index_dates(df_claims, source='iqvia', min_diagnosis_count=2, days_apart=60)
    >>> # More sensitive (fewer false negatives)
    >>> df_lenient = get_index_dates(df_claims, source='iqvia', min_diagnosis_count=1)
    >>> print(f"Strict: {len(df_strict):,}, Lenient: {len(df_lenient):,}")
    """
    # Determine column names
    if source:
        config = get_source_config(source)
        id_col = patient_id_col or config["columns"]["patient_id"]
        dt_col = date_col or config["columns"]["service_date"]
    else:
        id_col = patient_id_col or "pat_id"
        dt_col = date_col or "from_dt"

    # Ensure date column is datetime
    df = df_claims.copy()
    df[dt_col] = pd.to_datetime(df[dt_col])

    if min_diagnosis_count == 1:
        # Simple case: first diagnosis date per patient
        df_index = (
            df.groupby(id_col)[dt_col]
            .min()
            .reset_index()
            .rename(columns={dt_col: "index_date", id_col: "patient_id"})
        )
    elif min_diagnosis_count >= 2:
        # Require multiple diagnoses with minimum gap
        patient_dates = df.groupby(id_col)[dt_col].agg(["min", "max"])
        patient_dates["diff_days"] = (patient_dates["max"] - patient_dates["min"]).dt.days

        # Filter to patients meeting the criteria
        qualifying = patient_dates[patient_dates["diff_days"] >= days_apart]

        df_index = (
            qualifying["min"]
            .reset_index()
            .rename(columns={"min": "index_date", id_col: "patient_id"})
        )
    else:
        raise ValueError(f"min_diagnosis_count must be >= 1, got {min_diagnosis_count}")

    return df_index

add_demographics ¶

add_demographics(df_index: DataFrame, df_claims: DataFrame | None = None, df_demographics: DataFrame | None = None, source: str | None = None, patient_id_col: str | None = None, yob_col: str | None = None, sex_col: str | None = None) -> pd.DataFrame

Add demographics (age, sex) to an index dates DataFrame.

In RWE studies, age and sex are essential demographic variables. Age is calculated at the index date (age_at_index), which accounts for patients aging over the study period. Sex is typically coded as 'M'/'F' in claims databases, with 'U' (unknown) indicating missing or ambiguous data.

For IQVIA, demographics (der_yob, der_sex) are in the enroll table, not in claims. Use get_claims(..., include_demographics=True) to include them automatically.

Workflow:

get_claims(include_demographics=True) - Query claims with demographics
get_index_dates() - Identify index dates
add_demographics() <-- you are here
filter_demographics() - Apply age/sex criteria

Parameters:

Name	Type	Description	Default
`df_index`	`DataFrame`	Index dates DataFrame with patient_id and index_date columns. Output from get_index_dates().	required
`df_claims`	`DataFrame`	Claims data containing demographic columns. For IQVIA, use `get_claims(..., include_demographics=True)` to include der_yob/der_sex.	`None`
`df_demographics`	`DataFrame`	Separate demographics table. Takes precedence over df_claims.	`None`
`source`	`str`	Data source name for column defaults: 'iqvia', 'optum', 'komodo'.	`None`
`patient_id_col`	`str`	Patient ID column name. Defaults to source-specific column.	`None`
`yob_col`	`str`	Year of birth column name. Defaults to source-specific column.	`None`
`sex_col`	`str`	Sex column name. Defaults to source-specific column.	`None`

Returns:

Type	Description
`DataFrame`	Index DataFrame with added columns: - year_of_birth: Patient's year of birth - sex: Patient's sex ('M', 'F', or 'U') - age_at_index: Age at index date (index_year - year_of_birth)

See Also

get_claims : Use include_demographics=True for IQVIA filter_demographics : Next step - apply age/sex criteria get_index_dates : Previous step - identify index dates cohort.get_cohort : High-level function that handles all steps

Notes

IQVIA: Use get_claims(..., include_demographics=True) to include der_yob and der_sex from the enrollment table
Optum: Demographics (yrdob, gdr_cd) are in claims, no special handling needed
Age is calculated as index_year - year_of_birth (approximate, not exact)

Examples:

Recommended approach (IQVIA):

>>> df_claims = get_claims(conn, source='iqvia', ..., include_demographics=True)
>>> df_index = get_index_dates(df_claims, source='iqvia', min_diagnosis_count=2)
>>> df_cohort = add_demographics(df_index, df_claims=df_claims, source='iqvia')

Alternative - provide demographics separately:

>>> df_enroll = conn.query('SELECT pat_id, der_yob, der_sex FROM schema.enroll')
>>> df_cohort = add_demographics(df_index, df_demographics=df_enroll, source='iqvia')

Source code in alx_heor\claims\__init__.py

def add_demographics(
    df_index: pd.DataFrame,
    df_claims: pd.DataFrame | None = None,
    df_demographics: pd.DataFrame | None = None,
    source: str | None = None,
    patient_id_col: str | None = None,
    yob_col: str | None = None,
    sex_col: str | None = None,
) -> pd.DataFrame:
    """Add demographics (age, sex) to an index dates DataFrame.

    In RWE studies, age and sex are essential demographic variables. Age is
    calculated at the index date (age_at_index), which accounts for patients
    aging over the study period. Sex is typically coded as 'M'/'F' in claims
    databases, with 'U' (unknown) indicating missing or ambiguous data.

    For IQVIA, demographics (der_yob, der_sex) are in the `enroll` table,
    not in claims. Use `get_claims(..., include_demographics=True)` to
    include them automatically.

    **Workflow:**

    1. get_claims(include_demographics=True) - Query claims with demographics
    2. get_index_dates() - Identify index dates
    3. **add_demographics()** <-- you are here
    4. filter_demographics() - Apply age/sex criteria

    Parameters
    ----------
    df_index : pd.DataFrame
        Index dates DataFrame with patient_id and index_date columns.
        Output from get_index_dates().
    df_claims : pd.DataFrame, optional
        Claims data containing demographic columns. For IQVIA, use
        `get_claims(..., include_demographics=True)` to include der_yob/der_sex.
    df_demographics : pd.DataFrame, optional
        Separate demographics table. Takes precedence over df_claims.
    source : str, optional
        Data source name for column defaults: 'iqvia', 'optum', 'komodo'.
    patient_id_col : str, optional
        Patient ID column name. Defaults to source-specific column.
    yob_col : str, optional
        Year of birth column name. Defaults to source-specific column.
    sex_col : str, optional
        Sex column name. Defaults to source-specific column.

    Returns
    -------
    pd.DataFrame
        Index DataFrame with added columns:
        - year_of_birth: Patient's year of birth
        - sex: Patient's sex ('M', 'F', or 'U')
        - age_at_index: Age at index date (index_year - year_of_birth)

    See Also
    --------
    get_claims : Use include_demographics=True for IQVIA
    filter_demographics : Next step - apply age/sex criteria
    get_index_dates : Previous step - identify index dates
    cohort.get_cohort : High-level function that handles all steps

    Notes
    -----
    - IQVIA: Use `get_claims(..., include_demographics=True)` to include
      der_yob and der_sex from the enrollment table
    - Optum: Demographics (yrdob, gdr_cd) are in claims, no special handling needed
    - Age is calculated as index_year - year_of_birth (approximate, not exact)

    Examples
    --------
    Recommended approach (IQVIA):

    >>> df_claims = get_claims(conn, source='iqvia', ..., include_demographics=True)
    >>> df_index = get_index_dates(df_claims, source='iqvia', min_diagnosis_count=2)
    >>> df_cohort = add_demographics(df_index, df_claims=df_claims, source='iqvia')

    Alternative - provide demographics separately:

    >>> df_enroll = conn.query('SELECT pat_id, der_yob, der_sex FROM schema.enroll')
    >>> df_cohort = add_demographics(df_index, df_demographics=df_enroll, source='iqvia')
    """
    # Determine column names
    if source:
        config = get_source_config(source)
        id_col = patient_id_col or config["columns"]["patient_id"]
        yob = yob_col or config["columns"]["year_of_birth"]
        sex = sex_col or config["columns"]["sex"]
    else:
        id_col = patient_id_col or "pat_id"
        yob = yob_col or "der_yob"
        sex = sex_col or "der_sex"

    # Map index patient_id back to source column name
    index_id_col = "patient_id" if "patient_id" in df_index.columns else id_col

    # Determine which DataFrame has demographics
    if df_demographics is not None:
        df_demo_source = df_demographics
    elif df_claims is not None:
        if yob in df_claims.columns and sex in df_claims.columns:
            df_demo_source = df_claims
        else:
            raise ValueError(
                f"Demographics columns ({yob}, {sex}) not found in df_claims. "
                f"For IQVIA, use get_claims(..., include_demographics=True) "
                f"to include demographics from the enrollment table."
            )
    else:
        raise ValueError(
            "Must provide df_claims or df_demographics. "
            "For IQVIA, use get_claims(..., include_demographics=True)."
        )

    # Get unique demographics per patient
    df_demo = (
        df_demo_source[[id_col, yob, sex]]
        .drop_duplicates(subset=id_col)
        .rename(columns={id_col: index_id_col, yob: "year_of_birth", sex: "sex"})
    )

    # Merge
    df = df_index.merge(df_demo, on=index_id_col, how="left")

    # Calculate age at index
    df["index_date"] = pd.to_datetime(df["index_date"])
    df["age_at_index"] = df["index_date"].dt.year - df["year_of_birth"]

    return df

filter_demographics ¶

filter_demographics(df: DataFrame, min_age: int | None = None, max_age: int | None = None, valid_sex_only: bool = True, age_col: str = 'age_at_index', sex_col: str = 'sex') -> pd.DataFrame

Filter DataFrame by age and sex inclusion/exclusion criteria.

Demographic filtering is a standard step in RWE cohort identification. Most studies restrict to adults (≥18), and many require known sex for subgroup analyses. This function applies these filters while preserving a copy of the original data (non-destructive).

Common age restrictions: - Adults only: min_age=18 - Working age: min_age=18, max_age=65 - Elderly: min_age=65 - Pediatric: max_age=17

Workflow:

get_claims() - Query raw claims data
get_index_dates() - Identify index dates
add_demographics() - Add age/sex to cohort
filter_demographics() <-- you are here
Continue to enrollment module or use cohort.get_cohort()

Parameters:

Name	Type	Description	Default
`df`	`DataFrame`	DataFrame with age and sex columns (from add_demographics).	required
`min_age`	`int`	Minimum age (inclusive). None means no lower bound.	`None`
`max_age`	`int`	Maximum age (inclusive). None means no upper bound.	`None`
`valid_sex_only`	`bool`	If True, keep only rows where sex is 'M' or 'F'. Removes 'U' (unknown) and any other values.	`True`
`age_col`	`str`	Name of age column to filter on.	`'age_at_index'`
`sex_col`	`str`	Name of sex column to filter on.	`'sex'`

Returns:

Type	Description
`DataFrame`	Copy of input DataFrame with rows filtered by criteria.

See Also

add_demographics : Previous step - add age/sex columns enrollment.analyze_enrollment : Next step - enrollment requirements cohort.get_cohort : High-level function that handles all steps

Notes

Returns a COPY, does not modify the input DataFrame
If age_col doesn't exist, age filters are skipped (no error)
If sex_col doesn't exist, sex filter is skipped (no error)
Sex values 'U', '', NaN are removed when valid_sex_only=True

Examples:

Filter to adults only (most common for non-pediatric studies):

>>> df_adults = filter_demographics(df_cohort, min_age=18)
>>> print(f"Adults: {len(df_adults):,} of {len(df_cohort):,}")
Adults: 42,103 of 45,678

Filter to working-age adults:

>>> df_working = filter_demographics(df_cohort, min_age=18, max_age=64)
>>> print(f"Working age: {len(df_working):,}")
Working age: 28,456

Filter to elderly patients:

>>> df_elderly = filter_demographics(df_cohort, min_age=65)
>>> print(f"Elderly (65+): {len(df_elderly):,}")
Elderly (65+): 13,647

Keep patients with unknown sex (useful for sensitivity analysis):

>>> df_all_sex = filter_demographics(df_cohort, min_age=18, valid_sex_only=False)
>>> print(f"Including unknown sex: {len(df_all_sex):,}")

Check attrition from demographic filters:

>>> n_before = len(df_cohort)
>>> df_filtered = filter_demographics(df_cohort, min_age=18)
>>> n_after = len(df_filtered)
>>> print(f"Excluded by age filter: {n_before - n_after:,} ({(n_before-n_after)/n_before*100:.1f}%)")

Source code in alx_heor\claims\__init__.py

def filter_demographics(
    df: pd.DataFrame,
    min_age: int | None = None,
    max_age: int | None = None,
    valid_sex_only: bool = True,
    age_col: str = "age_at_index",
    sex_col: str = "sex",
) -> pd.DataFrame:
    """Filter DataFrame by age and sex inclusion/exclusion criteria.

    Demographic filtering is a standard step in RWE cohort identification.
    Most studies restrict to adults (≥18), and many require known sex for
    subgroup analyses. This function applies these filters while preserving
    a copy of the original data (non-destructive).

    Common age restrictions:
    - Adults only: min_age=18
    - Working age: min_age=18, max_age=65
    - Elderly: min_age=65
    - Pediatric: max_age=17

    **Workflow:**

    1. get_claims() - Query raw claims data
    2. get_index_dates() - Identify index dates
    3. add_demographics() - Add age/sex to cohort
    4. **filter_demographics()** <-- you are here
    5. Continue to enrollment module or use cohort.get_cohort()

    Parameters
    ----------
    df : pd.DataFrame
        DataFrame with age and sex columns (from add_demographics).
    min_age : int, optional
        Minimum age (inclusive). None means no lower bound.
    max_age : int, optional
        Maximum age (inclusive). None means no upper bound.
    valid_sex_only : bool, default=True
        If True, keep only rows where sex is 'M' or 'F'.
        Removes 'U' (unknown) and any other values.
    age_col : str, default='age_at_index'
        Name of age column to filter on.
    sex_col : str, default='sex'
        Name of sex column to filter on.

    Returns
    -------
    pd.DataFrame
        Copy of input DataFrame with rows filtered by criteria.

    See Also
    --------
    add_demographics : Previous step - add age/sex columns
    enrollment.analyze_enrollment : Next step - enrollment requirements
    cohort.get_cohort : High-level function that handles all steps

    Notes
    -----
    - Returns a COPY, does not modify the input DataFrame
    - If age_col doesn't exist, age filters are skipped (no error)
    - If sex_col doesn't exist, sex filter is skipped (no error)
    - Sex values 'U', '', NaN are removed when valid_sex_only=True

    Examples
    --------
    Filter to adults only (most common for non-pediatric studies):

    >>> df_adults = filter_demographics(df_cohort, min_age=18)
    >>> print(f"Adults: {len(df_adults):,} of {len(df_cohort):,}")
    Adults: 42,103 of 45,678

    Filter to working-age adults:

    >>> df_working = filter_demographics(df_cohort, min_age=18, max_age=64)
    >>> print(f"Working age: {len(df_working):,}")
    Working age: 28,456

    Filter to elderly patients:

    >>> df_elderly = filter_demographics(df_cohort, min_age=65)
    >>> print(f"Elderly (65+): {len(df_elderly):,}")
    Elderly (65+): 13,647

    Keep patients with unknown sex (useful for sensitivity analysis):

    >>> df_all_sex = filter_demographics(df_cohort, min_age=18, valid_sex_only=False)
    >>> print(f"Including unknown sex: {len(df_all_sex):,}")

    Check attrition from demographic filters:

    >>> n_before = len(df_cohort)
    >>> df_filtered = filter_demographics(df_cohort, min_age=18)
    >>> n_after = len(df_filtered)
    >>> print(f"Excluded by age filter: {n_before - n_after:,} ({(n_before-n_after)/n_before*100:.1f}%)")
    """
    df_out = df.copy()

    if min_age is not None:
        df_out = df_out[df_out[age_col] >= min_age]

    if max_age is not None:
        df_out = df_out[df_out[age_col] <= max_age]

    if valid_sex_only and sex_col in df_out.columns:
        df_out = df_out[df_out[sex_col].isin(["M", "F"])]

    return df_out

Claims¶

When to Use¶

Quick Example¶

Common Patterns¶

Custom Column Selection¶

Building SQL Manually¶

Related Modules¶

claims ¶

get_claims ¶

build_claims_sql ¶

get_index_dates ¶

add_demographics ¶

filter_demographics ¶