import pandas as pd import numpy as np import numpy_financial as npf from utils import execute_query, get_cloudwatch_logger, log_exception, apply_waiver_allocation, reconciliation from scripts.mysql_queries import emi_wise_payments_query, net_loan_irr_query, pf_dp_due_query, settlement_reconcile_v2_query from scripts.hm_update_data import hm_loan_schedule logger = get_cloudwatch_logger(stream_name="foreclosure_breakup") import asyncio # from pandasgui import show async def function1(data): """ returns loan schedule with updated payments and principal os, interest os, bounce and penal due for foreclosure. works for single application_id. """ application_id = data.get("application_id") chk_lender_qry = """ select lender_id from consumer_application where id in :application_ids """ chk_lender_df = execute_query(chk_lender_qry, params = {'application_ids': tuple([application_id,])}) if isinstance(chk_lender_df, str): return {"status": "ERROR", "message": chk_lender_df} if isinstance(chk_lender_df, pd.DataFrame) and chk_lender_df.empty: print(f"No data found in consumer_application table for Application_id: {application_id}") logger.send_log_event(f"No data found in consumer_application table for Application_id: {application_id}") return {"status": "FAILED", "message": f"No data found in consumer_application table for Application_id: {application_id}"} if chk_lender_df.iloc[0, 0] == 10: # if lender homeville then call separate hm_loan_schedule script as homeville rps is fetched via Homeville API hm_response = await hm_loan_schedule(application_id) return hm_response else: # Get EMI payments data for given application_id ep_df = execute_query(emi_wise_payments_query,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) # apply charges_waived # show(ep_df) # logger.send_log_event(f"INFO: RPS data(from emi_payments) fetched for application_id: {application_id}.") # Get net loan amount and monthly_irr_rate which will be used to reconcile payments 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."} net_loan_amount = float(irr_df['net_loan_amount'].iloc[0] if data.get("net_loan_amount") is None else data.get("net_loan_amount")) # 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. loan_type = irr_df['loan_type'].iloc[0] if loan_type == 'subvention': irr_rate = 0.00 else: if irr_df['pre_part_flag'].iloc[0] == 1: # if case is prepartpayment then use emi_calculation.rate column value tenure = irr_df['tenure'].iloc[0] emi_value = irr_df['emi_value'].iloc[0] irr_rate = npf.rate(nper = tenure, pmt = - emi_value, pv = net_loan_amount, fv = 0, when = 'end') else: # if case is interest based and prepartpayment not done then calculate monthly interest rate. net_loan_amount_list = [-int(net_loan_amount)] emis_list = irr_df['emis_list'].iloc[0].split(',') emis_list = [int(emi) for emi in emis_list] cash_flows = net_loan_amount_list + emis_list irr_rate = npf.irr(cash_flows) df_reconciled = reconciliation(ep_df, net_loan_amount, irr_rate) # logger.send_log_event(f"INFO: Payments reconciled") # df_reconciled.to_csv('final.csv',index=False) # show(df_reconciled) df_reconciled['emi_date'] = pd.to_datetime(df_reconciled['emi_date'], format='%d-%m-%Y').dt.date df_reconciled['due_date'] = pd.to_datetime(df_reconciled['due_date'], format='%d-%m-%Y').dt.date current_date = pd.Timestamp.today().date() ##### Foreclosure Breakup ##### pre_emi_due = round(df_reconciled['pre_emi_amount'].sum() - df_reconciled['pre_emi_paid'].sum()) # principal_outstanding = round(max(0,df_reconciled['principal_due'].sum() - df_reconciled['principal_comp_paid'].sum()),2) principal_outstanding = round(max(net_loan_amount - df_reconciled['principal_paid'].sum(),0),2) # Get interest due. formula = (interest due upto current + one future interest) - interest paid past_emis = df_reconciled[(df_reconciled['emi_date'] <= current_date) & (df_reconciled['emi_amount'] > 0)] 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) interest_outstanding = round(max(0,interest_due_until_now - interest_paid),2) # # Morat Due # morat_rows = df_reconciled[df_reconciled['morat_interest'] > 0] # Filter rows with morat_interest > 0 # morat_due = round( # max(0, morat_rows['morat_interest'].sum() - morat_rows['total_paid'].sum()), 2 # ) if not morat_rows.empty else 0.0 bounce_outstanding = max(0,round(df_reconciled['bounce_charges'].sum() - df_reconciled['bounce_paid'].sum(),2)) penal_outstanding = max(0,round(df_reconciled['delay_penalty'].sum() - df_reconciled['penal_paid'].sum(),2)) total_loan_outstanding = pre_emi_due + principal_outstanding + interest_outstanding + bounce_outstanding + penal_outstanding # Find Processing fee due & Down payment due pf_dp_df = execute_query(pf_dp_due_query, params = {'application_ids': tuple([application_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: return {"status": "FAILED", "message": f"No data found in emi_calculation table for Application_id: {application_id}"} pf_dp_df.select_dtypes(include=['number']).fillna(0) # Add total_amount_due & foreclosure amount for each EMI # df_reconciled['total_amount_due'] = ((df_reconciled['pre_emi_amount'] + df_reconciled['emi_amount'] + df_reconciled['bounce_charges'] # + df_reconciled['delay_penalty']) - df_reconciled['charges_waived'] - df_reconciled['total_amount_paid'] # ).clip(lower=0) foreclosure_amt_if_unpaid = ( # add pos + current interest due(if not paid) + future interest df_reconciled['closing_pos'] + (df_reconciled['interest_due'] - df_reconciled['interest_paid']) + df_reconciled['interest_due'].shift(-1).fillna(0) ) condition = ( # if emi not fully paid then above amount else show closing pos ((df_reconciled['interest_due'] + df_reconciled['principal_due']) - (df_reconciled['interest_paid'] + df_reconciled['principal_paid']) ).round() > 0 ) # Set foreclosure amount based on condition df_reconciled['foreclosure_amount'] = np.round(np.where(condition, foreclosure_amt_if_unpaid, df_reconciled['closing_pos']),2) df_reconciled.loc[df_reconciled.index[-1], 'foreclosure_amount'] = 0 # Set last amount as 0 # Add payment status column paymentconditions = [ df_reconciled['total_amount_paid'] == 0, df_reconciled['total_amount_paid'] >= df_reconciled['total_amount_due'] ] choices = [ 'pending', # total_amount_due = 0 'success' # total_amount_paid >= total_amount_due ] # default is 'PARTIAL' df_reconciled['payment_status'] = np.select(paymentconditions, choices, default='partial') # if emi_payment_id = 0 then prepartpayment else emi df_reconciled['emi_type'] = np.where(df_reconciled['emi_payment_id'] == 0, 'prepartpayment', 'emi') # convert empty date to blank df_reconciled['emi_date'] = df_reconciled['emi_date'].fillna('') df_reconciled['due_date'] = df_reconciled['due_date'].fillna('') # convert empty date to blank df_reconciled.loc[df_reconciled['emi_type'] == 'prepartpayment', 'total_amount_due'] = 0 # clip -ve values to 0 df_reconciled['opening_pos'] = df_reconciled['opening_pos'].clip(lower=0) df_reconciled['closing_pos'] = df_reconciled['closing_pos'].clip(lower=0) df_reconciled['foreclosure_amount'] = df_reconciled['foreclosure_amount'].clip(lower=0) df_reconciled['total_amount_due'] = (df_reconciled['total_amount_due'] - df_reconciled['total_amount_paid']).clip(lower=0) # subract settled amounts settled_amounts = execute_query(settlement_reconcile_v2_query, params = {'application_ids': tuple([application_id])}) if isinstance(settled_amounts, str): return {"status": "ERROR", "message": settled_amounts} 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 else: settled_amounts = settled_amounts.iloc[0].fillna(0).to_dict() settled_principal_amount = settled_amounts.get('settled_principal_amount') settled_interest_amount = settled_amounts.get('settled_interest_amount') settled_bounce_amount = settled_amounts.get('settled_bounce_amount') settled_penal_amount = settled_amounts.get('settled_penal_amount') principal_outstanding -= settled_principal_amount interest_outstanding -= settled_interest_amount interest_due -= settled_interest_amount bounce_outstanding -= settled_bounce_amount penal_outstanding -= settled_penal_amount 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(bounce_outstanding), "Penal charges due": float(penal_outstanding), "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) } # If the loan is foreclosed, truncate the returned schedule (data) at the foreclosure payment # date (inclusive) so the loan schedule / SOA stop at foreclosure. The dues above are computed # on the FULL schedule and are unaffected. Both legacy monolith and new # granular foreclosure rows are recognised, so flows that run pre-foreclosure # or use only 'message' (settlement, allocation, foreclosure amount) are untouched. fc_date_df = execute_query( """ select max(payment_date) as foreclosure_date from payment_transactions where application_id in :application_ids and payment_for in ('Foreclosure', 'foreclosure_bpi', 'foreclosure_interest', 'foreclosure_principal', 'foreclosure_bounce', 'foreclosure_penal', 'foreclosure_over_payment') and deleted_at is null and is_deleted_at is null and refunded_at is null """, params={'application_ids': tuple([application_id])} ) foreclosure_payment_date = fc_date_df['foreclosure_date'].iloc[0] if isinstance(fc_date_df, pd.DataFrame) and not fc_date_df.empty else None if foreclosure_payment_date is not None and pd.notna(foreclosure_payment_date): foreclosure_payment_date = pd.to_datetime(foreclosure_payment_date).date() emi_dates = pd.to_datetime(df_reconciled['emi_date'], errors='coerce').dt.date # keep rows on/before the foreclosure date; keep undated rows (e.g. pre-part-payment) as-is df_reconciled = df_reconciled[(emi_dates.isna()) | (emi_dates <= foreclosure_payment_date)] # Fetch the event rows for the inline foreclosure row. Legacy cases use # the monolith amount; split-only cases use the sum of their granular rows. fc_details_df = execute_query( """ select payment_for, amount, cf_transaction_id from payment_transactions where application_id in :application_ids and payment_for in ('Foreclosure', 'foreclosure_bpi', 'foreclosure_interest', 'foreclosure_principal', 'foreclosure_bounce', 'foreclosure_penal', 'foreclosure_over_payment') and deleted_at is null and is_deleted_at is null and refunded_at is null """, params={'application_ids': tuple([application_id])} ) fc_amount = 0 fc_principal = 0 fc_interest = 0 if isinstance(fc_details_df, pd.DataFrame) and not fc_details_df.empty: monolith_rows = fc_details_df[fc_details_df['payment_for'] == 'Foreclosure'] for _, row in fc_details_df.iterrows(): if row['payment_for'] == 'foreclosure_principal': fc_principal = float(row['amount']) elif row['payment_for'] in ('foreclosure_interest', 'foreclosure_bpi'): fc_interest += float(row['amount']) if not monolith_rows.empty: # Historical cases retain only the legacy monolith. fc_amount = float(monolith_rows['amount'].sum()) else: fc_amount = float(fc_details_df['amount'].sum()) transaction_ids = fc_details_df['cf_transaction_id'].dropna().unique().tolist() if transaction_ids: # The PF split deliberately reuses `processing_fee`; only # include it when it belongs to this foreclosure event. pf_df = execute_query( """ select coalesce(amount, 0) as amount from payment_transactions where application_id in :application_ids and cf_transaction_id in :transaction_ids and payment_for = 'processing_fee' and deleted_at is null and is_deleted_at is null and refunded_at is null order by id desc limit 1 """, params={ 'application_ids': tuple([application_id]), 'transaction_ids': tuple(transaction_ids), } ) if isinstance(pf_df, pd.DataFrame) and not pf_df.empty: fc_amount += float(pf_df['amount'].iloc[0] or 0) fc_date_str = pd.to_datetime(foreclosure_payment_date).strftime('%d-%m-%Y') # Build a foreclosure row matching df_reconciled columns fc_row = {col: 0 for col in df_reconciled.columns} fc_row.update({ 'emi_payment_id': -1, 'application_id': application_id, 'emi_date': '', 'due_date': '', 'pre_emi_amount': 0, 'moratorium_amount': 0, 'emi_amount': 0, 'bounce_charges': 0, 'delay_penalty': 0, 'payment_status': 'success', 'charges_waived': 0, 'total_paid': fc_amount, 'total_paid_for_bpi': 0, 'payment_details': '', 'latest_payment_date': fc_date_str, 'order_date': str(foreclosure_payment_date), 'opening_pos': 0, 'closing_pos': 0, 'principal_due': fc_principal, 'interest_due': fc_interest, 'principal_paid': fc_principal, 'interest_paid': fc_interest, 'bounce_paid': 0, 'penal_paid': 0, 'total_amount_due': 0, 'total_amount_paid': fc_amount, 'foreclosure_amount': 0, 'emi_type': 'foreclosure', 'payment_date': fc_date_str, }) df_reconciled = pd.concat([df_reconciled, pd.DataFrame([fc_row])], ignore_index=True) # show(df_reconciled) df_reconciled = df_reconciled.replace({np.nan: None}) return {"status": "Success", "message": foreclosure_breakup, "data": df_reconciled.to_dict(orient='records')} # if __name__ == '__main__': # data = {"application_id": 13446} # 240452, 271254, 280211 # print(asyncio.run(function1(data)))