import pandas as pd import ast, warnings import numpy as np from datetime import datetime from utils import execute_query, get_cloudwatch_logger from scripts.mysql_queries import allocation_query logger = get_cloudwatch_logger(stream_name="payment_reconcilition") warnings.filterwarnings('ignore') async def function1(data): application_id = data.get('application_id') new_payment_entry = data.get('payment_list') logger.send_log_event(f"INFO: In payment_reconciliation function - Request received: \n{data}") # get & run rps sql qyery which contains due amount and already paid amount for given application_id df = execute_query(allocation_query, params={"application_id": application_id}) df.select_dtypes(include=['number']).fillna(0) if df.empty: return {"error": f"No data found for application_id {application_id}"} # Initialize new/convert existing columns df['emi_date'] = pd.to_datetime(df['emi_date'], format='%d-%m-%Y').dt.date df['due_date'] = pd.to_datetime(df['due_date'], format='%d-%m-%Y').dt.date df['payment_list'] = df['payment_list'].apply(ast.literal_eval) df['payment_date'] = pd.NaT df['payment_details'] = '' df.loc[:, ['bounce_charges', 'delay_days', 'delay_penalty','interest_comp_paid', 'principal_comp_paid']] = 0 df.sort_values(by='due_date',inplace=True) # Extract payment date and amount from new_payment_entry new_payment_entry['payment_date'] = datetime.strptime(new_payment_entry['payment_date'], '%d-%m-%Y').date() new_amount_paid = new_payment_entry['amount'] new_payment_date = new_payment_entry['payment_date'] current_date = pd.Timestamp.today().date() # allocation logic for ind in df.index: if new_amount_paid <= 0: break # first allocate to bpi_interest if df.at[ind, 'total_paid_for_bpi'] < df.at[ind, 'bpi_interest']: remaining_bpi_interest = max(0, df.at[ind, 'bpi_interest'] - df.at[ind, 'total_paid_for_bpi']) allocatable_amount = min(remaining_bpi_interest, new_amount_paid) # allocatable_amount = min(df.at[ind, 'bpi_interest'], new_amount_paid) current_details = df.at[ind, 'payment_details'] new_details = f"{new_payment_date} - {allocatable_amount} - bpi" df.at[ind, 'payment_details'] = (current_details + ', ' if current_details else '') + new_details df.at[ind, 'total_paid'] += allocatable_amount df.at[ind, 'total_paid_for_bpi'] += allocatable_amount df.at[ind, 'amount_due'] -= allocatable_amount new_amount_paid -= allocatable_amount if new_amount_paid <= 0: break # allocate to morat interest if (df.at[ind, 'due_date'] <= current_date) and (df.at[ind, 'total_paid'] < (df.at[ind, 'morat_interest'] + df.at[ind, 'bpi_interest'])): allocatable_amount = min(df.at[ind, 'morat_interest'], new_amount_paid) current_details = df.at[ind, 'payment_details'] new_details = f"{new_payment_date} - {allocatable_amount} - morat_interest" df.at[ind, 'payment_details'] = (current_details + ', ' if current_details else '') + new_details df.at[ind, 'total_paid'] += allocatable_amount df.at[ind, 'amount_due'] -= allocatable_amount new_amount_paid -= allocatable_amount if new_amount_paid <= 0: break # allocate to previous or current unpaid emi if (df.at[ind, 'due_date'] <= current_date) and (df.at[ind, 'total_paid_for_emi'] < df.at[ind, 'emi_amount']): allocatable_amount = min(df.at[ind, 'pending_for_emi'],df.at[ind, 'amount_due'], new_amount_paid) current_details = df.at[ind, 'payment_details'] new_details = f"{new_payment_date} - {allocatable_amount} - emi" df.at[ind, 'payment_details'] = (current_details + ', ' if current_details else '') + new_details df.at[ind, 'total_paid'] += allocatable_amount df.at[ind, 'amount_due'] -= allocatable_amount df.at[ind, 'pending_for_emi'] -= allocatable_amount df.at[ind, 'total_paid_for_emi'] += allocatable_amount new_payment_dict = {datetime.strftime(new_payment_date, '%d-%m-%Y'):allocatable_amount} df.at[ind, 'payment_list'].append(new_payment_dict) if new_payment_date <= current_date else None # append current payment in payment_list new_amount_paid -= allocatable_amount if new_amount_paid <= 0: break # update payment_list def sort_payment_list(payment_list): return sorted(payment_list, key=lambda x: datetime.strptime(list(x.keys())[0], '%d-%m-%Y')) df['payment_list'] = df['payment_list'].apply(sort_payment_list) # --- Embedded lender-specific rules --- lender_bounce_penal_rules = { 'bounce': {3: 500, 4: 0, 9: 350, 10: 0, 20: 300}, 'penal': {3: (2*12/365)/100, 4: (2*12/365)/100, 9: (2*12/365)/100, 10: 0, 20: 0.0006}, 'default': {'bounce': 300, 'penal': 0.0015} # 0.15% daily } # set bounce charges for ind in df.index: if df.at[ind, 'emi_amount'] <= 0 or df.at[ind, 'due_date'] >= current_date: continue # if lender_id present in lender_bounce_penal_rules dict then apply bounce w.r.t dict else apply default bounce lender_id = df.at[ind, 'lender_id'] bounce_amount = lender_bounce_penal_rules['bounce'].get(lender_id, lender_bounce_penal_rules['default']['bounce']) paid_upto_now = sum( float(list(payment.values())[0]) for payment in df.at[ind, 'payment_list'] if datetime.strptime(list(payment.keys())[0], '%d-%m-%Y').date() <= df.at[ind, 'due_date'] ) df.at[ind, 'bounce_charges'] = 0 if paid_upto_now >= df.at[ind, 'emi_amount'] * 0.9 else bounce_amount # set penal charges for ind in df.index: if df.at[ind, 'emi_amount'] <= 0 or df.at[ind, 'due_date'] >= current_date: continue # if lender_id present in lender_bounce_penal_rules dict then apply penal w.r.t dict else apply default penal lender_id = df.at[ind, 'lender_id'] daily_rate = lender_bounce_penal_rules['penal'].get(lender_id, lender_bounce_penal_rules['default']['penal']) pending_amount = df.at[ind, 'emi_amount'] df.at[ind, 'delay_days'] = 0 df.at[ind, 'delay_penalty'] = 0 df.at[ind, 'payment_date'] = pd.NaT for payment_dict in df.at[ind, 'payment_list']: payment_date = datetime.strptime(list(payment_dict.keys())[0], '%d-%m-%Y').date() amount_paid = float(list(payment_dict.values())[0]) if pd.isna(df.at[ind, 'payment_date']): delay_days = max(0, (payment_date - df.at[ind, 'due_date']).days) else: delay_days = max(0, (payment_date - df.at[ind, 'payment_date']).days - 1) df.at[ind, 'delay_days'] += round(delay_days) if pending_amount > 0 else 0 df.at[ind, 'delay_penalty'] += round(pending_amount * daily_rate * delay_days) if pending_amount > 0 else 0 df.at[ind, 'payment_date'] = payment_date if payment_date > df.at[ind, 'due_date'] else pd.NaT pending_amount -= amount_paid df['total_payable'] = df['bpi_interest'] + df['morat_interest'] + df['emi_amount'] + df['bounce_charges'] + df['delay_penalty'] - df['charges_waived'] df['amount_due'] = np.maximum(0,df['total_payable'] - df['total_paid']) # allocate to previous bounce penal for ind in df.index: if new_amount_paid > 0: if (df.at[ind, 'due_date'] <= current_date) and (df.at[ind, 'amount_due'] > 0): allocatable_amount = min(df.at[ind, 'amount_due'], new_amount_paid) current_details = df.at[ind, 'payment_details'] new_details = f"{new_payment_date} - {allocatable_amount} - bounce-penal" df.at[ind, 'payment_details'] = (current_details + ', ' if current_details else '') + new_details df.at[ind, 'total_paid'] += allocatable_amount df.at[ind, 'amount_due'] -= allocatable_amount new_amount_paid -= allocatable_amount if new_amount_paid <= 0: break # allocate to future emi for ind in df.index: if new_amount_paid > 0: if df.at[ind, 'total_paid'] < df.at[ind, 'emi_amount']: allocatable_amount = min(df.at[ind, 'amount_due'], new_amount_paid) current_details = df.at[ind, 'payment_details'] new_details = f"{new_payment_date} - {allocatable_amount} - emi" df.at[ind, 'payment_details'] = (current_details + ', ' if current_details else '') + new_details df.at[ind, 'total_paid'] += allocatable_amount df.at[ind, 'amount_due'] -= allocatable_amount df.at[ind, 'pending_for_emi'] -= allocatable_amount df.at[ind, 'total_paid_for_emi'] += allocatable_amount new_amount_paid -= allocatable_amount if new_amount_paid <= 0: break # Reset column values df['payment_status'] = df.apply(lambda row: 'partial' if 0 < row['amount_due'] < row['total_payable'] else 'success' if row['amount_due'] == 0 else 'pending', axis=1) df['interest_comp_paid'] = np.where(df['total_paid_for_emi'] >= df['interest_comp'], df['interest_comp'], df['total_paid_for_emi']) df['principal_comp_paid'] = np.maximum(0, df['total_paid_for_emi'] - df['interest_comp']) # df['closing_principal_outstanding'] = df['opening_principal_outstanding'] - df['principal_comp_paid'] df['payment_date'] = np.where(df['payment_status'] == 'pending', '', np.where(df['payment_status'] == 'partial', new_payment_date, df['payment_date'])) # cleaning values in dataframe df['pending_for_emi'] = np.where(df['payment_status'] == 'success', 0, np.maximum(0, df['pending_for_emi'])) df = df.fillna(' ') # reordering columns df = df[['emi_payment_id', 'application_id', 'emi_date', 'due_date', 'bpi_interest', 'morat_interest', 'interest_comp', 'principal_comp', 'emi_amount', 'bounce_charges', 'delay_penalty', 'charges_waived', 'total_payable', 'amount_due', 'total_paid', 'total_paid_for_bpi', 'total_paid_for_emi', 'pending_for_emi', 'payment_status', 'delay_days', 'recent_payment_date','interest_comp_paid', 'principal_comp_paid', 'payment_details', 'principal_outstanding']] df.rename(columns={'recent_payment_date': 'payment_date'}, inplace=True) result = { "payment_allocation_json": df.to_dict(orient='records'), "over_payment": max(0,new_amount_paid) } logger.send_log_event(f"INFO: Function executed successfully.") return result # data = { # "application_id": 12474, # "payment_list": { # "payment_date": "30-08-2025", # "amount": 9800 # } # } # print(function1(data))