"""5% 익절 전략 백테스트 (손절 없음) 규칙: - 진입: 직전 포지션 청산 다음 캔들 종가 매수 - 청산: 고가 기준 +5% 달성 시 TP 가격으로 매도 - 손절: 없음 — 손실 중 무조건 보유 - 자본: 1,000,000 KRW (복리 없음 / 고정) - 캔들: 1h Output: - 종목별: 거래 수, TP 성공, 승률, 총 손익, 평균 보유시간, 최대 보유시간 - 현재 미결 포지션 현황 - 전략 한계 분석 """ import os as _os, sys as _sys _sys.path.insert(0, _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__)))) from __future__ import annotations import time from dataclasses import dataclass, field from typing import Optional import pyupbit import pandas as pd from datetime import datetime # ─── 파라미터 ────────────────────────────────────────────────── TICKERS = [ "KRW-WET", "KRW-FLOW", "KRW-BIO", "KRW-CYBER", "KRW-JTO", "KRW-AERO", "KRW-LAYER", "KRW-DKA", ] TP_PCT = 0.05 # 익절 목표 5% CAPITAL = 1_000_000 # 거래당 고정 자본 (복리 없음) INTERVAL = "minute60" COUNT = 1440 # 60일치 1h 캔들 # ─── 데이터 클래스 ────────────────────────────────────────────── @dataclass class Trade: entry_time: datetime exit_time: datetime entry_px: float exit_px: float pnl: float holding_h: int status: str # TP | HOLDING_LOSS | HOLDING_PROFIT # ─── 백테스트 ────────────────────────────────────────────────── def backtest_ticker(ticker: str) -> Optional[dict]: df = pyupbit.get_ohlcv(ticker, interval=INTERVAL, count=COUNT) if df is None or len(df) < 20: return None trades: list[Trade] = [] i = 0 while i < len(df) - 1: # 매수: 현재 캔들 종가 entry_px = float(df["close"].iloc[i]) entry_time = df.index[i] tp_px = entry_px * (1 + TP_PCT) hit = False for j in range(i + 1, len(df)): if float(df["high"].iloc[j]) >= tp_px: trades.append(Trade( entry_time = entry_time, exit_time = df.index[j], entry_px = entry_px, exit_px = tp_px, pnl = CAPITAL * TP_PCT, holding_h = j - i, status = "TP", )) i = j # 다음 진입은 j 캔들 종가 hit = True break if not hit: # 데이터 끝까지 TP 미달성 last_px = float(df["close"].iloc[-1]) pnl = CAPITAL * (last_px - entry_px) / entry_px status = "HOLDING_PROFIT" if pnl >= 0 else "HOLDING_LOSS" trades.append(Trade( entry_time = entry_time, exit_time = df.index[-1], entry_px = entry_px, exit_px = last_px, pnl = pnl, holding_h = len(df) - i - 1, status = status, )) break closed = [t for t in trades if t.status == "TP"] holding = [t for t in trades if t.status != "TP"] total_pnl = sum(t.pnl for t in trades) realized_pnl = sum(t.pnl for t in closed) return { "ticker": ticker, "n_total": len(trades), "n_tp": len(closed), "win_rate": len(closed) / len(trades) * 100 if trades else 0, "realized_pnl": realized_pnl, "total_pnl": total_pnl, "avg_h": sum(t.holding_h for t in closed) / len(closed) if closed else 0, "max_h": max(t.holding_h for t in trades) if trades else 0, "holding": holding, "trades": trades, } # ─── 메인 ────────────────────────────────────────────────────── def main() -> None: print("=" * 62) print(" 5% 익절 전략 백테스트 (손절 없음)") print(f" 기간: 최근 {COUNT}h ({COUNT//24}일) | 자본: {CAPITAL:,}원 | TP: {TP_PCT*100:.0f}%") print("=" * 62) all_results = [] for ticker in TICKERS: r = backtest_ticker(ticker) time.sleep(0.1) if r: all_results.append(r) # ─── 종목별 출력 ─────────────────────────────────────────── print(f"\n{'종목':<12} {'거래':<5} {'TP':>5} {'승률':>7} {'실현손익':>12} " f"{'미결손익':>12} {'평균보유':>8} {'최장보유':>8}") print("-" * 80) total_realized = 0 total_open_pnl = 0 for r in all_results: h_pnl = sum(t.pnl for t in r["holding"]) hold_str = "" if r["holding"]: h = r["holding"][0] mark = "⚠️ " if h.status == "HOLDING_LOSS" else "📈" hold_str = f" {mark}{h.pnl:+,.0f}원 ({h.holding_h}h 보유중)" print(f"{r['ticker']:<12} {r['n_total']:<5} {r['n_tp']:>5} " f"{r['win_rate']:>6.1f}% {r['realized_pnl']:>+12,.0f} " f"{h_pnl:>+12,.0f} {r['avg_h']:>7.1f}h {r['max_h']:>7}h" + hold_str) total_realized += r["realized_pnl"] total_open_pnl += h_pnl print("-" * 80) print(f"{'합계':<12} {'':5} {'':>5} {'':>7} {total_realized:>+12,.0f} " f"{total_open_pnl:>+12,.0f}") # ─── 미결 포지션 상세 ────────────────────────────────────── open_trades = [(r["ticker"], h) for r in all_results for h in r["holding"]] if open_trades: print("\n▶ 현재 미결 포지션") print(f" {'종목':<12} {'진입시각':<20} {'진입가':>12} {'현재가':>12} {'수익률':>8} {'보유시간':>8}") for ticker, h in open_trades: chg = (h.exit_px - h.entry_px) / h.entry_px * 100 print(f" {ticker:<12} {str(h.entry_time)[:16]:<20} " f"{h.entry_px:>12,.2f} {h.exit_px:>12,.2f} " f"{chg:>+7.2f}% {h.holding_h:>7}h") # ─── 최장 잠김 분석 ──────────────────────────────────────── all_trades_flat = [(r["ticker"], t) for r in all_results for t in r["trades"]] if all_trades_flat: worst = max(all_trades_flat, key=lambda x: x[1].holding_h) print(f"\n▶ 최장 자본 잠김") print(f" {worst[0]}: {worst[1].holding_h}h ({worst[1].holding_h/24:.1f}일) " f"진입 {str(worst[1].entry_time)[:16]} → " f"{'TP 달성' if worst[1].status == 'TP' else '미결'}") # ─── 전략 평가 ───────────────────────────────────────────── if all_results: total_trades = sum(r["n_total"] for r in all_results) total_tp = sum(r["n_tp"] for r in all_results) avg_hold_all = [r["avg_h"] for r in all_results if r["avg_h"] > 0] still_holding = sum(1 for r in all_results if r["holding"]) print("\n▶ 종합 평가") print(f" 전체 거래: {total_trades}건 | TP 성공: {total_tp}건 | " f"승률: {total_tp/total_trades*100:.1f}%") if avg_hold_all: print(f" TP까지 평균 보유: {sum(avg_hold_all)/len(avg_hold_all):.1f}h " f"({sum(avg_hold_all)/len(avg_hold_all)/24:.1f}일)") print(f" 현재 미결 포지션: {still_holding}개 종목") print(f" 총 실현 손익: {total_realized:+,.0f}원") print(f" 총 평가 손익: {total_realized + total_open_pnl:+,.0f}원") print("\n▶ 전략 한계") if still_holding > 0: print(f" ⚠️ {still_holding}개 종목이 TP 미달성 상태로 자본이 잠겨있습니다.") print(" ⚠️ 손절 없음 → 하락장에서 자본이 무한정 묶일 수 있습니다.") print(" ⚠️ 기회비용: 잠긴 자본으로 다른 거래 불가.") print(" 💡 개선안: 최대 보유기간(시간) 제한 또는 % 손절 추가 권장.") if __name__ == "__main__": main()