RelativeTimeRangeFilter

Bases: Filter

Filters events based on a time range relative to an anchor date.

Every time-relative operation in PhenEx requires an anchor date — the reference point from which "before" and "after" are measured. The anchor date is resolved as follows:

If anchor_phenotype is provided, the EVENT_DATE of that phenotype is used as the anchor. The anchor phenotype must be executed before this filter runs (PhenEx handles this automatically when phenotypes are composed via the Cohort).
If anchor_phenotype is None (the default), the INDEX_DATE column already present on the table is used. INDEX_DATE is set by the entry criterion of the cohort.

The when parameter controls the direction:

when='before': days prior to the anchor are positive, days after are negative.
when='after': days after the anchor are positive, days before are negative.

This convention means that min_days and max_days are always expressed as positive numbers regardless of direction.

Parameters:

Name	Type	Description	Default
`min_days`	`Optional[Value]`	Minimum number of days from the anchor date to include. Must use `GreaterThan` or `GreaterThanOrEqualTo`.	`GreaterThanOrEqualTo(0)`
`max_days`	`Optional[Value]`	Maximum number of days from the anchor date to include. Must use `LessThan` or `LessThanOrEqualTo`.	`None`
`anchor_phenotype`	`Optional[Phenotype]`	A phenotype whose EVENT_DATE is used as the anchor date. If None, the INDEX_DATE column on the input table is used.	`None`
`when`	`Optional[str]`	Direction relative to the anchor. Either 'before' or 'after'.	`'before'`

Examples:

Baseline period — one year before index date

from phenex.filters import RelativeTimeRangeFilter
from phenex.filters.value import LessThan, GreaterThan

one_year_preindex = RelativeTimeRangeFilter(
    max_days=LessThan(365),
    min_days=GreaterThan(0),
    when='before',
)

Follow-up period — anytime after index date

from phenex.filters import RelativeTimeRangeFilter
from phenex.filters.value import GreaterThanOrEqualTo

post_index = RelativeTimeRangeFilter(
    min_days=GreaterThanOrEqualTo(0),
    when='after',
)

Anchored to another phenotype — events within 90 days after AF diagnosis

from phenex.phenotypes import CodelistPhenotype
from phenex.codelists import Codelist
from phenex.filters import RelativeTimeRangeFilter
from phenex.filters.value import GreaterThanOrEqualTo, LessThanOrEqualTo

af = CodelistPhenotype(
    name='af',
    domain='CONDITION_OCCURRENCE',
    codelist=Codelist([313217]),
    return_date='first',
)

within_90d_of_af = RelativeTimeRangeFilter(
    anchor_phenotype=af,
    min_days=GreaterThanOrEqualTo(0),
    max_days=LessThanOrEqualTo(90),
    when='after',
)

Source code in phenex/filters/relative_time_range_filter.py

class RelativeTimeRangeFilter(Filter):
    """
    Filters events based on a time range relative to an anchor date.

    Every time-relative operation in PhenEx requires an **anchor date** — the reference point from which "before" and "after" are measured. The anchor date is resolved as follows:

    1. **If `anchor_phenotype` is provided**, the EVENT_DATE of that phenotype is used as the anchor. The anchor phenotype must be executed before this filter runs (PhenEx handles this automatically when phenotypes are composed via the Cohort).
    2. **If `anchor_phenotype` is None** (the default), the `INDEX_DATE` column already present on the table is used. INDEX_DATE is set by the entry criterion of the cohort.

    The `when` parameter controls the direction:

    - `when='before'`: days *prior to* the anchor are positive, days *after* are negative.
    - `when='after'`: days *after* the anchor are positive, days *before* are negative.

    This convention means that `min_days` and `max_days` are always expressed as positive numbers regardless of direction.

    Parameters:
        min_days: Minimum number of days from the anchor date to include. Must use `GreaterThan` or `GreaterThanOrEqualTo`.
        max_days: Maximum number of days from the anchor date to include. Must use `LessThan` or `LessThanOrEqualTo`.
        anchor_phenotype: A phenotype whose EVENT_DATE is used as the anchor date. If None, the INDEX_DATE column on the input table is used.
        when: Direction relative to the anchor. Either 'before' or 'after'.

    Examples:

    Example: Baseline period — one year before index date
        ```python
        from phenex.filters import RelativeTimeRangeFilter
        from phenex.filters.value import LessThan, GreaterThan

        one_year_preindex = RelativeTimeRangeFilter(
            max_days=LessThan(365),
            min_days=GreaterThan(0),
            when='before',
        )
        ```

    Example: Follow-up period — anytime after index date
        ```python
        from phenex.filters import RelativeTimeRangeFilter
        from phenex.filters.value import GreaterThanOrEqualTo

        post_index = RelativeTimeRangeFilter(
            min_days=GreaterThanOrEqualTo(0),
            when='after',
        )
        ```

    Example: Anchored to another phenotype — events within 90 days after AF diagnosis
        ```python
        from phenex.phenotypes import CodelistPhenotype
        from phenex.codelists import Codelist
        from phenex.filters import RelativeTimeRangeFilter
        from phenex.filters.value import GreaterThanOrEqualTo, LessThanOrEqualTo

        af = CodelistPhenotype(
            name='af',
            domain='CONDITION_OCCURRENCE',
            codelist=Codelist([313217]),
            return_date='first',
        )

        within_90d_of_af = RelativeTimeRangeFilter(
            anchor_phenotype=af,
            min_days=GreaterThanOrEqualTo(0),
            max_days=LessThanOrEqualTo(90),
            when='after',
        )
        ```
    """

    def __init__(
        self,
        min_days: Optional[Value] = GreaterThanOrEqualTo(0),
        max_days: Optional[Value] = None,
        when: Optional[str] = "before",
        anchor_phenotype: Optional["Phenotype"] = None,
    ):
        verify_relative_time_range_filter_input(min_days, max_days, when)

        self.min_days = min_days
        self.max_days = max_days
        self.when = when
        self.anchor_phenotype = anchor_phenotype
        super(RelativeTimeRangeFilter, self).__init__()

    def _filter(self, table: EventTable):
        if self.anchor_phenotype is not None:
            if self.anchor_phenotype.table is None:
                raise ValueError(
                    f"Dependent Phenotype {self.anchor_phenotype.name} must be executed before this node can run!"
                )
            else:
                anchor_table = self.anchor_phenotype.table
                reference_column = anchor_table.EVENT_DATE
                join_keys = [
                    k for k in _get_join_keys(table) if k in anchor_table.columns
                ]
                table = table.join(anchor_table, join_keys)
        else:
            assert (
                "INDEX_DATE" in table.columns
            ), f"INDEX_DATE column not found in table {table}"
            reference_column = table.INDEX_DATE

        DAYS_FROM_ANCHOR = reference_column.cast("timestamp").delta(
            table.EVENT_DATE.cast("timestamp"), "day"
        )
        if self.when == "after":
            DAYS_FROM_ANCHOR = -DAYS_FROM_ANCHOR

        table = table.mutate(DAYS_FROM_ANCHOR=DAYS_FROM_ANCHOR)

        conditions = []

        value_filter = ValueFilter(
            min_value=self.min_days,
            max_value=self.max_days,
            column_name="DAYS_FROM_ANCHOR",
        )

        return value_filter.filter(table)

`filter(table)`

Filters the given table according to the rules of the Filter.

Parameters:

Name	Type	Description	Default
`table`	`PhenexTable`	The table to be filtered.	required

Returns:

Name	Type	Description
`PhenexTable`	`PhenexTable`	The filtered table. The returned table has the exact same schema as the input table but has rows removed.

Source code in phenex/filters/filter.py

def filter(self, table: PhenexTable) -> PhenexTable:
    """
    Filters the given table according to the rules of the Filter.

    Args:
        table (PhenexTable): The table to be filtered.

    Returns:
        PhenexTable: The filtered table. The returned table has the exact same schema as the input table but has rows removed.
    """
    input_columns = table.columns
    filtered_table = self._filter(table)
    if not set(input_columns) <= set(filtered_table.columns):
        raise ValueError(f"Filter must not remove columns.")

    filtered_table = filtered_table.select(input_columns)
    if isinstance(table, PhenexTable):
        return type(table)(filtered_table)
    else:
        return filtered_table