import pandas as pd import duckdb import os import numpy as np from datetime import datetime from io import BytesIO from typing import Optional, Union, List import numpy_financial as npf from src.core.logger import with_cloudwatch_logging class ProcessPortfolioCutsUseCase: 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, jify_state_content: bytes, jify_score_content: bytes, task_id: str = "" ): """ Executes the portfolio cuts generation. Consolidates logic from portfolio_cuts, offbook_portfolio_cuts, and solar_portfolio_cuts. """ # 1. Load Primary Data df = pd.read_csv(BytesIO(after_adj_content), parse_dates=['analysis_date', 'disbursement_date'], dayfirst=True) # Global filter matching all portfolio cut notebooks: remove settled df = df[df['application_status'] != 'settled'].reset_index(drop=True) # Prepare analysis month (e.g., Jan_2026) if df.empty: return [] id_suffix = f"_{task_id}" if task_id else "" try: analysis_date = df['analysis_date'].iloc[0] analysis_month = analysis_date.strftime('%b_%Y') except: analysis_month = datetime.now().strftime('%b_%Y') analysis_month = f"{analysis_month}{id_suffix}" # 2. Pre-processing: Common logic for all cuts # Standardize columns and types float_cols = ['bureau_score', 'remaining_emi', 'emi_amount', 'net_disbursement_amount', 'hold_amount', 'principal_outstanding'] for col in float_cols: df[col] = pd.to_numeric(df[col], errors='coerce').fillna(0).astype('float64') if 'monthly_income' in df.columns: df['monthly_income'] = pd.to_numeric(df[df['monthly_income'].notna()]['monthly_income'].astype(str).str.replace(',', ''), errors='coerce').fillna(0) df['monthly_income'] = df['monthly_income'].fillna(0) # Calculate Annualised IRR df['annualised_irr_rate'] = df.apply( lambda row: 0.175/12 if row['merchant_name'] == 'Klub Works Private Limited' else npf.rate(row['remaining_emi'], row['emi_amount'], -row['net_disbursement_amount'], -row['hold_amount']) * 12, axis=1 ) # 3. Load External Data fintech_pos_df = pd.read_excel(BytesIO(fintech_content), sheet_name=0) # Rename for easier access fintech_pos_df = fintech_pos_df.rename(columns={'principal_outstanding_amount': 'principal_outstanding', 'outstanding_count': 'Number of Loans'}) jify_state_df = None jify_score_df = None if jify_state_content: jify_state_df = pd.read_excel(BytesIO(jify_state_content)) # Clean numeric columns in Jify (handle commas) for col in ['Number of Loans', 'Total Principal Outstanding (In Crs)', 'Principal without overdues (In Crs)']: if col in jify_state_df.columns: jify_state_df[col] = pd.to_numeric(jify_state_df[col].astype(str).str.replace(',', ''), errors='coerce').fillna(0) # Enforce strict columns to prevent merge pollution j_cols = ['State', 'Number of Loans', 'Total Principal Outstanding (In Crs)', 'Principal without overdues (In Crs)'] jify_state_df = jify_state_df[[c for c in j_cols if c in jify_state_df.columns]] if jify_score_content: jify_score_df = pd.read_excel(BytesIO(jify_score_content)) for col in ['Number of Loans', 'Total Principal Outstanding (In Crs)', 'Principal without overdues (In Crs)']: if col in jify_score_df.columns: jify_score_df[col] = pd.to_numeric(jify_score_df[col].astype(str).str.replace(',', ''), errors='coerce').fillna(0) # Enforce strict columns j_cols = ['Bureau Score', 'Number of Loans', 'Total Principal Outstanding (In Crs)', 'Principal without overdues (In Crs)'] jify_score_df = jify_score_df[[c for c in j_cols if c in jify_score_df.columns]] j_sums = None if jify_state_df is not None or jify_score_df is not None: # Pre-calculate j_sums for use in Category and Priority Sector cuts if jify_state_df is not None: j_sums = jify_state_df[['Number of Loans', 'Total Principal Outstanding (In Crs)', 'Principal without overdues (In Crs)']].sum() elif jify_score_df is not None: j_sums = jify_score_df[['Number of Loans', 'Total Principal Outstanding (In Crs)', 'Principal without overdues (In Crs)']].sum() ew_fintech = fintech_pos_df[fintech_pos_df['merchant_category'] == 'Earned Wage'] if not ew_fintech.empty and j_sums is not None: f_count = ew_fintech['Number of Loans'].sum() f_pos = ew_fintech['principal_outstanding'].sum() # Align State-wise if jify_state_df is not None: jify_state_df.at[0, 'Number of Loans'] += (f_count - jify_state_df['Number of Loans'].sum()) jify_state_df.at[0, 'Total Principal Outstanding (In Crs)'] += (f_pos - jify_state_df['Total Principal Outstanding (In Crs)'].sum()) jify_state_df.at[0, 'Principal without overdues (In Crs)'] += (f_pos - jify_state_df['Principal without overdues (In Crs)'].sum()) # Align Score-wise if jify_score_df is not None: jify_score_df.at[1, 'Number of Loans'] += (f_count - jify_score_df['Number of Loans'].sum()) jify_score_df.at[1, 'Total Principal Outstanding (In Crs)'] += (f_pos - jify_score_df['Total Principal Outstanding (In Crs)'].sum()) jify_score_df.at[1, 'Principal without overdues (In Crs)'] += (f_pos - jify_score_df['Principal without overdues (In Crs)'].sum()) # Re-calculate j_sums after alignment if jify_state_df is not None: j_sums = jify_state_df[['Number of Loans', 'Total Principal Outstanding (In Crs)', 'Principal without overdues (In Crs)']].sum() # 4. Generate Reports output_files = [] # --- GLOBAL CUTS --- (Logic from portfolio_cuts.ipynb) # Notebook excludes Fundly but INCLUDES KM Global for the main portfolio cuts global_cuts_df = df[df['merchant_name'] != 'Fundly'] global_cuts_file = self._generate_consolidated_cuts( global_cuts_df, analysis_month, "portfolio_cuts", fintech_pos_df, j_sums, jify_state_df, jify_score_df ) output_files.append(global_cuts_file) # Export Global Raw Data as requested global_raw_file = os.path.join(self.output_folder, f"portfolio_cuts_raw_data_{analysis_month}.csv") global_cuts_df.to_csv(global_raw_file, index=False) output_files.append(global_raw_file) # --- OFF-BOOK CUTS --- (Logic from offbook_portfolio_cuts.ipynb) offbook_df = df[(df['merchant_name'] != 'Fundly') & (df['lender_name'] != 'K. M. Global Credit Private Limited')] # Apply Lender-specific POS adjustments offbook_df = self._apply_offbook_adjustments(offbook_df) offbook_cuts_file = self._generate_consolidated_cuts( offbook_df, analysis_month, "offbook_portfolio_cuts", fintech_pos_df, is_offbook=True ) output_files.append(offbook_cuts_file) # --- SOLAR CUTS --- (Logic from solar_portfolio_cuts.ipynb) solar_df = df[(df['merchant_category'] == 'Solar and Rooftop Solutions') & (df['application_status'] != 'settled')] # Export Solar Raw Data as requested solar_raw_file = os.path.join(self.output_folder, f"solar_portfolio_cuts_raw_data_{analysis_month}.csv") solar_df.to_csv(solar_raw_file, index=False) output_files.append(solar_raw_file) solar_cuts_files = self._generate_solar_cuts(solar_df, analysis_month, fintech_pos_df) output_files.extend(solar_cuts_files) return output_files COL_ORDER = ["Number of Loans", "Total Principal Outstanding (In Crs)", "Principal without overdues (In Crs)", "No of cases in arrears", "No of loans written off", "Total PAR (In Crs)", "1-30", "31-60", "61-90", "91-120", "121-150", "151-180", "181-364", "Greater than 364"] def _generate_consolidated_cuts( self, df, analysis_month, report_base_name, fintech_pos_df, j_sums: Optional[pd.Series] = None, jify_state_df: Optional[pd.DataFrame] = None, jify_score_df: Optional[pd.DataFrame] = None, is_offbook: bool = False, is_solar: bool = False ): """Generates the standard 10 cuts in a single Excel file.""" df = self._apply_common_buckets(df) # Prepare fintech data for this report if is_offbook: # offbook_portfolio_cuts.ipynb only uses Fundly offbook f_report_df = fintech_pos_df[fintech_pos_df['merchant_category'] == 'Fundly offbook'] else: # portfolio_cuts.ipynb adds ALL fintechs (loc[1:] for states, sum() for Counts) # We use the full fintech_pos_df for global f_report_df = fintech_pos_df cut_list = [] # Geography geo_cut = self._run_cut_query(df, 'state_buckets', 'State') if is_offbook: # Merging logic for offbook: only Fundly offbook geo_cut.loc[geo_cut['State'] == 'Others', 'Number of Loans'] += f_report_df['Number of Loans'].sum() geo_cut.loc[geo_cut['State'] == 'Others', 'Total Principal Outstanding (In Crs)'] += f_report_df['principal_outstanding'].sum() geo_cut.loc[geo_cut['State'] == 'Others', 'Principal without overdues (In Crs)'] += f_report_df['principal_outstanding'].sum() else: # Merging logic for global: Jify State and select Fintech categories if jify_state_df is not None: geo_cut = pd.merge(geo_cut, jify_state_df, on='State', how='left').fillna(0) geo_cut['Number of Loans'] = geo_cut['Number of Loans_x'] + geo_cut['Number of Loans_y'] geo_cut['Total Principal Outstanding (In Crs)'] = geo_cut['Total Principal Outstanding (In Crs)_x'] + geo_cut['Total Principal Outstanding (In Crs)_y'] geo_cut['Principal without overdues (In Crs)'] = geo_cut['Principal without overdues (In Crs)_x'] + geo_cut['Principal without overdues (In Crs)_y'] geo_cut = geo_cut.drop(columns=[col for col in geo_cut.columns if '_x' in col or '_y' in col]) # Add other fintechs to Others state (notebook adds fintech_df to others) # We exclude Earned Wage from this addition because it's already in jify_state_df fintech_others = f_report_df[f_report_df['merchant_category'] != 'Earned Wage'] geo_cut.loc[geo_cut['State'] == 'Others', 'Number of Loans'] += fintech_others['Number of Loans'].sum() geo_cut.loc[geo_cut['State'] == 'Others', 'Total Principal Outstanding (In Crs)'] += fintech_others['principal_outstanding'].sum() geo_cut.loc[geo_cut['State'] == 'Others', 'Principal without overdues (In Crs)'] += fintech_others['principal_outstanding'].sum() # User requested sorting State by Total Principal Outstanding for both geo_cut = geo_cut.sort_values(by="Total Principal Outstanding (In Crs)", ascending=False).reset_index(drop=True) geo_cut = geo_cut[[geo_cut.columns[0]] + self.COL_ORDER] geo_cut = self._add_total_and_scale(geo_cut) cut_list.append(("Geography", geo_cut)) # Tenure tenure_cut = self._run_cut_query(df, 'tenure_buckets', 'Tenure') # Add all relevant fintechs to '1) 0 to 6 months' tenure_cut.loc[tenure_cut['Tenure'] == '1) 0 to 6 months', 'Number of Loans'] += f_report_df['Number of Loans'].sum() tenure_cut.loc[tenure_cut['Tenure'] == '1) 0 to 6 months', 'Total Principal Outstanding (In Crs)'] += f_report_df['principal_outstanding'].sum() tenure_cut.loc[tenure_cut['Tenure'] == '1) 0 to 6 months', 'Principal without overdues (In Crs)'] += f_report_df['principal_outstanding'].sum() tenure_cut = tenure_cut[[tenure_cut.columns[0]] + self.COL_ORDER] tenure_cut = self._add_total_and_scale(tenure_cut) cut_list.append(("Tenure", tenure_cut)) # Ticket Size ticket_cut = self._run_cut_query(df, 'ticket_size_buckets', 'Loan Ticket size') if is_offbook: ticket_cut.loc[ticket_cut['Loan Ticket size'] == '1) ₹ 0 to 50k', 'Number of Loans'] += f_report_df['Number of Loans'].sum() ticket_cut.loc[ticket_cut['Loan Ticket size'] == '1) ₹ 0 to 50k', 'Total Principal Outstanding (In Crs)'] += f_report_df['principal_outstanding'].sum() ticket_cut.loc[ticket_cut['Loan Ticket size'] == '1) ₹ 0 to 50k', 'Principal without overdues (In Crs)'] += f_report_df['principal_outstanding'].sum() else: # Solar CF to 500k-1000k, others to 0-50k scf = f_report_df[f_report_df['merchant_category'] == 'Solar Channel Financing'] ticket_cut.loc[ticket_cut['Loan Ticket size'] == '7) ₹ 500k to 1000k', 'Number of Loans'] += scf['Number of Loans'].sum() ticket_cut.loc[ticket_cut['Loan Ticket size'] == '7) ₹ 500k to 1000k', 'Total Principal Outstanding (In Crs)'] += scf['principal_outstanding'].sum() ticket_cut.loc[ticket_cut['Loan Ticket size'] == '7) ₹ 500k to 1000k', 'Principal without overdues (In Crs)'] += scf['principal_outstanding'].sum() others_f = f_report_df[f_report_df['merchant_category'] != 'Solar Channel Financing'] ticket_cut.loc[ticket_cut['Loan Ticket size'] == '1) ₹ 0 to 50k', 'Number of Loans'] += others_f['Number of Loans'].sum() ticket_cut.loc[ticket_cut['Loan Ticket size'] == '1) ₹ 0 to 50k', 'Total Principal Outstanding (In Crs)'] += others_f['principal_outstanding'].sum() ticket_cut.loc[ticket_cut['Loan Ticket size'] == '1) ₹ 0 to 50k', 'Principal without overdues (In Crs)'] += others_f['principal_outstanding'].sum() ticket_cut = ticket_cut[[ticket_cut.columns[0]] + self.COL_ORDER] ticket_cut = self._add_total_and_scale(ticket_cut) cut_list.append(("Ticket Size", ticket_cut)) # Category if not is_solar: cat_cut = self._run_cut_query(df, 'main_category', 'Category') # Add Jify as new row (Global only) if not is_offbook and j_sums is not None: j_row_dict = { cat_cut.columns[0]: 'Earned Wage', 'Number of Loans': j_sums.get('Number of Loans', 0), 'Total Principal Outstanding (In Crs)': j_sums.get('Total Principal Outstanding (In Crs)', 0), 'Principal without overdues (In Crs)': j_sums.get('Principal without overdues (In Crs)', 0) } j_row = pd.DataFrame([j_row_dict], columns=cat_cut.columns).fillna(0) cat_cut = pd.concat([cat_cut, j_row], ignore_index=True) # Map fintechs to categories cat_map = {'Healthcare Channel Financing': 'Healthcare', 'Solar Channel Financing': 'Solar/GE'} for f_cat, target in cat_map.items(): f_data = f_report_df[f_report_df['merchant_category'] == f_cat] if not f_data.empty: cat_cut.loc[cat_cut['Category'] == target, 'Number of Loans'] += f_data['Number of Loans'].sum() cat_cut.loc[cat_cut['Category'] == target, 'Total Principal Outstanding (In Crs)'] += f_data['principal_outstanding'].sum() cat_cut.loc[cat_cut['Category'] == target, 'Principal without overdues (In Crs)'] += f_data['principal_outstanding'].sum() # Fundly to RBF fundly = f_report_df[f_report_df['merchant_category'].str.contains('Fundly', na=False)] if not fundly.empty: cat_cut.loc[cat_cut['Category'] == 'Revenue Based Financing', 'Number of Loans'] += fundly['Number of Loans'].sum() cat_cut.loc[cat_cut['Category'] == 'Revenue Based Financing', 'Total Principal Outstanding (In Crs)'] += fundly['principal_outstanding'].sum() cat_cut.loc[cat_cut['Category'] == 'Revenue Based Financing', 'Principal without overdues (In Crs)'] += fundly['principal_outstanding'].sum() cat_cut = cat_cut.sort_values(by="Total Principal Outstanding (In Crs)", ascending=False).reset_index(drop=True) cat_cut = cat_cut[[cat_cut.columns[0]] + self.COL_ORDER] cat_cut = self._add_total_and_scale(cat_cut) cut_list.append(("Category", cat_cut)) # Priority Sector (Main Category) if not is_solar: ps_cut = self._run_cut_query(df, 'priority_sector_category', 'Main Category') # Add Jify (Global only) if not is_offbook and j_sums is not None: j_row_dict = { ps_cut.columns[0]: 'Earned Wage', 'Number of Loans': j_sums.get('Number of Loans', 0), 'Total Principal Outstanding (In Crs)': j_sums.get('Total Principal Outstanding (In Crs)', 0), 'Principal without overdues (In Crs)': j_sums.get('Principal without overdues (In Crs)', 0) } j_row = pd.DataFrame([j_row_dict], columns=ps_cut.columns).fillna(0) ps_cut = pd.concat([ps_cut, j_row], ignore_index=True) # Map fintechs for f_cat, target in {'Healthcare Channel Financing': 'Consumer(Home, Health, Oth)', 'Solar Channel Financing': 'Solar/GE'}.items(): f_data = f_report_df[f_report_df['merchant_category'] == f_cat] if not f_data.empty: ps_cut.loc[ps_cut['Main Category'] == target, 'Number of Loans'] += f_data['Number of Loans'].sum() ps_cut.loc[ps_cut['Main Category'] == target, 'Total Principal Outstanding (In Crs)'] += f_data['principal_outstanding'].sum() ps_cut.loc[ps_cut['Main Category'] == target, 'Principal without overdues (In Crs)'] += f_data['principal_outstanding'].sum() # Fundly to RBF fundly = f_report_df[f_report_df['merchant_category'].str.contains('Fundly', na=False)] if not fundly.empty: ps_cut.loc[ps_cut['Main Category'] == 'Revenue Based Financing', 'Number of Loans'] += fundly['Number of Loans'].sum() ps_cut.loc[ps_cut['Main Category'] == 'Revenue Based Financing', 'Total Principal Outstanding (In Crs)'] += fundly['principal_outstanding'].sum() ps_cut.loc[ps_cut['Main Category'] == 'Revenue Based Financing', 'Principal without overdues (In Crs)'] += fundly['principal_outstanding'].sum() ps_cut = ps_cut.sort_values(by="Total Principal Outstanding (In Crs)", ascending=False).reset_index(drop=True) ps_cut = ps_cut[[ps_cut.columns[0]] + self.COL_ORDER] ps_cut = self._add_total_and_scale(ps_cut) cut_list.append(("Priority Sector", ps_cut)) # IRR (Match notebook logic) irr_cut = self._run_cut_query(df, 'irr_buckets', 'IRR') # Add relevant fintechs to '1) Up to 16%' irr_cut.loc[irr_cut['IRR'] == '1) Up to 16%', 'Number of Loans'] += f_report_df['Number of Loans'].sum() irr_cut.loc[irr_cut['IRR'] == '1) Up to 16%', 'Total Principal Outstanding (In Crs)'] += f_report_df['principal_outstanding'].sum() irr_cut.loc[irr_cut['IRR'] == '1) Up to 16%', 'Principal without overdues (In Crs)'] += f_report_df['principal_outstanding'].sum() irr_cut = irr_cut.sort_values(by="IRR").reset_index(drop=True) irr_cut = irr_cut[[irr_cut.columns[0]] + self.COL_ORDER] irr_cut = self._add_total_and_scale(irr_cut) cut_list.append(("IRR", irr_cut)) # Bureau Score score_cut = self._run_cut_query(df, 'bureau_score_buckets', 'Bureau Score') if not is_offbook: # Global notebook adds ALL fintechs to 'Score NA' # First merge Jify data if jify_score_df is not None: score_cut = pd.merge(score_cut, jify_score_df, on='Bureau Score', how='left').fillna(0) score_cut['Number of Loans'] = score_cut['Number of Loans_x'] + score_cut['Number of Loans_y'] score_cut['Total Principal Outstanding (In Crs)'] = score_cut['Total Principal Outstanding (In Crs)_x'] + score_cut['Total Principal Outstanding (In Crs)_y'] score_cut['Principal without overdues (In Crs)'] = score_cut['Principal without overdues (In Crs)_x'] + score_cut['Principal without overdues (In Crs)_y'] score_cut = score_cut.drop(columns=[col for col in score_cut.columns if '_x' in col or '_y' in col]) # Then add OTHER fintechs (excluding Earned Wage which is in Jify) to Score NA f_others = f_report_df[f_report_df['merchant_category'] != 'Earned Wage'] score_cut.loc[score_cut['Bureau Score'] == 'Score NA', 'Number of Loans'] += f_others['Number of Loans'].sum() score_cut.loc[score_cut['Bureau Score'] == 'Score NA', 'Total Principal Outstanding (In Crs)'] += f_others['principal_outstanding'].sum() score_cut.loc[score_cut['Bureau Score'] == 'Score NA', 'Principal without overdues (In Crs)'] += f_others['principal_outstanding'].sum() else: # Offbook adds all relevant to Score NA score_cut.loc[score_cut['Bureau Score'] == 'Score NA', 'Number of Loans'] += f_report_df['Number of Loans'].sum() score_cut.loc[score_cut['Bureau Score'] == 'Score NA', 'Total Principal Outstanding (In Crs)'] += f_report_df['principal_outstanding'].sum() score_cut.loc[score_cut['Bureau Score'] == 'Score NA', 'Principal without overdues (In Crs)'] += f_report_df['principal_outstanding'].sum() score_cut = score_cut.sort_values(by="Bureau Score").reset_index(drop=True) score_cut = score_cut[[score_cut.columns[0]] + self.COL_ORDER] score_cut = self._add_total_and_scale(score_cut) cut_list.append(("Bureau Score", score_cut)) # Employee Status emp_cut = self._run_cut_query(df, 'employment_status_buckets', 'Customer Profile') # Notebook adds ALL fintech counts to 'Salaried' (even though jify is handles EW) emp_cut.loc[emp_cut['Customer Profile'] == 'Salaried', 'Number of Loans'] += f_report_df['Number of Loans'].sum() emp_cut.loc[emp_cut['Customer Profile'] == 'Salaried', 'Total Principal Outstanding (In Crs)'] += f_report_df['principal_outstanding'].sum() emp_cut.loc[emp_cut['Customer Profile'] == 'Salaried', 'Principal without overdues (In Crs)'] += f_report_df['principal_outstanding'].sum() emp_cut = emp_cut[[emp_cut.columns[0]] + self.COL_ORDER] emp_cut = self._add_total_and_scale(emp_cut) cut_list.append(("Employee Status", emp_cut)) # Monthly Income inc_cut = self._run_cut_query(df, 'monthly_income_buckets', 'Customer Income') # Notebook adds ALL fintech counts to '1) Below 15k' target_inc = '1) Below 15k' inc_cut.loc[inc_cut['Customer Income'] == target_inc, 'Number of Loans'] += f_report_df['Number of Loans'].sum() inc_cut.loc[inc_cut['Customer Income'] == target_inc, 'Total Principal Outstanding (In Crs)'] += f_report_df['principal_outstanding'].sum() inc_cut.loc[inc_cut['Customer Income'] == target_inc, 'Principal without overdues (In Crs)'] += f_report_df['principal_outstanding'].sum() inc_cut = inc_cut[[inc_cut.columns[0]] + self.COL_ORDER] inc_cut = self._add_total_and_scale(inc_cut) cut_list.append(("Monthly Income", inc_cut)) # Write to Excel output_path = os.path.join(self.output_folder, f"{report_base_name}_{analysis_month}.xlsx") # Standard separator for 15 columns (Label + 14 numeric) separator = pd.DataFrame([[''] * 15], columns=geo_cut.columns) # Determine category label for Counts sheet counts_label = 'All' if 'offbook' in report_base_name.lower(): counts_label = 'All' elif 'solar_overall' in report_base_name.lower(): counts_label = 'Solar' elif 'msme' in report_base_name.lower(): counts_label = 'MSME' elif 'rural' in report_base_name.lower(): counts_label = 'Solar Rural' # consistent with notebook logic elif 'residential' in report_base_name.lower(): counts_label = 'Solar Residential' counts_df = self._generate_counts_sheet(df, is_offbook, fintech_pos_df, counts_label, is_solar) with pd.ExcelWriter(output_path, engine='xlsxwriter') as writer: curr_row = 0 for name, cut_df in cut_list: cut_df.to_excel(writer, sheet_name='Sheet1', startrow=curr_row, index=False) curr_row += len(cut_df) + 1 separator.to_excel(writer, sheet_name='Sheet1', startrow=curr_row, index=False, header=False) curr_row += 2 # Add Counts sheet counts_df.to_excel(writer, sheet_name='Counts', index=False) return output_path def _apply_offbook_adjustments(self, df): df = df.copy() adjustments = { 'Singularity Creditworld Private Limited': 0.8, 'Mufin DA June 24': 0.9, 'Capri Global-DA': 0.9, 'HLF-PTC': 0.875, 'Avanse DA Solar': 0.9 } for lender, mult in adjustments.items(): df.loc[df['lender_name'] == lender, 'principal_outstanding'] *= mult return df def _apply_common_buckets(self, df): df = df.copy() # State buckets other_states = ['Himachal Pradesh', 'Chhattisgarh', 'Meghalaya', 'Andaman & Nicobar Islands', 'Jammu & Kashmir', 'Tripura', 'telangana ', 'Nagaland', 'Manipur', 'Sikkim', 'Chandigarh', 'Arunachal Pradesh', 'Goa', 'Pondicherry/Puducherry', 'Mizoram', 'Daman & Diu', 'Lakshadweep', 'Dadra & Nagar Haveli', 'Odisha', 'Jammu & amp; Kashmir', 'Jammu and Kashmir', 'Uttarakhand', 'Daman and Diu'] df['state_buckets'] = df['current_address_state'].apply(lambda x: 'Others' if (x in other_states or pd.isna(x)) else x) # Tenure buckets tenure_bins = [-1, 6, 12, 24, 36, 60, float('inf')] tenure_labels = ['1) 0 to 6 months', '2) 7 to 12 months', '3) 13 to 24 months', '4) 25 to 36 months', '5) 37 to 60 months', '6) Greater than 60 months'] df['tenure_buckets'] = pd.cut(df['remaining_emi'], bins=tenure_bins, labels=tenure_labels, right=True) # Ticket size buckets ts_bins = [0, 50000, 100000, 200000, 300000, 400000, 500000, 1000000, 2500000, 5000000, float('inf')] ts_labels = ['1) ₹ 0 to 50k', '2) ₹ 50k to 100k', '3) ₹ 100k to 200k', '4) ₹ 200k to 300k', '5) ₹ 300k to 400k', '6) ₹ 400k to 500k', '7) ₹ 500k to 1000k', '8) ₹ 1000k to 2500k', '9) ₹ 2500k to 5000k', '10) Greater than ₹ 5000k'] df['ticket_size_buckets'] = pd.cut(df['net_disbursement_amount'], bins=ts_bins, labels=ts_labels, right=True).fillna('1) ₹ 0 to 50k') # IRR buckets irr_bins = [-float('inf'), 0.16, 0.18, 0.20, 0.24, 0.30, float('inf')] irr_labels = ['1) Up to 16%', '2) 16% to 18%', '3) 18% to 20%', '4) 20% to 24%', '5) 24% to 30%', '6) 30% and higher'] df['irr_buckets'] = pd.cut(df['annualised_irr_rate'], bins=irr_bins, labels=irr_labels, right=True).fillna('1) Up to 16%') # Bureau Score buckets score_bins = [-1, 10, 300, 500, 600, 650, 700, 750, 800, float('inf')] score_labels = ['1) -1 to 10', '11 to 300', '2) 301 to 500', '3) 501 to 600', '4) 601 to 650', '5) 651 to 700', '6) 701 to 750', '7) 751 to 800', '8) Greater than 800'] df['bureau_score_buckets'] = pd.cut(df['bureau_score'], bins=score_bins, labels=score_labels, right=True).cat.add_categories('Score NA').fillna('Score NA') # Employee Status buckets emp_map = {'Salaried': ['SALARIED', 'PENSIONER', 'UNEMPLOYED'], 'Self Employed': ['BUSINESS', 'HOUSEWIFE', 'Self Employed - Business', 'Self Employed - Professional']} def map_emp(x): for k, v in emp_map.items(): if x in v: return k return 'Self Employed' # default for NaN or others as per notebook df['employment_status_buckets'] = df['employee_status'].apply(map_emp) # Monthly Income buckets inc_bins = [-float('inf'), 15000, 30000, 60000, float('inf')] inc_labels = ['1) Below 15k', '2) 15k to 30k', '3) 30k to 60k', '4) 60k and higher'] df['monthly_income_buckets'] = pd.cut(df['monthly_income'], bins=inc_bins, labels=inc_labels, right=True) # Priority Sector Category mapping (from notebook) ps_map = { 'Solar/GE' : ['Solar and Rooftop Solutions', 'Electronic Battery', 'Electronic Vehicles'], 'Consumer(Home, Health, Oth)' : ['Cosmetics', 'Dental Care', 'Equipment Financing', 'IVF and Fertility Treatments', 'Insurance Premium Financing', 'Multispeciality Treatments', 'Stemcell banking and treatment','Home Downpayment', 'Modular Kitchen, wardrobes and Interior', 'Painting Services', 'Water Purifier', 'Electronic Products', 'Loose Furniture', 'Personal Loan', 'Jewelry'], 'Education' : ['Education', 'Entrance prep', 'K12 Coaching', 'K12 Edtech', 'K12 Education', 'LSP - Green Energy & Education', 'Upskilling Edtech'], 'Revenue Based Financing' : ['Revenue Based Financing'] } def map_ps(cat): for target, source_list in ps_map.items(): if cat in source_list: return target return cat df['priority_sector_category'] = df['merchant_category'].apply(map_ps) # Ensure main_category exists (fallback to merchant_category if missing) if 'main_category' not in df.columns: df['main_category'] = df['merchant_category'] return df def _run_cut_query(self, df, group_col, label): qry = f""" SELECT {group_col} as "{label}", count(distinct case when (application_status not in ('closed', 'cancelled', 'settled')) and current_month_revised_dpd < 365 and principal_outstanding > 0 then application_id end) "Number of Loans", sum(case when (current_month_revised_dpd < 365) and (application_status not in ('closed', 'cancelled', 'settled')) then principal_outstanding end) "Total Principal Outstanding (In Crs)", sum(case when current_month_revised_dpd = 0 then principal_outstanding end) "Principal without overdues (In Crs)", count(distinct case when current_month_revised_dpd < 365 and current_month_revised_dpd > 0 and principal_outstanding > 0 then application_id end) "No of cases in arrears", count(distinct case when current_month_revised_dpd > 364 then application_id end) "No of loans written off", sum(case when (write_off_flag = 0) and (current_month_revised_dpd > 0 and current_month_revised_dpd <= 30) then principal_outstanding end) "1-30", sum(case when (write_off_flag = 0) and (current_month_revised_dpd > 30 and current_month_revised_dpd <= 60) then principal_outstanding end) "31-60", sum(case when (write_off_flag = 0) and (current_month_revised_dpd > 60 and current_month_revised_dpd <= 90) then principal_outstanding end) "61-90", sum(case when (write_off_flag = 0) and (current_month_revised_dpd > 90 and current_month_revised_dpd <= 120) then principal_outstanding end) "91-120", sum(case when (write_off_flag = 0) and (current_month_revised_dpd > 120 and current_month_revised_dpd <= 150) then principal_outstanding end) "121-150", sum(case when (write_off_flag = 0) and (current_month_revised_dpd > 150 and current_month_revised_dpd <= 180) then principal_outstanding end) "151-180", sum(case when (write_off_flag = 0) and (current_month_revised_dpd > 180 and current_month_revised_dpd <= 364) then principal_outstanding end) "181-364", 0 "Greater than 364" FROM df GROUP BY 1 """ res = duckdb.query(qry).df() # Fill numeric NaNs with 0, but keep string columns as is for col in res.columns: if res[col].dtype in ['float64', 'int64']: res[col] = res[col].fillna(0) res['Total PAR (In Crs)'] = res[["1-30","31-60","61-90","91-120","121-150","151-180","181-364"]].sum(axis=1) cols = [label, "Number of Loans", "Total Principal Outstanding (In Crs)", "Principal without overdues (In Crs)", "No of cases in arrears", "No of loans written off", "Total PAR (In Crs)", "1-30", "31-60", "61-90", "91-120", "121-150", "151-180", "181-364", "Greater than 364"] # Basic sorting by label as default return res[cols].sort_values(by=label) def _generate_counts_sheet(self, df, is_offbook, fintech_pos_df, category_label='All', is_solar: bool = False): qry = f""" SELECT '{category_label}' as "Category", count(distinct case when (application_status not in ('closed', 'cancelled', 'settled')) and current_month_revised_dpd < 365 and principal_outstanding > 0 then application_id end) "Number of Loans", count(case when (application_status not in ('closed', 'cancelled', 'settled')) and current_month_revised_dpd = 0 and principal_outstanding > 0 then application_id end) "Principal without overdues (In Crs)", count(distinct case when current_month_revised_dpd < 365 and current_month_revised_dpd > 0 and principal_outstanding > 0 then application_id end) "Total PAR (In Crs)", count(distinct case when (write_off_flag = 0) and (current_month_revised_dpd > 0 and current_month_revised_dpd <= 30) then application_id end) "1-30", count(case when (write_off_flag = 0) and (current_month_revised_dpd > 30 and current_month_revised_dpd <= 60) then application_id end) "31-60", count(case when (write_off_flag = 0) and (current_month_revised_dpd > 60 and current_month_revised_dpd <= 90) then application_id end) "61-90", count(case when (write_off_flag = 0) and (current_month_revised_dpd > 90 and current_month_revised_dpd <= 120) then application_id end) "91-120", count(case when (write_off_flag = 0) and (current_month_revised_dpd > 120 and current_month_revised_dpd <= 150) then application_id end) "121-150", count(case when (write_off_flag = 0) and (current_month_revised_dpd > 150 and current_month_revised_dpd <= 180) then application_id end) "151-180", count(case when (write_off_flag = 0) and (current_month_revised_dpd > 180 and current_month_revised_dpd <= 365) then application_id end) "181-364", 0 "Greater than 364" FROM df """ counts = duckdb.query(qry).df().fillna(0) # Add fintech counts if is_solar: # Only add Solar Channel Financing for solar reports f_data = fintech_pos_df[fintech_pos_df['merchant_category'] == 'Solar Channel Financing'] elif is_offbook: # offbook_portfolio_cuts.ipynb excludes others, only Fundly offbook f_data = fintech_pos_df[fintech_pos_df['merchant_category'] == 'Fundly offbook'] else: # portfolio_cuts.ipynb excludes Fundly offbook for global f_data = fintech_pos_df[fintech_pos_df['merchant_category'] != 'Fundly offbook'] f_counts = f_data['Number of Loans'].sum() counts['Number of Loans'] += f_counts counts['Principal without overdues (In Crs)'] += f_counts # Order columns as requested (Point 2) cols = ["Category", "Number of Loans", "Principal without overdues (In Crs)", "Total PAR (In Crs)", "1-30", "31-60", "61-90", "91-120", "121-150", "151-180", "181-364", "Greater than 364"] # Filter available columns final_cols = [c for c in cols if c in counts.columns] return counts[final_cols] def _add_total_and_scale(self, df): if df.empty: return df # Ensure numeric types for all columns except the first one for col in df.columns[1:]: df[col] = pd.to_numeric(df[col], errors='coerce').fillna(0) # Total row totals = df.iloc[:, 1:].sum() totals_row_list = ['Total'] + totals.tolist() # Verify length matches exactly if len(totals_row_list) != len(df.columns): # Fallback to manual construction if mismatch totals_row_list = ['Total'] + [0] * (len(df.columns) - 1) totals_row = pd.Series(totals_row_list, index=df.columns) res = pd.concat([df, totals_row.to_frame().T], ignore_index=True) # Scaling numeric_cols = res.select_dtypes(include=['number']).columns # Only scale POS related columns (index 2-3 and 6-14 in output) pos_cols = ["Total Principal Outstanding (In Crs)", "Principal without overdues (In Crs)", "Total PAR (In Crs)", "1-30", "31-60", "61-90", "91-120", "121-150", "151-180", "181-364"] for col in pos_cols: if col in res.columns: res[col] = pd.to_numeric(res[col], errors='coerce').fillna(0).apply(lambda x: round(x / 1e7, 2)) return res def _generate_solar_cuts(self, df, analysis_month, fintech_pos_df): files = [] scf = fintech_pos_df[fintech_pos_df['merchant_category'] == 'Solar Channel Financing'] segments = { 'residential': df[df['revised_type_of_customer'] == 'Residential'], 'rural': df[df['revised_type_of_customer'] == 'Rural C&I MSME(Attachakki, Oil Expeller etc.)'], 'msme': df[df['revised_type_of_customer'] == 'Urban C&I MSME(Factories, Hospitals, Hotels, Offices, etc.)'], 'solar_overall': df } for name, sub_df in segments.items(): # For MSME and Overall, add SCF merging f_scf = scf if name in ['msme', 'solar_overall'] else pd.DataFrame(columns=fintech_pos_df.columns) path = self._generate_consolidated_cuts(sub_df, analysis_month, f"{name}_portfolio_cuts", f_scf, is_solar=True) files.append(path) return files