import pandas as pd import matplotlib matplotlib.use('Agg') import matplotlib.pyplot as plt from io import BytesIO import os import fnmatch from datetime import datetime import numpy as np from typing import Optional, Union from src.core.logger import with_cloudwatch_logging class ProcessConsolidatedMISUseCase: def __init__(self): self.output_folder = "temp_reports" # Adjusted for backend context, or should be configurable if not os.path.exists(self.output_folder): os.makedirs(self.output_folder) @with_cloudwatch_logging(stream_name="merchant-risk") def execute(self, fintech_file: bytes, after_adj_content: bytes, task_id: str = ""): """ Executes the consolidated MIS report generation. """ id_suffix = f"_{task_id}" if task_id else "" # 1. Load Data # 1. Load Data df = pd.read_csv(BytesIO(after_adj_content), parse_dates=['analysis_date', 'disbursement_date'], dayfirst=True) df = df[df['merchant_name'] != 'Fundly'] df['disbursement_date'] = pd.to_datetime(df['disbursement_date']) # Standardize Application ID and other types df['application_id'] = pd.to_numeric(df['application_id'].astype(str).str.replace(',', ''), errors='coerce').fillna(0).astype('int64') df['bureau_score'] = pd.to_numeric(df['bureau_score'], errors='coerce').fillna(0) if 'monthly_income' in df.columns: df['monthly_income'] = pd.to_numeric(df['monthly_income'].astype(str).str.replace(',', ''), errors='coerce').fillna(0) # 2. Load Fintech Sheet fintech_disb_df, fintech_disb_df_count, fintech_pos_df = self._load_fintech_data(fintech_file) # Analysis Date for naming if df.empty: analysis_date_str = datetime.now().strftime('%b-%Y') output_file = os.path.join(self.output_folder, f"consolidated_mis_{analysis_date_str}{id_suffix}.xlsx") pd.DataFrame().to_excel(output_file) return output_file, [] analysis_date_str = datetime.strftime(df['analysis_date'].iloc[0], '%b-%Y') analysis_date_full = datetime.strftime(df['analysis_date'].iloc[0], '%Y_%m_%d') # 3. Process Reports (following notebook logic) # AUM Reports category_wise_POS = self._get_category_wise_aum(df, fintech_pos_df) main_category_wise_POS = self._get_main_category_wise_aum(category_wise_POS) merchant_wise_POS = self._get_merchant_wise_aum(df) lender_wise_aum = self._get_lender_wise_aum(df, fintech_pos_df) # Add Total rows to AUM reports as per notebook category_wise_POS = self._add_total_row(category_wise_POS, 'merchant_category') merchant_wise_POS = self._add_total_row(merchant_wise_POS, 'merchant_name') # Main Category wise AUM doesn't get a total row in the notebook but let's add for completeness if needed, # but let's stick to notebook first. # Actually notebook doesn't add total row to Merchant wise or Main Category wise. # Only Category wise AUM gets total row in notebook. lender_type_wise_aum = self._get_lender_type_wise_aum(lender_wise_aum) # Disbursement Reports final_sector_wise_gross_disb_df = self._get_sector_wise_gross_disb(df, fintech_disb_df) sector_wise_disb_count_df = self._get_sector_wise_disb_count(df, fintech_disb_df_count) main_cat_disb_df = self._get_main_category_gross_disb(final_sector_wise_gross_disb_df) main_cat_disb_count_df = self._get_main_category_disb_count(sector_wise_disb_count_df) final_sector_wise_net_disb_df = self._get_sector_wise_net_disb(df, fintech_disb_df) main_cat_net_disb_df = self._get_main_category_net_disb(final_sector_wise_net_disb_df) lender_type_wise_lender_disb_df = self._get_lender_type_wise_disb(df, final_sector_wise_net_disb_df) # 4. Generate Charts portfolio_chart_path = os.path.join(self.output_folder, f"portfolio_mix_{analysis_date_str}{id_suffix}.png") self._generate_portfolio_chart(main_category_wise_POS, portfolio_chart_path, analysis_date_str) onbook_offbook_chart_path = os.path.join(self.output_folder, f"onbook_offbook_{analysis_date_str}{id_suffix}.png") self._generate_onbook_offbook_chart(lender_type_wise_lender_disb_df, onbook_offbook_chart_path, analysis_date_str) # 5. Export to Multi-sheet Excel output_file = os.path.join(self.output_folder, f"consolidated_mis_{analysis_date_str}{id_suffix}.xlsx") # Scaling to Crores for specific sheets as per notebook with pd.ExcelWriter(output_file, engine='xlsxwriter') as writer: self._scale_and_export(category_wise_POS, writer, 'Category_wise_AUM', rename_aum=True) self._scale_and_export(main_category_wise_POS, writer, 'Main_Category_wise_AUM', rename_aum=True) self._scale_and_export(merchant_wise_POS, writer, 'Merchant_wise_AUM', rename_aum=True) self._scale_and_export(final_sector_wise_gross_disb_df, writer, 'Sector_wise_Gross_Disb_Value') self._scale_and_export(final_sector_wise_net_disb_df, writer, 'Sector_wise_Net_Disb_Value') sector_wise_disb_count_df.to_excel(writer, sheet_name='Sector_wise_Disb_Count', index=False) self._scale_and_export(main_cat_disb_df, writer, 'Main_Category_Gross_Disb_Value') main_cat_disb_count_df.to_excel(writer, sheet_name='Main_Category_Disb_Count', index=False) self._scale_and_export(main_cat_net_disb_df, writer, 'Main_Category_Net_Disb_Value') self._scale_and_export(lender_type_wise_lender_disb_df, writer, 'Lender_type_wise_Disburements', scale_cols=['Off-book', 'On-book', 'Total', 'Total Net Disb']) self._scale_and_export(lender_wise_aum, writer, 'Lender_wise_AUM', rename_aum=True) workbook = writer.book # Portoflio Mix Sheet p_mix_sheet = workbook.add_worksheet('Portfolio Mix') p_mix_sheet.insert_image('B2', portfolio_chart_path) # Onbook/Offbook Sheet onoff_sheet = workbook.add_worksheet('Onbook_Offbook_Disb') onoff_sheet.insert_image('B2', onbook_offbook_chart_path) return output_file, [portfolio_chart_path, onbook_offbook_chart_path] def _get_latest_after_adj_report(self): # In a real app, this might be passed as an ID or retrieved from a database # For now, we look in the output folder for the latest CSV search_dirs = ["temp_reports", "../output"] for d in search_dirs: if not os.path.exists(d): continue files = [f for f in os.listdir(d) if fnmatch.fnmatch(f, 'merchant_risk_after_adj*.csv')] if files: files.sort(key=lambda x: os.path.getmtime(os.path.join(d, x)), reverse=True) return os.path.join(d, files[0]) return None def _load_fintech_data(self, fintech_file): # Sheet 1: Gross/Net Disb fintech_disb_df = pd.read_excel(BytesIO(fintech_file), sheet_name=1) fintech_disb_df['disbursement_year_month'] = pd.to_datetime(fintech_disb_df['disbursement_date']).dt.to_period('M') fintech_disb_df = fintech_disb_df[['disbursement_year_month', 'Earned Wage', 'Fundly', 'Solar Channel Financing', 'Healthcare Channel Financing']] # Sheet 2: Disb Count fintech_disb_df_count = pd.read_excel(BytesIO(fintech_file), sheet_name=2) fintech_disb_df_count['disbursement_year_month'] = pd.to_datetime(fintech_disb_df_count['disbursement_date']).dt.to_period('M') fintech_disb_df_count = fintech_disb_df_count[['disbursement_year_month', 'Earned Wage', 'Fundly', 'Solar Channel Financing', 'Healthcare Channel Financing']] # Sheet 0: POS (AUM) fintech_pos_df = pd.read_excel(BytesIO(fintech_file), sheet_name=0) fintech_pos_df = fintech_pos_df[['merchant_category', 'principal_outstanding_amount']] fintech_pos_df = fintech_pos_df.rename(columns={'principal_outstanding_amount': 'principal_outstanding'}) return fintech_disb_df, fintech_disb_df_count, fintech_pos_df def _get_category_wise_aum(self, df, fintech_pos_df): # Exclude Closed, Cancelled, Settled and DPD >= 365 for AUM (as per notebook) df_active = df[(~df['application_status'].isin(['closed', 'cancelled', 'settled'])) & (df['current_month_revised_dpd'] < 365)] merchant_category_wise_POS = df_active.groupby('merchant_category').agg({'principal_outstanding': 'sum'}).reset_index() if not fintech_pos_df.empty: # Merge both as per notebook category_wise_POS = pd.concat([merchant_category_wise_POS, fintech_pos_df[['merchant_category', 'principal_outstanding']]], ignore_index=True) return category_wise_POS return merchant_category_wise_POS def _add_total_row(self, df, label_col): if df.empty: return df total_row = df.sum(numeric_only=True) total_df = pd.DataFrame([total_row]) total_df[label_col] = 'Total' return pd.concat([df, total_df], ignore_index=True) def _get_main_category_wise_aum(self, category_wise_POS): # Mappings as per notebook solar_cats = ['Solar and Rooftop Solutions', 'Electronic Vehicles', 'Electronic Battery', 'Solar Channel Financing'] health_home_cats = ['Cosmetics', 'Dental Care', 'Equipment Financing', 'IVF and Fertility Treatments', 'Insurance Premium Financing', 'Multispeciality Treatments', 'Stemcell banking and treatment', 'Healthcare Channel Financing', 'Home Downpayment', 'Loose Furniture', 'Modular Kitchen, wardrobes and Interior', 'Painting Services', 'Water Purifier', 'Electronic Products', 'Jewelry', 'Personal Loan'] edu_cats = ['Education', 'Entrance prep', 'K12 Coaching', 'K12 Edtech', 'K12 Education', 'LSP - Green Energy & Education', 'Upskilling Edtech'] rbf_cats = ['Revenue Based Financing', 'Fundly'] ew_cats = ['Earned Wage'] res = [] res.append(['Solar and Green Energy', category_wise_POS[category_wise_POS['merchant_category'].isin(solar_cats)]['principal_outstanding'].sum()]) res.append(['Health, Home and Others.', category_wise_POS[category_wise_POS['merchant_category'].isin(health_home_cats)]['principal_outstanding'].sum()]) res.append(['Education - Upskilling', category_wise_POS[category_wise_POS['merchant_category'].isin(edu_cats)]['principal_outstanding'].sum()]) res.append(['RBF/SME (Fintech Partners)', category_wise_POS[category_wise_POS['merchant_category'].isin(rbf_cats)]['principal_outstanding'].sum()]) res.append(['Earned Wage (Jify)', category_wise_POS[category_wise_POS['merchant_category'].isin(ew_cats)]['principal_outstanding'].sum()]) return pd.DataFrame(res, columns=['Main Category', 'principal_outstanding']) def _get_merchant_wise_aum(self, df): df_active = df[(~df['application_status'].isin(['closed', 'cancelled', 'settled'])) & (df['current_month_revised_dpd'] < 365)] return df_active.groupby(['merchant_name', 'merchant_category']).agg({'principal_outstanding': 'sum'}).sort_values(by='principal_outstanding', ascending=False).reset_index() def _get_lender_wise_aum(self, df, fintech_pos_df): df_active = df[(~df['application_status'].isin(['closed', 'cancelled', 'settled'])) & (df['current_month_revised_dpd'] < 365)] lender_pos = df_active.groupby('lender_name').agg({'principal_outstanding': 'sum'}).reset_index() if not fintech_pos_df.empty: fintech_pos_renamed = fintech_pos_df.rename(columns={'merchant_category': 'lender_name'}) lender_pos = pd.concat([lender_pos, fintech_pos_renamed], ignore_index=True) return lender_pos def _get_lender_type_wise_aum(self, lender_wise_aum): on_book_lenders = ['K. M. Global Credit Private Limited', 'Earned Wage', 'Fundly', 'Solar Channel Financing', 'Healthcare Channel Financing'] lender_wise_aum['lender_type'] = lender_wise_aum['lender_name'].apply(lambda x: 'On-book' if x in on_book_lenders else 'Off-book') return lender_wise_aum.groupby('lender_type').agg({'principal_outstanding': 'sum'}).reset_index() def _get_sector_wise_gross_disb(self, df, fintech_disb_df): df['disbursement_year_month'] = df['disbursement_date'].dt.to_period('M') # Use approved_amount as per notebook for gross disb pivot = pd.pivot_table(df, index='disbursement_year_month', columns='merchant_category', values='approved_amount', aggfunc='sum').sort_values(by='disbursement_year_month', ascending=False).fillna(0) pivot.reset_index(inplace=True) if not fintech_disb_df.empty: pivot = pd.merge(pivot, fintech_disb_df, on='disbursement_year_month', how='left').fillna(0) pivot['Total'] = pivot.select_dtypes(include=['number']).sum(axis=1) return pivot def _get_sector_wise_disb_count(self, df, fintech_disb_df_count): df['disbursement_year_month'] = df['disbursement_date'].dt.to_period('M') pivot = pd.pivot_table(df, index='disbursement_year_month', columns='merchant_category', values='application_id', aggfunc='count').sort_values(by='disbursement_year_month', ascending=False).fillna(0) pivot.reset_index(inplace=True) if not fintech_disb_df_count.empty: pivot = pd.merge(pivot, fintech_disb_df_count, on='disbursement_year_month', how='left').fillna(0) pivot['Total'] = pivot.select_dtypes(include=['number']).sum(axis=1) return pivot def _get_main_category_gross_disb(self, sector_df): return self._aggregate_by_main_cat(sector_df) def _get_main_category_disb_count(self, sector_df): return self._aggregate_by_main_cat(sector_df) def _get_sector_wise_net_disb(self, df, fintech_disb_df): # Use net_disbursement_amount as per notebook for net disb col = 'net_disbursement_amount' df['disbursement_year_month'] = df['disbursement_date'].dt.to_period('M') pivot = pd.pivot_table(df, index='disbursement_year_month', columns='merchant_category', values=col, aggfunc='sum').sort_values(by='disbursement_year_month', ascending=False).fillna(0) pivot.reset_index(inplace=True) if not fintech_disb_df.empty: pivot = pd.merge(pivot, fintech_disb_df, on='disbursement_year_month', how='left').fillna(0) pivot['Total'] = pivot.select_dtypes(include=['number']).sum(axis=1) return pivot def _get_main_category_net_disb(self, sector_df): return self._aggregate_by_main_cat(sector_df) def _aggregate_by_main_cat(self, df): if df.empty: return df res = df[['disbursement_year_month']].copy() solar = ['Solar and Rooftop Solutions', 'Solar Channel Financing', 'Electronic Vehicles', 'Electronic Battery'] health = ['Cosmetics', 'Dental Care', 'Equipment Financing', 'IVF and Fertility Treatments', 'Insurance Premium Financing', 'Multispeciality Treatments', 'Stemcell banking and treatment', 'Healthcare Channel Financing', 'Home Downpayment', 'Loose Furniture', 'Modular Kitchen, wardrobes and Interior', 'Painting Services', 'Water Purifier', 'Electronic Products', 'Jewelry', 'Personal Loan'] edu = ['Education', 'Entrance prep', 'K12 Coaching', 'K12 Edtech', 'K12 Education', 'LSP - Green Energy & Education', 'Upskilling Edtech'] rbf = ['Revenue Based Financing', 'Fundly'] ew = ['Earned Wage'] res['Solar and Green Energy'] = df[[c for c in solar if c in df.columns]].sum(axis=1) res['Health, Home and Others.'] = df[[c for c in health if c in df.columns]].sum(axis=1) res['Education - Upskilling'] = df[[c for c in edu if c in df.columns]].sum(axis=1) res['RBF/SME (Fintech Partners)'] = df[[c for c in rbf if c in df.columns]].sum(axis=1) res['Earned Wage (Jify)'] = df[[c for c in ew if c in df.columns]].sum(axis=1) res['Grand Total'] = res.select_dtypes(include=['number']).sum(axis=1) return res def _get_lender_type_wise_disb(self, df, net_disb_df): if net_disb_df.empty: return pd.DataFrame() df['lender_type'] = df['lender_name'].apply(lambda x: 'On-book' if x == 'K. M. Global Credit Private Limited' else 'Off-book') df['disbursement_year_month'] = df['disbursement_date'].dt.to_period('M') # Use approved_amount here as well for consistency with gross disb value pivot = pd.pivot_table(df, index='disbursement_year_month', columns='lender_type', values='approved_amount', aggfunc='sum').sort_values(by='disbursement_year_month', ascending=False).fillna(0) pivot.reset_index(inplace=True) # Ensure 'Off-book' and 'On-book' columns exist if 'Off-book' not in pivot.columns: pivot['Off-book'] = 0 if 'On-book' not in pivot.columns: pivot['On-book'] = 0 # Calculate Total for these two pivot['Total'] = pivot['Off-book'] + pivot['On-book'] # Merge with Total Net Disb from the other report pivot = pd.merge(pivot, net_disb_df[['disbursement_year_month', 'Total']], on='disbursement_year_month', how='left') pivot = pivot.rename(columns={'Total_x': 'Total', 'Total_y': 'Total Net Disb'}) pivot['% Off-book'] = (pivot['Off-book'] / pivot['Total'] * 100).fillna(0) pivot['% On-book'] = (100 - pivot['% Off-book']).fillna(0) return pivot def _generate_portfolio_chart(self, df, path, date_str): if df.empty: return plot_df = df.copy() total = plot_df['principal_outstanding'].sum() if total == 0: return plot_df['% of Portfolio'] = (plot_df['principal_outstanding'] / total) * 100 # Group < 1% into Others as per notebook plot_df['Main Category'] = plot_df.apply( lambda row: "Others" if row['% of Portfolio'] < 1 else row['Main Category'], axis=1 ) df_merged = plot_df.groupby("Main Category", as_index=False).agg({ "principal_outstanding": "sum", "% of Portfolio": "sum" }) # Color mapping as per notebook logic: # Education(Blue), Health/Home(Green), Others(Peach), Solar(Yellow) color_map = { 'Education - Upskilling': '#3333FF', 'Health, Home and Others.': '#58d68d', 'Others': '#f5cba7', 'Solar and Green Energy': '#DFFF00', 'RBF/SME (Fintech Partners)': '#873600', 'Earned Wage (Jify)': '#2E86C1' } # Sort alphabetically to ensure consistent color assignment if map fails df_merged = df_merged.sort_values("Main Category") colors = [color_map.get(cat, '#CCCCCC') for cat in df_merged['Main Category']] plt.figure(figsize=(10, 8)) plt.pie( df_merged['% of Portfolio'], autopct='%1.2f%%', colors=colors, wedgeprops={'width': 0.6}, textprops={'fontsize': 11, 'fontweight': 'bold'} ) plt.title(f'Portfolio Distribution by Main Category - {date_str}', fontweight='bold') plt.legend(df_merged['Main Category'], fontsize=12, ncol=4, loc='upper center') plt.tight_layout() plt.savefig(path) plt.close() def _generate_onbook_offbook_chart(self, df, path, date_str): if df.empty: return # Take latest 12 months and sort ascending for trend plot plot_df = df.head(12).copy() plot_df = plot_df.sort_values(by='disbursement_year_month', ascending=True) labels = plot_df['disbursement_year_month'].astype(str) plt.figure(figsize=(11, 5)) plt.plot(labels, plot_df['% Off-book'], label='% Off-book', color='blue') plt.plot(labels, plot_df['% On-book'], label='% On-book', color='red') # Add labels for i, (idx, row) in enumerate(plot_df.iterrows()): plt.text(i, row['% Off-book'], f"{row['% Off-book']:.2f}%", va='top', fontsize=7) plt.text(i, row['% On-book'], f"{row['% On-book']:.2f}%", va='top', fontsize=7) plt.title(f'% of Disbursement in Off-book & On-book over Time - {date_str}') plt.legend() plt.grid(True, linestyle='--', alpha=0.6) plt.savefig(path) plt.close() def _scale_and_export(self, df, writer, sheet_name, scale_cols=None, rename_aum=False): if df.empty: pd.DataFrame().to_excel(writer, sheet_name=sheet_name) return export_df = df.copy() # Convert Period to string for Excel if 'disbursement_year_month' in export_df.columns: export_df['disbursement_year_month'] = export_df['disbursement_year_month'].astype(str) # Scaling to Cr (10^7) if scale_cols: for col in scale_cols: if col in export_df.columns: export_df[col] = export_df[col] / 10**7 else: # Auto-scale numeric columns for col in export_df.columns: if export_df[col].dtype in ['float64', 'int64']: if 'count' not in col.lower() and 'percent' not in col.lower() and '%' not in col: export_df[col] = export_df[col] / 10**7 if rename_aum and 'principal_outstanding' in export_df.columns: export_df = export_df.rename(columns={'principal_outstanding': 'Principal Outstanding (in Cr.)'}) export_df.to_excel(writer, sheet_name=sheet_name, index=False)