AI API 사용량을 정확하게 추적하고 비용을 분석하는 것은 모든 개발팀에게 필수적인 과제입니다. 이번 튜토리얼에서는 HolySheep AI의 API를 활용하여 소비 명세서를 프로그래밍 방식으로 내보내는 방법을 단계별로 설명드리겠습니다. 대량 데이터 처리, 동시성 제어, 비용 최적화까지 프로덕션 환경에서 바로 적용 가능한 실전 코드를 제공합니다.
1. 소비 명세서 내보내기의 중요성과 아키텍처
AI API 비용은 사용량에 따라 급격히 변동합니다. 저는 이전 프로젝트에서 월간 API 비용이 3배 이상 불어나는 경험을 했고, 이때 소비 명세서 분석이 비용 절감의 첫걸음임을 깨달았습니다. HolySheep AI는 가입 시 무료 크레딧을 제공하며, 모든 주요 모델(GPT-4.1, Claude Sonnet, Gemini 2.5 Flash, DeepSeek V3.2)을 단일 API 키로 관리할 수 있어 명세서 통합이 간편합니다.
아키텍처 개요
소비 명세서 내보내기 시스템은 다음과 같은 구성 요소로 이루어집니다:
- API 클라이언트: HolySheep AI REST API 연동
- 데이터 수집기: 페이지네이션 기반 대량 데이터 패치
- 보고서 생성기: CSV/JSON/Excel 형태 내보내기
- 예약 실행기: Crontab 기반 자동화
2. REST API 기반 소비 명세서 조회
HolySheep AI의 소비 명세서 API는 필터링과 페이지네이션을 지원합니다. 먼저 API 엔드포인트 구조를 확인하고 기본적인 조회 방법을 구현하겠습니다.
import requests
import json
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import time
class HolySheepBillingExporter:
"""
HolySheep AI 소비 명세서 내보내기 클라이언트
HolySheep AI API 문서: https://docs.holysheep.ai
base_url: https://api.holysheep.ai/v1
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
def get_usage_summary(
self,
start_date: str,
end_date: str,
model: Optional[str] = None
) -> Dict:
"""
指定 기간의 사용량 요약 조회
Args:
start_date: 시작일 (YYYY-MM-DD)
end_date: 종료일 (YYYY-MM-DD)
model: 필터링할 모델명 (선택)
Returns:
사용량 요약 딕셔너리
"""
endpoint = f"{self.BASE_URL}/billing/usage"
params = {
"start_date": start_date,
"end_date": end_date
}
if model:
params["model"] = model
response = self.session.get(endpoint, params=params)
response.raise_for_status()
return response.json()
def get_cost_breakdown(
self,
start_date: str,
end_date: str,
granularity: str = "daily"
) -> List[Dict]:
"""
비용 상세 내역 조회 (모델별, 일별 구분)
HolySheep AI 가격 정책:
- GPT-4.1: $8/MTok
- Claude Sonnet 4.5: $15/MTok
- Gemini 2.5 Flash: $2.50/MTok
- DeepSeek V3.2: $0.42/MTok
"""
endpoint = f"{self.BASE_URL}/billing/costs"
params = {
"start_date": start_date,
"end_date": end_date,
"granularity": granularity # daily, monthly
}
response = self.session.get(endpoint, params=params)
response.raise_for_status()
return response.json().get("costs", [])
def get_detailed_logs(
self,
start_date: str,
end_date: str,
limit: int = 100,
after: Optional[str] = None
) -> Dict:
"""
세부 사용 로그 조회 (페이지네이션 지원)
대량 데이터 처리 시 필수
"""
endpoint = f"{self.BASE_URL}/billing/logs"
params = {
"start_date": start_date,
"end_date": end_date,
"limit": min(limit, 1000) # 최대 1000건
}
if after:
params["after"] = after
response = self.session.get(endpoint, params=params)
response.raise_for_status()
return response.json()
사용 예제
if __name__ == "__main__":
exporter = HolySheepBillingExporter("YOUR_HOLYSHEEP_API_KEY")
# 최근 7일치 사용량 요약 조회
end_date = datetime.now().strftime("%Y-%m-%d")
start_date = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d")
summary = exporter.get_usage_summary(start_date, end_date)
print(f"총 사용량: ${summary.get('total_cost', 0):.4f}")
print(f"총 토큰: {summary.get('total_tokens', 0):,}")
3. 대량 데이터 자동 수집 구현
실제 프로덕션 환경에서는 수만 건 이상의 로그를 처리해야 합니다. Cursor 기반 페이지네이션과 동시성 제어를 활용한 대량 데이터 수집기를 구현하겠습니다.
import asyncio
import aiohttp
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from typing import List, Dict, Generator
import csv
from pathlib import Path
@dataclass
class UsageRecord:
"""개별 사용 기록"""
timestamp: str
model: str
input_tokens: int
output_tokens: int
cost: float
request_id: str
class BulkUsageCollector:
"""
대량 사용 데이터 수집기
최적화 포인트:
- 비동기 병렬 처리로 수집 속도 향상
- Rate limiting 적용으로 API 제한 방지
- 배치 처리로 메모리 효율성 확보
"""
def __init__(
self,
api_key: str,
max_concurrent: int = 5,
requests_per_second: float = 10.0
):
self.api_key = api_key
self.max_concurrent = max_concurrent
self.requests_per_second = requests_per_second
self._rate_limiter = asyncio.Semaphore(max_concurrent)
async def fetch_page(
self,
session: aiohttp.ClientSession,
cursor: Optional[str],
start_date: str,
end_date: str
) -> Dict:
"""단일 페이지 데이터 조회"""
async with self._rate_limiter:
url = "https://api.holysheep.ai/v1/billing/logs"
params = {
"start_date": start_date,
"end_date": end_date,
"limit": 1000
}
if cursor:
params["after"] = cursor
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
async with session.get(url, params=params, headers=headers) as resp:
resp.raise_for_status()
data = await resp.json()
# Rate limiting 딜레이 적용
await asyncio.sleep(1.0 / self.requests_per_second)
return data
async def collect_all_usage(
self,
start_date: str,
end_date: str
) -> List[UsageRecord]:
"""전체 사용 기록 수집 (자동 페이지네이션)"""
all_records = []
cursor = None
connector = aiohttp.TCPConnector(limit=self.max_concurrent)
timeout = aiohttp.ClientTimeout(total=300)
async with aiohttp.ClientSession(
connector=connector,
timeout=timeout
) as session:
page_count = 0
while True:
page_count += 1
print(f"페이지 {page_count} 수집 중...")
data = await self.fetch_page(session, cursor, start_date, end_date)
logs = data.get("data", [])
if not logs:
break
for log in logs:
record = UsageRecord(
timestamp=log.get("created_at"),
model=log.get("model"),
input_tokens=log.get("usage", {}).get("prompt_tokens", 0),
output_tokens=log.get("usage", {}).get("completion_tokens", 0),
cost=log.get("cost", 0.0),
request_id=log.get("id", "")
)
all_records.append(record)
# 다음 페이지 커서
cursor = data.get("has_more", False) and data.get("next_cursor")
if not cursor:
break
print(f"총 {len(all_records):,}건 수집 완료")
return all_records
def export_to_csv(self, records: List[UsageRecord], filepath: str):
"""CSV 파일로 내보내기"""
fieldnames = ["timestamp", "model", "input_tokens", "output_tokens", "cost", "request_id"]
with open(filepath, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
for record in records:
writer.writerow({
"timestamp": record.timestamp,
"model": record.model,
"input_tokens": record.input_tokens,
"output_tokens": record.output_tokens,
"cost": f"{record.cost:.6f}",
"request_id": record.request_id
})
print(f"CSV 내보내기 완료: {filepath}")
async def main():
"""실행 예제"""
collector = BulkUsageCollector(
api_key="YOUR_HOLYSHEEP_API_KEY",
max_concurrent=5,
requests_per_second=10.0
)
end_date = datetime.now().strftime("%Y-%m-%d")
start_date = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d")
records = await collector.collect_all_usage(start_date, end_date)
# CSV 내보내기
output_path = f"usage_report_{start_date}_{end_date}.csv"
collector.export_to_csv(records, output_path)
# 통계 출력
total_cost = sum(r.cost for r in records)
model_costs = {}
for r in records:
model_costs[r.model] = model_costs.get(r.model, 0) + r.cost
print(f"\n=== 비용 분석 ===")
print(f"총 비용: ${total_cost:.4f}")
print(f"\n모델별 비용:")
for model, cost in sorted(model_costs.items(), key=lambda x: -x[1]):
print(f" {model}: ${cost:.4f} ({cost/total_cost*100:.1f}%)")
if __name__ == "__main__":
asyncio.run(main())
4. 비용 최적화 분석 및 리포트 생성
수집한 데이터를 기반으로 모델별 비용 분석, 트렌드 파악, 최적화 제안을 제공하는 분석 모듈을 구현하겠습니다. HolySheep AI의 경쟁력 있는 가격표를 활용하면 비용을 상당히 절감할 수 있습니다.
import pandas as pd
from datetime import datetime, timedelta
from collections import defaultdict
from typing import Dict, List
import json
HolySheep AI 모델 가격표 ($/1M 토큰)
MODEL_PRICING = {
"gpt-4.1": {"input": 8.00, "output": 8.00},
"gpt-4.1-turbo": {"input": 2.00, "output": 8.00},
"claude-sonnet-4.5": {"input": 15.00, "output": 15.00},
"claude-sonnet-4": {"input": 10.00, "output": 10.00},
"gemini-2.5-flash": {"input": 2.50, "output": 2.50},
"gemini-2.0-flash": {"input": 0.40, "output": 1.60},
"deepseek-v3.2": {"input": 0.42, "output": 0.42},
"deepseek-chat": {"input": 0.27, "output": 1.10},
}
class CostAnalyzer:
"""
AI API 비용 분석기
제공 기능:
- 모델별 비용 분석
- 일별/주별 트렌드
- 비용 최적화 추천
- 예상 비용 시뮬레이션
"""
def __init__(self, records: List[UsageRecord]):
self.records = records
self.df = self._build_dataframe()
def _build_dataframe(self) -> pd.DataFrame:
"""DataFrame 변환"""
data = [{
"timestamp": r.timestamp,
"model": r.model,
"input_tokens": r.input_tokens,
"output_tokens": r.output_tokens,
"total_tokens": r.input_tokens + r.output_tokens,
"cost": r.cost
} for r in self.records]
df = pd.DataFrame(data)
df["timestamp"] = pd.to_datetime(df["timestamp"])
df["date"] = df["timestamp"].dt.date
return df
def model_cost_breakdown(self) -> pd.DataFrame:
"""모델별 비용 분석"""
breakdown = self.df.groupby("model").agg({
"input_tokens": "sum",
"output_tokens": "sum",
"total_tokens": "sum",
"cost": "sum"
}).round(6)
breakdown["cost_share"] = (
breakdown["cost"] / breakdown["cost"].sum() * 100
).round(2)
breakdown["avg_cost_per_1m_tokens"] = (
breakdown["cost"] / breakdown["total_tokens"] * 1_000_000
).round(4)
return breakdown.sort_values("cost", ascending=False)
def daily_trend(self) -> pd.DataFrame:
"""일별 사용량 트렌드"""
daily = self.df.groupby("date").agg({
"total_tokens": "sum",
"cost": "sum"
}).round(4)
daily["avg_cost_per_token"] = (
daily["cost"] / daily["total_tokens"] * 1_000_000
).round(4)
return daily
def find_optimization_opportunities(self) -> List[Dict]:
"""
비용 최적화 기회 탐지
HolySheep AI 가격표 기반:
- DeepSeek V3.2: $0.42/MTok (최저가)
- Gemini 2.0 Flash: $0.40/MTok (입력만)
"""
recommendations = []
# 고비용 모델 사용량 분석
high_cost_models = ["gpt-4.1", "claude-sonnet-4.5"]
for model in high_cost_models:
model_data = self.df[self.df["model"] == model]
if not model_data.empty:
cost = model_data["cost"].sum()
tokens = model_data["total_tokens"].sum()
# 대안 모델 제안
if "gpt-4.1" in model:
alternative = "gpt-4.1-turbo"
alternative_cost = tokens * 2.0 / 1_000_000 # $2/M 입력
elif "claude-sonnet-4.5" in model:
alternative = "claude-sonnet-4"
alternative_cost = tokens * 10.0 / 1_000_000
else:
alternative = "gemini-2.5-flash"
alternative_cost = tokens * 2.5 / 1_000_000
potential_savings = cost - alternative_cost
recommendations.append({
"model": model,
"current_cost": cost,
"alternative": alternative,
"potential_savings": potential_savings,
"savings_percent": (potential_savings / cost * 100) if cost > 0 else 0,
"tokens": tokens,
"suggestion": f"{model} → {alternative} 전환 검토"
})
return sorted(recommendations, key=lambda x: -x["savings_percent"])
def generate_report(self) -> Dict:
"""종합 보고서 생성"""
model_breakdown = self.model_cost_breakdown()
daily_trend = self.daily_trend()
optimizations = self.find_optimization_opportunities()
report = {
"summary": {
"total_cost": round(self.df["cost"].sum(), 6),
"total_tokens": int(self.df["total_tokens"].sum()),
"total_requests": len(self.df),
"period_start": str(self.df["date"].min()),
"period_end": str(self.df["date"].max()),
"avg_cost_per_request": round(
self.df["cost"].sum() / len(self.df), 6
) if len(self.df) > 0 else 0
},
"model_breakdown": model_breakdown.to_dict("index"),
"daily_trend": daily_trend.to_dict("index"),
"optimizations": optimizations,
"potential_total_savings": round(
sum(o["potential_savings"] for o in optimizations), 4
)
}
return report
사용 예제
if __name__ == "__main__":
# CSV 파일에서 데이터 로드
df = pd.read_csv("usage_report_2025-01-01_2025-01-31.csv")
records = [
UsageRecord(
timestamp=row["timestamp"],
model=row["model"],
input_tokens=int(row["input_tokens"]),
output_tokens=int(row["output_tokens"]),
cost=float(row["cost"]),
request_id=row["request_id"]
)
for _, row in df.iterrows()
]
analyzer = CostAnalyzer(records)
report = analyzer.generate_report()
print("=== AI API 비용 분석 보고서 ===")
print(f"총 비용: ${report['summary']['total_cost']:.4f}")
print(f"총 토큰: {report['summary']['total_tokens']:,}")
print(f"총 요청: {report['summary']['total_requests']:,}")
print("\n=== 모델별 비용 ===")
for model, data in report["model_breakdown"].items():
print(f"{model}: ${data['cost']:.4f} ({data['cost_share']:.1f}%)")
print("\n=== 최적화 추천 ===")
for opt in report["optimizations"]:
print(f" {opt['suggestion']}")
print(f" 현재 비용: ${opt['current_cost']:.4f}")
print(f" 절감 가능: ${opt['potential_savings']:.4f} ({opt['savings_percent']:.1f}%)")
print(f"\n총 절감 가능 금액: ${report['potential_total_savings']:.4f}")
5. 자동화 스케줄러 설정
매일/매주 자동으로 소비 명세서를 수집하고 이메일까지 발송하는 스케줄러를 구현하겠습니다. Linux cron 또는 Windows Task Scheduler와 연동할 수 있습니다.
#!/usr/bin/env python3
"""
AI API 소비 명세서 자동 수집 및 보고서 발송
Crontab 설정: 0 6 * * * /usr/bin/python3 /opt/scripts/daily_billing_report.py
"""
import sys
import logging
from datetime import datetime, timedelta
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders
import smtplib
import json
로깅 설정
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
handlers=[
logging.FileHandler("/var/log/billing_exporter.log"),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
class AutomatedBillingReporter:
"""
자동화된 소비 명세서 리포터
기능:
- 일별/주별 자동 수집
- 이메일 보고서 발송
- Slack/Discord 웹훅 연동
- S3/GCS 백업
"""
def __init__(self, api_key: str, config: Dict):
self.api_key = api_key
self.config = config
self.exporter = HolySheepBillingExporter(api_key)
self.collector = BulkUsageCollector(
api_key,
max_concurrent=5,
requests_per_second=10.0
)
def run_daily_report(self):
"""일일 보고서 실행"""
logger.info("일일 보고서 수집 시작")
end_date = datetime.now().strftime("%Y-%m-%d")
start_date = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
try:
# 1. 데이터 수집
records = asyncio.run(
self.collector.collect_all_usage(start_date, end_date)
)
# 2. CSV 내보내기
output_path = f"/data/billing/{start_date}_usage.csv"
self.collector.export_to_csv(records, output_path)
# 3. 분석 실행
analyzer = CostAnalyzer(records)
report = analyzer.generate_report()
# 4. 보고서 저장
report_path = f"/data/billing/{start_date}_report.json"
with open(report_path, "w") as f:
json.dump(report, f, indent=2, default=str)
# 5. 알림 발송
self._send_email_alert(report, start_date)
logger.info(f"일일 보고서 완료: {start_date}")
logger.info(f"총 비용: ${report['summary']['total_cost']:.4f}")
except Exception as e:
logger.error(f"보고서 생성 실패: {e}")
self._send_error_alert(str(e))
raise
def run_monthly_backup(self):
"""월별 백업 실행"""
logger.info("월별 백업 시작")
end_date = datetime.now().strftime("%Y-%m-%d")
start_date = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d")
records = asyncio.run(
self.collector.collect_all_usage(start_date, end_date)
)
# 월별 CSV
backup_path = f"/data/billing/monthly_{start_date}_{end_date}.csv"
self.collector.export_to_csv(records, backup_path)
logger.info(f"월별 백업 완료: {backup_path}")
def _send_email_alert(self, report: Dict, date: str):
"""이메일 알림 발송"""
smtp_config = self.config.get("smtp", {})
if not smtp_config:
logger.warning("SMTP 설정 없음 - 이메일 발송 건너뜀")
return
msg = MIMEMultipart("alternative")
msg["Subject"] = f"[HolySheep AI] 일일 비용 보고서 - {date}"
msg["From"] = smtp_config["from"]
msg["To"] = smtp_config["to"]
# 텍스트 본문
text_body = self._build_text_report(report, date)
msg.attach(MIMEText(text_body, "plain"))
# HTML 본문
html_body = self._build_html_report(report, date)
msg.attach(MIMEText(html_body, "html"))
# SMTP 발송
with smtplib.SMTP(smtp_config["host"], smtp_config["port"]) as server:
if smtp_config.get("use_tls"):
server.starttls()
server.login(smtp_config["username"], smtp_config["password"])
server.send_message(msg)
logger.info(f"이메일 발송 완료: {smtp_config['to']}")
def _build_text_report(self, report: Dict, date: str) -> str:
"""텍스트 보고서 생성"""
lines = [
f"=== HolySheep AI 일일 비용 보고서: {date} ===",
"",
f"총 비용: ${report['summary']['total_cost']:.4f}",
f"총 토큰: {report['summary']['total_tokens']:,}",
f"총 요청: {report['summary']['total_requests']:,}",
"",
"모델별 비용:"
]
for model, data in report["model_breakdown"].items():
lines.append(
f" {model}: ${data['cost']:.4f} ({data['cost_share']:.1f}%)"
)
if report["optimizations"]:
lines.append("")
lines.append("비용 최적화 기회:")
for opt in report["optimizations"][:3]:
lines.append(f" - {opt['suggestion']}")
return "\n".join(lines)
def _build_html_report(self, report: Dict, date: str) -> str:
"""HTML 보고서 생성"""
model_rows = ""
for model, data in report["model_breakdown"].items():
model_rows += f"""
{model}
{data['total_tokens']:,}
${data['cost']:.4f}
{data['cost_share']:.1f}%
"""
return f"""
HolySheep AI 비용 보고서
HolySheep AI 비용 보고서
{date}
요약
총 비용: ${report['summary']['total_cost']:.4f}
총 토큰: {report['summary']['total_tokens']:,}
총 요청: {report['summary']['total_requests']:,}
모델별 비용
모델
토큰
비용
비율
{model_rows}
이 보고서는 HolySheep AI에서 자동 생성되었습니다.
"""
def _send_error_alert(self, error_message: str):
"""오류 알림 발송"""
logger.error(f"오류 발생: {error_message}")
# 오류 처리 로직 구현
Crontab 설정 가이드
CRON_CONFIG = """
매일早上 6시에 일일 보고서 실행
0 6 * * * /usr/bin/python3 /opt/scripts/daily_billing_report.py >> /var/log/cron_billing.log 2>&1
매월 1일 새벽 2시에 월별 백업 실행
0 2 1 * * /usr/bin/python3 /opt/scripts/monthly_backup.py >> /var/log/cron_backup.log 2>&1
"""
if __name__ == "__main__":
import os
api_key = os.environ.get("HOLYSHEEP_API_KEY")
if not api_key:
logger.error("HOLYSHEEP_API_KEY 환경 변수가 설정되지 않았습니다")
sys.exit(1)
config = {
"smtp": {
"host": "smtp.gmail.com",
"port": 587,
"use_tls": True,
"username": "[email protected]",
"password": "your-app-password",
"from": "[email protected]",
"to": "[email protected]"
}
}
reporter = AutomatedBillingReporter(api_key, config)
reporter.run_daily_report()
자주 발생하는 오류와 해결책
1. API 키 인증 실패 (401 Unauthorized)
# 오류 메시지
requests.exceptions.HTTPError: 401 Client Error: Unauthorized
원인: 잘못된 API 키 또는 환경 변수 미설정
해결 방법
import os
올바른 API 키 설정 확인
API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
API 키 유효성 검증
def validate_api_key(api_key: str) -> bool:
"""API 키 유효성 검증"""
if not api_key or len(api_key) < 20:
return False
response = requests.get(
"https://api.holysheep.ai/v1/models",
headers={"Authorization": f"Bearer {api_key}"}
)
return response.status_code == 200
사용
if not validate_api_key(API_KEY):
raise ValueError("유효하지 않은 HolySheep API 키입니다. https://www.holysheep.ai/register 에서 키를 확인하세요.")
2. Rate Limit 초과 (429 Too Many Requests)
# 오류 메시지
requests.exceptions.HTTPError: 429 Client Error: Too Many Requests
원인: API 요청 빈도가 제한을 초과
해결 방법: 지수 백오프와 재시도 로직 적용
import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class RateLimitResilientSession(requests.Session):
"""Rate Limit을 처리하는 세션 클래스"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Retry 전략 설정
retry_strategy = Retry(
total=5,
backoff_factor=2, # 2, 4, 8, 16, 32초 대기
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET", "POST"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
self.mount("https://", adapter)
self.mount("http://", adapter)
def request(self, method, url, **kwargs):
"""요청 전 Rate Limit 헤더 확인"""
response = super().request(method, url, **kwargs)
# Rate Limit 관련 헤더 처리
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
print(f"Rate Limit 도달. {retry_after}초 후 재시도...")
time.sleep(retry_after)
return super().request(method, url, **kwargs)
return response
사용 예제
session = RateLimitResilientSession()
session.headers.update({"Authorization": f"Bearer {API_KEY}"})
response = session.get("https://api.holysheep.ai/v1/billing/usage")
3. 대량 데이터 수집 시 메모리 부족 (MemoryError)
# 오류 메시지
MemoryError: Unable to allocate array
원인: 수백만 건의 레코드를 한 번에 메모리에 로드
해결 방법: 제너레이터 기반 스트리밍 처리
from typing import Iterator
import gc
class StreamingBillingCollector:
"""메모리 효율적인 스트리밍 수집기"""
def __init__(self, api_key: str):
self.api_key = api_key
self.exporter = HolySheepBillingExporter(api_key)
def stream_usage_records(
self,
start_date: str,
end_date: str,
batch_size: int = 1000
) -> Iterator[UsageRecord]:
"""
제너레이터로 레코드 스트리밍
메모리 사용량: O(batch_size) - 상수
"""
cursor = None
total_fetched = 0
while True:
# 배치 단위로 데이터 조회
data = self.exporter.get_detailed_logs(
start_date=start_date,
end_date=end_date,
limit=batch_size,
after=cursor
)
logs = data.get("data", [])
if not logs:
break
# 각 레코드를 제너레이터로 방출
for log in logs:
yield UsageRecord(
timestamp=log.get("created_at"),
model=log.get("model"),
input_tokens=log.get("usage", {}).get("prompt_tokens", 0),
output_tokens=log.get("usage", {}).get("completion_tokens", 0),
cost=log.get("cost", 0.0),
request_id=log.get("id", "")
)
total_fetched += 1
# 메모리 해제
del logs
gc.collect()
# 커서 업데이트
cursor = data.get("next_cursor")
if not cursor:
break
print(f"수집 완료: {total_fetched:,}건")
print(f"총 {total_fetched:,}건 스트리밍 완료")
사용 예제: CSV 파일로 직접 스트리밍
def export_streaming_to_csv(
start_date: str,
end_date: str,
output_path: str
):
"""스트리밍 방식으로 CSV 내보내기 (메모리 절약)"""
collector = StreamingBillingCollector(API_KEY)
with