AI Agent for Financial Analysis Development

Financial Agent Tools

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": "Query financial database (revenue, expenses, budget, actual)",
            "parameters": {
                "type": "object",
                "properties": {
                    "sql_query": {"type": "string"},
                    "description": {"type": "string"},
                },
                "required": ["sql_query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate_financial_metrics",
            "description": "Calculate financial indicators",
            "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": "Build financial model (DCF, budget, P&L forecast)",
            "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": "Generate financial report",
            "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:
    """Accurate calculation of financial metrics"""

    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"

Plan-Fact Analysis Agent

FINANCIAL_ANALYST_PROMPT = """You are a CFO-level financial analyst.

Your tasks:
1. Analyze financial data accurately and methodologically correctly
2. Use tools for calculations — never calculate in your head
3. For plan/fact variances — identify causes (price effect, volume effect, mix)
4. Give specific recommendations, not abstract observations
5. Point out anomalies and potential risks

Methodology:
- When analyzing P&L, break down variances into price and volume effects
- When assessing efficiency, compare with standard industry benchmarks
- For forecasts, indicate confidence interval and key assumptions"""

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"Data context:\n{json.dumps(context_data, ensure_ascii=False, indent=2)}"
        })

    # Agent loop
    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,
            })

Automatic Anomaly Detection

def detect_anomalies_in_data(financial_data: pd.DataFrame) -> list[dict]:
    """Statistical anomaly detection before passing to 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