S310 — 실적 가속 측정자 모듈 개발 계획서

날짜: 2026-05-24 (Sun) Topic: L (시스템 재설계) 선행: docs/research/2026-05-24_S310_earnings_signal_research.md (5 에이전트 리서치 통합) 목표: 한국 코퍼스에서 실적 가속 시그널 5+1을 산출하고 base-rate 측정

---

1. 모듈 구조 (MCP 기반)

중요: DART 데이터 수집은 korea-stock-mcp MCP 도구를 통해 Claude 세션이 직접 호출. Python 스크립트는 (a) 캐시된 parquet 읽기, (b) 지표 계산, (c) backtest 만 담당.

scripts/quant/earnings/
├── __init__.py
├── point_in_time.py        # vintage 관리, 50d cutoff, 4Q 90d (cache 기반)
├── cache_io.py             # data/dart_cache/ parquet 읽기/쓰기
├── e1_acceleration.py      # E1: Earnings Acceleration (cache → 산식)
├── e2_korea_sue.py         # E2: Korea-SUE
├── e3_revision.py          # E3: Consensus Revision (FnGuide 캐시 사용)
├── e4_revenue_accel.py     # E4: 매출 가속 Δ2차
├── e5_quality.py           # E5: CFO/NI
├── e6_buyback_event.py     # E6: 자사주 공시 (cache 기반)
├── composite_score.py      # EarningsScore = 4축 z 평균
└── gates.py                # 흑자/매출YoY/커버≥3

scripts/backfill/
├── dart_mcp_batch.md       # MCP 호출 배치 스펙 (Claude 세션에 인계)
├── universe_top500.py      # FDR로 시총 500 universe 산출
└── cache_validator.py      # 백필된 parquet 무결성 점검

data/dart_cache/
├── corp_code_map.parquet   # stock_code ↔ corp_code 매핑 (1회 생성)
├── fs_quarterly.parquet    # corp_code × fy × q × sj_nm × account → 값
├── disclosure_list.parquet # 자사주취득결정·잠정실적 공정공시 stream
└── _vintage_log.parquet    # (corp_code, fy, q, fetched_dt) 추적

MCP 호출 분리 원칙: - 캐시에 있으면 = Python 스크립트가 직접 parquet 로드 (백테스트 시 high-throughput) - 캐시에 없으면 = Claude 세션이 MCP 호출하여 캐시 채움 (백필 1회성)

---

2. 핵심 인터페이스 (point_in_time.py)

def usable_quarter(t: pd.Timestamp, code: str) -> tuple[int, int] | None:
    """시점 t에 종목 code의 사용 가능한 가장 최근 분기.

    Returns: (fy, q) 또는 None (가용 분기 없음).

    규칙:
      - t > 분기말 + 50d (보수) → 정식 보고서 사용 가능
      - 잠정실적 공정공시 있으면 vintage = announce_dt 사용 가능
      - 4Q는 90d cutoff (사업보고서)
      - 정정공시 시 vintage 업데이트
    """
    ...

def get_quarterly_is(code: str, fy: int, q: int) -> dict:
    """분기 손익 (revenue, op_income, net_income, eps, vintage_dt)."""
    ...

def get_quarterly_cf(code: str, fy: int, q: int) -> dict:
    """분기 현금흐름 (cfo)."""
    ...

---

3. 지표 모듈별 산식 + 입력 스키마

3.1 E1 — Earnings Acceleration

def compute_e1(code: str, t: pd.Timestamp) -> dict:
    """EA = (g_q) - (g_q-1), g = (EPS_q - EPS_q-4) / |EPS_q-4|

    Returns: {
      'ea': float,
      'g_curr': float,
      'g_prev': float,
      'vintage_dt': pd.Timestamp,
      'fy_q': str  # e.g. "2026Q1"
    }
    """
    fy_q = usable_quarter(t, code)
    if not fy_q: return {'_error': 'no_usable_quarter'}

    eps_q   = get_eps(code, fy_q)
    eps_q4  = get_eps(code, fy_q - 4 quarters)
    eps_q1  = get_eps(code, fy_q - 1 quarter)
    eps_q5  = get_eps(code, fy_q - 5 quarters)

    g_curr = (eps_q  - eps_q4) / abs(eps_q4)  if eps_q4 != 0 else NaN
    g_prev = (eps_q1 - eps_q5) / abs(eps_q5)  if eps_q5 != 0 else NaN
    ea = g_curr - g_prev
    return {'ea': ea, 'g_curr': g_curr, 'g_prev': g_prev, 'vintage_dt': ..., 'fy_q': ...}

3.2 E2 — Korea-SUE

def compute_e2(code: str, t: pd.Timestamp) -> dict:
    """Korea-SUE = (실제 OP - 컨센 OP) / max(|OP_q-4|, 매출액 × 0.01)

    분모 σ 대신 |OP_q-4|/매출액×0.01 채택 (적자전환 안정성, 실무 권고).
    """
    fy_q = usable_quarter(t, code)
    op_actual = get_quarterly_is(code, *fy_q)['op_income']
    op_consensus = get_consensus(code, fy_q)  # FnGuide
    op_q4 = get_quarterly_is(code, fy_q - 4 quarters)['op_income']
    rev_q = get_quarterly_is(code, *fy_q)['revenue']

    denom = max(abs(op_q4), rev_q * 0.01)
    sue = (op_actual - op_consensus) / denom if denom > 0 else NaN
    return {'sue': sue, ...}

3.3 E3 — Consensus Revision

def compute_e3(code: str, t: pd.Timestamp) -> dict:
    """Revision Ratio = (현재 컨센 EPS - 60d전 EPS) / 60d전 EPS"""
    eps_now = fnguide_consensus_eps(code, t)
    eps_60d = fnguide_consensus_eps(code, t - 60d)
    if eps_60d == 0: return {'_error': 'zero_denom'}
    rev = (eps_now - eps_60d) / abs(eps_60d)
    return {'revision_60d': rev, 'coverage_n': ..., 'as_of_dt': t}

3.4 E4 — 매출 가속 (Δ2차)

def compute_e4(code, t):
    """rev_QoQ_t - rev_QoQ_t-1, rev_QoQ = (rev_q - rev_q-4) / rev_q-4"""
    ...

3.5 E5 — 이익품질

def compute_e5(code, t):
    """CFO_q / NI_q. Sloan accruals 보완 (CF 사용)."""
    cfo = get_quarterly_cf(code, *fy_q)['cfo']
    ni  = get_quarterly_is(code, *fy_q)['net_income']
    return {'quality': cfo / ni if ni != 0 else NaN}

3.6 E6 — 자사주 취득 공시 (event)

def compute_e6_active_window(code, t, window_days=60):
    """t-window ~ t 사이 자사주취득결정 공시가 있는가."""
    events = dart_search(code, kind='자기주식취득결정', date_range=(t-window, t))
    return {'buyback_event': len(events) > 0, 'event_count': len(events),
            'latest_event_dt': max(e.dt for e in events) if events else None}

---

4. Gating (gates.py)

def passes_gate(features: dict) -> bool:
    return all([
        features.get('op_income_q', -1) > 0  # 흑자
            or features.get('op_income_q4', -1) < 0,  # 또는 흑자전환
        features.get('rev_yoy', -1) > 0,  # 매출 YoY > 0
        features.get('coverage_n', 0) >= 3,  # 컨센 ≥ 3명
    ])

수출주 잔차 SUE는 별도 모듈 (gates_export.py) — 수출비중 ≥50% 종목에만 USDKRW 변동률 통제.

---

5. Composite Score (composite_score.py)

def earnings_score(panel: pd.DataFrame) -> pd.DataFrame:
    """cross-section z-score 후 4축 평균.

    panel: code × [ea, sue, revision_60d, rev_accel, quality]
    """
    for col in ['ea', 'sue', 'revision_60d', 'rev_accel']:
        # percentile rank 후 z화 (한국 outlier 대응)
        panel[f'{col}_rank'] = panel[col].rank(pct=True)
        panel[f'{col}_z'] = scipy.stats.norm.ppf(panel[f'{col}_rank'].clip(0.001, 0.999))

    panel['quality_z'] = zscore_robust(panel['quality'])
    panel['earnings_score'] = panel[['ea_z', 'sue_z', 'revision_60d_z', 'rev_accel_z']].mean(axis=1)
    return panel

---

6. 백테스트 Harness (scripts/backtest/lxs_v5_earnings_base_rate.py)

# 매 분기말 + 60d 시점 (보수)에 cross-section snapshot
# universe: KOSPI+KOSDAQ 시총 상위 500
# 각 종목의 E1~E5 산출 → earnings_score
# decile 분할 → 상위10 vs 하위10 forward D+20/D+60 KOSPI-adjusted return
# 성공 기준: top decile - bottom decile spread ≥ 8%/yr (annualized 60D)

검증 단계: - Phase 1: E1 단독 base-rate (시장 가장 단순 시그널) - Phase 2: E2 단독 - Phase 3: E3 단독 - Phase 4: composite (E1+E2+E3 z 평균) - Phase 5: composite + gating - Phase 6: composite × L×S regime multiplier (v4c와 결합)

---

7. 데이터 수집 우선순위 (MCP 기반)

우선	항목	MCP 도구	호출 수	예상 소요
P0	corp_code ↔ stock_code 매핑	get_corp_code (배치 또는 1회 전체)	1~500	5~30분
P0	DART 분기 IS/CF 백필 (시총 500, 16분기)	get_financial_statement (sj_nm=포괄손익계산서·현금흐름표)	16,000 (500×16×2)	수 세션 분할
P0	잠정실적 공정공시 list (시총 500, 4년)	get_disclosure_list (pblntf_ty=B 주요사항)	~500 (종목당 1)	1 세션
P1	FnGuide 컨센 EPS 시계열 (시총 500, daily)	MCP 없음 → 별도 스크래핑 모듈	-	2~3시간
P1	KRX 일봉 수익률 (v2 historical 재사용)	기존 자산	-	-
P2	자사주 취득 결정 공시 stream	get_disclosure_list (pblntf_detail_ty=주요사항)	~500	1 세션
P3	USDKRW 일별 (수출주 잔차 SUE용)	FRED DEXKOUS (기존 fetcher 재사용)	-	즉시

MCP 백필 단계 분할

Phase B1 (1 세션 ~1000 호출): corp_code map 전체 생성 → corp_code_map.parquet Phase B2 (n 세션 분할): 시총 500 × 16분기 × 2 sj_nm = 16,000 호출 / 세션당 1,000건 = 16 세션 분할 - 각 세션: 50종목 × 16분기 × 2 = 1,600 호출 (~30~60분 추정) Phase B3 (1 세션): 잠정실적 공정공시 list 백필 Phase B4 (1 세션): 자사주 취득 결정 stream

대안 — 점진적 백필 (Recommended): - 1차 S311: 상위 100종목 × 8분기만 백필 (~1,600 호출, 1 세션) - E1 base-rate 시범 측정 → 신호 가치 확인 후 확장 - 신호 확인되면 S312/S313에서 500종목 × 16분기로 확장

---

8. 단계별 진행 (S311 ~ S314)

단계	작업	도구	산출물
S311a	universe_top100.py (FDR) + corp_code map	mcp__korea-stock-mcp__get_corp_code 100~500회	data/dart_cache/corp_code_map.parquet
S311b	cache_io.py + point_in_time.py + 5종목 dry-run 백필	mcp__...__get_financial_statement × 5×8×2=80	data/dart_cache/fs_quarterly_dryrun.parquet
S311c	시범 100종목 × 8분기 백필	MCP 1,600 호출, 1 세션	data/dart_cache/fs_quarterly_v1.parquet
S312	E1 단독 base-rate (Phase 1) 100종목 시험	Python 백테스트 (캐시 사용)	data/backtest/lxs_v5_e1/ + 리포트
S313	E1 신호 가치 확인 시 500종목 확장 + FnGuide 스크래핑 + E2/E3	MCP 14,400 호출 (~14 세션) + 스크래핑	data/dart_cache/fs_quarterly_v2.parquet + data/fnguide/consensus_v1.parquet
S314	composite + gating + L×S regime 결합 (Phase 5/6)	Python	data/backtest/lxs_v5_composite/ + 최종 리포트

각 단계 PM 결정 후 진행. S311c는 신호 가치 확인 게이트 — 100종목 시험에서 E1 alpha 없으면 500 확장 안 함.

---

9. 위험 / 검증

9.1 데이터 위험

• MCP 호출 분할 필요: 시총 500 × 16분기 × 2 sj_nm = 16,000 호출. 세션당 ~1,000~1,600 호출 한도로 16 세션 필요. 시범 100종목 × 8분기 = 1,600 호출 (1 세션)로 시작 권고.
• MCP 응답 대기: 단건 호출 응답이 빠르지 않으면 세션 시간 초과. 1차 dry-run으로 호출당 응답 시간 측정 필수.
• FnGuide 스크래핑 차단: User-Agent / IP 회전 / 1초 throttle. 차단 시 손수 수집 모드 fallback.
• K-IFRS 2011 단절: 본 backtest는 2022+만 → 단절 영향 없음.
• MCP 자체가 주가 조회 미지원: mcp__korea-stock-mcp의 get_stock_base_info/get_stock_trade_info는 KRX API 키 없어 사용 금지. 주가는 키움 REST / FDR 사용 (기존 자산).

9.2 검증 위험

• Look-ahead: point_in_time vintage 강제 검사 (assertion). t 이전 모든 데이터의 announce_dt ≤ t.
• Survivorship: 시총 500은 "현재 시점" 시총. 과거 시점은 다른 종목 셋이어야 함. 1차는 "현재 universe" + bias 명시. 2차는 KIS-Value 시총 history 도입.
• Look-back overfitting: 분기 발표 + 50d cutoff (보수)이 60D forward와 겹치지 않게 검증 윈도우 분리.

9.3 성공 기준

• Phase 1 (E1 단독): top - bottom 60D spread ≥ 5%/yr → 신호 가치
• Phase 4 (composite): ≥ 8%/yr → AQR QMJ Global 수준 (월 0.45% ≈ 5.4%/yr 상회)
• Phase 6 (composite × regime): ≥ 12%/yr → 본 시스템 의미 있음

---

10. 다음 세션 (S311) 시작 명령

S311은 3 sub-step으로 분할:

S311a (코드 구현, ~30분)

1. scripts/quant/earnings/ 디렉토리 신규
2. cache_io.py 구현 (parquet read/write, vintage 관리)
3. point_in_time.py 구현 (50d cutoff, 4Q 90d, 잠정공시 vintage)
4. scripts/backfill/universe_top500.py (FDR StockListing → 시총 500 list)

S311b (Dry-run, ~10분)

1. Claude 세션이 MCP로 5종목 corp_code 조회: 005930, 000660, 035720, 207940, 005380
2. 각 종목 2026Q1 분기 IS 1건씩 = 5 호출
3. 응답 형식 확인 + cache_io 검증

S311c (시범 백필, 1 세션)

1. 시총 100종목 × 8분기 (2024Q1~2026Q1) × 2 sj_nm (IS + CF) = 1,600 호출
2. 진행 중 진행도 표시 (50건마다 commit)
3. data/dart_cache/fs_quarterly_v1.parquet 산출
4. 검증: 행 수 / null 비율 / 동일 종목 분기별 일관성

→ S311c 완료 후 E1 단독 base-rate 측정 (S312)으로 진행.