import pandas as pd import swifter import io import ast from typing import Optional from src.application.services.merchant_risk.risk_engine import RiskEngine from src.core.logger import with_cloudwatch_logging class ProcessGrossRiskUseCase: def __init__(self, risk_engine: RiskEngine): self.risk_engine = risk_engine @with_cloudwatch_logging(stream_name="merchant-risk") def execute( self, master_data_content: bytes, lender_data_content: bytes, lender_name_change_content: bytes, adjustments_data_content: bytes ) -> bytes: # 1. Load Master Data df_master = pd.read_csv( io.BytesIO(master_data_content), parse_dates=['analysis_date', 'disbursement_date', 'first_emi_date', 'loan_close_date', 'last_emi_date'], dayfirst=True ) # Force conversion to ensure they are Timestamps, not strings or ints for col in ['analysis_date', 'disbursement_date', 'first_emi_date', 'loan_close_date', 'last_emi_date']: if col in df_master.columns: df_master[col] = pd.to_datetime(df_master[col], errors='coerce') # 2. Clean numeric columns columns_for_float = [ 'application_id', 'emi_amount', 'approved_amount', 'upfront_from_customer', 'emi_amount_expected', 'net_disbursement_amount', 'loan_principal', 'total_amount_paid_by_customer', 'amount_paid_for_emi', 'pre_part_payment_amount', 'irr_rate' ] for column in columns_for_float: if column in df_master.columns: # Handle commas and convert to numeric cleaned_series = df_master[column].astype(str).str.replace(',', '').replace('nan', '0') if column == 'application_id': df_master[column] = pd.to_numeric(cleaned_series, errors='coerce').fillna(0).astype('int64') else: df_master[column] = pd.to_numeric(cleaned_series, errors='coerce').fillna(0).astype(float) # 3. Parse emi_date_wise_amount_list def parse_emi_list(val): if pd.isna(val) or val == '': return {} try: if isinstance(val, str): return ast.literal_eval(val) return val except: return {} df_master['emi_date_wise_amount_list'] = df_master['emi_date_wise_amount_list'].apply(parse_emi_list) # 4. Perform POS Calculation df_non_null = df_master.dropna(subset=['emi_date_wise_amount_list']).copy() df_non_null = df_non_null[df_non_null['emi_date_wise_amount_list'].map(len) > 0].copy() if not df_non_null.empty: res = df_non_null.swifter.apply(self.risk_engine.pos_calculation, axis=1) df_non_null[['total_interest_paid', 'total_principal_paid', 'principal_outstanding', 'count_of_emi_paid']] = res df_non_null["principal_outstanding"] = df_non_null["principal_outstanding"].round().clip(lower=0) df_non_null['first_unpaid_emi_date'] = df_non_null.swifter.apply(self.risk_engine.calculate_first_unpaid_date, axis=1) df_nulls = df_master[~df_master.index.isin(df_non_null.index)].copy() for col in ['total_interest_paid', 'total_principal_paid', 'principal_outstanding', 'count_of_emi_paid']: df_nulls[col] = 0 df_nulls['first_unpaid_emi_date'] = pd.NaT df_master = pd.concat([df_non_null, df_nulls], ignore_index=True) # Ensure calculated and existing date columns are Timestamps for col in ['analysis_date', 'first_emi_date', 'first_unpaid_emi_date']: if col in df_master.columns: df_master[col] = pd.to_datetime(df_master[col], errors='coerce') # 5. Initial DPD Calculation df_master['current_dpd'] = 0 mask_dpd = ( (df_master['principal_outstanding'] > 0) & (df_master['first_emi_date'] < df_master['analysis_date']) & (~df_master['application_status'].isin(['closed', 'cancelled'])) & (pd.notnull(df_master['first_unpaid_emi_date'])) & (df_master['first_unpaid_emi_date'] < df_master['analysis_date']) ) df_master.loc[mask_dpd, 'current_dpd'] = ( pd.to_datetime(df_master.loc[mask_dpd, 'analysis_date']) - pd.to_datetime(df_master.loc[mask_dpd, 'first_unpaid_emi_date']) ).dt.days # 6. Apply sequence-sensitive overrides # 6a. Lender CSV Updates (Ported notebook cell 12-13) lender_df = None if lender_data_content: lender_df = pd.read_csv(io.BytesIO(lender_data_content)) df_master = self.risk_engine.apply_lender_overrides(df_master, lender_df) # 6b. Date Cleansing (Ported notebook cell 14 workaround for Analysis Date) df_master = self.risk_engine.clean_unpaid_dates(df_master) # 6c. Lender Name Changes (Ported notebook cell 15) if lender_name_change_content: lender_change_df = pd.read_excel(io.BytesIO(lender_name_change_content)) df_master = self.risk_engine.apply_lender_name_changes(df_master, lender_change_df) # 6d. Settlements (Ported notebook cell 17 - CRITICAL: Must happen before final POS zerofill) fraud_ids = None if adjustments_data_content: settlement_df = pd.read_excel(io.BytesIO(adjustments_data_content), sheet_name=1) df_master = self.risk_engine.handle_settlements(df_master, settlement_df, dpd_value=365) fraud_df = pd.read_excel(io.BytesIO(adjustments_data_content), sheet_name=2) fraud_ids = fraud_df['application_id'] # 6e. Fraud (Ported notebook cell 23) df_master = self.risk_engine.handle_fraud(df_master, fraud_ids) # 7. Final POS Cleansing & Buckets (Ported notebook cells 19, 20, 21, 24) df_master = self.risk_engine.apply_final_cleansing(df_master) # 8. Write-off Analytics (Ported notebook cell 25) df_master = self.risk_engine.calculate_write_off(df_master, fraud_ids) # 8. Reorder columns to match original original_cols = [ 'application_id', 'analysis_date', 'applicant_name', 'current_address_state', 'bureau_score', 'employee_status', 'monthly_income', 'merchant_name', 'anchor_merchant', 'merchant_category', 'type_of_customer', 'revised_type_of_customer', 'main_category', 'lender_name', 'application_status', 'disbursement_date', 'disbursement_type', 'partial_disbursement_status', 'first_emi_date', 'loan_close_date', 'last_emi_date', 'emi_amount', 'interest_rate', 'processing_fee', 'loan_amount', 'approved_amount', 'upfront_from_customer', 'remaining_emi', 'subvention', 'charge_subvention_on_net_loan_amount', 'irr_rate', 'hold_amount', 'net_disbursement_amount', 'loan_principal', 'emi_amount_expected', 'total_amount_paid_by_customer', 'amount_paid_for_emi', 'pre_part_payment_amount', 'emi_date_wise_amount_list', 'adjustment_type', 'date_of_adjustment', 'amt_to_be_adjusted', 'total_interest_paid', 'total_principal_paid', 'principal_outstanding', 'count_of_emi_paid', 'first_unpaid_emi_date', 'current_dpd', 'previous_month_dpd', 'current_month_revised_dpd', 'dpd_buckets', 'writeback_amount', 'write_off_flag', 'write_off_date', 'pre_emi_due', 'emi_due', 'underwriter' ] # Filter existing columns and keep others at the end cols_to_use = [c for c in original_cols if c in df_master.columns] other_cols = [c for c in df_master.columns if c not in original_cols] df_master = df_master[cols_to_use + other_cols] # 9. Return as CSV bytes output = io.BytesIO() df_master.to_csv(output, index=False) return output.getvalue()