import pandas as pd import numpy as np import asyncio, pytz from scripts.mysql_queries import emi_wise_payments_query, pf_dp_due_query from utils import execute_query, get_cloudwatch_logger, log_exception, get_hm_rps from datetime import datetime logger = get_cloudwatch_logger(stream_name="foreclosure_breakup") # from pandasgui import show local_tz = pytz.timezone("Asia/Kolkata") def update_tables(result_df, batch_size=3000): """ Updates consumer_application_additional (upsert) and consumer_application_element (update) using a single iteration over the DataFrame and batched execution. """ now = datetime.now(local_tz).strftime('%Y-%m-%d %H:%M:%S') print('update') sql_additional = """ INSERT INTO consumer_application_additional (application_id, pre_emi_due, principal_outstanding, interest_outstanding, bounce_charges_due, penal_charges_due, processing_fee_due, down_payment_due, first_unpaid_emi_date, dpd, dpd_bucket, emi_overdue_count, created_at, updated_at) VALUES (:application_id, :pre_emi_due, :principal_outstanding, :interest_outstanding, :bounce_charges_due, :penal_charges_due, :pf_due, :down_payment_due, :first_unpaid_emi_date, :dpd, :dpd_bucket, :emi_overdue_count, :created_at, :updated_at) ON DUPLICATE KEY UPDATE pre_emi_due = VALUES(pre_emi_due), principal_outstanding = VALUES(principal_outstanding), interest_outstanding = VALUES(interest_outstanding), bounce_charges_due = VALUES(bounce_charges_due), penal_charges_due = VALUES(penal_charges_due), processing_fee_due = VALUES(processing_fee_due), down_payment_due = VALUES(down_payment_due), first_unpaid_emi_date = VALUES(first_unpaid_emi_date), dpd = VALUES(dpd), dpd_bucket = VALUES(dpd_bucket), emi_overdue_count = VALUES(emi_overdue_count), updated_at = VALUES(updated_at); """ sql_element = """ UPDATE consumer_application_element SET principal_due = :principal_due, total_principal_due = :total_principal_due, amount_over_due = :amount_over_due, total_amount_due = :total_amount_due, emi_paid = :emi_paid, emi_pending = :emi_pending, updated_at = :updated_at WHERE application_id = :application_id """ # Prepare parameter dicts in a single loop params_additional = [] params_element = [] for row in result_df.itertuples(index=False): params_additional.append({ "application_id": row.application_id, "pre_emi_due": row.pre_emi_due, "principal_outstanding": row.principal_outstanding, "interest_outstanding": row.interest_outstanding, "bounce_charges_due": row.bounce_charges_due, "penal_charges_due": row.penal_charges_due, "pf_due": row.pf_due, "down_payment_due": row.down_payment_due, "first_unpaid_emi_date": row.first_unpaid_emi_date, "dpd": row.dpd, "dpd_bucket": row.dpd_bucket, "emi_overdue_count": row.emi_overdue_count, "created_at": now, "updated_at": now }) params_element.append({ "application_id": row.application_id, "principal_due": row.principal_due, "total_principal_due": row.total_principal_due, "amount_over_due": row.amount_over_due, "total_amount_due": row.total_amount_due, "emi_paid": row.emi_paid, "emi_pending": row.emi_pending, "updated_at": now }) logger.send_log_event(f"Updating element table with these values: application_id={row.application_id}, principal_due={row.principal_due}, total_principal_due={row.total_principal_due}, amount_over_due={row.amount_over_due}, total_amount_due={row.total_amount_due}, emi_paid={row.emi_paid}, emi_pending={row.emi_pending}, updated_at={now}") # Execute in batches using execute_query (automatically transactional) total_rows = len(result_df) for i in range(0, total_rows, batch_size): batch_additional = params_additional[i:i + batch_size] batch_element = params_element[i:i + batch_size] # Upsert consumer_application_additional execute_query(sql_additional, batch_additional) # Update consumer_application_element execute_query(sql_element, batch_element) return True async def hm_loan_schedule(app_id): try: hm_rps_df = get_hm_rps(app_id) # Get homeville rps via api if isinstance(hm_rps_df, dict) and hm_rps_df.get("status") == "Failed": log_exception(logger, f"Error fetching Homeville RPS for application_id {app_id}: {hm_rps_df}") return hm_rps_df logger.send_log_event(f"In hm_update_data func(). Homeville RPS fetched successfully for application_id: {app_id}") # Rename & assign new columns hm_rps_df.rename(columns={'repay_date':'emi_date', 'repay_amount':'emi_amount', 'repay_interest':'interest_due', 'repay_principle':'principal_due','amount_paid':'total_amount_paid','repay_os':'closing_pos', 'repaid_at':'latest_payment_date','repayment_type':'emi_type'}, inplace=True) hm_rps_df['emi_date'] = pd.to_datetime(hm_rps_df['emi_date']).dt.strftime('%d-%m-%Y') hm_rps_df['due_date'] = hm_rps_df['emi_date'] hm_rps_df['pre_emi_amount'] = np.where(hm_rps_df['emi_type'] == 'PEMI', hm_rps_df['emi_amount'], 0) hm_rps_df['interest_due'] = np.where(hm_rps_df['emi_type'] == 'EMI', hm_rps_df['interest_due'], 0) hm_rps_df['emi_amount'] = np.where(hm_rps_df['emi_type'] == 'EMI', hm_rps_df['emi_amount'], 0) hm_rps_df['bounce_charges'] = 0 hm_rps_df['delay_penalty'] = 0 hm_rps_df['charges_waived'] = 0 hm_rps_df['total_amount_due'] = hm_rps_df['interest_due'] + hm_rps_df['principal_due'] + hm_rps_df['pre_emi_amount'] hm_rps_df['pre_emi_paid'] = np.where(hm_rps_df['emi_type'] == 'PEMI', hm_rps_df['total_amount_paid'], 0) hm_rps_df['interest_paid'] = np.where( hm_rps_df['total_amount_paid'] >= hm_rps_df['interest_due'], hm_rps_df['interest_due'], np.where( hm_rps_df['total_amount_paid'] > 0, np.maximum(hm_rps_df['interest_due'] - hm_rps_df['total_amount_paid'], 0), 0 ) ) hm_rps_df['principal_paid'] = np.where( hm_rps_df['total_amount_paid'] >= (hm_rps_df['interest_due'] + hm_rps_df['principal_due']), hm_rps_df['principal_due'], # If total payment covers both interest and principal, pay the full principal np.where( hm_rps_df['total_amount_paid'] >= hm_rps_df['interest_due'], np.maximum(hm_rps_df['total_amount_paid'] - hm_rps_df['interest_due'], 0), # Pay remaining amount toward principal 0 # If no interest is paid, no principal can be paid ) ) hm_rps_df['bounce_paid'] = 0 hm_rps_df['penal_paid'] = 0 hm_rps_df['charges_to_be_waived'] = 0 hm_rps_df['unpaid_emi_date'] = np.where( (hm_rps_df['interest_paid'] + hm_rps_df['principal_paid']) < hm_rps_df['emi_amount'], hm_rps_df['emi_date'], pd.NaT ) # Foreclosure amount foreclosure_amt_if_unpaid = ( # add pos + current interest due(if not paid) + future interest hm_rps_df['closing_pos'] + (hm_rps_df['interest_due'] - hm_rps_df['interest_paid']) + hm_rps_df['interest_due'].shift(-1).fillna(0) ) condition = ( # if emi not fully paid then above amount else show closing pos ((hm_rps_df['interest_due'] + hm_rps_df['principal_due']) - (hm_rps_df['interest_paid'] + hm_rps_df['principal_paid']) ).round() > 0 ) # Set foreclosure amount based on condition hm_rps_df['foreclosure_amount'] = np.round(np.where(condition, foreclosure_amt_if_unpaid, hm_rps_df['closing_pos']),2) hm_rps_df.loc[hm_rps_df.index[-1], 'foreclosure_amount'] = 0 # Set last amount as 0 hm_rps_df['opening_pos'] = 0 # get emi_payment_id & payment_details ep_df = execute_query(emi_wise_payments_query,params={'application_ids': tuple([app_id])}) if isinstance(ep_df, str): return {"status": "ERROR", "message": ep_df} if isinstance(ep_df, pd.DataFrame) and ep_df.empty: return {"status": "FAILED", "message": f"No data found in emi_payments table for Application_id: {app_id}"} hm_rps_df = pd.merge(hm_rps_df, ep_df[['application_id', 'emi_date', 'emi_payment_id', 'payment_details', 'payment_status']], how='left', on=['application_id','emi_date']) hm_rps_df['emi_type'] = np.where(hm_rps_df['emi_type'] == 'part_payment', 'prepartpayment', 'emi') hm_rps_df['emi_amount_paid'] = np.where(hm_rps_df['emi_amount'] > 0, hm_rps_df['interest_paid'] + hm_rps_df['principal_paid'], 0) # Re-arrange columns hm_rps_df = hm_rps_df[['emi_payment_id', 'application_id', 'emi_date', 'due_date', 'pre_emi_amount', 'interest_due', 'principal_due', 'emi_amount', 'bounce_charges', 'delay_penalty', 'charges_waived', 'total_amount_due', 'total_amount_paid', 'emi_amount_paid', 'pre_emi_paid', 'interest_paid', 'principal_paid', 'bounce_paid', 'penal_paid', 'charges_to_be_waived', 'payment_details', 'unpaid_emi_date', 'opening_pos', 'closing_pos', 'latest_payment_date', 'foreclosure_amount', 'payment_status', 'emi_type']] hm_rps_df['emi_date'] = pd.to_datetime(hm_rps_df['emi_date'], dayfirst=True).dt.date hm_rps_df['unpaid_emi_date'] = pd.to_datetime(hm_rps_df['unpaid_emi_date'], format='%d-%m-%Y').dt.date pre_emi_due = round(hm_rps_df['pre_emi_amount'].sum() - hm_rps_df['pre_emi_paid'].sum()) principal_outstanding = round(max(hm_rps_df['principal_due'].sum() - hm_rps_df['principal_paid'].sum(),0), 2) # Get interest due. formula = (interest due upto current + one future interest) - interest paid current_date = pd.Timestamp.today().date() past_emis = hm_rps_df[(hm_rps_df['emi_date'] <= current_date) & (hm_rps_df['emi_amount'] > 0)] future_emis = hm_rps_df[hm_rps_df['emi_date'] > current_date].sort_values('emi_date') interest_due_until_now = past_emis['interest_due'].fillna(0).sum() interest_paid = hm_rps_df['interest_paid'].fillna(0).sum() future_emi_interest = 0 if not future_emis.empty: # One future EMI's interest next_emi = future_emis.iloc[0] next_interest_due = next_emi['interest_due'] or 0 next_interest_paid = next_emi['interest_paid'] or 0 if round(next_interest_due - next_interest_paid, 2) > 0: # Add future interest ONLY if not fully paid future_emi_interest = next_interest_due interest_due = round(max(0, interest_due_until_now - interest_paid) + future_emi_interest, 2) interest_outstanding = round(max(0,interest_due_until_now - interest_paid),2) # Find Processing fee due & Down payment due pf_dp_df = execute_query(pf_dp_due_query, params = {'application_ids': tuple([app_id])}) if isinstance(pf_dp_df, str): return {"status": "ERROR", "message": pf_dp_df} if isinstance(pf_dp_df, pd.DataFrame) and pf_dp_df.empty: print(f"No data found in emi_calculation table for Application_id: {app_id}") log_exception(logger, f"No data found in emi_calculation table for Application_id: {app_id}") return {"status": "FAILED", "message": f"No data found in emi_calculation table for Application_id: {app_id}"} pf_dp_df.select_dtypes(include=['number']).fillna(0) emi_overdue_count = hm_rps_df.loc[ (hm_rps_df['emi_date'] < current_date) & (pd.notna(hm_rps_df['unpaid_emi_date'])) ].shape[0] principal_due = int(max(0, round( hm_rps_df.loc[hm_rps_df['emi_date'] < current_date, 'emi_amount'].sum() - hm_rps_df['emi_amount_paid'].sum() ))) total_principal_due = int(max(0, round( (hm_rps_df['emi_amount'] - hm_rps_df['emi_amount_paid']).sum() ))) amount_over_due = int(max(0, round( hm_rps_df.loc[hm_rps_df['emi_date'] < current_date, 'total_amount_due'].sum() - (hm_rps_df.loc[hm_rps_df['emi_date'].notna(),'total_amount_paid'].sum()) # exclude prepartpayment rows ))) total_amount_due = int(max(0, round((hm_rps_df['total_amount_due'] - (hm_rps_df['total_amount_paid'])).sum()))) emi_paid = int(((hm_rps_df['total_amount_paid'] >= hm_rps_df['total_amount_due']) & (hm_rps_df['emi_payment_id'] != 0)).sum()) emi_pending = int(max(0, len(hm_rps_df[hm_rps_df['emi_payment_id'] != 0]) - emi_paid)) first_unpaid_emi = hm_rps_df.loc[hm_rps_df['unpaid_emi_date'].notna(),'unpaid_emi_date'].min() first_unpaid_emi_date = pd.NaT if pd.isna(first_unpaid_emi) else first_unpaid_emi dpd = 0 if pd.isna(first_unpaid_emi_date) else max((current_date - first_unpaid_emi_date).days,0) if dpd <= 0: dpd_bucket = 'Current' elif dpd <= 30: dpd_bucket = '1-30' elif dpd <= 60: dpd_bucket = '31-60' elif dpd <= 90: dpd_bucket = '61-90' elif dpd <= 120: dpd_bucket = '91-120' elif dpd <= 150: dpd_bucket = '121-150' elif dpd <= 180: dpd_bucket = '151-180' elif dpd <= 364: dpd_bucket = '181-364' else: dpd_bucket = '>=365' result_data = [] result_data.append({"application_id": app_id, "pre_emi_due": float(pre_emi_due), "principal_outstanding": float(principal_outstanding), "interest_outstanding": float(interest_outstanding), "bounce_charges_due": float(0), "penal_charges_due": float(0), "first_unpaid_emi_date": first_unpaid_emi_date, "dpd": dpd, "dpd_bucket": dpd_bucket, "emi_overdue_count": emi_overdue_count, "principal_due": principal_due, "total_principal_due": total_principal_due, "amount_over_due": amount_over_due, "total_amount_due": total_amount_due, "emi_paid": emi_paid, "emi_pending": emi_pending, "total_emi_count": len(hm_rps_df[hm_rps_df['emi_payment_id'] != 0]), "down_payment_due": float(pf_dp_df['down_payment_due'].iloc[0]) if not pf_dp_df.empty else 0, "pf_due": float(pf_dp_df['pf_due'].iloc[0]) if not pf_dp_df.empty else 0}) result_df = pd.DataFrame(result_data) # show(hm_rps_df) # show(result_df) update_result = update_tables(result_df) if update_result: logger.send_log_event(f"INFO: consumer_application_additional & consumer_application_element table successfully updated for application_id: {app_id}.") else: logger.send_log_event(f"INFO: consumer_application_additional & consumer_application_element table not updated for application_id: {app_id}.") logger.send_log_event(f"Homeville loan update data for application_id {app_id}: {result_data}") hm_rps_df['total_amount_due'] = (hm_rps_df['total_amount_due'] - hm_rps_df['total_amount_paid']).clip(lower=0) foreclosure_breakup = { "Pre-EMI due": float(pre_emi_due), "Principal due": float(principal_outstanding), "Interest due": float(interest_due), "Moratorium due": float(0), "Bounce charges due": float(0), "Penal charges due": float(0), "PF due": float(pf_dp_df['pf_due'].iloc[0]) if not pf_dp_df.empty else 0, "Advance due": float(pf_dp_df['down_payment_due'].iloc[0]) if not pf_dp_df.empty else 0, "Interest Outstanding": float(interest_outstanding) } return {"status": "Success", "message": foreclosure_breakup, "data": hm_rps_df.to_dict(orient='records')} except Exception as e: log_exception(logger, f"An unexpected error occurred: {e}") return {"status": "ERROR", "message": f"An unexpected error occurred: {e}"} # if __name__ == "__main__": # print(asyncio.run(hm_loan_schedule(254117)))