3
\$\begingroup\$

This is part of my research project with the main objective to select patients from various characteristics that are modeled as filters. The main filters include VALID_INDEX_STATUS, VALID_DX_STATUS, VALID_REG_STATUS, and VALID_RX_STATUS, where VALID_RX_STATUS describes if patient has certain treatment drug pattern (i.e., number of particular drugs before and after index date), which is handled by the RxStatus.

I created other python scripts that take care of data preparation as in sec1_data_preparation and data import as in sec2_prepped_data_import.

import sys
from abc import ABC, abstractmethod
import pandas as pd
import datetime
import ctypes
import numpy as np
import random
import pysnooper
import var_creator.var_creator as vc
import feature_tagger.feature_tagger as ft
import data_descriptor.data_descriptor as dd
import data_transformer.data_transformer as dt
import helper_functions.helper_functions as hf
import sec1_data_preparation as data_prep
import sec2_prepped_data_import as prepped_data_import

# Main class
class SubjectGrouping(ABC):
    def subject_selection_steps(self):
        self._pandas_output_setting()
        self.run_data_preparation()
        self.import_processed_main_data()
        self.initial_subject_pool()
        self.create_index_date()
        self.create_filter_tags()
        self.create_master_df()
        self._done_alert()      

    def _pandas_output_setting(self):
        '''Set pandas output display setting'''
        pd.set_option('display.max_rows', 500)
        pd.set_option('display.max_columns', 500)
        pd.set_option('display.width', 180)

    @abstractmethod
    def run_data_preparation(self):
        '''Run data_preparation_steps from base class'''
        pass

    @abstractmethod
    def import_processed_main_data(self):
        '''Import processed main data'''
        pass

    @abstractmethod
    def initial_subject_pool(self):
        '''Import processed main data'''
        pass

    @abstractmethod
    def create_index_date(self):
        '''Create index date'''
        pass

    @abstractmethod
    def create_filter_tags(self):
        '''Create tags to be used as filters'''
        pass

    @abstractmethod
    def create_master_df(self):
        '''Prepare dfs before filtering'''
        pass

    def _done_alert(self): 
        '''Alert when data processing is complete'''
        if self.control_panel['done_switch']:
            ctypes.windll.user32.MessageBoxA(0, b'Hello there', b'Program done.', 0x1000)

class SubjectGrouping_Project1(SubjectGrouping, data_prep.DataPreparation_Project1):
    def __init__(self):
        # Original df; never overwrite
        self.df_dad_origin = None
        self.df_pc_origin = None
        self.df_nacrs_origin = None
        self.df_pin_origin = None
        self.df_reg_origin = None
        # Working/processed df; can overwrite
        self.df_dad = None
        self.df_pc = None
        self.df_nacrs = None
        self.df_pin = None
        self.df_reg = None
        # Subject group holder
        self.df_validIndexDate = None
        self.df_initialPool = None
        self.df_master = None
        self.df_master_filtered = None
        self.control_panel = {
            'save_file_switch': False, # WARNING: Will overwrite existing files
            'df_subsampling_switch': False,  # WARNING: Only switch to True when testing
            'df_subsampling_n': 50000,
            'random_seed': 888,
            'df_remove_dup_switch': True,
            'parse_date_switch': True,
            'result_printout_switch': True,
            'comp_loc': 'office',
            'show_df_n_switch': False, # To be implemented. Show df length before and after record removal
            'done_switch': False,
            }

    def run_data_preparation(self):
        self.data_preparation_steps()

    def import_processed_main_data(self):
        t_obj = prepped_data_import.PreppedDataImport_Project1()
        t_obj.prepped_data_import_steps()
        df_dict = t_obj.return_all_dfs()
        self.df_dad_origin, self.df_pc_origin, self.df_nacrs_origin, self.df_pin_origin, self.df_reg_origin = (
            df_dict['DF_DAD'], df_dict['DF_PC'], df_dict['DF_NACRS'], df_dict['DF_PIN'], df_dict['DF_REG'])
        self.df_dad, self.df_pc, self.df_nacrs, self.df_pin, self.df_reg = (
            df_dict['DF_DAD'], df_dict['DF_PC'], df_dict['DF_NACRS'], df_dict['DF_PIN'], df_dict['DF_REG'])
        del t_obj, df_dict

    def initial_subject_pool(self, id_var='PATIENT_ID'):
        self.df_initialPool = total_subject_pool(self.df_dad_origin, self.df_pc_origin, 
            self.df_nacrs_origin, self.df_pin_origin, self.df_reg_origin, id_var=id_var)
        self.df_initialPool = self.df_initialPool[[id_var]]

    def create_index_date(self):
        # Age criteria
        self.df_pin = age_addOn_tag(df=self.df_pin)
        # Date range criteria
        self.df_pin = inDateRange_addOn_tag(
            df=self.df_pin, 
            start_date=self.inclusion_start_date, 
            end_date=self.inclusion_end_date
            )
        # Sp. Rx critiera
        self.df_pin = spRx_addOn_tag(
            df=self.df_pin,
            new_var='VALID_DRUG_A',
            rx_var='DRUG_DIN',
            rx_list=self.drug_a_fg_dinCode_list+self.drug_a_sg_dinCode_list
            )
        self.df_pin = spRx_addOn_tag(
            df=self.df_pin,
            new_var='VALID_FG_DRUG_A',
            rx_var='DRUG_DIN',
            rx_list=self.drug_a_fg_dinCode_list
            )
        self.df_pin = spRx_addOn_tag(
            df=self.df_pin,
            new_var='VALID_SG_DRUG_A',
            rx_var='DRUG_DIN',
            rx_list=self.drug_a_sg_dinCode_list
            )
        # Create INDEX_DATE
        self.df_validIndexDate = index_date(df=self.df_pin, rx_tag='VALID_SG_DRUG_A')

    def create_filter_tags(self):
        # Create df with only subjects with VALID_REG_STATUS==1
        self.df_validResidence_is1only = residence_is1only_tag(df=self.df_reg)
        # Create pre and post marks around index date
        surround_dates = {
            'PREINDEX_9YR':[9*365, 'subtract'],
            'PREINDEX_2YR':[2*365, 'subtract'],
            'PREINDEX_1YR':[1*365, 'subtract'],
            'PREINDEX_6MO':[0.5*365, 'subtract'],
            'POSTINDEX_6MO':[0.5*365, 'add'],
            'POSTINDEX_1YR':[1*365, 'add'],
            'POSTINDEX_2YR':[2*365, 'add'],
            }
        self.df_validIndexDate = add_surround_dates(df=self.df_validIndexDate, index_date='INDEX_DATE', 
            date_dict=surround_dates)
        # Create INDEX_AGE, INDEX_SEX, INDEX_RURAL
        self.df_validIndexDate = add_index_vars(df_index=self.df_validIndexDate, df_origin=self.df_pin_origin)
        # For DAD, create df with only subjects with VALID_DX status
        dx_obj1 = DxStatus(
            df=self.df_dad_origin,
            df_id='DAD',
            df_ref=self.df_validIndexDate)
        dx_obj1.dx_identification_steps()
        df_agg_dad_withDxTag = dx_obj1.return_agg_df()
        # For PC, create df with only subjects with VALID_DX status
        dx_obj2 = DxStatus(
            df=self.df_pc_origin,
            df_id='PC',
            df_ref=self.df_validIndexDate)
        dx_obj2.dx_identification_steps()
        df_agg_pc_withDxTag = dx_obj2.return_agg_df()
        # Merge the aggregated DAD and PC data with VALID_DX
        self.df_validDx = merge_dx_status_from_dfs(
            df_agg_dad_withDxTag,
            df_agg_pc_withDxTag,
            id_var='PATIENT_ID')
        # Obtain valid medication status with VALID_RX
        rx_obj1 = RxStatus(df=self.df_pin_origin, df_ref=self.df_validIndexDate)
        rx_obj1.rx_identification_steps()
        rx_filterCommand1 = [
            # Drug A-related
            ['&', 'PREINDEX2YR_N_DRUG_A_TAG', '>=', 0],
            ['&', 'POSTINDEX2YR_N_DRUG_A_TAG', '>=', 0],
            ['&', 'PREINDEX2YR_N_DRUG_A_FG_TAG', '>=', 0],
            ['&', 'POSTINDEX2YR_N_DRUG_A_FG_TAG', '>=', 0],
            ['&', 'PREINDEX2YR_N_DRUG_A_SG_TAG', '==', 0],
            ['&', 'POSTINDEX2YR_N_DRUG_A_SG_TAG', '>=', 2],
            # Control drug-related
            ['&', 'PREINDEX2YR_N_CONTROL_DRUG_TAG', '>=', 0],
            ['&', 'POSTINDEX2YR_N_CONTROL_DRUG_TAG', '>=', 0],
            ['&', 'PREINDEX2YR_N_CONTROL_DRUG_TYPICAL_TAG', '>=', 0],
            ['&', 'POSTINDEX2YR_N_CONTROL_DRUG_TYPICAL_TAG', '>=', 0],
            ['&', 'PREINDEX2YR_N_CONTROL_DRUG_ATYPICAL_TAG', '>=', 0],
            ['&', 'POSTINDEX2YR_N_CONTROL_DRUG_ATYPICAL_TAG', '>=', 0],
            ]
        rx_filterCommand2 = [
            # Drug A-related
            ['&', 'PREINDEX2YR_N_DRUG_A_TAG', '>=', 0],
            ['&', 'POSTINDEX2YR_N_DRUG_A_TAG', '>=', 0],
            ['&', 'PREINDEX2YR_N_DRUG_A_FG_TAG', '>=', 0],
            ['&', 'POSTINDEX2YR_N_DRUG_A_FG_TAG', '>=', 0],
            ['&', 'PREINDEX2YR_N_DRUG_A_SG_TAG', '==', 0],
            ['&', 'POSTINDEX2YR_N_DRUG_A_SG_TAG', '==', 0],
            # Control drug-related
            ['&', 'PREINDEX2YR_N_CONTROL_DRUG_TAG', '>=', 0],
            ['&', 'POSTINDEX2YR_N_CONTROL_DRUG_TAG', '>=', 0],
            ['&', 'PREINDEX2YR_N_CONTROL_DRUG_TYPICAL_TAG', '>=', 0],
            ['&', 'POSTINDEX2YR_N_CONTROL_DRUG_TYPICAL_TAG', '>=', 0],
            ['&', 'PREINDEX2YR_N_CONTROL_DRUG_ATYPICAL_TAG', '==', 0],
            ['&', 'POSTINDEX2YR_N_CONTROL_DRUG_ATYPICAL_TAG', '>=', 2],
            ]
        rx_filterCommand3 = [
            # Drug A-related
            ['&', 'PREINDEX2YR_N_DRUG_A_TAG', '==', 0],
            ['&', 'POSTINDEX2YR_N_DRUG_A_TAG', '>=', 2],
            ['&', 'PREINDEX2YR_N_DRUG_A_FG_TAG', '>=', 0],
            ['&', 'POSTINDEX2YR_N_DRUG_A_FG_TAG', '>=', 0],
            ['&', 'PREINDEX2YR_N_DRUG_A_SG_TAG', '>=', 0],
            ['&', 'POSTINDEX2YR_N_DRUG_A_SG_TAG', '>=', 0],
            # Control drug-related
            ['&', 'PREINDEX2YR_N_CONTROL_DRUG_TAG', '>=', 2],
            ['&', 'POSTINDEX2YR_N_CONTROL_DRUG_TAG', '>=', 0],
            ['&', 'PREINDEX2YR_N_CONTROL_DRUG_TYPICAL_TAG', '>=', 0],
            ['&', 'POSTINDEX2YR_N_CONTROL_DRUG_TYPICAL_TAG', '>=', 0],
            ['&', 'PREINDEX2YR_N_CONTROL_DRUG_ATYPICAL_TAG', '>=', 0],
            ['&', 'POSTINDEX2YR_N_CONTROL_DRUG_ATYPICAL_TAG', '>=', 0],
            ]
        rx_obj1.add_rxTag_to_agg_df(new_var='VALID_RX_STATUS_DRUG_A_SG', filter_command_list=rx_filterCommand1)
        rx_obj1.add_rxTag_to_agg_df(new_var='VALID_RX_STATUS_CONTROL_DRUG_ATYPICAL', filter_command_list=rx_filterCommand2)
        rx_obj1.add_rxTag_to_agg_df(new_var='VALID_RX_STATUS_DRUG_A', filter_command_list=rx_filterCommand3)
        self.df_validRx = rx_obj1.return_agg_df()

    def create_master_df(self):
        self.df_master = self.df_initialPool
        df_dict = {
            'df_validIndexDate':self.df_validIndexDate, 
            'df_validDx':self.df_validDx, 
            'df_validRx':self.df_validRx, 
            'df_validResidence_is1only':self.df_validResidence_is1only,
            }
        for df in df_dict.values():
            self.df_master = self.df_master.merge(df, on='PATIENT_ID', how='left')
        # Convert missing into '0'
        var_list = ['VALID_INDEX_STATUS', 'VALID_DX_STATUS', 'VALID_REG_STATUS', 'VALID_RX_STATUS_DRUG_A', 
            'VALID_RX_STATUS_CONTROL_DRUG_ATYPICAL', 'VALID_RX_STATUS_DRUG_A_SG']
        self.df_master[var_list] = self.df_master[var_list].fillna(0)

    def filter_master_df(self, new_var, filter_command_list):
        self.df_master_filtered = self.df_master.query((''.join([''.join(map(str, x)) for x in 
            filter_command_list])).strip('&'))
        self.df_master_filtered[new_var] = 1
        if len(self.df_master_filtered)==0:
            print('Warning: filter_master_df() resulted no "1" signal.')

    def return_master_df(self) -> object:
        return self.df_master

    def return_filtered_master_df(self) -> object:
        return self.df_master_filtered

# Helper class
######################################################################
class DxStatus():
    def __init__(self, df, df_id, df_ref, id_var='PATIENT_ID'):
        # Assert df_id input is 'DAD', 'NACRS', or 'PC' only
        assert df_id.lower() in ['dad', 'pc'], 'Assertion error: df_id needs to be DAD or PC.'
        # Composition, borrow from another class
        dataPrep_obj = data_prep.DataPreparation_Project1()
        dataPrep_obj.dir_name()
        dataPrep_obj.file_name()
        dataPrep_obj.import_ref_data()
        # Declare class vars
        self.df = df
        self.df_id = df_id
        self.df_ref = df_ref
        self.id_var = id_var
        self.ft_obj = ft.Tagger()
        self._dx_start_date_range = 'PREINDEX_9YR'
        self._dx_end_date_range = 'INDEX_DATE'
        if (self.df_id.lower() == 'dad'):
            self._var_base = 'DXCODE'
            self._repeat_var_start = 1
            self._repeat_var_end = 25
            self._cond_list = dataPrep_obj.dxIcd10Code_list
        elif (self.df_id.lower() == 'pc'):
            self._var_base = 'HLTH_DX_ICD9X_CODE_'
            self._repeat_var_start = 1
            self._repeat_var_end = 3    
            self._cond_list = dataPrep_obj.dxIcd9Code_list
        if (self.df_id.lower() == 'dad'):
            self._dx_event_date = 'ADMIT_DATE'
        elif (self.df_id.lower() == 'pc'):
            self._dx_event_date = 'SE_END_DATE'

    def dx_identification_steps(self):
        self.merge_data()
        self.dx_tag()
        self.date_tag()
        self.merge_tag()

    def merge_data(self):
        '''Remove rows if their id's are not in ref data; merge vars across dfs'''
        self.df = self.df_ref.merge(self.df, on=self.id_var, how='left')

    def dx_tag(self):
        self.df['ICD_TAG'] = self.ft_obj.multi_var_cond_tagger(
            self.df, 
            repeat_var_base_name=self._var_base, 
            repeat_var_start=self._repeat_var_start, 
            repeat_var_end=self._repeat_var_end, 
            cond_list=self._cond_list
            )

    def date_tag(self):
        self.df['DATE_TAG'] = np.where(
            (self.df[self._dx_event_date]>=self.df[self._dx_start_date_range]) 
            & (self.df[self._dx_event_date]<=self.df[self._dx_end_date_range]), 1, 0)

    def merge_tag(self):
        self.df['ICD_N_DATE_TAG'] = self.ft_obj.summing_all_tagger(
            self.df, tag_var_list=['ICD_TAG', 'DATE_TAG'])

    def return_df(self) -> object:
        return self.df

    def return_agg_df(self) -> object:
        if (self.df_id.lower() == 'dad'):
            return return_agg_df_dad_withDxTag(
                df=self.df,
                id_var=self.id_var,
                event_date='ADMIT_DATE',
                tag_var='ICD_N_DATE_TAG'
                )
        elif (self.df_id.lower() == 'pc'):
                return return_agg_df_pc_withDxTag(
                df=self.df,
                id_var=self.id_var,
                event_date='SE_END_DATE',
                tag_var='ICD_N_DATE_TAG'
                )

class RxStatus():
    def __init__(self, df, df_ref, id_var='PATIENT_ID'):
        self.df = df
        self.df_ref = df_ref
        self.id_var = id_var
        self.ft_obj = ft.Tagger()
        # Composition, borrow from another class
        self.dataPrep_obj = data_prep.DataPreparation_Project1()
        self.dataPrep_obj.dir_name()
        self.dataPrep_obj.file_name()
        self.dataPrep_obj.import_ref_data()

    def rx_identification_steps(self):
        self.filter_data()
        self.rx_tag()
        self.date_tag()
        self.merge_tag()
        self.agg_df()

    def filter_data(self, index_status='VALID_INDEX_STATUS'):
        self.df = self.df.merge(self.df_ref, on=self.id_var, how='left')
        self.df = self.df[self.df[index_status]==1]

    def rx_tag(self, din_var='DRUG_DIN', atc_var='SUPP_DRUG_ATC_CODE'):
        self.df['RX_DRUG_A_TAG'] = self.ft_obj.isin_tagger(self.df, din_var, self.dataPrep_obj.drug_a_dinCode_list)
        self.df['RX_DRUG_A_FG_TAG'] = self.ft_obj.isin_tagger(self.df, din_var, self.dataPrep_obj.drug_a_fg_dinCode_list)
        self.df['RX_DRUG_A_SG_TAG'] = self.ft_obj.isin_tagger(self.df, din_var, self.dataPrep_obj.drug_a_sg_dinCode_list)
        self.df['RX_CONTROL_DRUG_EXCLUDE_TAG'] = self.ft_obj.isin_tagger(self.df, din_var, self.dataPrep_obj.controlDrug_dinExclusion_list)
        self.df['RX_CONTROL_DRUG_PRETAG'] = self.ft_obj.isin_tagger(self.df, atc_var, self.dataPrep_obj.controlDrug_atcCode_list)
        self.df['RX_CONTROL_DRUG_TYPICAL_PRETAG'] = self.ft_obj.isin_tagger(self.df, atc_var, self.dataPrep_obj.controlDrug_typical_atcCode_list)
        self.df['RX_CONTROL_DRUG_ATYPICAL_PRETAG'] = self.ft_obj.isin_tagger(self.df, atc_var, self.dataPrep_obj.controlDrug_atypical_atcCode_list)
        self.df['RX_CONTROL_DRUG_TAG'] = np.where((self.df['RX_CONTROL_DRUG_PRETAG']==1)&(self.df['RX_CONTROL_DRUG_EXCLUDE_TAG']!=1), 1, 0)
        self.df['RX_CONTROL_DRUG_TYPICAL_TAG'] = np.where((self.df['RX_CONTROL_DRUG_TYPICAL_PRETAG']==1)
            &(self.df['RX_CONTROL_DRUG_EXCLUDE_TAG']!=1), 1, 0)
        self.df['RX_CONTROL_DRUG_ATYPICAL_TAG'] = np.where((self.df['RX_CONTROL_DRUG_ATYPICAL_PRETAG']==1)
            &(self.df['RX_CONTROL_DRUG_EXCLUDE_TAG']!=1), 1, 0)

    def date_tag(self, index_date='INDEX_DATE', event_date='DSPN_DATE'):
        self.df['IN_PREINDEX_2YR_TAG'] = self.ft_obj.date_range_tagger(self.df, event_date, 
            start_date_range=self.df['PREINDEX_2YR'], end_date_range=self.df[index_date], 
            include_start_date=True, include_end_date=False) # index date not included as pre-index period
        self.df['IN_POSTINDEX_2YR_TAG'] = self.ft_obj.date_range_tagger(self.df, event_date, 
            start_date_range=self.df[index_date], end_date_range=self.df['POSTINDEX_2YR'], 
            include_start_date=True, include_end_date=True) # index date included as post-index period

    def merge_tag(self):
        merge_dict = {
            'PREINDEX2YR_N_DRUG_A_TAG':['IN_PREINDEX_2YR_TAG', 'RX_DRUG_A_TAG'],
            'POSTINDEX2YR_N_DRUG_A_TAG':['IN_POSTINDEX_2YR_TAG', 'RX_DRUG_A_TAG'],
            'PREINDEX2YR_N_DRUG_A_FG_TAG':['IN_PREINDEX_2YR_TAG', 'RX_DRUG_A_FG_TAG'],
            'POSTINDEX2YR_N_DRUG_A_FG_TAG':['IN_POSTINDEX_2YR_TAG', 'RX_DRUG_A_FG_TAG'],
            'PREINDEX2YR_N_DRUG_A_SG_TAG':['IN_PREINDEX_2YR_TAG', 'RX_DRUG_A_SG_TAG'],
            'POSTINDEX2YR_N_DRUG_A_SG_TAG':['IN_POSTINDEX_2YR_TAG', 'RX_DRUG_A_SG_TAG'],
            'PREINDEX2YR_N_CONTROL_DRUG_TAG':['IN_PREINDEX_2YR_TAG', 'RX_CONTROL_DRUG_TAG'],
            'POSTINDEX2YR_N_CONTROL_DRUG_TAG':['IN_POSTINDEX_2YR_TAG', 'RX_CONTROL_DRUG_TAG'],
            'PREINDEX2YR_N_CONTROL_DRUG_TYPICAL_TAG':['IN_PREINDEX_2YR_TAG', 'RX_CONTROL_DRUG_TYPICAL_TAG'],
            'POSTINDEX2YR_N_CONTROL_DRUG_TYPICAL_TAG':['IN_POSTINDEX_2YR_TAG', 'RX_CONTROL_DRUG_TYPICAL_TAG'],
            'PREINDEX2YR_N_CONTROL_DRUG_ATYPICAL_TAG':['IN_PREINDEX_2YR_TAG', 'RX_CONTROL_DRUG_ATYPICAL_TAG'],
            'POSTINDEX2YR_N_CONTROL_DRUG_ATYPICAL_TAG':['IN_POSTINDEX_2YR_TAG', 'RX_CONTROL_DRUG_ATYPICAL_TAG'],
            }
        for key, val in merge_dict.items():
            self.df[key] = self.ft_obj.summing_all_tagger(self.df, tag_var_list=val)

    def agg_df(self, id_var='PATIENT_ID'):
        self.df_agg = pd.pivot_table(self.df, index=[id_var], values=[
            'INDEX_DATE', 'PREINDEX_2YR', 'POSTINDEX_2YR', 
            'PREINDEX2YR_N_DRUG_A_TAG', 'POSTINDEX2YR_N_DRUG_A_TAG', 
            'PREINDEX2YR_N_DRUG_A_FG_TAG', 'POSTINDEX2YR_N_DRUG_A_FG_TAG',
            'PREINDEX2YR_N_DRUG_A_SG_TAG', 'POSTINDEX2YR_N_DRUG_A_SG_TAG', 
            'PREINDEX2YR_N_CONTROL_DRUG_TAG', 'POSTINDEX2YR_N_CONTROL_DRUG_TAG',
            'PREINDEX2YR_N_CONTROL_DRUG_TYPICAL_TAG', 'POSTINDEX2YR_N_CONTROL_DRUG_TYPICAL_TAG',
            'PREINDEX2YR_N_CONTROL_DRUG_ATYPICAL_TAG', 'POSTINDEX2YR_N_CONTROL_DRUG_ATYPICAL_TAG'], aggfunc={
                'INDEX_DATE': 'first',
                'PREINDEX_2YR': 'first', 
                'POSTINDEX_2YR': 'first', 
                'PREINDEX2YR_N_DRUG_A_TAG':np.sum, 
                'POSTINDEX2YR_N_DRUG_A_TAG':np.sum,
                'PREINDEX2YR_N_DRUG_A_FG_TAG':np.sum,
                'POSTINDEX2YR_N_DRUG_A_FG_TAG':np.sum,
                'PREINDEX2YR_N_DRUG_A_SG_TAG':np.sum,
                'POSTINDEX2YR_N_DRUG_A_SG_TAG':np.sum,
                'PREINDEX2YR_N_CONTROL_DRUG_TAG':np.sum,
                'POSTINDEX2YR_N_CONTROL_DRUG_TAG':np.sum,
                'PREINDEX2YR_N_CONTROL_DRUG_TYPICAL_TAG':np.sum, 
                'POSTINDEX2YR_N_CONTROL_DRUG_TYPICAL_TAG':np.sum,
                'PREINDEX2YR_N_CONTROL_DRUG_ATYPICAL_TAG':np.sum, 
                'POSTINDEX2YR_N_CONTROL_DRUG_ATYPICAL_TAG':np.sum
                }
            )
        self.df_agg = pd.DataFrame(self.df_agg.to_records())

    def add_rxTag_to_agg_df(self, new_var, filter_command_list):
        self.df_agg_filtered = self.df_agg.query((''.join([''.join(map(str, x)) for x in 
            filter_command_list])).strip('&'))
        self.df_agg_filtered[new_var] = 1
        if len(self.df_agg_filtered)==0:
            print('Warning: add_rxTag_to_agg_df() resulted no RxTag=1 signal.')
        self.df_agg_filtered = self.df_agg_filtered[[new_var, self.id_var]]
        self.df_agg = self.df_agg.merge(self.df_agg_filtered, on=self.id_var, how='left')

    def return_agg_df(self) -> object:
        return self.df_agg

# Helper functions
######################################################################
def total_subject_pool(*dfs, id_var='PATIENT_ID') -> object:
    '''Return a new df by collecting all the unique values in id_var
    from input df's'''
    t_list = []
    for df in dfs:
        t_list += df[id_var].to_list()
    t_list = list(set(t_list))
    return pd.DataFrame(t_list, columns=[id_var])

def index_date(df, rx_tag, id_var='PATIENT_ID', event_date='DSPN_DATE') -> object:
    '''Return a new df with only subjects that have an existing index date'''
    # Remove if row not having valid age
    df = df[df['VALID_AGE']==1]
    # Sort
    df = df.sort_values([id_var, rx_tag, event_date], ascending=[True, False, True])
    # Aggregrate
    df_agg = aggregate_for_index_date(df, tag_filter_var=rx_tag, 
        pre_index_var=event_date, index=[id_var], values=[event_date, 'DRUG_DIN','SUPP_DRUG_ATC_CODE', 
        'SEX', 'RURAL'], aggfunc={event_date:np.min, 'DRUG_DIN':'first', 'SUPP_DRUG_ATC_CODE':'first', 
        'SEX':'first', 'RURAL':'first'})
    # Assert df_pin_patients_agg is not empty
    assert len(df_agg)>0, 'Assertion error: empty dataframe.'
    # Rename columns
    df_agg = df_agg.rename(columns={'DRUG_DIN':'INDEX_DIN', 'SUPP_DRUG_ATC_CODE':'INDEX_ATC'})
    # Add VALID_INDEX_STATUS==1 to later show these are the satisfied patients
    df_agg['VALID_INDEX_STATUS'] = 1
    # Normalize date (aka: removing hour/min/sec)
    df_agg['INDEX_DATE'] = pd.DatetimeIndex(df_agg['INDEX_DATE']).normalize()
    return df_agg

def add_surround_dates(df, date_dict, index_date='INDEX_DATE') -> object:
    for var, values in date_dict.items():
        df[var] = vc.date_adder(df, index_date, values[0], values[1])
        df[var] = pd.DatetimeIndex(df[var]).normalize()
    return df

def add_index_vars(df_index, df_origin, id_var='PATIENT_ID', index_var='INDEX_DATE', age_var='AGE', 
    sex_var='SEX', rural_var='RURAL') -> object:
    # Adding INDEX_AGE corresponding to INDEX_DATE
    df_temp = df_origin.merge(df_index, on=id_var, how='left')
    df_temp['MATCHED_DATES'] = np.where((df_temp['DSPN_DATE']==df_temp[index_var]), 1, 0)
    df_temp = df_temp[df_temp['MATCHED_DATES']==1]
    df_temp_agg = pd.pivot_table(df_temp, index=[id_var], values=[age_var], aggfunc='first')
    df_temp_agg = df_temp_agg.rename(columns={age_var:'INDEX_AGE'})
    df_temp_agg = pd.DataFrame(df_temp_agg.to_records())
    # Actually adding that to the original df
    df_index = df_index.merge(df_temp_agg, on=id_var, how='left')
    df_index = df_index.rename(columns={sex_var:'INDEX_SEX', rural_var:'INDEX_RURAL'})
    return df_index

def residence_is1only_tag(df, id_var='PATIENT_ID', criteria_year=[2015, 2016]) -> object:
    '''Return a new df tagged with 1 (who fulfilled the valid AB registry criterion). 
    Valid AB residence period - had a valid record in the provincial registry within 
    the inclusion period (Apr 2014 and Mar 2016).'''
    df['FYE_ACTIVE'] = np.where(df['ACTIVE_COVERAGE']==1, df['FYE'], np.nan)
    df_agg = df.groupby(by=id_var).agg({'FYE_ACTIVE':lambda x: list(x)})
    df_agg = df_agg.reset_index()
    df_agg['FYE_ACTIVE'] = df_agg['FYE_ACTIVE'].apply(lambda x: [i for i in x if ~np.isnan(i)]) # remove float nan
    df_agg['FYE_ACTIVE'] = df_agg['FYE_ACTIVE'].apply(lambda x: [int(i) for i in x]) # convert float to int
    df_agg['FYE_NEEDED'] = df_agg.apply(lambda x: criteria_year, axis=1)
    df_agg['VALID_REG_STATUS'] = df_agg.apply(compare_list_elements_btw_vars, axis=1)
    df_agg = df_agg[df_agg['VALID_REG_STATUS']==1]
    return df_agg[[id_var, 'VALID_REG_STATUS']]

def age_addOn_tag(df, age_var='AGE', new_var='VALID_AGE', criteria_age_min=18, criteria_age_max=999) -> object:
    '''Return an updated df tagged with 1 for correct age, 0 otherwise'''
    df[new_var] = np.where((df[age_var]>=criteria_age_min)&(df[age_var]<=criteria_age_max), 1, 0)   
    return df

def inDateRange_addOn_tag(df, start_date, end_date, event_date='DSPN_DATE', new_var='VALID_DATE', 
    include_start_date=True, include_end_date=True) -> object:
    '''Return an updated df tagged with 1 for date within data range, 0 otherwise'''
    ft_obj = ft.Tagger()
    df[new_var] = ft_obj.date_range_tagger(df, event_date, start_date_range=start_date, 
        end_date_range=end_date, include_start_date=include_start_date, include_end_date=include_end_date)
    return df

def spRx_addOn_tag(df, rx_var, new_var, rx_list) -> object:
    '''Return an updated df tagged with 1 who had a match with a sp Rx, 0 otherwise'''
    ft_obj = ft.Tagger()
    df[new_var] = ft_obj.isin_tagger(df, rx_var, rx_list)
    return df

def aggregate_for_index_date(df, tag_filter_var, pre_index_var, index, values, aggfunc, 
    index_var='INDEX_DATE') -> object:
    '''tag_filter_col_name refers to the column to retain patients>=1
    pre_index_col_name refers to the date column index date will be derived from 
    index_col_name refers to the column name of the index date'''
    df_filtered = df[df[tag_filter_var]>=1]
    df_agg = pd.pivot_table(df_filtered, index=index, values=values, 
        aggfunc=aggfunc)
    df_agg = pd.DataFrame(df_agg.to_records())
    df_agg = df_agg.rename(columns={pre_index_var:index_var})
    return df_agg

def compare_list_elements_btw_vars(row):
    if set(row['FYE_NEEDED']).issubset(row['FYE_ACTIVE']): return 1
    else: return 0

def return_agg_df_dad_withDxTag(df, id_var, event_date, tag_var) -> object:
    '''Aggregate by PATIENT_ID; return a df'''
    # Shorter df
    df = df.sort_values([id_var, tag_var, event_date], ascending=[True, 
        False, True])
    # Assign new column
    df_agg = (df.assign(ADMIT_DATE=df[event_date].where(
        df[tag_var].astype(bool)))
        .groupby(id_var)
        .agg({tag_var:'max', event_date:'first'})
        )
    df_agg = df_agg.rename(columns={event_date:'DAD_DX_DATE', 
        tag_var:'DAD_DX_TAG'})
    df_agg = pd.DataFrame(df_agg.to_records())
    return df_agg

def return_agg_df_pc_withDxTag(df, id_var, event_date, tag_var) -> object:
    df = df.sort_values([id_var, tag_var, event_date], ascending=[True, 
        False, True])
    # Aggregate by PATIENT_ID; tag_var will take the maximum value; SE_END_DATE will store the dates into
    # .. a list of dates if tag_var==1
    # Complex code below
    # Assign new column
    df_agg = (df.assign(SE_END_DATE=df[event_date].where(
        df[tag_var].astype(bool)))
        .groupby(id_var)
        .agg({tag_var:'max', event_date:lambda x: x.dropna().tolist()})
        )
    df_agg = pd.DataFrame(df_agg.to_records())
    # The dates in the PC_VISIT_DATE_LIST are date of visits that fulfill 1) dx icd and 2) within 9 years pre-index
    df_agg = df_agg.rename(columns={event_date:'PC_VISIT_DATE_LIST'})
    # Execute the function and reassign to new columns
    df_agg['temp'] = df_agg['PC_VISIT_DATE_LIST'].apply(date_scan_by_year_range)
    new_col_list = ['TOTAL_PC_DX_SIGNAL', 'PC_DX_TAG', 'PC_DX_DATE']    # WARNING: 'TOTAL_PC_DX_SIGNAL' gives strange figure, need to check later on
    for n, col in enumerate(new_col_list):
        df_agg[col] = df_agg['temp'].apply(lambda temp: temp[n])
    # Remove unused col and obj instance
    df_agg = df_agg.drop([tag_var, 'PC_VISIT_DATE_LIST', 'temp'], axis=1)
    return df_agg

def date_scan_by_year_range(date_list, num_of_events=3, num_of_years=3) -> tuple:
    '''return (signal_total, signal_binary, first_signal_date); i.e., if a signal is
    3 events over 3 years, then num_of_events=3 and num_of_years=3, if it is
    2 events over 4 years, then num_of_events=2 and num_of_years=4'''
    date_list.sort()
    signal_total = 0
    signal_binary = 0
    first_signal_date = None
    signal_date_list = []
    for i in range(0, len(date_list)):
        try:
            current_date = date_list[i]
            next_xx_date = date_list[i+(num_of_events-1)]
            days_diff = next_xx_date - current_date
            if days_diff <= datetime.timedelta(days=365*num_of_years):
                signal_total+=1
                signal_date_list.append(current_date)
        except Exception: pass
        # Get a binary signal summary
        if signal_total>=1:
            signal_binary=1
        # Sort signal_date_list and extract the first date
    signal_date_list.sort()
    try:
        first_signal_date = signal_date_list[0]
    except:
        first_signal_date = np.NaN
    return signal_total, signal_binary, first_signal_date

def merge_dx_status_from_dfs(*dfs, id_var) -> object:
    '''Param: *dfs as many dfs as user specified. id_var is the common identifier across dfs.
    Function: 1) Outer merge of all the dfs based on id_var; 2) assian 1 to VALID_DX_STATUS if at least one 
    {VAR}_DX_TAG variables is 1, otherwise 0; 3) assign earliest date from {VAR}_DX_DATE to DX_DATE;
    4) return processed df'''
    df_merged = pd.DataFrame(columns=[id_var])
    for df in dfs:
        df_merged = df_merged.merge(df, on=id_var, how='outer')
    dx_tag_cols = [col for col in df_merged.columns if '_DX_TAG' in col]
    df_merged['DX_TAG_SUM'] = df_merged[dx_tag_cols].sum(axis=1)
    df_merged['VALID_DX_STATUS'] = np.where((df_merged['DX_TAG_SUM']>=1), 1, 0)
    df_date_cols = [col for col in df_merged.columns if '_DX_DATE' in col]
    df_merged['DX_DATE'] = df_merged[df_date_cols].min(axis=1)
    return df_merged

# Execution
######################################################################
if __name__=='__main__':
    subjectGrp_filter_DRUG_A_sg = [
        ['&', 'VALID_INDEX_STATUS', '==', 1],
        ['&', 'VALID_DX_STATUS', '==', 1],
        ['&', 'VALID_REG_STATUS', '==', 1],
        ['&', 'VALID_RX_STATUS_DRUG_A_SG', '==', 1],
        ]

    subjectGrp_filter_CONTROL_DRUG_atypical = [
        ['&', 'VALID_INDEX_STATUS', '==', 1],
        ['&', 'VALID_DX_STATUS', '==', 1],
        ['&', 'VALID_REG_STATUS', '==', 1],
        ['&', 'VALID_RX_STATUS_CONTROL_DRUG_ATYPICAL', '==', 1],
        ]

    subjectGrp_filter_DRUG_A = [
        ['&', 'VALID_INDEX_STATUS', '==', 1],
        ['&', 'VALID_DX_STATUS', '==', 1],
        ['&', 'VALID_REG_STATUS', '==', 1],
        ['&', 'VALID_RX_STATUS_DRUG_A', '==', 1],
        ]

    subjectGrp_obj = SubjectGrouping_Project1()
    subjectGrp_obj.subject_selection_steps()
    subjectGrp_obj.filter_master_df(new_var='FINAL_GROUP_DRUG_A_SG', filter_command_list=subjectGrp_filter_DRUG_A_sg)
    df_final_DRUG_A_sg = subjectGrp_obj.return_filtered_master_df()
    print(len(df_final_DRUG_A_sg))
    print(df_final_DRUG_A_sg.PATIENT_ID.nunique())
    print(df_final_DRUG_A_sg.head())

    subjectGrp_obj.filter_master_df(new_var='FINAL_GROUP_CONTROL_DRUG_ATYPICAL', filter_command_list=subjectGrp_filter_CONTROL_DRUG_atypical)
    df_final_CONTROL_DRUG_atypical = subjectGrp_obj.return_filtered_master_df()
    print(len(df_final_CONTROL_DRUG_atypical))
    print(df_final_CONTROL_DRUG_atypical.PATIENT_ID.nunique())
    print(df_final_CONTROL_DRUG_atypical.head())

    subjectGrp_obj.filter_master_df(new_var='FINAL_GROUP_DRUG_A', filter_command_list=subjectGrp_filter_DRUG_A)
    df_final_DRUG_A = subjectGrp_obj.return_filtered_master_df()
    print(len(df_final_DRUG_A))
    print(df_final_DRUG_A.PATIENT_ID.nunique())
    print(df_final_DRUG_A.head())

I used a lot of new coding techniques (i.e., df.query, inheritance, composition) and design patterns (i.e., template method) that I haven't used extensively before. I wonder if overall I am incorporating them correctly. Any other tips/suggestions on how to improve my code will be greatly appreciated.

\$\endgroup\$
2
\$\begingroup\$

You have put up a lot of code for review1, so my feedback cuts some corners. Maybe other members of the community will jump on.


Style

Since your code base is quite substantial, there are some places where the style is not consistent. I will link to the relevant parts in Python's official Style Guide (PEP8) in the following. Before pointing out some of these spots, I would highly recommend to look into an IDE (e.g. PyCharm, Spyder, Visual Studio Code, Atom, ...) with a built-in style checker (e.g. pylint or flake8, both may also be used standalone). These will help you to avoid those pesky traps and the overall appearance will be more consistent.

Whitespace in Expressions
There are some places like

df_dict = {
    'df_validIndexDate':self.df_validIndexDate, 
    'df_validDx':self.df_validDx, 
    'df_validRx':self.df_validRx, 
    'df_validResidence_is1only':self.df_validResidence_is1only,
    }

where you miss out on spaces between the colons in dictionary definitions. The same goes for lines like signal_total+=1 (should be signal_total += 1), etc.

Names
The official recommendation is to use snake_case for variables and method/function names and CamelCase for classes, and most Python libraries consent with it. You also follow it most of the time, but some variable and function names use a funny mix one may call snake_camelCase, e.g. self.df_validIndexDate or def spRx_addOn_tag(...). IMHO it's best to pick either one and follow it consistently.

Blank lines
Use the power of blank lines to give a more robust visual structure to your code. E.g. it is common to seperate class definitions and top-level functions with two blank lines. You may also use a single blank line within function/method bodies where appropriate to group lines of code.

Imports
This topic is closely related to the previous one. The PEP8 recommendation is to group import by 1) standard library imports, 2) third-library imports, 3) imports from within your module/code-structure. It sometimes also makes sense to use subgroubs, e.g. if you import a lot of third-party libraries with different topics. Applying this principle to your code could look like this:

import sys
import random
import ctypes
import datetime
from abc import ABC, abstractmethod

import numpy as np
import pandas as pd
import pysnooper

import var_creator.var_creator as vc
import feature_tagger.feature_tagger as ft
import data_descriptor.data_descriptor as dd
import data_transformer.data_transformer as dt
import helper_functions.helper_functions as hf
import sec1_data_preparation as data_prep
import sec2_prepped_data_import as prepped_data_import

Side note: Visual Studio Code tells me that the imports of sys, random, and pysnooper (as well as dd, dt, and hf) are actually not used in the code you posted. But that might be an artifact of bringing the code to this site.

Documentation
I'ld say you did a reasonable job here. Most of your methods/functions have a little something of documentation. However, your classes lack any form of real documentation. There are some loose bits speaking of a Main class and a Helper class, but I think you could significantly improve it. Classes can also be documented using the '''docstring style''' and I would highly recommend to do so. Since you are working in the "scientific Python stack", you could also have a look at NumPy's docstring conventions to boost the expressiveness even further.

Code

After picking on the style for a while, lets have a look at some parts of the actual code.

Parenthesis in conditions
This is on the brink between style and code. Usually, you won't find lines like if (self.df_id.lower() == 'dad'): a lot in Python code. Most often it'll just be if self.df_id.lower() == 'dad':, since parenthesis are usually only put around the condition if it spans multiple lines.

Assignments and copies
From what I know about Python and pandas, I would say self.df_master = self.df_initialPool does not create a copy. Instead, you will also modify self.df_initialPool when manipulating self.df_master afterwards. Since you are using merge, and merge will create a copy if not told otherwise, I think you can get away with it at that point. You will have to check if this is acceptable if there are other instances of this "pattern".

Iterating over dictionary
There is

df_dict = {
    'df_validIndexDate': self.df_validIndexDate,
    'df_validDx': self.df_validDx,
    'df_validRx': self.df_validRx,
    'df_validResidence_is1only': self.df_validResidence_is1only,
}
for df in df_dict.values():
    ...

almost at the same spot as the previous one. If I have not missed out on a substantial part of the function, the keys of the dictionary are never actually used. This means, the same effect can be accomplished using a simple list:

df_list = [self.df_validIndexDate, self.df_validDx, 
           self.df_validRx, self.df_validResidence_is1only]
for df in df_list:
    self.df_master = self.df_master.merge(df, on='PATIENT_ID', how='left')

As a bonus, this will guarantee that the order in which these dataframes are merged is preserved also in Python versions prior to 3.6, where dicts where unordered (as far as a I know, this is still not an official language feature and considered an implementation detail). However, from what I can see that should be not an issue here.

Exception handling
Code parts like

try:
    first_signal_date = signal_date_list[0]
except:
    first_signal_date = np.NaN

can have unexpected/unwanted effects. Since you're catching all exceptions here, you may also miss things like wrong variables names or keyboard interrupts. So when catching exceptions, be as specific as possible on what you expect to go wrong. You can even catch multiple exceptions on a single line (just in case you were not aware of this). Using

try:
   ...
except Exception:
    pass

is only a miniscule improvement, since Exception is still quite high up in the exception hierarchy.

Memory management
I just wanted to bring to your attention that using del t_obj, df_dict does not immediately free the memory occupied by those variables. It only decrements the internal reference count. The excact moment when the memory will be freed still depends fully on the garbage collector. See also this SO post on that topic.


Well, that's it for the first round. Maybe other members of the community or future me can give you more detailed feedback regarding the coding techniques you asked about.


1 It may be worth to split those 600+ lines into more files for you to work with.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.