Liquidity Interest Rate Risk Assessor

def calculate_nii_sensitivity(cashflow_data: str, shock_scenarios: str = "[-200, -100, +100, +200]") -> str:
    """
    Calculate Net Interest Income(NII) sensitivity to interest rate shocks.
    
    Args:
        cashflow_data: Asset and liability data with Amount and Rate
        shock_scenarios: List of interest rate shock scenarios in basis points
    
    Returns:
        NII sensitivity analysis report
    """
    try:
        import json
        
        "color: #6b7280;"># Parse inputs
        if isinstance(cashflow_data, str):
            try:
                data = json.loads(cashflow_data)
            except json.JSONDecodeError:
                return "Error: Invalid JSON format for cashflow data"
        else:
            data = cashflow_data
        
        if isinstance(shock_scenarios, str):
            try:
                shocks = json.loads(shock_scenarios)
            except json.JSONDecodeError:
                shocks = [-200, -100, 100, 200]
        else:
            shocks = shock_scenarios
        
        assets = data.get('assets', [])
        liabilities = data.get('liabilities', [])
        
        "color: #6b7280;"># Calculate base NII
        def calculate_nii(assets_data, liabilities_data, rate_adjustment=0):
            interest_income = sum(
                item.get('amount', 0) * (item.get('rate', 0.05) + rate_adjustment / 10000)
                for item in assets_data
            )
            interest_expense = sum(
                item.get('amount', 0) * (item.get('rate', 0.03) + rate_adjustment / 10000)
                for item in liabilities_data
            )
            return interest_income - interest_expense
        
        base_nii = calculate_nii(assets, liabilities)
        
        "color: #6b7280;"># Calculate NII under shock scenarios
        report = "=== NII SENSITIVITY ANALYSIS ===\n\n"
        report += f"Base NII: {base_nii:,.2f}\n\n"
        report += f"{'Shock(bps)':<15} {'Interest Income':>18} {'Interest Expense':>18} {'NII':>15} {'NII Change':>15} {'% Change':>12}\n"
        report += "-" * 93 + "\n"
        
        for shock in shocks:
            shocked_nii = calculate_nii(assets, liabilities, shock)
            nii_change = shocked_nii - base_nii
            pct_change = (nii_change / base_nii * 100) if base_nii != 0 else 0
            
            "color: #6b7280;"># Calculate components
            shocked_income = sum(
                item.get('amount', 0) * (item.get('rate', 0.05) + shock / 10000)
                for item in assets
            )
            shocked_expense = sum(
                item.get('amount', 0) * (item.get('rate', 0.03) + shock / 10000)
                for item in liabilities
            )
            
            shock_sign = "+" if shock >= 0 else ""
            report += f"{shock_sign}{shock:<14} {shocked_income:>18,.2f} {shocked_expense:>18,.2f} {shocked_nii:>15,.2f} {nii_change:>15,.2f} {pct_change:>11.2f}%\n"
        
        "color: #6b7280;"># Add recommendations
        report += "\n=== FINDINGS & RECOMMENDATIONS ===\n"
        
        "color: #6b7280;"># Check if rising rates hurt NII
        positive_shock_nii = calculate_nii(assets, liabilities, 200)
        positive_shock_change = positive_shock_nii - base_nii
        
        if positive_shock_change < 0:
            report += "\n⚠️ NEGATIVE NII SENSITIVITY: Rising rates reduce NII.\n"
            report += "Recommendation: Balance sheet is liability-sensitive. Consider increasing fixed-rate assets or reducing floating-rate liabilities.\n"
        elif positive_shock_change > 0:
            report += "\n✓ POSITIVE NII SENSITIVITY: Rising rates increase NII.\n"
            report += "Recommendation: Balance sheet is asset-sensitive. Current positioning benefits from rising rate environment.\n"
        else:
            report += "\n✓ NEUTRAL NII SENSITIVITY: Balanced interest rate exposure.\n"
        
        return report
        
    except Exception as e:
        return f"Error calculating NII sensitivity: {str(e)}"

def calculate_eve_sensitivity(cashflow_data: str, shock_scenarios: str = "[-200, -100, +100, +200]") -> str:
    """
    Calculate Economic Value of Equity(EVE) sensitivity to interest rate shocks.
    
    Args:
        cashflow_data: Asset and liability cash flows with Amount, Rate, and Maturity
        shock_scenarios: List of interest rate shock scenarios in basis points
    
    Returns:
        EVE sensitivity analysis report
    """
    try:
        import json
        
        "color: #6b7280;"># Parse inputs
        if isinstance(cashflow_data, str):
            try:
                data = json.loads(cashflow_data)
            except json.JSONDecodeError:
                return "Error: Invalid JSON format for cashflow data"
        else:
            data = cashflow_data
        
        if isinstance(shock_scenarios, str):
            try:
                shocks = json.loads(shock_scenarios)
            except json.JSONDecodeError:
                shocks = [-200, -100, 100, 200]
        else:
            shocks = shock_scenarios
        
        assets = data.get('assets', [])
        liabilities = data.get('liabilities', [])
        
        "color: #6b7280;"># Calculate base EVE
        def calculate_present_value(cashflows, rate_adjustment=0):
            pv = 0
            for cf in cashflows:
                amount = cf.get('amount', 0)
                rate = cf.get('rate', 0.05) + rate_adjustment / 10000  "color: #6b7280;"># Convert bps to decimal
                maturity = cf.get('maturity_years', 1)
                discount_factor = 1 / ((1 + rate) ** maturity)
                pv += amount * discount_factor
            return pv
        
        base_asset_pv = calculate_present_value(assets)
        base_liability_pv = calculate_present_value(liabilities)
        base_eve = base_asset_pv - base_liability_pv
        
        "color: #6b7280;"># Calculate EVE under shock scenarios
        report = "=== EVE SENSITIVITY ANALYSIS ===\n\n"
        report += f"Base EVE: {base_eve:,.2f}\n\n"
        report += f"{'Shock(bps)':<15} {'Asset PV':>15} {'Liability PV':>15} {'EVE':>15} {'EVE Change':>15} {'% Change':>12}\n"
        report += "-" * 87 + "\n"
        
        for shock in shocks:
            shocked_asset_pv = calculate_present_value(assets, shock)
            shocked_liability_pv = calculate_present_value(liabilities, shock)
            shocked_eve = shocked_asset_pv - shocked_liability_pv
            eve_change = shocked_eve - base_eve
            pct_change = (eve_change / base_eve * 100) if base_eve != 0 else 0
            
            shock_sign = "+" if shock >= 0 else ""
            report += f"{shock_sign}{shock:<14} {shocked_asset_pv:>15,.2f} {shocked_liability_pv:>15,.2f} {shocked_eve:>15,.2f} {eve_change:>15,.2f} {pct_change:>11.2f}%\n"
        
        "color: #6b7280;"># Add recommendations
        report += "\n=== FINDINGS & RECOMMENDATIONS ===\n"
        
        worst_shock = min(shocks, key=lambda s: calculate_present_value(assets, s) - calculate_present_value(liabilities, s))
        worst_eve = calculate_present_value(assets, worst_shock) - calculate_present_value(liabilities, worst_shock)
        worst_change = worst_eve - base_eve
        worst_pct = (worst_change / base_eve * 100) if base_eve != 0 else 0
        
        report += f"\nWorst case scenario: {worst_shock:+}bps shock results in {worst_pct:.2f}% EVE decline.\n"
        
        if abs(worst_pct) > 15:
            report += "\n⚠️ HIGH RISK: EVE sensitivity exceeds 15% threshold.\n"
            report += "Recommendation: Consider hedging interest rate risk through duration matching or derivatives.\n"
        elif abs(worst_pct) > 10:
            report += "\n⚠️ MODERATE RISK: EVE sensitivity between 10-15%.\n"
            report += "Recommendation: Monitor interest rate exposure closely and prepare hedging strategies.\n"
        else:
            report += "\n✓ LOW RISK: EVE sensitivity within acceptable range (<10%).\n"
        
        return report
        
    except Exception as e:
        return f"Error calculating EVE sensitivity: {str(e)}"

def calculate_liquidity_gaps(balance_sheet_data: str) -> str:
    """
    Calculate liquidity gaps across time buckets.
    
    Args:
        balance_sheet_data: Balance sheet data in JSON or CSV format with Assets and Liabilities,
                          including amount and maturity buckets
    
    Returns:
        Liquidity gap analysis report with gaps by time bucket
    """
    try:
        import json
        
        "color: #6b7280;"># Parse input data
        if isinstance(balance_sheet_data, str):
            try:
                data = json.loads(balance_sheet_data)
            except json.JSONDecodeError:
                return "Error: Invalid JSON format for balance sheet data"
        else:
            data = balance_sheet_data
        
        "color: #6b7280;"># Time buckets for gap analysis
        buckets = ['0-1M', '1-3M', '3-6M', '6-12M', '1-2Y', '2-5Y', '5Y+']
        
        "color: #6b7280;"># Initialize gaps
        gaps = {bucket: 0 for bucket in buckets}
        cumulative_gaps = {bucket: 0 for bucket in buckets}
        
        "color: #6b7280;"># Process assets and liabilities
        assets = data.get('assets', [])
        liabilities = data.get('liabilities', [])
        
        "color: #6b7280;"># Calculate net position per bucket
        for bucket in buckets:
            asset_total = sum(item['amount'] for item in assets if item.get('maturity') == bucket)
            liability_total = sum(item['amount'] for item in liabilities if item.get('maturity') == bucket)
            gaps[bucket] = asset_total - liability_total
        
        "color: #6b7280;"># Calculate cumulative gaps
        cumulative = 0
        for bucket in buckets:
            cumulative += gaps[bucket]
            cumulative_gaps[bucket] = cumulative
        
        "color: #6b7280;"># Generate report
        report = "=== LIQUIDITY GAP ANALYSIS ===\n\n"
        report += f"{'Time Bucket':<15} {'Assets':>15} {'Liabilities':>15} {'Gap':>15} {'Cumulative':>15}\n"
        report += "-" * 75 + "\n"
        
        for bucket in buckets:
            asset_total = sum(item['amount'] for item in assets if item.get('maturity') == bucket)
            liability_total = sum(item['amount'] for item in liabilities if item.get('maturity') == bucket)
            gap = gaps[bucket]
            cumulative = cumulative_gaps[bucket]
            
            report += f"{bucket:<15} {asset_total:>15,.2f} {liability_total:>15,.2f} {gap:>15,.2f} {cumulative:>15,.2f}\n"
        
        "color: #6b7280;"># Add recommendations
        report += "\n=== FINDINGS & RECOMMENDATIONS ===\n"
        
        negative_gaps = [bucket for bucket, gap in gaps.items() if gap < 0]
        if negative_gaps:
            report += f"\n⚠️ Negative gaps identified in: {', '.join(negative_gaps)}\n"
            report += "Recommendation: Consider increasing short-term liquid assets or extending liability maturities.\n"
        
        if cumulative_gaps['0-1M'] < 0:
            report += "\n⚠️ Immediate liquidity shortfall detected(0-1M bucket).\n"
            report += "Recommendation: Activate contingency funding plan immediately.\n"
        
        if all(gap > 0 for gap in gaps.values()):
            report += "\n✓ All time buckets show positive liquidity positions.\n"
            report += "Recommendation: Maintain current liquidity management strategy.\n"
        
        return report
        
    except Exception as e:
        return f"Error calculating liquidity gaps: {str(e)}"