I work in the population health industry and get contracts from commercial companies to conduct research on their products. This is the general code to identify target patient groups from a provincial datasets, including DAD (hospital discharge), PC (physician claims), NACRS (emergency room visit), PIN (drug dispensation), and REG (provincial registry). Same patients can have multiple rows in each of the databases. For example, if a patient was hospitalized 3 times, s/he will show up as three separate rows in DAD data. The code does the followings:
- Import data from csv files into individual Pandas dataframes (df's)
- Then it goes through some initial data cleaning and processing (such as random sampling, date formatting, calling additional reference information (such as icd code for study condition)
- Under the section
1) Identify patients for case defn'n #1
, a series of steps have been done to label (as tags) each of the relevant data and filtering based on these tags. Datasets are linked together to see if a particular patient fulfills the diagnostic code requirement. - Information also needs to be aggregated by unique patient level via the
pivot_table
function to summarize by unique patients - At the end, the final patient dataframe is saved into local directory, and analytic results are printed
- I also made my own modules
feature_tagger
to house some of the more frequently-used functions away from this main code
# Overall steps:
# 1) Patient defintiion: Had a ICD code and a procedure code within a time period
# 2) Output: A list of PHN_ENC of included patients; corresponding index date
# .. 'CaseDefn1_PatientDict_FINAL.txt'
# .. 'CaseDefn1_PatientDf_FINAL.csv'
# 3) Results: Analytic results
# ----------------------------------------------------------------------------------------------------------
import pandas as pd
import datetime
import random
import feature_tagger.feature_tagger as ft
import data_descriptor.data_descriptor as dd
import data_transformer.data_transformer as dt
import var_creator.var_creator as vc
# Unrestrict pandas' output display
pd.set_option('display.max_rows', 500)
pd.set_option('display.max_columns', 500)
pd.set_option('display.width', 120)
# Control panel
save_file_switch = False # WARNING: will overwrite existing when == True
df_subsampling_switch = False # WARNING: make to sure turn off for final results
edge_date_inclusion = True # whether to include the last date in the range of inclusion criteria
testing_printout_switch = False
result_printout_switch = True
done_switch = True
df_subsampling_n = 15000
random_seed = 888
# Instantiate objects
ft_obj = ft.Tagger()
dt_obj = dt.Data_Transformer()
# Import data
loc = 'office'
if loc == 'office':
directory = r'E:\My_Working_Primary\Projects\Data_Analysis\\'
elif loc == 'home':
directory = r'C:\Users\MyStuff\Dropbox\Projects\Data_Analysis\\'
else: pass
refDataDir = r'_Data\RefData\\'
realDataDir = r'_Data\RealData\\'
resultDir = r'_Results\\'
file_dad = 'Prepped_DAD_Data.csv'
file_pc = 'Prepped_PC_Data.csv'
file_nacrs = 'Prepped_NACRS_Data.csv'
file_pin = 'Prepped_PIN_Data.csv'
file_reg = 'Prepped_REG_Data.csv'
df_dad = pd.read_csv(directory+realDataDir+file_dad, dtype={'PHN_ENC': str}, encoding='utf-8', low_memory=False)
df_pc = pd.read_csv(directory+realDataDir+file_pc, dtype={'PHN_ENC': str}, encoding='utf-8', low_memory=False)
df_nacrs = pd.read_csv(directory+realDataDir+file_nacrs, dtype={'PHN_ENC': str}, encoding='utf-8', low_memory=False)
df_pin = pd.read_csv(directory+realDataDir+file_pin, dtype={'PHN_ENC': str}, encoding='utf-8', low_memory=False)
df_reg = pd.read_csv(directory+realDataDir+file_reg, dtype={'PHN_ENC': str}, encoding='utf-8', low_memory=False)
# Create random sampling of df's to run codes faster
if df_subsampling_switch==True:
if (df_subsampling_n>len(df_dad))|(df_subsampling_n>len(df_pc))|(df_subsampling_n>len(df_nacrs))|(df_subsampling_n>len(df_pin)):
print ('Warning: Specified subsample size is larger than the total no. of row of some of the dataset,')
print ('As a result, resampling with replacement will be done to reach specified subsample size.')
df_dad = dt_obj.random_n(df_dad, n=df_subsampling_n, on_switch=df_subsampling_switch, random_state=random_seed)
df_pc = dt_obj.random_n(df_pc, n=df_subsampling_n, on_switch=df_subsampling_switch, random_state=random_seed)
df_nacrs = dt_obj.random_n(df_nacrs, n=df_subsampling_n, on_switch=df_subsampling_switch, random_state=random_seed)
df_pin = dt_obj.random_n(df_pin, n=df_subsampling_n, on_switch=df_subsampling_switch, random_state=random_seed)
# Format variable type
df_dad['ADMIT_DATE'] = pd.to_datetime(df_dad['ADMIT_DATE'], format='%Y-%m-%d')
df_dad['DIS_DATE'] = pd.to_datetime(df_dad['DIS_DATE'], format='%Y-%m-%d')
df_pc['SE_END_DATE'] = pd.to_datetime(df_pc['SE_END_DATE'], format='%Y-%m-%d')
df_pc['SE_START_DATE'] = pd.to_datetime(df_pc['SE_START_DATE'], format='%Y-%m-%d')
df_nacrs['ARRIVE_DATE'] = pd.to_datetime(df_nacrs['ARRIVE_DATE'], format='%Y-%m-%d')
df_pin['DSPN_DATE'] = pd.to_datetime(df_pin['DSPN_DATE'], format='%Y-%m-%d')
df_reg['PERS_REAP_END_RSN_DATE'] = pd.to_datetime(df_reg['PERS_REAP_END_RSN_DATE'], format='%Y-%m-%d')
# Import reference codes
file_rxCode = '_InStudyCodes_ATC&DIN.csv'
file_icdCode = '_InStudyCodes_DxICD.csv'
file_serviceCode = '_InStudyCodes_ServiceCode.csv'
df_rxCode = pd.read_csv(directory+refDataDir+file_rxCode, dtype={'ICD_9': str}, encoding='utf-8', low_memory=False)
df_icdCode = pd.read_csv(directory+refDataDir+file_icdCode, encoding='utf-8', low_memory=False)
df_serviceCode = pd.read_csv(directory+refDataDir+file_serviceCode, encoding='utf-8', low_memory=False)
# Defining study's constant variables
inclusion_start_date = datetime.datetime(2017, 4, 1, 00, 00, 00)
inclusion_end_date = datetime.datetime(2018, 3, 31, 23, 59, 59)
sp_serviceCode_dict = {df_serviceCode['Short_Desc'][0]:df_serviceCode['Health_Service_Code'][0]}
sp_serviceCode_val = sp_serviceCode_dict['ABC injection']
sp_dxCode_dict = {'DIABETES_ICD9': df_icdCode['ICD_9'][0], 'DIABETES_ICD10': df_icdCode['ICD_10'][0]}
sp_dxCode_val_icd9 = sp_dxCode_dict['DIABETES_ICD9']
sp_dxCode_val_icd10 = sp_dxCode_dict['DIABETES_ICD10']
# ----------------------------------------------------------------------------------------------------------
# 1) Identify patients for case def'n #1.
# Step 1 - Aged between 18 and 100 years old on the index date
# Step 2 - Had at least 1 recorded ICD diagnostic code based on physician visit (ICD-9-CA=9999 in PC) or
# hospitalization (ICD-10-CA=G9999 in DAD) during the inclusion period
# Step 3.1 - Had at least 1 specific procedure code (99.999O) during
# the inclusion period (Note: earliest ABC injection code date is the Index date)
# Step 3.2 - Construct index date
# Step 4 - Registered as a valid Alberta resident for 2 years before the index date and 1 year after the
# index date (determined from PR)
# 1.1) Get age at each service, then delete rows with age falling out of 18-100 range
df_dad_ageTrimmed = df_dad.copy()
df_dad_ageTrimmed = df_dad_ageTrimmed[(df_dad_ageTrimmed['AGE']>=18) & (df_dad_ageTrimmed['AGE']<=100)]
df_pc_ageTrimmed = df_pc.copy()
df_pc_ageTrimmed = df_pc_ageTrimmed[(df_pc_ageTrimmed['AGE']>=18) & (df_pc_ageTrimmed['AGE']<=100)]
# 1.2) Tag appropriate date within sp range > tag DIABETES code > combine tags
df_dad_ageTrimmed['DAD_DATE_TAG'] = ft_obj.date_range_tagger(df_dad_ageTrimmed, 'ADMIT_DATE',
start_date_range=inclusion_start_date, end_date_range=inclusion_end_date, edge_date_inclusion=
edge_date_inclusion)
df_dad_ageTrimmed['DAD_ICD_TAG'] = ft_obj.multi_var_cond_tagger(df_dad_ageTrimmed, repeat_var_base_name='DXCODE',
repeat_var_start=1, repeat_var_end=25, cond_list=[sp_dxCode_val_icd10])
df_dad_ageTrimmed['DAD_DATE_ICD_TAG'] = ft_obj.summing_all_tagger(df_dad_ageTrimmed, tag_var_list=['DAD_DATE_TAG',
'DAD_ICD_TAG'])
df_pc_ageTrimmed['PC_DATE_TAG'] = ft_obj.date_range_tagger(df_pc_ageTrimmed, 'SE_END_DATE',
start_date_range=inclusion_start_date, end_date_range=inclusion_end_date, edge_date_inclusion=
edge_date_inclusion)
df_pc_ageTrimmed['PC_ICD_TAG'] = ft_obj.multi_var_cond_tagger(df_pc_ageTrimmed, repeat_var_base_name='HLTH_DX_ICD9X_CODE_',
repeat_var_start=1, repeat_var_end=3, cond_list=[str(sp_dxCode_val_icd9)])
df_pc_ageTrimmed['PC_DATE_ICD_TAG'] = ft_obj.summing_all_tagger(df_pc_ageTrimmed, tag_var_list=['PC_DATE_TAG',
'PC_ICD_TAG'])
# Output a list of all patients PHN_ENC who satisfy the Date and DIABETES code criteria
df_dad_ageDateICDtrimmed = df_dad_ageTrimmed[df_dad_ageTrimmed['DAD_DATE_ICD_TAG']==1]
df_pc_ageDateICDtrimmed = df_pc_ageTrimmed[df_pc_ageTrimmed['PC_DATE_ICD_TAG']==1]
dad_patientList_diabetes_Code = df_dad_ageDateICDtrimmed['PHN_ENC'].unique().tolist()
pc_patientList_diabetes_Code = df_pc_ageDateICDtrimmed['PHN_ENC'].unique().tolist()
dad_pc_patientList_diabetes_Code = list(set(dad_patientList_diabetes_Code)|set(pc_patientList_diabetes_Code))
dad_pc_patientList_diabetes_Code.sort()
# 1.3.1) Tag appropriate date within sp range > tag ABC injection code > combine tags
df_pc_ageTrimmed['PC_PROC_TAG'] = df_pc_ageTrimmed['ABC_INJECT']
df_pc_ageTrimmed['PC_DATE_PROC_TAG'] = ft_obj.summing_all_tagger(df_pc_ageTrimmed, tag_var_list=['PC_DATE_TAG',
'PC_PROC_TAG'])
df_pc_ageDateProcTrimmed = df_pc_ageTrimmed[df_pc_ageTrimmed['PC_DATE_PROC_TAG']==1]
pc_patientList_procCode = df_pc_ageDateProcTrimmed['PHN_ENC'].unique().tolist()
dad_pc_patientList_diabetes_NprocCode = list(set(dad_pc_patientList_diabetes_Code)&set(pc_patientList_procCode))
dad_pc_patientList_diabetes_NprocCode.sort()
# 1.3.2) Find Index date
df_pc_ageDateProcTrimmed_pivot = pd.pivot_table(df_pc_ageDateProcTrimmed, index=['PHN_ENC'],
values=['SE_END_DATE', 'AGE', 'SEX', 'RURAL'], aggfunc={'SE_END_DATE':np.min, 'AGE':np.min,
'SEX':'first', 'RURAL':'first'})
df_pc_ageDateProcTrimmed_pivot = pd.DataFrame(df_pc_ageDateProcTrimmed_pivot.to_records())
df_pc_ageDateProcTrimmed_pivot = df_pc_ageDateProcTrimmed_pivot.rename(columns={'SE_END_DATE':'INDEX_DT'})
# 1.4) Filter by valid registry
# Create a list variable (based on index date) to indicate which fiscal years need to be valid according to
# the required 2 years before index and 1 year after index date, in df_pc_ageDateProcTrimmed_pivot
def extract_needed_fiscal_years(row): # extract 2 years before and 1 year after index date
if int(row['INDEX_DT'].month) >= 4:
index_yr = int(row['INDEX_DT'].year)+1
else:
index_yr = int(row['INDEX_DT'].year)
first_yr = index_yr-2
four_yrs_str = str(first_yr)+','+str(first_yr+1)+','+str(first_yr+2)+','+str(first_yr+3)
return four_yrs_str
df_temp = df_pc_ageDateProcTrimmed_pivot.copy()
df_temp['FYE_NEEDED'] = df_temp.apply(extract_needed_fiscal_years, axis=1)
df_temp['FYE_NEEDED'] = df_temp['FYE_NEEDED'].apply(lambda x: x[0:].split(',')) # from whole string to list of string items
df_temp['FYE_NEEDED'] = df_temp['FYE_NEEDED'].apply(lambda x: [int(i) for i in x]) # from list of string items to list of int items
# Create a list variable to indicate the active fiscal year, in df_reg
df_reg['FYE_ACTIVE'] = np.where(df_reg['ACTIVE_COVERAGE']==1, df_reg['FYE'], np.nan)
df_reg_agg = df_reg.groupby(by='PHN_ENC').agg({'FYE_ACTIVE':lambda x: list(x)})
df_reg_agg = df_reg_agg.reset_index()
df_reg_agg['FYE_ACTIVE'] = df_reg_agg['FYE_ACTIVE'].apply(lambda x: [i for i in x if ~np.isnan(i)]) # remove float nan
df_reg_agg['FYE_ACTIVE'] = df_reg_agg['FYE_ACTIVE'].apply(lambda x: [int(i) for i in x]) # convert float to int
# Merge df's and create tag, if active years do not cover all the required fiscal year, exclude patients
# Create inclusion/exclusion patient list to apply to obtain patient cohort based on case def'n #1
df_temp_v2 = df_temp.merge(df_reg_agg, on='PHN_ENC', how='left')
df_temp_v2_trimmed = df_temp_v2[(df_temp_v2['FYE_NEEDED'].notnull())&(df_temp_v2['FYE_ACTIVE'].notnull())]
# Remove rows with missing on either variables
def compare_list_elements_btw_cols(row):
if set(row['FYE_NEEDED']).issubset(row['FYE_ACTIVE']):
return 1
else:
return 0
df_temp_v2_trimmed['VALID_REG'] = df_temp_v2_trimmed.apply(compare_list_elements_btw_cols, axis=1)
df_temp_v2_trimmed_v2 = df_temp_v2_trimmed[df_temp_v2_trimmed['VALID_REG']==1]
reg_patientList = df_temp_v2_trimmed_v2['PHN_ENC'].unique().tolist()
# Apply inclusion/exclusion patient list (from REG) to find final patients
# Obtain final patient list
df_final_defn1 = df_pc_ageDateProcTrimmed_pivot.merge(df_temp_v2_trimmed_v2, on='PHN_ENC', how='inner')
df_final_defn1 = df_final_defn1[['PHN_ENC', 'AGE_x', 'SEX_x', 'RURAL_x', 'INDEX_DT_x']]
df_final_defn1 = df_final_defn1.rename(columns={'AGE_x':'AGE', 'SEX_x':'SEX', 'RURAL_x':'RURAL', 'INDEX_DT_x':'INDEX_DT',})
df_final_defn1['PREINDEX_1Yr'] = (df_final_defn1['INDEX_DT']-pd.Timedelta(days=364)) # 364 because index date is counted as one pre-index date
df_final_defn1['PREINDEX_2Yr'] = (df_final_defn1['INDEX_DT']-pd.Timedelta(days=729)) # 729 because index date is counted as one pre-index date
df_final_defn1['POSTINDEX_1Yr'] = (df_final_defn1['INDEX_DT']+pd.Timedelta(days=364))
list_final_defn1 = df_final_defn1['PHN_ENC'].unique().tolist()
dict_final_defn1 = {'Final unique patients of case definition #1':list_final_defn1}
# Additional ask (later on)
# How: Create INDEX_DT_FIS_YR (index date fiscal year) by mapping INDEX_DT to fiscal year
def index_date_fiscal_year(row):
if ((row['INDEX_DT'] >= datetime.datetime(2015, 4, 1, 00, 00, 00)) &
(row['INDEX_DT'] < datetime.datetime(2016, 4, 1, 00, 00, 00))):
return '2015/2016'
elif ((row['INDEX_DT'] >= datetime.datetime(2016, 4, 1, 00, 00, 00)) &
(row['INDEX_DT'] < datetime.datetime(2017, 4, 1, 00, 00, 00))):
return '2016/2017'
else:
return 'Potential error'
df_final_defn1['INDEX_DT_FIS_YR'] = df_final_defn1.apply(index_date_fiscal_year, axis=1)
# 2) Output final patient list for future access
# WARNING: will overwrite existing
if save_file_switch == True:
if df_subsampling_switch == True:
f = open(directory+resultDir+'_CaseDefn1_PatientDict_Subsample.txt',"w")
f.write(str(dict_final_defn1)+',')
f.close()
df_final_defn1.to_csv(directory+resultDir+'_CaseDefn1_PatientDf_Subsample.csv', sep=',', encoding='utf-8')
elif df_subsampling_switch == False:
f = open(directory+resultDir+'CaseDefn1_PatientDict_FINAL.txt',"w")
f.write(str(dict_final_defn1)+',')
f.close()
df_final_defn1.to_csv(directory+resultDir+'CaseDefn1_PatientDf_FINAL.csv', sep=',', encoding='utf-8')
# 3) Results: Analytic results
if result_printout_switch == True:
print ('Unique PHN_ENC N, (aged 18 to 100 during inclusion period) from DAD:')
print (df_dad_ageTrimmed['PHN_ENC'].nunique())
print ('Unique PHN_ENC N, (aged 18 to 100 during inclusion period) from PC:')
print (df_pc_ageTrimmed['PHN_ENC'].nunique())
print ('Unique PHN_ENC N, (aged 18 to 100 during inclusion period) from DAD or PC:')
dd_obj = dd.Data_Comparator(df_dad_ageTrimmed, df_pc_ageTrimmed, 'PHN_ENC')
print (dd_obj.unique_n_union())
print ('Unique PHN_ENC N, (aged 18 to 100) and (had DIABETES code during inclusion period) from DAD:')
print (df_dad_ageDateICDtrimmed['PHN_ENC'].nunique())
print ('Unique PHN_ENC N, (aged 18 to 100) and (had DIABETES code during inclusion period) from PC:')
print (df_pc_ageDateICDtrimmed['PHN_ENC'].nunique())
print ('Unique PHN_ENC N, (aged 18 to 100) and (had DIABETES code during inclusion period) from DAD or PC:')
print (len(dad_pc_patientList_diabetes_Code))
print ('Unique PHN_ENC N, (aged 18 to 100) and (had DIABETES code during inclusion period)\
and (had ABC injection code) from DAD and PC:')
print (df_pc_ageDateProcTrimmed_pivot['PHN_ENC'].nunique())
print ('Unique PHN_ENC N, (aged 18 to 1005) and (had DIABETES code during inclusion period)\
and (had ABC injection code) and (had AB resident around index date) from DAD, PC, and REG [Case Def #1]:')
print (df_final_defn1['PHN_ENC'].nunique())
# Additional analytic ask (later on)
print ('Patient N by index date as corresponding fiscal year:')
print (df_final_defn1['INDEX_DT_FIS_YR'].value_counts())
if done_switch == True:
ctypes.windll.user32.MessageBoxA(0, b'Hello there', b'Program done.', 3)
My questions are:
- This is a code for a specific project, other projects from other companies while aren't exactly the same, they usually follow similar overall steps including cleaning data, linking data, creating tags, filtering tags, aggregating data, saving files, and producing data. How can I refactor my code to be maintainable within this specific project, as well as reusable across similar projects?
- Many times, once I have run the code and produce the results, clients may come back to ask for additional follow-up information (i.e., the ones under
# Additional ask (later on)
). How can I deal with additional asks more effectively with maintainability and expandability in mind? - Any areas I can try using some design patterns?
- Any other suggestions on how I can write better python code are more than welcome.
df_subsampling_switch
to sample data for code development. Another characteristic that deems this problem to be big data is that they are linkable, which created a lot more complexity. \$\endgroup\$ – KubiK888 Apr 12 '19 at 19:52pd.set_option('display.max_rows', 500)
. \$\endgroup\$ – 200_success Apr 12 '19 at 19:54