백테스트 데이터 전처리가 중요한 이유
백테스트 데이터 전처리는 자동매매 전략의 성패를 좌우하는 핵심 단계입니다. 아무리 정교한 전략이라도 데이터에 결측값, 이상치, 중복, 시간대 불일치가 있으면 백테스트 결과를 신뢰할 수 없습니다. “Garbage In, Garbage Out”은 퀀트 트레이딩에서도 예외 없이 적용됩니다.
실전에서 흔히 발생하는 문제는 다음과 같습니다:
- 결측 캔들: 거래소 점검, 네트워크 장애로 빠진 봉
- 이상 가격: 플래시 크래시, API 오류로 인한 극단값
- 중복 데이터: 동일 타임스탬프에 여러 레코드
- 시간대 불일치: UTC, KST 혼재로 인한 타이밍 오류
- 생존자 편향: 상장폐지된 코인 데이터 누락
OHLCV 데이터 수집과 정규화
먼저 거래소 API에서 수집한 원시 데이터를 표준 형태로 변환합니다.
import pandas as pd
import numpy as np
from datetime import datetime, timezone
def load_and_normalize(filepath: str, timeframe: str = '1h') -> pd.DataFrame:
"""OHLCV CSV 로드 및 정규화
Args:
filepath: CSV 파일 경로
timeframe: 캔들 주기 ('1m', '5m', '1h', '1d')
Returns:
정규화된 DataFrame
"""
df = pd.read_csv(filepath)
# 컬럼명 표준화
column_map = {
'timestamp': 'datetime', 'time': 'datetime', 'date': 'datetime',
'Open': 'open', 'High': 'high', 'Low': 'low',
'Close': 'close', 'Volume': 'volume',
}
df.rename(columns=column_map, inplace=True)
# 타임스탬프 처리 (밀리초/초 자동 감지)
if df['datetime'].dtype in ['int64', 'float64']:
ts = df['datetime'].iloc[0]
if ts > 1e12: # 밀리초
df['datetime'] = pd.to_datetime(df['datetime'], unit='ms', utc=True)
else: # 초
df['datetime'] = pd.to_datetime(df['datetime'], unit='s', utc=True)
else:
df['datetime'] = pd.to_datetime(df['datetime'], utc=True)
# 타입 강제 변환
for col in ['open', 'high', 'low', 'close', 'volume']:
df[col] = pd.to_numeric(df[col], errors='coerce')
df.set_index('datetime', inplace=True)
df.sort_index(inplace=True)
return df
# 사용 예시
# df = load_and_normalize('btc_usdt_1h.csv', '1h')결측 데이터 탐지와 보간
거래소 점검이나 API 장애로 빠진 캔들을 찾아 적절히 처리해야 합니다. 결측 비율에 따라 보간 방식이 달라집니다.
| 결측 유형 | 처리 방법 | 적합한 경우 |
|---|---|---|
| 단일 캔들 누락 | 전후 값 보간 (forward fill) | 1~2개 연속 누락 |
| 연속 구간 누락 | 다른 거래소 데이터 대체 | 3개 이상 연속 |
| 장기간 누락 | 해당 구간 제외 | 수시간 이상 |
def detect_missing_candles(df: pd.DataFrame, timeframe: str = '1h') -> pd.DataFrame:
"""결측 캔들 탐지 및 보고"""
freq_map = {'1m': 'T', '5m': '5T', '15m': '15T', '1h': 'H', '4h': '4H', '1d': 'D'}
freq = freq_map.get(timeframe, 'H')
# 완전한 시간 인덱스 생성
full_index = pd.date_range(
start=df.index.min(),
end=df.index.max(),
freq=freq,
tz='UTC'
)
missing = full_index.difference(df.index)
if len(missing) > 0:
print(f"⚠️ 결측 캔들: {len(missing)}개 / 전체 {len(full_index)}개")
print(f" 결측 비율: {len(missing)/len(full_index)*100:.2f}%")
# 연속 결측 구간 찾기
gaps = []
gap_start = missing[0]
prev = missing[0]
for ts in missing[1:]:
if (ts - prev) > pd.Timedelta(freq):
gaps.append((gap_start, prev, (prev - gap_start).total_seconds() / 3600))
gap_start = ts
prev = ts
gaps.append((gap_start, prev, (prev - gap_start).total_seconds() / 3600))
print(f" 연속 결측 구간: {len(gaps)}개")
for start, end, hours in sorted(gaps, key=lambda x: -x[2])[:5]:
print(f" {start} ~ {end} ({hours:.1f}시간)")
else:
print("✅ 결측 캔들 없음")
return missing
def fill_missing_candles(df: pd.DataFrame, timeframe: str = '1h',
max_gap: int = 3) -> pd.DataFrame:
"""결측 캔들 보간 처리
Args:
max_gap: forward fill 최대 허용 갭 수
"""
freq_map = {'1m': 'T', '5m': '5T', '15m': '15T', '1h': 'H', '4h': '4H', '1d': 'D'}
freq = freq_map.get(timeframe, 'H')
# 리인덱스
full_index = pd.date_range(
start=df.index.min(), end=df.index.max(),
freq=freq, tz='UTC'
)
df = df.reindex(full_index)
# OHLC: forward fill (max_gap 제한)
for col in ['open', 'high', 'low', 'close']:
df[col] = df[col].fillna(method='ffill', limit=max_gap)
# Volume: 결측은 0으로 (거래 없음)
df['volume'] = df['volume'].fillna(0)
# max_gap 초과 구간은 NaN 유지 → 나중에 제외
remaining_na = df['close'].isna().sum()
if remaining_na > 0:
print(f"⚠️ 보간 불가 구간: {remaining_na}개 (max_gap={max_gap} 초과)")
return df이상치 탐지와 처리
플래시 크래시, API 오류, 유동성 부족으로 발생한 이상 가격은 백테스트 결과를 왜곡합니다. 통계적 방법으로 이상치를 탐지하고 처리합니다.
def detect_outliers(df: pd.DataFrame, method: str = 'zscore',
threshold: float = 4.0) -> pd.Series:
"""이상 가격 탐지
Args:
method: 'zscore' 또는 'iqr'
threshold: Z-score 기준값 또는 IQR 배수
Returns:
이상치 불리언 마스크
"""
returns = df['close'].pct_change().dropna()
if method == 'zscore':
# 롤링 Z-score (적응적)
rolling_mean = returns.rolling(100, min_periods=20).mean()
rolling_std = returns.rolling(100, min_periods=20).std()
zscore = (returns - rolling_mean) / (rolling_std + 1e-10)
outliers = zscore.abs() > threshold
elif method == 'iqr':
q1 = returns.rolling(100, min_periods=20).quantile(0.25)
q3 = returns.rolling(100, min_periods=20).quantile(0.75)
iqr = q3 - q1
outliers = (returns < q1 - threshold * iqr) | (returns > q3 + threshold * iqr)
print(f"이상치 탐지: {outliers.sum()}개 / {len(returns)}개")
return outliers
def handle_outliers(df: pd.DataFrame, outlier_mask: pd.Series,
method: str = 'clip') -> pd.DataFrame:
"""이상치 처리
Args:
method: 'clip' (윈저화), 'nan' (제거), 'median' (중앙값 대체)
"""
df = df.copy()
returns = df['close'].pct_change()
if method == 'clip':
# 상하위 0.5% 윈저화
lower = returns.quantile(0.005)
upper = returns.quantile(0.995)
clipped = returns.clip(lower, upper)
# 가격 재구성
df['close'] = df['close'].iloc[0] * (1 + clipped).cumprod()
df['close'] = df['close'].fillna(method='ffill')
elif method == 'nan':
df.loc[outlier_mask.index[outlier_mask], ['open','high','low','close']] = np.nan
df = df.fillna(method='ffill')
return dfOHLCV 무결성 검증
가격 데이터의 논리적 일관성을 검증하는 것도 중요합니다. High가 Low보다 낮거나, Close가 High/Low 범위 밖에 있는 경우를 잡아냅니다.
def validate_ohlcv(df: pd.DataFrame) -> dict:
"""OHLCV 무결성 검증"""
issues = {}
# 1. High >= Low 검증
bad_hl = df[df['high'] < df['low']]
if len(bad_hl) > 0:
issues['high_lt_low'] = len(bad_hl)
print(f"❌ High < Low: {len(bad_hl)}건")
# 2. Close가 High/Low 범위 내인지
bad_close = df[(df['close'] > df['high']) | (df['close'] < df['low'])]
if len(bad_close) > 0:
issues['close_out_range'] = len(bad_close)
print(f"❌ Close 범위 이탈: {len(bad_close)}건")
# 3. Open이 High/Low 범위 내인지
bad_open = df[(df['open'] > df['high']) | (df['open'] < df['low'])]
if len(bad_open) > 0:
issues['open_out_range'] = len(bad_open)
print(f"❌ Open 범위 이탈: {len(bad_open)}건")
# 4. 음수 가격
neg_price = df[(df[['open','high','low','close']] <= 0).any(axis=1)]
if len(neg_price) > 0:
issues['negative_price'] = len(neg_price)
print(f"❌ 음수/0 가격: {len(neg_price)}건")
# 5. 음수 거래량
neg_vol = df[df['volume'] < 0]
if len(neg_vol) > 0:
issues['negative_volume'] = len(neg_vol)
print(f"❌ 음수 거래량: {len(neg_vol)}건")
# 6. 중복 타임스탬프
dupes = df.index.duplicated().sum()
if dupes > 0:
issues['duplicates'] = dupes
print(f"❌ 중복 타임스탬프: {dupes}건")
if not issues:
print("✅ OHLCV 무결성 검증 통과")
return issues
def fix_ohlcv_issues(df: pd.DataFrame) -> pd.DataFrame:
"""OHLCV 무결성 문제 자동 수정"""
df = df.copy()
# 중복 제거 (마지막 값 유지)
df = df[~df.index.duplicated(keep='last')]
# High/Low 정규화
actual_high = df[['open','high','low','close']].max(axis=1)
actual_low = df[['open','high','low','close']].min(axis=1)
df['high'] = actual_high
df['low'] = actual_low
# 음수 가격 제거
mask = (df[['open','high','low','close']] <= 0).any(axis=1)
df.loc[mask, ['open','high','low','close']] = np.nan
df = df.fillna(method='ffill')
# 음수 거래량 → 0
df['volume'] = df['volume'].clip(lower=0)
return df데이터 전처리 파이프라인
위의 모든 단계를 하나의 파이프라인으로 통합하면 재사용성이 높아집니다.
class BacktestDataPipeline:
"""백테스트 데이터 전처리 파이프라인"""
def __init__(self, timeframe: str = '1h', outlier_threshold: float = 4.0,
max_fill_gap: int = 3):
self.timeframe = timeframe
self.outlier_threshold = outlier_threshold
self.max_fill_gap = max_fill_gap
self.report = {}
def process(self, df: pd.DataFrame) -> pd.DataFrame:
"""전체 전처리 파이프라인 실행"""
print("=" * 50)
print("📊 백테스트 데이터 전처리 시작")
print("=" * 50)
original_len = len(df)
# 1단계: 무결성 검증 및 수정
print("n[1/4] OHLCV 무결성 검증...")
issues = validate_ohlcv(df)
if issues:
df = fix_ohlcv_issues(df)
self.report['fixed_issues'] = issues
# 2단계: 결측 캔들 처리
print(f"n[2/4] 결측 캔들 탐지...")
missing = detect_missing_candles(df, self.timeframe)
self.report['missing_candles'] = len(missing)
df = fill_missing_candles(df, self.timeframe, self.max_fill_gap)
# 3단계: 이상치 처리
print(f"n[3/4] 이상치 탐지...")
outliers = detect_outliers(df, threshold=self.outlier_threshold)
self.report['outliers'] = outliers.sum()
if outliers.sum() > 0:
df = handle_outliers(df, outliers, method='clip')
# 4단계: NaN 제거
print(f"n[4/4] 최종 정리...")
before = len(df)
df = df.dropna(subset=['close'])
dropped = before - len(df)
self.report['dropped_rows'] = dropped
# 최종 보고
print(f"n{'=' * 50}")
print(f"✅ 전처리 완료")
print(f" 원본: {original_len}행 → 처리 후: {len(df)}행")
print(f" 결측 보간: {self.report['missing_candles']}개")
print(f" 이상치 처리: {self.report['outliers']}개")
print(f" 제거 행: {dropped}개")
return df
# 사용 예시
# pipeline = BacktestDataPipeline(timeframe='1h')
# clean_df = pipeline.process(raw_df)
# clean_df.to_csv('btc_usdt_1h_clean.csv')전처리 품질 체크리스트
| 검증 항목 | 기준 | 조치 |
|---|---|---|
| 결측 비율 | 1% 미만 | 초과 시 데이터 소스 변경 |
| 이상치 비율 | 0.1% 미만 | 초과 시 threshold 조정 |
| OHLCV 무결성 | 에러 0건 | 자동 수정 후 재검증 |
| 시간 연속성 | 갭 없음 | 보간 또는 구간 제외 |
| 데이터 기간 | 최소 1년 | 다양한 시장 환경 포함 |
마무리
백테스트 데이터 전처리는 화려하지 않지만 전략 신뢰도를 결정하는 기반 작업입니다. 결측 보간, 이상치 처리, 무결성 검증을 체계적으로 수행하면 백테스트와 실전 매매의 괴리를 크게 줄일 수 있습니다.
전처리된 데이터로 백테스트 과적합 방지법을 적용하면 더욱 견고한 전략을 개발할 수 있습니다. 또한 켈리 기준 자금 관리법과 결합하면 데이터 품질에 기반한 안정적 포지션 사이징도 가능합니다.