import pandas as pd from datetime import datetime, timedelta import os, sys from utils import execute_query, get_cloudwatch_logger, log_exception, engine, local_tz from scripts.mysql_queries import get_loan_info_query, get_emi_schedules_query, get_payment_transactions_query logger = get_cloudwatch_logger(stream_name="da-daily-metrics") # Ensure we can import utils from the parent directory script_dir = os.path.dirname(os.path.abspath(__file__)) parent_dir = os.path.dirname(os.path.dirname(script_dir)) if parent_dir not in sys.path: sys.path.append(parent_dir) def log_and_print(msg): print(msg) logger.send_log_event(msg) def calculate_accrued_revenue(loan_ids: list = None, as_on_date_str: str = None, start_date_str: str = None, full_history: bool = False): """ Calculates daily accrued revenue and summaries for specific Blacksoil loans. """ try: if not as_on_date_str: as_on_date_str = datetime.now().strftime('%Y-%m-%d') as_on_date = datetime.strptime(as_on_date_str, '%Y-%m-%d').date() if start_date_str: start_date = datetime.strptime(start_date_str, '%Y-%m-%d').date() else: start_date = as_on_date if loan_ids: loans_str = ",".join([f"'{str(lid)}'" for lid in loan_ids]) log_and_print(f"Running metrics calculation for {len(loan_ids)} provided loan_ids: {loan_ids}") else: loans_str = None log_and_print("No loan_ids provided. Fetching all active loans with POS >= 1.") # 1. Fetch Basic Info from da_system_data and blacksoil_attributes, filtering by POS > 0 log_and_print(f"Fetching active loan info...") q_info = get_loan_info_query(loans_str) df_info = execute_query(q_info) if df_info.empty: log_and_print("No active loans found.") return active_loans = df_info['loan_no'].tolist() active_loans_str = ",".join([f"'{lid}'" for lid in active_loans]) # 2. Fetch EMIs from blacksoil_emi_payments log_and_print(f"Fetching EMI schedules...") q_emi = get_emi_schedules_query(active_loans_str) df_emi = execute_query(q_emi) if not df_emi.empty: df_emi['emi_date'] = pd.to_datetime(df_emi['emi_date']).dt.date # 3. Fetch Payments from blacksoil_payment_transactions log_and_print(f"Fetching payment transactions...") q_pay = get_payment_transactions_query(active_loans_str) df_pay = execute_query(q_pay) if not df_pay.empty: df_pay['payment_date'] = pd.to_datetime(df_pay['payment_date']).dt.date else: df_pay = pd.DataFrame(columns=['loan_id', 'principal_received', 'payment_date', 'payment_for']) daily_metrics = [] summary = [] for row in df_info.itertuples(index=False): loan_id = str(row.loan_no) roi_pct = float(row.int_rate) if pd.notnull(row.int_rate) else 0.0 roi = roi_pct / 100.0 dpd = int(row.dpd) l_emi = df_emi[df_emi['loan_id'].astype(str) == loan_id].copy() if l_emi.empty: log_and_print(f"No EMI schedule found for loan {loan_id}. Skipping.") continue total_loan_amount = float(l_emi['principal_due'].sum()) first_emi_date = l_emi['emi_date'].min() first_emi_interest = float(l_emi[l_emi['emi_date'] == first_emi_date]['interest_due'].sum()) l_pay = df_pay[df_pay['loan_id'].astype(str) == loan_id].copy() current_pos = total_loan_amount current_date = first_emi_date # We need a function to get total payments in the month corresponding to the EMI date. def get_payments_for_month(dt): m_start = dt.replace(day=1) # Find next month start if m_start.month == 12: m_end = m_start.replace(year=m_start.year+1, month=1) else: m_end = m_start.replace(month=m_start.month+1) p = l_pay[(l_pay['payment_date'] >= m_start) & (l_pay['payment_date'] < m_end)] has_foreclosure = any(p['payment_for'].str.lower().isin(['foreclosure', 'foreclosure_principal'])) return float(p['principal_received'].sum()), has_foreclosure # Get all EMI dates as a set for faster lookup emi_dates = set(l_emi['emi_date'].unique()) pos_tracker = current_pos accrued_not_due_sum = 0 recent_emi_due_date = None log_and_print(f"Processing loan {loan_id} starting from {first_emi_date} to {as_on_date}...") while current_date <= as_on_date: if pos_tracker <= 0: break # Add first EMI date interest specifically if current_date == first_emi_date: daily_interest = first_emi_interest # Also drop POS if there are payments in the first month and it is an EMI date monthly_pay, has_foreclosure = get_payments_for_month(current_date) if has_foreclosure: pos_tracker = 0 else: pos_tracker = max(0, pos_tracker - monthly_pay) recent_emi_due_date = current_date accrued_not_due_sum = 0 else: # Check if today is an EMI date, if so, drop the POS based on payments in this month if current_date in emi_dates: monthly_pay, has_foreclosure = get_payments_for_month(current_date) if has_foreclosure: pos_tracker = 0 else: pos_tracker = max(0, pos_tracker - monthly_pay) recent_emi_due_date = current_date accrued_not_due_sum = 0 daily_interest = (pos_tracker * roi * 1) / 365 if full_history or current_date >= start_date: daily_metrics.append({ 'loan_id': loan_id, 'as_on_date': current_date.strftime('%Y-%m-%d'), 'interest_due': round(daily_interest, 6), 'principal_outstanding': round(pos_tracker, 2) }) if recent_emi_due_date and current_date > recent_emi_due_date: accrued_not_due_sum += daily_interest current_date += timedelta(days=1) # Summary calculations interest_due_not_paid = (roi * pos_tracker * dpd) / 365 no_of_days = (as_on_date - recent_emi_due_date).days if recent_emi_due_date else 0 if pos_tracker > 0: summary.append({ 'Loan ID': loan_id, 'Customer Name': row.member_name, 'Partner Name': row.partner_id, 'UCIC': '', 'Principal Outstanding': round(pos_tracker, 2), 'No of days': no_of_days, 'ROI%': roi_pct, 'Interest Accured but not due': round(accrued_not_due_sum, 6), 'Interest due which is not paid': round(interest_due_not_paid, 6), 'DPD Days': dpd, 'Recent EMI due date': recent_emi_due_date.strftime('%Y-%m-%d') if recent_emi_due_date else '' }) if daily_metrics: df_daily = pd.DataFrame(daily_metrics) # Add local time for created_at and updated_at now_str = datetime.now(local_tz).strftime('%Y-%m-%d %H:%M:%S') df_daily['created_at'] = now_str df_daily['updated_at'] = now_str records = df_daily.to_dict('records') batch_size = 5000 log_and_print(f"Upserting {len(records)} rows into da_daily_metrics in batches of {batch_size}...") from sqlalchemy import text insert_stmt = text(""" INSERT INTO da_daily_metrics (loan_id, as_on_date, principal_outstanding, interest_due, created_at, updated_at) VALUES (:loan_id, :as_on_date, :principal_outstanding, :interest_due, :created_at, :updated_at) ON DUPLICATE KEY UPDATE principal_outstanding = VALUES(principal_outstanding), interest_due = VALUES(interest_due), updated_at = VALUES(updated_at) """) with engine.begin() as conn: for i in range(0, len(records), batch_size): batch = records[i:i+batch_size] conn.execute(insert_stmt, batch) log_and_print("Upsert complete.") log_and_print(f"Successfully upserted {len(records)} metric rows for {len(active_loans)} loans in batches.") except Exception as e: log_exception(logger, f"Error in calculate_accrued_revenue: {str(e)}") # if __name__ == "__main__": # d = sys.argv[1] if len(sys.argv) > 1 else None # # For testing, provide loan_ids or let the endpoint handle it # calculate_accrued_revenue(['1003102000213', '1003101000582'], d)