테스트 프레임워크 설정 및 프로젝트 구조

금융 시계열 데이터를 처리하는 분석 파이프라인에서는 수치 계산의 정확성과 외부 API 호출의 안정성이 필수적입니다. 이를 위해 pytest와 커버리지 측정 플러그인을 설치하여 테스트 환경을 구성합니다.

pip install pytest pytest-cov pytest-asyncio freezegun

효율적인 테스트 관리를 위해 소스 코드와 테스트 코드를 명확히 분리하고, 도메인별로 모듈을 나누는 것이 좋습니다.

quant_analysis_pipeline/
├── core/
│   ├── market_data/      # 시세 데이터 수집 모듈
│   ├── indicators/       # 기술적 지표 연산 모듈
│   └── alerts/           # 알림 전송 모듈
├── tests/
│   ├── conftest.py       # 전역 Fixture 및 설정
│   ├── test_market_data.py
│   ├── test_indicators.py
│   └── test_alerts.py
└── pyproject.toml

외부 의존성 격리를 위한 Mock 객체 활용

실제 네트워크 요청 없이 데이터 수집 로직을 검증하기 위해 unittest.mock을 사용하여 외부 API 응답을 모방합니다.

# core/market_data/equity_client.py
def fetch_equity_history(ticker: str, interval: str = "1d") -> dict:
    """외부 벤더를 통해 주식 히스토리 데이터를 조회합니다."""
    # 실제 API 호출 로직
    pass

테스트 코드에서는 네트워크 계층을 패치(Patch)하여 고정된 응답을 반환하도록 설정합니다.

# tests/test_market_data.py
import pytest
from unittest.mock import patch, MagicMock
from core.market_data.equity_client import fetch_equity_history

def test_fetch_equity_history_returns_valid_payload():
    """정상적인 API 응답 시 데이터 파싱 로직을 검증합니다."""
    mock_api_response = MagicMock()
    mock_api_response.json.return_value = {
        "symbol": "MSFT",
        "prices": [{"date": "2023-10-01", "close": 310.5}]
    }
    mock_api_response.status_code = 200

    with patch('core.market_data.equity_client.requests.get', return_value=mock_api_response):
        payload = fetch_equity_history("MSFT")
        
        assert payload is not None
        assert payload["symbol"] == "MSFT"
        assert len(payload["prices"]) == 1

수치 연산 로직 및 엣지 케이스 검증

기술적 지표 계산은 금융 시스템의 핵심입니다. 이동평균(SMA)이나 상대강도지수(RSI)와 같은 지표가 경계값이나 극단적인 데이터에서 어떻게 동작하는지 확인해야 합니다.

# tests/test_indicators.py
from core.indicators.technical_metrics import compute_sma, compute_rsi

def test_compute_sma_with_custom_window():
    """지정된 기간의 단순이동평균(SMA) 계산 정확도를 테스트합니다."""
    price_series = [10.0, 20.0, 30.0, 40.0, 50.0, 60.0]
    
    sma_values = compute_sma(price_series, window_size=4)
    
    # 4일 SMA는 (10+20+30+40)/4=25, (20+30+40+50)/4=35, (30+40+50+60)/4=45
    assert len(sma_values) == 3
    assert sma_values[-1] == 45.0

def test_compute_rsi_overbought_oversold_conditions():
    """RSI 지표의 과매수 및 과매도 구간 판별 로직을 검증합니다."""
    # 지속적 상승 시나리오
    bullish_trend = [100 + i * 2 for i in range(15)]
    assert compute_rsi(bullish_trend, period=14) > 70.0

    # 지속적 하락 시나리오
    bearish_trend = [100 - i * 2 for i in range(15)]
    assert compute_rsi(bearish_trend, period=14) < 30.0

알림 시스템 테스트 및 예외 처리

슬랙(Slack)이나 이메일과 같은 외부 알림 서비스 연동 시, 성공 및 실패 상황에 대한 예외 처리 로직을 모의 객체로 테스트합니다.

# tests/test_alerts.py
from unittest.mock import patch, MagicMock
from core.alerts.slack_bot import dispatch_slack_notification

def test_dispatch_notification_handles_server_error():
    """슬랙 API 서버 오류(5xx) 발생 시 재시도 또는 실패 플래그 반환을 확인합니다."""
    error_response = MagicMock()
    error_response.status_code = 503
    error_response.raise_for_status.side_effect = Exception("Service Unavailable")

    with patch('core.alerts.slack_bot.httpx.post', return_value=error_response):
        is_sent = dispatch_slack_notification(channel="#quant-signals", text="Anomaly detected")
        
        assert is_sent is False

고급 테스트 기법: 파라미터화 및 Fixture

다양한 입력값에 대한 검증을 단순화하기 위해 @pytest.mark.parametrize를 사용하며, 반복되는 테스트 데이터 준비는 Fixture로 추상화합니다.

# tests/test_indicators.py (계속)
import pytest
import pandas as pd
from core.indicators.volatility import check_price_anomaly

@pytest.fixture
def mock_ohlcv_dataframe():
    """테스트용 OHLCV 데이터프레임을 생성합니다."""
    return pd.DataFrame({
        'open': [100, 102, 101, 105, 120],
        'high': [103, 104, 102, 108, 125],
        'low': [99, 101, 100, 104, 118],
        'close': [102, 101, 101, 107, 122],
        'volume': [5000, 4800, 5100, 6000, 15000]
    })

@pytest.mark.parametrize("threshold,expected_anomaly", [
    (0.05, True),   # 5% 변동성 임계값: 마지막 날 급등으로 이상치 탐지
    (0.30, False),  # 30% 변동성 임계값: 정상 범위 내로 간주
])
def test_price_anomaly_detection(mock_ohlcv_dataframe, threshold, expected_anomaly):
    """가격 변동성 임계값에 따른 이상치 탐지 알고리즘을 테스트합니다."""
    result = check_price_anomaly(mock_ohlcv_dataframe, volatility_threshold=threshold)
    assert result == expected_anomaly

시간 의존적 로직 및 비동기 코드 테스트

장 마감 시간 확인이나 비동기 데이터 스트리밍과 같이 시간 및 이벤트 루프에 의존하는 코드는 freezegun과 pytest-asyncio를 활용해 제어합니다.

# tests/test_market_data.py (계속)
from freezegun import freeze_time
from core.market_data.scheduler import is_market_open

@freeze_time("2023-11-10 14:30:00") # 미국 동부 시간 기준 장중
def test_market_status_during_trading_hours():
    """특정 시간대에 시장 개방 상태를 올바르게 판단하는지 검증합니다."""
    assert is_market_open(exchange="NYSE") is True

import pytest
from core.market_data.async_stream import stream_realtime_quotes

@pytest.mark.asyncio
async def test_async_quote_streaming():
    """비동기 웹소켓 또는 스트리밍 클라이언트의 데이터 수신 로직을 테스트합니다."""
    quotes = await stream_realtime_quotes(symbols=["TSLA", "NVDA"], duration_sec=2)
    assert len(quotes) > 0
    assert "TSLA" in quotes[0].symbol

테스트 커버리지 측정 및 CI/CD 파이프라인 통합

코드베이스의 테스트 품질을 정량적으로 파악하기 위해 커버리지 리포트를 생성하고, GitHub Actions를 통해 자동화합니다.

# 터미널에서 커버리지 리포트 생성
pytest --cov=core --cov-report=term-missing --cov-report=html tests/

아래는 코드 푸시 및 PR 생성 시 자동으로 테스트를 실행하고 커버리지를 보고하는 GitHub Actions 워크플로우 구성 예시입니다.

# .github/workflows/ci_pipeline.yml
name: Quant Pipeline CI

on:
  push:
    branches: [ main, develop ]
  pull_request:
    branches: [ main ]

jobs:
  run_unit_tests:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout Repository
        uses: actions/checkout@v4

      - name: Initialize Python Environment
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
          cache: 'pip'

      - name: Install Project Dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements-dev.txt

      - name: Execute Pytest with Coverage
        run: |
          pytest --cov=core --cov-report=xml --cov-fail-under=80 tests/

      - name: Publish Coverage Report
        uses: codecov/codecov-action@v3
        with:
          files: ./coverage.xml
          fail_ci_if_error: true