import pandas as pd import numpy as np import numpy_financial as npf import math, pytz, json, asyncio, os from datetime import datetime from dateutil.relativedelta import relativedelta from utils import execute_query, apply_waiver_allocation, insert_payment_transaction, get_cloudwatch_logger, log_exception from scripts.loan_schedule_script import generateLoanSchedule from scripts.mysql_queries import net_loan_irr_query, emi_id_wise_payments # from pandasgui import show logger = get_cloudwatch_logger(stream_name="pre_part_payment") current_date = pd.Timestamp.today().date() local_tz = pytz.timezone("Asia/Kolkata") now = datetime.now(local_tz).strftime('%Y-%m-%d %H:%M:%S') def df_to_mysql_json_safe(df, drop_columns=None): df_clean = df.copy() # Drop unwanted columns if drop_columns: df_clean = df_clean.drop(columns=drop_columns, errors="ignore") # Replace NaN, inf, -inf with None df_clean = df_clean.replace([np.nan, np.inf, -np.inf], None) # Optional: round floats to 2 decimals float_cols = df_clean.select_dtypes(include=["float", "float64"]).columns df_clean[float_cols] = df_clean[float_cols].applymap(lambda x: round(x, 2) if x is not None else None) # Serialize to JSON return json.dumps(df_clean.to_dict(orient="records"), default=str) def compute_irr(loan_type, pre_part_flag, net_loan_amount, emis_list, tenure, emi_value): if loan_type == "subvention": return 0.00 if pre_part_flag == 1: return npf.rate(nper = tenure, pmt = - emi_value, pv = net_loan_amount, fv = 0, when = 'end') # Build cash flows cash_flows = [-int(net_loan_amount)] cash_flows.extend(map(int, emis_list.split(","))) irr = npf.irr(cash_flows) return irr if not math.isnan(irr) else 0.00 def reconciliation(df): """ This function reconciles payments recieved against interest, principal, charges and calulates opening and closing pos on emi level """ beginning_balance = df['opening_pos'].iloc[0] results = [] for row in df.itertuples(index=False): pre_emi = getattr(row, 'pre_emi_amount', 0) emi_amount = getattr(row, 'emi_amount', 0) bounce = getattr(row, 'bounce_charges', 0) penal = getattr(row, 'delay_penalty', 0) total_paid = getattr(row, 'total_paid', 0) total_paid_for_bpi = getattr(row,'total_paid_for_bpi') interest_due = getattr(row, 'interest_due', 0) principal_due = getattr(row, 'principal_due', 0) # Step 1: Begin allocation from total_paid of that EMI remaining_amount = total_paid pre_emi_paid = min(total_paid_for_bpi, pre_emi) # allocate total_paid_for_bpi directly to pre_emi # morat_paid = min(remaining_amount, moratorium_amount) # remaining_amount -= morat_paid interest_paid = min(remaining_amount, interest_due) remaining_amount -= interest_paid principal_paid = min(remaining_amount, principal_due) remaining_amount -= principal_paid bounce_paid = min(remaining_amount, bounce) remaining_amount -= bounce_paid penal_paid = min(remaining_amount, penal) remaining_amount -= penal_paid # Compute closing position -- don't require this code as principal outstanding is calculated separately below # closing_pos = opening_pos - principal_due # beginning_balance = closing_pos # Record the results for the row results.append({ 'emi_payment_id': row.emi_payment_id, 'application_id': row.application_id, 'emi_date': row.emi_date, 'pre_emi_amount': pre_emi, 'interest_due': interest_due, 'principal_due': principal_due, 'emi_amount': emi_amount, 'bounce_charges': bounce, 'delay_penalty': penal, 'charges_waived': row.charges_waived, 'total_amount_due': (pre_emi + emi_amount + bounce + penal), 'total_amount_paid': round(total_paid + total_paid_for_bpi, 2), 'pre_emi_paid': pre_emi_paid, 'interest_paid': interest_paid, 'principal_paid': principal_paid, 'bounce_paid': bounce_paid, 'penal_paid': penal_paid, 'unpaid_emi_date': row.emi_date if round(interest_paid + principal_paid) < round(emi_amount) else pd.NaT, 'opening_pos': getattr(row, 'opening_pos', 0), 'closing_pos': getattr(row, 'closing_pos', 0) }) return pd.DataFrame(results) def compute_pos(application_id): """ Compute POS, first unpaid emi date, last paid emi date etc. for a given application_id """ # fetch data from loan schedule table as it as precise(not rounded) interest, principal and emi values schedule_qry = """ select emi_payment_id,application_id, emi_date, due_date, pre_emi_amount, interest_due, principal_due, emi_amount, opening_pos, closing_pos from loan_schedule_v2 where application_id in :application_ids and deleted_at is null order by application_id, emi_date """ schedule_df = execute_query(schedule_qry, params = {'application_ids': tuple([application_id,])}) if isinstance(schedule_df, str): log_exception(f"ERROR: {schedule_df}") return {"status": "ERROR", "message": schedule_df} if isinstance(schedule_df, pd.DataFrame) and schedule_df.empty: # if no record present for given application_id in loan_Schedule_v2 table then call generateLoanSchedule which creates and inserts schedule in the table logger.send_log_event(f"INFO: No schedule present in loan_schedule_v2 for application_id: {application_id}. Inserting schedule in loan_schedule_v2.") generateLoanSchedule(application_id) amort_df = execute_query(schedule_qry, params = {'application_ids': tuple([application_id,])}) else: amort_df = schedule_df.copy() # else use existing records from loan_schedule_v2 # Get emi_payment_id wise payments and merge it with loan ammortization df payments_df = execute_query(emi_id_wise_payments,params={'application_ids': tuple([application_id])}) if isinstance(payments_df, str): return {"status": "ERROR", "message": payments_df} if isinstance(payments_df, pd.DataFrame) and payments_df.empty: return {"status": "FAILED", "message": f"No data found in emi_payments table for Application_id: {application_id}"} amort_df = pd.merge(amort_df, payments_df, on=['application_id', 'emi_date'], how='inner') amort_df.rename(columns={'emi_payment_id_y': 'emi_payment_id'}, inplace=True) amort_df = amort_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', 'opening_pos', 'closing_pos', 'total_paid', 'total_paid_for_bpi', 'latest_payment_date']] # show(amort_df) amort_df = apply_waiver_allocation(amort_df) df_reconciled = reconciliation(amort_df) 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['foreclosure_amt'] = df_reconciled.apply( lambda row: row['opening_pos'] + max(row['interest_due'] - row['interest_paid'], 0) + ( df_reconciled.iloc[row.name + 1]['interest_due'] if row.name < len(df_reconciled) - 1 else 0 ) if pd.notnull(row['unpaid_emi_date']) else row['closing_pos'], axis=1) # logger.send_log_event(f"INFO: Payments reconciled for application_id: {application_id}.") # Prepare data required to compute pre part payment principal_outstanding = round(max(df_reconciled['principal_due'].sum() - df_reconciled['principal_paid'].sum(), 0), 2) total_amount_due = int(max(0, round((df_reconciled['total_amount_due'] - (df_reconciled['total_amount_paid'])).sum()))) tenure = int(df_reconciled.loc[df_reconciled['emi_amount'] > 0].shape[0]) first_unpaid_emi = df_reconciled.loc[df_reconciled['unpaid_emi_date'].notna(),'unpaid_emi_date'].min() first_unpaid_emi_date = (first_unpaid_emi if pd.notna(first_unpaid_emi) else pd.NaT) last_paid_emi = df_reconciled.loc[(df_reconciled['unpaid_emi_date'].isna()) &(df_reconciled['emi_amount'] > 0),'emi_date'].max() last_paid_emi_date = pd.NaT if pd.isna(last_paid_emi) else last_paid_emi # check if atleast one emi is paid atleast_one_emi_paid = df_reconciled.loc[(df_reconciled['unpaid_emi_date'].isna()) &(df_reconciled['emi_amount'] > 0)].shape[0] > 0 # get last paid emi amount # if pd.notna(last_paid_emi_date): # last_paid_emi_amount = ( # df_reconciled.loc[ # (df_reconciled['emi_date'] == last_paid_emi_date) & # (df_reconciled['unpaid_emi_date'].isna()) & # (df_reconciled['emi_amount'] > 0), # 'emi_amount' # ].iloc[-1] # safe even if multiple rows exist # ) # else: # last_paid_emi_amount = 0.0 # get last paid emi amount filtered = df_reconciled.loc[df_reconciled['unpaid_emi_date'].notna(), 'emi_amount'] last_paid_emi_amount = round(filtered.iloc[0] if not filtered.empty else 0) # print(last_paid_emi_amount) # Get Current Foreclosure amount; interest due. formula = (interest due upto current + one future interest) - interest paid past_emis = df_reconciled[df_reconciled['emi_date'] <= current_date] future_emis = df_reconciled[df_reconciled['emi_date'] > current_date].sort_values('emi_date') interest_due_until_now = past_emis['interest_due'].fillna(0).sum() interest_paid = df_reconciled['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) foreclosure_amount = interest_due + principal_outstanding irr_df = execute_query(net_loan_irr_query, params = {'application_ids': tuple([application_id,])}) if isinstance(irr_df, str): return {"status": "ERROR", "message": irr_df} if isinstance(irr_df, pd.DataFrame) and irr_df.empty: return {"status": "FAILED", "message": f"No data found in emi_calculation table for Application_id: {application_id}"} empty_column = irr_df.columns[irr_df.isna().all()].tolist() if empty_column: return {"status": "FAILED", "message": f"{empty_column} is empty in emi_calculation table."} row = irr_df.iloc[0] monthly_interest_rate = compute_irr(loan_type=row["loan_type"], pre_part_flag=row["pre_part_flag"], net_loan_amount=row["net_loan_amount"], emis_list=row["emis_list"], tenure=row["tenure"], emi_value=row["emi_value"]) net_loan_amount = float(row["net_loan_amount"]) annual_interest_rate = monthly_interest_rate * 12 logger.send_log_event(f"INFO: net_loan_amount: {net_loan_amount}, monthly_interest_rate: {monthly_interest_rate} & annual_interest_rate: {annual_interest_rate} calculated for application_id: {application_id}.") compute_pos_data_dict = { "principal_outstanding": principal_outstanding, "total_outstanding": total_amount_due, "foreclosure_amount": foreclosure_amount, "first_unpaid_emi_date": first_unpaid_emi_date, "last_paid_emi_date": last_paid_emi_date, "atleast_one_emi_paid": atleast_one_emi_paid, "last_paid_emi_amount": last_paid_emi_amount, "monthly_interest_rate": monthly_interest_rate, "annual_interest_rate": annual_interest_rate, "total_tenure": tenure, "df_reconciled": df_reconciled.to_dict(orient='records') } return compute_pos_data_dict def reduce_emi(existing_rps, annual_interest_rate, current_pos, prepart_payment_date, prepart_amount, total_tenure, first_unpaid_emi_date, last_paid_emi_date): """ Calculates the Repayment Schedule after a part payment, keeping tenure constant and reducing EMI. """ days_since_old_pos = (prepart_payment_date - last_paid_emi_date).days # days between last paid emi and pre-part payment date interest_before_pp = round(current_pos * days_since_old_pos * (annual_interest_rate / 360), 2) logger.send_log_event(f"INFO: Interest for {days_since_old_pos} days is ₹{interest_before_pp}.") adj_prepart_amount = round(max(prepart_amount - interest_before_pp,0), 2) logger.send_log_event(f"INFO: Pre part payment amount after adjustment: ₹{adj_prepart_amount}.") new_pos = round(current_pos - adj_prepart_amount, 2) logger.send_log_event(f"INFO: New POS after pre part payment.: ₹{new_pos}.") paid_months = existing_rps[existing_rps['unpaid_emi_date'].isna() & (existing_rps['emi_amount'] > 0)].shape[0] pending_months = total_tenure - paid_months logger.send_log_event(f"INFO: Pending months where emi amount will change: {pending_months}") new_emi_amount = round(npf.pmt(annual_interest_rate/12, pending_months, -new_pos, 0), 2) if annual_interest_rate != 0 else round(new_pos / pending_months, 2) logger.send_log_event(f"INFO: New EMI amount: {new_emi_amount}") # First month calculation days_after_new_pos = (first_unpaid_emi_date - prepart_payment_date).days logger.send_log_event(f"INFO: Days after new pos: {days_after_new_pos}") interest_m1 = round(new_pos * days_after_new_pos * (annual_interest_rate / 360), 2) principal_m1 = round(new_emi_amount - interest_m1, 2) closing_pos = new_pos - principal_m1 logger.send_log_event(f"INFO: Interest first emi: {interest_m1}, Principal first emi: {principal_m1}, Closing POS: {closing_pos}") schedule_data = [] schedule_data.append({ 'installment_number':paid_months + 1, 'emi_date': first_unpaid_emi_date, 'emi_amount': new_emi_amount, 'bpi_interest': 0, 'morat_interest': 0, 'interest_comp': float(interest_m1), 'principal_comp': float(principal_m1), 'principal_outstanding': float(closing_pos), 'opening_pos1': float(new_pos), 'closing_pos1': float(closing_pos) }) # Subsequent months calculation pos = closing_pos emi_date = first_unpaid_emi_date + relativedelta(months=1) for m in range(paid_months + 2, total_tenure + 1): opening_pos = round(pos, 2) interest_comp = round(opening_pos * 30 * (annual_interest_rate / 360), 2) if m == total_tenure: # Check for final month (Last EMI amount adjustment) emi = round(pos + interest_comp, 2) principal_comp = round(pos, 2) closing_pos = 0 pos = 0 else: # Normal calculation as per EMI amount emi = new_emi_amount principal_comp = emi - interest_comp # pos = pos - principal_comp closing_pos = round(opening_pos - principal_comp, 2) pos = closing_pos schedule_data.append({ 'installment_number': m, 'emi_date': emi_date, 'emi_amount': emi, 'bpi_interest': 0, 'morat_interest': 0, 'interest_comp': float(interest_comp), 'principal_comp': float(principal_comp), 'principal_outstanding': float(pos), 'opening_pos1': float(opening_pos), 'closing_pos1': float(closing_pos) }) emi_date = emi_date + relativedelta(months=1) revised_tenure = pd.DataFrame(schedule_data) revised_tenure['total_payable'] = revised_tenure[['bpi_interest', 'morat_interest', 'interest_comp', 'principal_comp']].sum(axis=1).round() revised_tenure['foreclosure_amount'] = revised_tenure['principal_outstanding'] + revised_tenure['interest_comp'].shift(-1).fillna(0) final_tenure = existing_rps.merge(revised_tenure, on="emi_date", how = "left") # Remove extra columns and keep necessary final_tenure['interest_due'] = final_tenure['interest_comp'].fillna(final_tenure['interest_due']) final_tenure['principal_due'] = final_tenure['principal_comp'].fillna(final_tenure['principal_due']) final_tenure['emi_amount'] = final_tenure['emi_amount_y'].fillna(final_tenure['emi_amount_x']) final_tenure['principal_outstanding'] = final_tenure['principal_outstanding'].fillna(final_tenure['closing_pos']) final_tenure['foreclosure_amount'] = final_tenure['foreclosure_amount'].fillna(final_tenure['foreclosure_amt']) final_tenure = final_tenure[final_tenure['emi_date'].notna()].copy() start = 0 if final_tenure.iloc[0]['pre_emi_amount'] > 0 else 1 final_tenure['installment_number'] = range(start, start + len(final_tenure)) final_tenure = final_tenure[['emi_payment_id', 'installment_number','application_id', 'emi_date', 'pre_emi_amount', 'interest_due', 'principal_due', 'emi_amount', 'bounce_charges', 'delay_penalty', 'charges_waived', 'principal_outstanding', 'foreclosure_amount', 'opening_pos1', 'closing_pos1']] return final_tenure def reduce_tenure(existing_rps, annual_interest_rate, current_pos, prepart_payment_date, prepart_amount, first_unpaid_emi_date, last_paid_emi_date, last_paid_emi_amount): """ Calculates the Repayment Schedule after a part payment, keeping EMI constant and reducing tenure. """ days_since_old_pos = (prepart_payment_date - last_paid_emi_date).days # days between last paid emi and pre-part payment date interest_before_pp = round(current_pos * days_since_old_pos * (annual_interest_rate / 360), 2) logger.send_log_event(f"INFO: Interest for {days_since_old_pos} days is ₹{interest_before_pp}.") adj_prepart_amount = round(max(prepart_amount - interest_before_pp,0), 2) logger.send_log_event(f"INFO: Pre part payment amount after adjustment: ₹{adj_prepart_amount}.") new_pos = round(current_pos - adj_prepart_amount, 2) logger.send_log_event(f"INFO: New POS after pre part payment.: ₹{new_pos}.") days_after_new_pos = (first_unpaid_emi_date - prepart_payment_date).days logger.send_log_event(f"INFO: Days after new pos: {days_after_new_pos}") paid_months = existing_rps[existing_rps['unpaid_emi_date'].isna() & (existing_rps['emi_amount'] > 0)].shape[0] schedule_data = [] # First Month Adjustment interest_m1 = round(new_pos * days_after_new_pos * (annual_interest_rate / 360), 2) principal_m1 = round(last_paid_emi_amount - interest_m1, 2) closing_pos = new_pos - principal_m1 schedule_data.append({ 'installment_number':paid_months + 1, 'emi_date': first_unpaid_emi_date, 'emi_amount': last_paid_emi_amount, 'bpi_interest': 0, 'morat_interest': 0, 'interest_comp': float(interest_m1), 'principal_comp': float(principal_m1), 'principal_outstanding': float(closing_pos), 'opening_pos1': float(new_pos), 'closing_pos1': float(closing_pos) }) # Subsequent months calculation pos = closing_pos emi_date = first_unpaid_emi_date + relativedelta(months=1) standard_emi = last_paid_emi_amount m = paid_months + 2 while pos > 0: # Keep calculating until POS = 0 opening_pos = round(pos, 2) interest_comp = round(opening_pos * 30 * (annual_interest_rate / 360), 2) # Check if balance + interest is less than one full EMI (Last Month) if (opening_pos + interest_comp) <= standard_emi: # Check if pending amount in less then EMI (Last EMI amount adjustment) emi = opening_pos + interest_comp principal_comp = opening_pos closing_pos = 0.0 pos = 0 else: emi = standard_emi principal_comp = emi - interest_comp # pos = pos - principal_comp closing_pos = round(opening_pos - principal_comp, 2) pos = closing_pos schedule_data.append({ 'installment_number':m, 'emi_date': emi_date, 'emi_amount': emi, 'bpi_interest': 0, 'morat_interest': 0, 'interest_comp': float(interest_comp), 'principal_comp': float(principal_comp), 'principal_outstanding': float(pos), 'opening_pos1': float(opening_pos), 'closing_pos1': float(closing_pos) }) m += 1 emi_date = emi_date + relativedelta(months=1) revised_tenure = pd.DataFrame(schedule_data) revised_tenure['total_payable'] = revised_tenure[['bpi_interest', 'morat_interest', 'interest_comp', 'principal_comp']].sum(axis=1).round() revised_tenure['foreclosure_amount'] = revised_tenure['principal_outstanding'] + revised_tenure['interest_comp'].shift(-1).fillna(0) final_tenure = existing_rps.merge(revised_tenure, on = "emi_date", how = "left") final_tenure = final_tenure[final_tenure["emi_date"] <= revised_tenure["emi_date"].max()] # remove excess tenure # Remove extra columns and keep necessary final_tenure['interest_due'] = final_tenure['interest_comp'].fillna(final_tenure['interest_due']) final_tenure['principal_due'] = final_tenure['principal_comp'].fillna(final_tenure['principal_due']) final_tenure['emi_amount'] = final_tenure['emi_amount_y'].fillna(final_tenure['emi_amount_x']) final_tenure['principal_outstanding'] = final_tenure['principal_outstanding'].fillna(final_tenure['closing_pos']) final_tenure['foreclosure_amount'] = final_tenure['foreclosure_amount'].fillna(final_tenure['foreclosure_amt']) final_tenure = final_tenure[final_tenure['emi_date'].notna()].copy() start = 0 if final_tenure.iloc[0]['pre_emi_amount'] > 0 else 1 final_tenure['installment_number'] = range(start, start + len(final_tenure)) final_tenure = final_tenure[['emi_payment_id', 'installment_number','application_id', 'emi_date', 'pre_emi_amount', 'interest_due', 'principal_due', 'emi_amount', 'bounce_charges', 'delay_penalty', 'charges_waived', 'principal_outstanding', 'foreclosure_amount', 'opening_pos1', 'closing_pos1']] return final_tenure def update_tables(existing_rps, rps_to_be_updated, final_rps, data): """ Updates emi_payments and rps_logs and loan_schedule_v2 correctly: - Updates rows that should remain - Deletes / soft-deletes rows that should NOT remain - Inserts payment transaction and audit log """ # ---------------- SQL ---------------- # update_ep_sql = """ update emi_payments set emi_amount = :emi_amount, updated_at = :updated_at where application_id = :application_id and id = :id; """ delete_ep_sql = """ delete from emi_payments where application_id = :application_id and id not in ({ids}); """ # update_lsv2_sql = """ # update loan_schedule_v2 # set interest_due = :interest_due, principal_due = :principal_due, emi_amount = :emi_amount, opening_pos = :opening_pos, # closing_pos = :closing_pos, updated_at = :updated_at # where application_id = :application_id and emi_payment_id = :id and emi_date > current_date and deleted_at is null; # """ update_lsv2_sql = """ update loan_schedule_v2 set interest_due = :interest_due, principal_due = :principal_due, emi_amount = :emi_amount, opening_pos = :opening_pos, closing_pos = :closing_pos, updated_at = :updated_at where application_id = :application_id and emi_date = :emi_date and emi_date > current_date and deleted_at is null; """ # delete_lsv2_sql = """ # delete from loan_schedule_v2 # where application_id = :application_id and id not in ({ids}); # """ delete_lsv2_sql = """ update loan_schedule_v2 set deleted_at = :deleted_at where application_id = :application_id and emi_date not in ({emi_dates}) and deleted_at is null; """ update_rps_sql = """ update rps_logs set interest_comp = :interest_comp, principal_comp = :principal_comp, emi_amount = :emi_amount, total_payable = :total_payable, principal_outstanding = :principal_outstanding, foreclosure_amount = :foreclosure_amount, updated_at = :updated_at where application_id = :application_id and emi_date = :emi_date and deleted_at is null; """ delete_rps_sql = """ update rps_logs set deleted_at = :deleted_at where application_id = :application_id and emi_date NOT IN ({emi_dates}) and deleted_at is null; """ # ---------------- Prepare data ---------------- # data["payment_for"] = "pre_part_payment" application_id = data["application_id"] # IDs & dates that should remain emi_payment_ids = [] emi_dates = [] # ---------------- Updates ---------------- # insert_payment_transaction(data) for row in final_rps.itertuples(index=False): emi_payment_ids.append(row.emi_payment_id) # get list of emi_payment_ids to be updated emi_dates.append(row.emi_date) # get list of emi_dates to be updated execute_query( update_ep_sql, { "id": row.emi_payment_id, "application_id": row.application_id, "emi_amount": round(row.emi_amount), "updated_at": now, }, ) execute_query( update_lsv2_sql, { "emi_date": row.emi_date, "application_id": row.application_id, "interest_due": row.interest_due, "principal_due": row.principal_due, "emi_amount": row.emi_amount, "opening_pos": None if pd.isna(row.opening_pos1) else row.opening_pos1, "closing_pos": None if pd.isna(row.closing_pos1) else row.closing_pos1, "updated_at": now, }, ) execute_query( update_rps_sql, { "application_id": row.application_id, "emi_date": row.emi_date, "interest_comp": row.interest_due, "principal_comp": row.principal_due, "emi_amount": round(row.emi_amount), "total_payable": row.emi_amount, "principal_outstanding": row.principal_outstanding, "foreclosure_amount": row.foreclosure_amount, "updated_at": now, }, ) # ---------------- Deletes (ONLY ONCE) ---------------- # if data.get("type") == "tenure": # emi_payments delete ids_placeholder = ",".join(str(i) for i in emi_payment_ids) execute_query( delete_ep_sql.format(ids=ids_placeholder), {"application_id": application_id}, ) # loan_schedule_v2 soft delete # ids_placeholder = ",".join(str(i) for i in emi_payment_ids) # execute_query( # delete_lsv2_sql.format(ids=ids_placeholder), # {"application_id": application_id}, # ) # loan_schedule_v2 soft delete based on emi_date emi_date_placeholders = ",".join(f"'{d}'" for d in emi_dates) execute_query( delete_lsv2_sql.format(emi_dates=emi_date_placeholders), { "application_id": application_id, "deleted_at": now, }, ) # rps_logs soft delete emi_date_placeholders = ",".join(f"'{d}'" for d in emi_dates) execute_query( delete_rps_sql.format(emi_dates=emi_date_placeholders), { "application_id": application_id, "deleted_at": now, }, ) # ---------------- Prepartpayment_rps_logs ---------------- # insert_prepartpayment_rps_logs_sql = """ insert into prepartpayment_rps_logs (application_id, old_rps, changed_rps, final_rps, user_id, created_at, updated_at) values (:application_id, :old_rps, :changed_rps, :final_rps, :user_id, :created_at, :updated_at); """ # Clean dataframe before converting to json existing_rps = existing_rps[["emi_payment_id", "application_id", "emi_date", "pre_emi_amount", "interest_due", "principal_due", "emi_amount", "bounce_charges", "delay_penalty", "charges_waived", "opening_pos", "closing_pos", "foreclosure_amt"]] rps_to_be_updated = rps_to_be_updated[["emi_payment_id", "installment_number", "application_id", "emi_date", "pre_emi_amount", "interest_due", "principal_due", "emi_amount", "bounce_charges", "delay_penalty", "charges_waived", "principal_outstanding", "foreclosure_amount", "opening_pos1", "closing_pos1"]] final_rps = final_rps[["emi_payment_id", "installment_number", "application_id", "emi_date", "pre_emi_amount", "interest_due", "principal_due", "emi_amount", "bounce_charges", "delay_penalty", "charges_waived", "principal_outstanding", "foreclosure_amount", "opening_pos1", "closing_pos1"]] old_rps_json = df_to_mysql_json_safe(existing_rps) changed_rps_json = df_to_mysql_json_safe(rps_to_be_updated) final_rps_json = df_to_mysql_json_safe(final_rps) execute_query( insert_prepartpayment_rps_logs_sql, { "application_id": application_id, "old_rps": old_rps_json, "changed_rps": changed_rps_json, "final_rps": final_rps_json, "user_id": data.get("user_id", 107), "created_at": now, "updated_at": now, }, ) return True async def function1(data): request_time = datetime.now() logger.send_log_event(f"INFO: In prepartpayment_v2 function - Request received:\n{data}") application_id = data.get("application_id") prepart_amount = data.get("amount") prepart_date = data.get("payment_date") prepart_payment_date = datetime.strptime(prepart_date, "%Y-%m-%d").date() negative_bpi_qry = """ select application_id from emi_payments where pre_emi_amount < 0 and application_id = :application_id """ negative_bpi_df = execute_query(negative_bpi_qry, params = {'application_id': application_id}) if isinstance(negative_bpi_df, str): logger.send_log_event(f"ERROR: {negative_bpi_df}") return {"status": "ERROR", "message": negative_bpi_df} if isinstance(negative_bpi_df, pd.DataFrame) and not negative_bpi_df.empty: logger.send_log_event(f"ERROR: Application_id:{application_id} has -ve pre_emi.") return {"status": "ERROR", "message": f"Application_id:{application_id} has -ve BPI. Pre part payment not allowed."} compute_pos_data_dict = compute_pos(application_id) existing_rps = pd.DataFrame(compute_pos_data_dict.get("df_reconciled")) current_pos = compute_pos_data_dict.get("principal_outstanding") last_paid_emi_date = compute_pos_data_dict.get("last_paid_emi_date") last_paid_emi_amount = compute_pos_data_dict.get("last_paid_emi_amount") annual_interest_rate = compute_pos_data_dict.get("annual_interest_rate") total_tenure = compute_pos_data_dict.get("total_tenure") first_unpaid_emi_date = compute_pos_data_dict.get("first_unpaid_emi_date") # Validation Block if pd.notna(first_unpaid_emi_date) and first_unpaid_emi_date <= current_date: return {"status": "FAILED", "message": f"Application_id {application_id} has unpaid EMI(s) on {first_unpaid_emi_date}"} if compute_pos_data_dict.get("atleast_one_emi_paid") == False: return {"status": "FAILED", "message": f"application_id {application_id} has no paid emi(s). Atleast 1 EMI should be paid before pre-part payment."} if compute_pos_data_dict.get("principal_outstanding") <= 0: return {"status": "FAILED", "message": f"POS for application_id {application_id} = 0."} if prepart_amount > compute_pos_data_dict.get("total_outstanding"): return {"status": "FAILED", "message": f"Pre-part payment amount {prepart_amount} is greater than total outstanding amount {compute_pos_data_dict.get('total_outstanding')}."} if prepart_amount <= last_paid_emi_amount: return {"status": "FAILED", "message": f"Pre-part payment amount {prepart_amount} is less than or equal to EMI amount {round(last_paid_emi_amount)}."} foreclosure = compute_pos_data_dict.get("foreclosure_amount", 0) if (foreclosure - 100) <= prepart_amount < foreclosure: return {"status": "FAILED", "message": f"Pre-part payment amount {prepart_amount} is equal to or within ₹100 of the foreclosure amount. Pre-Part Payment cannot be initiated. Please proceed with Foreclosure.."} if prepart_payment_date <= last_paid_emi_date: return {"status": "FAILED", "message": f"Pre-part payment date {prepart_date} should be after the last paid EMI date: {last_paid_emi_date}."} if data.get("type") == "emi": final_rps = reduce_emi(existing_rps, annual_interest_rate, current_pos, prepart_payment_date, prepart_amount, total_tenure, first_unpaid_emi_date, last_paid_emi_date) # show(final_rps) if data.get("is_preview_flag") == 1: float_cols = final_rps.select_dtypes(include=["float", "float64"]).columns final_rps[float_cols] = final_rps[float_cols].applymap(lambda x: round(x, 2) if x is not None else None) final_rps['emi_amount'] = final_rps["emi_amount"].round() final_rps.drop(columns=['opening_pos1','closing_pos1'],axis=1,inplace=True) response_time = datetime.now() elapsed_time = (response_time - request_time).total_seconds() logger.send_log_event(f"INFO: RPS generated(preview) for application_id: {application_id}. Response time: {elapsed_time:.2f} seconds.") return {"status": "SUCCESS", "message": f"Please review the RPS for application_id: {application_id}", "data": final_rps.to_dict(orient='records')} else: rps_to_be_updated = final_rps[final_rps['emi_date'] > current_date] update_tables(existing_rps, rps_to_be_updated, final_rps, data) response_time = datetime.now() elapsed_time = (response_time - request_time).total_seconds() logger.send_log_event(f"INFO: RPS updated in system for application_id: {application_id}. Response time: {elapsed_time:.2f} seconds.") return {"status": "SUCCESS", "message": f"Pre part payment done for application_id: {application_id}. RPS updated successfully."} else: final_rps = reduce_tenure(existing_rps, annual_interest_rate, current_pos, prepart_payment_date, prepart_amount, first_unpaid_emi_date, last_paid_emi_date, last_paid_emi_amount) # show(final_rps) if data.get("is_preview_flag") == 1: float_cols = final_rps.select_dtypes(include=["float", "float64"]).columns final_rps[float_cols] = final_rps[float_cols].applymap(lambda x: round(x, 2) if x is not None else None) final_rps['emi_amount'] = final_rps["emi_amount"].round() final_rps.drop(columns=['opening_pos1','closing_pos1'],axis=1,inplace=True) response_time = datetime.now() elapsed_time = (response_time - request_time).total_seconds() logger.send_log_event(f"INFO: RPS generated(preview) for application_id: {application_id}. Response time: {elapsed_time:.2f} seconds.") return {"status": "SUCCESS", "message": f"Please review the RPS for application_id: {application_id}", "data": final_rps.to_dict(orient='records')} else: rps_to_be_updated = final_rps[final_rps['emi_date'] > current_date] update_tables(existing_rps, rps_to_be_updated, final_rps, data) response_time = datetime.now() elapsed_time = (response_time - request_time).total_seconds() logger.send_log_event(f"INFO: RPS updated in system for application_id: {application_id}. Response time: {elapsed_time:.2f} seconds.") return {"status": "SUCCESS", "message": f"Pre part payment done for application_id: {application_id}. RPS updated successfully."} # if __name__ == "__main__": # data = { # "application_id": 12630, # 13446, 13306, 12474 # "payment_date": "2026-01-17", # pre-part payment date # "amount": 418, # pre-part payment amount # "is_preview_flag": 1, # 1 for preview, 0 for update # "type": "emi", # either reduce emi or reduce tenure # "user_id": 107 # } # result = asyncio.run(function1(data)) # print(result)