Liquidity Gap Calculation:

Liquidity_Gap(t) = Assets_Maturing(t) - Liabilities_Maturing(t)

Cumulative_Liquidity_Gap(t) = Σ_{i=1}^{t} Liquidity_Gap(i)

Positive Gap: More assets maturing than liabilities (liquidity surplus)
Negative Gap: More liabilities maturing than assets (liquidity deficit)

Time Bands:
- Overnight
- 2-7 days
- 8-30 days
- 31-90 days
- 91-365 days
- >365 days

def calculate_lcr(bank_data):
    """
    Calculate Liquidity Coverage Ratio
    
    Parameters:
    - bank_data: Dictionary with bank balance sheet data
    
    Returns:
    - LCR ratio and pass/fail status
    """
    # HQLA Calculation
    hqla = 0
    hqla += bank_data['cash'] * 1.00
    hqla += bank_data['central_bank_reserves'] * 1.00
    hqla += bank_data['sovereign_debt'] * 1.00
    hqla += bank_data['corporate_bonds_aa'] * 0.85
    hqla += bank_data['corporate_bonds_bbb'] * 0.50
    
    # Cash Outflows
    outflows = 0
    outflows += bank_data['retail_deposits_stable'] * 0.05
    outflows += bank_data['retail_deposits_unstable'] * 0.10
    outflows += bank_data['wholesale_deposits_operational'] * 0.25
    outflows += bank_data['wholesale_deposits_non_operational'] * 0.40
    outflows += bank_data['commercial_paper'] * 1.00
    outflows += bank_data['undrawn_credit_lines'] * 0.50
    outflows += bank_data['undrawn_liquidity_lines'] * 1.00
    
    # Cash Inflows
    inflows = 0
    inflows += bank_data['maturing_retail_loans'] * 0.50
    inflows += bank_data['maturing_wholesale_loans'] * 0.50
    inflows += bank_data['contractual_inflows'] * 1.00
    
    # Apply inflow cap (75% of outflows)
    max_inflows = outflows * 0.75
    adjusted_inflows = min(inflows, max_inflows)
    
    # Net outflows
    net_outflows = outflows - adjusted_inflows
    
    # LCR
    lcr = hqla / net_outflows if net_outflows > 0 else float('inf')
    
    return {
        'hqla': hqla,
        'outflows': outflows,
        'inflows': inflows,
        'net_outflows': net_outflows,
        'lcr': lcr,
        'pass': lcr >= 1.0
    }

def calculate_nsfr(bank_data):
    """
    Calculate Net Stable Funding Ratio
    
    Parameters:
    - bank_data: Dictionary with bank balance sheet data
    
    Returns:
    - NSFR ratio and pass/fail status
    """
    # Available Stable Funding (ASF)
    asf = 0
    asf += bank_data['cet1_capital'] * 1.00
    asf += bank_data['additional_tier1'] * 1.00
    asf += bank_data['tier2_capital'] * 1.00
    asf += bank_data['retail_deposits_stable'] * 0.95
    asf += bank_data['retail_deposits_unstable'] * 0.90
    asf += bank_data['wholesale_deposits_1yr'] * 1.00
    asf += bank_data['wholesale_deposits_6mo'] * 0.50
    
    # Required Stable Funding (RSF)
    rsf = 0
    rsf += bank_data['cash'] * 0.00
    rsf += bank_data['sovereign_debt_0rw'] * 0.05
    rsf += bank_data['sovereign_debt_20rw'] * 0.20
    rsf += bank_data['corporate_debt_aa'] * 0.50
    rsf += bank_data['corporate_debt_bbb'] * 0.50
    rsf += bank_data['mortgages'] * 0.85
    rsf += bank_data['commercial_loans'] * 0.85
    rsf += bank_data['equities'] * 1.00
    rsf += bank_data['other_assets'] * 1.00
    
    # NSFR
    nsfr = asf / rsf if rsf > 0 else float('inf')
    
    return {
        'asf': asf,
        'rsf': rsf,
        'nsfr': nsfr,
        'pass': nsfr >= 1.0
    }

import numpy as np

def calculate_duration(cash_flows, discount_rate):
    """
    Calculate Macaulay Duration
    
    Parameters:
    - cash_flows: Array of cash flows
    - discount_rate: Yield to maturity
    
    Returns:
    - Duration in years
    """
    times = np.arange(1, len(cash_flows) + 1)
    
    # Present value of each cash flow
    pv_cash_flows = cash_flows / (1 + discount_rate) ** times
    
    # Weighted present value
    weighted_pv = pv_cash_flows * times
    
    # Duration
    duration = np.sum(weighted_pv) / np.sum(pv_cash_flows)
    
    return duration

def calculate_duration_gap(assets, liabilities):
    """
    Calculate duration gap for ALM
    
    Parameters:
    - assets: List of asset dictionaries with 'value' and 'duration'
    - liabilities: List of liability dictionaries with 'value' and 'duration'
    
    Returns:
    - Duration gap
    """
    # Weighted average duration of assets
    asset_value = sum(a['value'] for a in assets)
    asset_duration = sum(a['value'] * a['duration'] for a in assets) / asset_value
    
    # Weighted average duration of liabilities
    liability_value = sum(l['value'] for l in liabilities)
    liability_duration = sum(l['value'] * l['duration'] for l in liabilities) / liability_value
    
    # Duration gap
    duration_gap = asset_duration - liability_duration * (liability_value / asset_value)
    
    return {
        'asset_duration': asset_duration,
        'liability_duration': liability_duration,
        'duration_gap': duration_gap,
        'asset_value': asset_value,
        'liability_value': liability_value
    }

def calculate_convexity(cash_flows, discount_rate):
    """
    Calculate Convexity for more accurate interest rate risk measurement
    
    Parameters:
    - cash_flows: Array of cash flows
    - discount_rate: Yield to maturity
    
    Returns:
    - Convexity
    """
    times = np.arange(1, len(cash_flows) + 1)
    
    # Present value of each cash flow
    pv_cash_flows = cash_flows / (1 + discount_rate) ** times
    
    # Weighted by time*(time+1)
    weighted_pv = pv_cash_flows * times * (times + 1)
    
    # Convexity
    convexity = np.sum(weighted_pv) / ((1 + discount_rate) ** 2 * np.sum(pv_cash_flows))
    
    return convexity

def calculate_irrbb(assets, liabilities, shock_rates):
    """
    Calculate Interest Rate Risk in Banking Book
    
    Parameters:
    - assets: List of asset dictionaries
    - liabilities: List of liability dictionaries
    - shock_rates: Array of rate shocks to test
    
    Returns:
    - Impact on economic value and net interest income
    """
    results = []
    
    for shock in shock_rates:
        # Revalue assets
        new_asset_value = 0
        new_asset_income = 0
        
        for asset in assets:
            # Price change due to rate shock
            price_change = -asset['duration'] * shock * asset['value']
            
            # Convexity adjustment
            if 'convexity' in asset:
                convexity_adjustment = 0.5 * asset['convexity'] * (shock ** 2) * asset['value']
                price_change += convexity_adjustment
            
            new_asset_value += asset['value'] + price_change
            
            # Impact on income (assuming variable rates)
            new_asset_income += asset['yield'] * (1 + shock) * asset['value']
        
        # Revalue liabilities
        new_liability_value = 0
        new_liability_cost = 0
        
        for liability in liabilities:
            # Liability value change
            price_change = -liability['duration'] * shock * liability['value']
            new_liability_value += liability['value'] + price_change
            
            # Impact on funding cost
            new_liability_cost += liability['cost'] * (1 + shock) * liability['value']
        
        # Calculate impact
        economic_value_change = new_asset_value - new_liability_value - (sum(a['value'] for a in assets) - sum(l['value'] for l in liabilities))
        
        net_interest_income = new_asset_income - new_liability_cost
        nim_change = net_interest_income / new_asset_value
        
        results.append({
            'shock': shock,
            'economic_value_change': economic_value_change,
            'net_interest_income': net_interest_income,
            'nim': nim_change
        })
    
    return results

import pandas as pd
from datetime import datetime, timedelta

class IntradayLiquidityManager:
    """
    Intraday liquidity management system
    """
    def __init__(self, starting_balance, payment_schedule):
        self.balance = starting_balance
        self.payment_schedule = payment_schedule
        self.transaction_log = []
    
    def process_payments(self):
        """
        Process payments throughout the day
        """
        for payment in self.payment_schedule:
            # Check if we have sufficient balance
            if self.balance < payment['amount']:
                # Need to source liquidity
                self.sources_liquidity(payment['amount'] - self.balance)
            
            # Process payment
            self.balance -= payment['amount']
            self.transaction_log.append({
                'time': payment['time'],
                'type': 'outgoing',
                'amount': payment['amount'],
                'balance': self.balance
            })
        
        return self.transaction_log
    
    def sources_liquidity(self, needed_amount):
        """
        Source intraday liquidity
        """
        # Options:
        # 1. Use central bank facilities
        # 2. Repo with counterparties
        # 3. Draw down credit lines
        # 4. Sell assets
        
        # For simulation, assume central bank facility
        self.balance += needed_amount
        self.transaction_log.append({
            'time': datetime.now(),
            'type': 'liquidity_injection',
            'amount': needed_amount,
            'balance': self.balance
        })
    
    def calculate_intraday_liquidity_metrics(self):
        """
        Calculate key intraday liquidity metrics
        """
        df = pd.DataFrame(self.transaction_log)
        
        metrics = {
            'max_intraday_deficit': abs(min(df['balance'])),
            'min_intraday_balance': min(df['balance']),
            'peak_intraday_balance': max(df['balance']),
            'total_payments': df[df['type'] == 'outgoing']['amount'].sum(),
            'total_liquidity_injections': df[df['type'] == 'liquidity_injection']['amount'].sum(),
            'number_of_liquidity_events': len(df[df['type'] == 'liquidity_injection']),
            'coverage_hours': self.calculate_coverage_hours()
        }
        
        return metrics
    
    def calculate_coverage_hours(self):
        """
        Calculate hours of coverage for remaining balance
        """
        # Project future payments
        remaining_payments = [p for p in self.payment_schedule if p['time'] > datetime.now()]
        
        if not remaining_payments:
            return float('inf')
        
        # Calculate survival time
        hourly_outflow = sum(p['amount'] for p in remaining_payments) / 24
        coverage_hours = self.balance / hourly_outflow if hourly_outflow > 0 else float('inf')
        
        return coverage_hours

import numpy as np
import pandas as pd
from scipy.stats import poisson, norm

class DigitalBankRunSimulator:
    """
    Simulate digital bank runs with stochastic models
    """
    def __init__(self, initial_deposits, hqla, run_intensity=0.10, acceleration=0.15):
        self.initial_deposits = initial_deposits
        self.hqla = hqla
        self.run_intensity = run_intensity
        self.acceleration = acceleration
    
    def simulate_run_path(self, days=30, n_simulations=1000):
        """
        Simulate multiple bank run paths
        """
        results = []
        
        for _ in range(n_simulations):
            path = self.simulate_single_path(days)
            results.append(path)
        
        return results
    
    def simulate_single_path(self, days):
        """
        Simulate a single bank run path
        """
        deposits = self.initial_deposits
        hqla = self.hqla
        path = []
        
        for day in range(days):
            # Run intensity increases over time (social media effect)
            daily_intensity = self.run_intensity * (1 + self.acceleration * day / days)
            
            # Stochastic run amount (jump process)
            run_amount = deposits * daily_intensity + norm.rvs(0, 0.01 * deposits)
            
            # Deposits decrease
            deposits -= run_amount
            
            # HQLA decreases as depositors withdraw
            hqla -= run_amount
            
            # Record path
            path.append({
                'day': day,
                'deposits': max(0, deposits),
                'hqla': max(0, hqla),
                'run_amount': run_amount
            })
            
            # Check if bank has failed
            if hqla <= 0 or deposits <= 0:
                break
        
        return path
    
    def analyze_results(self, results):
        """
        Analyze simulation results
        """
        # Extract survival times
        survival_times = []
        final_deposits = []
        final_hqla = []
        
        for path in results:
            survival_times.append(len(path))
            final_deposits.append(path[-1]['deposits'])
            final_hqla.append(path[-1]['hqla'])
        
        # Calculate statistics
        survival_times = np.array(survival_times)
        final_deposits = np.array(final_deposits)
        final_hqla = np.array(final_hqla)
        
        return {
            'mean_survival_time': np.mean(survival_times),
            'median_survival_time': np.median(survival_times),
            'fail_probability': np.mean(survival_times < 30),
            'mean_final_deposits': np.mean(final_deposits),
            'mean_final_hqla': np.mean(final_hqla),
            'survival_distribution': survival_times,
            'fail_paths': [r for r in results if len(r) < 30]
        }