시작하기 전에: 왜 거래 데이터 정제가 중요한가
저는 3개월 전 암호화폐 트레이딩 봇 개발 프로젝트를 진행하면서 중요한发现问题를 했습니다. Tardis에서 받은 Kraken BTC/USD 데이터가，看似 정돈된 CSV처럼 보였지만, 실제로 파싱해보면timestamp 불일치, 거래량 음수값, 중복 레코드 등 치명적인 문제들이 있었죠.
결국 2주간 삽질한末에 안정적인 데이터 파이프라인을 구축했습니다. 이 튜토리얼에서는 제가 실제로 사용한 코드와 같은 실전 경험을 공유하겠습니다.
Tardis와 Kraken 데이터 구조 이해
Tardis 메시지 포맷 개요
Tardis는 암호화폐 거래소 실시간 데이터를 제공하는 API 서비스입니다. Kraken 현물 거래소의 데이터는 다음과 같은 포맷으로 전송됩니다:
- trade: 개별 거래 실행 내역
- book: 오더북 업데이트
- ticker: 실시간 시세 정보
CSV 변환의 필요성
실시간 스트리밍 대신 CSV로 저장하면:
- 과거 데이터 백테스팅 용이
- 머신러닝 모델 학습 데이터셋 구축
- 여러 거래소 데이터 통합 분석
- 비용 절감 (실시간 스트리밍은 비쌈)
환경 설정
# 필요한 패키지 설치
pip install pandas numpy python-dateutil
프로젝트 디렉토리 구조
project/
├── raw_data/ # Tardis에서 다운로드한 원본 CSV
├── cleaned_data/ # 정제된 CSV 저장
├── scripts/ # 파싱 스크립트
└── config.py # 설정 파일Tardis CSV 데이터 파싱实战
1단계: 기본 CSV 파싱
import pandas as pd
from datetime import datetime, timezone
import numpy as np
def parse_tardis_trades(csv_path: str) -> pd.DataFrame:
"""
Tardis에서 다운로드한 Kraken 거래 데이터 CSV 파싱
"""
# Tardis CSV는 쉼표 구분자 사용
df = pd.read_csv(csv_path)
# 기본 컬럼 확인
print("원본 컬럼:", df.columns.tolist())
print("데이터 건수:", len(df))
return df
실행 예제
raw_df = parse_tardis_trades("raw_data/kraken_btcusd_trades_2024.csv")
print(raw_df.head())Tardis에서 다운로드한 CSV의典型적인 컬럼 구조는 다음과 같습니다:
timestamp: 거래 발생 시각 (ISO 8601)side: buy 또는 sellprice: 거래 가격amount: 거래 수량trade_id: 고유 거래 ID
2단계: 데이터 타입 변환과 정제
def clean_trade_data(df: pd.DataFrame) -> pd.DataFrame:
"""
Kraken 거래 데이터 정제 로직
"""
df_clean = df.copy()
# 1. Timestamp 변환 (UTC 기준)
df_clean['timestamp'] = pd.to_datetime(
df_clean['timestamp'],
utc=True
).dt.tz_convert('UTC')
# 2. 숫자형 컬럼 타입 변환
numeric_cols = ['price', 'amount']
for col in numeric_cols:
df_clean[col] = pd.to_numeric(df_clean[col], errors='coerce')
# 3. 거래량 음수값 제거 (정상 거래는 양수)
invalid_trades = df_clean[df_clean['amount'] <= 0]
print(f"잘못된 거래량 레코드: {len(invalid_trades)}건")
df_clean = df_clean[df_clean['amount'] > 0]
# 4. 가격 범위 검증 (비정상적 가격 제거)
# BTC/USD의 합리적 범위: $1,000 ~ $500,000
df_clean = df_clean[
(df_clean['price'] >= 1000) &
(df_clean['price'] <= 500000)
]
# 5. 중복 레코드 제거
before_count = len(df_clean)
df_clean = df_clean.drop_duplicates(subset=['trade_id'], keep='first')
removed = before_count - len(df_clean)
print(f"중복 제거: {removed}건")
# 6. 거래 방향 정규화
df_clean['side'] = df_clean['side'].str.lower().map({'buy': 'buy', 'sell': 'sell'})
return df_clean
정제 실행
cleaned_df = clean_trade_data(raw_df)
print(f"\n정제 후 데이터 건수: {len(cleaned_df)}")
print(cleaned_df.dtypes)3단계: 고급 정제 - 이상치 탐지와 처리
def detect_and_handle_outliers(df: pd.DataFrame, window: int = 100) -> pd.DataFrame:
"""
이동 평균 기반 이상치 탐지 및 처리
"""
df_out = df.copy().sort_values('timestamp').reset_index(drop=True)
# 이동 평균과 표준편차 계산
df_out['ma_price'] = df_out['price'].rolling(window=window, center=True).mean()
df_out['std_price'] = df_out['price'].rolling(window=window, center=True).std()
# Z-score 기반 이상치 탐지 (3 sigma)
df_out['z_score'] = np.abs(
(df_out['price'] - df_out['ma_price']) / df_out['std_price']
)
# 이상치 마스킹 (is_outlier 플래그)
df_out['is_outlier'] = df_out['z_score'] > 3
outliers_count = df_out['is_outlier'].sum()
print(f"탐지된 이상치: {outliers_count}건 ({outliers_count/len(df_out)*100:.2f}%)")
# 이상치를 NaN으로 대체하고 보간
df_out.loc[df_out['is_outlier'], 'price'] = np.nan
df_out['price'] = df_out['price'].interpolate(method='linear')
# 임시 컬럼 제거
df_out = df_out.drop(columns=['ma_price', 'std_price', 'z_score', 'is_outlier'])
return df_out
이상치 처리 실행
final_df = detect_and_handle_outliers(cleaned_df)
정제 결과 저장
final_df.to_csv(
'cleaned_data/kraken_btcusd_cleaned.csv',
index=False,
compression='gzip'
)
print("정제 완료! 저장 경로: cleaned_data/kraken_btcusd_cleaned.csv")실전 활용: AI 기반 거래 분석 파이프라인
정제된 데이터를 AI 모델과 연동하면 자동화된 거래 분석 시스템을 구축할 수 있습니다. HolySheep AI를 사용하면 해외 신용카드 없이도 GPT-4.1, Claude, Gemini 등 주요 모델을 단일 API 키로 통합할 수 있습니다.
import open