import pandas as pd import numpy as np import numpy_financial as npf from utils import execute_query, get_cloudwatch_logger, apply_waiver_allocation, reconciliation, zip_file from scripts.mysql_queries import emi_wise_payments_query, net_loan_irr_query, pf_dp_due_query, closed_cancelled_settled_check, \ settlement_validation_qry, settlement_reconcile_v2_query import pytz, asyncio, os from datetime import datetime from scripts.bounce_penal_update import function1 as bounce_penal_function from core_utilities import mail_service logger = get_cloudwatch_logger(stream_name="update_loan_data") local_tz = pytz.timezone("Asia/Kolkata") UPDATE_LOANS_DIR = "update_loans_logs_folder" os.makedirs(UPDATE_LOANS_DIR, exist_ok=True) # from pandasgui import show 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') 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 loan_metrics_calculation(data: dict): """ Calculates pos, interest os, dpd etc. for a batch of application IDs and updates in db. Accepts a dictionary of the form: {"app_id_list": [id1, id2, ...]} """ application_ids = data.get("app_id_list") or [] if application_ids: # Use provided IDs original_app_ids = set(application_ids) else: # Fetch all applicable application IDs from DB all_app_query = """ select id from consumer_application where state in ('disbursed', 'nach registered') and (deleted_at is null and is_deleted is null and merchant_id not in (73, 9421) and lender_id not in (18, 21)) """ all_app_df = execute_query(all_app_query) original_app_ids = set(all_app_df['id'].tolist()) # Update bounce penal to get updated data await bounce_penal_function({"app_id_list": original_app_ids}) logger.send_log_event(f"Fetching data from emi_payments table for {len(original_app_ids)} application_ids.") ep_df = execute_query(emi_wise_payments_query, params={'application_ids': original_app_ids}) if isinstance(ep_df, str): return {"status": "ERROR", "message": ep_df} found_app_ids = set(ep_df['application_id'].unique()) missing_ep = original_app_ids - found_app_ids failed_app_ids = set() failed_app_ids.update(missing_ep) if ep_df.empty: return {"status": "FAILED", "message": "No data found in emi_payments for any of the given Application_ids"} irr_df = execute_query(net_loan_irr_query, params={"application_ids": list(found_app_ids)}) if isinstance(irr_df, str): return {"status": "ERROR", "message": irr_df} found_irr_ids = set(irr_df['application_id'].unique()) missing_irr = found_app_ids - found_irr_ids failed_app_ids.update(missing_irr) # found in ep but not in irr valid_app_ids = found_app_ids - failed_app_ids # get only those aplication_id where net_loan amount and irr is present. if not valid_app_ids: return {"status": "FAILED", "message": "No valid application data to process."} ep_df = ep_df[ep_df['application_id'].isin(valid_app_ids)] ep_df = ep_df.groupby('application_id').apply(apply_waiver_allocation).reset_index(drop=True) ep_df = ep_df.merge(irr_df, on='application_id') def compute_irr(row): # IRR computation # Instead of fetching monthly interest componenet from db, calculate it by npf.irr function by passing # net_loan_amount(in -ve) & list of emi amounts from emi_payments table. try: if row['loan_type'] == 'subvention': return 0.00 if row['pre_part_flag'] == 1: tenure = row['tenure'] emi_value = row['emi_value'] net_loan_amount = row['net_loan_amount'] return npf.rate(nper = tenure, pmt = - emi_value, pv = net_loan_amount, fv = 0, when = 'end') if pd.isna(row.get('emis_list')) or not row.get('emis_list'): return 0.00 cash_flows = [-int(row['net_loan_amount'])] + list(map(int, str(row['emis_list']).split(','))) irr = npf.irr(cash_flows) return 0.00 if np.isnan(irr) else irr except Exception as e: print(f"Error computing IRR for application_id {row.name}: {e}") return 0.00 irr_map = (ep_df.drop_duplicates('application_id').set_index('application_id').apply(compute_irr, axis=1)) ep_df['irr_rate'] = ep_df['application_id'].map(irr_map) result_data = [] for app_id, group in ep_df.groupby('application_id'): # perform reconciliation for all application_ids try: net_loan_amount = group['net_loan_amount'].iloc[0] irr_rate = group['irr_rate'].iloc[0] df_reconciled = reconciliation(group, loan_principal=net_loan_amount, irr_rate=irr_rate) if pd.isna(net_loan_amount) or pd.isna(irr_rate): failed_app_ids.add(app_id) continue df_reconciled['emi_date'] = pd.to_datetime(df_reconciled['emi_date'], format='%d-%m-%Y').dt.date df_reconciled['unpaid_emi_date'] = pd.to_datetime(df_reconciled['unpaid_emi_date'], format='%d-%m-%Y').dt.date df_reconciled['emi_amount_paid'] = np.where(df_reconciled['emi_amount'] > 0, df_reconciled['interest_paid'] + df_reconciled['principal_paid'], 0) current_date = pd.Timestamp.today().date() # show(df_reconciled) # subract settled amounts settled_amounts = execute_query(settlement_reconcile_v2_query, params = {'application_ids': tuple([app_id])}) if isinstance(settled_amounts, str): failed_app_ids.add(app_id) continue if isinstance(settled_amounts, pd.DataFrame) and settled_amounts.empty: settled_principal_amount = 0 settled_interest_amount = 0 settled_bounce_amount = 0 settled_penal_amount = 0 settled_total_amount = 0 settlement_done_date = None else: settled_amounts = settled_amounts.iloc[0].to_dict() settled_principal_amount = settled_amounts.get('settled_principal_amount', 0) settled_interest_amount = settled_amounts.get('settled_interest_amount', 0) settled_bounce_amount = settled_amounts.get('settled_bounce_amount', 0) settled_penal_amount = settled_amounts.get('settled_penal_amount', 0) settled_total_amount = (settled_principal_amount + settled_interest_amount + settled_bounce_amount + settled_penal_amount) settlement_done_date = settled_amounts.get('settlement_done_date', None) pre_emi_due = round(df_reconciled['pre_emi_amount'].sum() - df_reconciled['pre_emi_paid'].sum()) principal_outstanding = round(max(net_loan_amount - df_reconciled['principal_paid'].sum() - settled_principal_amount,0), 2) past_emis = df_reconciled[(df_reconciled['emi_date'] <= current_date) & (df_reconciled['emi_amount'] > 0)] interest_due_until_now = past_emis['interest_due'].fillna(0).sum() interest_paid = df_reconciled['interest_paid'].fillna(0).sum() + settled_interest_amount # interest_due = round(max(0, interest_due_until_now - interest_paid) + (future_emi_interest or 0), 2) interest_outstanding = round(max(0, interest_due_until_now - interest_paid), 2) bounce_outstanding = round(max(df_reconciled['bounce_charges'].sum() - df_reconciled['bounce_paid'].sum() - settled_bounce_amount, 0), 2) penal_outstanding = round(max(df_reconciled['delay_penalty'].sum() - df_reconciled['penal_paid'].sum() - settled_penal_amount, 0), 2) emi_overdue_count = df_reconciled.loc[ (df_reconciled['emi_date'] < current_date) & (pd.notna(df_reconciled['unpaid_emi_date'])) ].shape[0] principal_due = int(max(0, round( df_reconciled.loc[df_reconciled['emi_date'] < current_date, 'emi_amount'].sum() - df_reconciled['emi_amount_paid'].sum() - (settled_principal_amount + settled_interest_amount) ))) total_principal_due = int(max(0, round( (df_reconciled['emi_amount'] - df_reconciled['emi_amount_paid']).sum() - (settled_principal_amount + settled_interest_amount) ))) amount_over_due = int(max(0, round( df_reconciled.loc[df_reconciled['emi_date'] < current_date, 'total_amount_due'].sum() - (df_reconciled.loc[df_reconciled['emi_date'].notna(),'total_amount_paid'].sum()) # exclude prepartpayment rows - settled_total_amount ))) total_amount_due = int(max(0, round((df_reconciled['total_amount_due'] - (df_reconciled['total_amount_paid'])).sum() - settled_total_amount))) emi_paid = int(((df_reconciled['total_amount_paid'] >= df_reconciled['total_amount_due']) & (df_reconciled['emi_payment_id'] != 0)).sum()) emi_pending = int(max(0, len(df_reconciled[df_reconciled['emi_payment_id'] != 0]) - emi_paid)) first_unpaid_emi = df_reconciled.loc[df_reconciled['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 pd.isna(first_unpaid_emi_date): dpd = 0 elif settlement_done_date is not None: dpd = max((settlement_done_date - first_unpaid_emi_date).days,0) else: dpd = 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.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(bounce_outstanding), "penal_charges_due": float(penal_outstanding), "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(df_reconciled[df_reconciled['emi_payment_id'] != 0])}) logger.send_log_event(f"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(bounce_outstanding)}, penal_charges_due={float(penal_outstanding)}, \ 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(df_reconciled[df_reconciled['emi_payment_id'] != 0])}") except Exception as e: print(f"Error for app_id={app_id}: {e}") failed_app_ids.add(app_id) if not result_data: return {"status": "FAILED", "message": "No data processed."} result_df = pd.DataFrame(result_data) # Get pf due and downpayment due columns for all application_ids pf_dp_df = execute_query(pf_dp_due_query, params={"application_ids": list(result_df['application_id'])}) if isinstance(pf_dp_df, pd.DataFrame) and not pf_dp_df.empty: result_df = result_df.merge(pf_dp_df, on='application_id', how='left') result_df[['pf_due', 'down_payment_due']] = result_df[['pf_due', 'down_payment_due']].fillna(0) else: result_df['pf_due'] = 0 result_df['down_payment_due'] = 0 if failed_app_ids: logger.send_log_event(f"WARNING: Could not process the following application_ids: {sorted(list(failed_app_ids))}") print(f"Could not process the following application_ids: {sorted(list(failed_app_ids))}") # if case is foreclosed or closed/cancelled then update all values to 0 foreclosed_settled_df = execute_query(settlement_validation_qry, params={"application_ids": list(result_df['application_id'])}) if isinstance(foreclosed_settled_df, pd.DataFrame) and not foreclosed_settled_df.empty: result_df = result_df.merge(foreclosed_settled_df, on='application_id', how='left') mask = (result_df['foreclosed'] == 1) | (result_df['app_status'].isin(['closed','cancelled'])) | (result_df['settled'] == 1) cols = ['pre_emi_due', 'principal_outstanding', 'interest_outstanding', 'bounce_charges_due', 'penal_charges_due', 'down_payment_due', 'pf_due', 'dpd', 'emi_overdue_count', 'principal_due', 'total_principal_due', 'amount_over_due', 'total_amount_due', 'emi_pending'] result_df.loc[mask, cols] = 0 result_df.loc[mask, 'first_unpaid_emi_date'] = pd.NaT result_df.loc[mask, 'dpd_bucket'] = 'Current' result_df.loc[mask, 'emi_paid'] = result_df.loc[mask, 'total_emi_count'] 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 {len(list(result_df['application_id']))} records.") if not application_ids: # If running via cron now_str = datetime.now(local_tz).strftime('%Y-%m-%d_%H-%M-%S') for f in os.listdir(UPDATE_LOANS_DIR): file_path = os.path.join(UPDATE_LOANS_DIR, f) if os.path.isfile(file_path): os.remove(file_path) log_file = os.path.join(UPDATE_LOANS_DIR, f"update_loans_log_{now_str}.csv") save_df = result_df[['application_id', 'principal_outstanding', 'dpd', 'interest_outstanding', 'bounce_charges_due', 'penal_charges_due']] save_df.to_csv(log_file, index=False) zip_path = log_file.replace('.csv', '_compressed.zip') zip_file(log_file, zip_path) msg = f"Update Loans Logs {now_str}" email_list = ["rudraksha.vaity@creditfair.in"] mail_service.send_mail_with_attachment(email_list, f"Update Loans Log File\n {msg}", msg, zip_path) return {"status": "Success", "message": "Successfully processed loans data"} # if __name__ == '__main__': # # data = {"app_id_list": [12474,13677,13394, 10648]} # data = {"app_id_list": [78558]} # 276320, 280211,276320,1011,1215,266080,265895 # result = asyncio.run(loan_metrics_calculation(data)) # # result = loan_metrics_calculation(data) # print(result)