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 ProcessAfterAdjustmentUseCase: def __init__(self, risk_engine: RiskEngine): self.risk_engine = risk_engine @with_cloudwatch_logging(stream_name="merchant-risk") def execute( self, before_adj_content: bytes, adjustments_content: bytes, previous_month_content: bytes, lender_data_content: bytes ) -> bytes: # 1. Load Before Adjustment Data df = pd.read_csv( io.BytesIO(before_adj_content), parse_dates=['analysis_date', 'disbursement_date', 'first_emi_date', 'loan_close_date', 'last_emi_date', 'first_unpaid_emi_date'], dayfirst=True ) # Ensure Timestamps and IDs df['application_id'] = df['application_id'].astype(int) for col in ['analysis_date', 'disbursement_date', 'first_emi_date', 'loan_close_date', 'last_emi_date', 'first_unpaid_emi_date']: if col in df.columns: df[col] = pd.to_datetime(df[col], errors='coerce') # 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['emi_date_wise_amount_list'] = df['emi_date_wise_amount_list'].swifter.apply(parse_emi_list) # 2. Get Merchant Wise Adjustments (Sheet 0) additional_adj_df = pd.read_excel(io.BytesIO(adjustments_content), sheet_name=0) merchant_wise_adj = self.risk_engine.calculate_merchant_adjustments(df, additional_adj_df) # 3. Waterfall Allocation (Sheet 3 exclusions) exclusion_df = pd.read_excel(io.BytesIO(adjustments_content), sheet_name=3) if 'date_of_adjustment' in exclusion_df.columns: exclusion_df['date_of_adjustment'] = pd.to_datetime(exclusion_df['date_of_adjustment'], errors='coerce') df = self.risk_engine.apply_waterfall_allocation(df, merchant_wise_adj, exclusion_df) # 4. Apply manual app-level adjustments (Sheet 3) # Standardizing as int64 for reliable update() matching (handles commas) df['application_id'] = pd.to_numeric(df['application_id'].astype(str).str.replace(',', ''), errors='coerce').fillna(0).astype('int64') exclusion_df['application_id'] = pd.to_numeric(exclusion_df['application_id'].astype(str).str.replace(',', ''), errors='coerce').fillna(0).astype('int64') df.set_index('application_id', inplace=True) exclusion_df.set_index('application_id', inplace=True) df.update(exclusion_df[['date_of_adjustment', 'adjustment_type', 'amt_to_be_adjusted']]) df.reset_index(inplace=True) # 5. Recalculate POS/DPD after adjustments df_non_null = df.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_after_adj, 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) # Re-calculate first unpaid date based on new emi count df_non_null['first_unpaid_emi_date'] = df_non_null.swifter.apply(self.risk_engine.calculate_first_unpaid_date, axis=1) df_non_null['first_unpaid_emi_date'] = pd.to_datetime(df_non_null['first_unpaid_emi_date'], errors='coerce') df_nulls = df[~df.index.isin(df_non_null.index)].copy() df = pd.concat([df_non_null, df_nulls], ignore_index=True) # 5b. Final overrides that affect "Original DPD" (Ported from notebook cells 96-106) # Initial DPD before overrides/renaming df['current_dpd'] = 0 mask_dpd = ( (df['principal_outstanding'] > 0) & (df['first_emi_date'] < df['analysis_date']) & (~df['application_status'].isin(['closed', 'cancelled'])) & (pd.notnull(df['first_unpaid_emi_date'])) & (df['first_unpaid_emi_date'] < df['analysis_date']) ) df.loc[mask_dpd, 'current_dpd'] = ( pd.to_datetime(df.loc[mask_dpd, 'analysis_date']) - pd.to_datetime(df.loc[mask_dpd, 'first_unpaid_emi_date']) ).dt.days # Lender POS-DPD Change lender_df = None if lender_data_content: lender_df = pd.read_csv(io.BytesIO(lender_data_content)) df = self.risk_engine.apply_lender_overrides(df, lender_df) # Settlements & Fraud overrides (Sheet 1 & 2) if adjustments_content: settlement_df = pd.read_excel(io.BytesIO(adjustments_content), sheet_name=1) df = self.risk_engine.handle_settlements(df, settlement_df, dpd_value=0) fraud_df = pd.read_excel(io.BytesIO(adjustments_content), sheet_name=2) df = self.risk_engine.handle_fraud(df, fraud_df['application_id']) # Merchant List full collection & other cleansing df = self.risk_engine.apply_final_cleansing(df) df = self.risk_engine.clean_unpaid_dates(df) # Rename to Original DPD (Notebook cell 107) df.rename(columns={'current_dpd': 'current_month_original_dpd'}, inplace=True) # 6. Rolling DPD & Writeback if previous_month_content: prev_df = pd.read_csv( io.BytesIO(previous_month_content), usecols=['application_id', 'current_month_revised_dpd', 'principal_outstanding', 'write_off_date'], parse_dates=['write_off_date'], dayfirst=True ).rename(columns={ 'current_month_revised_dpd': 'previous_month_dpd', 'principal_outstanding': 'previous_month_pos', 'write_off_date': 'previous_write_off_date' }) prev_df['application_id'] = prev_df['application_id'].astype(int) df = self.risk_engine.calculate_revised_dpd(df, prev_df) df = self.risk_engine.calculate_writeback(df) else: df['previous_month_dpd'] = 0 df['current_month_revised_dpd'] = df['current_month_original_dpd'] df['writeback_amount'] = 0.0 # 7. Final Cleansing & Write-off Analytics df = self.risk_engine.apply_final_cleansing(df) # Determine fraud IDs for write-off analytics (Notebook cell 106) fraud_df = pd.read_excel(io.BytesIO(adjustments_content), sheet_name=2) fraud_ids = fraud_df['application_id'] df = self.risk_engine.calculate_write_off(df, 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_month_original_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.columns] other_cols = [c for c in df.columns if c not in original_cols] df = df[cols_to_use + other_cols] # 9. Return CSV output = io.BytesIO() df.to_csv(output, index=False) return output.getvalue()