2026-04-16 강의 정리

---

검색용 문서를 만들 때는 설명문과 필터에 사용할 구조화 데이터를 구분해야 한다. ETF 정보를 예제로 수치 계산, 설명 생성, 인덱싱을 연결하되 원본 속성이 생성 문장에 묻히지 않게 구성한다. 검색 결과를 설명할 수 있도록 문서와 원본 항목의 연결도 함께 유지한다.

목차

1. 프로젝트 아키텍처 개요
2. ETF 데이터 수집 & 성과 지표 계산
3. LLM으로 ETF 설명 생성 (Document 구성)
4. FAISS 벡터스토어 구축
5. BM25 키워드 검색 (Kiwi 형태소 분석)
6. 필터 기반 검색 (filtered_search)
7. BM25 파라미터 튜닝
8. 자주 나오는 실수 / 주의사항
9. [보충] 금융 챗봇 전체 파이프라인
10. 검색 문서와 원본 데이터의 연결 유지

---

1. 프로젝트 아키텍처 개요

[ETF 금융 챗봇 파이프라인]

Step 1. 데이터 수집
  FinanceDataReader → ETF 가격 데이터 (1년치)

Step 2. 성과 지표 계산
  수익률, 변동성, MDD → etf_data[]

Step 3. LLM 설명 생성
  GPT-4o-mini → ETF 한 줄 설명 → Document

Step 4. 벡터스토어 구축
  OpenAIEmbeddings → FAISS 인덱싱

Step 5. 검색 (RAG Retriever)
  벡터 검색 + BM25 키워드 검색

Step 6. 사용자 쿼리 응답
  검색 결과 → GPT-4o-mini → 답변

---

2. ETF 데이터 수집 & 성과 지표 계산

종목 목록 정의

tickers_info = {
    "069500": {"category": "국내주식", "expense_ratio": 0.15, "dividend_yield": 1.8,
               "keywords": ["코스피", "대형주", "인덱스", "분산투자", "국내주식"]},
    "379800": {"category": "해외주식", "expense_ratio": 0.05, "dividend_yield": 0.0,
               "keywords": ["미국", "S&P500", "대형주", "성장", "해외주식"]},
    "132030": {"category": "원자재",   "expense_ratio": 0.68, "dividend_yield": 0.0,
               "keywords": ["금", "원자재", "인플레이션", "헤지", "안전자산"]},
    "214980": {"category": "채권",     "expense_ratio": 0.05, "dividend_yield": 3.2,
               "keywords": ["단기채", "안전", "예금대안", "저위험", "이자"]},
    # ... 총 12개 종목
}

가격 데이터 수집

def collect_etf_price(ticker, days=365):
    end_date   = datetime.now().strftime('%Y-%m-%d')
    start_date = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
    df = fdr.DataReader(ticker, start_date, end_date)
    return {'status': 'real', 'data': df, 'ticker': ticker}

성과 지표 계산

def compute_metrics(df):
    close   = df['Close'].ffill()
    ret     = close.pct_change().dropna()
    ann_ret = ((1 + ret.mean()) ** 252 - 1) * 100    # 연간 수익률 (%)
    ann_vol = ret.std() * np.sqrt(252) * 100           # 연간 변동성 (%)
    cum     = (1 + ret).cumprod()
    mdd     = ((cum - cum.cummax()) / cum.cummax()).min() * 100  # MDD (%)
    return {'return_1y': round(ann_ret, 2), 'volatility': round(ann_vol, 2), 'mdd': round(mdd, 2)}

def risk_from_vol(vol):
    if vol < 5:   return "낮음"
    if vol < 15:  return "약간 낮음"
    if vol < 25:  return "중간"
    return "높음"

ETF 데이터 통합

etf_data = []
for t, info in tickers_info.items():
    result  = collect_etf_price(t, days=365)
    metrics = compute_metrics(result['data'])
    name    = name_by_ticker.get(t, f"ETF_{t}")
    etf_data.append({
        "ticker": t, "name": name,
        **info,
        **metrics,
        'risk_level':   risk_from_vol(metrics['volatility']),
        'data_status':  result['status']
    })

실행 결과 예시:

{'ticker': '069500', 'name': 'KODEX 200', 'category': '국내주식',
 'expense_ratio': 0.15, 'dividend_yield': 1.8,
 'return_1y': 227.44, 'volatility': 36.58, 'mdd': -20.62,
 'risk_level': '높음', 'data_status': 'real'}

---

3. LLM으로 ETF 설명 생성 (Document 구성)

생성한 설명은 검색 표현을 풍부하게 만드는 용도로 사용할 수 있지만 수치의 원본을 대체하지는 않는다. 비용이나 분류 조건처럼 정확한 비교가 필요한 값은 메타데이터에 보존하고, 설명문과 값이 어긋나는지 확인한다.

ETF 설명 생성

def generate_description(etf):
    prompt = f"""다음 ETF의 특징을 한국어 1~2문장으로 간결히 설명하세요.
    이름: {etf['name']}
    카테고리: {etf['category']}
    키워드: {','.join(etf['keywords'])}
    수수료: {etf['expense_ratio']}% / 배당수익률: {etf['dividend_yield']}%
    1년수익률: {etf['return_1y']}% / 변동성: {etf['volatility']}% / MDD: {etf['mdd']}%
    """
    return llm.invoke([{'role': 'user', 'content': prompt}]).content.strip()

Document 구성

RAG에서 검색 단위가 되는 Document 객체 생성.

from langchain_core.documents import Document

documents = []
for etf in etf_data:
    desc = generate_description(etf)

    # page_content: 검색에 사용될 텍스트
    text = (f"{etf['name']} ({etf['category']}): {desc} "
            f"키워드: {', '.join(etf['keywords'])}"
            f"수수료 {etf['expense_ratio']}%, 배당수익률: {etf['dividend_yield']}%"
            f"수익률: {etf['return_1y']}% / 변동성: {etf['volatility']}% / MDD: {etf['mdd']}%")

    # metadata: 필터링에 사용될 구조화 정보
    metadata = {k: v for k, v in etf.items() if k != "keywords"}
    metadata['keywords'] = ', '.join(etf['keywords'])

    documents.append(Document(page_content=text, metadata=metadata))

Document 구조:

page_content: "KODEX 200 (국내주식): KODEX 200은 코스피 대형주에 투자하는 국내주식 인덱스 ETF로 ..."
metadata:     {'ticker': '069500', 'category': '국내주식', 'expense_ratio': 0.15, ...}

page_content는 임베딩 검색/BM25 대상, metadata는 필터 조건으로 활용

---

4. FAISS 벡터스토어 구축

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS as LangchainFAISS

embeddings = OpenAIEmbeddings(model='text-embedding-3-small')

# 문서 임베딩 + 인덱스 생성 (원샷)
vectorstore = LangchainFAISS.from_documents(documents, embeddings)

vectorstore.index.ntotal  # → 12 (총 12개 ETF 문서)

벡터 검색

query = '미국 기술주에 투자하고 싶어요'
results = vectorstore.similarity_search_with_score(query, k=3)

# 결과: [(Document, score), ...]
# score: L2 거리 (낮을수록 유사)

실행 결과:

1. TIGER 미국필라델피아반도체나스닥  (score: 1.18) ← 가장 관련
2. KODEX 미국나스닥100             (score: 1.25)
3. ACE KRX금현물                  (score: 1.37) ← 배당 ETF이지만 "미국" 키워드 매칭

---

5. BM25 키워드 검색 (Kiwi 형태소 분석)

왜 한국어 형태소 분석이 필요한가?

단순 공백 분리로는 “배당수익률”, “배당” “수익” “률”로 분리 불가. Kiwi가 형태소 단위로 분리.

from kiwipiepy import Kiwi
from rank_bm25 import BM25Okapi

kiwi = Kiwi()

def kiwi_tokenize(text):
    tokens = kiwi.tokenize(text)
    result = []
    for token in tokens:
        # SL: 외래어/영어, SN: 숫자, NNG: 일반명사, NNP: 고유명사
        if token.tag in ("SL", "SN", "NNG", "NNP"):
            result.append(token.form.lower())
    return result

예시:

kiwi_tokenize("KODEX 200 ETF는 어떤 종류의 주식에 투자하나요?")
# → ['kodex', '200', 'etf', '종류', '주식', '투자']

kiwi_tokenize("배당수익률 3% 이상인 미국 ETF를 추천해주세요")
# → ['배당', '수익', '3', '이상', '미국', 'etf', '추천']

BM25 인덱스 구축

corpus = [kiwi_tokenize(doc.page_content) for doc in documents]
bm25   = BM25Okapi(corpus)

def bm25_search(query, k=5):
    tokens = kiwi_tokenize(query)
    scores = bm25.get_scores(tokens)
    top_idx = np.argsort(scores)[::-1][:k]
    return [(documents[i], scores[i]) for i in top_idx if scores[i] > 0]

벡터 검색 vs BM25 비교

쿼리              | 벡터 검색 결과        | BM25 결과
"KODEX 200"      | KODEX 200           | KODEX 200           ← 동일
"안전한 투자 상품"  | KODEX 단기채권PLUS   | KODEX 단기채권PLUS   ← 동일
"배당 ETF"        | ACE KRX금현물        | ACE KRX금현물        ← 동일
"인플레이션 방어"   | KIWOOM 국고채10년    | KODEX 골드선물(H)    ← 다름!

“인플레이션 방어”에서 벡터는 채권을 찾지만, BM25는 “인플레이션 헤지” 키워드가 있는 골드를 찾음 → 두 방법 모두 필요

---

6. 필터 기반 검색 (filtered_search)

필터의 적용 순서는 반환 후보에 영향을 준다. 검색 후 일부 문서를 제거하는 방식이라면 남은 개수와 조건 만족 여부를 함께 확인하고, 결과가 적다는 이유로 조건에 맞는 문서 자체가 없다고 단정하지 않는다.

메타데이터 조건으로 결과를 필터링하는 커스텀 검색.

def filtered_search(vectorstore, query, filters=None, k=5, fetch_k=20):
    """
    filters 형식:
      {"category": "해외주식"}                           # 등호 조건
      {"expense_ratio": {"less_than": 0.5}}             # 범위 조건
      {"category": "해외주식", "expense_ratio": {...}}   # 복합 조건
    """
    results = vectorstore.similarity_search_with_score(query, fetch_k)
    if not filters:
        return results[:k]

    filtered = []
    for doc, score in results:
        match = True
        for key, condition in filters.items():
            val = doc.metadata.get(key)
            if isinstance(condition, dict):
                if "less_than"    in condition and val > condition["less_than"]:    match = False
                if "greater_than" in condition and val < condition["greater_than"]: match = False
            elif val != condition:
                match = False
        if match:
            filtered.append((doc, score))

    return filtered[:k]

사용 예

results = filtered_search(
    vectorstore,
    "저비용 해외 ETF",
    filters={"category": "해외주식", "expense_ratio": {"less_than": 0.5}},
    k=3
)

---

7. BM25 파라미터 튜닝 (k1, b)

query  = "배당 ETF"
tokens = kiwi_tokenize(query)
params = [(0.5, 0.25), (1.5, 0.75), (2.5, 0.9)]

for k1, b in params:
    bm25_temp = BM25Okapi(corpus, k1=k1, b=b)
    scores    = bm25_temp.get_scores(tokens)
    top3      = np.argsort(scores)[::-1][:3]
    print(f"k1={k1}, b={b}")
    for idx in top3:
        print(f"  [{scores[idx]:.3f}] {documents[idx].metadata['name']}")

결과: k1, b 값에 무관하게 상위 ETF 순서 동일 → 이 도메인에서는 기본값으로도 충분

파라미터 의미:

파라미터	역할	기본값
k1	단어 빈도 포화점 (클수록 빈도 효과 강함)	1.5
b	문서 길이 정규화 (1.0 = 완전 정규화, 0 = 정규화 없음)	0.75

---

자주 나오는 실수 / 주의사항

• generate_description을 루프에서 반복 호출: API 요금 누적, 속도 느림 → 결과를 캐싱하거나 배치 처리
• Document metadata에 numpy float 타입 저장: FAISS 저장/로드 시 직렬화 오류 → float(metrics['return_1y']) 등 Python 기본 타입으로 변환
• BM25 점수 0인 문서 포함: if scores[i] > 0 조건으로 관련 없는 문서 제거
• similarity_search_with_score의 score: 벡터 거리(L2), 낮을수록 좋음 ↔ BM25 점수는 높을수록 좋음 → 두 점수 합산 시 정규화 필수

---

[보충] 금융 챗봇 전체 파이프라인

[ETF 금융 챗봇]

1. 데이터 수집
   └─ FinanceDataReader → 가격 데이터
   └─ tickers_info → 카테고리, 수수료, 키워드

2. 데이터 생성 (RAG용 Document)
   └─ compute_metrics() → 수익률, 변동성, MDD
   └─ generate_description() → LLM 설명 문구
   └─ Document(page_content, metadata)

3. 인덱싱
   └─ FAISS vectorstore (벡터 검색)
   └─ BM25Okapi (키워드 검색)

4. 검색 (사용자 쿼리)
   └─ filtered_search() → 조건 + 벡터 검색
   └─ bm25_search() → 키워드 검색
   └─ hybrid 결합 → 상위 문서 선별

5. 답변 생성
   └─ 상위 문서 컨텍스트 → GPT-4o-mini → 최종 답변

---

10. 검색 문서와 원본 데이터의 연결 유지

인덱스를 만들 때는 항목 식별자와 데이터 기준 시점을 남긴다. 설명문만 저장하면 정보가 갱신됐을 때 어떤 문서를 다시 생성하고 교체해야 하는지 찾기 어려워진다.

평가 질문은 표현 유사도와 명시적 조건을 나누어 준비한다. 자연어 설명이 잘 맞는지와 숫자·분류 조건을 만족하는지는 다른 확인 항목이며, 하나의 검색 점수로 둘을 대신하지 않는다.