📈 시장 데이터 에이전트 ReAct 패턴

import yfinance as yf
from langchain_core.tools import tool
from typing import Optional
import pandas as pd

@tool
def get_us_stock_price(ticker: str) -> dict:
    """
    미국 주식 종목의 현재가, 등락률, 시가총액, PER을 조회합니다.
    ticker: 미국 주식 심볼 (예: AAPL, MSFT, NVDA, TSLA)
    """
    t = yf.Ticker(ticker.upper())
    info = t.info
    hist = t.history(period="2d")

    if hist.empty or not info:
        return {"error": f"'{ticker}' 데이터를 찾을 수 없습니다."}

    prev_close = hist['Close'].iloc[-2] if len(hist) >= 2 else hist['Close'].iloc[-1]
    curr_close = hist['Close'].iloc[-1]
    change_pct = ((curr_close - prev_close) / prev_close) * 100

    return {
        "ticker": ticker.upper(),
        "name": info.get("longName", ticker),
        "price": round(curr_close, 2),
        "currency": "USD",
        "change_pct": round(change_pct, 2),
        "market_cap": info.get("marketCap"),
        "pe_ratio": info.get("trailingPE"),
        "52w_high": info.get("fiftyTwoWeekHigh"),
        "52w_low": info.get("fiftyTwoWeekLow"),
    }

@tool
def get_us_index(index: str = "SP500") -> dict:
    """
    미국 주요 지수를 조회합니다.
    index: SP500 | NASDAQ | DOW | VIX
    """
    symbol_map = {
        "SP500":  "^GSPC",
        "NASDAQ": "^IXIC",
        "DOW":    "^DJI",
        "VIX":    "^VIX",
    }
    symbol = symbol_map.get(index.upper(), "^GSPC")
    t = yf.Ticker(symbol)
    hist = t.history(period="2d")

    if hist.empty:
        return {"error": f"지수 데이터를 가져올 수 없습니다: {index}"}

    curr = hist['Close'].iloc[-1]
    prev = hist['Close'].iloc[-2] if len(hist) >= 2 else curr
    return {
        "index": index.upper(),
        "value": round(curr, 2),
        "change_pct": round(((curr - prev) / prev) * 100, 2),
    }

@tool
def compare_us_stocks(tickers: str, period: str = "1mo") -> str:
    """
    여러 미국 주식의 수익률을 비교합니다.
    tickers: 쉼표 구분 심볼 (예: "AAPL,MSFT,NVDA")
    period: 1d | 5d | 1mo | 3mo | 6mo | 1y
    """
    symbols = [t.strip().upper() for t in tickers.split(",")]
    data = yf.download(symbols, period=period, auto_adjust=True)["Close"]

    if data.empty:
        return "데이터를 가져올 수 없습니다."

    returns = ((data.iloc[-1] - data.iloc[0]) / data.iloc[0] * 100).round(2)
    result = [f"{sym}: {ret:+.2f}%" for sym, ret in returns.items()]
    return f"{period} 수익률 비교:\n" + "\n".join(result)

from pykrx import stock
from langchain_core.tools import tool
from datetime import datetime, timedelta
import pandas as pd

def _today() -> str:
    return datetime.today().strftime("%Y%m%d")

def _prev_day(days: int = 1) -> str:
    return (datetime.today() - timedelta(days=days)).strftime("%Y%m%d")

@tool
def get_kr_stock_price(ticker: str) -> dict:
    """
    한국 주식 종목의 현재 시세를 조회합니다.
    ticker: 종목 코드 6자리 (예: 005930=삼성전자, 000660=SK하이닉스, 035420=NAVER)
    """
    try:
        name = stock.get_market_ticker_name(ticker)
        df = stock.get_market_ohlcv_by_date(_prev_day(5), _today(), ticker)
        df = df.dropna()
        if df.empty:
            return {"error": f"'{ticker}' 데이터가 없습니다."}

        row = df.iloc[-1]
        prev_close = df.iloc[-2]["종가"] if len(df) >= 2 else row["시가"]
        change_pct = ((row["종가"] - prev_close) / prev_close) * 100

        return {
            "ticker": ticker,
            "name": name,
            "close": int(row["종가"]),
            "currency": "KRW",
            "change_pct": round(change_pct, 2),
            "volume": int(row["거래량"]),
            "open": int(row["시가"]),
            "high": int(row["고가"]),
            "low": int(row["저가"]),
        }
    except Exception as e:
        return {"error": str(e)}

@tool
def search_kr_ticker(company_name: str) -> list:
    """
    회사명으로 한국 주식 종목 코드를 검색합니다.
    company_name: 회사명 (예: 삼성전자, 카카오, 현대차)
    """
    all_tickers = stock.get_market_ticker_list(market="ALL")
    results = []
    for t in all_tickers:
        name = stock.get_market_ticker_name(t)
        if company_name in name:
            results.append({"ticker": t, "name": name})
        if len(results) >= 5:
            break
    return results

@tool
def get_kr_top_stocks(market: str = "KOSPI", by: str = "거래량", limit: int = 5) -> str:
    """
    한국 주식 시장에서 상위 종목을 조회합니다.
    market: KOSPI | KOSDAQ
    by: 거래량 | 거래대금 | 시가총액
    limit: 조회할 종목 수 (최대 10)
    """
    df = stock.get_market_ohlcv_by_ticker(_today(), market=market)
    if df.empty:
        df = stock.get_market_ohlcv_by_ticker(_prev_day(1), market=market)

    top = df.nlargest(min(limit, 10), by)
    lines = [f"{market} {by} 상위 {limit}종목:"]
    for ticker, row in top.iterrows():
        name = stock.get_market_ticker_name(ticker)
        lines.append(f"  {name}({ticker}): {by} {row[by]:,}")
    return "\n".join(lines)

import yfinance as yf
from langchain_core.tools import tool

@tool
def get_exchange_rate(pair: str = "USDKRW") -> dict:
    """
    환율을 조회합니다.
    pair: 통화쌍 (예: USDKRW, EURKRW, JPYKRW, USDEUR)
    """
    # Yahoo Finance 환율 심볼 형식: USDKRW=X
    symbol = f"{pair.upper()}=X"
    t = yf.Ticker(symbol)
    hist = t.history(period="5d")

    if hist.empty:
        return {"error": f"환율 데이터를 찾을 수 없습니다: {pair}"}

    rate = hist['Close'].iloc[-1]
    prev = hist['Close'].iloc[-2] if len(hist) >= 2 else rate
    return {
        "pair": pair.upper(),
        "rate": round(rate, 4),
        "change_pct": round(((rate - prev) / prev) * 100, 4),
    }

from langchain_ollama import ChatOllama
from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.prompts import PromptTemplate
from langchain import hub

from tools.us_stock import get_us_stock_price, get_us_index, compare_us_stocks
from tools.kr_stock import get_kr_stock_price, search_kr_ticker, get_kr_top_stocks
from tools.forex import get_exchange_rate

# 전체 Tool 목록
MARKET_TOOLS = [
    get_us_stock_price,
    get_us_index,
    compare_us_stocks,
    get_kr_stock_price,
    search_kr_ticker,
    get_kr_top_stocks,
    get_exchange_rate,
]

SYSTEM_PROMPT = """당신은 미국과 한국 주식 시장 전문가입니다.
사용자의 질문에 답하기 위해 제공된 도구를 사용하세요.
수치는 항상 구체적으로 제시하고, 단위(USD/KRW/%)를 명확히 표기하세요.
투자 권유는 하지 않으며, 객관적인 데이터만 제공합니다."""

def build_market_agent(model: str = "llama3.1:8b") -> AgentExecutor:
    llm = ChatOllama(model=model, temperature=0)

    # ReAct 프롬프트 템플릿
    prompt = PromptTemplate.from_template(
        SYSTEM_PROMPT + """

사용 가능한 도구:
{tools}

도구 이름 목록: {tool_names}

질문: {input}
{agent_scratchpad}"""
    )

    agent = create_react_agent(llm, MARKET_TOOLS, prompt)
    return AgentExecutor(
        agent=agent,
        tools=MARKET_TOOLS,
        verbose=True,           # 추론 과정 출력
        max_iterations=5,       # 최대 Tool 호출 횟수
        handle_parsing_errors=True,
    )

# 싱글턴
_agent: AgentExecutor | None = None

def get_market_agent() -> AgentExecutor:
    global _agent
    if _agent is None:
        _agent = build_market_agent()
    return _agent

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from agents.market_agent import get_market_agent

router = APIRouter(prefix="/market", tags=["시장 데이터"])

class MarketQuery(BaseModel):
    question: str

class MarketResponse(BaseModel):
    question: str
    answer: str

@router.post("/ask", response_model=MarketResponse)
async def ask_market(query: MarketQuery):
    """자연어 질문으로 시장 데이터를 조회합니다."""
    try:
        agent = get_market_agent()
        result = agent.invoke({"input": query.question})
        return MarketResponse(question=query.question, answer=result["output"])
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

from agents.market_agent import build_market_agent

agent = build_market_agent()

questions = [
    "애플(AAPL) 현재 주가와 PER을 알려줘",
    "오늘 S&P500과 NASDAQ 지수 현황은?",
    "삼성전자 오늘 종가가 얼마야?",
    "NVDA, AAPL, MSFT 최근 1개월 수익률을 비교해줘",
    "달러 원화 환율이 얼마야?",
    "KOSPI 거래량 상위 5종목은?",
]

for q in questions:
    print(f"\nQ: {q}")
    result = agent.invoke({"input": q})
    print(f"A: {result['output']}")