import os, math import pandas as pd import numpy as np import numpy_financial as npf import pytz from datetime import datetime from utils import execute_query, get_cloudwatch_logger, log_exception from scripts.mysql_queries import accrued_revenue_loans_query, accrued_revenue_emi_payments_query, upsert_accrued_revenue_query, cleanup_accrued_revenue_query def generate_accrued_revenue_report(app_id_list=None): logger = get_cloudwatch_logger(stream_name="accrued_revenue_rps_update") local_tz = pytz.timezone("Asia/Kolkata") try: logger.send_log_event("Fetching loans...") loans_df = execute_query(accrued_revenue_loans_query) if loans_df.empty: logger.send_log_event("No loans found.") return if app_id_list: app_ids = app_id_list loans_df = loans_df[loans_df['application_id'].isin(app_ids)] else: app_ids = loans_df['application_id'].tolist() logger.send_log_event(f"Fetching EMI payments for {len(app_ids)} loans...") chunk_size = 5000 all_emis = [] for i in range(0, len(app_ids), chunk_size): chunk_ids = app_ids[i:i + chunk_size] query_ids = f"({','.join(map(str, chunk_ids))})" query = accrued_revenue_emi_payments_query.replace(':application_ids', query_ids) emis_df = execute_query(query) if not emis_df.empty: all_emis.append(emis_df) if not all_emis: logger.send_log_event("No EMI payments found.") return emis_master_df = pd.concat(all_emis, ignore_index=True) emis_master_df['emi_date'] = pd.to_datetime(emis_master_df['emi_date'], errors='coerce') emi_groups = emis_master_df.groupby('application_id') all_db_rows = [] logger.send_log_event("Calculating schedules...") for idx, loan in loans_df.iterrows(): try: app_id = loan['application_id'] if app_id not in emi_groups.groups: continue emi_data = emi_groups.get_group(app_id).sort_values('emi_date') approved_amount = float(loan['approved_amount'] or 0) subvention_rate = float(loan['subvention'] or 0) net_pf = float(loan['pf_excl_gst'] or 0) # 1. Base Variables loan_amount = float(loan['net_loan_amount'] or 0) total_subvention_pool = float(loan['total_subvention_pool'] or 0) # 2. Starting Outlays net_principal_outlay = float(loan['net_principal_outlay'] or 0) eir_outlay = float(loan['outstanding_as_per_eir'] or 0) # 3. Build Cashflows cf_in_standard = [] cf_in_eir = [] for emi_row in emi_data.itertuples(index=False): base_emi = float(getattr(emi_row, 'emi_amount', 0) or 0) pre_emi = float(getattr(emi_row, 'pre_emi_amount', 0) or 0) cashflows = float(getattr(emi_row, 'cashflows', base_emi + pre_emi)) if pd.isna(cashflows): cashflows = base_emi + pre_emi if base_emi > 0: cf_in_standard.append(base_emi) cf_in_eir.append(cashflows) if not cf_in_standard: cf_in_standard = cf_in_eir cf_customer = [-loan_amount] + cf_in_standard cf_company = [-net_principal_outlay] + cf_in_standard cf_eir = [-eir_outlay] + cf_in_eir # 4. Calculate IRRs pre_part_flag = loan.get('pre_part_flag', 0) emi_value = float(loan.get('emi_value') or 0) remaining_tenure = int(loan.get('remaining_emi') or 0) if pre_part_flag == 1 and remaining_tenure > 0 and emi_value > 0: cust_irr = npf.rate(nper=remaining_tenure, pmt=-emi_value, pv=loan_amount, fv=0, when='end') comp_irr = npf.rate(nper=remaining_tenure, pmt=-emi_value, pv=net_principal_outlay, fv=0, when='end') cust_irr = 0.0 if math.isnan(cust_irr) else cust_irr comp_irr = 0.0 if math.isnan(comp_irr) else comp_irr eir_irr = npf.irr(cf_eir) eir_irr = 0.0 if math.isnan(eir_irr) or eir_irr < 0 else eir_irr else: cust_irr = npf.irr(cf_customer) cust_irr = 0.0 if math.isnan(cust_irr) or cust_irr < 0 else cust_irr comp_irr = npf.irr(cf_company) comp_irr = 0.0 if math.isnan(comp_irr) or comp_irr < 0 else comp_irr eir_irr = npf.irr(cf_eir) eir_irr = 0.0 if math.isnan(eir_irr) or eir_irr < 0 else eir_irr interest_rate = float(loan.get('interest_rate') or 0.0) if interest_rate == 0.0: cust_irr = 0.0 # 5. Amortization Loop opening_standard = loan_amount opening_net = net_principal_outlay opening_eir = eir_outlay subvention_balance = total_subvention_pool disbursement_date = loan.get('disbursed_date') if pd.notna(disbursement_date): if isinstance(disbursement_date, str): disbursement_date = pd.to_datetime(disbursement_date, errors='coerce') if pd.notna(disbursement_date): disb_date_str = disbursement_date.strftime('%Y-%m-%d') else: disb_date_str = '' now = datetime.now(local_tz).strftime('%Y-%m-%d %H:%M:%S') valid_dates_for_app = [] if disb_date_str: valid_dates_for_app.append(disb_date_str) all_db_rows.append({ 'application_id': app_id, 'emi_date': disb_date_str, 'net_loan_amount': loan_amount, 'customer_interest_rate': str(cust_irr), 'company_irr': str(comp_irr), 'eir': str(eir_irr), 'opening_outstanding': round(opening_standard, 2), 'opening_net_principal_outlay': round(opening_net, 2), 'opening_subvention_balance': 0.0, 'principal_payable': 0.0, 'emi_amount': 0.0, 'closing_outstanding': round(opening_standard, 2), 'revenue': 0.0, 'interest': 0.0, 'subvention': 0.0, 'processing_fees': round(net_pf, 2), 'bounce_charges': 0.0, 'penal_interest': 0.0, 'subvention_balance': 0.0, 'closing_net_principal_outlay': round(opening_net, 2), 'opening_outstanding_as_per_eir': round(opening_eir, 2), 'interest_as_per_eir': 0.0, 'outflows': 0.0, 'closing_outstanding_as_per_eir': round(opening_eir, 2), 'now': now }) previous_emi_date = disbursement_date if pd.notna(disbursement_date) else pd.NaT previous_emipayment_id = None for emi_row in emi_data.itertuples(index=False): emi_payment_id = int(getattr(emi_row, 'emi_payment_id', 1)) current_date = pd.to_datetime(getattr(emi_row, 'emi_date', pd.NaT), errors='coerce') row_opening_std = opening_standard row_opening_net = opening_net row_opening_eir = opening_eir row_opening_subv = subvention_balance if pd.notna(previous_emi_date) and emi_payment_id == 0: days_diff = max(0, (current_date - previous_emi_date).days) if pd.notna(current_date) else 0 cust_interest = opening_standard * days_diff * ((cust_irr * 12) / 360) total_revenue = opening_net * days_diff * ((comp_irr * 12) / 360) eir_interest = opening_eir * eir_irr subvention_recognized = total_revenue - cust_interest base_emi = float(getattr(emi_row, 'emi_amount', 0) or 0) total_cf = base_emi principal_payable = max(0, base_emi - cust_interest) subvention_balance -= subvention_recognized opening_standard -= principal_payable opening_net = opening_net - principal_payable + subvention_recognized opening_eir = opening_eir - total_cf + eir_interest else: base_emi = float(getattr(emi_row, 'emi_amount', 0) or 0) pre_emi = float(getattr(emi_row, 'pre_emi_amount', 0) or 0) total_cf = base_emi + pre_emi if previous_emipayment_id == 0 and pd.notna(previous_emi_date): days_diff = max(0, (current_date - previous_emi_date).days) if pd.notna(current_date) else 0 cust_interest = opening_standard * days_diff * ((cust_irr * 12) / 360) total_revenue = opening_net * days_diff * ((comp_irr * 12) / 360) eir_interest = opening_eir * eir_irr else: cust_interest = opening_standard * cust_irr total_revenue = opening_net * comp_irr eir_interest = opening_eir * eir_irr if base_emi > 0: subvention_recognized = total_revenue - cust_interest principal_payable = base_emi - cust_interest subvention_balance -= subvention_recognized opening_standard -= principal_payable opening_net = opening_net - principal_payable + subvention_recognized else: cust_interest = 0.0 total_revenue = 0.0 subvention_recognized = 0.0 principal_payable = 0.0 opening_eir = opening_eir - total_cf + eir_interest is_success = str(getattr(emi_row, 'payment_status', '')).lower() == 'success' penal_interest = float(getattr(emi_row, 'delay_penalty', 0) or 0) if is_success else 0.0 final_bounce = float(getattr(emi_row, 'bounce_charges', 0) or 0) if is_success else 0.0 if pd.notna(current_date): previous_emi_date = current_date previous_emipayment_id = emi_payment_id emi_dt = getattr(emi_row, 'emi_date', '') if emi_dt: if type(emi_dt) != str: emi_dt = emi_dt.strftime('%Y-%m-%d') valid_dates_for_app.append(emi_dt) all_db_rows.append({ 'application_id': app_id, 'emi_date': emi_dt, 'net_loan_amount': loan_amount, 'customer_interest_rate': str(cust_irr), 'company_irr': str(comp_irr), 'eir': str(eir_irr), 'opening_outstanding': round(row_opening_std, 2), 'opening_net_principal_outlay': round(row_opening_net, 2), 'opening_subvention_balance': round(row_opening_subv, 2), 'principal_payable': round(principal_payable, 2), 'emi_amount': round(base_emi, 2), 'closing_outstanding': round(opening_standard, 2), 'revenue': round(total_revenue, 2), 'interest': round(cust_interest, 2), 'subvention': round(subvention_recognized, 2), 'processing_fees': 0.0, 'bounce_charges': round(final_bounce, 2), 'penal_interest': round(penal_interest, 2), 'subvention_balance': round(subvention_balance, 2), 'closing_net_principal_outlay': round(opening_net, 2), 'opening_outstanding_as_per_eir': round(row_opening_eir, 2), 'interest_as_per_eir': round(eir_interest, 2), 'outflows': round(total_cf, 2), 'closing_outstanding_as_per_eir': round(opening_eir, 2), 'now': now }) if valid_dates_for_app: cleanup_q = cleanup_accrued_revenue_query.replace(':valid_dates', "('" + "','".join(valid_dates_for_app) + "')") execute_query(cleanup_q, params={'application_id': app_id, 'now': now}) except Exception as e: logger.send_log_event(f"Error processing application_id {app_id}: {str(e)}") continue if all_db_rows: logger.send_log_event(f"Upserting {len(all_db_rows)} rows to the database in chunks...") db_chunk_size = 5000 for i in range(0, len(all_db_rows), db_chunk_size): chunk = all_db_rows[i:i + db_chunk_size] execute_query(upsert_accrued_revenue_query, params=chunk) logger.send_log_event("Upsert complete.") except Exception as e: log_exception(logger, "Error generating accrued revenue report") raise if __name__ == '__main__': generate_accrued_revenue_report()