포스트

2026-04-14 강의 정리

추천 시스템 - 데이터 수집 및 전처리 (시장, 상품 데이터)

게시
By kimseokryu
12 분읽는 시간
2026-04-14 강의 정리

시계열 데이터를 정리할 때는 비어 있는 값을 채우는 작업도 분석 가정을 바꾼다. ETF 데이터를 예제로 스키마, 결측치, 이상치, 파생 변수를 연결하되 원본과 변환 결과를 구분해서 관리한다. 전처리 파이프라인의 판단 기준은 데이터가 깨끗해 보이는지가 아니라 어떤 값을 왜 바꿨는지 설명할 수 있는지다.

목차

  1. ETF 데이터 구조 개요
  2. dataclass로 ETFSchema 설계
  3. FinanceDataReader로 실데이터 수집
  4. 결측치 처리
  5. 이상치 탐지
  6. Feature Engineering
  7. preprocess_pipeline 완성
  8. 이동평균 & 트렌드 분해 개념
  9. 자주 나오는 실수 / 주의사항
  10. [보충] pandas 시계열 기초
  11. 전처리로 바뀐 값을 추적하기

1. ETF 데이터 구조 개요

1
2
3
4
5
6
7
8
9
10
11
12
13

# ETF 카테고리별 종목 수
categories = {
    "국내주식": 145, "해외주식": 120, "채권": 85,
    "섹터": 95, "원자재": 25, "부동산": 15, "기타": 30,
}

# 개별 ETF 데이터 예시
etf_universe = [
    {"name": "KODEX 200",    "cat": "국내주식", "ret_1y": 12.3,  "risk": 15.2},
    {"name": "TIGER S&P500", "cat": "해외주식", "ret_1y": 18.5,  "risk": 13.8},
    {"name": "KODEX 2차전지",  "cat": "섹터",   "ret_1y": -5.2,  "risk": 32.1},
    ...
]

리스크-수익률 시각화 (산포도)

1
2
3
4
5
6
7
8

fig, ax = plt.subplots(figsize=(8, 5))
for cat in df_etf['cat'].unique():
    sub = df_etf[df_etf['cat'] == cat]
    ax.scatter(sub['risk'], sub['ret_1y'], label=cat)
    for _, row in sub.iterrows():
        ax.annotate(row['name'][:8], (row['risk'], row['ret_1y']))

ax.axhline(y=10, color='gray', linestyle='--', alpha=0.5)  # 기준선 10%

2. dataclass로 ETFSchema 설계

스키마는 컬럼 이름뿐 아니라 날짜 기준과 값의 단위를 맞추는 출발점이다. 데이터 수집처가 달라져도 같은 의미로 처리할 수 있도록 필수 항목과 허용 범위를 구분하고, 원본 값과 검증 결과를 함께 남긴다.

왜 dataclass를 쓰나?

  • __init__ 자동 생성, 코드 간결
  • __post_init__ 으로 유효성 검증 가능
  • asdict() → dict → JSON 직렬화 편리
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

from dataclasses import dataclass, field, asdict
from typing import List

@dataclass
class ETFSchema:
    ticker: str          # ex) "069500"
    name: str
    category: str
    expense_ratio: float
    risk_level: str = "중간"          # 기본값
    aum_billion: float = 0.0
    description: str = ""
    keywords: List[str] = field(default_factory=list)  # 가변 기본값은 field() 필수

    def __post_init__(self):
        if self.expense_ratio < 0 or self.expense_ratio > 5:
            raise ValueError(f"수수료 범위 오류: {self.expense_ratio}")
        valid_risks = ["매우낮음", "낮음", "중간", "높음", "매우높음"]
        if self.risk_level not in valid_risks:
            raise ValueError(f"리스크 등급 오류: {self.risk_level}")

field(default_factory=list) — 리스트를 기본값으로 쓸 때 반드시 이렇게 작성. keywords: List[str] = [] 로 쓰면 모든 인스턴스가 같은 리스트를 공유하는 버그 발생!

ETFSchema 사용 예

1
2
3
4
5
6
7
8
9
10
11
12
13
14

etf = ETFSchema(
    ticker="069500", name="KODEX 200",
    category="국내주식", expense_ratio=0.15,
    keywords=["코스피", "대형주", "인덱스"]
)

# dict 변환
etf_dict = asdict(etf)

# JSON 직렬화
etf_json = json.dumps(etf_dict, ensure_ascii=False, indent=2)

# JSON → dataclass 복원
ETFSchema(**json.loads(etf_json))

ETFReturns — 수익률 유효성 검증

1
2
3
4
5
6
7
8
9
10
11
12
13
14

@dataclass
class ETFReturns:
    ticker: str
    return_1m: float
    return_3m: float
    return_1y: float
    return_3y: float

    def __post_init__(self):
        # 반복 검증: 필드 이름 리스트로 처리 (DRY)
        for field_name in ['return_1m', 'return_3m', 'return_1y', 'return_3y']:
            value = getattr(self, field_name)
            if value < -100 or value > 500:
                raise ValueError(f"{field_name} 오류: {value}")

실행 결과:

1
2

ETFReturns("123456", -200.1, 5.4, 12.3, 28.5)
→ ValueError: return_1m 오류: -200.1

3. FinanceDataReader로 실데이터 수집

1
2
3
4
5
6
7
8
9
10
11
12
13

import FinanceDataReader as fdr

# 한국 ETF 전체 목록
etf_list = fdr.StockListing('ETF/KR')
# → 총 1,088개 종목

# 개별 종목 가격 데이터 (1년치)
ticker = "069500"  # KODEX 200
end_date   = datetime.now().strftime('%Y-%m-%d')
start_date = (datetime.now() - timedelta(days=365)).strftime('%Y-%m-%d')

price_df = fdr.DataReader(ticker, start_date, end_date)
# → 244개 거래일 데이터

price_df 컬럼 구조:

1
2
3
4
5

             Open   High    Low  Close    Volume    Change
Date
2025-04-14  32343  32570  32271  32373  5621869   0.006342
...
2026-04-14  90245  91600  89955  90430 14863983   0.029017

데이터 수집 함수화

1
2
3
4
5
6
7
8
9

def collect_etf_price(ticker, days=365):
    end_date   = datetime.now().strftime('%Y-%m-%d')
    start_date = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
    try:
        df = fdr.DataReader(ticker, start_date, end_date)
        if len(df) > 0:
            return {'status': 'real', 'data': df, 'ticker': ticker}
    except Exception:
        pass

4. 결측치 처리

언제 결측치가 생기나?

1

주말, 공휴일, 거래정지 → 해당 날짜 row 전체가 없거나 일부 컬럼이 NaN

처리 방법 비교

1
2
3
4
5
6
7
8
9
10
11
12

# 1. Forward Fill (앞 값으로 채우기)
ffill_df = sample_df.copy()
ffill_df['Close'] = ffill_df['Close'].ffill()

# 2. Linear Interpolation (선형 보간)
interp_df = sample_df.copy()
interp_df['Close'] = interp_df['Close'].interpolate(method='linear')

# 3. Rolling Average (이동평균으로 채우기)
rolling_df = sample_df.copy()
rolling_mean = rolling_df['Close'].rolling(5, min_periods=1).mean()
rolling_df['Close'] = rolling_df['Close'].fillna(rolling_mean)
방법특징적합한 경우
Forward Fill직전 값 그대로 복사단기 결측, 급격한 변화 없을 때
Interpolation앞뒤 값의 중간값으로 매끄럽게연속적인 시계열 데이터
Rolling Average주변 N일 평균노이즈가 많은 데이터

5. 이상치 탐지

IQR 방법

1
2
3
4

def detect_iqr(series, factor=1.5):
    q1, q3 = series.quantile(0.25), series.quantile(0.75)
    iqr = q3 - q1
    return (series < q1 - factor * iqr) | (series > q3 + factor * iqr)

Z-score 방법

1
2
3

def detect_z(series, threshold=3.0):
    z = (series - series.mean()) / series.std()
    return z.abs() > threshold

실행 결과 (KODEX 200 244일 데이터 기준):

1
2
3

Z-score 이상치:  Volume=3개, Change=7개
IQR 이상치:      Volume=10개, Change=23개
교집합 (확실한 이상치): Volume=3개, Change=7개

z_outliers & iqr_outliers — 두 방법 모두 이상치로 판정된 경우가 가장 확실한 이상치

6. Feature Engineering

1
2
3
4
5
6
7
8
9
10
11

features = ffill_df[['Close']].copy()

# 일일 수익률 (%)
features['return'] = features['Close'].pct_change()

# 로그 수익률 (정규분포에 더 가까움, 포트폴리오 계산에 유리)
features['log_return'] = np.log(features['Close'] / features['Close'].shift(1))

# 변동성 (5일, 20일 롤링 표준편차)
features['vol_5d']  = features['return'].rolling(5).std()
features['vol_20d'] = features['return'].rolling(20).std()

pct_change vs log_return 비교:

1
2

return     = (P_t - P_{t-1}) / P_{t-1}   ← 일반 수익률
log_return = ln(P_t / P_{t-1})            ← 로그 수익률 (작은 값에서 거의 동일)

7. preprocess_pipeline 완성

아래 코드는 전처리 단계를 연결하는 예제이며 모든 시계열에 공통 적용할 규칙은 아니다. 특히 보간과 뒤쪽 값으로 채우기는 이후 시점의 정보를 사용할 수 있으므로, 과거 시점에서의 판단을 재현하는 분석에서는 사용 가능한 데이터 범위를 따로 제한해야 한다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

def preprocess_pipeline(df):
    result = df[['Close']].copy()

    # Step 1: 결측치 처리 (linear interpolation + ffill + bfill)
    result['Close'] = result['Close'].interpolate(method='linear')
    result['Close'] = result['Close'].ffill().bfill()

    # Step 2: 이상치 처리 (IQR * 1.5 초과 → ffill 대체)
    ret = result['Close']
    q1, q3 = ret.quantile(0.25), ret.quantile(0.75)
    iqr = q3 - q1
    outlier_mask = (ret < q1 - 1.5 * iqr) | (ret > q3 + 1.5 * iqr)
    result.loc[outlier_mask, 'Close'] = np.nan
    result['Close'] = result['Close'].ffill()

    # Step 3: Feature 추가
    result['return']     = result['Close'].pct_change()
    result['log_return'] = np.log(result['Close'] / result['Close'].shift(1))

    return result

실행 결과:

1
2
3
4

              Close    return  log_return
2025-04-14  32373.0       NaN         NaN
2025-04-15  32692.0  0.009854    0.009806
2025-04-16  32183.0 -0.015570   -0.015692

8. 이동평균 & 트렌드 분해 개념

이동평균 (Moving Average)

  • 노이즈 제거, 추세 파악에 사용
1
2

SMA (Standard MA): 최근 N개의 단순 평균
EMA (Exponential MA): 최근 값에 더 큰 가중치를 부여

트렌드 분해 (Decomposition)

시계열 = 트렌드 + 계절성 + 잔차

1
2
3

트렌드(Trend):      장기적인 방향성 (예: 시장이 전반적으로 성장)
계절성(Seasonality): 반복 패턴 (예: 여름엔 에어컨 ETF↑)
잔차(Residual):     나머지 (설명되지 않는 노이즈)

자주 나오는 실수 / 주의사항

  • keywords: List[str] = [] 금지: 클래스 변수로 공유되어 서로 다른 인스턴스가 같은 리스트를 수정하는 버그 → field(default_factory=list) 사용
  • 결측치와 이상치 순서: 반드시 결측치 처리 먼저 → 이상치 제거. 순서 반대로 하면 이상치를 결측으로 오인할 수 있음
  • pct_change()의 첫 행: 항상 NaN → dropna() 호출 필요
  • fdr.DataReader 예외처리: 상장 폐지/데이터 없는 종목 있음 → try-except 필수

[보충] pandas 시계열 기초

1
2
3
4
5
6
7
8
9
10
11
12
13

# 날짜 인덱스로 필터링
df['2026-01':'2026-03']       # 슬라이싱
df.loc['2026-04-14']          # 특정 날짜

# 리샘플링 (일별 → 주별 마지막 값)
weekly = df['Close'].resample('W').last()

# 롤링 (슬라이딩 윈도우)
df['Close'].rolling(20).mean()   # 20일 이동평균
df['Close'].rolling(20).std()    # 20일 표준편차

# 누적 합/곱
(1 + df['return']).cumprod()     # 누적 수익률

11. 전처리로 바뀐 값을 추적하기

결측치나 이상치를 일괄 대체하기 전에 수집 실패인지 실제 변화인지 구분한다. 원본을 보존하고 수정한 위치와 이유를 기록해 두면 지표가 달라졌을 때 전처리 단계까지 원인을 되짚을 수 있다.

파이프라인을 확인할 때는 결측 행 수, 대체된 값, 날짜 정렬과 중복 여부를 전후로 비교한다. 첫 수익률처럼 계산에 필요한 이전 값이 없어 생긴 결측치와 수집 오류도 같은 방식으로 처리하지 않는다.

모두의연구소, AI/LLM 서비스 개발 과정
AI LLM
이 기사는 저작권자의 CC BY 4.0 라이센스를 따릅니다.

인기 태그

AI LLM