상승 다이버전스 듀얼 타임프레임 퀀트 시스템 설계 명세서

1. 핵심 가설 및 시스템 개요

2. 매일 실행되는 동적 유니버스 스크리닝 알고리즘 (Universe Selection)

[미국 상장 주식 전체 풀]
       │
       ▼ (1단계: 유동성 필터)
[슬리피지 제로 유니버스]
       │
       ▼ (2단계: 재무 건전성 필터)
[부도 리스크 제로 우량주 풀]
       │
       ▼ (3단계: 스타일 및 인덱스 필터)
[최종 트레이딩 적격 유니버스 (EOD 확정)]

[필터 1단계] 유동성 필터 (Liquidity Filter)

[필터 2단계] 재무 건전성 필터 (Fundamental Quality Filter)

[필터 3단계] 스타일 및 인덱스 필터 (Style & Index Filter)

3. 기술적 지표 및 상승 다이버전스 수학적 정의

변수 정의

주가 저점 (Pivot Low) 정의

상승 다이버전스 판정 공식

4. 진입 및 리스크 관리 규칙 (Execution & Exit Rules)

진입 시점 (Entry Trigger)

손절 규칙 (Stop Loss)

익절 규칙 (Take Profit)

5. 자금 관리 및 포트폴리오 배분 규칙 (Position Sizing & Portfolio Allocation)

포트폴리오 한도 (Portfolio Cap)

단일 거래 리스크 배분 공식 (Position Sizing Formula)

6. 15년 백테스트 성과 요약 (2011-06-28 ~ 2026-06-28)

부록: 파이썬 백테스트 소스 코드 (yfinance 멀티인덱스 버그 디버깅 완결판)

import yfinance as yf
import pandas as pd
import numpy as np

# ==========================================
# 1. 시스템 파라미터 설정 (튜닝 가능)
# ==========================================
CANDIDATE_TICKERS = ["AAPL", "MSFT", "NVDA", "TSM", "AVGO", "MU", "AMD"]

START_DATE = "2011-06-28"
END_DATE = "2026-06-28"

RSI_PERIOD = 14
EMA_PERIOD = 20
ATR_PERIOD = 14
PIVOT_WINDOW = 5        # K=5 (저점 판정 기준)
MIN_GAP = 10            # W_min (두 저점 최소 간격)
MAX_GAP = 40            # W_max (두 저점 최대 간격)
OVERSOLD_LIMIT = 35     # OS (과매도 기준)

INITIAL_CAPITAL = 100000  # 초기 자본금 ($)
RISK_PER_TRADE = 0.01    # 단일 거래 최대 리스크 (총 자산의 1%)
MAX_PORTFOLIO_PCT = 0.10 # 단일 종목 최대 진입 비중 (10%)

# ==========================================
# 2. 동적 유니버스 스크리닝 필터 (yfinance 버그 방어)
# ==========================================
def filter_universe(ticker_list):
    valid_tickers = []
    print("\n=========================================")
    print(" [1단계] 동적 유니버스 스크리닝 시작")
    print("=========================================")
    
    for ticker in ticker_list:
        try:
            t = yf.Ticker(ticker)
            info = t.info
            
            market_cap = info.get("marketCap", 0)
            sector = info.get("sector", "")
            industry = info.get("industry", "")
            
            is_cap_ok = market_cap >= 10000000000
            is_sector_ok = (sector == "Technology") or ("Semiconductor" in industry)
            
            if is_cap_ok and is_sector_ok:
                valid_tickers.append(ticker)
                print(f"  [PASS] {ticker:<5} | 시총: ${market_cap/1e9:>6.1f}B | 섹터: {sector:<10}")
            else:
                print(f"  [FAIL] {ticker:<5} | 시총: ${market_cap/1e9:>6.1f}B | 섹터: {sector:<10}")
                
        except Exception as e:
            # API 제한 등으로 info 실패 시 백테스트 진행을 위해 기본 패스 처리 우회
            valid_tickers.append(ticker)
            print(f"  [PASS_BYPASS] {ticker:<5} (Metadata API 제한 우회)")
            
    print(f"\n➔ 최종 트레이딩 적격 유니버스: {valid_tickers}")
    print("=========================================\n")
    return valid_tickers

# ==========================================
# 3. 기술적 지표 계산 함수 (멀티인덱스 완벽 방어)
# ==========================================
def calculate_indicators(df):
    # yfinance 멀티 인덱스 강제 플래티닝 (핵심 디버깅 포인트)
    if isinstance(df.columns, pd.MultiIndex):
        df.columns = df.columns.get_level_values(0)
        
    # 데이터 타입 강제 1차원 Series 변환
    close_series = pd.Series(df['Close'].values.flatten(), index=df.index)
    high_series = pd.Series(df['High'].values.flatten(), index=df.index)
    low_series = pd.Series(df['Low'].values.flatten(), index=df.index)
    
    # RSI (Wilder's Smoothing)
    delta = close_series.diff()
    gain = (delta.where(delta > 0, 0)).copy()
    loss = (-delta.where(delta < 0, 0)).copy()
    
    avg_gain = gain.ewm(alpha=1/RSI_PERIOD, adjust=False).mean()
    avg_loss = loss.ewm(alpha=1/RSI_PERIOD, adjust=False).mean()
    
    rs = avg_gain / avg_loss
    df['RSI'] = (100 - (100 / (1 + rs))).values
    
    # EMA 20
    df['EMA20'] = close_series.ewm(span=EMA_PERIOD, adjust=False).mean().values
    
    # ATR 14
    high_low = high_series - low_series
    high_close = np.abs(high_series - close_series.shift())
    low_close = np.abs(low_series - close_series.shift())
    ranges = pd.concat([high_low, high_close, low_close], axis=1)
    true_range = ranges.max(axis=1)
    df['ATR'] = true_range.ewm(alpha=1/ATR_PERIOD, adjust=False).mean().values
    
    # 데이터 정제 (NaN 제거)
    df = df.dropna(subset=['RSI', 'EMA20', 'ATR'])
    return df

# ==========================================
# 4. Pivot Low (저점) 및 상승 다이버전스 감지
# ==========================================
def detect_bullish_divergence(df):
    df['Pivot_Low'] = False
    df['Divergence_Signal'] = False
    df['T1_Idx'] = -1
    df['T2_Idx'] = -1
    
    close_values = df['Close'].values
    rsi_values = df['RSI'].values
    
    # Pivot Low 감지 (좌우 K개 캔들 비교)
    for i in range(PIVOT_WINDOW, len(df) - PIVOT_WINDOW):
        current_val = close_values[i]
        left_sub = close_values[i-PIVOT_WINDOW:i]
        right_sub = close_values[i+1:i+PIVOT_WINDOW+1]
        
        if current_val == min(current_val, left_sub.min(), right_sub.min()):
            df.iloc[i, df.columns.get_loc('Pivot_Low')] = True
            
    # 상승 다이버전스 패턴 매칭
    pivot_indices = np.where(df['Pivot_Low'] == True)[0]
    
    for i in range(1, len(pivot_indices)):
        pos2 = pivot_indices[i] # 현재 바닥 (T2)
        pos1 = pivot_indices[i-1] # 과거 바닥 (T1)
        gap = pos2 - pos1
        
        # 조건 검증 (간격, 주가 저점 하락, RSI 저점 상승, 첫 바닥 과매도)
        if MIN_GAP <= gap <= MAX_GAP:
            if close_values[pos2] < close_values[pos1]:
                if rsi_values[pos2] > rsi_values[pos1]:
                    if rsi_values[pos1] <= OVERSOLD_LIMIT:
                        df.iloc[pos2, df.columns.get_loc('Divergence_Signal')] = True
                        df.iloc[pos2, df.columns.get_loc('T1_Idx')] = pos1
                        df.iloc[pos2, df.columns.get_loc('T2_Idx')] = pos2
                        
    return df

# ==========================================
# 5. 백테스트 시뮬레이터 실행 (조건문 완벽 작동 보장)
# ==========================================
def run_backtest(df):
    capital = INITIAL_CAPITAL
    position = 0
    entry_price = 0
    stop_loss = 0
    target_1 = 0
    half_exited = False
    trades = []
    
    active_signal = False
    signal_t2_price = 0
    signal_t1_pos = -1
    signal_t2_pos = -1
    
    close_vals = df['Close'].values
    high_vals = df['High'].values
    low_vals = df['Low'].values
    rsi_vals = df['RSI'].values
    ema_vals = df['EMA20'].values
    atr_vals = df['ATR'].values
    div_signals = df['Divergence_Signal'].values
    t1_indices = df['T1_Idx'].values
    t2_indices = df['T2_Idx'].values
    
    for i in range(len(df)):
        current_date = df.index[i]
        current_close = close_vals[i]
        current_high = high_vals[i]
        current_low = low_vals[i]
        current_rsi = rsi_vals[i]
        current_ema = ema_vals[i]
        current_atr = atr_vals[i]
        
        # 1) 신호 대기 상태
        if div_signals[i]:
            active_signal = True
            signal_t2_price = current_close
            signal_t1_pos = int(t1_indices[i])
            signal_t2_pos = int(t2_indices[i])
            
        # 2) 진입 (EMA 20 상향 돌파 시)
        if active_signal and position == 0:
            if current_close > current_ema:
                entry_price = current_close
                stop_loss = signal_t2_price - (1.5 * current_atr)
                target_1 = high_vals[signal_t1_pos:signal_t2_pos+1].max()
                
                risk_amount = capital * RISK_PER_TRADE
                loss_per_share = entry_price - stop_loss
                
                if loss_per_share > 0:
                    qty_by_risk = risk_amount / loss_per_share
                    qty_by_cap = (capital * MAX_PORTFOLIO_PCT) / entry_price
                    position = min(qty_by_risk, qty_by_cap)
                else:
                    position = (capital * MAX_PORTFOLIO_PCT) / entry_price
                    
                half_exited = False
                active_signal = False
                
                trades.append({
                    'Entry_Date': current_date,
                    'Entry_Price': entry_price,
                    'Stop_Loss': stop_loss,
                    'Target_1': target_1,
                    'Qty': position,
                    'Status': 'OPEN'
                })
                
        # 3) 포지션 보유 중 리스크/익절 관리 (실제 시뮬레이션 활성화)
        elif position > 0:
            trade_info = trades[-1]
            
            # A. 손절 처리
            if current_low < stop_loss:
                exit_price = min(current_close, stop_loss)
                pnl = (exit_price - entry_price) * position
                capital += pnl
                trade_info['Exit_Date'] = current_date
                trade_info['Exit_Price'] = exit_price
                trade_info['PnL'] = pnl
                trade_info['Return_Pct'] = (exit_price / entry_price - 1) * 100
                trade_info['Status'] = 'STOP_LOSS'
                position = 0
                
            # B. 1차 익절 (50% 물량)
            elif not half_exited and current_high >= target_1:
                exit_price = target_1
                pnl = (exit_price - entry_price) * (position * 0.5)
                capital += pnl
                position = position * 0.5
                half_exited = True
                trade_info['Half_Exit_Date'] = current_date
                trade_info['Half_Exit_Price'] = exit_price
                # 본전 보존을 위해 손절가를 진입가로 상향 조정 (Breakeven)
                stop_loss = entry_price 
                
            # C. 2차 익절 (RSI 70 이상 과열 진입 시)
            elif current_rsi >= 70:
                exit_price = current_close
                pnl = (exit_price - entry_price) * position
                capital += pnl
                trade_info['Exit_Date'] = current_date
                trade_info['Exit_Price'] = exit_price
                trade_info['PnL'] = trade_info.get('PnL', 0) + pnl
                trade_info['Return_Pct'] = (exit_price / entry_price - 1) * 100
                trade_info['Status'] = 'TAKE_PROFIT'
                position = 0
                
    return pd.DataFrame(trades), capital

# ==========================================
# 6. 실행 및 결과 출력
# ==========================================
if __name__ == "__main__":
    active_universe = filter_universe(CANDIDATE_TICKERS)
    
    for ticker in active_universe:
        data = yf.download(ticker, start=START_DATE, end=END_DATE)
        
        if not data.empty:
            data = calculate_indicators(data)
            data = detect_bullish_divergence(data)
            trades_df, final_capital = run_backtest(data)
            
            print(f"\n➔ [{ticker}] 15년 백테스트 최종 성과")
            print("-" * 40)
            print(f"  * 초기 자본금: ${INITIAL_CAPITAL:,.2f} ➔ 최종 자본금: ${final_capital:,.2f}")
            
            if len(trades_df) > 0:
                completed_trades = trades_df[trades_df['Status'] != 'OPEN']
                total_trades = len(completed_trades)
                winning_trades = len(completed_trades[completed_trades['Return_Pct'] > 0])
                win_rate = (winning_trades / total_trades) * 100 if total_trades > 0 else 0
                avg_return = completed_trades['Return_Pct'].mean()
                
                # MDD 계산
                trades_df['Cum_Capital'] = INITIAL_CAPITAL + trades_df['PnL'].copy().fillna(0).cumsum()
                cum_max = trades_df['Cum_Capital'].cummax()
                drawdown = (trades_df['Cum_Capital'] - cum_max) / cum_max * 100
                mdd = drawdown.min()
                
                # 평균 보유기간 계산
                hold_days = []
                for _, row in completed_trades.iterrows():
                    days = (row['Exit_Date'] - row['Entry_Date']).days
                    hold_days.append(days)
                avg_hold_days = np.mean(hold_days) if len(hold_days) > 0 else 0
                
                print(f"  * 총 거래 횟수: {total_trades}회")
                print(f"  * 평균 승률: {win_rate:.2f}%")
                print(f"  * 거래당 평균 수익률: {avg_return:.2f}%")
                print(f"  * 평균 보유기간: {avg_hold_days:.1f}일")
                print(f"  * 최대 낙폭 (MDD): {mdd:.2f}%")
            else:
                print("  * 조건에 맞는 거래 신호가 발생하지 않았습니다.")
        else:
            print(f"  * {ticker} 데이터를 가져오지 못했습니다.")