AI в финансах: fraud detection, robo-advisors, compliance

from dataclasses import dataclass
from datetime import datetime
from typing import Optional
import numpy as np


@dataclass
class Transaction:
    """Сырая транзакция из процессинга."""
    transaction_id: str
    user_id: str
    amount: float
    currency: str
    merchant_id: str
    merchant_category_code: str  # MCC
    timestamp: datetime
    device_id: str
    ip_address: str
    card_present: bool  # физическая или онлайн-транзакция
    country: str
    city: str


def extract_features(
    tx: Transaction,
    user_history: list[Transaction],
    window_hours: int = 24,
) -> dict[str, float]:
    """
    Извлечь признаки для fraud scoring.
    Каждый признак должен быть обоснован бизнес-логикой.
    """
    now = tx.timestamp
    recent = [t for t in user_history if (now - t.timestamp).total_seconds() < window_hours * 3600]

    # Признаки транзакции
    features = {
        "amount": tx.amount,
        "amount_log": np.log1p(tx.amount),  # лог-трансформация для скошенных распределений

        # Отклонение от среднего пользователя
        "amount_vs_user_mean": tx.amount / (np.mean([t.amount for t in user_history]) + 1e-9),
        "amount_vs_user_std_z": (
            (tx.amount - np.mean([t.amount for t in user_history]))
            / (np.std([t.amount for t in user_history]) + 1e-9)
        ),

        # Временные признаки
        "hour_of_day": now.hour,
        "day_of_week": now.weekday(),
        "is_weekend": int(now.weekday() >= 5),
        "is_night": int(now.hour < 6 or now.hour > 23),

        # Скорость транзакций (velocity)
        "tx_count_24h": len(recent),
        "tx_sum_24h": sum(t.amount for t in recent),
        "unique_merchants_24h": len({t.merchant_id for t in recent}),
        "unique_countries_24h": len({t.country for t in recent}),

        # Географические признаки
        "new_country": int(tx.country not in {t.country for t in user_history}),
        "new_merchant": int(tx.merchant_id not in {t.merchant_id for t in user_history}),

        # Канальные признаки
        "is_card_not_present": int(not tx.card_present),
        "new_device": int(tx.device_id not in {t.device_id for t in user_history}),
    }

    return features

import joblib
import lightgbm as lgb
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler


def build_fraud_model(X_train, y_train, X_val, y_val) -> lgb.LGBMClassifier:
    """
    LightGBM — стандарт для транзакционного фрода:
    - быстрый inference (< 1 мс)
    - интерпретируемость через SHAP
    - хорошо работает на несбалансированных данных

    Типичный дисбаланс: 0.1-0.5% фрода.
    """
    model = lgb.LGBMClassifier(
        n_estimators=500,
        learning_rate=0.05,
        max_depth=6,
        num_leaves=31,
        # Критично для несбалансированных данных:
        class_weight="balanced",
        # или: scale_pos_weight=neg_count/pos_count
        min_child_samples=50,
        subsample=0.8,
        colsample_bytree=0.8,
        random_state=42,
        n_jobs=-1,
    )

    model.fit(
        X_train, y_train,
        eval_set=[(X_val, y_val)],
        eval_metric="auc",
        callbacks=[lgb.early_stopping(50, verbose=False)],
    )

    return model


def get_fraud_score_with_explanation(
    model: lgb.LGBMClassifier,
    features: dict[str, float],
    top_n_reasons: int = 3,
) -> dict:
    """
    Вернуть score + объяснение через SHAP.
    Требование регулятора: каждое решение должно быть объяснимо.
    """
    import shap

    X = np.array([list(features.values())])
    feature_names = list(features.keys())

    score = model.predict_proba(X)[0][1]  # вероятность фрода

    # SHAP для объяснимости
    explainer = shap.TreeExplainer(model)
    shap_values = explainer.shap_values(X)[1]  # для класса 1 (фрод)

    # Топ-N признаков, которые повлияли на решение
    shap_pairs = sorted(
        zip(feature_names, shap_values[0]),
        key=lambda x: abs(x[1]),
        reverse=True,
    )
    top_reasons = [
        {"feature": name, "shap_value": float(val), "direction": "increase" if val > 0 else "decrease"}
        for name, val in shap_pairs[:top_n_reasons]
    ]

    return {
        "fraud_score": float(score),
        "decision": "block" if score > 0.8 else ("review" if score > 0.4 else "approve"),
        "top_reasons": top_reasons,
    }

import anthropic
from pathlib import Path


FINANCIAL_ANALYST_PROMPT = """Ты опытный финансовый аналитик (buy-side, equity research).

Твои задачи:
- Анализировать финансовую отчётность объективно и критически
- Выявлять расхождения между словами менеджмента и цифрами
- Всегда указывать на ключевые риски и неопределённости
- Давать структурированный анализ, удобный для принятия инвестиционных решений

ВАЖНО: Ты не даёшь инвестиционных рекомендаций (buy/sell/hold). Только анализ.
Все выводы — на основе предоставленных данных, без домыслов.
"""


def analyze_earnings_report(
    report_text: str,
    company: str,
    quarter: str,
    prior_guidance: Optional[str] = None,
) -> dict:
    """
    Анализ квартального отчёта через LLM.

    Args:
        report_text: текст отчёта или транскрипт earnings call
        company: название компании
        quarter: квартал (например, "Q3 2025")
        prior_guidance: предыдущий прогноз менеджмента (если есть)
    """
    client = anthropic.Anthropic()

    context = f"Компания: {company}\nОтчётный период: {quarter}\n"
    if prior_guidance:
        context += f"Предыдущий guidance менеджмента:\n{prior_guidance}\n\n"
    context += f"Текст отчёта/транскрипта:\n{report_text}"

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2000,
        system=FINANCIAL_ANALYST_PROMPT,
        messages=[
            {
                "role": "user",
                "content": f"""{context}

Проведи анализ по следующей структуре:

1. КЛЮЧЕВЫЕ РЕЗУЛЬТАТЫ
   - Выручка: факт vs guidance/consensus
   - Margins: динамика и объяснение
   - EPS: факт vs ожидания

2. КАЧЕСТВО ПРИБЫЛИ
   - Органический vs неорганический рост
   - Разовые статьи, влияющие на результат
   - FCF conversion

3. FORWARD-LOOKING
   - Новый guidance vs предыдущий
   - Качество комментариев менеджмента (конкретные цифры или размытые формулировки?)

4. КРАСНЫЕ ФЛАГИ
   - Необъяснённые отклонения
   - Изменения в учётной политике
   - Нетипичный язык или уклончивые ответы на конкретные вопросы

5. КЛЮЧЕВЫЕ ВОПРОСЫ
   - 3 вопроса, которые необходимо прояснить""",
            }
        ],
    )

    return {
        "company": company,
        "quarter": quarter,
        "analysis": response.content[0].text,
        "tokens_used": response.usage.input_tokens + response.usage.output_tokens,
    }


def compliance_check_document(
    document_text: str,
    regulation_type: str = "AML",
    jurisdiction: str = "RU",
) -> dict:
    """
    Проверка документа на соответствие регуляторным требованиям через LLM.
    Не заменяет юридическую экспертизу — флагует потенциальные проблемы.
    """
    client = anthropic.Anthropic()

    COMPLIANCE_PROMPTS = {
        "AML_RU": """Ты комплаенс-специалист по 115-ФЗ (Противодействие легализации доходов).
        Анализируй документ на наличие признаков, требующих обязательного контроля или сообщения в Росфинмониторинг.
        Ссылайся на конкретные статьи 115-ФЗ.
        Указывай уровень уверенности: высокий/средний/низкий риск.""",

        "GDPR_FINANCE": """Ты DPO (Data Protection Officer) в финансовой компании.
        Проверь документ на соответствие GDPR применительно к финансовым данным.
        Основные проверки: правовое основание обработки, retention periods, трансграничная передача.""",
    }

    system = COMPLIANCE_PROMPTS.get(f"{regulation_type}_{jurisdiction}", COMPLIANCE_PROMPTS["AML_RU"])

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1500,
        system=system,
        messages=[
            {
                "role": "user",
                "content": f"Проверь следующий документ:\n\n{document_text}\n\nВыдай: 1) уровень риска (LOW/MEDIUM/HIGH), 2) конкретные проблемные места с цитатами, 3) рекомендуемые действия.",
            }
        ],
    )

    return {
        "regulation": f"{regulation_type}/{jurisdiction}",
        "analysis": response.content[0].text,
        "disclaimer": "AI-анализ носит информационный характер и не является юридическим заключением. Требуется верификация квалифицированным специалистом.",
    }

from scipy.optimize import minimize
import numpy as np


def modern_portfolio_optimization(
    expected_returns: np.ndarray,  # годовая доходность по каждому активу
    cov_matrix: np.ndarray,        # ковариационная матрица доходностей
    risk_tolerance: float,          # 0.0 (консервативный) до 1.0 (агрессивный)
    constraints: Optional[dict] = None,
) -> dict:
    """
    Оптимизация портфеля по Марковицу (Modern Portfolio Theory).
    Максимизируем risk-adjusted return для заданной терпимости к риску.
    """
    n_assets = len(expected_returns)

    def portfolio_volatility(weights):
        return np.sqrt(weights @ cov_matrix @ weights)

    def portfolio_return(weights):
        return weights @ expected_returns

    def negative_sharpe(weights, risk_free_rate=0.05):
        ret = portfolio_return(weights)
        vol = portfolio_volatility(weights)
        return -(ret - risk_free_rate) / (vol + 1e-9)

    # risk_tolerance влияет на целевую функцию
    def objective(weights):
        ret = portfolio_return(weights)
        vol = portfolio_volatility(weights)
        # Комбинация: (1-tol)*risk_minimization + tol*return_maximization
        return (1 - risk_tolerance) * vol - risk_tolerance * ret

    # Ограничения: сумма весов = 1, все веса >= 0 (нет коротких позиций)
    cons = [{"type": "eq", "fun": lambda w: np.sum(w) - 1}]
    bounds = [(0, 1)] * n_assets

    if constraints:
        # Дополнительные ограничения (например, max 30% в один актив)
        if "max_weight" in constraints:
            bounds = [(0, constraints["max_weight"])] * n_assets

    result = minimize(
        objective,
        x0=np.ones(n_assets) / n_assets,  # равновесный старт
        method="SLSQP",
        bounds=bounds,
        constraints=cons,
        options={"ftol": 1e-9, "maxiter": 1000},
    )

    optimal_weights = result.x
    return {
        "weights": optimal_weights.tolist(),
        "expected_return": float(portfolio_return(optimal_weights)),
        "volatility": float(portfolio_volatility(optimal_weights)),
        "sharpe_ratio": float(
            (portfolio_return(optimal_weights) - 0.05)
            / (portfolio_volatility(optimal_weights) + 1e-9)
        ),
        "optimization_success": result.success,
    }


def rebalance_portfolio(
    current_weights: dict[str, float],
    target_weights: dict[str, float],
    portfolio_value: float,
    threshold: float = 0.05,  # ребалансировать если отклонение > 5%
    transaction_cost: float = 0.001,  # 0.1% комиссия
) -> list[dict]:
    """
    Вычислить необходимые сделки для ребалансировки.
    Минимизируем транзакционные издержки: торгуем только при значимых отклонениях.
    """
    trades = []
    for asset, target_w in target_weights.items():
        current_w = current_weights.get(asset, 0.0)
        delta_w = target_w - current_w

        if abs(delta_w) > threshold:
            amount = delta_w * portfolio_value
            cost = abs(amount) * transaction_cost
            trades.append({
                "asset": asset,
                "action": "buy" if delta_w > 0 else "sell",
                "amount": abs(amount),
                "transaction_cost": cost,
                "current_weight": current_w,
                "target_weight": target_w,
            })

    return sorted(trades, key=lambda x: abs(x["amount"]), reverse=True)