"""공포탐욕지수(F&G) 조건별 백테스트 질문: "F&G 수치에 따라 진입 조건을 제한하면 성과가 나아지는가?" 테스트 구간: A. 필터 없음 (baseline) B. F&G > 25 (Extreme Fear 구간 제외) C. F&G > 45 (Fear 이하 구간 제외, Neutral 이상) D. F&G > 55 (Greed 이상만 진입) E. F&G 20~45 (Fear 구간만 진입 – 역발상 매수) 진입 전략: 모멘텀 (거래량 급증 + 횡보 → +3% 상승 시 매수) 청산: 트레일링 스탑 1.5% + 타임 스탑 24h/+3% """ 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 json, time, sys, datetime, urllib.request import pyupbit import pandas as pd from collections import defaultdict from dataclasses import dataclass # ── 종목 ────────────────────────────────────────────────────── TICKERS = [ "KRW-BTC", "KRW-ETH", "KRW-XRP", "KRW-SOL", "KRW-DOGE", "KRW-ADA", "KRW-DOT", "KRW-NEAR", "KRW-AVAX", "KRW-LINK", "KRW-SUI", "KRW-HBAR", "KRW-TRX", "KRW-VIRTUAL", "KRW-SXP", "KRW-CFG", "KRW-HOLO", "KRW-ENSO", "KRW-KAVA", "KRW-KNC", "KRW-OM", "KRW-STEEM", "KRW-ORBS", ] # ── 공통 파라미터 ───────────────────────────────────────────── VOL_MULT = 2.0 QUIET_2H = 2.0 SIG_TO_H = 8 MOM_THR = 3.0 SIG_CANCEL = 3.0 TRAIL_STOP = 0.015 # 1.5% (앞 테스트에서 최적) TIME_H = 24 TIME_MIN = 3.0 # ── F&G 데이터 로드 ─────────────────────────────────────────── def load_fng(days: int = 365) -> dict[str, int]: """날짜(YYYY-MM-DD) → F&G 지수 딕셔너리 반환.""" url = f"https://api.alternative.me/fng/?limit={days}&format=json" with urllib.request.urlopen(url, timeout=10) as r: data = json.loads(r.read()) result = {} for d in data["data"]: dt_str = datetime.datetime.fromtimestamp(int(d["timestamp"])).strftime("%Y-%m-%d") result[dt_str] = int(d["value"]) return result def fng_at(fng_map: dict, ts: pd.Timestamp) -> int: """타임스탬프에 해당하는 F&G 값 반환. 없으면 50(Neutral) 반환.""" return fng_map.get(ts.strftime("%Y-%m-%d"), 50) # ── F&G 필터 정의 ───────────────────────────────────────────── CONFIGS = [ (None, None, "필터 없음 (baseline)"), (26, None, "F&G > 25 (Extreme Fear 제외)"), (46, None, "F&G > 45 (Neutral 이상)"), (56, None, "F&G > 55 (Greed 이상)"), (20, 45, "F&G 20~45 (Fear 역발상 매수)"), ] # ── 시뮬레이션 ──────────────────────────────────────────────── @dataclass class Trade: pnl: float h: int fng: int exit: str def simulate(df, fng_map: dict, fng_lo=None, fng_hi=None) -> list[Trade]: closes = df["close"].values vols = df["volume"].values idx = df.index trades: list[Trade] = [] sig_px = sig_i = None pos_buy = pos_peak = pos_i = pos_fng = None for i in range(7, len(closes) - max(TIME_H + 4, 10)): # ── 포지션 관리 ────────────────────────────────────── if pos_buy is not None: cur = closes[i] if cur > pos_peak: pos_peak = cur if (pos_peak - cur) / pos_peak >= TRAIL_STOP: pnl = (cur - pos_buy) / pos_buy * 100 trades.append(Trade(pnl, i - pos_i, pos_fng, "trail")) pos_buy = pos_peak = pos_i = pos_fng = sig_px = sig_i = None continue if i - pos_i >= TIME_H: pnl = (cur - pos_buy) / pos_buy * 100 if pnl < TIME_MIN: trades.append(Trade(pnl, i - pos_i, pos_fng, "time")) pos_buy = pos_peak = pos_i = pos_fng = sig_px = sig_i = None continue continue # ── 신호 유효성 ────────────────────────────────────── if sig_px is not None: if i - sig_i > SIG_TO_H: sig_px = sig_i = None elif (closes[i] - sig_px) / sig_px * 100 < -SIG_CANCEL: sig_px = sig_i = None # ── 축적 감지 ───────────────────────────────────────── if sig_px is None: vol_avg = vols[i - 6:i - 1].mean() if vol_avg <= 0: continue if vols[i - 1] / vol_avg >= VOL_MULT: if abs(closes[i] - closes[i - 2]) / closes[i - 2] * 100 < QUIET_2H: sig_px = closes[i] sig_i = i continue # ── F&G 필터 체크 ───────────────────────────────────── fng_val = fng_at(fng_map, idx[i]) if fng_lo is not None and fng_val < fng_lo: continue if fng_hi is not None and fng_val > fng_hi: continue # ── 모멘텀 진입 ─────────────────────────────────────── if (closes[i] - sig_px) / sig_px * 100 >= MOM_THR: pos_buy = pos_peak = closes[i] pos_i = i pos_fng = fng_val sig_px = sig_i = None return trades def stats(trades: list[Trade]) -> dict: if not trades: return dict(n=0, wr=0, avg_pnl=0, total_pnl=0, rr=0, avg_win=0, avg_loss=0, avg_fng=0, max_dd=0) wins = [t for t in trades if t.pnl > 0] losses = [t for t in trades if t.pnl <= 0] aw = sum(t.pnl for t in wins) / len(wins) if wins else 0 al = sum(t.pnl for t in losses) / len(losses) if losses else 0 rr = abs(aw / al) if al else 0 cum = 0.0; pk = 0.0; max_dd = 0.0 for t in trades: cum += t.pnl if cum > pk: pk = cum if pk - cum > max_dd: max_dd = pk - cum return dict( n = len(trades), wr = len(wins) / len(trades) * 100, avg_pnl = sum(t.pnl for t in trades) / len(trades), total_pnl= sum(t.pnl for t in trades), rr = rr, avg_win = aw, avg_loss = al, avg_fng = sum(t.fng for t in trades) / len(trades), max_dd = max_dd, ) def main() -> None: print("F&G 데이터 로드 중...") fng_map = load_fng(365) print(f" {len(fng_map)}일치 F&G 데이터 로드 완료\n") print(f"종목 데이터 수집 중 ({len(TICKERS)}개, 60일 1h 캔들)...") datasets: dict = {} for i, tk in enumerate(TICKERS): try: df = pyupbit.get_ohlcv(tk, interval="minute60", count=1440) if df is not None and len(df) > 50: datasets[tk] = df sys.stderr.write(f"\r {i+1}/{len(TICKERS)} {tk} ") time.sleep(0.08) except Exception: pass sys.stderr.write("\n") print(f" 완료: {len(datasets)}개 종목\n") # ── 결과 집계 ────────────────────────────────────────────── print("=" * 78) print(" F&G 조건별 백테스트 결과 (60일 / 모멘텀 진입 / 스탑 1.5%)") print("=" * 78) print(f" {'조건':<30} {'거래':>5} {'승률':>6} {'평균PnL':>8} " f"{'손익비':>6} {'총PnL':>8} {'MaxDD':>7} {'진입F&G':>7}") print(" " + "-" * 74) all_results = {} for fng_lo, fng_hi, label in CONFIGS: all_trades: list[Trade] = [] for df in datasets.values(): all_trades.extend(simulate(df, fng_map, fng_lo, fng_hi)) s = stats(all_trades) all_results[label] = (s, all_trades) marker = " ◀ 현행" if "없음" in label else "" if s["n"] == 0: print(f" {label:<30} 거래 없음") continue print( f" {label:<30} {s['n']:>5}건 {s['wr']:>5.1f}% " f"{s['avg_pnl']:>+7.3f}% {s['rr']:>5.2f} " f"{s['total_pnl']:>+7.1f}% -{s['max_dd']:>5.1f}% " f"{s['avg_fng']:>6.1f}{marker}" ) # ── F&G 구간별 세부 성과 ────────────────────────────────── print() print(" F&G 수치 구간별 실제 거래 성과 (필터 없음 전체 거래 기준):") base_trades = all_results["필터 없음 (baseline)"][1] zones = [ (0, 25, "Extreme Fear (0~25) "), (26, 45, "Fear (26~45)"), (46, 55, "Neutral (46~55)"), (56, 75, "Greed (56~75)"), (76, 100, "Extreme Greed(76~100)"), ] print(f" {'F&G 구간':<28} {'건수':>5} {'승률':>6} {'평균PnL':>9} {'손익비':>6}") print(" " + "-" * 60) for lo, hi, name in zones: subset = [t for t in base_trades if lo <= t.fng <= hi] if not subset: continue s = stats(subset) print( f" {name:<28} {s['n']:>5}건 {s['wr']:>5.1f}% " f"{s['avg_pnl']:>+8.3f}% {s['rr']:>5.2f}" ) # ── 월별 F&G 흐름과 성과 ───────────────────────────────── print() print(" F&G 추이와 진입 시점 분포:") sorted_fng = sorted(fng_map.items())[-70:] # 최근 70일 for dt_str, val in sorted_fng[::7]: # 1주 간격 bar_len = val // 5 bar = "█" * bar_len zone = ("극공포" if val <= 25 else "공포" if val <= 45 else "중립" if val <= 55 else "탐욕") print(f" {dt_str} {val:>3} {bar:<20} {zone}") if __name__ == "__main__": main()