import json, math, os, requests, calendar, pytz import pandas as pd import numpy_financial as npf from datetime import datetime from dateutil.relativedelta import relativedelta from functools import lru_cache from utils import get_cloudwatch_logger, log_exception, execute_query, days360, apply_rounding, calculate_emi, calculate_bpi_raw, calculate_interest_raw from scripts.blacksoil_lms.co_lending_update_loans_data import process_loan_updates from utils import local_tz from datetime import datetime local_tz = pytz.timezone("Asia/Kolkata") now = datetime.now(local_tz).strftime('%Y-%m-%d %H:%M:%S') from scripts.blacksoil_lms.co_lending_utils import fetch_product_config logger = get_cloudwatch_logger(stream_name="co_lending_rps") def calculate_emi_and_bpi_dates(disbursement_date: str, emi_due_day: int, config_bpi_applicable: bool): disb_dt = datetime.strptime(disbursement_date, "%Y-%m-%d") d = disb_dt.day f_day = int(emi_due_day) if d < f_day: first_emi_dt = (disb_dt + relativedelta(months=1)).replace(day=f_day) bpi_applicable = config_bpi_applicable elif d > f_day: first_emi_dt = (disb_dt + relativedelta(months=2)).replace(day=f_day) bpi_applicable = config_bpi_applicable else: first_emi_dt = (disb_dt + relativedelta(months=1)).replace(day=f_day) bpi_applicable = False return first_emi_dt.strftime("%Y-%m-%d"), bpi_applicable def generate_standard_rps(loan_amount: float, annual_rate: float, tenure: int, disbursement_date: str, first_emi_date: str, calculation_basis: str = "ACTUAL_365", bpi_day_convention: str = "ACTUAL_365", bpi_applicable: bool = True, bpi_treatment: str = "SEPARATE_BEFORE_EMI", rounding_config: dict = None) -> list: """ Generates a single Repayment Schedule using actual days for interest calculation. """ if not rounding_config: rounding_config = {"emi_rounding": "ROUND", "emi_rounding_digits": 0, "interest_rounding": "ROUND", "interest_rounding_digits": 0, "bpi_rounding": "ROUND", "bpi_rounding_digits": 0} disb_dt = datetime.strptime(disbursement_date, "%Y-%m-%d") first_emi_dt = datetime.strptime(first_emi_date, "%Y-%m-%d") bpi_due_dt = first_emi_dt - relativedelta(months=1) schedule = [] # Month 0: BPI bpi_days = (bpi_due_dt - disb_dt).days bpi_interest_raw = 0.0 if bpi_applicable: bpi_interest_raw = calculate_bpi_raw(loan_amount, annual_rate, disb_dt, bpi_due_dt, bpi_day_convention, tenure) bpi_interest = apply_rounding(bpi_interest_raw, rounding_config["bpi_rounding"], rounding_config["bpi_rounding_digits"]) if bpi_applicable and bpi_interest < 0: raise ValueError(f"Calculated BPI is negative ({bpi_interest}). The first_emi_date is likely too close or prior to the disbursement_date.") if bpi_applicable and bpi_treatment == "ADD_TO_LOAN": loan_amount = loan_amount + bpi_interest pos = loan_amount emi = calculate_emi(loan_amount, annual_rate, tenure, rounding_config) elif bpi_applicable and bpi_treatment == "ADD_IN_EMI": pos = loan_amount emi = calculate_emi(loan_amount + bpi_interest, annual_rate, tenure, rounding_config) else: pos = loan_amount emi = calculate_emi(loan_amount, annual_rate, tenure, rounding_config) if bpi_applicable and bpi_treatment == "SEPARATE_BEFORE_EMI": schedule.append({ "installment_number": 0, "due_date": bpi_due_dt.strftime("%Y-%m-%d"), "days_diff": bpi_days, "bpi_due": bpi_interest, "emi_amount_due": 0.0, "interest_due": 0.0, "principal_due": 0.0, "closing_pos": pos }) prev_date = bpi_due_dt else: prev_date = disb_dt current_date = first_emi_dt for month in range(1, tenure + 1): days = (current_date - prev_date).days interest_raw = calculate_interest_raw( principal=pos, rate=annual_rate, start_date=prev_date, end_date=current_date, calculation_basis=calculation_basis, period=month, tenure=tenure, original_loan_amount=loan_amount ) interest = apply_rounding(interest_raw, rounding_config["interest_rounding"], rounding_config["interest_rounding_digits"]) # Final month adjustment if month == tenure: principal = pos payment_due = round(principal + interest, 2) pos = 0.0 else: if month == 1 and interest > emi: principal = 0.0 payment_due = interest else: payment_due = emi principal = payment_due - interest pos = pos - principal schedule.append({ "installment_number": month, "due_date": current_date.strftime("%Y-%m-%d"), "days_diff": days, "bpi_due": 0.0, "emi_amount_due": payment_due, "interest_due": interest, "principal_due": principal, "closing_pos": pos }) prev_date = current_date current_date = current_date + relativedelta(months=1) return schedule def generate_co_lending_rps(loan_amount: float, interest_rate: float, tenure: int, disbursement_date: str, first_emi_date: str, hurdle_rate: float = 0.18, partner_a_share: float = 0.8, calculation_basis: str = "ACTUAL_365", bpi_day_convention: str = "ACTUAL_365", bpi_applicable: bool = True, bpi_treatment: str = "SEPARATE_BEFORE_EMI", rounding_config: dict = None) -> dict: """ Generates the 4 required types of RPS and returns them as a Python dictionary. 1. RPS 1: 100% of loan amount at original interest rate 2. RPS 2: Partner A share (e.g. 80%) at original interest rate 3. RPS 3: Partner B share (e.g. 20%) at original interest rate (100% - 80%) 4. RPS 4: Partner A share (e.g. 80%) at hurdle rate """ if not rounding_config: rounding_config = {"emi_rounding": "ROUND", "emi_rounding_digits": 0, "interest_rounding": "ROUND", "interest_rounding_digits": 0, "bpi_rounding": "ROUND", "bpi_rounding_digits": 0} partner_b_share = round(1.0 - partner_a_share, 2) rps_100_orig = generate_standard_rps(loan_amount, interest_rate, tenure, disbursement_date, first_emi_date, calculation_basis, bpi_day_convention, bpi_applicable, bpi_treatment, rounding_config) a_loan_amount = round(loan_amount * partner_a_share, 2) rps_partner_a_orig = generate_standard_rps( loan_amount=a_loan_amount, annual_rate=interest_rate, tenure=tenure, disbursement_date=disbursement_date, first_emi_date=first_emi_date, calculation_basis=calculation_basis, bpi_day_convention=bpi_day_convention, bpi_applicable=bpi_applicable, bpi_treatment=bpi_treatment, rounding_config=rounding_config ) b_loan_amount = round(loan_amount - a_loan_amount, 2) rps_partner_b_orig = generate_standard_rps( loan_amount=b_loan_amount, annual_rate=interest_rate, tenure=tenure, disbursement_date=disbursement_date, first_emi_date=first_emi_date, calculation_basis=calculation_basis, bpi_day_convention=bpi_day_convention, bpi_applicable=bpi_applicable, bpi_treatment=bpi_treatment, rounding_config=rounding_config ) rps_partner_a_hurdle = [] pos_hurdle = a_loan_amount disb_dt = datetime.strptime(disbursement_date, "%Y-%m-%d") first_emi_dt = datetime.strptime(first_emi_date, "%Y-%m-%d") bpi_due_dt = first_emi_dt - relativedelta(months=1) if bpi_applicable and bpi_treatment == "ADD_TO_LOAN": hurdle_bpi_raw = calculate_bpi_raw(pos_hurdle, hurdle_rate, disb_dt, bpi_due_dt, bpi_day_convention, tenure) hurdle_bpi = apply_rounding(hurdle_bpi_raw, rounding_config["bpi_rounding"], rounding_config["bpi_rounding_digits"]) pos_hurdle += hurdle_bpi prev_date_hurdle = disb_dt for month_data in rps_100_orig: m = month_data["installment_number"] d = month_data["due_date"] days = month_data["days_diff"] current_date_hurdle = datetime.strptime(d, "%Y-%m-%d") if m == 0: bpi_due_dt = datetime.strptime(d, "%Y-%m-%d") hurdle_bpi_raw = calculate_bpi_raw(pos_hurdle, hurdle_rate, disb_dt, bpi_due_dt, bpi_day_convention, tenure) hurdle_bpi = apply_rounding(hurdle_bpi_raw, rounding_config["bpi_rounding"], rounding_config["bpi_rounding_digits"]) rps_partner_a_hurdle.append({ "installment_number": m, "due_date": d, "days_diff": days, "bpi_due": hurdle_bpi, "emi_amount_due": 0.0, "interest_due": 0.0, "principal_due": 0.0, "closing_pos": pos_hurdle }) prev_date_hurdle = current_date_hurdle continue hurdle_principal = month_data["principal_due"] * partner_a_share hurdle_interest_raw = calculate_interest_raw( principal=pos_hurdle, rate=hurdle_rate, start_date=prev_date_hurdle, end_date=current_date_hurdle, calculation_basis=calculation_basis, period=m, tenure=tenure, original_loan_amount=a_loan_amount ) hurdle_interest = apply_rounding(hurdle_interest_raw, rounding_config["interest_rounding"], rounding_config["interest_rounding_digits"]) hurdle_payment_due = round(hurdle_principal + hurdle_interest, 2) if m == tenure: pos_hurdle = 0.0 else: pos_hurdle = pos_hurdle - hurdle_principal rps_partner_a_hurdle.append({ "installment_number": m, "due_date": d, "days_diff": days, "bpi_due": 0.0, "emi_amount_due": hurdle_payment_due, "interest_due": hurdle_interest, "principal_due": hurdle_principal, "closing_pos": pos_hurdle }) prev_date_hurdle = current_date_hurdle p_a = int(partner_a_share * 100) p_b = int(partner_b_share * 100) result = { "orig": rps_100_orig, "partner_a": rps_partner_a_orig, "partner_b": rps_partner_b_orig, "hurdle": rps_partner_a_hurdle, "p_a": p_a, "p_b": p_b } return result def bulk_generate(data_source, user_id=107) -> list: """Reads bulk loans from Excel/CSV file path or a list of dicts and returns a list of dictionaries with their RPS.""" # Clear the LRU cache at the start of each bulk upload # This guarantees we get the freshest LMS configs per upload, while still # hitting the API only once per product_id for the duration of this loop. fetch_product_config.cache_clear() if isinstance(data_source, str): logger.send_log_event(f"INFO: Starting bulk RPS generation for file: {data_source}") if data_source.endswith(".csv"): df = pd.read_csv(data_source) else: df = pd.read_excel(data_source) elif isinstance(data_source, list): logger.send_log_event("INFO: Starting bulk RPS generation from JSON list.") df = pd.DataFrame(data_source) else: raise ValueError("Invalid data_source. Must be a file path string or a list of dictionaries.") df = df.dropna(how='all') output = [] db_insert_params = [] # Pre-check for existing loans in DB to avoid massive IN clauses issues loan_ids_to_check = df['loan_id'].dropna().astype(str).tolist() existing_loan_ids = set() if loan_ids_to_check: chunk_size = 1000 for i in range(0, len(loan_ids_to_check), chunk_size): chunk = loan_ids_to_check[i:i + chunk_size] format_strings = ','.join([':l_' + str(j) for j in range(len(chunk))]) sql_check = f"SELECT DISTINCT loan_id FROM co_lending_emi_payments WHERE loan_id IN ({format_strings}) AND deleted_at IS NULL" params_check = {f'l_{j}': lid for j, lid in enumerate(chunk)} try: existing_df = execute_query(sql_check, params=params_check) if existing_df is not None and not existing_df.empty: existing_loan_ids.update(existing_df['loan_id'].tolist()) except Exception as e: log_exception(logger, f"Error checking existing loan IDs: {str(e)}") required_cols = {'loan_id', 'loan_amount', 'interest_rate', 'tenure', 'disbursement_date', 'product_id', 'bounce_type'} if not required_cols.issubset(set(df.columns)): error_msg = f"Missing required columns. Expected: {required_cols}" logger.send_log_event(f"ERROR: {error_msg}") return [{"error": error_msg}] logger.send_log_event(f"INFO: Found {len(df)} rows to process.") try: created_by = int(user_id) if user_id is not None and user_id != "" else 107 except: created_by = 107 for row in df.itertuples(index=False): loan_id = getattr(row, 'loan_id', None) if pd.isna(loan_id): continue loan_id = str(loan_id) bounce_type = str(getattr(row, 'bounce_type', '')).lower() if bounce_type not in ['nach', 'cheque']: error_msg = f"Invalid bounce_type '{bounce_type}' for loan_id {loan_id}. Must be 'nach' or 'cheque'." logger.send_log_event(f"ERROR: {error_msg}") output.append({"loan_id": loan_id, "status": "skipped", "message": error_msg}) continue # Skip if already in database if loan_id in existing_loan_ids: logger.send_log_event(f"INFO: RPS for loan_id {loan_id} already exists in DB. Skipping.") output.append({ "loan_id": loan_id, "status": "skipped", "message": "Data already exists in database" }) continue try: loan_amount = float(row.loan_amount) interest_rate = float(row.interest_rate) tenure = int(row.tenure) product_id = int(row.product_id) if loan_amount <= 0 or interest_rate < 0 or tenure <= 0: raise ValueError("loan_amount, interest_rate, and tenure must be positive numbers.") d_date = pd.to_datetime(row.disbursement_date, errors='coerce') if pd.isna(d_date): raise ValueError("Incorrect date format for disbursement_date.") disbursement_date = d_date.strftime("%Y-%m-%d") partner_a_share, hurdle_rate, calculation_basis, bpi_day_convention, config_bpi_applicable, bpi_treatment, rounding_config, raw_config, disbursement_logic, fixed_day_of_month, emi_due_day = fetch_product_config(product_id) first_emi_date, bpi_applicable = calculate_emi_and_bpi_dates(disbursement_date, emi_due_day, config_bpi_applicable) # Log the extracted configuration components for debugging logger.send_log_event(f"INFO: Config for loan_id {loan_id} (product_id {product_id}): " f"rps={json.dumps(raw_config.get('rps'))}, interest={json.dumps(raw_config.get('interest'))}") rps_dict = generate_co_lending_rps( loan_amount=loan_amount, interest_rate=interest_rate, tenure=tenure, disbursement_date=disbursement_date, first_emi_date=first_emi_date, hurdle_rate=hurdle_rate, partner_a_share=partner_a_share, calculation_basis=calculation_basis, bpi_day_convention=bpi_day_convention, bpi_applicable=bpi_applicable, bpi_treatment=bpi_treatment, rounding_config=rounding_config ) output.append({ "loan_id": loan_id, "schedules": rps_dict }) # Prepare DB insert params for rps_type, schedule_list in rps_dict.items(): if rps_type in ["p_a", "p_b"]: continue # Skip the percent values mapping keys # Determine schedule percent value if rps_type == "orig": sch_pct = 100.0 elif rps_type in ["partner_a", "hurdle"]: sch_pct = float(rps_dict.get("p_a", 0.0)) elif rps_type == "partner_b": sch_pct = float(rps_dict.get("p_b", 0.0)) else: sch_pct = 0.0 for r in schedule_list: due_date_str = r["due_date"] db_insert_params.append({ "loan_id": loan_id, "schedule_percent": sch_pct, "rps_type": rps_type, "installment_number": r["installment_number"], "emi_date": due_date_str, "due_date": due_date_str, "days_diff": r["days_diff"], "bpi_due": float(r.get("bpi_due", 0.0)), "emi_amount_due": float(r["emi_amount_due"]), "interest_due": float(r["interest_due"]), "principal_due": float(r["principal_due"]), "closing_pos": float(r["closing_pos"]), "created_by": created_by, "created_at": now, "updated_at": now }) except Exception as e: log_exception(logger, f"Error processing loan_id {loan_id}: {str(e)}") output.append({ "loan_id": loan_id, "error": str(e) }) # Perform bulk insert of all accumulated rows if db_insert_params: logger.send_log_event(f"INFO: Inserting {len(db_insert_params)} RPS rows into database...") insert_sql = """ INSERT INTO co_lending_emi_payments ( loan_id, schedule_percent, rps_type, installment_number, emi_date, due_date, days_diff, bpi_due, emi_amount_due, interest_due, principal_due, closing_pos, created_by, created_at, updated_at ) VALUES ( :loan_id, :schedule_percent, :rps_type, :installment_number, :emi_date, :due_date, :days_diff, :bpi_due, :emi_amount_due, :interest_due, :principal_due, :closing_pos, :created_by, :created_at, :updated_at ) """ try: execute_query(insert_sql, params=db_insert_params) logger.send_log_event(f"INFO: Successfully inserted RPS rows into database.") successful_loan_ids = list(set([p['loan_id'] for p in db_insert_params])) if successful_loan_ids: logger.send_log_event(f"INFO: Triggering process_loan_updates for {len(successful_loan_ids)} loans.") process_loan_updates(loan_ids=successful_loan_ids) except Exception as e: log_exception(logger, f"Database insert failed: {str(e)}") logger.send_log_event(f"INFO: Completed processing. Total outputs: {len(output)}") return output # if __name__ == "__main__": # from dotenv import load_dotenv # load_dotenv(dotenv_path=r"D:\Python scripts\cf-python-scripts\.env") # test_payload = [{"loan_id": 800001000009, "loan_amount": 200000, "interest_rate": 0.299, "tenure": 36, "disbursement_date": "2026-04-28", "first_emi_date": "2026-06-05", "product_id": 1}] # print(f"Processing bulk RPS from local JSON payload...") # bulk_data = bulk_generate(test_payload) # output_file = r"D:\Python scripts\cf-python-scripts\testing_dir\bulk_rps_output.json" # with open(output_file, "w") as f: # json.dump(bulk_data, f, indent=4) # print(f"Bulk RPS JSON saved to {output_file}")