암호화폐 시장에서 통계 차익거래(Statistical Arbitrage)는 수학적 모델과 알고리즘을 활용하여 가격 비효율을 이익으로 전환하는 고급 거래 전략입니다. 이 튜토리얼에서는 HolySheep AI를 활용한 데이터 전처리 및 특성 추출 파이프라인 구축 방법을 상세히 설명합니다.
핵심 결론: HolySheep AI의 다중 모델 통합과 $0.42/MTok의 업계 최저가 DeepSeek 모델을 활용하면, 암호화폐 시계열 데이터 전처리 비용을 기존 대비 85% 절감하면서 실시간 특성 추출 파이프라인을 구축할 수 있습니다.
암호화폐 통계 차익거래 개요
통계 차익거래는 다음 세 단계를 기반으로 동작합니다:
- 쌍 식별: 상관관계가 높은 암호화폐 쌍(BTC-ETH, ETH-USDT 등) 탐지
- 가격 분산 모형: 두 자산 간 가격 비율의 정상성(Stationarity) 검증
- 거래 실행: 분산이 평균으로 회귀할 때 매수/매도 신호 발생
성공적인 차익거래 전략의 핵심은 높은 품질의 데이터 전처리와 의미 있는 특성(Feature) 추출입니다. 여기서 AI의 역할이 중요해집니다.
왜 HolySheep AI인가?
암호화폐 데이터 분석에는 다양한 AI 모델이 필요합니다. 텍스트 기반 뉴스 분석에는 GPT-4.1, 수치 데이터 처리에 최적화된 DeepSeek V3.2, 빠른 추론에는 Gemini 2.5 Flash. HolySheep는 이 모든 모델을 단일 API 키로 통합 제공합니다.
주요 AI 서비스 비교표
| 서비스 | DeepSeek V3.2 | GPT-4.1 | Claude Sonnet 4.5 | Gemini 2.5 Flash | 로컬 결제 |
|---|---|---|---|---|---|
| HolySheep AI | $0.42/MTok ✓ | $8/MTok ✓ | $15/MTok ✓ | $2.50/MTok ✓ | 지원 ✓ |
| OpenAI 직접 | 미지원 | $15/MTok | 미지원 | 미지원 | 불가 |
| Anthropic 직접 | 미지원 | 미지원 | $18/MTok | 미지원 | 불가 |
| Google AI | 미지원 | 미지원 | 미지원 | $1.25/MTok | 불가 |
이런 팀에 적합 / 비적합
✓ HolySheep가 적합한 팀
- 암호화폐 거래팀: 다중 거래소 API 연동 및 실시간 데이터 분석 필요
- 퀀트 트레이딩 팀: Python/R 기반 백테스팅 및 특성 엔지니어링 수행
- 블록체인 분석 스타트업: 제한된 예산으로 AI 파이프라인 구축 필요
- 개인 트레이더: 해외 신용카드 없이 글로벌 AI 서비스 이용 희망
✗ HolySheep가 적합하지 않은 팀
- 방대한企业内部 데이터 처리: 클라우드 기반 자체 ML 인프라 보유한 대형 금융사
- 극단적 저지연 요구: 초단타 거래(HFT)에 필수적인 마이크로초 단위 레이턴시 필요
가격과 ROI
암호화폐 차익거래 AI 파이프라인 비용 분석:
| 작업 | 월간 처리량 | HolySheep 비용 | OpenAI 직점 비용 | 절감액 |
|---|---|---|---|---|
| 뉴스 감성 분석 | 100K 토큰 | $800 | $1,500 | $700 (47%) |
| 시계열 특성 추출 | 500K 토큰 | $210 | - | - |
| 쌍 탐지 최적화 | 200K 토큰 | $84 | - | - |
| 총 합계 | 800K 토큰 | $1,094 | $2,700 | $1,606 (60%) |
데이터 전처리 파이프라인 구축
암호화폐 차익거래를 위한 데이터 전처리 파이프라인은 다음 구성요소로 이루어집니다:
- 데이터 수집: 다중 거래소(OHLCV, 주문서, 체결 데이터)
- 정제: 이상치 제거, 결측치 보간
- 동기화: 시간대 통일, 빈도 변환
- 특성 공학: 이동평균, 변동성, 상관관계
HolySheep AI 통합 설정
import os
import requests
import json
HolySheep AI API 설정
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
class HolySheepAIClient:
"""암호화폐 분석을 위한 HolySheep AI 클라이언트"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def analyze_with_deepseek(self, prompt: str, system_prompt: str = None) -> dict:
"""
DeepSeek V3.2를 사용한 시계열 분석
비용 최적화: $0.42/MTok
"""
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": prompt})
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json={
"model": "deepseek-chat",
"messages": messages,
"temperature": 0.3,
"max_tokens": 2000
}
)
if response.status_code != 200:
raise Exception(f"HolySheep API 오류: {response.status_code} - {response.text}")
return response.json()
def analyze_market_sentiment(self, news_text: str) -> dict:
"""
GPT-4.1을 사용한 시장 분위기 분석
복잡한 텍스트 이해에 최적화
"""
system_prompt = """당신은 암호화폐 시장 전문 애널리스트입니다.
뉴스 기사를 분석하여 다음 정보를 반환하세요:
- sentiment: bullish/bearish/neutral
- confidence: 0~1 사이 신뢰도
- key_factors: 주요影响因素 배열
- price_impact: 예상 가격 영향 (short/medium/long term)
"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"다음 암호화폐 뉴스를 분석하세요:\n\n{news_text}"}
]
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json={
"model": "gpt-4.1",
"messages": messages,
"temperature": 0.2,
"response_format": {"type": "json_object"}
}
)
if response.status_code != 200:
raise Exception(f"시장 분위기 분석 실패: {response.text}")
return response.json()
클라이언트 인스턴스 생성
ai_client = HolySheepAIClient(HOLYSHEEP_API_KEY)
DeepSeek를 사용한 차익거래 쌍 탐지
prompt = """
다음 암호화폐 쌍의 상관관계를 분석하고 통계 차익거래 적합성을 평가하세요:
쌍 1: BTC/USDT vs ETH/USDT (최근 30일 일별 데이터)
쌍 2: SOL/USDT vs MATIC/USDT (최근 30일 일별 데이터)
쌍 3: LINK/USDT vs DOT/USDT (최근 30일 일별 데이터)
각 쌍에 대해 다음을 분석하세요:
1. Pearson 상관계수
2. 공적분(Cointegration) 가능성
3. 평균 회귀 가능성 (Halflife 추정)
4. 차익거래 적합성 점수 (1-10)
Python 코드로 특성 추출 로직을 제공하세요.
"""
시계열 데이터 특성 추출
import pandas as pd
import numpy as np
from typing import Dict, List, Tuple
from scipy import stats
class CryptoFeatureExtractor:
"""암호화폐 시계열 데이터 특성 추출기"""
def __init__(self, ai_client: HolySheepAIClient):
self.ai_client = ai_client
self.features_cache = {}
def extract_basic_features(self, df: pd.DataFrame, price_col: str = 'close') -> pd.DataFrame:
"""
기본 재무 특성 추출
"""
df = df.copy()
# 이동평균
for window in [5, 10, 20, 50]:
df[f'SMA_{window}'] = df[price_col].rolling(window=window).mean()
df[f'EMA_{window}'] = df[price_col].ewm(span=window).mean()
# 변동성 지표
df['volatility_20'] = df[price_col].rolling(window=20).std()
df['volatility_50'] = df[price_col].rolling(window=50).std()
# 모멘텀
df['returns'] = df[price_col].pct_change()
df['momentum_10'] = df[price_col] / df[price_col].shift(10) - 1
df['momentum_20'] = df[price_col] / df[price_col].shift(20) - 1
# RSI 계산
delta = df[price_col].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / loss
df['RSI'] = 100 - (100 / (1 + rs))
# MACD
exp1 = df[price_col].ewm(span=12, adjust=False).mean()
exp2 = df[price_col].ewm(span=26, adjust=False).mean()
df['MACD'] = exp1 - exp2
df['MACD_signal'] = df['MACD'].ewm(span=9, adjust=False).mean()
df['MACD_hist'] = df['MACD'] - df['MACD_signal']
return df
def extract_arbitrage_features(self, df_pair: pd.DataFrame) -> Dict:
"""
차익거래 특화 특성 추출
"""
price_a = df_pair['price_a'].values
price_b = df_pair['price_b'].values
# 가격 비율 (Spread)
spread = price_a / price_b
spread_mean = np.mean(spread)
spread_std = np.std(spread)
z_score = (spread - spread_mean) / spread_std
# Halflife 계산 (Ornstein-Uhlenbeck process)
spread_lag = np.roll(spread, 1)
spread_lag[0] = spread_lag[1]
delta_spread = np.diff(spread)
# 회귀 분석으로 halflife 추정
try:
X = spread_lag[:-1].reshape(-1, 1)
y = delta_spread
from sklearn.linear_model import LinearRegression
model = LinearRegression()
model.fit(X, y)
theta = -model.coef_[0]
halflife = np.log(2) / theta if theta > 0 else np.inf
except:
halflife = np.inf
# ADF 검정 (정상성)
from statsmodels.tsa.stattools import adfuller
try:
adf_result = adfuller(spread, maxlag=1)
adf_statistic = adf_result[0]
adf_pvalue = adf_result[1]
is_stationary = adf_pvalue < 0.05
except:
adf_statistic = 0
adf_pvalue = 1
is_stationary = False
# Hurst 지수 (mean reversion 경향)
hurst = self._calculate_hurst_exponent(spread)
return {
'spread_mean': spread_mean,
'spread_std': spread_std,
'current_z_score': z_score[-1] if len(z_score) > 0 else 0,
'halflife': halflife,
'adf_statistic': adf_statistic,
'adf_pvalue': adf_pvalue,
'is_stationary': is_stationary,
'hurst_exponent': hurst,
'correlation': np.corrcoef(price_a, price_b)[0, 1]
}
def _calculate_hurst_exponent(self, series: np.ndarray, max_k: int = 100) -> float:
"""
Hurst 지수 계산
H < 0.5: 평균 회귀 경향
H = 0.5: 랜덤 워크
H > 0.5: 추세 지속 경향
"""
n = len(series)
if n < 100:
return 0.5
max_k = min(max_k, n // 2)
RS_list = []
for k in range(10, max_k, 10):
rs_values = []
for start in range(0, n - k, k):
subseries = series[start:start + k]
mean_subseries = np.mean(subseries)
cumulative_deviation = np.cumsum(subseries - mean_subseries)
R = np.max(cumulative_deviation) - np.min(cumulative_deviation)
S = np.std(subseries)
if S > 0:
rs_values.append(R / S)
if rs_values:
RS_list.append(np.mean(rs_values))
if len(RS_list) < 2:
return 0.5
# 로그-로그 회귀
log_k = np.log(np.arange(10, 10 + len(RS_list) * 10, 10))
log_RS = np.log(RS_list)
slope, _, _, _, _ = stats.linregress(log_k, log_RS)
return slope
def generate_ai_enhanced_features(self, market_data: Dict) -> Dict:
"""
HolySheep AI를 활용한 고급 특성 생성
DeepSeek V3.2 ($0.42/MTok) 활용
"""
prompt = f"""
암호화폐 시장 데이터를 분석하여 차익거래 전략에 사용할 수 있는
고급 특성을 제안하고 구현 코드를 생성하세요.
시장 데이터 요약:
- 변동성: {market_data.get('volatility', 'N/A')}
- 거래량 트렌드: {market_data.get('volume_trend', 'N/A')}
- 최근 급등락: {market_data.get('recent_moves', 'N/A')}
- 시장 뉴스: {market_data.get('news_summary', 'N/A')}
다음 형식으로 응답하세요:
1. 추천 특성 목록 (이름, 정의, 예상 효과)
2. 각 특성을 계산하는 Python 코드 스니펫
3. 특성 조합 추천
"""
try:
result = self.ai_client.analyze_with_deepseek(
prompt=prompt,
system_prompt="""당신은 암호화폐 퀀트 트레이딩 전문가입니다.
통계 차익거래 전략에 최적화된 특성 공학을 수행합니다."""
)
return {
'status': 'success',
'analysis': result['choices'][0]['message']['content'],
'model_used': 'deepseek-chat',
'cost': 0.42 # $/MTok
}
except Exception as e:
return {'status': 'error', 'message': str(e)}
사용 예제
def main():
# HolySheep AI 클라이언트 초기화
ai_client = HolySheepAIClient(HOLYSHEEP_API_KEY)
# 특성 추출기 초기화
extractor = CryptoFeatureExtractor(ai_client)
# 샘플 데이터 (실제 거래소 API에서 가져와야 함)
sample_data = {
'BTC/USDT': pd.DataFrame({
'timestamp': pd.date_range('2024-01-01', periods=100),
'close': np.random.randn(100).cumsum() + 50000
}),
'ETH/USDT': pd.DataFrame({
'timestamp': pd.date_range('2024-01-01', periods=100),
'close': np.random.randn(100).cumsum() + 3000
})
}
# 기본 특성 추출
btc_features = extractor.extract_basic_features(sample_data['BTC/USDT'])
eth_features = extractor.extract_basic_features(sample_data['ETH/USDT'])
# 차익거래 특성 추출
pair_data = pd.DataFrame({
'price_a': btc_features['close'],
'price_b': eth_features['close']
})
arb_features = extractor.extract_arbitrage_features(pair_data)
# AI Enhanced 특성
market_info = {
'volatility': '높음 (>5%)',
'volume_trend': '증가 추세',
'recent_moves': 'BTC 3% 상승, ETH 2.5% 상승',
'news_summary': '기관 투자자 유입 증가, 규제 완화 기대'
}
ai_features = extractor.generate_ai_enhanced_features(market_info)
print("차익거래 특성:")
for key, value in arb_features.items():
print(f" {key}: {value}")
print("\nAI 분석 결과:")
print(ai_features['analysis'][:500] + "...")
if __name__ == "__main__":
main()
실시간 차익거래 신호 생성 시스템
import asyncio
import websockets
import json
from datetime import datetime
from typing import Optional
class ArbitrageSignalGenerator:
"""실시간 차익거래 신호 생성기"""
def __init__(self, ai_client: HolySheepAIClient, config: dict):
self.ai_client = ai_client
self.config = config
self.entry_threshold = config.get('entry_zscore', 2.0)
self.exit_threshold = config.get('exit_zscore', 0.5)
self.position_size = config.get('position_size', 0.1)
self.active_positions = {}
async def monitor_pair(self, pair_name: str, websocket_url: str):
"""
암호화폐 쌍 실시간 모니터링
"""
async with websockets.connect(websocket_url) as ws:
await ws.send(json.dumps({
"method": "subscribe",
"params": [f"{pair_name}@trade", f"{pair_name}@bookTicker"]
}))
while True:
try:
message = await asyncio.recv()
data = json.loads(message)
if data.get('e') == 'bookTicker':
await self._process_book_ticker(data, pair_name)
elif data.get('e') == 'trade':
await self._process_trade(data, pair_name)
except Exception as e:
print(f"오류 발생: {e}")
await asyncio.sleep(1)
async def _process_book_ticker(self, data: dict, pair_name: str):
"""
호가창 데이터 처리 및 스프레드 계산
"""
symbol = data.get('s')
bid_price = float(data.get('b', 0))
ask_price = float(data.get('a', 0))
if symbol not in self.active_positions:
self.active_positions[symbol] = {
'bid': bid_price,
'ask': ask_price,
'spread': ask_price - bid_price,
'spread_pct': (ask_price - bid_price) / bid_price * 100
}
else:
# 실시간 스프레드 업데이트
self.active_positions[symbol].update({
'bid': bid_price,
'ask': ask_price,
'spread': ask_price - bid_price,
'spread_pct': (ask_price - bid_price) / bid_price * 100
})
async def _process_trade(self, data: dict, pair_name: str):
"""
체결 데이터 처리 및 신호 생성
"""
symbol = data.get('s')
price = float(data.get('p'))
quantity = float(data.get('q'))
trade_time = datetime.fromtimestamp(data.get('T', 0) / 1000)
# 10초 간격으로 신호 평가
if not hasattr(self, '_last_signal_check') or \
(datetime.now() - self._last_signal_check).total_seconds() > 10:
signal = await self._evaluate_arbitrage_signal(pair_name)
if signal:
await self._execute_signal(signal)
self._last_signal_check = datetime.now()
async def _evaluate_arbitrage_signal(self, pair_name: str) -> Optional[dict]:
"""
HolySheep AI를 활용한 차익거래 신호 평가
"""
# 현재 시장 상태 수집
market_state = self._collect_market_state(pair_name)
prompt = f"""
다음 암호화폐 차익거래 쌍의 신호를 평가하세요:
쌍명: {pair_name}
시장 상태:
- 현재 스프레드: {market_state['spread']:.4f}
- 스프레드 Z-점수: {market_state['z_score']:.2f}
- 평균 스프레드: {market_state['mean_spread']:.4f}
- 스프레드 표준편차: {market_state['std_spread']:.4f}
- 최근 거래량: {market_state['volume']}
- 호가 스프레드: {market_state['bid_ask_spread']:.2f}%
판단 기준:
- 진입 Z-점수 임계값: ±{self.entry_threshold}
- 청산 Z-점수 임계값: ±{self.exit_threshold}
다음 형식으로 응답하세요:
{{
"signal": "BUY_A_SELL_B" | "BUY_B_SELL_A" | "CLOSE" | "HOLD",
"confidence": 0.0~1.0,
"reason": "판단 근거",
"expected_profit_pct": number,
"risk_factors": ["위험 요소 배열"]
}}
"""
try:
# Gemini 2.5 Flash로 빠른 판단 ($2.50/MTok)
response = self.ai_client.analyze_with_gemini(
prompt=prompt,
system_prompt="당신은 고속 암호화폐 차익거래 트레이딩 봇입니다."
)
content = response['choices'][0]['message']['content']
# JSON 파싱
import re
json_match = re.search(r'\{[^{}]*\}', content, re.DOTALL)
if json_match:
signal_data = json.loads(json_match.group())
return signal_data
except Exception as e:
print(f"AI 신호 평가 실패: {e}")
return None
def _collect_market_state(self, pair_name: str) -> dict:
"""
시장 상태 데이터 수집
"""
# 실제 구현에서는 거래소 API 또는 DB에서 데이터 조회
return {
'spread': 0.0234,
'z_score': 1.8,
'mean_spread': 0.02,
'std_spread': 0.0015,
'volume': 1500000,
'bid_ask_spread': 0.05
}
async def _execute_signal(self, signal: dict):
"""
신호 실행 (시뮬레이션)
"""
if signal['signal'] == 'HOLD':
return
action = signal['signal']
confidence = signal['confidence']
if confidence < 0.7:
print(f"신뢰도 부족 ({confidence:.2f}), 신호 무시")
return
print(f"""
╔══════════════════════════════════════════════════════╗
║ 차익거래 신호 발생 ║
╠══════════════════════════════════════════════════════╣
║ 동작: {action}
║ 신뢰도: {confidence:.2%}
║ 예상 수익: {signal.get('expected_profit_pct', 0):.3f}%
║ 판단 근거: {signal['reason']}
║ 위험 요소: {', '.join(signal.get('risk_factors', []))}
╚══════════════════════════════════════════════════════╝
""")
async def main():
# HolySheep AI 클라이언트
ai_client = HolySheepAIClient(HOLYSHEEP_API_KEY)
# 신호 생성기 설정
config = {
'entry_zscore': 2.0,
'exit_zscore': 0.5,
'position_size': 0.1,
'max_pairs': 5
}
generator = ArbitrageSignalGenerator(ai_client, config)
# BTC-ETH 차익거래 모니터링 시작
await generator.monitor_pair(
"BTC-ETH",
"wss://stream.binance.com:9443/ws"
)
HolySheep AI Gemini 모델 호출 추가
def extend_holy_sheep_client():
"""기존 HolySheepAIClient에 Gemini 메서드 추가"""
def analyze_with_gemini(self, prompt: str, system_prompt: str = None) -> dict:
"""Gemini 2.5 Flash를 사용한 고속 분석"""
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": prompt})
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json={
"model": "gemini-2.5-flash",
"messages": messages,
"temperature": 0.1,
"max_tokens": 1000
}
)
if response.status_code != 200:
raise Exception(f"Gemini API 오류: {response.text}")
return response.json()
HolySheepAIClient.analyze_with_gemini = analyze_with_gemini
extend_holy_sheep_client()
자주 발생하는 오류 해결
1. API 인증 오류: "Invalid API Key"
# 오류 증상
{"error": {"message": "Invalid API Key", "type": "invalid_request_error"}}
해결 방법
1. API 키 형식 확인
print(f"API 키 길이: {len(HOLYSHEEP_API_KEY)}")
print(f"API 키 시작: {HOLYSHEEP_API_KEY[:10]}...")
2. 환경 변수에서 올바르게 로드되었는지 확인
import os
api_key = os.environ.get('HOLYSHEEP_API_KEY', '')
if not api_key:
raise ValueError("HOLYSHEEP_API_KEY 환경 변수가 설정되지 않았습니다.")
3. base_url이 정확한지 확인 (공백 제거)
base_url = "https://api.holysheep.ai/v1".strip()
if not base_url.endswith('/v1'):
base_url = base_url.rstrip('/') + '/v1'
4. 헤더 형식 확인
headers = {
"Authorization": f"Bearer {api_key}", # Bearer 키워드 필수
"Content-Type": "application/json"
}
5. 테스트 요청
test_response = requests.get(
f"{base_url}/models",
headers={"Authorization": f"Bearer {api_key}"}
)
print(f"연결 테스트 결과: {test_response.status_code}")
if test_response.status_code == 401:
print("API 키가 유효하지 않습니다. HolySheep 대시보드에서 확인하세요.")
2. Rate Limit 초과 오류
# 오류 증상
{"error": {"message": "Rate limit exceeded", "type": "rate_limit_error"}}
해결 방법
import time
from functools import wraps
class RateLimitedClient:
"""Rate Limit 관리를 위한 래퍼 클래스"""
def __init__(self, original_client, max_requests_per_minute=60):
self.client = original_client
self.max_rpm = max_requests_per_minute
self.request_times = []
def _check_rate_limit(self):
"""최근 1분 내 요청 수 확인"""
current_time = time.time()
self.request_times = [
t for t in self.request_times
if current_time - t < 60
]
if len(self.request_times) >= self.max_rpm:
sleep_time = 60 - (current_time - self.request_times[0])
print(f"Rate limit 도달. {sleep_time:.1f}초 대기...")
time.sleep(sleep_time)
def analyze_with_deepseek(self, prompt: str, system_prompt: str = None) -> dict:
self._check_rate_limit()
self.request_times.append(time.time())
return self.client.analyze_with_deepseek(prompt, system_prompt)
def analyze_with_gemini(self, prompt: str, system_prompt: str = None) -> dict:
self._check_rate_limit()
self.request_times.append(time.time())
return self.client.analyze_with_gemini(prompt, system_prompt)
사용
limited_client = RateLimitedClient(ai_client, max_requests_per_minute=30)
재시도 로직과 함께 사용
def retry_with_backoff(func, max_retries=3, initial_delay=1):
"""지수 백오프와 함께 재시도"""
for attempt in range(max_retries):
try:
return func()
except Exception as e:
if "rate limit" in str(e).lower():
delay = initial_delay * (2 ** attempt)
print(f"재시도 {attempt + 1}/{max_retries}, {delay}초 후 재시도...")
time.sleep(delay)
else:
raise
raise Exception(f"최대 재시도 횟수({max_retries}) 초과")
3. 토큰 제한 초과 및 응답 잘림
# 오류 증상
{"error": {"message": "This model's maximum context length is 8192 tokens"}}
해결 방법
def chunk_large_data(data: str, max_tokens: int = 3000) -> list:
"""대규모 데이터를 청크로 분할"""
# 토큰 추정 (한국어의 경우 문자 수 기반)
# 영어는 대략 4글자 = 1토큰, 한국어는 2글자 = 1토큰 추정
chars_per_token = 2
chunks = []
current_chunk = []
current_length = 0
for line in data.split('\n'):
line_length = len(line) // chars_per_token
if current_length + line_length > max_tokens:
if current_chunk:
chunks.append('\n'.join(current_chunk))
current_chunk = [line]
current_length = line_length
else:
current_chunk.append(line)
current_length += line_length
if current_chunk:
chunks.append('\n'.join(current_chunk))
return chunks
def summarize_and_combine(client, chunks: list, combine_prompt: str) -> dict:
"""각 청크 요약 후 통합"""
summaries = []
for i, chunk in enumerate(chunks):
prompt = f"""
다음 데이터를 간결하게 요약하세요 (500토큰 이내):
데이터 {i+1}/{len(chunks)}:
{chunk}
"""
result = client.analyze_with_deepseek(
prompt=prompt,
system_prompt="한국어로 간결하게 요약"
)
summaries.append(result['choices'][0]['message']['content'])
# 통합 분석
combined_prompt = f"""
다음 요약들을 통합하여 최종 분석을 제공하세요:
{combine_prompt}
부분 요약:
{chr(10).join([f'{i+1}. {s}' for i, s in enumerate(summaries)])}
"""
return client.analyze_with_deepseek(
prompt=