AI Digital Risk Manager Development

Development of an AI-based digital risk manager

An AI Risk Manager is a digital worker that combines risk monitoring, quantitative assessment, and report generation into a single autonomous agent. In banks and insurance companies, this system processes thousands of risk signals in real time and generates recommendations for the risk committee without the need for manual analyst work.

Digital Risk Manager Architecture

Agent Architecture (LangGraph):

from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic

class RiskManagerAgent:
    def __init__(self):
        self.llm = ChatAnthropic(model='claude-opus-4')
        self.tools = [
            market_risk_calculator,
            credit_risk_scorer,
            liquidity_risk_monitor,
            compliance_checker,
            report_generator
        ]
        self.graph = self.build_graph()

    def build_graph(self):
        workflow = StateGraph(AgentState)
        workflow.add_node('assess_risks', self.assess_all_risks)
        workflow.add_node('identify_breaches', self.check_limit_breaches)
        workflow.add_node('generate_actions', self.recommend_actions)
        workflow.add_node('escalate', self.escalate_to_human)
        workflow.add_node('generate_report', self.create_risk_report)

        workflow.add_conditional_edges(
            'identify_breaches',
            lambda state: 'escalate' if state['critical_breaches'] else 'generate_actions'
        )
        return workflow.compile()

Risk categories monitored:

• Market risk: VaR, CVaR, Delta, option Greeks, beta, DV01 for bonds
• Credit risk: PD (probability of default), LGD, EAD, EL, RAROC
• Liquidity risk: LCR (Liquidity Coverage Ratio), NSFR, GAP analysis
• Operational risk: incidents, KRI (Key Risk Indicators), Basel AMA
• Concentration risk: exposure by counterparty, sector, geography

Real-time market risk monitoring

VaR and CVaR calculation:

import numpy as np
from scipy.stats import norm

def calculate_portfolio_var(returns, weights, confidence=0.99, horizon=1):
    """
    Historical Simulation VaR (предпочтительнее параметрического)
    """
    portfolio_returns = returns @ weights
    var = np.percentile(portfolio_returns, (1 - confidence) * 100)
    cvar = portfolio_returns[portfolio_returns <= var].mean()
    return var * np.sqrt(horizon), cvar * np.sqrt(horizon)

def parametric_var(portfolio_return, portfolio_vol, confidence=0.99, horizon=1):
    """
    Параметрический VaR (предполагает нормальность)
    """
    z_score = norm.ppf(1 - confidence)
    return (portfolio_return * horizon + z_score * portfolio_vol * np.sqrt(horizon))

Risk Limit Monitoring:

risk_limits = {
    'var_99_1d': 5_000_000,       # VaR 99% за 1 день
    'cvar_95_10d': 15_000_000,    # CVaR 95% за 10 дней
    'single_name_concentration': 0.05,  # max 5% на один эмитент
    'sector_concentration': 0.20,
    'duration_limit': 7.5,         # лет для облигационного портфеля
    'delta_equity': 100_000_000    # нетто-дельта акций
}

def check_limit_breaches(current_metrics, limits):
    breaches = {}
    for metric, limit in limits.items():
        if current_metrics[metric] > limit:
            breaches[metric] = {
                'current': current_metrics[metric],
                'limit': limit,
                'breach_pct': (current_metrics[metric] / limit - 1) * 100
            }
    return breaches

Credit scoring and monitoring

Automated Credit Monitoring:

def credit_portfolio_monitor(credit_portfolio, market_data):
    """
    Ежедневный пересмотр кредитного портфеля
    Триггеры: изменение рейтинга, CDS spread spike, новости
    """
    alerts = []

    for exposure in credit_portfolio:
        # Рыночный сигнал: скачок CDS
        cds_change = market_data['cds_spread'][exposure.counterparty]
        if cds_change > 50:  # > 50 bp за день
            alerts.append({
                'counterparty': exposure.counterparty,
                'type': 'cds_spike',
                'severity': 'high' if cds_change > 100 else 'medium',
                'cds_change': cds_change,
                'exposure': exposure.notional
            })

        # Рейтинговое событие
        current_rating = get_latest_rating(exposure.counterparty)
        if rating_downgrade(exposure.last_known_rating, current_rating) >= 2:
            alerts.append({
                'counterparty': exposure.counterparty,
                'type': 'rating_downgrade',
                'old_rating': exposure.last_known_rating,
                'new_rating': current_rating,
                'exposure': exposure.notional
            })

    return sorted(alerts, key=lambda x: x['exposure'], reverse=True)

Expected Loss Calculation:

def calculate_el(pd, lgd, ead):
    """EL = PD × LGD × EAD"""
    return pd * lgd * ead

def raroc(expected_return, el, economic_capital, cost_of_capital=0.12):
    """RAROC = (Expected Return - EL) / Economic Capital"""
    return (expected_return - el) / economic_capital

NLP news monitoring

Automatic news monitoring:

from transformers import pipeline
import feedparser

def monitor_risk_news(watchlist_entities):
    """
    Мониторинг новостей по списку контрагентов и отраслей
    Классификация: financial_stress, legal_regulatory, geopolitical
    """
    sentiment_analyzer = pipeline(
        'text-classification',
        model='ProsusAI/finbert'
    )
    risk_classifier = pipeline(
        'text-classification',
        model='custom_risk_event_model'
    )

    for entity in watchlist_entities:
        articles = fetch_news(entity, days=1)
        for article in articles:
            sentiment = sentiment_analyzer(article.text[:512])[0]
            risk_events = risk_classifier(article.text[:512])[0]

            if sentiment['label'] == 'negative' and sentiment['score'] > 0.8:
                yield RiskAlert(
                    entity=entity,
                    source=article.url,
                    headline=article.title,
                    risk_type=risk_events['label'],
                    severity=classify_severity(risk_events)
                )

Automatic reporting

Daily Risk Report Generation:

def generate_daily_risk_report(portfolio_data, limit_breaches, alerts, llm):
    """
    Структурированный дневной риск-отчёт для риск-комитета
    """
    report_data = {
        'executive_summary': generate_executive_summary(portfolio_data, limit_breaches),
        'market_risk': format_market_risk_section(portfolio_data),
        'credit_risk': format_credit_section(portfolio_data, alerts),
        'limit_breaches': format_breaches(limit_breaches),
        'recommended_actions': generate_recommendations(limit_breaches, portfolio_data)
    }

    # LLM генерирует нарратив по структурированным данным
    narrative = llm.invoke(
        f"Generate a concise risk report summary based on these metrics: {report_data}"
    )

    return Report(data=report_data, narrative=narrative, timestamp=datetime.now())

Regulatory reporting:

• FRTB (Fundamental Review of Trading Book): SA and IMA approaches
• Basel III/IV: NSFR, LCR calculations
• Bank of Russia: Form 634-P, PCR for systemically important banks

Integration

Data Sources:

• Bloomberg, Refinitiv Eikon: market data, CDS spreads, ratings
• Murex, Calypso, Openlink: trading systems (positions, P&L)
• Core Banking: loan portfolio, limits
• NewsAPI, Factiva, Interfax: news feed

Workflow: All agent recommendations → through the approval workflow to the risk manager. Critical limit violations → automatic notification to the CRO + emergency meeting.

Deadlines: VaR/CVaR engine + limit monitoring + basic alerts — 6-8 weeks. LangGraph agent, NLP news monitoring, automated reporting, regulatory forms — 5-6 months.