Financial Data Analyst — Phân tích dữ liệu và biểu đồ tự động

Phân tích tài chính truyền thống đòi hỏi analyst phải viết code pandas, tạo biểu đồ matplotlib, và soạn báo cáo — mỗi bước tốn nhiều thời gian. Financial Data Analyst Agent tự động hóa toàn bộ pipeline này: nhận câu hỏi bằng ngôn ngữ tự nhiên, tự viết và thực thi code phân tích, sinh biểu đồ, và tổng hợp báo cáo.

Bài viết này xây dựng agent phân tích cổ phiếu và portfolio với Claude sonnet làm coding engine.

Kiến trúc: Code Execution Agent

User: "Phan tich co phieu VNM trong 6 thang qua"
      |
      v
[Claude: Viet code pandas de phan tich]
      |
      v
[Execute code trong sandbox]
      |
      v
[Thu thap ket qua + output charts]
      |
      v
[Claude: Interpret ket qua, viet bao cao]
      |
      v
Report + Charts

Điểm mấu chốt: Claude không chỉ sinh code — nó còn thực thi code đó, đọc kết quả, và diễn giải ý nghĩa tài chính. Đây là pattern "code interpreter" tương tự ChatGPT Code Interpreter.

Setup môi trường

pip install anthropic pandas numpy matplotlib yfinance scipy

Cấu trúc project:

financial-analyst/
  agent.py          -- Agent chính
  code_executor.py  -- Sandbox thực thi code
  data_tools.py     -- Lấy dữ liệu tài chính
  output/           -- Charts và reports được lưu ở đây

Code Executor — Sandbox an toàn

# code_executor.py
import sys
import io
import traceback
import matplotlib
matplotlib.use('Agg')  # Non-interactive backend
import matplotlib.pyplot as plt
import base64
import os

ALLOWED_IMPORTS = {
    'pandas', 'numpy', 'matplotlib', 'matplotlib.pyplot',
    'scipy', 'math', 'statistics', 'json', 'datetime'
}

def execute_python(code: str, context: dict = None) -> dict:
    """
    Thuc thi Python code trong sandbox.
    Tra ve: output text, loi, va danh sach charts (base64).
    """
    if context is None:
        context = {}

    # Capture stdout
    old_stdout = sys.stdout
    sys.stdout = captured_output = io.StringIO()

    charts = []
    errors = None

    try:
        # Inject san variables tien ich
        exec_globals = {
            '__builtins__': {
                'print': print,
                'range': range,
                'len': len,
                'list': list,
                'dict': dict,
                'str': str,
                'int': int,
                'float': float,
                'bool': bool,
                'round': round,
                'sorted': sorted,
                'enumerate': enumerate,
                'zip': zip,
                'min': min,
                'max': max,
                'sum': sum,
                'abs': abs,
            }
        }

        # Cho phep import thu vien an toan
        import pandas as pd
        import numpy as np
        import matplotlib.pyplot as plt

        exec_globals.update({
            'pd': pd,
            'np': np,
            'plt': plt,
            **context
        })

        # Thuc thi code
        exec(code, exec_globals)

        # Thu thap tat ca charts
        for fig_num in plt.get_fignums():
            fig = plt.figure(fig_num)
            buf = io.BytesIO()
            fig.savefig(buf, format='png', dpi=150, bbox_inches='tight')
            buf.seek(0)
            chart_b64 = base64.b64encode(buf.read()).decode()
            charts.append(chart_b64)
            plt.close(fig)

    except Exception as e:
        errors = traceback.format_exc()

    finally:
        sys.stdout = old_stdout
        plt.close('all')

    output = captured_output.getvalue()

    return {
        "output": output,
        "errors": errors,
        "charts": charts,
        "charts_count": len(charts),
        "success": errors is None
    }

Data Tools — Lấy dữ liệu tài chính thực

# data_tools.py
import json

def get_stock_data(ticker: str, period: str = "6mo") -> str:
    """
    Lay du lieu lich su co phieu tu Yahoo Finance.
    period: 1d, 5d, 1mo, 3mo, 6mo, 1y, 2y, 5y, ytd, max
    """
    try:
        import yfinance as yf
        stock = yf.Ticker(ticker)
        hist = stock.history(period=period)

        if hist.empty:
            return json.dumps({"error": f"Khong tim thay du lieu cho {ticker}"})

        # Chuyen sang JSON-friendly format
        data = {
            "ticker": ticker,
            "period": period,
            "records": len(hist),
            "start_date": str(hist.index[0].date()),
            "end_date": str(hist.index[-1].date()),
            "latest_close": round(float(hist['Close'].iloc[-1]), 2),
            "data_csv": hist.to_csv()
        }
        return json.dumps(data, ensure_ascii=False)

    except Exception as e:
        return json.dumps({"error": str(e)})


def get_portfolio_data(holdings: list) -> str:
    """
    Lay du lieu cho nhieu co phieu.
    holdings: [{"ticker": "VNM", "shares": 100, "avg_cost": 85000}, ...]
    """
    try:
        import yfinance as yf
        portfolio = []

        for h in holdings:
            ticker = h["ticker"]
            shares = h["shares"]
            avg_cost = h.get("avg_cost", 0)

            stock = yf.Ticker(ticker)
            info = stock.fast_info

            current_price = round(float(info.last_price or 0), 2)
            market_value = current_price * shares
            cost_basis = avg_cost * shares
            pnl = market_value - cost_basis
            pnl_pct = (pnl / cost_basis * 100) if cost_basis > 0 else 0

            portfolio.append({
                "ticker": ticker,
                "shares": shares,
                "current_price": current_price,
                "market_value": round(market_value, 2),
                "cost_basis": round(cost_basis, 2),
                "pnl": round(pnl, 2),
                "pnl_pct": round(pnl_pct, 2)
            })

        total_value = sum(p["market_value"] for p in portfolio)
        total_pnl = sum(p["pnl"] for p in portfolio)

        return json.dumps({
            "holdings": portfolio,
            "total_market_value": round(total_value, 2),
            "total_pnl": round(total_pnl, 2),
            "total_pnl_pct": round(total_pnl / (total_value - total_pnl) * 100, 2) if total_value > total_pnl else 0
        }, ensure_ascii=False)

    except Exception as e:
        return json.dumps({"error": str(e)})

Tool Schemas và Agent

# agent.py
import anthropic
import json
from code_executor import execute_python
from data_tools import get_stock_data, get_portfolio_data

client = anthropic.Anthropic()

analyst_tools = [
    {
        "name": "execute_python",
        "description": (
            "Thuc thi Python code de phan tich du lieu tai chinh. "
            "Co san: pandas (pd), numpy (np), matplotlib (plt). "
            "Dung plt.figure() de tao bieu do. Dung print() de hien thi ket qua. "
            "Bien 'stock_data' co san neu da get_stock_data truoc."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "code": {
                    "type": "string",
                    "description": "Python code can thuc thi"
                }
            },
            "required": ["code"]
        }
    },
    {
        "name": "get_stock_data",
        "description": "Lay du lieu lich su gia co phieu tu Yahoo Finance",
        "input_schema": {
            "type": "object",
            "properties": {
                "ticker": {
                    "type": "string",
                    "description": "Ma co phieu, vi du: AAPL, TSLA, VNM.VN"
                },
                "period": {
                    "type": "string",
                    "description": "Khoang thoi gian: 1mo, 3mo, 6mo, 1y, 2y, max",
                    "enum": ["1mo", "3mo", "6mo", "1y", "2y", "5y", "max"]
                }
            },
            "required": ["ticker"]
        }
    },
    {
        "name": "get_portfolio_data",
        "description": "Lay du lieu hien tai cua danh muc dau tu",
        "input_schema": {
            "type": "object",
            "properties": {
                "holdings": {
                    "type": "array",
                    "description": "Danh sach co phieu nam giu",
                    "items": {
                        "type": "object",
                        "properties": {
                            "ticker": {"type": "string"},
                            "shares": {"type": "number"},
                            "avg_cost": {"type": "number"}
                        }
                    }
                }
            },
            "required": ["holdings"]
        }
    }
]

ANALYST_SYSTEM = """Ban la mot chuyen gia phan tich tai chinh co kinh nghiem.
Khi duoc hoi, hay:
1. Thu thap du lieu can thiet bang tools
2. Viet va thuc thi code phan tich chi tiet
3. Tao bieu do truc quan khi phu hop
4. Dien giai ket qua theo goc do tai chinh, khong chi mo ta so lieu
5. Dua ra nhan xet va khuyen nghi thuc te

Luon giai thich y nghia cua chi so, khong chi liet ke so lieu."""


def run_financial_analyst(question: str) -> dict:
    messages = [{"role": "user", "content": question}]
    all_charts = []
    stock_context = {}

    for _ in range(20):
        response = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=8192,
            system=ANALYST_SYSTEM,
            tools=analyst_tools,
            messages=messages
        )

        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason == "end_turn":
            final_text = next(
                (b.text for b in response.content if hasattr(b, "text")), ""
            )
            return {
                "analysis": final_text,
                "charts": all_charts,
                "charts_count": len(all_charts)
            }

        tool_results = []
        for block in response.content:
            if block.type != "tool_use":
                continue

            if block.name == "get_stock_data":
                result = get_stock_data(**block.input)
                data = json.loads(result)
                # Luu CSV vao context de code co the dung
                if "data_csv" in data:
                    stock_context["_last_csv"] = data["data_csv"]
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result
                })

            elif block.name == "get_portfolio_data":
                result = get_portfolio_data(**block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result
                })

            elif block.name == "execute_python":
                # Inject CSV data neu co
                code = block.input["code"]
                if "_last_csv" in stock_context:
                    # Them code load CSV vao dau
                    load_code = (
                        "import io
"
                        "_csv_data = '''" + stock_context["_last_csv"] + "'''
"
                        "stock_data = pd.read_csv(io.StringIO(_csv_data), index_col=0, parse_dates=True)
"
                    )
                    code = load_code + code

                exec_result = execute_python(code)
                all_charts.extend(exec_result.get("charts", []))

                result_str = exec_result["output"]
                if exec_result["errors"]:
                    result_str += "
LOI:
" + exec_result["errors"]
                if exec_result["charts_count"] > 0:
                    result_str += f"
[Da tao {exec_result['charts_count']} bieu do]"

                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result_str
                })

        messages.append({"role": "user", "content": tool_results})

    return {"analysis": "Timeout", "charts": all_charts, "charts_count": len(all_charts)}

Demo: Phân tích cổ phiếu Apple

result = run_financial_analyst(
    "Phan tich co phieu AAPL trong 6 thang qua. "
    "Tinh: return, volatility, Sharpe ratio, va ve bieu do gia + volume."
)

print(result["analysis"])
print(f"Da tao {result['charts_count']} bieu do")

Claude sẽ tự động sinh và chạy code như sau:

# Code Claude tu sinh va thuc thi:
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

# stock_data da duoc inject tu get_stock_data()
close = stock_data['Close']
returns = close.pct_change().dropna()

# Tinh chi so
total_return = (close.iloc[-1] / close.iloc[0] - 1) * 100
volatility = returns.std() * np.sqrt(252) * 100
sharpe = (returns.mean() / returns.std()) * np.sqrt(252)

print(f"Total Return: {total_return:.2f}%")
print(f"Annualized Volatility: {volatility:.2f}%")
print(f"Sharpe Ratio: {sharpe:.2f}")

# Bieu do
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8))
ax1.plot(close.index, close.values)
ax1.set_title('AAPL - Gia co phieu')
ax2.bar(stock_data.index, stock_data['Volume'])
ax2.set_title('Volume')
plt.tight_layout()

Demo: Portfolio Review

portfolio_result = run_financial_analyst(
    "Review danh muc cua toi: AAPL 50 co phieu (mua gia 150), "
    "MSFT 30 co phieu (mua gia 380), GOOGL 20 co phieu (mua gia 140). "
    "Tinh P&L, phan bo danh muc, va khuyen nghi re-balance."
)

print(portfolio_result["analysis"])

Tổng kết: Khả năng của Financial Analyst Agent

Financial Analyst Agent cho thấy sức mạnh của pattern "LLM + Code Execution": thay vì dùng pre-built tools cứng nhắc, Claude tự viết code phù hợp với từng câu hỏi cụ thể — linh hoạt như một data scientist thực sự.

Muốn mở rộng? Thêm dữ liệu thực từ SSI, VPS, hoặc Bloomberg API; kết hợp với Customer Support Agent để tạo robo-advisor; hoặc đọc về LLM Agent từ đầu để nắm vững kiến trúc nền tảng.

Bài viết liên quan

• Claude cho Tài chính: Tổng quan Plugin
• Claude cho Finance: Quản lý close sổ kế toán
• Claude cho Finance: Phân tích báo cáo tài chính
• Customer Support Agent — Chatbot hỗ trợ production-grade
• Claude cho Legal: Đánh giá rủi ro pháp lý