import os, io, time, warnings, argparse, boto3 import numpy as np import numpy_financial as npf import pandas as pd from joblib import Parallel, delayed from datetime import datetime from dateutil.relativedelta import relativedelta from dotenv import load_dotenv load_dotenv() warnings.filterwarnings('ignore') pd.set_option('future.no_silent_downcasting', True) from utils import execute_query, reconciliation, apply_waiver_allocation from scripts.pincode_geocoder import sync_pincode_master from scripts.mysql_queries import ( active_applications_for_metrics_query, historical_payments_query, net_loan_irr_query, historical_emi_payments_query ) # ── Configuration ────────────────────────────────────────────────────────────── CHUNK_SIZE = 5000 N_JOBS = -1 # Use all available CPU cores # ── Date Utilities ───────────────────────────────────────────────────────────── def get_run_dates(is_historical): """ Determine START_DATE and END_DATE based on run mode. END_DATE is always the last day of the previous calendar month. START_DATE is '2026-06-30' for historical backfill, or equal to END_DATE for incremental. """ end_date_str = (datetime.today().replace(day=1) - relativedelta(days=1)).strftime('%Y-%m-%d') start_date_str = '2026-06-30' if is_historical else end_date_str return start_date_str, end_date_str def get_target_dates(start, end): """Return a list of month-end dates between start and end, inclusive.""" return [d.date() for d in pd.date_range(start=start, end=end, freq='M')] # ── Data Fetching ────────────────────────────────────────────────────────────── def build_target_dataset(start_date, end_date): """ Fetch active applications for each month-end target date and combine into one DataFrame. Each row represents one application active as of a given target date. """ target_dates = get_target_dates(start_date, end_date) dfs = [] for td in target_dates: print(f"Fetching active applications as of {td}...") df = execute_query(active_applications_for_metrics_query, params={'as_on_date': td}) if not df.empty: df['target_date'] = td dfs.append(df) return pd.concat(dfs, ignore_index=True) def fetch_raw_timelines(app_id_list): """ Fetch loan info, EMI schedule, and full payment history for a list of application IDs. Returns: (loan_info_df, emi_payments_df, payments_df) """ loan_info_df = execute_query(net_loan_irr_query, params={'application_ids': tuple(app_id_list)}) payments_df = execute_query(historical_payments_query, params={'app_id_list': tuple(app_id_list)}) if not payments_df.empty: payments_df['payment_date'] = pd.to_datetime(payments_df['payment_date'], errors='coerce').dt.date payments_df = payments_df.dropna(subset=['payment_date']) emi_payments_df = execute_query(historical_emi_payments_query, params={'application_ids': tuple(app_id_list)}) emi_payments_df = apply_waiver_allocation(emi_payments_df) return loan_info_df, emi_payments_df, payments_df # ── AWS S3 ───────────────────────────────────────────────────────────────────── def get_s3_client(): """ Initialize a boto3 S3 client from environment variables. Returns (s3_client, bucket_name), or (None, None) if credentials are missing. """ aws_access_key = os.getenv('AWS_ACCESS_KEY_ID') aws_secret_key = os.getenv('AWS_SECRET_ACCESS_KEY') aws_region = os.getenv('AWS_REGION_NAME', 'ap-south-1') bucket_name = os.getenv('AWS_BUCKET') if aws_access_key and aws_secret_key and bucket_name: s3_client = boto3.client('s3', aws_access_key_id=aws_access_key, aws_secret_access_key=aws_secret_key, region_name=aws_region) return s3_client, bucket_name return None, None def export_snapshots(final_df, s3_client, bucket_name): """ Partition snapshots by target_month (YYYY-MM) and upload each partition to S3. Falls back to saving locally if S3 is not configured. """ final_df['target_month'] = pd.to_datetime(final_df['target_date']).dt.strftime('%Y-%m') total_uploaded = 0 for month, group in final_df.groupby('target_month'): group_to_save = group.drop(columns=['target_month']) s3_key = f"loan_pos_dpd/target_month={month}/snapshot.parquet" if s3_client: buf = io.BytesIO() group_to_save.to_parquet(buf, index=False, engine='pyarrow') buf.seek(0) print(f"Uploading target_month={month} → s3://{bucket_name}/{s3_key}") s3_client.upload_fileobj(buf, bucket_name, s3_key) total_uploaded += len(group_to_save) else: output_dir = os.path.join(os.path.dirname(__file__), '..', 'output', 'partitions', f"target_month={month}") os.makedirs(output_dir, exist_ok=True) output_file = os.path.join(output_dir, 'snapshot.parquet') group_to_save.to_parquet(output_file, index=False, engine='pyarrow') print(f"Saved target_month={month} locally → {output_file}") if s3_client: print(f"Uploaded {total_uploaded} snapshots across {final_df['target_month'].nunique()} partitions.") # ── Snapshot Calculation Helpers ─────────────────────────────────────────────── def get_dpd_bucket(dpd): """Map a DPD (days past due) integer to its reporting bucket label.""" if dpd == 0: return 'Current' if dpd <= 30: return '1-30' if dpd <= 60: return '31-60' if dpd <= 90: return '61-90' if dpd <= 120: return '91-120' if dpd <= 150: return '121-150' if dpd <= 180: return '151-180' if dpd <= 364: return '181-364' return '365+' def build_amort_table(app_emi_payments, pt_filtered, app_id): """ Build a chronological amortization table for a single application by merging the EMI schedule with filtered payment transactions. Pre-part payments (emi_payment_id=0) are injected as separate date-stamped rows, mirroring the SQL UNION logic in foreclosure_breakup. """ # Group regular payments (linked to an EMI) by emi_payment_id if pt_filtered.empty: pt_grouped = pd.DataFrame(columns=['emi_payment_id', 'total_paid', 'total_paid_for_bpi', 'latest_payment_date']) else: regular_pt = pt_filtered[pt_filtered['emi_payment_id'] != 0] pt_grouped = regular_pt.groupby('emi_payment_id').agg( total_paid=('amount_regular', 'sum'), total_paid_for_bpi=('amount_pre_emi', 'sum'), latest_payment_date=('payment_date_dt', 'max') ).reset_index() amort_df = pd.merge(app_emi_payments, pt_grouped, on='emi_payment_id', how='left') amort_df['total_paid'] = amort_df['total_paid'].fillna(0) amort_df['total_paid_for_bpi'] = amort_df['total_paid_for_bpi'].fillna(0) amort_df['payment_details'] = '' amort_df['application_id'] = app_id if 'emi_date' not in amort_df.columns: amort_df['emi_date'] = pd.NaT amort_df['emi_date'] = pd.to_datetime(amort_df['emi_date'], dayfirst=True) amort_df['emi_date'] = amort_df['emi_date'].fillna(pd.to_datetime(amort_df['latest_payment_date'])) # Inject pre-part payments (emi_payment_id == 0) as individual rows by payment_date pre_parts = pt_filtered[pt_filtered['emi_payment_id'] == 0] if not pre_parts.empty: pre_parts_grouped = pre_parts.groupby('payment_date_dt').agg( total_paid=('amount_regular', 'sum') ).reset_index() new_rows = [ { 'emi_payment_id': 0, 'application_id': app_id, 'emi_date': pd.to_datetime(r.payment_date_dt), 'emi_amount': r.total_paid, 'total_paid': r.total_paid, 'total_paid_for_bpi': 0, 'pre_emi_amount': 0, 'moratorium_amount': 0, 'bounce_charges': 0, 'delay_penalty': 0, 'latest_payment_date': pd.to_datetime(r.payment_date_dt), 'payment_details': '' } for r in pre_parts_grouped.itertuples() ] amort_df = pd.concat([amort_df, pd.DataFrame(new_rows)], ignore_index=True) # Sort chronologically and format date columns amort_df['emi_date_obj'] = pd.to_datetime(amort_df['emi_date'], errors='coerce') amort_df = amort_df.sort_values(['emi_date_obj', 'emi_payment_id']) amort_df['emi_date'] = amort_df['emi_date_obj'].dt.strftime('%d-%m-%Y') if 'due_date' in amort_df.columns: amort_df['due_date'] = pd.to_datetime(amort_df['due_date'], errors='coerce').dt.strftime('%d-%m-%Y') for col in ['bounce_charges', 'delay_penalty', 'pre_emi_amount', 'moratorium_amount', 'emi_amount']: if col in amort_df.columns: amort_df[col] = pd.to_numeric(amort_df[col], errors='coerce').fillna(0).infer_objects(copy=False) # For pre-part rows, ensure emi_amount reflects the actual payment amount mask = amort_df['emi_payment_id'] == 0 if mask.any(): amort_df.loc[mask, 'emi_amount'] = amort_df.loc[mask, 'total_paid'] return amort_df # ── Core Processing ──────────────────────────────────────────────────────────── def process_chunk(app_ids, all_target_dates_df): """ Calculate POS and DPD for each application in app_ids across all its target dates. Designed to run in parallel via joblib. Returns a list of snapshot dicts. """ pd.set_option('future.no_silent_downcasting', True) loan_info, emi_payments_df, payment_data = fetch_raw_timelines(app_ids) chunk_snapshots = [] reserved_cols = {'application_id', 'target_date', 'application_status', 'status', 'pos', 'dpd', 'dpd_bucket'} static_cols = [c for c in all_target_dates_df.columns if c not in reserved_cols] payment_data['amount_pre_emi'] = np.where(payment_data['payment_for'] == 'Pre-EMI', payment_data['amount'], 0) payment_data['amount_regular'] = np.where(payment_data['payment_for'] != 'Pre-EMI', payment_data['amount'], 0) for app_id in app_ids: app_target_dates = all_target_dates_df[all_target_dates_df['application_id'] == app_id] if app_target_dates.empty: continue app_loan_info = loan_info[loan_info['application_id'] == app_id] if app_loan_info.empty: continue net_loan_amount = float(app_loan_info['net_loan_amount'].iloc[0]) pre_part_flag = app_loan_info['pre_part_flag'].iloc[0] # Compute the periodic interest rate for the reconciliation engine if pre_part_flag == 1: if pd.isna(app_loan_info['tenure'].iloc[0]) or pd.isna(app_loan_info['emi_value'].iloc[0]): irr_rate = 0.05 else: tenure = int(app_loan_info['tenure'].iloc[0]) emi_value = float(app_loan_info['emi_value'].iloc[0]) irr_rate = npf.rate(nper=tenure, pmt=-emi_value, pv=net_loan_amount, fv=0, when='end') else: emis_list_str = str(app_loan_info['emis_list'].iloc[0]) if emis_list_str.strip() and emis_list_str.lower() != 'nan': emi_values = [int(v) for v in emis_list_str.split(',') if v.strip().isdigit()] irr_rate = npf.irr([-int(net_loan_amount)] + emi_values) else: irr_rate = 0.05 app_payments = payment_data[payment_data['application_id'] == app_id].copy() app_emi_payments = emi_payments_df[emi_payments_df['application_id'] == app_id] for row in app_target_dates.itertuples(): target_date = row.target_date status = str(row.application_status).lower() meta_dict = {col: getattr(row, col, None) for col in static_cols} # Closed/settled loans: POS and DPD are zero if status in ['closed', 'cancelled', 'foreclosed', 'settled']: chunk_snapshots.append({ 'application_id': app_id, **meta_dict, 'target_date': target_date, 'status': status, 'pos': 0, 'dpd': 0, 'dpd_bucket': 'Current' }) continue # Filter payments strictly up to the target date target_date_dt = pd.to_datetime(target_date) app_payments['payment_date_dt'] = pd.to_datetime(app_payments['payment_date']) pt_filtered = app_payments[app_payments['payment_date_dt'] <= target_date_dt] amort_df = build_amort_table(app_emi_payments, pt_filtered, app_id) try: df_reconciled = reconciliation(amort_df, net_loan_amount, irr_rate) except Exception as e: print(f"Failed reconciliation for {app_id} on {target_date}: {e}") continue # POS: net loan amount minus all principal repaid up to target date principal_paid = df_reconciled['principal_paid'].sum() if 'principal_paid' in df_reconciled.columns else 0 pos = round(max(net_loan_amount - principal_paid, 0), 2) # DPD: days elapsed since the earliest EMI not yet fully paid dpd = 0 if 'principal_due' in df_reconciled.columns and 'interest_due' in df_reconciled.columns: for emi in df_reconciled.itertuples(): if getattr(emi, 'emi_payment_id', 0) == 0: continue p_due, p_paid = getattr(emi, 'principal_due', 0), getattr(emi, 'principal_paid', 0) i_due, i_paid = getattr(emi, 'interest_due', 0), getattr(emi, 'interest_paid', 0) if round(p_paid, 2) < round(p_due, 2) or round(i_paid, 2) < round(i_due, 2): first_unpaid = getattr(emi, 'emi_date') if isinstance(first_unpaid, str): first_unpaid = pd.to_datetime(first_unpaid, format='%d-%m-%Y', errors='coerce').date() if pd.notna(first_unpaid): dpd = max((target_date_dt.date() - first_unpaid).days, 0) break chunk_snapshots.append({ 'application_id': app_id, **meta_dict, 'target_date': target_date, 'status': status, 'pos': pos, 'dpd': dpd, 'dpd_bucket': get_dpd_bucket(dpd) }) return chunk_snapshots # ── Entry Point ──────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser(description="Generate monthly Loan POS and DPD snapshots.") parser.add_argument('--historical', action='store_true', help="Backfill from 2026-06-30 to previous month-end") parser.add_argument('--incremental', action='store_true', help="Process previous month only (default)") args = parser.parse_args() is_historical = args.historical start_date, end_date = get_run_dates(is_historical) start_time = time.time() print(f"\n{'─' * 45}") print(f" Mode : {'Historical' if is_historical else 'Incremental'}") print(f" Start Date: {start_date}") print(f" End Date : {end_date}") print(f"{'─' * 45}\n") # 1. Build target dataset (active applications per month-end) all_target_dates_df = build_target_dataset(start_date, end_date) # 2. Initialize S3 s3_client, bucket_name = get_s3_client() # 3. Sync pincode master with any new pincodes discovered in this run if 'customer_address_pincode' in all_target_dates_df.columns: print("\n--- Syncing Pincode Master ---") try: sync_pincode_master(all_target_dates_df['customer_address_pincode'], s3_client=s3_client, bucket_name=bucket_name) except Exception as e: print(f"Warning: Pincode sync encountered an issue: {e}") print(f"{'─' * 45}\n") # 4. Run parallel POS & DPD calculations across all application chunks unique_app_ids = all_target_dates_df['application_id'].unique().tolist() # unique_app_ids = [255265] print(f"Found {len(unique_app_ids)} unique applications. Processing in chunks of {CHUNK_SIZE}...") chunks = [unique_app_ids[i:i + CHUNK_SIZE] for i in range(0, len(unique_app_ids), CHUNK_SIZE)] results = Parallel(n_jobs=N_JOBS)( delayed(process_chunk)(chunk, all_target_dates_df[all_target_dates_df['application_id'].isin(chunk)]) for chunk in chunks ) all_snapshots = [snap for chunk_result in results for snap in chunk_result] # 5. Export partitioned snapshots to S3 (or local fallback) final_df = pd.DataFrame(all_snapshots) # final_df.to_csv('target.csv',index=False) export_snapshots(final_df, s3_client, bucket_name) print(f"\nTotal Execution Time: {round(time.time() - start_time, 2)} seconds") if __name__ == "__main__": main()