[Lap 2] [모듈 1: DataPortal] 미래 데이터를 보면 죽는 시점 포털

# 흔히 저지르는 치명적 실수
def get_daily_signal(df, today):
    # 오늘까지의 데이터를 슬라이싱한다고 믿음
    history = df.loc[:today]
    signal = calculate_momentum(history)
    return signal

# tests/test_pit_data_portal.py
import pytest
from datetime import date
from quant_backtester.data.portal import PITDataPortal
from quant_backtester.errors import LookAheadBiasError

def test_cannot_query_future_date_raises_lookahead_error(mock_duckdb_conn):
    """
    as_of_date가 2024-03-15일 때, 
    2024-03-16일 이후 데이터를 조회하려 하면 즉시 LookAheadBiasError가 터져야 한다.
    """
    portal = PITDataPortal(conn=mock_duckdb_conn, as_of_date=date(2024, 3, 15))
    
    # 1. 과거 데이터 조회: 정상 통과해야 함
    past_data = portal.get_history(ticker="AAPL", end_date=date(2024, 3, 15), lookback=20)
    assert len(past_data) == 20
    assert past_data.index.max() <= date(2024, 3, 15)

    # 2. 내일 데이터 조회 시도: 반드시 예외 발생
    with pytest.raises(LookAheadBiasError) as exc_info:
        portal.get_history(ticker="AAPL", end_date=date(2024, 3, 16), lookback=20)
    
    assert "Look-Ahead Bias Detected!" in str(exc_info.value)
    assert "Requested date 2024-03-16 is ahead of current as_of_date 2024-03-15" in str(exc_info.value)

def test_financial_statement_uses_filed_date_not_period_end(mock_duckdb_conn):
    """
    2023 Q3 재무제표는 2023-09-30 기준이지만 실제 공시는 2023-11-02에 이루어졌다.
    as_of_date가 2023-10-15일 때 이 재무제표는 조회되어서는 안 된다.
    """
    portal = PITDataPortal(conn=mock_duckdb_conn, as_of_date=date(2023, 10, 15))
    
    fundamentals = portal.get_latest_fundamentals(ticker="MSFT")
    
    # 2023-10-15 시점에서는 Q3 실적이 아니라 Q2 실적만 잡혀야 함!
    assert fundamentals['fiscal_quarter'] == "2023Q2"

$ pytest tests/test_pit_data_portal.py
============================= FAILURES =============================
ImportError: cannot import name 'PITDataPortal' from 'quant_backtester.data.portal'
========================= 2 failed in 0.08s =========================

# quant_backtester/data/portal.py
from datetime import date
import duckdb
import pandas as pd
from quant_backtester.errors import LookAheadBiasError

class PITDataPortal:
    """
    Point-In-Time 시점 격리 데이터 포털.
    주어진 as_of_date 이후의 미래 데이터 접근을 물리적으로 차단한다.
    """
    def __init__(self, conn: duckdb.DuckDBPyConnection, as_of_date: date):
        self._conn = conn
        self._as_of_date = as_of_date

    @property
    def as_of_date(self) -> date:
        return self._as_of_date

    def get_history(self, ticker: str, end_date: date, lookback: int) -> pd.DataFrame:
        # 1. 시점 침범 여부 엄격 검사 (Hard Gate)
        if end_date > self._as_of_date:
            raise LookAheadBiasError(
                f"💥 Look-Ahead Bias Detected! "
                f"Requested date {end_date} is ahead of current as_of_date {self._as_of_date}. "
                f"Strategy tried to peek into future market data!"
            )
        
        # 2. DuckDB 파라미터 쿼리로 시점 이하 데이터만 슬라이싱
        query = """
            SELECT date, ticker, open, high, low, close, volume, adj_close
            FROM daily_prices
            WHERE ticker = ? AND date <= ?
            ORDER BY date DESC
            LIMIT ?
        """
        df = self._conn.execute(query, [ticker, end_date, lookback]).df()
        
        # 날짜 오름차순 정렬 및 DatetimeIndex 지정
        df['date'] = pd.to_datetime(df['date'])
        df = df.sort_values('date').set_index('date')
        return df

    def get_latest_fundamentals(self, ticker: str) -> dict:
        """
        회계 기준일(period_end)이 아닌 실제 공시일(filed_date) 기준으로만 재무제표 조회
        """
        query = """
            SELECT *
            FROM company_financials
            WHERE ticker = ? AND filed_date <= ?
            ORDER BY filed_date DESC
            LIMIT 1
        """
        result = self._conn.execute(query, [ticker, self._as_of_date]).fetchone()
        if not result:
            return {}
        
        columns = [desc[0] for desc in self._conn.description]
        return dict(zip(columns, result))

$ pytest tests/test_pit_data_portal.py -v
tests/test_pit_data_portal.py::test_cannot_query_future_date_raises_lookahead_error PASSED [ 50%]
tests/test_pit_data_portal.py::test_financial_statement_uses_filed_date_not_period_end PASSED [100%]

========================= 2 passed in 0.12s =========================

# 35년 4,400만 행 Parquet 테이블 대상 1,000회 연속 PIT 쿼리 벤치마크
import time

start = time.perf_counter()
for _ in range(1000):
    portal.get_history("AAPL", end_date=date(2024, 3, 15), lookback=250)
elapsed = time.perf_counter() - start

print(f"1회 조회당 평균 소요 시간: {elapsed / 1000 * 1000:.3f} ms")
# 출력: 1회 조회당 평균 소요 시간: 1.842 ms