import json import os import requests import pandas as pd from datetime import datetime import pytz from sqlalchemy import text from functools import lru_cache from utils import execute_query, get_cloudwatch_logger, log_exception, apply_rounding logger = get_cloudwatch_logger(stream_name="co_lending_regular_allocation") _session = requests.Session() @lru_cache(maxsize=128) def fetch_product_id(loan_id): """ Fetches the product_id associated with a loan to dynamically retrieve Master LMS configs. """ sql = "SELECT product_id FROM co_lending_loans WHERE loan_id = :loan_id and deleted_at is null" df = execute_query(sql, params={"loan_id": loan_id}) if df is not None and not df.empty: return df.iloc[0]['product_id'] raise ValueError(f"product_id not found for loan_id {loan_id}") def fetch_all_rps(loan_id): """ Fetches all RPS types (orig, partner_a, partner_b, hurdle) in a single query and returns a dictionary of dataframes mapped by rps_type. """ sql = """ SELECT * FROM co_lending_emi_payments WHERE loan_id = :loan_id and deleted_at is null ORDER BY due_date ASC, installment_number ASC """ df = execute_query(sql, params={"loan_id": loan_id}) if df is None or df.empty: raise ValueError(f"No RPS found for loan_id {loan_id}") rps_dfs = { 'orig': df[df['rps_type'] == 'orig'].set_index('installment_number'), 'partner_a': df[df['rps_type'] == 'partner_a'].set_index('installment_number'), 'partner_b': df[df['rps_type'] == 'partner_b'].set_index('installment_number'), 'hurdle': df[df['rps_type'] == 'hurdle'].set_index('installment_number') } partner_a_share = 0.0 if not rps_dfs['partner_a'].empty: partner_a_share = float(rps_dfs['partner_a'].iloc[0]['schedule_percent']) / 100.0 return rps_dfs, partner_a_share from scripts.blacksoil_lms.co_lending_utils import fetch_master_config_no_cache def get_due_amount(row, component, payment_date_obj=None): if component == 'bounce': if payment_date_obj is not None: row_due_date = row['due_date'] if isinstance(row_due_date, pd.Timestamp): row_due_date = row_due_date.date() elif isinstance(row_due_date, str): row_due_date = datetime.strptime(row_due_date, "%Y-%m-%d").date() if payment_date_obj <= row_due_date: return 0.0 return float(row.get('bounce_charges', 0.0) or 0.0) if component == 'penal': if payment_date_obj is not None: row_due_date = row['due_date'] if isinstance(row_due_date, pd.Timestamp): row_due_date = row_due_date.date() elif isinstance(row_due_date, str): row_due_date = datetime.strptime(row_due_date, "%Y-%m-%d").date() if payment_date_obj <= row_due_date: return 0.0 return float(row.get('delay_penalty', 0.0) or 0.0) if pd.isna(row.get(f'{component}_due')): return 0.0 return float(row[f'{component}_due']) def fetch_historical_payments(loan_id): """ Aggregates all previously paid amounts for this loan grouped by EMI row and component type. This prevents database thrashing by fetching all history in one highly optimized query. """ sql = """ SELECT pt.co_lending_emi_payments_id, pt.payment_for, SUM(pt.amount) as paid_amount FROM co_lending_payment_transactions pt JOIN co_lending_emi_payments ep ON ep.id = pt.co_lending_emi_payments_id and ep.deleted_at is null WHERE ep.loan_id = :loan_id AND pt.deleted_at IS NULL GROUP BY pt.co_lending_emi_payments_id, pt.payment_for """ df = execute_query(sql, params={"loan_id": loan_id}) history = {} if df is not None and not df.empty: for _, row in df.iterrows(): emi_id = int(row['co_lending_emi_payments_id']) comp = str(row['payment_for']) paid = float(row['paid_amount']) if emi_id not in history: history[emi_id] = {} history[emi_id][comp] = paid return history def get_paid_amount(history, row, component): if row is None or pd.isna(row.get('id')): return 0.0 emi_id = int(row['id']) return history.get(emi_id, {}).get(component, 0.0) def allocate_independent_rps(rps_df, rps_type, cash_available, priority_sequence, history, park_advance_collection=False, payment_date_obj=None, caps_per_inst=None): """ Runs the allocation loop entirely independently for a single RPS. This strictly enforces that the provided `cash_available` pays down the schedule sequentially based on the `priority_sequence` fetched from Master LMS. Returns: allocations: List of transaction objects remaining_cash: Any leftover cash that wasn't consumed by this schedule """ allocations = [] remaining_cash = cash_available for inst_num, row in rps_df.iterrows(): if remaining_cash <= 0: break # Parked transaction cut-off logic if park_advance_collection and payment_date_obj is not None: # check if due_date is strictly greater than payment_date row_due_date = row['due_date'] if isinstance(row_due_date, pd.Timestamp): row_due_date = row_due_date.date() elif isinstance(row_due_date, str): row_due_date = datetime.strptime(row_due_date, "%Y-%m-%d").date() # assuming db format if row_due_date > datetime.today().date(): break effective_sequence = [] for comp in priority_sequence: if comp == "interest": effective_sequence.append("bpi") effective_sequence.append("interest") elif comp == "charges": effective_sequence.append("bounce") effective_sequence.append("penal") else: effective_sequence.append(comp) inst_cap = caps_per_inst.get(inst_num, 0.0) if caps_per_inst is not None else float('inf') inst_consumed = 0.0 for comp in effective_sequence: if remaining_cash <= 0 or inst_consumed >= inst_cap: break due = get_due_amount(row, comp, payment_date_obj) paid = get_paid_amount(history, row, comp) unpaid = max(0.0, due - paid) if unpaid > 0: available_for_this = min(remaining_cash, inst_cap - inst_consumed) allocatable = min(unpaid, available_for_this) if allocatable > 0: allocations.append({ "rps_type": rps_type, "co_lending_emi_payments_id": int(row['id']), "installment_number": int(inst_num), "component": comp, "amount": float(round(allocatable, 2)) }) remaining_cash -= allocatable inst_consumed += allocatable if remaining_cash > 0 and rps_type == 'orig': comp_name = "parked_transaction" if park_advance_collection else "over_payment" allocations.append({ "rps_type": rps_type, "co_lending_emi_payments_id": 0, "installment_number": 0, "component": comp_name, "amount": float(round(remaining_cash, 2)) }) remaining_cash = 0.0 return allocations, round(remaining_cash, 2) def save_to_db(loan_id, amount_paid, payment_date_str, allocations, payment_list=None, user_id=None): if payment_list is None: payment_list = {} try: p_date = datetime.strptime(payment_date_str, "%d-%m-%Y").strftime("%Y-%m-%d") except ValueError: p_date = payment_date_str # Fallback if already correctly formatted ist_now = datetime.now(pytz.timezone('Asia/Kolkata')).strftime('%Y-%m-%d %H:%M:%S') original_payment_log_id = payment_list.get("original_payment_log_id") # Insert log payment_for_val = payment_list.get("payment_for", "EMI") log_sql = """ INSERT INTO co_lending_payment_logs (loan_id, amount, payment_date, payment_for, transaction_number, transaction_mode, utr, user_id, created_at, updated_at) VALUES (:loan_id, :amount, :payment_date, :payment_for_val, :transaction_number, :transaction_mode, :utr, :user_id, :ist_now, :ist_now) """ # Since execute_query doesn't return lastrowid, we can fetch it or just not rely on it # For now, we will use engine directly for the transaction to get lastrowid from utils import engine from sqlalchemy import text try: with engine.begin() as conn: if original_payment_log_id: payment_log_id = original_payment_log_id else: result = conn.execute(text(log_sql), { "loan_id": str(loan_id), "amount": amount_paid, "payment_date": p_date, "payment_for_val": payment_for_val, "transaction_number": payment_list.get("transaction_number"), "transaction_mode": payment_list.get("transaction_mode"), "utr": payment_list.get("utr"), "user_id": user_id, "ist_now": ist_now }) payment_log_id = result.lastrowid txn_sql = text(""" INSERT INTO co_lending_payment_transactions (loan_id, co_lending_emi_payments_id, co_lending_payment_logs_id, payment_date, amount, payment_for, created_at, updated_at) VALUES (:loan_id, :emi_id, :log_id, :payment_date, :amount, :payment_for, :ist_now, :ist_now) """) parked_sql = text(""" INSERT INTO co_lending_parked_transactions (loan_id, co_lending_payment_logs_id, original_amount, available_amount, status, created_at, updated_at) VALUES (:loan_id, :log_id, :amount, :amount, 'AVAILABLE', :ist_now, :ist_now) """) total_parked_amount = sum(alloc["amount"] for alloc in allocations if alloc["component"] == "parked_transaction" and alloc["rps_type"] == "orig") if total_parked_amount > 0 and not original_payment_log_id: conn.execute(parked_sql, { "loan_id": str(loan_id), "log_id": payment_log_id, "amount": float(round(total_parked_amount, 2)), "ist_now": ist_now }) for alloc in allocations: if alloc["component"] != "parked_transaction": db_payment_for = "pre_emi" if alloc["component"] == "bpi" else alloc["component"] conn.execute(txn_sql, { "loan_id": str(loan_id), "emi_id": alloc["co_lending_emi_payments_id"], "log_id": payment_log_id, "payment_date": p_date, "amount": alloc["amount"], "payment_for": db_payment_for, "ist_now": ist_now }) return payment_log_id except Exception as e: log_exception(logger, f"Database error in save_to_db: {str(e)}") raise e def co_lending_regular_allocation(data): loan_id = data.get("loan_id") payment_list = data.get("payment_list") dry_run = data.get("dry_run", True) user_id = data.get("user_id") if not loan_id or not payment_list: return {"error": "loan_id and payment_list are required"} amount_paid = float(payment_list.get("amount", 0)) payment_date_str = payment_list.get("payment_date") try: payment_date_obj = None if payment_date_str: try: payment_date_obj = datetime.strptime(payment_date_str, "%d-%m-%Y").date() except ValueError: try: payment_date_obj = datetime.strptime(payment_date_str, "%Y-%m-%d").date() except ValueError: pass product_id = fetch_product_id(loan_id) allocation_type = data.get("allocation_type", "REGULAR_EMI") priority_sequence, rounding_config, park_advance_collection = fetch_master_config_no_cache(product_id, allocation_type=allocation_type) rps_dfs, partner_a_share = fetch_all_rps(loan_id) history = fetch_historical_payments(loan_id) # 1. Split the cash upfront based on partner_a_share # This prevents negative pending issues by decoupling Partner A and B schedules # from the 'orig' schedule, allowing them to amortize perfectly independently. r_type = rounding_config.get("emi_rounding", "ROUND") r_digits = rounding_config.get("emi_rounding_digits", 0) pa_cash = apply_rounding(amount_paid * partner_a_share, r_type, r_digits) pb_cash = apply_rounding(amount_paid - pa_cash, r_type, r_digits) all_transactions = [] # 2. Allocate independently # Each schedule engine gets its exact slice of cash and consumes its own due amounts sequentially orig_allocs, orig_rem = allocate_independent_rps(rps_dfs['orig'], 'orig', amount_paid, priority_sequence, history, park_advance_collection, payment_date_obj) pa_allocs, pa_rem = allocate_independent_rps(rps_dfs['partner_a'], 'partner_a', pa_cash, priority_sequence, history, park_advance_collection, payment_date_obj) pb_allocs, pb_rem = allocate_independent_rps(rps_dfs['partner_b'], 'partner_b', pb_cash, priority_sequence, history, park_advance_collection, payment_date_obj) # 3. Hurdle Allocation # Partner A's negotiated hurdle schedule is paid off using strictly the cash received by Partner A # Calculate exactly how much Partner A consumed per installment pa_cash_per_inst = {} for alloc in pa_allocs: if alloc["component"] not in ["parked_transaction", "over_payment"]: inst = alloc['installment_number'] pa_cash_per_inst[inst] = pa_cash_per_inst.get(inst, 0.0) + alloc['amount'] # Pass pa_cash_per_inst as caps to prevent hurdle schedule from consuming future service fee cash hurdle_allocs, hurdle_rem = allocate_independent_rps( rps_dfs['hurdle'], 'hurdle', pa_cash, priority_sequence, history, park_advance_collection, payment_date_obj, caps_per_inst=pa_cash_per_inst ) # 4. Service Fee Generation # Service fee is the difference between Partner A's allocated cash and Hurdle's allocated cash for each installment hurdle_cash_per_inst = {} hurdle_emi_ids = {} for alloc in hurdle_allocs: inst = alloc['installment_number'] if inst != 0: hurdle_cash_per_inst[inst] = hurdle_cash_per_inst.get(inst, 0.0) + alloc['amount'] hurdle_emi_ids[inst] = alloc['co_lending_emi_payments_id'] for inst, pa_allocated in pa_cash_per_inst.items(): hurdle_allocated = hurdle_cash_per_inst.get(inst, 0.0) diff = round(pa_allocated - hurdle_allocated, 2) if diff > 0: emi_id = hurdle_emi_ids.get(inst) if not emi_id: try: emi_id = int(rps_dfs['hurdle'].loc[inst, 'id']) except KeyError: emi_id = 0 hurdle_allocs.append({ "rps_type": "hurdle", "co_lending_emi_payments_id": emi_id, "installment_number": inst, "component": "service_fee", "amount": diff }) hurdle_rem = max(0.0, round(hurdle_rem - diff, 2)) # Combine everything for DB insertion all_transactions.extend(orig_allocs) all_transactions.extend(pa_allocs) all_transactions.extend(pb_allocs) all_transactions.extend(hurdle_allocs) payment_log_id = None if not dry_run: payment_log_id = save_to_db(loan_id, amount_paid, payment_date_str, all_transactions, payment_list, user_id) ist_now_date = datetime.now(pytz.timezone('Asia/Kolkata')).date() if payment_date_obj and payment_date_obj < ist_now_date: def run_recalc(): try: from scripts.blacksoil_lms.co_lending_bounce_penal import process_bounce_penal_updates process_bounce_penal_updates([str(loan_id)]) except Exception as e: print(f"Error in background bounce penal refresh: {e}") from scripts.blacksoil_lms.co_lending_historical_recalc import recalculate_historical_metrics recalculate_historical_metrics([str(loan_id)], payment_date_obj.strftime("%Y-%m-%d"), ist_now_date.strftime("%Y-%m-%d")) import threading threading.Thread(target=run_recalc).start() return { "status": "success", "dry_run": dry_run, "payment_log_id": payment_log_id, "loan_id": loan_id, "payment_amount": amount_paid, "partner_a_cash": pa_cash, "partner_b_cash": pb_cash, "transactions": all_transactions, "hurdle_allocations": hurdle_allocs } except Exception as e: log_exception(logger, f"Error processing regular allocation for loan_id {loan_id}: {str(e)}") return {"error": str(e)} # if __name__ == "__main__": # from dotenv import load_dotenv # load_dotenv(dotenv_path=r"D:\Python scripts\cf-python-scripts\.env") # test_payload = { # "loan_id": 60000, # "dry_run": False, # Change to False to test actual DB inserts # "payment_list": { # "payment_date": '30-08-2026', # "amount": 10000, # "payment_for": 'EMI', # "transaction_number": 'blacksoil123', # "transaction_mode": 'NEFT', # "utr": 'IDFC232' # } # } # try: # result = co_lending_regular_allocation(test_payload) # if "error" not in result: # print(f"Successfully processed loan_id: {test_payload['loan_id']}") # else: # print(f"Error processing loan: {result['error']}") # finally: # # Aggressively dispose of connection pools to prevent 'Too many connections' on subsequent runs # try: # from utils import engine, engine_read # engine.dispose() # engine_read.dispose() # except Exception: # pass