Risk Model Builder

def project_liquidity(cash_flows: str, off_balance_sheet: Optional[str] = None, stress_factor: float = 1.0) -> str:
    """
    Projects liquidity positions over time, including off-balance-sheet items and optional stress adjustments.

    Parameters:
    - cash_flows: JSON string — list of objects with keys Date(str), Inflows(float), Outflows(float).
                  Example: '[{"Date":"2024-01","Inflows":500000,"Outflows":300000}]'
    - off_balance_sheet: Optional JSON string — list of objects with keys Date(str), Commitments(float).
    - stress_factor: Multiplier to simulate stressed outflows(>1 means more outflows)

    Returns:
    - JSON string with projected liquidity and net positions
    """
    try:
        data = json.loads(cash_flows) if isinstance(cash_flows, str) else cash_flows
        df = pd.DataFrame(data)
        df['NetCashFlow'] = df['Inflows'] - df['Outflows'] * stress_factor

        if off_balance_sheet:
            obs_data = json.loads(off_balance_sheet) if isinstance(off_balance_sheet, str) else off_balance_sheet
            obs_df = pd.DataFrame(obs_data)
            df = df.merge(obs_df, on='Date', how='left')
            df['Commitments'] = df['Commitments'].fillna(0)
            df['NetCashFlow'] -= df['Commitments']

        df['CumulativeLiquidity'] = df['NetCashFlow'].cumsum()
        df['LiquidityGap'] = df['CumulativeLiquidity'].apply(lambda x: min(x, 0))

        result = df[['Date', 'Inflows', 'Outflows', 'NetCashFlow', 'CumulativeLiquidity', 'LiquidityGap']]
        return result.to_json(orient='records', indent=2)

    except Exception as e:
        return json.dumps({'error': str(e)})

def calculate_interest_rate_sensitivity(portfolio: str, rate_shifts: List[float]) -> str:
    """
    Calculates sensitivity of Net Interest Income(NII) and Economic Value of Equity(EVE) to interest rate changes.

    Parameters:
    - portfolio: JSON string — list of objects with keys Instrument(str), Type('Asset' or 'Liability'),
                 Amount(float), Rate(float), Duration(float).
                 Example: '[{"Instrument":"Bond","Type":"Asset","Amount":1000000,"Rate":0.05,"Duration":3}]'
    - rate_shifts: List of interest rate changes in decimals, e.g. [0.01, -0.01, 0.02] for +1%, -1%, +2%.

    Returns:
    - JSON string with NII and EVE sensitivity for each rate shift.
    """
    try:
        data = json.loads(portfolio) if isinstance(portfolio, str) else portfolio
        port_df = pd.DataFrame(data)

        results = []
        for shift in rate_shifts:
            df = port_df.copy()
            df['ShiftedRate'] = df['Rate'] + shift
            df['NII_Impact'] = df.apply(
                lambda x: x['Amount'] * (x['ShiftedRate'] - x['Rate']) * (1 if x['Type'] == 'Asset' else -1),
                axis=1
            )
            df['EVE_Impact'] = df.apply(
                lambda x: x['Amount'] * x['Duration'] * (x['ShiftedRate'] - x['Rate']) * (1 if x['Type'] == 'Asset' else -1),
                axis=1
            )
            results.append({
                'RateShift': shift,
                'TotalNIIImpact': round(float(df['NII_Impact'].sum()), 2),
                'TotalEVEImpact': round(float(df['EVE_Impact'].sum()), 2)
            })

        return json.dumps(results, indent=2)

    except Exception as e:
        return json.dumps({'error': str(e)})