import pandas as pd import numpy as np import numpy_financial as npf import math, pytz, asyncio, os from datetime import datetime from utils import execute_query, apply_waiver_allocation, get_cloudwatch_logger from scripts.mysql_queries import loan_schedule_query_prepart, net_loan_irr_query # from pandasgui import show logger = get_cloudwatch_logger(stream_name="pre_part_payment") local_tz = pytz.timezone("Asia/Kolkata") now = datetime.now(local_tz).strftime('%Y-%m-%d %H:%M:%S') # UPLOAD_DIR = "pre_part_payment_uploads" # os.makedirs(UPLOAD_DIR, exist_ok=True) 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 loan_amortization(df, loan_principal, annual_interest_rate, tenure, last_emi_date): beginning_balance = loan_principal results = [] previous_emi_date = pd.NaT previous_emipayment_id = None for row in df.itertuples(index=True): pre_emi = getattr(row, 'pre_emi_amount', 0) opening_pos = beginning_balance emi_date = getattr(row, 'emi_date', pd.NaT) emi_date = pd.to_datetime(emi_date, format='%d-%m-%Y') emi_payment_id = getattr(row, 'emi_payment_id', 0) if getattr(row, 'emi_amount', 0) != 0: # If EMI amount is present then calculate amortization for that EMI, else Pre-EMI interest_comp = round(opening_pos * 30 * (annual_interest_rate / 360), 2) if pd.notna(previous_emi_date) and emi_payment_id == 0: # Calculation for pre part payment row days_since_old_pos = (emi_date - previous_emi_date).days # calculate days_diff betn prev emi date and current emi date(prepartpayment date) interest_comp = round(opening_pos * days_since_old_pos * (annual_interest_rate / 360), 2) # calculate interest for above days emi_amount = getattr(row, 'emi_amount', 0) # get emi amount as is from db table principal_comp = round(max(emi_amount - interest_comp,0), 2) closing_pos = opening_pos - principal_comp beginning_balance = closing_pos else: if previous_emipayment_id == 0: # if previous row was of prepartpayment days_after_new_pos = (emi_date - previous_emi_date).days # calculate days_diff betn prev emi date(prepartpayment date) and current emi date interest_comp = round(opening_pos * days_after_new_pos * (annual_interest_rate / 360), 2) # calculate interest fro above days emi_amount = getattr(row, 'emi_amount', 0) principal_comp = round(max(emi_amount - interest_comp,0), 2) closing_pos = opening_pos - principal_comp beginning_balance = closing_pos else: if emi_date.date() == last_emi_date: # Last EMI amount adjustment interest_comp = round(opening_pos * 30 * (annual_interest_rate / 360), 2) emi_amount = round(opening_pos + interest_comp, 2) principal_comp = round(opening_pos, 2) closing_pos = 0 else: interest_comp = round(opening_pos * 30 * (annual_interest_rate / 360), 2) emi_amount = npf.pmt(annual_interest_rate / 12, tenure, -loan_principal, 0) if annual_interest_rate != 0.00 else loan_principal / tenure principal_comp = round(max(emi_amount - interest_comp,0), 2) closing_pos = opening_pos - principal_comp beginning_balance = closing_pos # Update the beginning balance for the next row interest_component = interest_comp principal_component = principal_comp else: # Only Pre-EMI is present in current row emi_amount, opening_pos, closing_pos, interest_component, principal_component = 0, 0, 0, 0, 0 # Update previous_emi_date & previous emi_payment_id to current values for the next iteration previous_emi_date = emi_date previous_emipayment_id = emi_payment_id results.append({'emi_payment_id': getattr(row, 'emi_payment_id', 0), 'application_id': row.application_id, 'emi_date': row.emi_date, 'due_date': row.due_date, 'pre_emi_amount': pre_emi, 'interest_due': interest_component, 'principal_due': principal_component, 'emi_amount': emi_amount, 'opening_pos': opening_pos, 'closing_pos': closing_pos, }) return pd.DataFrame(results) def insert_loan_schedule_v2_table(df, batch_size=500): """ Insert loan schedule records into loan_schedule_v2 table. """ sql_query = """ INSERT INTO loan_schedule_v2 (application_id, emi_date, due_date, pre_emi_amount, interest_due, principal_due, emi_amount, opening_pos, closing_pos, deleted_at, created_at, updated_at, emi_payment_id ) VALUES ( :application_id, :emi_date, :due_date, :pre_emi_amount, :interest_due, :principal_due, :emi_amount, :opening_pos, :closing_pos, :deleted_at, :created_at, :updated_at, :emi_payment_id ); """ params = [] for row in df.itertuples(index=False): params.append({ "application_id": row.application_id, "emi_date": row.emi_date, "due_date": row.due_date, "pre_emi_amount": row.pre_emi_amount, "interest_due": row.interest_due, "principal_due": row.principal_due, "emi_amount": row.emi_amount, "opening_pos": row.opening_pos, "closing_pos": row.closing_pos, "deleted_at": None, "created_at": now, "updated_at": now, "emi_payment_id": row.emi_payment_id }) # Batch insert total_rows = len(params) for i in range(0, total_rows, batch_size): batch = params[i:i + batch_size] execute_query(sql_query, batch) return True def generateLoanSchedule(application_id): logger.send_log_event(f"INFO: In generateLoanSchedule for application_id: {application_id}.") ep_df = execute_query(loan_schedule_query_prepart,params={'application_ids': tuple([application_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: {application_id}"} ep_df = apply_waiver_allocation(ep_df) 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 tenure = row["tenure"] last_emi_date = row["last_emi_date"] 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}.") loan_amortization_df = loan_amortization(ep_df, net_loan_amount, annual_interest_rate, tenure, last_emi_date) loan_amortization_df = loan_amortization_df[loan_amortization_df['emi_payment_id'] > 0] # only keep actual emi rows loan_amortization_df["emi_date"] = pd.to_datetime(loan_amortization_df["emi_date"], format="%d-%m-%Y", errors="coerce").dt.strftime("%Y-%m-%d") loan_amortization_df["due_date"] = pd.to_datetime(loan_amortization_df["due_date"], format="%d-%m-%Y", errors="coerce").dt.strftime("%Y-%m-%d") # file_name = f"loan_amort_sheet_{application_id}.csv" # file_path = os.path.join(UPLOAD_DIR, file_name) # loan_amortization_df.to_csv(file_path, index=False) # show(loan_amortization_df) insert_loan_schedule_v2_table(loan_amortization_df) logger.send_log_event(f"INFO: Loan Schedule created for application_id in loan_schedule_v2: {application_id}.") # if __name__ == "__main__": # print(generateLoanSchedule(13170))