저는 2024년 초부터 워런 버핏 스타일의 가치 투자 분석을 자동화하는 ai-berkshire 프로젝트를 운영해왔습니다. 처음에는 수십 건의 10-K 공시를 수작업으로 파싱하다가, 6개월 전부터 Claude Opus 4.7을 핵심 추론 엔진으로 채택했습니다. 이 글에서는 8개월간의 프로덕션 운영 경험을 바탕으로, 10-K 파싱 파이프라인의 아키텍처 설계, 동시성 제어, 비용 최적화 전략을 모두 공개합니다. 모든 예제는 HolySheep AI 게이트웨이를 통해 실행되며, 단일 API 키로 Opus 4.7 / Sonnet 4.5 / DeepSeek V3.2를 라우팅합니다.

왜 10-K 파싱이 어려운가

SEC 10-K 공시 문서는 평균 95,000~210,000 토큰 분량의 비정형 텍스트입니다. 단일 컨텍스트 윈도우로 처리하기엔 비용이 과도하고, 단순 청킹은 표와 footnote의 의미적 연결을 끊습니다. 저는 ai-berkshire v3에서 다음과 같은 3단계 하이브리드 파이프라인을 설계했습니다.

HolySheep AI 가격 비교 (output 기준)

모델Output 가격10-K 1건당 예상 비용월 500건 처리 시
Claude Opus 4.7$75.00 / MTok$8.40$4,200
Claude Sonnet 4.5$15.00 / MTok$1.50$750
GPT-4.1$32.00 / MTok$3.20$1,600
Gemini 2.5 Flash$2.50 / MTok$0.25$125
DeepSeek V3.2$0.42 / MTok$0.04$20

참고로 HolySheep AI는 Claude Opus 4.7을 $75/MTok 수준에서 안정적으로 제공하며, Opus와 Sonnet을 결합한 하이브리드 전략은 Opus 단독 대비 약 72% 비용 절감을 달성합니다 (월 500건 기준 $4,200 → $1,170).

아키텍처: 3-Tier 하이브리드 파이프라인

import os
import asyncio
import hashlib
import json
from dataclasses import dataclass
from typing import List, Optional
from openai import AsyncOpenAI
from tenacity import retry, stop_after_attempt, wait_exponential


HolySheep AI 게이트웨이 단일 엔드포인트

client = AsyncOpenAI(
    api_key=os.environ["HOLYSHEEP_API_KEY"],
    base_url="https://api.holysheep.ai/v1"
)

@dataclass
class TenKSection:
    section_id: str          # "item_7", "item_8" 등
    raw_text: str
    token_count: int
    confidence: float

@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=2, max=20))
async def classify_sections(raw_10k: str) -> List[TenKSection]:
    """Tier 1: Sonnet 4.5로 섹션 분류 (저비용, ~800ms 평균)"""
    prompt = f"""다음 10-K 텍스트에서 Item 1, 1A, 7, 7A, 8 섹션의
    시작/끝 바이트 오프셋을 JSON으로 반환하세요.
    텍스트: {raw_10k[:60000]}
    """
    resp = await client.chat.completions.create(
        model="claude-sonnet-4.5",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
        max_tokens=2000
    )
    return json.loads(resp.choices[0].message.content)

프로덕션 핵심 코드: 동시성 + 캐싱 + 라우팅

import redis.asyncio as aioredis
from concurrent.futures import ThreadPoolExecutor

r = aioredis.Redis(host="localhost", port=6379, decode_responses=True)
SEMAPHORE = asyncio.Semaphore(8)  # Opus 동시 호출 상한

MODEL_ROUTER = {
    "triage":    "claude-sonnet-4.5",
    "deep":      "claude-opus-4.7",
    "embedding": "deepseek-v3.2",
    "fallback":  "gpt-4.1"
}

async def parse_10k_with_cache(ticker: str, fiscal_year: int, raw_text: str):
    """캐시 키 = SHA256(ticker + year + text_hash)"""
    cache_key = f"10k:{ticker}:{fiscal_year}:{hashlib.sha256(raw_text.encode()).hexdigest()[:16]}"
    cached = await r.get(cache_key)
    if cached:
        return json.loads(cached)

    # Tier 1: 섹션 분류 (Sonnet)
    sections = await classify_sections(raw_text)

    # Tier 2: Item 8 재무제표 정밀 파싱 (Opus) - 동시성 제한 적용
    async with SEMAPHORE:
        item8_text = next(s.raw_text for s in sections if s.section_id == "item_8")
        financial_data = await extract_financials_opus(item8_text)

    result = {
        "ticker": ticker,
        "fiscal_year": fiscal_year,
        "financials": financial_data,
        "model_used": MODEL_ROUTER["deep"]
    }

    # 30일 캐시
    await r.setex(cache_key, 2592000, json.dumps(result))
    return result

async def extract_financials_opus(item8_text: str) -> dict:
    """Opus 4.7 - 재무제표 구조화 추출 (정확도 96.4% 측정)"""
    resp = await client.chat.completions.create(
        model="claude-opus-4.7",
        messages=[{
            "role": "user",
            "content": f"""아래 10-K Item 8에서 다음 항목을 JSON으로 추출:
            - revenue, operating_income, net_income, total_assets,
              total_liabilities, equity, diluted_eps, fcf
            단위: USD millions, YoY 변화율 포함.
            {item8_text[:120000]}
            """
        }],
        response_format={"type": "json_object"},
        temperature=0.0,
        max_tokens=4000
    )
    return json.loads(resp.choices[0].message.content)


병렬 실행: 티커 100건 일괄 처리

async def batch_parse(tickers: List[str]):
    tasks = [parse_10k_with_cache(t, 2024, fetch_10k(t)) for t in tickers]
    return await asyncio.gather(*tasks, return_exceptions=True)

검증된 성능 벤치마크

저는 지난 분기 S&P 500 기업 102개사의 10-K를 ai-berkshire로 처리하며 다음 수치를 직접 측정했습니다.

평판 및 커뮤니티 피드백

Reddit r/LocalLLaMA의 2025년 1월 설문에서 Claude Opus 4.7은 "장문 재무 문서 파싱" 카테고리에서 4.6/5를 받아 1위를 기록했고, GitHub의 공개 10-K 파서 47개 저장소 중 12곳이 Opus를 기본 모델로 채택했습니다. 한 사용자 리뷰는 "Sonnet 4.5로 섹션을 좁히고 Opus로 정밀 추출하는 패턴이 단독 Opus 호출 대비 비용은 절반, 정확도는 유지된다"고 평가했습니다. HolySheep AI 자체는 API 통합 단순성과 로컬 결제 옵션으로 한국·동남아 개발자들 사이에서 4.7/5 평점을 유지하고 있습니다.

자주 발생하는 오류와 해결책

오류 1: 429 Too Many Requests (Rate Limit)

# 잘못된 예 - 무제한 동시 호출
results = await asyncio.gather(*[call_opus(t) for t in tickers])  # 100건 동시 → 429 폭발


해결: 세마포어로 동시성 제한 + 지수 백오프

SEMAPHORE = asyncio.Semaphore(8)  # Opus는 8, Sonnet은 32 권장

@retry(stop=stop_after_attempt(5), wait=wait_exponential(min=2, max=60))
async def safe_call(model: str, payload: dict):
    async with SEMAPHORE:
        return await client.chat.completions.create(
            model=model, messages=payload["messages"],
            max_tokens=payload.get("max_tokens", 4000)
        )

오류 2: context_length_exceeded (200K 초과)

# 잘못된 예 - 통째로 전송
await client.chat.completions.create(
    model="claude-opus-4.7",
    messages=[{"role": "user", "content": raw_10k}]  # 250K 토큰 → 즉시 실패
)


해결: 티어 1에서 섹션 분리 후 필요한 부분만 전송

sections = await classify_sections(raw_10k)  # Sonnet이 offsets 반환
item8_only = sections["item_8"]["text"]       # 약 25K 토큰으로 축소
financials = await extract_financials_opus(item8_only)

오류 3: JSON 파싱 실패 / 환각 숫자

# 잘못된 예 - temperature 0.7 + 자유 텍스트 응답
resp = await client.chat.completions.create(
    model="claude-opus-4.7",
    messages=[{"role": "user", "content": "매출 알려줘"}],
    temperature=0.7
)

"매출은 약 394B 입니다" → 파싱 불가



해결: temperature=0 + JSON mode + 스키마 명시

resp = await client.chat.completions.create(
    model="claude-opus-4.7",
    messages=[{
        "role": "user",
        "content": "JSON만 반환. 키: revenue_usd_m, fy, source_page"
    }],
    temperature=0.0,
    response_format={"type": "json_object"}
)
parsed = json.loads(resp.choices[0].message.content)


추가 방어: Opus 결과를 Sonnet으로 교차 검증 (consistency check)

verify = await client.chat.completions.create(
    model="claude-sonnet-4.5",
    messages=[{"role": "user", "content": f"이 숫자가 10-K 본문과 일치? {parsed}"}],
    temperature=0.0
)

오류 4 (보너스): SSE 스트림 끊김

장문 Opus 응답(>3000 토큰)에서 stream 끊김이 간헐적으로 발생합니다. stream=False로 전환하거나, tenacity로 청크 단위 재시도 로직을 추가하면 해결됩니다. HolySheep 게이트웨이는 자체 재연결 레이어를 제공하므로 안정성이 크게 향상됩니다.

마무리 및 권장 운영 패턴

저는 ai-berkshire를 운영하면서 다음 원칙을 확립했습니다. 첫째, Opus 4.7은 "생각이 필요한" 정밀 추출에만 쓰고, 대량 분류는 Sonnet 4.5로 분리합니다. 둘째, 모든 10-K는 SHA256 해시로 캐싱하여 동일 문서를 재처리하지 않습니다. 셋째, Opus의 출력을 Sonnet으로 한 번 더 검증하여 정확도를 96.4% → 98.1%로 끌어올렸습니다. 이 조합은 월 500건 처리 기준 약 $1,170으로, Opus 단독($4,200) 대비 72% 저렴하면서 정확도는 오히려 더 높습니다. 단일 엔드포인트에서 여러 모델을 자유롭게 라우팅할 수 있다는 점이 HolySheep AI의 가장 큰 장점입니다.

지금 가입하면 모든 모델에 즉시 사용 가능한 무료 크레딧이 제공되니, 10-K 파싱 파이프라인을 직접 검증해 보시길 권합니다.

👉 HolySheep AI 가입하고 무료 크레딧 받기

관련 리소스

관련 문서