Разработка AI-цифрового риск-менеджера (AI Risk Manager)

Разработка AI-цифрового риск-менеджера

AI Risk Manager — цифровой работник, объединяющий мониторинг рисков, количественную оценку и генерацию отчётов в единый автономный агент. В банках и страховых компаниях такая система обрабатывает тысячи сигналов риска в режиме реального времени и формирует рекомендации для риск-комитета без ручной работы аналитиков.

Архитектура цифрового риск-менеджера

Агентная архитектура (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()

Категории отслеживаемых рисков:

• Рыночный риск: VaR, CVaR, Delta, греки опционов, beta, DV01 для облигаций
• Кредитный риск: PD (probability of default), LGD, EAD, EL, RAROC
• Ликвидностный риск: LCR (Liquidity Coverage Ratio), NSFR, GAP-анализ
• Операционный риск: инциденты, KRI (Key Risk Indicators), Basel AMA
• Концентрационный риск: exposure by counterparty, sector, geography

Мониторинг рыночного риска в реальном времени

VaR и CVaR вычисление:

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

Кредитный скоринг и мониторинг

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-мониторинг новостей

Автоматический новостной мониторинг:

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)
                )

Автоматическая отчётность

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())

Регуляторная отчётность:

• FRTB (Fundamental Review of Trading Book): SA и IMA подходы
• Basel III / IV: NSFR, LCR расчёты
• Банк России: форма 634-П, ПКР для системно значимых банков

Интеграция

Data Sources:

• Bloomberg, Refinitiv Eikon: рыночные данные, CDS spreads, рейтинги
• Murex, Calypso, Openlink: торговые системы (позиции, P&L)
• Core Banking: кредитный портфель, лимиты
• NewsAPI, Factiva, Интерфакс: новостной поток

Workflow: Все рекомендации агента → через approval workflow к риск-менеджеру. Критические лимитные нарушения → автоматическое уведомление CRO + экстренное совещание.

Сроки: VaR/CVaR движок + limit monitoring + базовые алерты — 6-8 недель. LangGraph агент, NLP новостной мониторинг, automated reporting, регуляторные формы — 5-6 месяцев.