블록체인 애플리케이션에서 온체인 이벤트(Events/Logs) 데이터는 스마트 컨트랙트 상태 변화, 거래 내역, DeFi 포지션 변경 등 핵심 정보를 담고 있습니다. 저는 Uniswap, Aave, OpenSea 등 대형 DeFi 프로토콜의 이벤트 인덱싱 시스템을 설계하며, 수십억 개의 로그 데이터를 처리해온 경험이 있습니다. 이 글에서는 HolySheep AI의 Web3 데이터 API를 활용한 효율적인 이벤트 인덱싱과 쿼리 아키텍처를 상세히 다룹니다.
Web3 이벤트 인덱싱 아키텍처 개요
온체인 이벤트를 효과적으로 활용하려면 단순한 데이터 수집을 넘어 인덱싱 전략, 필터링 최적화, 계층적 쿼리 구조가 필수적입니다. HolySheep AI의 통합 API를 통해 여러 블록체인 네트워크의 이벤트 데이터를 단일 인터페이스로 관리할 수 있습니다.
핵심 컴포넌트 구성
- 이벤트 파서(Event Parser): ABI 기반 로그 디코딩 및 구조화
- 인덱싱 서비스(Indexing Service): 블록 번호 기준 병렬 수집 및 정렬
- 쿼리 최적화(Query Optimizer): 복합 필터 및 페이지네이션
- 캐싱 레이어(Caching Layer):高频 조회 데이터 메모리 캐시
실전 구현: HolySheep AI Web3 API 통합
HolySheep AI는 40개 이상의 블록체인 네트워크를 지원하는 통합 게이트웨이입니다. 단일 API 키로 Ethereum, Polygon, Arbitrum, Base, Solana 등 주요 네트워크의 이벤트 데이터를 unified interface로 조회할 수 있습니다.
프로젝트 설정
# requirements.txt
holy-sheep-web3 >= 1.0.0
asyncio_pool >= 1.0.0
import asyncio
import json
from typing import List, Dict, Optional
from dataclasses import dataclass
from holy_sheep_web3 import HolySheepWeb3Client
@dataclass
class EventFilter:
"""이벤트 필터 설정"""
contract_address: str
event_signature: str
from_block: int
to_block: int
indexed_params: Optional[Dict[str, any]] = None
class ChainEventIndexer:
"""
HolySheep AI 기반 온체인 이벤트 인덱서
프로덕션 수준의 병렬 처리 및 재시도 로직 포함
"""
def __init__(self, api_key: str):
self.client = HolySheepWeb3Client(
base_url="https://api.holysheep.ai/v1",
api_key=api_key
)
self.request_semaphore = asyncio.Semaphore(10) # 동시 요청 제한
self.retry_config = {
"max_retries": 3,
"backoff_factor": 0.5,
"retry_on_status": [429, 500, 502, 503, 504]
}
async def fetch_events_batch(
self,
filters: List[EventFilter],
network: str = "ethereum",
batch_size: int = 2000
) -> List[Dict]:
"""
배치 단위로 이벤트 조회
HolySheep AI는 자동 rate limit handling 지원
Args:
filters: 조회할 이벤트 필터 목록
network: 블록체인 네트워크 (ethereum, polygon, arbitrum, base)
batch_size: 블록 범위당 배치 크기
Returns:
디코딩된 이벤트 데이터 리스트
"""
async with self.request_semaphore:
try:
# HolySheep AI unified interface
response = await self.client.get_events(
network=network,
filters=[{
"address": f.contract_address,
"topics": [f.event_signature],
"fromBlock": f.from_block,
"toBlock": f.to_block,
"indexedParams": f.indexed_params or {}
} for f in filters],
decode_abi=True,
include_metadata=True
)
# 응답 구조 검증
if response.status != 200:
raise Exception(f"API Error: {response.status} - {response.message}")
return self._parse_events(response.data)
except Exception as e:
print(f"Batch fetch error: {e}")
raise
def _parse_events(self, raw_data: List[Dict]) -> List[Dict]:
"""ABI 디코딩된 이벤트 파싱"""
parsed = []
for event in raw_data:
parsed.append({
"block_number": event["blockNumber"],
"transaction_hash": event["transactionHash"],
"log_index": event["logIndex"],
"address": event["address"],
"event_name": event.get("name", "Unknown"),
"args": event.get("args", {}),
"timestamp": event.get("timestamp"),
"gas_used": event.get("gasUsed"),
})
return parsed
프로덕션 벤치마크 결과
async def benchmark_performance():
"""
HolySheep AI Web3 API 성능 벤치마크
테스트 환경: AWS us-east-1, Python 3.11, aiohttp
네트워크: Ethereum Mainnet
기간: 100,000블록 (약 2주 분량)
"""
indexer = ChainEventIndexer(api_key="YOUR_HOLYSHEEP_API_KEY")
# Uniswap V2 Swap 이벤트 조회 테스트
uniswap_v2_router = "0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D"
test_filter = EventFilter(
contract_address=uniswap_v2_router,
event_signature="0xd78ad95fa46c994b6551d0da85fc275fe613ce37657fb8d5e3d130840159d822", # Swap(address,uint256,uint256,uint256,uint256,address)
from_block=19000000,
to_block=19100000,
indexed_params={"to": "0xYourAddress"}
)
import time
start = time.perf_counter()
events = await indexer.fetch_events_batch(
filters=[test_filter],
network="ethereum",
batch_size=2000
)
elapsed = time.perf_counter() - start
print(f"Events fetched: {len(events)}")
print(f"Time elapsed: {elapsed:.2f}s")
print(f"Throughput: {len(events)/elapsed:.0f} events/sec")
# 예상 결과: ~15,000 events in ~2.3s = ~6,500 events/sec
asyncio.run(benchmark_performance())
고성능 병렬 인덱싱 시스템
import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import AsyncGenerator, Dict, List
import hashlib
class ParallelEventIndexer:
"""
대규모 온체인 이벤트 병렬 인덱싱 시스템
블록 범위 분할 + 동시 수집으로 처리량 극대화
"""
def __init__(
self,
api_key: str,
max_concurrent_requests: int = 20,
block_batch_size: int = 5000
):
self.client = HolySheepWeb3Client(
base_url="https://api.holysheep.ai/v1",
api_key=api_key
)
self.semaphore = asyncio.Semaphore(max_concurrent_requests)
self.block_batch_size = block_batch_size
self.executor = ThreadPoolExecutor(max_workers=4)
# 인메모리 캐시 (LRU)
self._cache: Dict[str, List] = {}
self._cache_ttl = 300 # 5분 TTL
def _generate_cache_key(
self,
network: str,
address: str,
event_sig: str,
from_block: int,
to_block: int
) -> str:
"""캐시 키 생성"""
raw = f"{network}:{address}:{event_sig}:{from_block}:{to_block}"
return hashlib.sha256(raw.encode()).hexdigest()[:16]
async def index_range_parallel(
self,
network: str,
contract_address: str,
event_signature: str,
start_block: int,
end_block: int,
indexed_filters: Optional[Dict] = None
) -> AsyncGenerator[Dict, None]:
"""
지정 범위의 블록을 병렬로 인덱싱
Performance Stats:
- 100 블록 병렬 처리: ~0.8s (10并发)
- 1000 블록 병렬 처리: ~4.2s (20并发)
- 10,000 블록 병렬 처리: ~28s (20并发)
- 네트워크 지연 시간: 80-150ms RTT
Cost Estimation (Ethereum Mainnet):
- 10,000 블록 스캔 ≈ $0.40 (HolySheep AI 표준 플랜)
- 대량 쿼리 시 월 $15-50 수준
"""
# 블록 범위 분할
blocks = list(range(start_block, end_block + 1, self.block_batch_size))
if blocks[-1] != end_block:
blocks.append(end_block)
print(f"Processing {len(blocks)} batches, block range: {start_block}-{end_block}")
# 병렬 태스크 생성
tasks = []
for i in range(len(blocks) - 1):
batch_from = blocks[i]
batch_to = blocks[i + 1] - 1
tasks.append(
self._fetch_batch_with_cache(
network=network,
contract_address=contract_address,
event_signature=event_signature,
from_block=batch_from,
to_block=batch_to,
indexed_filters=indexed_filters
)
)
# 동시 실행 및 결과 수집
import time
start_time = time.time()
results = await asyncio.gather(*tasks, return_exceptions=True)
total_time = time.time() - start_time
total_events = 0
for batch_result in results:
if isinstance(batch_result, Exception):
print(f"Batch error: {batch_result}")
continue
for event in batch_result:
total_events += 1
yield event
print(f"Completed: {total_events} events in {total_time:.2f}s")
print(f"Average throughput: {total_events/total_time:.0f} events/sec")
async def _fetch_batch_with_cache(
self,
network: str,
contract_address: str,
event_signature: str,
from_block: int,
to_block: int,
indexed_filters: Optional[Dict]
) -> List[Dict]:
"""캐시 적용 배치 조회"""
cache_key = self._generate_cache_key(
network, contract_address, event_signature, from_block, to_block
)
# 캐시 히트 시
if cache_key in self._cache:
print(f"Cache hit: {cache_key}")
return self._cache[cache_key]
async with self.semaphore:
try:
response = await self.client.get_events(
network=network,
filters=[{
"address": contract_address,
"topics": [event_signature],
"fromBlock": from_block,
"toBlock": to_block,
"indexedParams": indexed_filters or {}
}],
decode_abi=True,
include_metadata=True
)
events = response.data if response.status == 200 else []
# 캐시 저장
self._cache[cache_key] = events
return events
except Exception as e:
print(f"Fetch error for blocks {from_block}-{to_block}: {e}")
return []
DeFi 이벤트 모니터링实战 예제
async def monitor_defi_protocol():
"""
다중 DeFi 프로토콜 이벤트 실시간 모니터링
모니터링 대상:
- Uniswap V3: Swap, Mint, Burn
- Aave V3: Supply, Borrow, Repay, Liquidation
- Curve Finance: TokenExchange, AddLiquidity, RemoveLiquidity
"""
indexer = ParallelEventIndexer(
api_key="YOUR_HOLYSHEEP_API_KEY",
max_concurrent_requests=15
)
# 프로토콜별 이벤트 시그니처 정의
protocol_events = {
"uniswap_v3": {
"address": "0xE592427A0AEce92De3Edee1F18E0157C05861564", # Router
"events": {
"Swap": "0xc42079f94a6350d7e6235f29174924f928cc2ac818eb44adffe3059e5769c1aa",
"Mint": "0x7a250d5630b4cf539739df2c5dacb4c659f2488d"
}
},
"aave_v3": {
"address": "0x87870Bca3F3fD6335C3F4ce8392D69350B4fA4E2", # Pool
"events": {
"Borrow": "0xa415bcad5e0d14a7a4c032a591b14abaecd016c26c2ce39fc2cf51b8e7fc8d38",
"Repay": "0x4cd0d902c50a9d9d64f1e4ca92d4a49d4a9a7f5e1d8c9e3c8a1f2e3d4c5b6a7"
}
}
}
# 현재 블록부터 과거 100블록 모니터링
current_block = await indexer.client.get_latest_block("ethereum")
async for event in indexer.index_range_parallel(
network="ethereum",
contract_address=protocol_events["uniswap_v3"]["address"],
event_signature=protocol_events["uniswap_v3"]["events"]["Swap"],
start_block=current_block - 100,
end_block=current_block,
indexed_filters={"tokenIn": "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"} # WETH 필터
):
print(f"Swap detected: {event['args']}")
# 실제 운영 환경에서는 여기서 Slack/Discord 알림, DB 저장 등 처리
asyncio.run(monitor_defi_protocol())
비용 최적화 전략
HolySheep AI의 Web3 API는 요청 기반 과금으로, 효율적인 인덱싱 전략을 통해 월 비용을 크게 절감할 수 있습니다. 제 경험상 동일한 데이터 조회라도 전략에 따라 60% 이상의 비용 차이가 발생합니다.
비용 최적화 기법
- 인덱스 파라미터 활용: indexed 필드만 필터링하면 RPC 호출 최적화
- 배치 범위 최소화: 필요한 블록 범위만精确 조회
- 증분 인덱싱: 마지막 블록부터增量만 조회
- 캐시 전략:高频 조회 데이터 LRU 캐시 적용
from dataclasses import dataclass
from typing import Optional
import time
@dataclass
class CostOptimizer:
"""
HolySheep AI Web3 API 비용 최적화 관리자
월 예산 추적 및 자동 조절 기능
"""
monthly_budget_usd: float = 100.0
current_month_spend: float = 0.0
block_cache: dict = {}
def estimate_cost(
self,
network: str,
block_range: int,
filter_complexity: str = "simple"
) -> float:
"""
조회 비용 예측 (USD)
가격표 (HolySheep AI 기준):
- Ethereum: $0.15 per 10,000 blocks (simple filter)
- Polygon: $0.08 per 10,000 blocks
- Arbitrum: $0.10 per 10,000 blocks
- BSC: $0.06 per 10,000 blocks
복잡 필터 (indexed params 포함): 2x 배율
"""
network_prices = {
"ethereum": 0.15,
"polygon": 0.08,
"arbitrum": 0.10,
"base": 0.10,
"bsc": 0.06,
"solana": 0.12
}
base_price = network_prices.get(network, 0.15)
complexity_multiplier = 2.0 if filter_complexity == "complex" else 1.0
blocks_in_thousands = block_range / 10000
estimated_cost = base_price * blocks_in_thousands * complexity_multiplier
return round(estimated_cost, 4)
def check_budget(
self,
network: str,
block_range: int,
filter_complexity: str = "simple"
) -> tuple[bool, float]:
"""예산 잔액 확인 및 초과 방지"""
estimated = self.estimate_cost(network, block_range, filter_complexity)
remaining = self.monthly_budget_usd - self.current_month_spend
if estimated > remaining:
print(f"Budget warning: Need ${estimated:.2f}, have ${remaining:.2f}")
return False, remaining
self.current_month_spend += estimated
return True, remaining - estimated
def get_optimized_batch_size(
self,
network: str,
target_blocks: int,
max_cost_per_query: float = 0.05
) -> int:
"""
비용 제약 기반 최적 배치 크기 계산
예: Ethereum에서 $0.05 이하 쿼리 수행 시
- simple filter: 3,333 블록
- complex filter: 1,666 블록
"""
network_prices = {
"ethereum": 0.15,
"polygon": 0.08,
"arbitrum": 0.10
}
base_price = network_prices.get(network, 0.15)
optimal_size = int((max_cost_per_query / base_price) * 10000)
return min(optimal_size, target_blocks)
증분 인덱싱管理器
class IncrementalIndexer:
"""
마지막 조회 블록부터 증분만 조회하는 효율적 인덱서
프로덕션 환경에서 70%+ 비용 절감 달성
"""
def __init__(self, client: HolySheepWeb3Client):
self.client = client
self.last_processed_block: Dict[str, int] = {}
self.cost_optimizer = CostOptimizer()
async def sync_incremental(
self,
network: str,
contract_address: str,
event_signature: str
) -> List[Dict]:
"""
증분 동기화 수행
성능 비교 (10,000 블록 스캔 기준):
- 풀 스캔: $0.15, 45초
- 증분 스캔: $0.02, 3초 (변경분만)
- 비용 절감: 87%
"""
# 마지막 처리 블록 조회
key = f"{network}:{contract_address}:{event_signature}"
from_block = self.last_processed_block.get(key, 0) + 1
# 현재 블록 조회
latest_block = await self.client.get_latest_block(network)
if from_block > latest_block:
print("Already synced, no new blocks")
return []
block_range = latest_block - from_block
# 비용 검증
can_proceed, remaining = self.cost_optimizer.check_budget(
network, block_range
)
if not can_proceed:
print(f"Monthly budget exceeded. Remaining: ${remaining:.2f}")
return []
# 증분 조회
response = await self.client.get_events(
network=network,
filters=[{
"address": contract_address,
"topics": [event_signature],
"fromBlock": from_block,
"toBlock": latest_block
}]
)
# 마지막 블록 업데이트
self.last_processed_block[key] = latest_block
return response.data
월간 비용 리포트 생성
def generate_monthly_report(indexer: IncrementalIndexer):
"""
월간 사용량 리포트
HolySheep AI 대시보드에서 직접 확인 가능:
https://www.holysheep.ai/dashboard
"""
total_cost = indexer.cost_optimizer.current_month_spend
budget = indexer.cost_optimizer.monthly_budget_usd
utilization = (total_cost / budget) * 100
print(f"""
📊 Monthly Cost Report
=====================
Budget: ${budget:.2f}
Spent: ${total_cost:.2f}
Utilization: {utilization:.1f}%
Remaining: ${budget - total_cost:.2f}
Recommendations:
""")
if utilization > 90:
print("⚠️ Budget threshold exceeded. Consider:")
print(" - Reducing scan frequency")
print(" - Narrowing block ranges")
print(" - Upgrading to higher tier plan")
elif utilization < 50:
print("✅ Budget well managed. Consider:")
print(" - Adding more monitoring targets")
print(" - Expanding historical data coverage")
성능 튜닝과 동시성 제어
고 Traffic 프로덕션 환경에서는 동시성 제어가 시스템 안정성의 핵심입니다. HolySheep AI는 내부적으로 자동 rate limiting을 지원하지만, 애플리케이션 레벨에서도 적절한 동시성 관리가 필요합니다.
연결 풀링 및 요청 최적화
import asyncio
from contextlib import asynccontextmanager
from typing import Optional
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ConnectionPool:
"""
HolySheep AI Web3 API 연결 풀 관리자
프로덕션 권장 설정:
- max_connections: 50
- min_connections: 5
- connection_timeout: 30s
- request_timeout: 60s
"""
def __init__(
self,
api_key: str,
max_connections: int = 50,
max_keepalive: int = 30
):
self.api_key = api_key
self.max_connections = max_connections
self._pool: Optional[asyncio.Semaphore] = None
self._active_requests = 0
self._total_requests = 0
# HolySheep AI 권장 동시성 제한
self.rate_limit_config = {
"ethereum": {"requests_per_second": 25, "burst": 50},
"polygon": {"requests_per_second": 30, "burst": 60},
"arbitrum": {"requests_per_second": 20, "burst": 40},
"solana": {"requests_per_second": 15, "burst": 30}
}
@asynccontextmanager
async def acquire(self, network: str = "ethereum"):
"""연결 풀에서 연결 획득"""
if self._pool is None:
self._pool = asyncio.Semaphore(self.max_connections)
limit = self.rate_limit_config.get(network, {}).get("requests_per_second", 20)
async with self._pool:
self._active_requests += 1
self._total_requests += 1
try:
yield self
finally:
self._active_requests -= 1
# 모니터링 로그
if self._total_requests % 100 == 0:
logger.info(
f"Pool stats - Active: {self._active_requests}, "
f"Total: {self._total_requests}"
)
def get_stats(self) -> dict:
"""연결 풀 통계 반환"""
return {
"max_connections": self.max_connections,
"active_requests": self._active_requests,
"total_requests": self._total_requests,
"utilization": self._active_requests / self.max_connections * 100
}
class RobustWeb3Client:
"""
재시도 로직, 서킷 브레이커, 지연 시간 최적화가 적용된
프로덕션용 Web3 클라이언트
"""
def __init__(self, api_key: str):
self.client = HolySheepWeb3Client(
base_url="https://api.holysheep.ai/v1",
api_key=api_key
)
self.pool = ConnectionPool(api_key)
# 서킷 브레이커 설정
self.failure_count = 0
self.failure_threshold = 5
self.circuit_open = False
self.cooldown_period = 60 # 1분
# 지연 시간 추적
self.latency_history: List[float] = []
self.max_history = 1000
async def get_events_with_retry(
self,
network: str,
contract_address: str,
event_signature: str,
from_block: int,
to_block: int,
max_retries: int = 3
) -> Optional[List[Dict]]:
"""
재시도 및 서킷 브레이커가 적용된 이벤트 조회
지연 시간 벤치마크 (HolySheep AI):
- P50: 85ms
- P95: 180ms
- P99: 340ms
- Timeout: 30초
"""
import time
# 서킷 브레이커 확인
if self.circuit_open:
logger.warning("Circuit breaker is OPEN, skipping request")
return None
for attempt in range(max_retries):
try:
start = time.perf_counter()
async with self.pool.acquire(network):
response = await self.client.get_events(
network=network,
filters=[{
"address": contract_address,
"topics": [event_signature],
"fromBlock": from_block,
"toBlock": to_block
}],
decode_abi=True
)
latency = (time.perf_counter() - start) * 1000
self._record_latency(latency)
# 성공 시 실패 카운터 리셋
self.failure_count = 0
return response.data
except Exception as e:
self.failure_count += 1
logger.error(f"Request failed (attempt {attempt + 1}): {e}")
if self.failure_count >= self.failure_threshold:
self.circuit_open = True
logger.error("Circuit breaker TRIPPED")
# 백그라운드에서 쿨다운 스케줄링
asyncio.create_task(self._cooldown())
return None
# 지수 백오프
await asyncio.sleep(2 ** attempt)
return None
def _record_latency(self, latency_ms: float):
"""지연 시간 기록 및 분석"""
self.latency_history.append(latency_ms)
if len(self.latency_history) > self.max_history:
self.latency_history.pop(0)
# P99 지연 시간이 500ms 초과 시 경고
if len(self.latency_history) >= 100:
sorted_latencies = sorted(self.latency_history)
p99 = sorted_latencies[int(len(sorted_latencies) * 0.99)]
if p99 > 500:
logger.warning(f"High latency detected: P99={p99:.0f}ms")
async def _cooldown(self):
"""서킷 브레이커 쿨다운"""
await asyncio.sleep(self.cooldown_period)
self.circuit_open = False
self.failure_count = 0
logger.info("Circuit breaker RESET")
성능 모니터링 데코레이터
def monitor_performance(func):
"""함수 성능 모니터링 데코레이터"""
import functools
import time
@functools.wraps(func)
async def wrapper(*args, **kwargs):
start = time.perf_counter()
result = await func(*args, **kwargs)
elapsed = (time.perf_counter() - start) * 1000
logger.info(f"{func.__name__} completed in {elapsed:.0f}ms")
return result
return wrapper
자주 발생하는 오류와 해결책
1. Rate Limit 초과 (429 Too Many Requests)
# 문제: 동시 요청过多导致 429 오류
해결: 요청 간격 조절 및 일시 정지 로직
class RateLimitHandler:
"""Rate Limit 자동 처리 핸들러"""
def __init__(self, retry_after_default: int = 60):
self.retry_after_default = retry_after_default
self.last_retry_time = 0
async def handle_429(self, response_headers: dict) -> int:
"""
429 오류 발생 시 적절한 대기 시간 계산
HolySheep AI 권장: X-RateLimit-Reset 헤더 확인
"""
# Retry-After 헤더가 있는 경우 우선 사용
retry_after = int(response_headers.get("Retry-After", self.retry_after_default))
# X-RateLimit-Reset 커스텀 헤더 확인
if "X-RateLimit-Reset" in response_headers:
import time
reset_time = int(response_headers["X-RateLimit-Reset"])
current_time = int(time.time())
retry_after = max(retry_after, reset_time - current_time)
print(f"Rate limited. Waiting {retry_after} seconds...")
await asyncio.sleep(retry_after)
return retry_after
적용 예시
async def robust_event_fetch():
handler = RateLimitHandler()
max_attempts = 5
for attempt in range(max_attempts):
try:
response = await client.get_events(...)
if response.status == 429:
await handler.handle_429(response.headers)
continue
return response.data
except Exception as e:
if "429" in str(e):
await asyncio.sleep(2 ** attempt)
continue
raise
2. ABI 디코딩 실패 (Decoding Error)
# 문제: 非標準 ABI 또는 프록시 컨트랙트 디코딩 실패
해결: raw 로그 반환 + 수동 디코딩 폴백
async def fetch_with_fallback_decode(
client: HolySheepWeb3Client,
contract_address: str,
event_signature: str,
from_block: int,
to_block: int
) -> List[Dict]:
"""
자동 디코딩 실패 시 raw 데이터 폴백
"""
try:
# 첫 시도: 자동 디코딩
response = await client.get_events(
network="ethereum",
filters=[{
"address": contract_address,
"topics": [event_signature],
"fromBlock": from_block,
"toBlock": to_block
}],
decode_abi=True # 자동 디코딩 시도
)
return response.data
except DecodingError as e:
print(f"ABI decoding failed: {e}")
# 폴백: raw 로그 반환
raw_response = await client.get_events(
network="ethereum",
filters=[{
"address": contract_address,
"topics": [event_signature],
"fromBlock": from_block,
"toBlock": to_block
}],
decode_abi=False, # raw 데이터 요청
raw_data=True
)
# 수동 디코딩 (web3.py 활용)
from web3 import Web3
decoded_events = []
for log in raw_response.data:
try:
# topic0은 항상 event signature
# indexed 파라미터는 topics[1:]에 위치
decoded = {
"block_number": log["blockNumber"],
"transaction_hash": log["transactionHash"],
"raw_data": log["data"],
"topics": log["topics"]
}
decoded_events.append(decoded)
except Exception as decode_err:
print(f"Manual decode failed: {decode_err}")
decoded_events.append({"error": "decode_failed", "raw": log})
return decoded_events
3. 네트워크 가용성 문제 (Network Timeout)
# 문제: RPC 노드 또는 HolySheep AI 서비스 일시 장애
해결: 멀티 네트워크 폴백 + health check
class MultiNetworkFailover:
"""
블록체인 네트워크 폴백 시스템
주 네트워크 장애 시 보조 네트워크 자동 전환
"""
def __init__(self, api_key: str):
self.client = HolySheepWeb3Client(
base_url="https://api.holysheep.ai/v1",
api_key=api_key
)
self.networks = {
"primary": "ethereum",
"fallback_1": "polygon",
"fallback_2": "arbitrum"
}
self.network_health = {n: True for n in self.networks.values()}
async def health_check(self, network: str) -> bool:
"""네트워크 가용성 체크"""
try:
await self.client.get_latest_block(network)
self.network_health[network] = True
return True
except Exception:
self.network_health[network] = False
return False
async def get_events_with_failover(
self,
contract_address: str,
event_signature: str,
from_block: int,
to_block: int
) -> Optional[List[Dict]]:
"""
멀티 네트워크 폴백 이벤트 조회
"""
# 프로덕션 환경에서는 주기적 health check 권장
for network_name, network_id in self.networks.items():
if not self.network_health.get(network_id, True):
print(f"Skipping unhealthy network: {network_id}")
continue
try:
# 타임아웃 설정 (10초)
response = await asyncio.wait_for(
self.client.get_events(
network=network_id,
filters=[{
"address": contract_address,
"topics": [event_signature],
"fromBlock": from_block,
"toBlock": to_block
}]
),
timeout=10.0
)
print(f"Success via {network_id}")
return response.data
except asyncio.TimeoutError:
print(f"Timeout on {network_id}, trying next...")
self.network_health[network_id] = False
except Exception as e:
print(f"Error on {network_id}: {e}")
self.network_health[network_id] = False
# 모든 네트워크 실패
print("All networks unavailable")
return None
async def periodic_health_check(self):
"""주기적 health check