Custom Variables Guide

AI-generated content

This page was generated by AI and has not been fully reviewed by a human. Content may be inaccurate or incomplete. If you find any issues, please create an issue on the GitHub repository.

This guide explains how to build variables that are not in the catalogue: aggregations of existing time series, derived values computed by your own Python functions, and — for a deployment that has bound a store — native extractions.

All the classes live in corr_vars.sources.local_datasource. Despite the name, the aggregation and derivation classes are source-agnostic: they compute from data that is already in the cohort, so they work on a cohort built from any source.

Understanding Variable Architecture

Variable Type Comparison

Variable Type	Input	Purpose	Class
NativeDynamic	Backing store	New time series fetched from a store	local_datasource.NativeDynamic
ComplexVariable	Whatever the py function builds	Time series or single value produced entirely in Python	local_datasource.ComplexVariable
NativeStatic	ONE existing dynamic variable	Aggregation (!first, !max, !mean, …)	local_datasource.NativeStatic
DerivedStatic	Existing columns on the observation frame	One value per observation, from an expression or py	local_datasource.DerivedStatic
DerivedDynamic	Existing variables in requires	New time series computed in Python	local_datasource.DerivedDynamic

Only NativeDynamic touches a backing store. Everything else processes data that is already loaded, which is why the same classes work on a reprodICU cohort.

The two paths

There are two ways to add a custom variable to a cohort.

Path 1 — direct construction

Build the object and hand it to add_variable(). This works on any cohort:

from corr_vars import Cohort
from corr_vars.sources.local_datasource import (
    DerivedStatic,
    NativeStatic,
)

cohort = Cohort(
    obs_level="icu_stay",
    sources={"reprodicu": {"path": "/path/to/reprodICU_files"}},
    project="my_project",
)

cohort.add_variable(NativeStatic(
    var_name="max_sodium_icu",
    select="!max value",
    base_var="blood_sodium",
    tmin="icu_admission",
    tmax="icu_discharge",
))
cohort.add_variable(DerivedStatic(
    var_name="hypernatremia",
    requires=["max_sodium_icu"],
    expression="max_sodium_icu > 145",
))

base_var and requires resolve through the cohort’s own sources, so blood_sodium above is served by reprodicu. Pass tmin/tmax inside the constructor; passing them to add_variable() for an object emits a UserWarning.

Path 2 — declarative definition

Register a definition dict, then add the variable by name. This is the same shape a source’s bundled vars.json and the CORR Concepts API use:

cohort.add_variable_definition("max_sodium_icu", {
    "type": "native_static",
    "select": "!max value",
    "base_var": "blood_sodium",   # NOT "requires"
})
cohort.add_variable("max_sodium_icu", tmin="icu_admission", tmax="icu_discharge")

For native_static the field is base_var (the legacy alias variable is still accepted), never requires.

Warning

Which path works where. A project-local definition is dispatched to the VariableLoader of every source configured on the cohort, and only a loader that knows the type can build it. local_datasource’s loader knows native_static, derived_static, derived_dynamic, native_dynamic, complex and free_days. The reprodicu loader does not know select / base_var / expression and raises TypeError on them.

On a reprodICU-only cohort, Path 2 is for overrides of served definitions (cleaning, filter, …) and for reprodICU’s own fields. For the aggregation types, use Path 1. Path 2 for aggregation types is for a deployment that has bound local_datasource to a store (see Local Data Source (skeleton)), and for authoring definitions in vars.json or the Concepts API.

Derived Variables with Custom Functions (py Parameter)

Anything an expression cannot express goes into a py function.

Function Signature

def my_custom_function(var, cohort):
    """Custom variable calculation.

    Args:
        var: The variable object. Useful attributes: ``var.var_name``,
            ``var.time_window``, ``var.data`` (this variable's own extracted
            rows, for native and complex variables) and
            ``var.required_vars["name"].data`` (a dependency's data).
        cohort: The cohort, e.g. ``cohort.obs`` and ``cohort.primary_key``.

    Returns:
        polars.DataFrame: The variable's data.
    """
    ...

Always pass py_ready_polars=True for new code — the function then receives and returns polars frames directly.

DerivedDynamic: time-series output

A shock index (heart rate / systolic blood pressure) as a new time series:

import polars as pl

from corr_vars import Cohort
from corr_vars.sources.local_datasource import DerivedDynamic

def calculate_shock_index(var, cohort):
    """Calculate the shock index from heart rate and blood pressure."""
    hr_data = var.required_vars["heart_rate"].data
    sbp_data = var.required_vars["blood_pressure_sys"].data

    merged = hr_data.join(
        sbp_data,
        on=["icu_stay_id", "recordtime"],
        how="inner",
        suffix="_sbp",
    )

    return merged.select(
        "icu_stay_id",
        "recordtime",
        (pl.col("value") / pl.col("value_sbp")).alias("value"),
    )

shock_index_var = DerivedDynamic(
    var_name="shock_index_dynamic",
    requires=["heart_rate", "blood_pressure_sys"],
    py=calculate_shock_index,
    py_ready_polars=True,
    cleaning={"value": {"low": 0.1, "high": 5.0}},
)

cohort = Cohort(
    obs_level="icu_stay",
    sources={"reprodicu": {"path": "/path/to/reprodICU_files"}},
    project="my_project",
    load_default_vars=False,
)
cohort.add_variable(shock_index_var)

The dependencies in requires are loaded automatically; adding them by hand first is optional.

DerivedStatic: single value per observation

A simplified organ-dysfunction score built from admission values:

import polars as pl

from corr_vars.sources.local_datasource import DerivedStatic

def calculate_dysfunction_score(var, cohort):
    """Score creatinine, platelets and bilirubin at their first measurement."""

    def _first(name, alias):
        return (
            var.required_vars[name]
            .data.group_by("icu_stay_id")
            .agg(pl.col("value").first().alias(alias))
        )

    result = (
        cohort.obs.select("icu_stay_id")
        .join(_first("blood_creatinine", "creat"), on="icu_stay_id", how="left")
        .join(_first("blood_platelets", "plt"), on="icu_stay_id", how="left")
        .join(_first("blood_bilirubin", "bili"), on="icu_stay_id", how="left")
    )

    creat_score = (
        pl.when(pl.col("creat") < 1.2).then(0)
        .when(pl.col("creat") < 2.0).then(1)
        .when(pl.col("creat") < 3.5).then(2)
        .otherwise(3)
    )
    plt_score = (
        pl.when(pl.col("plt") >= 150).then(0)
        .when(pl.col("plt") >= 100).then(1)
        .when(pl.col("plt") >= 50).then(2)
        .otherwise(3)
    )
    bili_score = (
        pl.when(pl.col("bili") < 1.2).then(0)
        .when(pl.col("bili") < 2.0).then(1)
        .when(pl.col("bili") < 6.0).then(2)
        .otherwise(3)
    )

    return result.select(
        "icu_stay_id",
        (creat_score + plt_score + bili_score).alias("dysfunction_score"),
    )

dysfunction_var = DerivedStatic(
    var_name="dysfunction_score",
    requires=["blood_creatinine", "blood_platelets", "blood_bilirubin"],
    py=calculate_dysfunction_score,
    py_ready_polars=True,
)
cohort.add_variable(dysfunction_var)

For something this simple, an expression is usually the better tool — reserve py for logic SQL cannot express.

ComplexVariable: build the frame yourself

ComplexVariable runs no query and no aggregation: the py function produces the frame from required_vars, and dynamic declares whether the result is a time series or one value per observation.

import polars as pl

from corr_vars.sources.local_datasource import ComplexVariable

def glucose_category(var, cohort):
    """Categorise every glucose measurement."""
    data = var.required_vars["blood_glucose"].data
    return data.select(
        "icu_stay_id",
        "recordtime",
        pl.when(pl.col("value") < 70).then(pl.lit("hypoglycemia"))
        .when(pl.col("value") < 140).then(pl.lit("normal"))
        .when(pl.col("value") < 200).then(pl.lit("hyperglycemia"))
        .otherwise(pl.lit("severe_hyperglycemia"))
        .alias("value"),
    )

glucose_var = ComplexVariable(
    var_name="glucose_category",
    dynamic=True,
    requires=["blood_glucose"],
    py=glucose_category,
    py_ready_polars=True,
)
cohort.add_variable(glucose_var)

Simple Aggregations: NativeStatic

If all you need is one statistic of one dynamic variable, do not write a function — use NativeStatic:

from corr_vars.sources.local_datasource import NativeStatic

# First blood pressure measurement
first_bp = NativeStatic(
    var_name="first_blood_pressure_sys",
    select="!first value",
    base_var="blood_pressure_sys",
)

# Maximum heart rate in the first 24 hours
max_hr_24h = NativeStatic(
    var_name="max_heart_rate_24h",
    select="!max value",
    base_var="heart_rate",
    tmin="icu_admission",
    tmax=("icu_admission", "+24h"),
)

# Blood pressure closest to admission, within two hours
admission_bp = NativeStatic(
    var_name="admission_blood_pressure",
    select="!closest(icu_admission, 0, 2h) value",
    base_var="blood_pressure_sys",
)

for variable in (first_bp, max_hr_24h, admission_bp):
    cohort.add_variable(variable)

The full list of aggregation functions and of the where filters is in Aggregated Variables.

Native Variables (store extraction)

NativeDynamic fetches new rows from a backing store. It is the one class here that only works in a deployment that has bound the source skeleton to a store — that is, one that subclassed NativeExtractor, implemented extract(), and set NativeDynamic.extractor_class. Without that binding, extraction raises NotImplementedError.

from corr_vars.sources.local_datasource import NativeDynamic

# Only in a deployment that bound NativeDynamic.extractor_class to its store.
custom_lab_var = NativeDynamic(
    var_name="blood_custom_marker",
    dynamic=True,
    table="labs",
    where="analyte = 'custom marker'",
    value_dtype="Float64",
    cleaning={"value": {"low": 0.1, "high": 100.0}},
)
cohort.add_variable(custom_lab_var)

table, where, value_dtype and columns are handed to the extractor; the BaseDynamic pipeline around it adds caching, the time filter, cleaning and optional py post-processing. Because NativeDynamic also accepts py, a native variable can post-process its own extraction: var.data then holds the raw rows.

See Local Data Source (skeleton) for how to bind the skeleton to a store, and for the configuration and mapping files a deployment fills in.

Debugging and Testing

Use a small cohort

Iterating on a py function against the full dataset is slow. Limit a reprodICU cohort to one dataset with include_datasets:

test_cohort = Cohort(
    obs_level="icu_stay",
    sources={"reprodicu": {
        "path": "/path/to/reprodICU_files",
        "include_datasets": ["eicu"],
    }},
    project="my_project",
    load_default_vars=False,
)

test_cohort.add_variable(my_custom_variable)
print(test_cohort.obs["my_variable_name"].describe())

Inspect the inputs

Most failures are shape problems, not logic problems. Print what the function actually received:

def debug_custom_function(var, cohort):
    print(f"variable: {var.var_name}, window: {var.time_window}")
    print(f"cohort rows: {len(cohort.obs)}")
    for name, req in var.required_vars.items():
        rows = 0 if req.data is None else len(req.data)
        print(f"  {name}: {rows} rows, columns {req.data.columns}")
    ...

Checklist

1. The returned frame carries the cohort’s primary key (cohort.primary_key).

2. A dynamic variable also returns recordtime; a static one must not.

3. The value column is named value (or is renamed by the caller for multi-column selects).

4. Empty inputs return an empty frame with the right columns rather than raising.

5. cleaning bounds match the unit the values are actually in.

Common Patterns

Rolling average

import polars as pl

from corr_vars.sources.local_datasource import DerivedDynamic

def rolling_average_function(var, cohort):
    """Six-hour rolling mean of the lactate time series."""
    data = var.required_vars["blood_lactate"].data

    return (
        data.sort(["icu_stay_id", "recordtime"])
        .with_columns(
            pl.col("value")
            .rolling_mean_by("recordtime", window_size="6h")
            .over("icu_stay_id")
            .alias("value")
        )
        .select("icu_stay_id", "recordtime", "value")
    )

rolling_lactate = DerivedDynamic(
    var_name="lactate_rolling_6h",
    requires=["blood_lactate"],
    py=rolling_average_function,
    py_ready_polars=True,
)

Mean arterial pressure from two series

import polars as pl

from corr_vars.sources.local_datasource import DerivedDynamic

def calculate_map_function(var, cohort):
    """MAP = (2 * diastolic + systolic) / 3, matched within five minutes."""
    sys_data = var.required_vars["blood_pressure_sys"].data
    dia_data = var.required_vars["blood_pressure_dia"].data

    joined = sys_data.sort("recordtime").join_asof(
        dia_data.sort("recordtime").rename({"value": "dia_value"}),
        on="recordtime",
        by="icu_stay_id",
        tolerance="5m",
    )

    return joined.select(
        "icu_stay_id",
        "recordtime",
        ((2 * pl.col("dia_value") + pl.col("value")) / 3).alias("value"),
    )

map_calculated = DerivedDynamic(
    var_name="blood_pressure_mean_calculated",
    requires=["blood_pressure_sys", "blood_pressure_dia"],
    py=calculate_map_function,
    py_ready_polars=True,
    cleaning={"value": {"low": 30, "high": 150}},
)

Summary

Do you need new rows from a backing store?

• Yes → NativeDynamic, in a deployment that bound an extractor. See Local Data Source (skeleton).

• No → process what is already in the cohort:

  • one statistic of one dynamic variable → NativeStatic

  • one value per observation from existing columns → DerivedStatic (expression first, py if needed)

  • a new time series → DerivedDynamic

  • a frame your function builds outright → ComplexVariable

Prefer Path 1 (direct construction) in analysis code. Use Path 2 (declarative definitions) where a loader that understands the type will see them: a bound local_datasource, a source’s vars.json, or the Concepts API.