import ast import pandas as pd import pytz import asyncio from datetime import datetime from scripts.mysql_queries import bounce_penal_query from utils import execute_query, get_cloudwatch_logger, log_exception from scripts.blacksoil_lms.co_lending_utils import fetch_product_config_no_cache from scripts.mysql_queries import CO_LENDING_BOUNCE_PENAL_QUERY from scripts.blacksoil_lms.co_lending_update_loans_data import process_loan_updates import warnings warnings.filterwarnings('ignore') local_tz = pytz.timezone("Asia/Kolkata") logger = get_cloudwatch_logger(stream_name="co_lending_bounce_penal") def update_emi_payment_details(calculated_df, batch_size=3000): sql = """ UPDATE co_lending_emi_payments SET bounce_charges = :bounce_charges, delay_penalty = :delay_penalty, delay_days = :delay_days, total_payable = :total_payable, amount_due = :amount_due, payment_status = :payment_status, updated_at = NOW() WHERE loan_id = :loan_id AND id = :id """ all_params = [ { "loan_id": row.loan_id, "id": row.emi_payment_id, "bounce_charges": row.bounce_charges, "delay_penalty": row.delay_penalty, "delay_days": row.delay_days, "total_payable": row.total_payable, "amount_due": row.amount_due, "payment_status": row.payment_status } for row in calculated_df.itertuples(index=False) ] for i in range(0, len(all_params), batch_size): batch = all_params[i:i + batch_size] execute_query(sql, batch) def calculate_charges(df): df['emi_date'] = pd.to_datetime(df['emi_date'], format='%d-%m-%Y').dt.date df['due_date'] = pd.to_datetime(df['due_date'], format='%d-%m-%Y').dt.date def parse_payment_list(val): if isinstance(val, (bytes, bytearray)): val = val.decode('utf-8') return ast.literal_eval(val) if val else [] df['payment_list'] = df['payment_list'].apply(parse_payment_list) df.loc[:, ['bounce_charges', 'delay_days', 'delay_penalty']] = 0 df.sort_values(by='due_date', inplace=True) df['payment_date'] = pd.NaT current_date = pd.Timestamp.today().date() def get_total_pending(target_loan_id, target_due_date, before_payment_date=None, up_to_due_date=False): mask = (df['loan_id'] == target_loan_id) & (df['due_date'] == target_due_date) & (df['rps_type'] == 'orig') total_emi = df.loc[mask, 'emi_amount'].sum() total_paid = 0 for _, r in df[mask].iterrows(): for p_dict in r['payment_list']: p_date = datetime.strptime(list(p_dict.keys())[0], '%d-%m-%Y').date() if up_to_due_date: if p_date <= target_due_date: total_paid += float(list(p_dict.values())[0]) else: if p_date < before_payment_date: total_paid += float(list(p_dict.values())[0]) return total_emi - total_paid for ind in df.index: if df.at[ind, 'emi_amount'] <= 0 or df.at[ind, 'due_date'] >= current_date: continue product_id = df.at[ind, 'product_id'] try: config_data = fetch_product_config_no_cache(product_id)[7] # The 8th returned value is the raw data dict charges = config_data.get("charges", []) except Exception as e: logger.send_log_event(f"Failed to fetch config for product {product_id}: {e}") charges = [] penal_conf = next((c for c in charges if c["charge_master_id"] == 1), None) nach_bounce_conf = next((c for c in charges if c["charge_master_id"] == 2), None) cheque_bounce_conf = next((c for c in charges if c["charge_master_id"] == 3), None) # Read the bounce_type from the DataFrame row bounce_type = str(df.at[ind, 'bounce_type']).lower().strip() # Dynamically assign the config based on the type if bounce_type == 'cheque': bounce_conf = cheque_bounce_conf else: # Default to NACH if missing or anything else bounce_conf = nach_bounce_conf rps_type = df.at[ind, 'rps_type'] # Determine bounce paid_upto_now = sum( float(list(payment.values())[0]) for payment in df.at[ind, 'payment_list'] if datetime.strptime(list(payment.keys())[0], '%d-%m-%Y').date() <= df.at[ind, 'due_date'] ) if paid_upto_now < df.at[ind, 'emi_amount']: # Bounced if bounce_conf: val = float(bounce_conf["value"]) partner_split = bounce_conf.get("partner_split") or {} lookup_type = 'partner_a' if rps_type == 'hurdle' else rps_type split_pct = float(partner_split.get(lookup_type, 100.0)) / 100.0 if bounce_conf["charge_type"] == "PERCENTAGE": total_bounced_pending = get_total_pending(df.at[ind, 'loan_id'], df.at[ind, 'due_date'], up_to_due_date=True) df.at[ind, 'bounce_charges'] = (val / 100.0) * total_bounced_pending * split_pct else: df.at[ind, 'bounce_charges'] = val * split_pct # Calculate penal pending_amount = df.at[ind, 'emi_amount'] df.at[ind, 'delay_days'] = 0 df.at[ind, 'delay_penalty'] = 0 df.at[ind, 'payment_date'] = pd.NaT for payment_dict in df.at[ind, 'payment_list']: payment_date = datetime.strptime(list(payment_dict.keys())[0], '%d-%m-%Y').date() amount_paid = float(list(payment_dict.values())[0]) if pd.isna(df.at[ind, 'payment_date']): delay_days = max(0, (payment_date - df.at[ind, 'due_date']).days) else: delay_days = max(0, (payment_date - df.at[ind, 'payment_date']).days - 1) df.at[ind, 'delay_days'] += round(delay_days) if pending_amount > 0 else 0 if penal_conf and pending_amount > 0 and delay_days > 0: val = float(penal_conf["value"]) partner_split = penal_conf.get("partner_split") or {} lookup_type = 'partner_a' if rps_type == 'hurdle' else rps_type split_pct = float(partner_split.get(lookup_type, 100.0)) / 100.0 if penal_conf["charge_type"] == "PERCENTAGE": # Option A: calculate on TOTAL pending across all RPS rows, then split total_pending_at_payment = get_total_pending(df.at[ind, 'loan_id'], df.at[ind, 'due_date'], before_payment_date=payment_date) df.at[ind, 'delay_penalty'] += round(total_pending_at_payment * (val / 100.0) * delay_days * split_pct) else: # One time flat penalty per EMI if it's delayed if df.at[ind, 'delay_penalty'] == 0: df.at[ind, 'delay_penalty'] = val * split_pct df.at[ind, 'payment_date'] = payment_date if payment_date > df.at[ind, 'due_date'] else pd.NaT pending_amount -= amount_paid return df def process_bounce_penal_updates(loan_ids=None): try: query = CO_LENDING_BOUNCE_PENAL_QUERY params = {} if loan_ids: format_strings = ','.join([':l_' + str(j) for j in range(len(loan_ids))]) where_clause = f"WHERE ep.deleted_at IS NULL AND ep.loan_id IN ({format_strings})" query = query.replace("WHERE ep.deleted_at IS NULL", where_clause) params = {f'l_{j}': lid for j, lid in enumerate(loan_ids)} ep_df = execute_query(query, params=params) if ep_df.empty: logger.send_log_event("INFO: No co-lending EMI payment records found.") else: calculated_df = (ep_df.groupby("loan_id", group_keys=False) .apply(calculate_charges).reset_index(drop=True)) # Recalculate full EMI amounts and status calculated_df['total_payable'] = ( calculated_df['emi_amount'] + calculated_df['bpi_due'].fillna(0) + calculated_df['bounce_charges'] + calculated_df['delay_penalty'] - calculated_df['charges_waived'].fillna(0) ) calculated_df['amount_due'] = (calculated_df['total_payable'] - calculated_df['total_paid']).clip(lower=0) def get_payment_status(row): if row['amount_due'] <= 0: return 'success' if row['total_paid'] > 0 and row['amount_due'] > 0: return 'partial' return 'pending' calculated_df['payment_status'] = calculated_df.apply(get_payment_status, axis=1) update_emi_payment_details(calculated_df, batch_size=3000) logger.send_log_event(f"INFO: Bounce & penalty updated successfully for {len(ep_df['loan_id'].unique())} loans.") process_loan_updates(loan_ids=loan_ids) # Automatically cascade to update POS/DPD metrics except Exception as e: log_exception(logger, f"Error in bounce penal update: {e}") if __name__ == '__main__': process_bounce_penal_updates()