포스트

2026-04-14 강의 정리

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

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

목차

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

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()     # 누적 수익률

끝

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

인기 태그

AI LLM