import pandas as pd import numpy as np import os import ast from datetime import datetime, date from numba import njit, prange from io import BytesIO from typing import List from pandas.tseries.offsets import MonthEnd from dateutil.relativedelta import relativedelta import xlsxwriter from src.core.logger import with_cloudwatch_logging @njit(parallel=True) def run_vintage_calculation_numba( principals, irrs, total_paid_array, adj_amounts, emi_schedules_amounts, max_periods=24 ): n = len(principals) pos_matrix = np.full((n, max_periods), 0.0) for i in prange(n): loan_principal = principals[i] irr = irrs[i] curr_total_paid = total_paid_array[i] curr_adj = adj_amounts[i] beginning_balance = loan_principal for p in range(max_periods): emi_amount = emi_schedules_amounts[i, p] if emi_amount == 0 and beginning_balance <= 0: break interest_component = beginning_balance * irr if interest_component > emi_amount: interest_component = emi_amount interest_paid = min(curr_total_paid, interest_component) curr_total_paid -= interest_paid principal_component = max(emi_amount - interest_component, 0.0) principal_available = curr_total_paid + curr_adj principal_paid = min(principal_available, principal_component) if principal_paid <= curr_total_paid: curr_total_paid -= principal_paid else: adjusted_draw = principal_paid - curr_total_paid curr_total_paid = 0.0 curr_adj -= adjusted_draw ending_balance = beginning_balance - principal_paid pos_matrix[i, p] = ending_balance beginning_balance = ending_balance if beginning_balance <= 0: break return pos_matrix class ProcessStaticPoolUseCase: def __init__(self): self.output_folder = "temp_reports" if not os.path.exists(self.output_folder): os.makedirs(self.output_folder) @with_cloudwatch_logging(stream_name="merchant-risk") def execute( self, after_adj_content: bytes, fintech_content: bytes, task_id: str = "" ) -> List[str]: # Constants L1 = ['Singularity Creditworld Private Limited', 'Vivriti Capital Limited', 'CAPRI GLOBAL CAPITAL LIMITED'] L2 = [ "Pristyn Care (GHV)", "PRISTYN CARE - REIMBURSEMENT INSURANCE", "Klub Works Private Limited", "Hexa Health (Insurance Reimbursement)", "EasyAspataal (Insurance Reimbursement)", "Ayu Health (Insurance Reimbursement)", "PRISTYN CARE - BNPL", "EasyAspataal - BNPL", "CLAIMBUDDY (Insurance Reimbursement)" ] # 1. Load Data df_raw = pd.read_csv(BytesIO(after_adj_content), low_memory=False) if df_raw.empty: return [] # Robust Date Parsing for col in ['analysis_date', 'first_emi_date', 'first_unpaid_emi_date']: if col in df_raw.columns: df_raw[col] = pd.to_datetime(df_raw[col], errors='coerce') analysis_date = pd.to_datetime(df_raw['analysis_date'].dropna().iloc[0]) if not df_raw['analysis_date'].dropna().empty else pd.Timestamp.now() start_date = pd.Timestamp('2018-12-31') # Rule: Filter out future dates for first_unpaid_emi_date mask_future = (df_raw['first_unpaid_emi_date'].notnull()) & (df_raw['first_unpaid_emi_date'] > analysis_date) df_raw.loc[mask_future, 'first_unpaid_emi_date'] = pd.NaT # Rule: Clip first_unpaid_emi_date to first_emi_date mask_before = (df_raw['first_unpaid_emi_date'].notnull()) & (df_raw['first_emi_date'].notnull()) & (df_raw['first_unpaid_emi_date'] < df_raw['first_emi_date']) df_raw.loc[mask_before, 'first_unpaid_emi_date'] = df_raw.loc[mask_before, 'first_emi_date'] # Preprocessing parity with notebook (df_master for denominators) df_master = df_raw[(df_raw['first_emi_date'] > start_date)].copy() # Numerator df (excludes closed/cancelled) df_active = df_master[~df_master['application_status'].isin(['closed', 'cancelled'])].copy() n = len(df_active) principals = df_active['loan_principal'].values.astype(np.float64) irrs = (df_active['irr_rate'].values / 100.0).astype(np.float64) adj_amounts = df_active['amt_to_be_adjusted'].values.astype(np.float64) pos_outstandings = df_active['principal_outstanding'].values.astype(np.float64) # Overrides logic is_special_lender = df_active['lender_name'].isin(L1) is_special_merchant = df_active['merchant_name'].isin(L2) mask_total_paid = is_special_lender | is_special_merchant total_paid_array = np.where(mask_total_paid, df_active['loan_principal'], df_active['amount_paid_for_emi']).astype(np.float64) mask_dpd_masking = is_special_lender & is_special_merchant f_unpaid_vals = df_active['first_unpaid_emi_date'].mask(mask_dpd_masking, pd.NaT).values.astype('datetime64[ns]') # EMI schedules emi_amounts_matrix = np.zeros((n, 24), dtype=np.float64) emi_dict_col = df_active['emi_date_wise_amount_list'].values first_emi_dates = df_active['first_emi_date'].values.astype('datetime64[ns]') for idx in range(n): try: row_dict = ast.literal_eval(emi_dict_col[idx]) except: row_dict = {} f_emi = pd.Timestamp(first_emi_dates[idx]) if pd.isnull(f_emi): continue sorted_items = sorted(row_dict.items(), key=lambda item: datetime.strptime(item[0], '%d-%m-%Y')) last_amt = 0.0 if sorted_items: last_amt = float(sorted_items[0][1]) for d_str, amt in sorted_items: try: d = datetime.strptime(d_str, '%d-%m-%Y') mob = (d.year - f_emi.year) * 12 + (d.month - f_emi.month) if 0 <= mob < 24: emi_amounts_matrix[idx, mob] = float(amt) except: continue curr_amt = last_amt for p in range(24): if emi_amounts_matrix[idx, p] > 0: curr_amt = emi_amounts_matrix[idx, p] else: emi_amounts_matrix[idx, p] = curr_amt # 2. RUN NUMBA pos_matrix = run_vintage_calculation_numba(principals, irrs, total_paid_array, adj_amounts, emi_amounts_matrix) # 3. Generate Case-wise rows rows = [] app_ids = df_active['application_id'].values lender_names = df_active['lender_name'].values for i in range(n): app_id = app_ids[i] f_unpaid = f_unpaid_vals[i] f_emi = pd.Timestamp(first_emi_dates[i]) in_l1 = lender_names[i] in L1 for p in range(24): pos = pos_matrix[i, p] curr_emi_date = (f_emi + relativedelta(months=p)).date() emi_np_date = np.datetime64(curr_emi_date) if emi_np_date > np.datetime64(analysis_date.date()): break if in_l1: pos = pos_outstandings[i] dpd = 0 if not pd.isnull(f_unpaid): eo_month = pd.Timestamp(curr_emi_date) + MonthEnd(0) if emi_np_date >= f_unpaid: dpd = (eo_month.normalize() - pd.Timestamp(f_unpaid).normalize()).days rows.append([app_id, p + 1, pos, emi_np_date, dpd]) if pos <= 0: break rps_df = pd.DataFrame(rows, columns=['application_id', 'installment_number', 'pos', 'emi_date', 'dpd']) df_active['emi_month_year'] = (pd.to_datetime(df_active['first_emi_date']) + MonthEnd(0)).dt.strftime('%Y-%m-%d') df_master['emi_month_year'] = (pd.to_datetime(df_master['first_emi_date']) + MonthEnd(0)).dt.strftime('%Y-%m-%d') # 4. Segment Logic Preparation fintech_df = pd.read_excel(BytesIO(fintech_content), sheet_name='Sheet2') fintech_df['disbursement_date'] = (pd.to_datetime(fintech_df['disbursement_date']) + MonthEnd(0)).dt.strftime('%Y-%m-%d') # Convert numeric columns explicitly to float for c in ['Earned Wage', 'Fundly', 'Solar Channel Financing', 'Healthcare Channel Financing']: if c in fintech_df.columns: fintech_df[c] = fintech_df[c].apply(lambda x: float(str(x).replace(',', '')) if pd.notnull(x) and str(x).strip() != '' else 0.0) # Ensure rps_df has merged info with consistent date type (String) rps_df = pd.merge(rps_df, df_active[['application_id', 'emi_month_year', 'merchant_category', 'revised_type_of_customer']], on='application_id', how='left') # Segment definitions SEGMENTS = { 'Overall': { 'df_filter': lambda x: x, # all 'fintech_cols': ['Earned Wage', 'Fundly', 'Solar Channel Financing', 'Healthcare Channel Financing'] }, 'Core': { 'df_filter': lambda x: x[~x['merchant_category'].isin(['Revenue Based Financing', 'Personal Loan'])], 'fintech_cols': [] }, 'Solar + SCF': { 'df_filter': lambda x: x[x['merchant_category'] == 'Solar and Rooftop Solutions'], 'fintech_cols': ['Solar Channel Financing'] # SCF }, 'Residential': { 'df_filter': lambda x: x[(x['merchant_category'] == 'Solar and Rooftop Solutions') & (x['revised_type_of_customer'] == 'Residential')], 'fintech_cols': [] }, 'Rural': { 'df_filter': lambda x: x[(x['merchant_category'] == 'Solar and Rooftop Solutions') & (x['revised_type_of_customer'] == 'Rural C&I MSME(Attachakki, Oil Expeller etc.)')], 'fintech_cols': [] }, 'MSME + SCF': { 'df_filter': lambda x: x[(x['merchant_category'] == 'Solar and Rooftop Solutions') & (x['revised_type_of_customer'] == 'Urban C&I MSME(Factories, Hospitals, Hotels, Offices, etc.)')], 'fintech_cols': ['Solar Channel Financing'] } } # Date range for all pivots # Date range for all pivots (Strings) report_start_str = '2019-02-28' emi_max_val = df_master['emi_month_year'].max() if not df_master.empty else report_start_str emi_max_dt = pd.to_datetime(emi_max_val) analysis_dt_val = pd.to_datetime(analysis_date) end_dt = min(emi_max_dt, analysis_dt_val) full_date_idx_str = pd.date_range(report_start_str, end_dt, freq='M').strftime('%Y-%m-%d').tolist() def get_vintages_for_segment(seg_name, dpd_bucket): cfg = SEGMENTS[seg_name] # 1. Denominator (Net Disb) seg_master = cfg['df_filter'](df_master).copy() # Explicit numeric aggregation to avoid date summation errors base_agg_dict = {} for col_val in seg_master['emi_month_year'].unique(): dt_key = str(col_val) subset = seg_master[seg_master['emi_month_year'] == col_val] val_m = 0.0 for v in subset['net_disbursement_amount']: try: val_m += float(str(v).replace(',', '')) except: pass base_agg_dict[dt_key] = val_m base_disb = pd.DataFrame(list(base_agg_dict.items()), columns=['emi_month_year', 'net_disbursement_amount']) if cfg['fintech_cols']: f_agg_dict = {} for d_val in fintech_df['disbursement_date'].unique(): d_key = str(d_val) subset_f = fintech_df[fintech_df['disbursement_date'] == d_val] f_row_sum = 0.0 for col_c in cfg['fintech_cols']: for v_f in subset_f[col_c]: try: f_row_sum += float(str(v_f).replace(',', '')) except: pass f_agg_dict[d_key] = f_row_sum fintech_agg = pd.DataFrame(list(f_agg_dict.items()), columns=['emi_month_year', 'fintech_total']) all_disb = pd.merge(base_disb, fintech_agg, on='emi_month_year', how='outer').fillna(0) # Final safe sum for net_disbursement_amount new_disb_vals = [] for _, row_d in all_disb.iterrows(): b_v = 0.0 f_v = 0.0 try: b_v = float(str(row_d['net_disbursement_amount']).replace(',', '')) except: pass try: f_v = float(str(row_d['fintech_total']).replace(',', '')) except: pass new_disb_vals.append(b_v + f_v) all_disb['net_disbursement_amount'] = new_disb_vals all_disb = all_disb[['emi_month_year', 'net_disbursement_amount']] else: all_disb = base_disb all_disb = all_disb.set_index('emi_month_year').reindex(full_date_idx_str, fill_value=0).reset_index().rename(columns={'index': 'emi_month_year'}) # 2. Numerator (POS) seg_rps = cfg['df_filter'](rps_df).copy() filtered_rps = seg_rps[seg_rps['dpd'] > dpd_bucket].copy() # Explicit numeric aggregation for POS pos_agg_dict = {} for col_v in filtered_rps['emi_month_year'].unique(): dt_key = str(col_v) subset_r = filtered_rps[filtered_rps['emi_month_year'] == col_v] for mob_v in subset_r['installment_number'].unique(): mob_key = int(mob_v) mob_subset = subset_r[subset_r['installment_number'] == mob_v] val_r = 0.0 for v_pos in mob_subset['pos']: try: val_r += float(str(v_pos).replace(',', '')) except: pass if dt_key not in pos_agg_dict: pos_agg_dict[dt_key] = {} pos_agg_dict[dt_key][mob_key] = val_r # Convert dict back to DataFrame rows_abs = [] for d_k, mobs_v in pos_agg_dict.items(): r_abs = {'emi_month_year': d_k} r_abs.update({f'emi_{m_k}': m_v for m_k, m_v in mobs_v.items()}) rows_abs.append(r_abs) piv_abs = pd.DataFrame(rows_abs) if piv_abs.empty: piv_abs = pd.DataFrame(columns=['emi_month_year'] + [f'emi_{i}' for i in range(1, 25)]) else: # Ensure all 24 MOB columns exist for i in range(1, 25): if f'emi_{i}' not in piv_abs.columns: piv_abs[f'emi_{i}'] = 0.0 # Selection should strictly include emi_month_year and emi_1...emi_24 final_ordered_cols = ['emi_month_year'] + [f'emi_{i}' for i in range(1, 25)] piv_abs = piv_abs[final_ordered_cols].fillna(0) piv_abs = pd.merge(all_disb, piv_abs, on='emi_month_year', how='left').fillna(0) piv_abs.set_index('emi_month_year', inplace=True) # 3. Percentage piv_pct = piv_abs.copy() for col in piv_pct.columns: if col != 'net_disbursement_amount': piv_pct[col] = piv_pct[col] / piv_pct['net_disbursement_amount'].replace(0, np.nan) return piv_abs.fillna(0), piv_pct.fillna(0) def post_process_vintage(pivot_df, add_average=True): df_out = pivot_df.reset_index().sort_values('emi_month_year') for idx in df_out.index: raw_val = df_out.loc[idx, 'emi_month_year'] if pd.isnull(raw_val) or raw_val == 'Average': continue emi_month = pd.to_datetime(raw_val) mob_limit = (analysis_date.year - emi_month.year) * 12 + (analysis_date.month - emi_month.month) + 1 for col in df_out.columns: if col.startswith('emi_'): try: mob_num = int(col.split('_')[1]) if mob_num > mob_limit: df_out.loc[idx, col] = np.nan except: continue if add_average: val_cols = [c for c in df_out.columns if c.startswith('emi_')] # Explicit safe mean calculation avg_row = {c: np.nan for c in df_out.columns} avg_row['emi_month_year'] = 'Average' for c in val_cols: v_list = [] for v in df_out[c]: try: f_v = float(v) if pd.notnull(f_v): v_list.append(f_v) except: pass avg_row[c] = np.mean(v_list) if v_list else np.nan # Use simple DataFrame constructor and append vs concat to be safer df_avg = pd.DataFrame([avg_row]) df_out = pd.concat([df_out, df_avg], ignore_index=True) return df_out def format_pct(df_pct): df_f = df_pct.copy() for col in df_f.columns: if col.startswith('emi_'): df_f[col] = df_f[col].apply(lambda x: f"{x*100:.2f}%" if pd.notnull(x) and not isinstance(x, str) else "") return df_f # 5. Build Final Excel id_suffix = f"_{task_id}" if task_id else "" final_file = os.path.join(self.output_folder, f"vintage_curves_{datetime.strftime(analysis_date, '%b_%Y')}{id_suffix}.xlsx") with pd.ExcelWriter(final_file, engine='xlsxwriter') as writer: workbook = writer.book bold = workbook.add_format({'bold': True}) for d_bucket in [90, 60, 30]: sheet_name = f'static_pool_{d_bucket}' worksheet = workbook.add_worksheet(sheet_name) writer.sheets[sheet_name] = worksheet curr_row = 0 for seg_name in SEGMENTS: abs_p, pct_p = get_vintages_for_segment(seg_name, d_bucket) abs_processed = post_process_vintage(abs_p, add_average=False) pct_processed = post_process_vintage(pct_p, add_average=True) pct_final = format_pct(pct_processed) # Titles worksheet.write(curr_row, 0, f'Performance Months (PAR >{d_bucket}+ Abs) - {seg_name} Portfolio', bold) start_col_pct = 26 + 2 worksheet.write(curr_row, start_col_pct, f'Performance Months (PAR >{d_bucket}+ %) - {seg_name} Portfolio', bold) # Write Data abs_processed.to_excel(writer, sheet_name=sheet_name, startrow=curr_row+1, index=False) pct_final.to_excel(writer, sheet_name=sheet_name, startrow=curr_row+1, startcol=start_col_pct, index=False) # Increment row for next segment (data rows + spacing) curr_row += len(abs_processed) + 5 # Case wise pos dpd (Explicit Loop version) audit_dict = {} for _, r_r in rps_df.iterrows(): a_id = r_r['application_id'] inst = int(r_r['installment_number']) if a_id not in audit_dict: audit_dict[a_id] = {} audit_dict[a_id][f'dpd_{inst}'] = r_r['dpd'] audit_dict[a_id][f'pos_{inst}'] = r_r['pos'] audit_rows = [] for a_id, dvals in audit_dict.items(): r_aud = {'application_id': a_id} r_aud.update(dvals) audit_rows.append(r_aud) audit_df = pd.DataFrame(audit_rows) audit_df = pd.merge(audit_df, df_active[['application_id', 'first_emi_date', 'merchant_category', 'revised_type_of_customer']], on='application_id', how='left') # Sort columns dpd_cols = sorted([c for c in audit_df.columns if c.startswith('dpd_')], key=lambda x: int(x.split('_')[1])) pos_cols = sorted([c for c in audit_df.columns if c.startswith('pos_')], key=lambda x: int(x.split('_')[1])) final_cols = ['application_id', 'first_emi_date', 'merchant_category', 'revised_type_of_customer'] + dpd_cols + pos_cols audit_df[final_cols].to_excel(writer, index=False, sheet_name='Case wise pos dpd') return [final_file]