"""3월 백테스트 — 현재 tick_trader 설정 기준 (1분봉 시뮬레이션) 20초봉 → 1분봉 환산: VOL_LOOKBACK 61(20s) → 20(1min) 트레일/손절은 가격 기반이라 동일 타임아웃 4h = 240분 """ import time from collections import defaultdict, deque from datetime import datetime, timedelta import pyupbit from dotenv import load_dotenv load_dotenv() # ── 설정 (현재 tick_trader 기준) ── TICKERS = [ 'KRW-ETH', 'KRW-XRP', 'KRW-SOL', 'KRW-DOGE', 'KRW-SIGN', 'KRW-BARD', 'KRW-KITE', 'KRW-CFG', 'KRW-SXP', 'KRW-ARDR', ] VOL_LOOKBACK = 20 # 1분봉 기준 VOL_MIN = 5.0 VOL_KRW_MIN = 5_000_000 # 1분봉 거래대금 (20초봉 500만 × 3) SPREAD_MIN = 0.3 # 횡보 필터: 5봉 변동폭 % (15봉/3) HIGHPOS_BARS = 30 # 고점 필터: 30봉(30분) HIGHPOS_THR = 0.9 HIGHPOS_MOVE = 1.0 # 구간 변동 1%+ TRAIL_PCT = 0.015 MIN_PROFIT_PCT = 0.005 STOP_LOSS_PCT = 0.02 TIMEOUT_BARS = 240 # 4시간 MAX_POS = 3 PER_POS = 33_333 FEE = 0.0005 def calc_vr(bars, idx): start = max(0, idx - VOL_LOOKBACK) end = max(0, idx - 1) baseline = sorted(bars[i]['volume'] for i in range(start, end)) if not baseline: return 0.0 trim = max(1, len(baseline) // 10) trimmed = baseline[:len(baseline) - trim] if not trimmed: return 0.0 avg = sum(trimmed) / len(trimmed) return bars[idx]['volume'] / avg if avg > 0 else 0.0 def load_data(ticker, start_date='2026-03-01', end_date='2026-03-07'): """Oracle backtest_ohlcv에서 1분봉 로드, 없으면 pyupbit 페이징.""" import os bars = [] # Oracle DB 시도 try: import oracledb kwargs = dict( user=os.environ['ORACLE_USER'], password=os.environ['ORACLE_PASSWORD'], dsn=os.environ['ORACLE_DSN'], ) if w := os.environ.get('ORACLE_WALLET'): kwargs['config_dir'] = w conn = oracledb.connect(**kwargs) cur = conn.cursor() cur.execute(""" SELECT ts, open_p, high_p, low_p, close_p, volume_p FROM backtest_ohlcv WHERE ticker = :t AND interval_cd = 'minute1' AND ts >= TO_TIMESTAMP(:s, 'YYYY-MM-DD') AND ts < TO_TIMESTAMP(:e, 'YYYY-MM-DD') ORDER BY ts """, {'t': ticker, 's': start_date, 'e': end_date}) rows = cur.fetchall() conn.close() if rows: for ts, o, h, l, c, v in rows: bars.append({'ts': ts, 'open': float(o), 'high': float(h), 'low': float(l), 'close': float(c), 'volume': float(v)}) return bars except Exception as e: pass # Fallback: pyupbit 페이징 import pandas as pd all_df = [] to = end_date + ' 23:59:59' for _ in range(60): df = pyupbit.get_ohlcv(ticker, interval='minute1', count=200, to=to) if df is None or df.empty: break all_df.append(df) earliest = df.index[0] if str(earliest)[:10] < start_date: break to = (earliest - timedelta(minutes=1)).strftime('%Y-%m-%d %H:%M:%S') time.sleep(0.12) if not all_df: return [] merged = pd.concat(all_df).sort_index() merged = merged[~merged.index.duplicated(keep='first')] merged = merged[start_date:end_date] for ts, row in merged.iterrows(): bars.append({'ts': ts, 'open': row['open'], 'high': row['high'], 'low': row['low'], 'close': row['close'], 'volume': row['volume']}) return bars def run_backtest(): print("=" * 80) print("3월 백테스트 (1분봉 시뮬레이션)") print(f"종목: {len(TICKERS)}개 | VOL >= {VOL_MIN}x | 거래대금 >= {VOL_KRW_MIN/1e6:.0f}M") print(f"트레일: 고점-{TRAIL_PCT*100:.1f}% (최소 +{MIN_PROFIT_PCT*100:.1f}%) | 손절: -{STOP_LOSS_PCT*100:.1f}% | 타임아웃: {TIMEOUT_BARS//60}h") print(f"포지션: 최대 {MAX_POS} | 종목당 {PER_POS:,}원") print("=" * 80) # 데이터 로드 all_bars = {} for t in TICKERS: print(f" 로딩 {t}...", end=' ') bars = load_data(t) print(f"{len(bars)}봉") all_bars[t] = bars time.sleep(0.2) # 모든 종목의 타임스탬프를 합쳐서 시간순 이벤트 생성 events = [] for t in TICKERS: for i, b in enumerate(all_bars[t]): events.append((b['ts'], t, i)) events.sort(key=lambda x: x[0]) # 시뮬레이션 positions = {} trades = [] signals_count = 0 skipped_spread = 0 skipped_highpos = 0 skipped_maxpos = 0 skipped_greens = 0 for ts, ticker, idx in events: bar_list = all_bars[ticker] b = bar_list[idx] # ── 포지션 관리 (매 봉마다) ── for t in list(positions.keys()): pos = positions[t] # 해당 종목의 현재 봉 찾기 t_bars = all_bars[t] t_idx = pos['last_idx'] # 시간이 현재 이벤트 이후인 봉이 있으면 업데이트 while t_idx + 1 < len(t_bars) and t_bars[t_idx + 1]['ts'] <= ts: t_idx += 1 pos['last_idx'] = t_idx cur_bar = t_bars[t_idx] cur_price = cur_bar['close'] cur_high = cur_bar['high'] cur_low = cur_bar['low'] # peak 갱신 (봉 고가 기준) pos['running_peak'] = max(pos['running_peak'], cur_high) bars_held = t_idx - pos['entry_idx'] # 봉 내 손절 체크 (저가 기준) loss = (cur_low - pos['entry_price']) / pos['entry_price'] if loss <= -STOP_LOSS_PCT: exit_p = pos['entry_price'] * (1 - STOP_LOSS_PCT) pnl = -STOP_LOSS_PCT * 100 krw = PER_POS * (-STOP_LOSS_PCT) - PER_POS * FEE * 2 trades.append({ 'ticker': t, 'entry_ts': pos['entry_ts'], 'exit_ts': cur_bar['ts'], 'entry': pos['entry_price'], 'exit': exit_p, 'pnl_pct': pnl, 'pnl_krw': krw, 'bars': bars_held, 'tag': 'stoploss', 'peak': pos['running_peak'], 'vol_ratio': pos.get('vol_ratio', 0), 'prev_greens': pos.get('prev_greens', 0), 'bar_rise_pct': pos.get('bar_rise_pct', 0), }) del positions[t] continue # 트레일링 체크 (종가 기준) profit = (cur_price - pos['entry_price']) / pos['entry_price'] drop = (pos['running_peak'] - cur_price) / pos['running_peak'] if pos['running_peak'] > 0 else 0 if profit >= MIN_PROFIT_PCT and drop >= TRAIL_PCT: pnl = profit * 100 krw = PER_POS * profit - PER_POS * FEE * 2 trades.append({ 'ticker': t, 'entry_ts': pos['entry_ts'], 'exit_ts': cur_bar['ts'], 'entry': pos['entry_price'], 'exit': cur_price, 'pnl_pct': pnl, 'pnl_krw': krw, 'bars': bars_held, 'tag': 'trail', 'peak': pos['running_peak'], 'vol_ratio': pos.get('vol_ratio', 0), 'prev_greens': pos.get('prev_greens', 0), 'bar_rise_pct': pos.get('bar_rise_pct', 0), }) del positions[t] continue # 타임아웃 if bars_held >= TIMEOUT_BARS: pnl = profit * 100 krw = PER_POS * profit - PER_POS * FEE * 2 trades.append({ 'ticker': t, 'entry_ts': pos['entry_ts'], 'exit_ts': cur_bar['ts'], 'entry': pos['entry_price'], 'exit': cur_price, 'pnl_pct': pnl, 'pnl_krw': krw, 'bars': bars_held, 'tag': 'timeout', 'peak': pos['running_peak'], 'vol_ratio': pos.get('vol_ratio', 0), 'prev_greens': pos.get('prev_greens', 0), 'bar_rise_pct': pos.get('bar_rise_pct', 0), }) del positions[t] continue # ── 시그널 감지 ── if idx < VOL_LOOKBACK + 3: continue if ticker in positions: continue if len(positions) >= MAX_POS: continue # 양봉 if b['close'] <= b['open']: continue # 거래량 배수 vr = calc_vr(bar_list, idx) if vr < VOL_MIN: continue # 거래대금 하한 bar_krw = b['close'] * b['volume'] if bar_krw < VOL_KRW_MIN: continue # 횡보 필터 (5봉) recent = bar_list[idx-4:idx+1] p_high = max(x['high'] for x in recent) p_low = min(x['low'] for x in recent) if p_low > 0: spread = (p_high - p_low) / p_low * 100 if spread < SPREAD_MIN: skipped_spread += 1 continue # 고점 필터 long_start = max(0, idx - HIGHPOS_BARS) long_bars = bar_list[long_start:idx+1] l_high = max(x['high'] for x in long_bars) l_low = min(x['low'] for x in long_bars) if l_high > l_low: pos_in_range = (b['close'] - l_low) / (l_high - l_low) move_pct = (l_high - l_low) / l_low * 100 if pos_in_range > HIGHPOS_THR and move_pct > HIGHPOS_MOVE: skipped_highpos += 1 continue # 진입 전 연속 양봉 수 prev_greens = 0 for k in range(idx-1, max(idx-10, 0), -1): if bar_list[k]['close'] > bar_list[k]['open']: prev_greens += 1 else: break # 연속 양봉 필터: 2봉 이상 연속 양봉 필요 if prev_greens < 2: skipped_greens += 1 continue signals_count += 1 # 진입봉 상승폭 bar_rise_pct = (b['close'] - b['open']) / b['open'] * 100 if b['open'] > 0 else 0 # 매수 (현재가) entry_price = b['close'] positions[ticker] = { 'entry_price': entry_price, 'entry_ts': b['ts'], 'entry_idx': idx, 'last_idx': idx, 'running_peak': b['high'], 'vol_ratio': vr, 'prev_greens': prev_greens, 'bar_rise_pct': bar_rise_pct, } # 미청산 포지션 강제 종료 for t, pos in list(positions.items()): t_bars = all_bars[t] last_bar = t_bars[-1] cur_price = last_bar['close'] profit = (cur_price - pos['entry_price']) / pos['entry_price'] krw = PER_POS * profit - PER_POS * FEE * 2 trades.append({ 'ticker': t, 'entry_ts': pos['entry_ts'], 'exit_ts': last_bar['ts'], 'entry': pos['entry_price'], 'exit': cur_price, 'pnl_pct': profit * 100, 'pnl_krw': krw, 'bars': len(t_bars) - 1 - pos['entry_idx'], 'tag': 'open', 'peak': pos['running_peak'], 'vol_ratio': pos.get('vol_ratio', 0), 'prev_greens': pos.get('prev_greens', 0), 'bar_rise_pct': pos.get('bar_rise_pct', 0), }) # ── 결과 출력 ── print("\n" + "=" * 80) print("결과") print("=" * 80) print(f"\n시그널 발생: {signals_count}건") print(f"필터 제외: 횡보 {skipped_spread}건, 고점 {skipped_highpos}건, 연속양봉<2 {skipped_greens}건") print(f"총 거래: {len(trades)}건\n") if not trades: print("거래 없음") return wins = [t for t in trades if t['pnl_pct'] > 0] losses = [t for t in trades if t['pnl_pct'] <= 0] total_krw = sum(t['pnl_krw'] for t in trades) win_rate = len(wins) / len(trades) * 100 print(f"승률: {win_rate:.1f}% ({len(wins)}승 {len(losses)}패)") print(f"총 PNL: {total_krw:+,.0f}원") print(f"평균 PNL: {sum(t['pnl_pct'] for t in trades)/len(trades):+.2f}%") if wins: print(f"평균 수익 (승): +{sum(t['pnl_pct'] for t in wins)/len(wins):.2f}%") if losses: print(f"평균 손실 (패): {sum(t['pnl_pct'] for t in losses)/len(losses):+.2f}%") # 청산 유형별 by_tag = defaultdict(list) for t in trades: by_tag[t['tag']].append(t) print("\n[청산 유형별]") for tag, tlist in sorted(by_tag.items()): avg_pnl = sum(t['pnl_pct'] for t in tlist) / len(tlist) total = sum(t['pnl_krw'] for t in tlist) print(f" {tag:10s}: {len(tlist):3d}건 avg {avg_pnl:+.2f}% 합계 {total:+,.0f}원") # 종목별 by_ticker = defaultdict(list) for t in trades: by_ticker[t['ticker']].append(t) print("\n[종목별]") for ticker in TICKERS: tlist = by_ticker.get(ticker, []) if not tlist: continue w = sum(1 for t in tlist if t['pnl_pct'] > 0) total = sum(t['pnl_krw'] for t in tlist) avg = sum(t['pnl_pct'] for t in tlist) / len(tlist) print(f" {ticker:12s}: {len(tlist):3d}건 {w}승{len(tlist)-w}패 avg {avg:+.2f}% 합계 {total:+,.0f}원") # 승패별 시그널 특성 분석 print("\n[승패별 시그널 특성]") win_vrs = [t['vol_ratio'] for t in trades if t['pnl_pct'] > 0] loss_vrs = [t['vol_ratio'] for t in trades if t['pnl_pct'] <= 0] print(f" 승리 거래량비: 평균 {sum(win_vrs)/len(win_vrs):.1f}x 중간값 {sorted(win_vrs)[len(win_vrs)//2]:.1f}x") print(f" 패배 거래량비: 평균 {sum(loss_vrs)/len(loss_vrs):.1f}x 중간값 {sorted(loss_vrs)[len(loss_vrs)//2]:.1f}x") # 거래량비 구간별 승률 print("\n[거래량비 구간별 승률]") for lo, hi in [(5,7),(7,10),(10,15),(15,30),(30,999)]: subset = [t for t in trades if lo <= t['vol_ratio'] < hi] if not subset: continue w = sum(1 for t in subset if t['pnl_pct'] > 0) avg_pnl = sum(t['pnl_pct'] for t in subset) / len(subset) print(f" {lo:3d}~{hi:3d}x: {len(subset):3d}건 승률 {w/len(subset)*100:.0f}% avg {avg_pnl:+.2f}%") # 연속 양봉 여부별 승률 print("\n[진입 전 연속 양봉 수별 승률]") for n_green in [1, 2, 3]: subset = [t for t in trades if t.get('prev_greens', 0) >= n_green] if not subset: continue w = sum(1 for t in subset if t['pnl_pct'] > 0) avg_pnl = sum(t['pnl_pct'] for t in subset) / len(subset) print(f" {n_green}봉+ 연속양봉: {len(subset):3d}건 승률 {w/len(subset)*100:.0f}% avg {avg_pnl:+.2f}%") # 진입봉 상승폭별 승률 print("\n[진입봉 상승폭별 승률]") for lo, hi in [(0,0.3),(0.3,0.7),(0.7,1.5),(1.5,100)]: subset = [t for t in trades if lo <= t.get('bar_rise_pct', 0) < hi] if not subset: continue w = sum(1 for t in subset if t['pnl_pct'] > 0) avg_pnl = sum(t['pnl_pct'] for t in subset) / len(subset) print(f" {lo:.1f}~{hi:.1f}%: {len(subset):3d}건 승률 {w/len(subset)*100:.0f}% avg {avg_pnl:+.2f}%") # 상위 거래 trades_sorted = sorted(trades, key=lambda x: -x['pnl_pct']) print("\n[TOP 10 수익]") for t in trades_sorted[:10]: peak_pnl = (t['peak'] - t['entry']) / t['entry'] * 100 print(f" {t['ticker']:12s} {str(t['entry_ts'])[5:16]} {t['entry']:>12,.1f} → {t['exit']:>12,.1f} " f"PNL {t['pnl_pct']:+.2f}% 고점 +{peak_pnl:.1f}% {t['bars']}봉 [{t['tag']}]") print("\n[WORST 10 손실]") for t in trades_sorted[-10:]: print(f" {t['ticker']:12s} {str(t['entry_ts'])[5:16]} {t['entry']:>12,.1f} → {t['exit']:>12,.1f} " f"PNL {t['pnl_pct']:+.2f}% {t['bars']}봉 [{t['tag']}]") # 일별 PNL by_date = defaultdict(float) by_date_cnt = defaultdict(int) for t in trades: d = str(t['exit_ts'])[:10] by_date[d] += t['pnl_krw'] by_date_cnt[d] += 1 print("\n[일별 PNL]") for d in sorted(by_date.keys()): print(f" {d}: {by_date_cnt[d]:3d}건 {by_date[d]:+,.0f}원") if __name__ == '__main__': run_backtest()