#17 [모듈2-1] 2-Pass 아키텍처 피처 엔지니어링

피처 엔지니어링

2-Pass 아키텍처

피어 의존성을 해결하는 피처 파이프라인 설계

Data Collect → NLP Sentiment → Feature Eng. → ML Signal

이 글은 AI_Finance(ALCHEMETRIC) 프로젝트의 피처 엔지니어링 모듈에서 Phase 6 섹터/상대강도 피처가 만든 순환 의존성을 해결하기 위해 도입한 2-Pass 아키텍처를 다룬다. A 종목의 섹터 피처를 계산하려면 동일 섹터 B, C, D 종목의 수익률이 먼저 필요하고, B 종목도 마찬가지라는 피어 의존성(peer dependency) 문제를 어떻게 파이프라인 분리로 풀었는지 상세히 해부한다.

목차

1. 문제 정의 — Phase 6이 만든 순환 의존성
2. 설계 선택지 — 왜 2-Pass인가
3. Pass 1 상세 — Phase 1~5+7 종목별 계산
4. Pass 2 상세 — Phase 6 섹터/뉴스 정제 계산
5. 증분 최적화 — _get_phase6_targets()
6. 데이터 흐름 다이어그램
7. 파이프라인 컨텍스트 — 75개 피처 전체 맵

1. 문제 정의 — Phase 6이 만든 순환 의존성

ALCHEMETRIC의 피처 파이프라인은 종목별로 독립 계산이 가능하도록 설계되어 있다. Phase 1(가격/기술지표)부터 Phase 5(공시/수급)까지는 각 종목의 데이터만으로 피처를 계산할 수 있다. 그러나 Phase 6에서 도입된 섹터/상대강도 피처는 이 전제를 깨뜨렸다.

피어 의존성이란

Phase 6A의 핵심 피처인 sector_return_1d(섹터 평균 1일 수익률)를 계산하려면 동일 섹터에 속한 모든 피어 종목의 return_1d가 필요하다. relative_strength_1d(상대강도)는 return_1d - sector_return_1d로 계산되므로, 결국 피어 종목의 Phase 1 데이터에 의존한다.

구체적인 시나리오를 보자. 반도체 섹터에 삼성전자(005930), SK하이닉스(000660), DB하이텍(000990)이 있다고 하자.

계산 대상	필요한 데이터	문제
005930의 sector_return_1d	000660, 000990의 return_1d	피어의 Phase 1이 먼저 완료되어야 함
000660의 sector_return_1d	005930, 000990의 return_1d	동일한 문제
000990의 sector_return_1d	005930, 000660의 return_1d	동일한 문제

종목별 순차 처리에서 005930을 먼저 계산하면, 000660의 return_1d가 아직 feature_store에 없다. 모든 종목의 Phase 1~5가 완료된 후에야 Phase 6을 계산할 수 있다는 뜻이다.

핵심 모순: Phase 6은 피어 종목의 Phase 1~5 결과를 읽어야 하지만, 종목별 독립 계산 구조에서는 피어의 결과가 아직 존재하지 않는다. 한 패스(single-pass)로는 해결 불가능하다.

Phase 6B의 sector_news_sentiment도 같은 문제를 갖는다. 섹터 내 피어 종목들의 news_sentiment(Phase 4에서 계산)를 집계해야 하므로, 피어의 Phase 4가 완료되어야 한다.

2. 설계 선택지 — 왜 2-Pass인가

피어 의존성을 해결하기 위해 검토한 세 가지 접근법과 각각의 트레이드오프를 비교한다.

접근법	동작 방식	장점	단점
종목별 순차 계산	종목 A 처리 시 피어 B,C,D의 raw price를 직접 조회하여 수익률 계산	단일 루프	O(N x M) DB 조회. 섹터 30종목이면 종목당 30회 price 조회. 전체 복잡도 O(N²)
2-Pass 분리	Pass 1에서 전종목 Phase 1~5 저장 → Pass 2에서 feature_store를 bulk read하여 Phase 6 계산	DB 조회 최소화 (벌크), 증분 최적화 가능	2회 순회. 코드 분리 필요
배치 후처리	전종목 계산 완료 후 별도 배치 작업으로 Phase 6 업데이트	기존 코드 변경 없음	별도 스케줄링 필요. 파이프라인 일관성 보장 어려움. 트랜잭션 분리 문제

2-Pass가 최적인 이유

종목별 순차 계산은 코드가 단순해 보이지만 실제로는 매우 비효율적이다. KOSPI 900종목 기준, 평균 섹터 크기 30종목이라면 Pass 1에서 이미 저장한 return_1d를 다시 raw price로부터 재계산하게 된다.

2-Pass 분리는 Pass 1에서 이미 feature_store에 저장된 return_1d, return_5d, return_20d를 get_sector_features_bulk()로 한 번에 가져온다. DB I/O를 최소화하면서 compute_all() 메서드 하나에 전체 흐름이 캡슐화되어 파이프라인 일관성도 보장된다.

설계 결정: 2-Pass 분리를 채택한다. compute_all() 내부에서 Pass 1과 Pass 2를 순차 실행하며, include_phase6 플래그로 Pass 1에서 Phase 6 계산을 스킵한다.

3. Pass 1 상세 — Phase 1~5+7 종목별 계산

Pass 1은 compute_all()에서 각 종목에 대해 compute_features(include_phase6=False)를 호출한다. 이 단계에서 Phase 1(가격/기술지표), Phase 2(재무), Phase 3(거시), Phase 4(뉴스), Phase 5(공시/수급), Phase 7(대안 데이터)을 계산하고 feature_store에 INSERT/UPSERT한다.

compute_all() — Pass 1 루프

# Pass 1: Phase 1~5 피처 계산 (Phase 6 제외)
logger.info(f"Pass 1/2: Phase 1~5 피처 계산 ({total}종목)", "compute_all")
phase1_updated_codes = set()

for code in codes:
    try:
        count = self.compute_features(
            market, code, start_date, end_date,
            target_days=target_days,
            incremental=incremental,
            include_phase6=False,   # Phase 6 스킵
        )
        if count > 0:
            success += 1
            phase1_updated_codes.add(code)  # Pass 2 대상 추적
        else:
            skipped += 1
    except Exception as e:
        failed += 1

핵심은 phase1_updated_codes 집합이다. Pass 1에서 실제로 신규 데이터를 저장한 종목만 추적하여, Pass 2에서 불필요한 재계산을 방지한다.

compute_features() 내부 — include_phase6 분기

compute_features()는 단일 종목의 피처 계산 메서드다. include_phase6 파라미터가 False이면 Phase 6A/6B를 건너뛴다.

def compute_features(
    self, market, code, start_date=None, end_date=None,
    target_days=None, incremental=True,
    include_phase6=True,       # 기본값은 True (단일 종목 호출 시)
) -> int:
    # ... Phase 1~5+7 계산 ...

    # 3.10 대안 데이터 피처 병합 (Phase 7)
    features_df = self._merge_alternative_features(
        features_df, market, code, session,
    )

    # compute_all()에서는 include_phase6=False로 Pass 1 수행 후
    # Pass 2에서 _compute_phase6_only()로 별도 처리
    if include_phase6:
        features_df = self._merge_sector_features(
            features_df, market, code, session,
        )
        features_df = self._merge_news_refined_features(
            features_df, market, code, session,
        )

주의: include_phase6의 기본값은 True이다. 단일 종목 API 호출(compute_features() 직접 호출)에서는 피어 데이터가 이미 feature_store에 존재한다고 가정하고 Phase 6을 포함한다. compute_all()에서만 False로 호출한다.

증분 모드와 _LOOKBACK_CALENDAR_DAYS

증분 모드(incremental=True)에서는 마지막 계산일 이후의 신규 데이터만 처리한다. 그러나 SMA60 같은 지표는 60 영업일(약 120 캘린더일)의 lookback이 필요하므로, price 데이터는 last_date - 120일부터 조회한다.

# SMA60이 가장 긴 lookback → 캘린더일 120일 ≈ 영업일 80일
_LOOKBACK_CALENDAR_DAYS = 120

# 증분 모드: 마지막 계산일 확인
if incremental:
    last_feature = repo.get_latest_features(market, code)
    if last_feature:
        last_date = last_feature.date

if last_date:
    lookback_start = last_date - timedelta(days=_LOOKBACK_CALENDAR_DAYS)
    price_start = str(lookback_start)

4. Pass 2 상세 — Phase 6 섹터/뉴스 정제 계산

Pass 2는 Pass 1이 완료된 후 실행된다. 이 시점에서 전종목의 Phase 1~5+7 피처가 feature_store에 저장되어 있으므로, 피어 종목의 수익률을 자유롭게 조회할 수 있다.

compute_all() — Pass 2 루프

# Pass 2: Phase 6 섹터/뉴스정제 피처 계산
phase6_targets = self._get_phase6_targets(
    market, codes, phase1_updated_codes, incremental,
)
logger.info(
    f"Pass 2/2: Phase 6 피처 계산 ({len(phase6_targets)}/{total}종목)",
    "compute_all",
)

phase6_success = 0
phase6_failed = 0
for code in phase6_targets:
    try:
        count = self._compute_phase6_only(
            market, code, start_date, end_date,
        )
        if count > 0:
            phase6_success += 1
    except Exception as e:
        phase6_failed += 1

_compute_phase6_only() — Phase 6 전용 계산기

이 메서드는 Pass 2의 핵심이다. feature_store에서 해당 종목의 기존 레코드를 읽고, sector_return_1d가 NULL인 행(Phase 6 미계산)만 필터링하여 Phase 6A + 6B를 계산한 뒤 UPDATE한다.

def _compute_phase6_only(self, market, code, start_date=None, end_date=None) -> int:
    with database.session() as session:
        ml_repo = MLRepository(session)

        features = ml_repo.get_features(market, code, start_date, end_date)
        if not features:
            return 0

        # Phase 6 미계산 레코드만 처리 (sector_return_1d가 NULL)
        features = [f for f in features if f.sector_return_1d is None]
        if not features:
            return 0

        # Phase 6 계산에 필요한 최소 컬럼만 로드
        df = pd.DataFrame([{
            "date": f.date,
            "return_1d": float(f.return_1d) if f.return_1d is not None else None,
            "return_5d": float(f.return_5d) if f.return_5d is not None else None,
            "return_20d": float(f.return_20d) if f.return_20d is not None else None,
            "news_sentiment": float(f.news_sentiment) if f.news_sentiment is not None else None,
        } for f in features])

        # 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
        phase6_cols = _SECTOR_FEATURE_COLUMNS + _NEWS_REFINED_FEATURE_COLUMNS
        records = []
        for _, row in df.iterrows():
            record = {"market": market, "code": code, "date": row["date"]}
            for col in phase6_cols:
                # ... type 변환 ...
                record[col] = val
            records.append(record)

        saved = ml_repo.upsert_features(records)  # UPDATE only phase6 cols
        return saved

Phase 6A — _merge_sector_features()

Phase 6A는 7개의 섹터/상대강도 피처를 계산한다. 핵심은 MLRepository.get_sector_features_bulk()로 동일 섹터 전체 종목의 return_1d, return_5d, return_20d를 한 번의 벌크 쿼리로 가져오는 것이다.

# MLRepository.get_sector_features_bulk()
def get_sector_features_bulk(self, market, codes, start_date, end_date):
    return (
        self.session.query(
            FeatureStore.code,
            FeatureStore.date,
            FeatureStore.return_1d,
            FeatureStore.return_5d,
            FeatureStore.return_20d,
            FeatureStore.news_sentiment,  # Phase 6B에서도 활용
        )
        .filter(
            FeatureStore.market == market,
            FeatureStore.code.in_(codes),     # IN 절로 벌크 조회
            FeatureStore.date >= start_date,
            FeatureStore.date <= end_date,
        )
        .order_by(FeatureStore.date)
        .all()
    )

이 벌크 데이터를 기반으로 날짜별 섹터 집계(groupby)와 상대강도를 계산한다.

피처	계산 로직
sector_return_1d	섹터 내 전종목 return_1d 평균
sector_return_5d	섹터 내 전종목 return_5d 평균
relative_strength_1d	종목 return_1d - sector_return_1d
relative_strength_5d	종목 return_5d - sector_return_5d
relative_strength_20d	종목 return_20d - 섹터 평균 return_20d
sector_momentum_rank	섹터 내 return_5d 백분위 순위 (0.0~1.0)
sector_breadth	섹터 내 return_1d > 0인 종목 비율 (시장폭)

Phase 6B — _merge_news_refined_features()

Phase 6B는 3개의 뉴스 정제 피처를 계산한다. sector_news_sentiment는 Phase 6A와 동일한 벌크 쿼리 결과에서 news_sentiment 컬럼을 활용한다.

피처	계산 로직
news_relevance_ratio	뉴스 제목에 종목명이 포함된 비율
news_sentiment_filtered	종목명 포함 뉴스만의 필터링된 센티먼트
sector_news_sentiment	동일 섹터 피어의 news_sentiment 평균 (feature_store bulk read)

upsert 전략: _compute_phase6_only()는 Phase 6 컬럼만 포함한 record를 upsert_features()에 전달한다. upsert_features()는 record의 키에서 market, code, date를 제외한 나머지를 update_fields로 동적 추출하므로, Phase 1~5 컬럼은 건드리지 않고 Phase 6 컬럼만 UPDATE한다.

5. 증분 최적화 — _get_phase6_targets()

전체 모드에서는 당연히 전종목을 Pass 2 대상으로 삼는다. 그러나 증분 모드에서 매번 900종목 전체를 Pass 2로 돌리면 비효율적이다. _get_phase6_targets()는 실제 업데이트가 필요한 종목만 선별한다.

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인 종목)
    with database.session() as session:
        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]  # 순서 유지

대상 선정 로직

Pass 2 대상은 두 가지 조건의 합집합이다.

조건	소스	이유
Pass 1 갱신 종목	phase1_updated_codes (set)	새 날짜가 추가되었으므로 Phase 6도 계산 필요
Phase 6 미계산 종목	get_codes_with_phase6()	이전 실행에서 Phase 6이 실패했거나 처음 계산하는 종목

get_codes_with_phase6() — 미계산 종목 탐지

# MLRepository
def get_codes_with_phase6(self, market, codes) -> set[str]:
    """Phase 6 데이터가 있는 종목코드 집합"""
    rows = (
        self.session.query(FeatureStore.code)
        .filter(
            FeatureStore.market == market,
            FeatureStore.code.in_(codes),
            FeatureStore.sector_return_1d.isnot(None),  # NULL이 아닌 것만
        )
        .distinct()
        .all()
    )
    return {r[0] for r in rows}

sector_return_1d IS NOT NULL을 Phase 6 계산 완료의 마커로 사용한다. 이 컬럼이 NULL이면 해당 종목의 Phase 6이 아직 처리되지 않았다는 뜻이다.

시나리오 예시

KOSPI 900종목, 일일 증분 모드 기준으로 실제 동작을 시뮬레이션한다.

상태	종목수	Pass 2 대상?
Pass 1에서 신규 데이터 저장됨	850	O (updated_codes)
Pass 1에서 변동 없음 (skipped), Phase 6 존재	40	X (스킵)
Pass 1에서 변동 없음 (skipped), Phase 6 NULL	10	O (미계산 복구)

결과: 전종목 900개 중 860개만 Pass 2를 실행한다. 40종목(4.4%)을 스킵하여 불필요한 DB 조회와 계산을 절약한다. 특히 Phase 6 미계산 종목 10개는 이전 실행에서 에러로 실패한 케이스를 자동 복구하는 효과가 있다.

6. 데이터 흐름 다이어그램

2-Pass 아키텍처의 전체 데이터 흐름을 시각화한다.

                        compute_all(market, codes)
                                  |
            +-------------------------------------------+
            |                PASS 1 (종목별)              |
            +-------------------------------------------+
            |  for code in codes:                       |
            |    compute_features(include_phase6=False) |
            |                                           |
            |  +------ Phase 1: 가격/기술지표 ----------+  |
            |  |  return_1d, sma_20, rsi_14, ...     |  |
            |  +-------------------------------------+  |
            |            |                              |
            |  +------ Phase 2: 재무 -----------------+  |
            |  |  per, pbr, roe, debt_ratio, ...     |  |
            |  +-------------------------------------+  |
            |            |                              |
            |  +------ Phase 3: 거시 -----------------+  |
            |  |  vix, kospi_index, krw_usd, ...     |  |
            |  +-------------------------------------+  |
            |            |                              |
            |  +------ Phase 4: 뉴스 센티먼트 ----------+  |
            |  |  news_sentiment, news_volume, ...   |  |
            |  +-------------------------------------+  |
            |            |                              |
            |  +------ Phase 5: 공시/수급 -------------+  |
            |  |  disclosure_count, short_sell, ...  |  |
            |  +-------------------------------------+  |
            |            |                              |
            |  +------ Phase 7: 대안 데이터 ------------+  |
            |  |  google_trend, community, ...       |  |
            |  +-------------------------------------+  |
            |            |                              |
            |      [Phase 6 SKIP]                       |
            |            |                              |
            |       INSERT/UPSERT                       |
            +-----------|-------------------------------+
                        v
         +==============================+
         |    feature_store (DB)        |
         |  Phase 1~5+7: FILLED         |
         |  Phase 6:     NULL           |
         +==============================+
                        |
             _get_phase6_targets()
           updated_codes + NULL codes
                        |
            +-----------------------------------------+
            |              PASS 2 (종목별)              |
            +-----------------------------------------+
            |  for code in phase6_targets:             |
            |    _compute_phase6_only(code)            |
            |                                          |
            |    1. get_features(code)                 |
            |       sector_return_1d IS NULL 필터       |
            |                                          |
            |    2. get_sector_features_bulk()         |
            |       피어의 return_1d/5d/20d BULK READ    |
            |                                           |
            |  +------ Phase 6A: 섹터/상대강도 --------+   |
            |  |  sector_return, relative_strength  |   |
            |  |  sector_momentum_rank, breadth     |   |
            |  +------------------------------------+   |
            |            |                              |
            |  +------ Phase 6B: 뉴스 정제 -----------+   |
            |  |  news_relevance, sector_news_sent  |   |
            |  +------------------------------------+   |
            |            |                              |
            |  UPDATE (Phase 6 columns only)            |
            +-----------|-------------------------------+
                        v
         +==============================+
         |    feature_store (DB)        |
         |  Phase 1~5+7: FILLED         |
         |  Phase 6:     FILLED         |
         +==============================+

피처 엔지니어링

2-Pass 아키텍처

피어 의존성을 해결하는 피처 파이프라인 설계

Data Collect → NLP Sentiment → Feature Eng. → ML Signal

이 글은 AI_Finance(ALCHEMETRIC) 프로젝트의 피처 엔지니어링 모듈에서 Phase 6 섹터/상대강도 피처가 만든 순환 의존성을 해결하기 위해 도입한 2-Pass 아키텍처를 다룬다. A 종목의 섹터 피처를 계산하려면 동일 섹터 B, C, D 종목의 수익률이 먼저 필요하고, B 종목도 마찬가지라는 피어 의존성(peer dependency) 문제를 어떻게 파이프라인 분리로 풀었는지 상세히 해부한다.

목차

1. 문제 정의 — Phase 6이 만든 순환 의존성
2. 설계 선택지 — 왜 2-Pass인가
3. Pass 1 상세 — Phase 1~5+7 종목별 계산
4. Pass 2 상세 — Phase 6 섹터/뉴스 정제 계산
5. 증분 최적화 — _get_phase6_targets()
6. 데이터 흐름 다이어그램
7. 파이프라인 컨텍스트 — 75개 피처 전체 맵

1. 문제 정의 — Phase 6이 만든 순환 의존성

ALCHEMETRIC의 피처 파이프라인은 종목별로 독립 계산이 가능하도록 설계되어 있다. Phase 1(가격/기술지표)부터 Phase 5(공시/수급)까지는 각 종목의 데이터만으로 피처를 계산할 수 있다. 그러나 Phase 6에서 도입된 섹터/상대강도 피처는 이 전제를 깨뜨렸다.

피어 의존성이란

Phase 6A의 핵심 피처인 sector_return_1d(섹터 평균 1일 수익률)를 계산하려면 동일 섹터에 속한 모든 피어 종목의 return_1d가 필요하다. relative_strength_1d(상대강도)는 return_1d - sector_return_1d로 계산되므로, 결국 피어 종목의 Phase 1 데이터에 의존한다.

구체적인 시나리오를 보자. 반도체 섹터에 삼성전자(005930), SK하이닉스(000660), DB하이텍(000990)이 있다고 하자.

계산 대상	필요한 데이터	문제
005930의 sector_return_1d	000660, 000990의 return_1d	피어의 Phase 1이 먼저 완료되어야 함
000660의 sector_return_1d	005930, 000990의 return_1d	동일한 문제
000990의 sector_return_1d	005930, 000660의 return_1d	동일한 문제

종목별 순차 처리에서 005930을 먼저 계산하면, 000660의 return_1d가 아직 feature_store에 없다. 모든 종목의 Phase 1~5가 완료된 후에야 Phase 6을 계산할 수 있다는 뜻이다.

핵심 모순: Phase 6은 피어 종목의 Phase 1~5 결과를 읽어야 하지만, 종목별 독립 계산 구조에서는 피어의 결과가 아직 존재하지 않는다. 한 패스(single-pass)로는 해결 불가능하다.

Phase 6B의 sector_news_sentiment도 같은 문제를 갖는다. 섹터 내 피어 종목들의 news_sentiment(Phase 4에서 계산)를 집계해야 하므로, 피어의 Phase 4가 완료되어야 한다.

2. 설계 선택지 — 왜 2-Pass인가

피어 의존성을 해결하기 위해 검토한 세 가지 접근법과 각각의 트레이드오프를 비교한다.

접근법	동작 방식	장점	단점
종목별 순차 계산	종목 A 처리 시 피어 B,C,D의 raw price를 직접 조회하여 수익률 계산	단일 루프	O(N x M) DB 조회. 섹터 30종목이면 종목당 30회 price 조회. 전체 복잡도 O(N²)
2-Pass 분리	Pass 1에서 전종목 Phase 1~5 저장 → Pass 2에서 feature_store를 bulk read하여 Phase 6 계산	DB 조회 최소화 (벌크), 증분 최적화 가능	2회 순회. 코드 분리 필요
배치 후처리	전종목 계산 완료 후 별도 배치 작업으로 Phase 6 업데이트	기존 코드 변경 없음	별도 스케줄링 필요. 파이프라인 일관성 보장 어려움. 트랜잭션 분리 문제

2-Pass가 최적인 이유

종목별 순차 계산은 코드가 단순해 보이지만 실제로는 매우 비효율적이다. KOSPI 900종목 기준, 평균 섹터 크기 30종목이라면 Pass 1에서 이미 저장한 return_1d를 다시 raw price로부터 재계산하게 된다.

2-Pass 분리는 Pass 1에서 이미 feature_store에 저장된 return_1d, return_5d, return_20d를 get_sector_features_bulk()로 한 번에 가져온다. DB I/O를 최소화하면서 compute_all() 메서드 하나에 전체 흐름이 캡슐화되어 파이프라인 일관성도 보장된다.

설계 결정: 2-Pass 분리를 채택한다. compute_all() 내부에서 Pass 1과 Pass 2를 순차 실행하며, include_phase6 플래그로 Pass 1에서 Phase 6 계산을 스킵한다.

3. Pass 1 상세 — Phase 1~5+7 종목별 계산

Pass 1은 compute_all()에서 각 종목에 대해 compute_features(include_phase6=False)를 호출한다. 이 단계에서 Phase 1(가격/기술지표), Phase 2(재무), Phase 3(거시), Phase 4(뉴스), Phase 5(공시/수급), Phase 7(대안 데이터)을 계산하고 feature_store에 INSERT/UPSERT한다.

compute_all() — Pass 1 루프

# Pass 1: Phase 1~5 피처 계산 (Phase 6 제외)
logger.info(f"Pass 1/2: Phase 1~5 피처 계산 ({total}종목)", "compute_all")
phase1_updated_codes = set()

for code in codes:
    try:
        count = self.compute_features(
            market, code, start_date, end_date,
            target_days=target_days,
            incremental=incremental,
            include_phase6=False,   # Phase 6 스킵
        )
        if count > 0:
            success += 1
            phase1_updated_codes.add(code)  # Pass 2 대상 추적
        else:
            skipped += 1
    except Exception as e:
        failed += 1

핵심은 phase1_updated_codes 집합이다. Pass 1에서 실제로 신규 데이터를 저장한 종목만 추적하여, Pass 2에서 불필요한 재계산을 방지한다.

compute_features() 내부 — include_phase6 분기

compute_features()는 단일 종목의 피처 계산 메서드다. include_phase6 파라미터가 False이면 Phase 6A/6B를 건너뛴다.

def compute_features(
    self, market, code, start_date=None, end_date=None,
    target_days=None, incremental=True,
    include_phase6=True,       # 기본값은 True (단일 종목 호출 시)
) -> int:
    # ... Phase 1~5+7 계산 ...

    # 3.10 대안 데이터 피처 병합 (Phase 7)
    features_df = self._merge_alternative_features(
        features_df, market, code, session,
    )

    # compute_all()에서는 include_phase6=False로 Pass 1 수행 후
    # Pass 2에서 _compute_phase6_only()로 별도 처리
    if include_phase6:
        features_df = self._merge_sector_features(
            features_df, market, code, session,
        )
        features_df = self._merge_news_refined_features(
            features_df, market, code, session,
        )

주의: include_phase6의 기본값은 True이다. 단일 종목 API 호출(compute_features() 직접 호출)에서는 피어 데이터가 이미 feature_store에 존재한다고 가정하고 Phase 6을 포함한다. compute_all()에서만 False로 호출한다.

증분 모드와 _LOOKBACK_CALENDAR_DAYS

증분 모드(incremental=True)에서는 마지막 계산일 이후의 신규 데이터만 처리한다. 그러나 SMA60 같은 지표는 60 영업일(약 120 캘린더일)의 lookback이 필요하므로, price 데이터는 last_date - 120일부터 조회한다.

# SMA60이 가장 긴 lookback → 캘린더일 120일 ≈ 영업일 80일
_LOOKBACK_CALENDAR_DAYS = 120

# 증분 모드: 마지막 계산일 확인
if incremental:
    last_feature = repo.get_latest_features(market, code)
    if last_feature:
        last_date = last_feature.date

if last_date:
    lookback_start = last_date - timedelta(days=_LOOKBACK_CALENDAR_DAYS)
    price_start = str(lookback_start)

4. Pass 2 상세 — Phase 6 섹터/뉴스 정제 계산

Pass 2는 Pass 1이 완료된 후 실행된다. 이 시점에서 전종목의 Phase 1~5+7 피처가 feature_store에 저장되어 있으므로, 피어 종목의 수익률을 자유롭게 조회할 수 있다.

compute_all() — Pass 2 루프

# Pass 2: Phase 6 섹터/뉴스정제 피처 계산
phase6_targets = self._get_phase6_targets(
    market, codes, phase1_updated_codes, incremental,
)
logger.info(
    f"Pass 2/2: Phase 6 피처 계산 ({len(phase6_targets)}/{total}종목)",
    "compute_all",
)

phase6_success = 0
phase6_failed = 0
for code in phase6_targets:
    try:
        count = self._compute_phase6_only(
            market, code, start_date, end_date,
        )
        if count > 0:
            phase6_success += 1
    except Exception as e:
        phase6_failed += 1

_compute_phase6_only() — Phase 6 전용 계산기

이 메서드는 Pass 2의 핵심이다. feature_store에서 해당 종목의 기존 레코드를 읽고, sector_return_1d가 NULL인 행(Phase 6 미계산)만 필터링하여 Phase 6A + 6B를 계산한 뒤 UPDATE한다.

def _compute_phase6_only(self, market, code, start_date=None, end_date=None) -> int:
    with database.session() as session:
        ml_repo = MLRepository(session)

        features = ml_repo.get_features(market, code, start_date, end_date)
        if not features:
            return 0

        # Phase 6 미계산 레코드만 처리 (sector_return_1d가 NULL)
        features = [f for f in features if f.sector_return_1d is None]
        if not features:
            return 0

        # Phase 6 계산에 필요한 최소 컬럼만 로드
        df = pd.DataFrame([{
            "date": f.date,
            "return_1d": float(f.return_1d) if f.return_1d is not None else None,
            "return_5d": float(f.return_5d) if f.return_5d is not None else None,
            "return_20d": float(f.return_20d) if f.return_20d is not None else None,
            "news_sentiment": float(f.news_sentiment) if f.news_sentiment is not None else None,
        } for f in features])

        # 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
        phase6_cols = _SECTOR_FEATURE_COLUMNS + _NEWS_REFINED_FEATURE_COLUMNS
        records = []
        for _, row in df.iterrows():
            record = {"market": market, "code": code, "date": row["date"]}
            for col in phase6_cols:
                # ... type 변환 ...
                record[col] = val
            records.append(record)

        saved = ml_repo.upsert_features(records)  # UPDATE only phase6 cols
        return saved

Phase 6A — _merge_sector_features()

Phase 6A는 7개의 섹터/상대강도 피처를 계산한다. 핵심은 MLRepository.get_sector_features_bulk()로 동일 섹터 전체 종목의 return_1d, return_5d, return_20d를 한 번의 벌크 쿼리로 가져오는 것이다.

# MLRepository.get_sector_features_bulk()
def get_sector_features_bulk(self, market, codes, start_date, end_date):
    return (
        self.session.query(
            FeatureStore.code,
            FeatureStore.date,
            FeatureStore.return_1d,
            FeatureStore.return_5d,
            FeatureStore.return_20d,
            FeatureStore.news_sentiment,  # Phase 6B에서도 활용
        )
        .filter(
            FeatureStore.market == market,
            FeatureStore.code.in_(codes),     # IN 절로 벌크 조회
            FeatureStore.date >= start_date,
            FeatureStore.date <= end_date,
        )
        .order_by(FeatureStore.date)
        .all()
    )

이 벌크 데이터를 기반으로 날짜별 섹터 집계(groupby)와 상대강도를 계산한다.

피처	계산 로직
sector_return_1d	섹터 내 전종목 return_1d 평균
sector_return_5d	섹터 내 전종목 return_5d 평균
relative_strength_1d	종목 return_1d - sector_return_1d
relative_strength_5d	종목 return_5d - sector_return_5d
relative_strength_20d	종목 return_20d - 섹터 평균 return_20d
sector_momentum_rank	섹터 내 return_5d 백분위 순위 (0.0~1.0)
sector_breadth	섹터 내 return_1d > 0인 종목 비율 (시장폭)

Phase 6B — _merge_news_refined_features()

Phase 6B는 3개의 뉴스 정제 피처를 계산한다. sector_news_sentiment는 Phase 6A와 동일한 벌크 쿼리 결과에서 news_sentiment 컬럼을 활용한다.

피처	계산 로직
news_relevance_ratio	뉴스 제목에 종목명이 포함된 비율
news_sentiment_filtered	종목명 포함 뉴스만의 필터링된 센티먼트
sector_news_sentiment	동일 섹터 피어의 news_sentiment 평균 (feature_store bulk read)

upsert 전략: _compute_phase6_only()는 Phase 6 컬럼만 포함한 record를 upsert_features()에 전달한다. upsert_features()는 record의 키에서 market, code, date를 제외한 나머지를 update_fields로 동적 추출하므로, Phase 1~5 컬럼은 건드리지 않고 Phase 6 컬럼만 UPDATE한다.

5. 증분 최적화 — _get_phase6_targets()

전체 모드에서는 당연히 전종목을 Pass 2 대상으로 삼는다. 그러나 증분 모드에서 매번 900종목 전체를 Pass 2로 돌리면 비효율적이다. _get_phase6_targets()는 실제 업데이트가 필요한 종목만 선별한다.

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인 종목)
    with database.session() as session:
        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]  # 순서 유지

대상 선정 로직

Pass 2 대상은 두 가지 조건의 합집합이다.

조건	소스	이유
Pass 1 갱신 종목	phase1_updated_codes (set)	새 날짜가 추가되었으므로 Phase 6도 계산 필요
Phase 6 미계산 종목	get_codes_with_phase6()	이전 실행에서 Phase 6이 실패했거나 처음 계산하는 종목

get_codes_with_phase6() — 미계산 종목 탐지

# MLRepository
def get_codes_with_phase6(self, market, codes) -> set[str]:
    """Phase 6 데이터가 있는 종목코드 집합"""
    rows = (
        self.session.query(FeatureStore.code)
        .filter(
            FeatureStore.market == market,
            FeatureStore.code.in_(codes),
            FeatureStore.sector_return_1d.isnot(None),  # NULL이 아닌 것만
        )
        .distinct()
        .all()
    )
    return {r[0] for r in rows}

sector_return_1d IS NOT NULL을 Phase 6 계산 완료의 마커로 사용한다. 이 컬럼이 NULL이면 해당 종목의 Phase 6이 아직 처리되지 않았다는 뜻이다.

시나리오 예시

KOSPI 900종목, 일일 증분 모드 기준으로 실제 동작을 시뮬레이션한다.

상태	종목수	Pass 2 대상?
Pass 1에서 신규 데이터 저장됨	850	O (updated_codes)
Pass 1에서 변동 없음 (skipped), Phase 6 존재	40	X (스킵)
Pass 1에서 변동 없음 (skipped), Phase 6 NULL	10	O (미계산 복구)

결과: 전종목 900개 중 860개만 Pass 2를 실행한다. 40종목(4.4%)을 스킵하여 불필요한 DB 조회와 계산을 절약한다. 특히 Phase 6 미계산 종목 10개는 이전 실행에서 에러로 실패한 케이스를 자동 복구하는 효과가 있다.

6. 데이터 흐름 다이어그램

2-Pass 아키텍처의 전체 데이터 흐름을 시각화한다.

                        compute_all(market, codes)
                                  |
            +-------------------------------------------+
            |                PASS 1 (종목별)              |
            +-------------------------------------------+
            |  for code in codes:                       |
            |    compute_features(include_phase6=False) |
            |                                           |
            |  +------ Phase 1: 가격/기술지표 ----------+  |
            |  |  return_1d, sma_20, rsi_14, ...     |  |
            |  +-------------------------------------+  |
            |            |                              |
            |  +------ Phase 2: 재무 -----------------+  |
            |  |  per, pbr, roe, debt_ratio, ...     |  |
            |  +-------------------------------------+  |
            |            |                              |
            |  +------ Phase 3: 거시 -----------------+  |
            |  |  vix, kospi_index, krw_usd, ...     |  |
            |  +-------------------------------------+  |
            |            |                              |
            |  +------ Phase 4: 뉴스 센티먼트 ----------+  |
            |  |  news_sentiment, news_volume, ...   |  |
            |  +-------------------------------------+  |
            |            |                              |
            |  +------ Phase 5: 공시/수급 -------------+  |
            |  |  disclosure_count, short_sell, ...  |  |
            |  +-------------------------------------+  |
            |            |                              |
            |  +------ Phase 7: 대안 데이터 ------------+  |
            |  |  google_trend, community, ...       |  |
            |  +-------------------------------------+  |
            |            |                              |
            |      [Phase 6 SKIP]                       |
            |            |                              |
            |       INSERT/UPSERT                       |
            +-----------|-------------------------------+
                        v
         +==============================+
         |    feature_store (DB)        |
         |  Phase 1~5+7: FILLED         |
         |  Phase 6:     NULL           |
         +==============================+
                        |
             _get_phase6_targets()
           updated_codes + NULL codes
                        |
            +-----------------------------------------+
            |              PASS 2 (종목별)              |
            +-----------------------------------------+
            |  for code in phase6_targets:             |
            |    _compute_phase6_only(code)            |
            |                                          |
            |    1. get_features(code)                 |
            |       sector_return_1d IS NULL 필터       |
            |                                          |
            |    2. get_sector_features_bulk()         |
            |       피어의 return_1d/5d/20d BULK READ    |
            |                                           |
            |  +------ Phase 6A: 섹터/상대강도 --------+   |
            |  |  sector_return, relative_strength  |   |
            |  |  sector_momentum_rank, breadth     |   |
            |  +------------------------------------+   |
            |            |                              |
            |  +------ Phase 6B: 뉴스 정제 -----------+   |
            |  |  news_relevance, sector_news_sent  |   |
            |  +------------------------------------+   |
            |            |                              |
            |  UPDATE (Phase 6 columns only)            |
            +-----------|-------------------------------+
                        v
         +==============================+
         |    feature_store (DB)        |
         |  Phase 1~5+7: FILLED         |
         |  Phase 6:     FILLED         |
         +==============================+

핵심 포인트

1) 데이터 분리: Pass 1에서 feature_store는 중간 저장소(intermediate store) 역할을 한다. Phase 1~5+7은 채워지고 Phase 6 컬럼은 NULL인 상태로 커밋된다.

2) 읽기 방향 역전: Pass 1에서는 외부 데이터 소스(StockPrice, StockFundamental 등)에서 읽지만, Pass 2에서는 feature_store 자체에서 읽는다. get_sector_features_bulk()가 이 역할을 담당한다.

3) 부분 업데이트: Pass 2의 upsert_features()는 record에 포함된 키만 UPDATE한다. Phase 6 컬럼 10개만 담긴 record를 전달하면 나머지 65개 컬럼은 그대로 유지된다.

7. 파이프라인 컨텍스트 — 75개 피처 전체 맵

2-Pass 아키텍처는 전체 Phase 1~7 파이프라인 중 Phase 6의 계산 순서를 재배치하는 설계이다. 아래 표는 feature_store에 저장되는 75개 피처의 전체 구조와 각 Phase가 어느 Pass에서 실행되는지를 보여준다.

Phase	카테고리	피처 수	Pass	대표 피처
Phase 1	가격 / 기술지표 / 파생	24	Pass 1	return_1d sma_20 rsi_14 macd bb_pctb
Phase 2	재무 / 기관외인	9	Pass 1	per pbr roe foreign_ratio
Phase 3	거시경제	11	Pass 1	vix krw_usd kospi_index fed_rate
Phase 4	뉴스 센티먼트	5	Pass 1	news_sentiment news_volume
Phase 5A	공시	5	Pass 1	disclosure_count_30d disclosure_sentiment
Phase 5B	수급 (공매도/프로그램)	5	Pass 1	short_selling_ratio program_net_volume
Phase 6A	섹터 / 상대강도	7	Pass 2	sector_return_1d relative_strength_5d
Phase 6B	뉴스 정제	3	Pass 2	sector_news_sentiment news_relevance_ratio
Phase 7	대안 데이터	6	Pass 1	google_trend_score community_post_volume
합계		75	Pass 1: 65 / Pass 2: 10

여기에 target_class_1d, target_class_5d, target_return_1d, target_return_5d 4개의 타겟 변수를 더하면 feature_store 테이블은 총 79개의 데이터 컬럼을 갖는다.

Phase 번호와 실행 순서의 차이

Phase 번호(1~7)는 피처의 논리적 분류이고, Pass(1~2)는 실행 순서이다. Phase 7(대안 데이터)은 번호가 6보다 크지만 피어 의존성이 없으므로 Pass 1에서 계산된다. Phase 6만이 피어 의존성 때문에 Pass 2로 분리된다.

실행 순서 (compute_features 내부):

Pass 1
  3.  _compute_all_features()     # Phase 1: 가격/기술지표/파생
  3.5 _merge_fundamental_features()# Phase 2: 재무
  3.6 _merge_macro_features()      # Phase 3: 거시
  3.7 _merge_news_features()       # Phase 4: 뉴스
  3.8 _merge_disclosure_features() # Phase 5A: 공시
  3.9 _merge_supply_demand()       # Phase 5B: 수급
  3.10 _merge_alternative_features()# Phase 7: 대안 데이터
  3.11 [SKIP if include_phase6=False]

Pass 2 (compute_all → _compute_phase6_only)
       _merge_sector_features()     # Phase 6A: 섹터/상대강도
       _merge_news_refined_features()# Phase 6B: 뉴스 정제

설계 원칙: 피처 파이프라인에서 종목-독립 피처(Phase 1~5, 7)와 피어-의존 피처(Phase 6)를 분리하는 것이 2-Pass 아키텍처의 본질이다. 이 분리는 include_phase6 하나의 boolean 파라미터로 제어되며, feature_store를 Pass 1과 Pass 2 사이의 공유 저장소(shared intermediate store)로 활용한다.

ALCHEMETRIC — M2 Feature Engineering

2-Pass Architecture / Document M2-1

핵심 포인트

1) 데이터 분리: Pass 1에서 feature_store는 중간 저장소(intermediate store) 역할을 한다. Phase 1~5+7은 채워지고 Phase 6 컬럼은 NULL인 상태로 커밋된다.

2) 읽기 방향 역전: Pass 1에서는 외부 데이터 소스(StockPrice, StockFundamental 등)에서 읽지만, Pass 2에서는 feature_store 자체에서 읽는다. get_sector_features_bulk()가 이 역할을 담당한다.

3) 부분 업데이트: Pass 2의 upsert_features()는 record에 포함된 키만 UPDATE한다. Phase 6 컬럼 10개만 담긴 record를 전달하면 나머지 65개 컬럼은 그대로 유지된다.

7. 파이프라인 컨텍스트 — 75개 피처 전체 맵

2-Pass 아키텍처는 전체 Phase 1~7 파이프라인 중 Phase 6의 계산 순서를 재배치하는 설계이다. 아래 표는 feature_store에 저장되는 75개 피처의 전체 구조와 각 Phase가 어느 Pass에서 실행되는지를 보여준다.

Phase	카테고리	피처 수	Pass	대표 피처
Phase 1	가격 / 기술지표 / 파생	24	Pass 1	return_1d sma_20 rsi_14 macd bb_pctb
Phase 2	재무 / 기관외인	9	Pass 1	per pbr roe foreign_ratio
Phase 3	거시경제	11	Pass 1	vix krw_usd kospi_index fed_rate
Phase 4	뉴스 센티먼트	5	Pass 1	news_sentiment news_volume
Phase 5A	공시	5	Pass 1	disclosure_count_30d disclosure_sentiment
Phase 5B	수급 (공매도/프로그램)	5	Pass 1	short_selling_ratio program_net_volume
Phase 6A	섹터 / 상대강도	7	Pass 2	sector_return_1d relative_strength_5d
Phase 6B	뉴스 정제	3	Pass 2	sector_news_sentiment news_relevance_ratio
Phase 7	대안 데이터	6	Pass 1	google_trend_score community_post_volume
합계		75	Pass 1: 65 / Pass 2: 10

여기에 target_class_1d, target_class_5d, target_return_1d, target_return_5d 4개의 타겟 변수를 더하면 feature_store 테이블은 총 79개의 데이터 컬럼을 갖는다.

Phase 번호와 실행 순서의 차이

Phase 번호(1~7)는 피처의 논리적 분류이고, Pass(1~2)는 실행 순서이다. Phase 7(대안 데이터)은 번호가 6보다 크지만 피어 의존성이 없으므로 Pass 1에서 계산된다. Phase 6만이 피어 의존성 때문에 Pass 2로 분리된다.

실행 순서 (compute_features 내부):

Pass 1
  3.  _compute_all_features()     # Phase 1: 가격/기술지표/파생
  3.5 _merge_fundamental_features()# Phase 2: 재무
  3.6 _merge_macro_features()      # Phase 3: 거시
  3.7 _merge_news_features()       # Phase 4: 뉴스
  3.8 _merge_disclosure_features() # Phase 5A: 공시
  3.9 _merge_supply_demand()       # Phase 5B: 수급
  3.10 _merge_alternative_features()# Phase 7: 대안 데이터
  3.11 [SKIP if include_phase6=False]

Pass 2 (compute_all → _compute_phase6_only)
       _merge_sector_features()     # Phase 6A: 섹터/상대강도
       _merge_news_refined_features()# Phase 6B: 뉴스 정제

설계 원칙: 피처 파이프라인에서 종목-독립 피처(Phase 1~5, 7)와 피어-의존 피처(Phase 6)를 분리하는 것이 2-Pass 아키텍처의 본질이다. 이 분리는 include_phase6 하나의 boolean 파라미터로 제어되며, feature_store를 Pass 1과 Pass 2 사이의 공유 저장소(shared intermediate store)로 활용한다.

ALCHEMETRIC — M2 Feature Engineering

2-Pass Architecture / Document M2-1