Разработка AI-агента для финансового анализа

Инструменты финансового агента

from openai import OpenAI from pydantic import BaseModel from typing import Literal, Optional import pandas as pd import json client = OpenAI() financial_tools = [ { "type": "function", "function": { "name": "query_financial_database", "description": "Запрос к финансовой базе данных (выручка, расходы, бюджет, факт)", "parameters": { "type": "object", "properties": { "sql_query": {"type": "string"}, "description": {"type": "string"}, }, "required": ["sql_query"] } } }, { "type": "function", "function": { "name": "calculate_financial_metrics", "description": "Рассчитать финансовые показатели", "parameters": { "type": "object", "properties": { "metric": { "type": "string", "enum": ["EBITDA", "ROE", "ROA", "ROIC", "NPV", "IRR", "payback_period", "gross_margin", "operating_margin", "net_margin", "current_ratio", "debt_to_equity", "working_capital"] }, "input_data": {"type": "object"}, }, "required": ["metric", "input_data"] } } }, { "type": "function", "function": { "name": "build_financial_model", "description": "Построить финансовую модель (DCF, бюджет, P&L прогноз)", "parameters": { "type": "object", "properties": { "model_type": {"type": "string", "enum": ["dcf", "budget_variance", "pnl_forecast"]}, "parameters": {"type": "object"}, }, "required": ["model_type", "parameters"] } } }, { "type": "function", "function": { "name": "generate_financial_report", "description": "Сформировать финансовый отчёт", "parameters": { "type": "object", "properties": { "report_type": {"type": "string"}, "period": {"type": "string"}, "data": {"type": "object"}, }, "required": ["report_type", "period"] } } }, ] def calculate_financial_metrics(metric: str, input_data: dict) -> str: """Точный расчёт финансовых метрик""" calculators = { "EBITDA": lambda d: d["revenue"] - d["cogs"] - d["opex"] + d.get("da", 0), "gross_margin": lambda d: (d["revenue"] - d["cogs"]) / d["revenue"] * 100, "operating_margin": lambda d: d["ebit"] / d["revenue"] * 100, "ROE": lambda d: d["net_income"] / d["equity"] * 100, "ROA": lambda d: d["net_income"] / d["total_assets"] * 100, "current_ratio": lambda d: d["current_assets"] / d["current_liabilities"], "debt_to_equity": lambda d: d["total_debt"] / d["equity"], } calculator = calculators.get(metric) if not calculator: return f"Metric {metric} not implemented" try: result = calculator(input_data) return json.dumps({ "metric": metric, "result": round(result, 4), "unit": "%" if metric in ["gross_margin", "operating_margin", "ROE", "ROA"] else "x", }) except KeyError as e: return f"Missing required field: {e}" except ZeroDivisionError: return "Division by zero: check denominator values" 

Агент план-факт анализа

FINANCIAL_ANALYST_PROMPT = """Ты — финансовый аналитик CFO-уровня. Твои задачи: 1. Анализировать финансовые данные точно и методологически корректно 2. Использовать инструменты для расчётов — никогда не считай в уме 3. При отклонениях план/факт — выявлять причины (price effect, volume effect, mix) 4. Давать конкретные рекомендации, а не абстрактные наблюдения 5. Указывать на аномалии и потенциальные риски Методология: - При анализе P&L разбивай отклонения на ценовые и объёмные - При оценке эффективности используй сравнение со стандартными отраслевыми бенчмарками - При прогнозах — указывай confidence interval и ключевые допущения""" def financial_analysis_agent(question: str, context_data: dict = None) -> str: messages = [ {"role": "system", "content": FINANCIAL_ANALYST_PROMPT}, {"role": "user", "content": question}, ] if context_data: messages.insert(1, { "role": "system", "content": f"Контекст данных:\n{json.dumps(context_data, ensure_ascii=False, indent=2)}" }) # Агентный цикл for _ in range(8): response = client.chat.completions.create( model="gpt-4o", messages=messages, tools=financial_tools, ) msg = response.choices[0].message messages.append(msg) if not msg.tool_calls: return msg.content for tool_call in msg.tool_calls: tool_name = tool_call.function.name tool_args = json.loads(tool_call.function.arguments) result = execute_financial_tool(tool_name, tool_args) messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": result, }) 

Автоматическая аномалия-детекция

def detect_anomalies_in_data(financial_data: pd.DataFrame) -> list[dict]: """Статистическое выявление аномалий перед передачей в LLM""" anomalies = [] for column in financial_data.select_dtypes(include="number").columns: mean = financial_data[column].mean() std = financial_data[column].std() z_scores = (financial_data[column] - mean) / std outliers = financial_data[abs(z_scores) > 2.5] if not outliers.empty: for idx, row in outliers.iterrows(): anomalies.append({ "column": column, "value": row[column], "z_score": round(z_scores[idx], 2), "period": str(idx), }) return anomalies