Необходим обзор: Методология очистки данных для временных рядов CGM – первый реальный набор данных

Я работаю над обработкой данных непрерывного мониторинга глюкозы (CGM) из XDrip+ и буду признателен за отзывы о моей методологии очистки данных. Это мой первый опыт работы с “грязными” медицинскими данными из реального мира после изучения основ науки о данных.

Вот мой текущий подход:

Обработка данных глюкозы:

• Округление временных меток до 5 минут
• Создание полной временной линии с флагами отсутствующих значений
• Интерполяция пробелов до 20 минут (4 измерения) с использованием линейной интерполяции
• Ограничение значений в физиологических пределах (39.64-360.36 мг/дл)
• Конвертация в миллиграммы/дл и ммоль/л

Обработка данных инсулина:

• Парсинг JSON-метаданных для определения типов инсулина (Novorapid/Levemir)
• Классификация не маркированного инсулина по дозировке (≤8 единиц = болюс, >8 единиц = базальный)
• Флагирование и исключение подозрительных записей (не маркированные дозы >15 единиц)
• Отслеживание уверенности с флагами маркированного/немаркированного

Данные углеводов:

• Фильтрация записей <1 г (вероятные ошибки)
• Удаление дублирующихся временных меток

Выравнивание данных:

• Использование временной линии глюкозы в качестве основного индекса
• Суммирование нескольких записей инсулина/углеводов в одном 5-минутном окне
• Заполнение отсутствующих обработок нулем
• Сохранение значений NaN для глюкозы и флагов пробелов

Вопросы:

1. Разумны ли мои правила и пороги валидации для медицинских данных?
2. Подходит ли линейная интерполяция для пробелов в данных по глюкозе?
3. Как мне проверить качество моего очищенного набора данных?
4. Какие критические шаги очищения я мог пропустить?
5. Существуют ли установленные рекомендации для данных CGM, которым я должен следовать?

Код приведен ниже. Спасибо за вашу помощь!

https://github.com/Warren8824/cgm-data-processor

import pandas as pd
import numpy as np
import json
from typing import Dict


def clean_classify_insulin(df, bolus_limit=8, max_limit=15):
    """
    Очистка и классификация данных инсулина на базальные и болюсные категории.
    Параметры:
        df: DataFrame с колонками инсулина и insulinJSON
        bolus_limit: Порог единиц для классификации немаркированного инсулина как болюс
        max_limit: Максимальное количество единиц для немаркированного инсулина


    Возвращает:
        df_clean: DataFrame с базальными, болюсными и колонками labeled_insulin с
        индексом datetime.
    """
    df_clean = df[df['insulin'] > 0.0].copy()
    df_clean = df_clean[~df_clean.index.duplicated(keep='first')]

    # Инициализация колонок
    df_clean['bolus'] = 0.0
    df_clean['basal'] = 0.0
    # Инициализация с явным типом bool и установка на False
    df_clean['labeled_insulin'] = pd.Series(False, index=df_clean.index, dtype=bool)

    # Обработка маркированного инсулина из JSON
    for idx, row in df_clean.iterrows():
        try:
            insulin_data = json.loads(row['insulinJSON'])
            if insulin_data and isinstance(insulin_data, list):
                insulin_type = insulin_data[0].get('insulin', '').lower()
                if 'novorapid' in insulin_type:
                    df_clean.at[idx, 'bolus'] = row['insulin']
                    df_clean.at[idx, 'labeled_insulin'] = True  # Устанавливается как маркированный только если явно помечен
                elif 'levemir' in insulin_type:
                    df_clean.at[idx, 'basal'] = row['insulin']
                    df_clean.at[idx, 'labeled_insulin'] = True  # Устанавливается как маркированный только если явно помечен
        except (json.JSONDecodeError, IndexError, KeyError, AttributeError):
            continue

    # Создание маски для немаркированных доз инсулина и для доз выше 15 единиц
    unlabeled = (df_clean['bolus'] == 0) & (df_clean['basal'] == 0)
    valid_insulin = df_clean['insulin'] <= max_limit

    # Удаление доз инсулина, которые остались немаркированными и выходят за пределы нашего определенного диапазона - > 15 единиц
    df_clean = df_clean[~(unlabeled & ~valid_insulin)]

    # Классификация оставшегося действительного немаркированного инсулина на основе единиц - 1-8 = болюс, >8 = базальный
    # Примечание: Они остаются помеченными как немаркированные даже после классификации
    df_clean.loc[unlabeled & valid_insulin & (df_clean['insulin'] <= bolus_limit), 'bolus'] = df_clean['insulin']
    df_clean.loc[unlabeled & valid_insulin & (df_clean['insulin'] > bolus_limit), 'basal'] = df_clean['insulin']

    # Возврат датафрейма, содержащего только данные, связанные с инсулином
    df_clean = df_clean[['basal', 'bolus', 'labeled_insulin']]

    return df_clean



def clean_classify_carbs(df):
    # Создание копии, чтобы избежать изменения оригинального датафрейма
    df_clean = df.copy()

    # Оставить только строки, где углеводы >= 1.0 грамм
    df_clean = df_clean[df_clean['carbs'] >= 1.0]

    # Удаление строк, где индекс (временная метка) дублируется
    df_clean = df_clean[~df_clean.index.duplicated(keep='first')]

    # Возврат DataFrame, содержащего только данные о приеме пищи
    df_clean = df_clean[['carbs']]

    return df_clean


def clean_glucose(df, interpolation_limit=4):
    # Создание копии, чтобы избежать изменения оригинального датафрейма
    clean_df = df.copy()

    # Округление всех временных меток до ближайшего 5-минутного интервала
    clean_df.index = clean_df.index.round('5min')

    # Оставить только числовую колонку 'calculated_value' перед сгруппировкой
    clean_df = clean_df[['calculated_value']]

    # Создание полного 5-минутного индексного массива
    full_index = pd.date_range(
        start=clean_df.index.min(),
        end=clean_df.index.max(),
        freq='5min'
    )

    # Переиндексация, чтобы включить все интервалы и обработать дублирующиеся метки времени
    clean_df = clean_df.groupby(level=0).mean().reindex(full_index)

    # Создание флага для всех строк с отсутствующими данными
    clean_df['missing'] = clean_df['calculated_value'].isna()

    # Создание групп последовательных отсутствующих значений
    # Когда отсутствующие значения меняются (True на False или наоборот), cumsum увеличивается
    clean_df['gap_group'] = (~clean_df['missing']).cumsum()

    # В рамках каждой группы False (где missing=True) подсчитываем размер группы
    gap_sizes = clean_df[clean_df['missing']].groupby('gap_group').size()

    # Выявление групп пробелов, которые больше interpolation_limit
    large_gaps = gap_sizes[gap_sizes > interpolation_limit].index

    # Интерполяция всех пробелов изначально
    clean_df['calculated_value'] = clean_df['calculated_value'].interpolate(
        method='linear',
        limit=interpolation_limit,
        limit_direction='forward'
    )

    # Сброс интерпольированных значений обратно в NaN для больших пробелов
    for gap_group in large_gaps:
        mask = (clean_df['gap_group'] == gap_group) & clean_df['missing']
        clean_df.loc[mask, 'calculated_value'] = np.nan

    # Переименование колонки 'calculated_value' в 'mg_dl'
    clean_df.rename(columns={'calculated_value': 'mg_dl'}, inplace=True)

    # Ограничение значений колонки 'mg_dl' диапазоном от 39.64 до 360.36 (2.2 - 20.0 ммоль/л)
    clean_df['mg_dl'] = clean_df['mg_dl'].clip(lower=39.64, upper=360.36)

    # Создание новой колонки 'mmol_l' путем конвертации 'calculated_value' из мг/дл в ммоль/л
    clean_df['mmol_l'] = clean_df['mg_dl'] * 0.0555

    # Удаление всех колонок, кроме mg_dl и mmol_l
    clean_df = clean_df[['mg_dl', 'mmol_l', 'missing']]

    return clean_df

import pandas as pd
import numpy as np
from typing import Dict
from datetime import datetime


def align_diabetes_data(
        bg_df: pd.DataFrame,
        carb_df: pd.DataFrame,
        insulin_df: pd.DataFrame
) -> pd.DataFrame:
    """
    Выравнивает данные по диабету (уровень глюкозы в крови, углеводы, инсулин) до регулярных 5-минутных интервалов,
    уже установленных в датафрейме bg_df.

    Эта функция принимает три датафрейма с временными метками по диабету и создает один
    датафрейм с регулярными 5-минутными интервалами. Углеводы и записи инсулина суммируются. Этот
    набор данных подходит для машинного обучения и других приложений, где требуется единая временная метка.

    Аргументы:
        bg_df: DataFrame с индексом временной метки и колонками ['mg_dl', 'mmol_l', 'missing']
              Измерения уровня глюкозы в крови в нерегулярных интервалах
        carb_df: DataFrame с индексом временной метки и колонкой ['carbs']
                Спорадические записи углеводов
        insulin_df: DataFrame с индексом временной метки и колонками ['bolus', 'basal', 'labeled_insulin']
                   Спорадические записи инсулина

    Шаги обработки:
    1. Создайте копии датафреймов, чтобы избежать предупреждений
    3. Округлите временные метки carb_df и insulin_df до ближайших 5 минут
    3a. Оформите данные по инсулину - суммируйте все записи в каждом 5-минутном окне
    3b. Переиндексируйте этот датафрейм в bg_df.index для выравнивания временных меток
    4a. Оформите данные по углеводам - суммируйте все записи в каждом 5-минутном окне
    4b. Переиндексируйте этот датафрейм в bg_df.index для выравнивания временных меток
    5. Объедините все данные и убедитесь, что все интервалы существуют
    6. Заполните отсутствующие значения обработок нулями (сохранив BG как NaN)
    7. Заполните отсутствующие строки insulin_labeled (где bg_df имеет индекс, а insulin_resampled нет)
    значением False

    Возвращает:
        DataFrame с:
        - Регулярным индексом 5-минутного интервала
        - Колонки: ['mg_dl', 'mmol_l', 'missing', 'carbs', 'bolus', 'basal', 'labeled_insulin']
        - Значения BG, усредненные в рамках интервалов, NaN, где отсутствуют
        - Значения обработки, суммированные в интервалах, 0, где отсутствуют
    """
    # Сначала округлите временные метки во всех датафреймах до 5-минутных интервалов
    bg_df = bg_df.copy()
    carb_df = carb_df.copy()
    insulin_df = insulin_df.copy()

    # Округление временных меток до ближайших 5 минут
    carb_df.index = carb_df.index.round('5min')
    insulin_df.index = insulin_df.index.round('5min')

    # Оформление и переиндексация инсулина - суммируйте по окнам
    insulin_resampled = insulin_df.resample('5min').agg({
        'bolus': 'sum',
        'basal': 'sum',
        # Используйте лямбду, чтобы убедиться, что пустые группы являются явно False - Пустые группы по умолчанию равны True
        'labeled_insulin': lambda x: x.all() if len(x) > 0 else False
    }).reindex(bg_df.index)

    # Заполнение отсутствующих значений labeled_insulin значением False, явным преобразованием в логический тип, чтобы избежать предупреждений
    insulin_resampled['labeled_insulin'] = insulin_resampled['labeled_insulin'].astype('boolean').fillna(False)

    # Оформление и переиндексация углеводов - суммируйте по окнам
    carb_resampled = carb_df.resample('5min').agg({
        'carbs': 'sum'
    }).reindex(bg_df.index)

    # Объединение всех данных
    aligned_df = pd.concat([bg_df, carb_resampled, insulin_resampled], axis=1)

    # Заполнение отсутствующих значений обработки нулями
    aligned_df['carbs'] = aligned_df['carbs'].fillna(0)
    aligned_df[['basal', 'bolus']] = aligned_df[['basal', 'bolus']].fillna(0)


    return aligned_df