증분 파이프라인 + Lookback 버퍼
SMA60을 위한 120일 버퍼, 신규 데이터만 저장하는 온라인 피처 계산
이 글은 AI_Finance(ALCHEMETRIC) 프로젝트의 피처 엔지니어링 모듈에서 증분(incremental) 파이프라인과 Lookback 버퍼 설계를 다룬다. KOSPI 350종목 x 250거래일의 피처를 매일 전체 재계산하는 대신, _LOOKBACK_CALENDAR_DAYS = 120일의 버퍼를 두고 신규 데이터만 계산하여 upsert하는 구조를 상세히 분석한다.
목차
1. 문제 정의 — 전체 재계산의 비용
ALCHEMETRIC의 피처 엔지니어링 모듈은 Phase 1부터 Phase 7까지 총 75개 피처를 계산한다. KOSPI 기준 약 350종목, 1년치 데이터가 약 250거래일이므로 전체 재계산 시 처리량은 다음과 같다.
| 항목 | 수량 |
|---|---|
| 종목 수 | ~350 |
| 종목당 거래일 | ~250일/년 |
| 피처 수 (Phase 1~7) | 75개 |
| 전체 재계산 셀 수 | 350 x 250 x 75 = ~6,562,500 |
매일 시장이 끝나고 추가되는 신규 데이터는 종목당 1~2일치에 불과하다. 1~2일치를 추가하기 위해 매번 250일 x 350종목을 전부 다시 계산하는 것은 명백한 낭비다.
그러나 단순히 "오늘 데이터만" 가져와서 피처를 계산하면 문제가 생긴다. SMA60(60일 이동평균), RSI14(14일 상대강도지수), Bollinger Bands(20일) 등 롤링 지표는 과거 N일의 가격 데이터가 반드시 필요하기 때문이다.
→ Lookback 버퍼 전략이 이 딜레마를 해결한다. 계산에는 과거 데이터를 포함하되, 저장은 신규분만.
2. Lookback 버퍼 설계 — 왜 120일인가
피처 엔지니어링 모듈 상단에는 다음과 같은 상수가 정의되어 있다.
# SMA60이 가장 긴 lookback → 캘린더일 120일 ≈ 영업일 80일 _LOOKBACK_CALENDAR_DAYS = 120
달력일 vs 거래일
주식 시장의 거래일과 달력일은 같지 않다. 주말(토, 일) 및 공휴일을 제외하면 1달 ≈ 21~22거래일이다. SMA60은 60거래일의 데이터가 필요하므로, 달력일로 환산하면 다음과 같다.
| 롤링 지표 | 필요 거래일 | 달력일 환산 (약) |
|---|---|---|
| SMA5, return_1d | 5일 | ~7일 |
| RSI14 | 14일 | ~20일 |
| SMA20, BB(20), volatility_20d | 20일 | ~28일 |
| EMA26, MACD | 26일 (+9 signal) | ~50일 |
| SMA60 | 60일 | ~85일 |
SMA60이 가장 긴 lookback window이며, 달력일로 약 85일이 필요하다. 여기에 공휴일 집중 구간(설, 추석 등)에 대한 여유분을 더해 120일로 설정했다. 120달력일 ≈ 80거래일이므로 SMA60의 60거래일을 넉넉하게 커버한다.
최소 데이터 검증
lookback 버퍼로 확장된 데이터를 가져온 후에도, 데이터가 충분한지 검증하는 안전장치가 있다.
if len(rows) < 60: logger.warning( f"데이터 부족: {market}:{code} ({len(rows)}행, 최소 60행 필요)", "compute_features", ) return 0
최소 60행(60거래일)이 없으면 SMA60을 계산할 수 없으므로 해당 종목은 건너뛴다. 신규 상장 종목이나 거래 정지 종목이 이 조건에 해당한다.
3. 증분 계산 흐름 — compute_features()
compute_features()는 단일 종목의 피처 계산을 담당하는 핵심 메서드다. incremental=True(기본값)일 때의 흐름을 단계별로 분석한다.
Step 1: last_date 조회
먼저 feature_store에서 해당 종목의 마지막 계산일을 조회한다.
# 증분 모드: 마지막 계산일 확인 if incremental: last_feature = repo.get_latest_features(market, code) if last_feature: last_date = last_feature.date
get_latest_features()는 ORDER BY date DESC LIMIT 1로 최신 레코드를 가져온다. 기존 데이터가 없으면 last_date = None이 되어 전체 계산으로 fallback한다.
Step 2: Lookback 범위 확장
price_start = None if last_date: lookback_start = last_date - timedelta(days=_LOOKBACK_CALENDAR_DAYS) price_start = str(lookback_start) elif start_date: price_start = str(start_date)
last_date가 2025-03-10이라면, lookback_start는 2024-11-11이 된다. 이 날짜부터 가격 데이터를 조회하여 SMA60 등 롤링 지표가 정확하게 계산될 수 있도록 한다.
Step 3: 새 데이터 없으면 스킵
# 증분: 새 데이터 없으면 스킵 if last_date: max_price_date = rows[-1].date if max_price_date <= last_date: return 0
가격 데이터의 최신 날짜가 last_date 이하이면 새로 계산할 것이 없으므로 즉시 0을 반환한다. 이 early return 덕분에 스케줄러가 주말이나 공휴일에 실행되어도 불필요한 계산이 발생하지 않는다.
Step 4: 전체 범위에서 피처 계산
# 3. 피처 계산 (lookback 포함 전체 구간) features_df = self._compute_all_features(df, target_days=target_days) # 3.5~3.11: Phase 2~7 피처 병합 features_df = self._merge_fundamental_features(features_df, market, code, session) features_df = self._merge_macro_features(features_df, session) features_df = self._merge_news_features(features_df, market, code, session) features_df = self._merge_disclosure_features(features_df, market, code, session) features_df = self._merge_supply_demand_features(features_df, market, code, session) features_df = self._merge_alternative_features(features_df, market, code, session)
이 단계에서는 lookback 버퍼를 포함한 전체 범위(120일 전 ~ 오늘)에서 모든 피처를 계산한다. SMA60, RSI14 등 롤링 지표가 과거 데이터로부터 정확하게 계산된다.
Step 5: 신규분만 필터링
# 증분: 신규분만 필터링 if last_date: features_df = features_df[features_df["date"] > last_date] if features_df.empty: return 0
여기가 증분 파이프라인의 핵심이다. 120일치로 계산했지만, 저장하는 것은 last_date 이후의 신규 행만이다. 나머지 lookback 구간의 데이터는 계산 정확성을 위해 사용된 후 폐기된다.
Step 6: Upsert 저장
# 5. DB 저장 saved = repo.upsert_features(records) mode = "증분" if last_date else "전체" logger.info( f"피처 저장 완료 ({mode}): {market}:{code} ({saved}행)", "compute_features", )
upsert_features()는 ON CONFLICT (market, code, date) DO UPDATE 방식으로 동작한다. 이미 존재하는 날짜의 레코드는 업데이트하고, 새 날짜는 삽입한다. 멱등성이 보장되므로 동일한 날짜에 대해 여러 번 실행해도 안전하다.
전체 흐름 다이어그램
증분 파이프라인 흐름 (incremental=True) ─────────────────────────────────────────────────────────── feature_store stock_price ┌─────────────┐ ┌──────────────┐ │ last_date │ │ 전체 가격 │ │ = 2025-3-10 │ │ 2024-11-11 ~ │ └──────┬──────┘ └──────┬───────┘ │ │ ▼ │ lookback_start │ = 3-10 - 120일 │ = 2024-11-11 │ │ │ └─────────┐ ┌─────────────┘ ▼ ▼ ┌───────────────────┐ │ price query │ │ 2024-11-11 ~ 현재 │ │ (~80거래일 + 신규) │ └─────────┬─────────┘ ▼ ┌───────────────────┐ │ _compute_all_ │ │ features() │ │ Phase 1~7 전부 계산 │ │ (lookback 포함) │ └─────────┬─────────┘ ▼ ┌───────────────────┐ │ date > last_date │ │ 필터링 │ │ → 3-11, 3-12만 │ │ 남김 │ └─────────┬─────────┘ ▼ ┌───────────────────┐ │ upsert_features() │ │ 2행만 INSERT/UPDATE│ └───────────────────┘
4. 롤링 지표 정합성 — SMA60 예시
SMA60(60일 단순이동평균)을 예시로 lookback 버퍼의 필요성을 구체적으로 살펴보자.
시나리오: 2025-03-11 피처 계산
feature_store의 마지막 계산일이 2025-03-10이고, 3월 11일의 새 가격 데이터가 들어왔다고 가정하자.
| 방식 | 가져오는 데이터 | SMA60 결과 |
|---|---|---|
| lookback 없음 | 3-11 하루만 | NaN (데이터 1행, 60행 필요) |
| last_date부터 | 3-10 ~ 3-11 (2행) | NaN (데이터 2행, 60행 필요) |
| lookback 120일 | 2024-11-11 ~ 3-11 (~80거래일) | 정확한 값 (60행 이상) |
3월 11일의 SMA60을 정확히 계산하려면 60거래일 전인 대략 12월 초의 종가부터 필요하다. 120달력일(약 80거래일)을 가져오면 SMA60뿐 아니라 SMA20, RSI14, MACD, Bollinger Bands 등 모든 롤링 지표가 정확한 값을 갖게 된다.
코드에서의 피처 계산
# _compute_all_features() 내부 # === 가격 피처 === result["return_1d"] = df["close"].pct_change(1) result["return_5d"] = df["close"].pct_change(5) result["return_20d"] = df["close"].pct_change(20) result["volatility_20d"] = df["close"].pct_change().rolling(20).std() # === 기술적 지표 (가장 긴 window: SMA60) === sma60 = calc_sma(df, 60)["sma"] # ← 60거래일 필요! result["sma_60"] = sma60 # === 파생 피처 === result["price_to_sma60"] = df["close"] / sma60.replace(0, np.nan) result["golden_cross"] = (sma5 > sma20).astype(int)
5. 타겟 변수 동적 생성 — _compute_targets
피처 엔지니어링 파이프라인은 피처뿐 아니라 타겟 변수(label)도 함께 계산한다. ML 학습에 사용할 미래 수익률과 등락 분류를 생성하는 로직이 _compute_all_features() 내부에 포함되어 있다.
# === 타겟 변수 (동적 생성) === for days in target_days: # 기본값: [1, 5] col_return = f"target_return_{days}d" col_class = f"target_class_{days}d" # 미래 N일 후 수익률 = (미래 종가 / 현재 종가) - 1 result[col_return] = df["close"].shift(-days) / df["close"] - 1 # 상승 = 1, 하락 = 0 result[col_class] = (result[col_return] > 0).astype("Int64") result.loc[result[col_return].isna(), col_class] = pd.NA
target_days 파라미터
target_days는 리스트로 전달되며, 기본값은 [1, 5]이다. 이를 통해 다음 4개의 타겟 컬럼이 동적으로 생성된다.
| 컬럼명 | 설명 | 타입 |
|---|---|---|
| target_return_1d | 1일 후 수익률 | float (-0.05 ~ +0.05 등) |
| target_class_1d | 1일 후 등락 (상승=1, 하락=0) | Int64 (nullable) |
| target_return_5d | 5일 후 수익률 | float |
| target_class_5d | 5일 후 등락 (상승=1, 하락=0) | Int64 (nullable) |
shift(-N)과 자연스러운 NULL
df["close"].shift(-5)는 현재 행에서 5행 뒤의 종가를 가져온다. 데이터의 마지막 5행은 아직 미래 데이터가 존재하지 않으므로 자연스럽게 NaN이 된다.
date close shift(-5) target_return_5d target_class_5d ────────── ─────── ────────── ──────────────── ─────────────── 2025-03-05 50,000 51,500 +0.030 1 2025-03-06 50,200 52,000 +0.036 1 2025-03-07 49,800 NaN NULL NULL 2025-03-10 50,100 NaN NULL NULL 2025-03-11 51,500 NaN NULL NULL
6. 성능 비교 — 전체 vs 증분
compute_features()의 incremental 파라미터에 따른 동작 차이를 비교한다.
| 항목 | incremental=False (전체) | incremental=True (증분) |
|---|---|---|
| DB 조회 | 전체 기간 가격 데이터 | last_date - 120일 ~ 현재 |
| 계산 범위 | 전체 기간 (250일+) | ~80거래일 (lookback 포함) |
| 저장 행 수 | 250행 (전체) | 1~2행 (신규분만) |
| 350종목 기준 | 87,500행 INSERT/UPDATE | 350~700행 INSERT/UPDATE |
| 스킵 로직 | 없음 | 새 데이터 없으면 즉시 return 0 |
incremental=False가 필요한 경우
증분 모드가 기본이지만, 다음 상황에서는 incremental=False로 전체 재계산이 필요하다.
| 상황 | 이유 |
|---|---|
| 새 피처 컬럼 추가 | 기존 레코드에 새 컬럼 값이 NULL. 전체 재계산으로 과거 행도 채워야 한다. |
| 피처 계산 로직 변경 | RSI 계산 방식 변경 등. 과거 값도 새 로직으로 다시 계산해야 일관성 유지. |
| 과거 가격 데이터 수정 | 수정된 종가가 롤링 지표에 영향. 증분으로는 과거 행이 업데이트되지 않는다. |
| Phase 업그레이드 | Phase 5 → Phase 6 전환 시 기존 데이터에 섹터/상대강도 피처가 없음. |
7. 코드 구조 & 파이프라인 컨텍스트
compute_all(): 2-Pass 아키텍처
compute_all()은 마켓 전체 종목의 피처를 계산하는 오케스트레이터로, 2-Pass 구조를 사용한다.
def compute_all(self, market, codes=None, ..., incremental=True): """마켓 전체 종목의 피처를 계산 (2-pass)""" # Pass 1: Phase 1~5 + Phase 7 피처 계산 logger.info(f"Pass 1/2: Phase 1~5 피처 계산 ({total}종목)") for code in codes: count = self.compute_features( market, code, ..., incremental=incremental, include_phase6=False, # Phase 6은 Pass 2에서 ) if count > 0: phase1_updated_codes.add(code) # Pass 2: Phase 6 섹터/뉴스정제 피처 phase6_targets = self._get_phase6_targets(...) for code in phase6_targets: self._compute_phase6_only(market, code, ...)
Phase 6(섹터 상대강도, 뉴스 정제)은 동일 섹터 피어 종목들의 Phase 1~5 데이터가 필요하므로, 모든 종목의 Phase 1~5가 완료된 후에 별도 Pass로 실행된다.
_compute_phase6_only(): 증분 지원
Phase 6 계산도 증분을 지원한다. sector_return_1d가 NULL인 레코드만 필터링하여 처리한다.
def _compute_phase6_only(self, market, code, ...): # feature_store에서 Phase 1~5 데이터 로드 features = ml_repo.get_features(market, code, ...) # Phase 6 미계산 레코드만 처리 (sector_return_1d가 NULL) features = [f for f in features if f.sector_return_1d is None] # Phase 6A: 섹터/상대강도 df = self._merge_sector_features(df, market, code, session) # Phase 6B: 뉴스 정제 df = self._merge_news_refined_features(df, market, code, session) # Phase 6 컬럼만 추출하여 upsert saved = ml_repo.upsert_features(records)
_get_phase6_targets(): 증분 모드의 타겟 결정
증분 모드에서 Pass 2의 대상 종목을 결정하는 로직이다. Pass 1에서 신규 데이터가 있던 종목과, Phase 6 컬럼이 아직 NULL인 종목을 합집합으로 처리한다.
def _get_phase6_targets(self, market, all_codes, updated_codes, incremental): if not incremental: return all_codes # 전체 모드: 전종목 # Pass 1에서 갱신된 종목은 무조건 포함 targets = set(updated_codes) # Phase 6 미계산 종목 추가 (sector_return_1d가 NULL인 종목) codes_with_phase6 = MLRepository(session).get_codes_with_phase6(market, all_codes) for code in all_codes: if code not in codes_with_phase6: targets.add(code) return [c for c in all_codes if c in targets]
Upsert 메커니즘 (MLRepository)
upsert_features()는 PostgreSQL의 ON CONFLICT DO UPDATE 또는 SQLite의 동등 구문을 사용한다. unique constraint는 (market, code, date) 복합키이다.
def upsert_features(self, records: list[dict]) -> int: """피처 데이터 Upsert""" # records의 키에서 market, code, date를 제외한 나머지가 update 대상 sample = records[0] update_fields = [k for k in sample.keys() if k not in ("market", "code", "date")] return self._upsert( FeatureStore, records, "uq_feature_store", # constraint name ["market", "code", "date"], # index_elements (SQLite) update_fields, extra_set={"created_at": datetime.now()}, )
이 설계 덕분에 Phase 6의 _compute_phase6_only()가 Phase 6 컬럼만 담은 records를 upsert하면, 기존 Phase 1~5 데이터는 보존되면서 Phase 6 컬럼만 업데이트된다. update_fields가 동적으로 결정되기 때문이다.
전체 파이프라인에서의 위치
┌─────────────────────────────────────────────────────────────────┐ │ ALCHEMETRIC 75-Feature Pipeline │ ├─────────────────────────────────────────────────────────────────┤ │ │ │ Scheduler (매일 18:30) │ │ │ │ │ ▼ │ │ full_collect() │ │ │ │ │ ├── [1] 가격/재무/거시/뉴스/공시 수집 │ │ │ │ │ ├── [2] NLP 센티먼트 분석 │ │ │ │ │ ├── [3] FeatureEngineer.compute_all(incremental=True) │ │ │ ├── Pass 1: Phase 1~5,7 (종목별 증분) │ │ │ └── Pass 2: Phase 6 (섹터/뉴스정제) │ │ │ │ │ └── [4] ML 학습/예측 │ │ │ └─────────────────────────────────────────────────────────────────┘
75개 피처 Phase 구성
| Phase | 카테고리 | 피처 수 | 증분 전략 |
|---|---|---|---|
| Phase 1 | 가격/기술지표/파생 | 24 | Lookback 120일 |
| Phase 2 | 재무 (PER, PBR, ROE 등) | 9 | date join + forward-fill |
| Phase 3 | 거시경제 | 11 | date join + forward-fill |
| Phase 4 | 뉴스 센티먼트 | 5 | date join |
| Phase 5 | 공시 + 수급 | 10 | 60일 lookback window |
| Phase 6 | 섹터 상대강도 + 뉴스 정제 | 10 | Pass 2 별도 실행 |
| Phase 7 | 대안 데이터 (트렌드, 커뮤니티) | 6 | 10일 lookback |
'Project > 🚀 AI Finance 2026' 카테고리의 다른 글
| #21 [모듈3] Machine Learning (0) | 2026.03.13 |
|---|---|
| #20 [모듈2-4] 피어 비교 피처 — 섹터 상대강도 + 복합 인덱스 (0) | 2026.03.13 |
| #18 [모듈2-2] 결측 데이터 전략 - Graceful NULL + 이중 Forward-Fill (0) | 2026.03.13 |
| #17 [모듈2-1] 2-Pass 아키텍처 피처 엔지니어링 (0) | 2026.03.13 |
| #16 [모듈2] Feature Engineering (0) | 2026.03.13 |
