📊 기술적 분석 에이전트 Tool Use 패턴

import yfinance as yf
import pandas as pd
import pandas_ta as ta
from langchain_core.tools import tool
from pykrx import stock
from datetime import datetime, timedelta

def _get_ohlcv_us(ticker: str, period: str = "6mo") -> pd.DataFrame:
    df = yf.Ticker(ticker.upper()).history(period=period)
    df.index = pd.to_datetime(df.index).tz_localize(None)
    return df[["Open", "High", "Low", "Close", "Volume"]].dropna()

def _get_ohlcv_kr(ticker: str, days: int = 180) -> pd.DataFrame:
    end = datetime.today().strftime("%Y%m%d")
    start = (datetime.today() - timedelta(days=days)).strftime("%Y%m%d")
    df = stock.get_market_ohlcv_by_date(start, end, ticker)
    df.columns = ["Open", "High", "Low", "Close", "Volume"]
    return df.dropna()

@tool
def get_moving_average(ticker: str, market: str = "US", windows: str = "5,20,60") -> str:
    """
    이동평균선과 골든크로스/데드크로스 여부를 계산합니다.
    ticker: 종목코드 (미국: AAPL, 한국: 005930)
    market: US | KR
    windows: 이동평균 기간 (쉼표 구분, 예: "5,20,60,120")
    """
    df = _get_ohlcv_us(ticker) if market == "US" else _get_ohlcv_kr(ticker)
    periods = [int(w.strip()) for w in windows.split(",")]

    lines = [f"[{ticker} 이동평균 분석]"]
    curr_price = df["Close"].iloc[-1]
    lines.append(f"현재가: {curr_price:,.2f}")

    ma_values = {}
    for p in periods:
        if len(df) >= p:
            ma = df["Close"].rolling(p).mean().iloc[-1]
            ma_values[p] = ma
            position = "위" if curr_price > ma else "아래"
            lines.append(f"MA{p}: {ma:,.2f} (현재가 {position})")

    # 골든크로스 판단 (단기 MA가 장기 MA 위로)
    sorted_periods = sorted(ma_values.keys())
    if len(sorted_periods) >= 2:
        short, long_ = sorted_periods[0], sorted_periods[-1]
        short_ma_prev = df["Close"].rolling(short).mean().iloc[-2]
        long_ma_prev  = df["Close"].rolling(long_).mean().iloc[-2]
        short_ma_now  = ma_values[short]
        long_ma_now   = ma_values[long_]

        if short_ma_now > long_ma_now and short_ma_prev <= long_ma_prev:
            lines.append(f"🟢 골든크로스 발생 (MA{short} > MA{long_})")
        elif short_ma_now < long_ma_now and short_ma_prev >= long_ma_prev:
            lines.append(f"🔴 데드크로스 발생 (MA{short} < MA{long_})")
        else:
            trend = "MA 정배열" if short_ma_now > long_ma_now else "MA 역배열"
            lines.append(f"현재 추세: {trend}")

    return "\n".join(lines)

@tool
def get_rsi(ticker: str, market: str = "US", period: int = 14) -> str:
    """
    RSI(상대강도지수)를 계산해 과매수/과매도 여부를 판단합니다.
    ticker: 종목코드
    market: US | KR
    period: RSI 계산 기간 (기본 14일)
    """
    df = _get_ohlcv_us(ticker) if market == "US" else _get_ohlcv_kr(ticker)
    rsi_series = ta.rsi(df["Close"], length=period)

    if rsi_series is None or rsi_series.dropna().empty:
        return "RSI를 계산할 수 없습니다 (데이터 부족)."

    rsi_now  = rsi_series.iloc[-1]
    rsi_prev = rsi_series.iloc[-2]

    if rsi_now >= 70:
        signal = "🔴 과매수 구간 (매도 고려)"
    elif rsi_now <= 30:
        signal = "🟢 과매도 구간 (매수 고려)"
    elif rsi_now > rsi_prev:
        signal = "↗ RSI 상승 중 (모멘텀 강화)"
    else:
        signal = "↘ RSI 하락 중 (모멘텀 약화)"

    return (
        f"[{ticker} RSI({period})]\n"
        f"현재 RSI: {rsi_now:.1f}\n"
        f"전일 RSI: {rsi_prev:.1f}\n"
        f"판단: {signal}"
    )

@tool
def get_macd(ticker: str, market: str = "US") -> str:
    """
    MACD(이동평균 수렴확산지수)를 계산해 추세 전환 시그널을 제공합니다.
    ticker: 종목코드
    market: US | KR
    """
    df = _get_ohlcv_us(ticker) if market == "US" else _get_ohlcv_kr(ticker)
    macd_df = ta.macd(df["Close"], fast=12, slow=26, signal=9)

    if macd_df is None or macd_df.dropna().empty:
        return "MACD를 계산할 수 없습니다."

    macd     = macd_df["MACD_12_26_9"].iloc[-1]
    signal   = macd_df["MACDs_12_26_9"].iloc[-1]
    hist     = macd_df["MACDh_12_26_9"].iloc[-1]
    hist_prev = macd_df["MACDh_12_26_9"].iloc[-2]

    if hist > 0 and hist_prev <= 0:
        cross = "🟢 MACD 골든크로스 (상승 전환 시그널)"
    elif hist < 0 and hist_prev >= 0:
        cross = "🔴 MACD 데드크로스 (하락 전환 시그널)"
    elif hist > 0:
        cross = "📈 MACD 양봉 유지 (상승 추세)"
    else:
        cross = "📉 MACD 음봉 유지 (하락 추세)"

    return (
        f"[{ticker} MACD(12,26,9)]\n"
        f"MACD: {macd:.4f}\n"
        f"Signal: {signal:.4f}\n"
        f"Histogram: {hist:.4f}\n"
        f"판단: {cross}"
    )

@tool
def get_bollinger_bands(ticker: str, market: str = "US", period: int = 20) -> str:
    """
    볼린저밴드를 계산해 변동성과 가격 위치를 분석합니다.
    ticker: 종목코드
    market: US | KR
    period: 이동평균 기간 (기본 20일)
    """
    df = _get_ohlcv_us(ticker) if market == "US" else _get_ohlcv_kr(ticker)
    bb = ta.bbands(df["Close"], length=period, std=2)

    if bb is None or bb.dropna().empty:
        return "볼린저밴드를 계산할 수 없습니다."

    upper  = bb[f"BBU_{period}_2.0"].iloc[-1]
    middle = bb[f"BBM_{period}_2.0"].iloc[-1]
    lower  = bb[f"BBL_{period}_2.0"].iloc[-1]
    curr   = df["Close"].iloc[-1]
    width  = (upper - lower) / middle * 100  # 밴드폭 %

    pct_b  = (curr - lower) / (upper - lower) * 100  # 밴드 내 위치

    if curr >= upper:
        position = "🔴 상단 밴드 터치/돌파 (과매수 경고)"
    elif curr <= lower:
        position = "🟢 하단 밴드 터치/돌파 (과매도, 반등 기대)"
    elif pct_b >= 70:
        position = "↗ 상단 밴드 근접 (강세)"
    elif pct_b <= 30:
        position = "↘ 하단 밴드 근접 (약세)"
    else:
        position = "↔ 밴드 중앙부 (방향 탐색 중)"

    return (
        f"[{ticker} 볼린저밴드({period})]\n"
        f"상단: {upper:,.2f}\n"
        f"중심(MA{period}): {middle:,.2f}\n"
        f"하단: {lower:,.2f}\n"
        f"현재가: {curr:,.2f} (밴드 내 위치: {pct_b:.1f}%)\n"
        f"밴드폭: {width:.1f}%\n"
        f"판단: {position}"
    )

from langchain_core.tools import tool
from tools.technical import (
    _get_ohlcv_us, _get_ohlcv_kr,
    get_rsi, get_macd, get_bollinger_bands, get_moving_average,
)
import pandas_ta as ta

@tool
def get_composite_signal(ticker: str, market: str = "US") -> str:
    """
    RSI, MACD, 볼린저밴드, 이동평균을 종합한 매수·매도 시그널을 생성합니다.
    ticker: 종목코드
    market: US | KR
    """
    df = _get_ohlcv_us(ticker) if market == "US" else _get_ohlcv_kr(ticker)

    scores = []

    # RSI
    rsi = ta.rsi(df["Close"], length=14).iloc[-1]
    if rsi <= 30:   scores.append(("RSI", +2, f"과매도({rsi:.1f})"))
    elif rsi <= 45: scores.append(("RSI", +1, f"중립 하단({rsi:.1f})"))
    elif rsi >= 70: scores.append(("RSI", -2, f"과매수({rsi:.1f})"))
    elif rsi >= 55: scores.append(("RSI", -1, f"중립 상단({rsi:.1f})"))
    else:           scores.append(("RSI",  0, f"중립({rsi:.1f})"))

    # MACD
    macd_df = ta.macd(df["Close"])
    if macd_df is not None and not macd_df.dropna().empty:
        hist = macd_df.iloc[-1, 2]
        prev = macd_df.iloc[-2, 2]
        if hist > 0 and prev <= 0:   scores.append(("MACD", +2, "골든크로스"))
        elif hist > 0:               scores.append(("MACD", +1, "양봉"))
        elif hist < 0 and prev >= 0: scores.append(("MACD", -2, "데드크로스"))
        else:                        scores.append(("MACD", -1, "음봉"))

    # 볼린저밴드
    bb = ta.bbands(df["Close"])
    if bb is not None and not bb.dropna().empty:
        curr = df["Close"].iloc[-1]
        upper = bb.iloc[-1, 0]
        lower = bb.iloc[-1, 2]
        pct_b = (curr - lower) / (upper - lower) * 100
        if pct_b <= 20:   scores.append(("BB", +2, f"하단근접({pct_b:.0f}%)"))
        elif pct_b >= 80: scores.append(("BB", -2, f"상단근접({pct_b:.0f}%)"))
        else:             scores.append(("BB",  0, f"중앙({pct_b:.0f}%)"))

    # 이동평균 정배열
    ma20  = df["Close"].rolling(20).mean().iloc[-1]
    ma60  = df["Close"].rolling(60).mean().iloc[-1]
    curr_p = df["Close"].iloc[-1]
    if curr_p > ma20 > ma60: scores.append(("MA", +1, "정배열"))
    elif curr_p < ma20 < ma60: scores.append(("MA", -1, "역배열"))
    else:                      scores.append(("MA",  0, "혼조"))

    total = sum(s for _, s, _ in scores)
    detail = " | ".join(f"{name}:{label}" for name, _, label in scores)

    if   total >= 4:  judgment = "🟢 강한 매수"
    elif total >= 2:  judgment = "🟡 약한 매수"
    elif total <= -4: judgment = "🔴 강한 매도"
    elif total <= -2: judgment = "🟠 약한 매도"
    else:             judgment = "⚪ 중립"

    return (
        f"[{ticker} 종합 시그널]\n"
        f"점수: {total:+d}/8\n"
        f"판단: {judgment}\n"
        f"근거: {detail}"
    )

from langchain_core.tools import tool
import yfinance as yf
import pandas as pd
import pandas_ta as ta

@tool
def run_rsi_backtest(ticker: str, period: str = "1y",
                     rsi_buy: float = 30, rsi_sell: float = 70) -> str:
    """
    RSI 기반 단순 매수·매도 전략을 백테스팅합니다.
    ticker: 미국 주식 심볼 (예: AAPL)
    period: 백테스팅 기간 (1y | 2y | 5y)
    rsi_buy: 매수 기준 RSI (기본 30, 과매도)
    rsi_sell: 매도 기준 RSI (기본 70, 과매수)
    """
    df = yf.Ticker(ticker.upper()).history(period=period)
    df["RSI"] = ta.rsi(df["Close"], length=14)
    df = df.dropna()

    position = False
    buy_price = 0.0
    trades = []

    for i, row in df.iterrows():
        if not position and row["RSI"] <= rsi_buy:
            position = True
            buy_price = row["Close"]
        elif position and row["RSI"] >= rsi_sell:
            ret = (row["Close"] - buy_price) / buy_price * 100
            trades.append(ret)
            position = False

    if not trades:
        return f"{ticker} {period} 백테스트: 거래 발생 없음 (RSI {rsi_buy}/{rsi_sell} 조건 미충족)"

    total_ret = sum(trades)
    win_rate  = sum(1 for t in trades if t > 0) / len(trades) * 100
    bh_ret    = (df["Close"].iloc[-1] - df["Close"].iloc[0]) / df["Close"].iloc[0] * 100

    return (
        f"[{ticker} RSI({rsi_buy}/{rsi_sell}) 백테스트 — {period}]\n"
        f"총 거래 횟수: {len(trades)}회\n"
        f"승률: {win_rate:.1f}%\n"
        f"전략 누적 수익률: {total_ret:+.1f}%\n"
        f"Buy & Hold 수익률: {bh_ret:+.1f}%\n"
        f"초과 수익: {total_ret - bh_ret:+.1f}%"
    )

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

from tools.technical import get_moving_average, get_rsi, get_macd, get_bollinger_bands
from tools.signal import get_composite_signal
from tools.backtest import run_rsi_backtest
from tools.us_stock import get_us_stock_price
from tools.kr_stock import get_kr_stock_price

TECHNICAL_TOOLS = [
    get_composite_signal,
    get_moving_average,
    get_rsi,
    get_macd,
    get_bollinger_bands,
    run_rsi_backtest,
    get_us_stock_price,
    get_kr_stock_price,
]

SYSTEM = """당신은 주식 기술적 분석 전문가입니다.
RSI, MACD, 볼린저밴드, 이동평균 등 기술적 지표를 해석해 투자자에게 명확한 분석을 제공합니다.

분석 원칙:
- 먼저 종합 시그널을 확인하고, 세부 지표로 보완합니다
- 지표 수치를 구체적으로 제시합니다
- 기술적 분석의 한계를 명시합니다
- 투자 권유가 아닌 분석 정보만 제공합니다"""

def build_technical_agent() -> AgentExecutor:
    llm = ChatOllama(model="llama3.1:8b", temperature=0)
    prompt = PromptTemplate.from_template(
        SYSTEM + "\n\n사용 가능한 도구:\n{tools}\n\n도구 이름: {tool_names}\n\n질문: {input}\n{agent_scratchpad}"
    )
    agent = create_react_agent(llm, TECHNICAL_TOOLS, prompt)
    return AgentExecutor(agent=agent, tools=TECHNICAL_TOOLS, verbose=True,
                         max_iterations=6, handle_parsing_errors=True)

from agents.technical_agent import build_technical_agent

agent = build_technical_agent()

questions = [
    "AAPL 현재 RSI와 MACD를 분석해줘",
    "삼성전자(005930) 기술적 분석 종합 시그널은?",
    "NVDA 볼린저밴드 위치와 이동평균 정배열 여부를 알려줘",
    "AAPL에서 RSI 30/70 전략으로 1년 백테스팅 해줘",
    "SK하이닉스(000660) 골든크로스 발생했어?",
]

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