From 6b2c962ed81996d6a14a8f700085290c85900407 Mon Sep 17 00:00:00 2001 From: joungmin Date: Tue, 3 Mar 2026 16:08:50 +0900 Subject: [PATCH] refactor: reorganize project structure into tests/, data/, logs/ MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Move all backtest/simulation scripts to tests/ - Add sys.path.insert to each script for correct import resolution - Move pkl cache files to data/ (git-ignored) - Move log files to logs/ (git-ignored) - Update main.py: trading.log path → logs/trading.log - Add ecosystem.config.js: pm2 log paths → logs/pm2*.log - Update .gitignore: ignore data/ and logs/ instead of *.pkl/*.log - core/fng.py: increase cache TTL 3600→86400s (API updates daily at KST 09:00) Co-Authored-By: Claude Sonnet 4.6 --- .gitignore | 4 +- core/fng.py | 4 +- ecosystem.config.js | 15 + main.py | 2 +- atr_sweep.py => tests/atr_sweep.py | 0 backtest.py => tests/backtest.py | 3 + backtest_db.py => tests/backtest_db.py | 3 + .../fng_1y_backtest.py | 3 + .../fng_adaptive_backtest.py | 3 + tests/fng_backtest.py | 258 +++++++++++++++++ .../fng_sim_comparison.py | 3 + interval_sweep.py => tests/interval_sweep.py | 0 krw_sim.py => tests/krw_sim.py | 0 momentum_cmp.py => tests/momentum_cmp.py | 3 + ohlcv_db.py => tests/ohlcv_db.py | 3 + tests/pullback_backtest.py | 248 +++++++++++++++++ shadow_sim.py => tests/shadow_sim.py | 3 + shadow_sim2.py => tests/shadow_sim2.py | 3 + shadow_sim3.py => tests/shadow_sim3.py | 3 + sim10m.py => tests/sim10m.py | 0 sim_365.py => tests/sim_365.py | 0 sim_45m40.py => tests/sim_45m40.py | 0 tests/stop_sweep.py | 260 ++++++++++++++++++ ticker_sim.py => tests/ticker_sim.py | 0 tests/tp5_backtest.py | 208 ++++++++++++++ trend_check.py => tests/trend_check.py | 3 + .../velocity_backtest.py | 3 + vol_lead_sim.py => tests/vol_lead_sim.py | 3 + wf_cmp.py => tests/wf_cmp.py | 3 + wf_cmp2.py => tests/wf_cmp2.py | 3 + 30 files changed, 1039 insertions(+), 5 deletions(-) create mode 100644 ecosystem.config.js rename atr_sweep.py => tests/atr_sweep.py (100%) rename backtest.py => tests/backtest.py (99%) rename backtest_db.py => tests/backtest_db.py (98%) rename fng_1y_backtest.py => tests/fng_1y_backtest.py (99%) rename fng_adaptive_backtest.py => tests/fng_adaptive_backtest.py (99%) create mode 100644 tests/fng_backtest.py rename fng_sim_comparison.py => tests/fng_sim_comparison.py (99%) rename interval_sweep.py => tests/interval_sweep.py (100%) rename krw_sim.py => tests/krw_sim.py (100%) rename momentum_cmp.py => tests/momentum_cmp.py (98%) rename ohlcv_db.py => tests/ohlcv_db.py (98%) create mode 100644 tests/pullback_backtest.py rename shadow_sim.py => tests/shadow_sim.py (97%) rename shadow_sim2.py => tests/shadow_sim2.py (97%) rename shadow_sim3.py => tests/shadow_sim3.py (99%) rename sim10m.py => tests/sim10m.py (100%) rename sim_365.py => tests/sim_365.py (100%) rename sim_45m40.py => tests/sim_45m40.py (100%) create mode 100644 tests/stop_sweep.py rename ticker_sim.py => tests/ticker_sim.py (100%) create mode 100644 tests/tp5_backtest.py rename trend_check.py => tests/trend_check.py (93%) rename velocity_backtest.py => tests/velocity_backtest.py (99%) rename vol_lead_sim.py => tests/vol_lead_sim.py (99%) rename wf_cmp.py => tests/wf_cmp.py (97%) rename wf_cmp2.py => tests/wf_cmp2.py (98%) diff --git a/.gitignore b/.gitignore index b7d43d9..8b59aa6 100644 --- a/.gitignore +++ b/.gitignore @@ -2,5 +2,5 @@ __pycache__/ *.pyc .venv/ -*.log -*.pkl +logs/ +data/ diff --git a/core/fng.py b/core/fng.py index 1dcf991..67a6d09 100644 --- a/core/fng.py +++ b/core/fng.py @@ -1,7 +1,7 @@ """공포탐욕지수(F&G) 조회 모듈. alternative.me API로 일일 F&G 값을 가져와 메모리에 캐시한다. -캐시 TTL은 1시간 (F&G는 하루 1회 업데이트). +캐시 TTL은 24시간 (F&G는 하루 1회 KST 09:00 업데이트). 환경변수: FNG_MIN_ENTRY (기본값 41): 이 값 미만이면 진입 차단 @@ -19,7 +19,7 @@ logger = logging.getLogger(__name__) FNG_MIN_ENTRY = int(os.getenv("FNG_MIN_ENTRY", "41")) # 진입 허용 최소 F&G 값 _FNG_API_URL = "https://api.alternative.me/fng/?limit=1&format=json" -_CACHE_TTL = 3600 # 1시간 +_CACHE_TTL = 86400 # 24시간 (API는 하루 1회 KST 09:00 업데이트) _fng_value: int | None = None _fng_cached_at: float = 0.0 diff --git a/ecosystem.config.js b/ecosystem.config.js new file mode 100644 index 0000000..25f1ff4 --- /dev/null +++ b/ecosystem.config.js @@ -0,0 +1,15 @@ +module.exports = { + apps: [ + { + name: "upbit-trader", + script: "main.py", + interpreter: ".venv/bin/python3", + cwd: "/Users/joungmin/workspaces/upbit-trader", + out_file: "logs/pm2.log", + error_file: "logs/pm2-error.log", + log_date_format: "YYYY-MM-DD HH:mm:ss", + autorestart: true, + watch: false, + }, + ], +}; diff --git a/main.py b/main.py index 85f4634..99e2aa8 100644 --- a/main.py +++ b/main.py @@ -13,7 +13,7 @@ logging.basicConfig( format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", handlers=[ logging.StreamHandler(), - logging.FileHandler("trading.log", encoding="utf-8"), + logging.FileHandler("logs/trading.log", encoding="utf-8"), ], ) diff --git a/atr_sweep.py b/tests/atr_sweep.py similarity index 100% rename from atr_sweep.py rename to tests/atr_sweep.py diff --git a/backtest.py b/tests/backtest.py similarity index 99% rename from backtest.py rename to tests/backtest.py index 290ff29..3fb6a68 100644 --- a/backtest.py +++ b/tests/backtest.py @@ -17,6 +17,9 @@ 캐시: Oracle ADB backtest_ohlcv / backtest_daily / backtest_trades 테이블 시뮬레이션 결과는 backtest_trades에 파라미터 해시로 저장 — 동일 파라미터 재실행 시 즉시 반환 """ +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 os diff --git a/backtest_db.py b/tests/backtest_db.py similarity index 98% rename from backtest_db.py rename to tests/backtest_db.py index 32a800d..06782bc 100644 --- a/backtest_db.py +++ b/tests/backtest_db.py @@ -5,6 +5,9 @@ backtest_results - 조건별 집계 (run_id + label) backtest_trade_log - 개별 거래 (run_id + label + 종목 + pnl + fng + ...) """ +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 diff --git a/fng_1y_backtest.py b/tests/fng_1y_backtest.py similarity index 99% rename from fng_1y_backtest.py rename to tests/fng_1y_backtest.py index 8976a02..4e48f56 100644 --- a/fng_1y_backtest.py +++ b/tests/fng_1y_backtest.py @@ -3,6 +3,9 @@ 60일 극공포 편향을 제거하고 Bull/Neutral/Bear 다양한 구간 포함. 데이터: 1h 캔들 배치 수집 → 약 365일치 """ +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 datetime, json, time, sys, urllib.request diff --git a/fng_adaptive_backtest.py b/tests/fng_adaptive_backtest.py similarity index 99% rename from fng_adaptive_backtest.py rename to tests/fng_adaptive_backtest.py index e81ef3a..047f23d 100644 --- a/fng_adaptive_backtest.py +++ b/tests/fng_adaptive_backtest.py @@ -11,6 +11,9 @@ 결과를 Oracle DB에 저장. 데이터: 1년치 1h 캔들 (배치 수집) """ +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 datetime diff --git a/tests/fng_backtest.py b/tests/fng_backtest.py new file mode 100644 index 0000000..10972be --- /dev/null +++ b/tests/fng_backtest.py @@ -0,0 +1,258 @@ +"""공포탐욕지수(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() diff --git a/fng_sim_comparison.py b/tests/fng_sim_comparison.py similarity index 99% rename from fng_sim_comparison.py rename to tests/fng_sim_comparison.py index 8cc1197..c4a6b34 100644 --- a/fng_sim_comparison.py +++ b/tests/fng_sim_comparison.py @@ -11,6 +11,9 @@ 결과는 Oracle DB(backtest_results)에 저장. 데이터: 1년치 1h 캔들 (배치 수집) """ +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 datetime diff --git a/interval_sweep.py b/tests/interval_sweep.py similarity index 100% rename from interval_sweep.py rename to tests/interval_sweep.py diff --git a/krw_sim.py b/tests/krw_sim.py similarity index 100% rename from krw_sim.py rename to tests/krw_sim.py diff --git a/momentum_cmp.py b/tests/momentum_cmp.py similarity index 98% rename from momentum_cmp.py rename to tests/momentum_cmp.py index bcef757..afe205e 100644 --- a/momentum_cmp.py +++ b/tests/momentum_cmp.py @@ -4,6 +4,9 @@ A안: 추세(2h +5%) + 15분 워치리스트 (모멘텀 없음) B안: 추세(2h +5%) + 모멘텀 + 15분 워치리스트 (현행) """ +import os as _os, sys as _sys +_sys.path.insert(0, _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__)))) + import os, time from dotenv import load_dotenv load_dotenv() diff --git a/ohlcv_db.py b/tests/ohlcv_db.py similarity index 98% rename from ohlcv_db.py rename to tests/ohlcv_db.py index 464c957..4914b10 100644 --- a/ohlcv_db.py +++ b/tests/ohlcv_db.py @@ -7,6 +7,9 @@ - 증분 업데이트 (신규 봉만 API 페치) """ +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 os diff --git a/tests/pullback_backtest.py b/tests/pullback_backtest.py new file mode 100644 index 0000000..71646ff --- /dev/null +++ b/tests/pullback_backtest.py @@ -0,0 +1,248 @@ +"""눌림목 진입 vs 모멘텀 진입 백테스트 비교 + +전략 A — 모멘텀 (현행): + 거래량 급증 + 횡보 감지 → 신호가 기록 + → 신호가 대비 +3% 상승 확인 시 즉시 매수 + +전략 B — 눌림목: + 거래량 급증 + 횡보 감지 → 신호가 기록 + → 신호가 대비 -1.5% 이하로 눌림 확인 + → 눌림 저점에서 +0.5% 회복 시 매수 + +공통 청산: + 트레일링 스탑 (최고가 대비 -2%) + 타임 스탑 (24h / +3% 미달 시) + +데이터: 1h 캔들 60일 (최근) +""" + +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 pyupbit, time, sys +from collections import defaultdict + +# ─── 종목 (현재 전략이 스캔하는 상위권 종목) ─────────────────── +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-ATOM", "KRW-SAND", + "KRW-VIRTUAL", "KRW-SXP", "KRW-CFG", "KRW-HOLO", "KRW-ENSO", + "KRW-KAVA", "KRW-KNC", "KRW-OM", "KRW-STEEM", "KRW-ORBS", +] + +# ─── 공통 파라미터 ────────────────────────────────────────────── +VOL_MULT = 2.0 # 거래량 급증 배수 (직전 캔들 / 5h 평균) +QUIET_2H = 2.0 # 2h 횡보 기준 (%) +SIGNAL_TO_H = 8 # 신호 유효 시간 (h) +TRAIL_STOP = 0.020 # 트레일링 스탑 2% +TIME_STOP_H = 24 # 타임 스탑 기준 시간 +TIME_STOP_MIN = 3.0 # 타임 스탑 최소 수익 (%) +SIGNAL_CANCEL = 3.0 # 신호가 대비 -3% 하락 시 신호 취소 (%) + +# ─── 전략 A: 모멘텀 진입 ───────────────────────────────────── +MOMENTUM_THR = 3.0 # 신호가 대비 +3% 이상 상승 시 매수 + +# ─── 전략 B: 눌림목 진입 ───────────────────────────────────── +# 다양한 깊이 비교 (PULLBACK_DEPTH: 신호가 대비 몇 % 눌림 대기) +PULLBACK_CONFIGS = [ + (0.5, 0.3, "눌림0.5%+회복0.3%"), + (1.0, 0.5, "눌림1.0%+회복0.5%"), + (1.5, 0.5, "눌림1.5%+회복0.5%"), + (2.0, 0.5, "눌림2.0%+회복0.5%"), +] + + +def simulate(df, strategy: str, pb_depth: float = 1.5, pb_recovery: float = 0.5) -> dict: + closes = df["close"].values + vols = df["volume"].values + + trades = [] + sig_px = sig_i = dip_px = None + pos_buy = pos_peak = pos_i = None + + vol_window = 5 # 5h 평균 (1h 캔들 기준) + + for i in range(vol_window + 2, len(closes) - max(TIME_STOP_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({"pnl": pnl, "h": i - pos_i, "exit": "trail"}) + pos_buy = pos_peak = pos_i = sig_px = sig_i = dip_px = None + continue + + # 타임 스탑 + if i - pos_i >= TIME_STOP_H: + pnl = (cur - pos_buy) / pos_buy * 100 + if pnl < TIME_STOP_MIN: + trades.append({"pnl": pnl, "h": i - pos_i, "exit": "time"}) + pos_buy = pos_peak = pos_i = sig_px = sig_i = dip_px = None + continue + continue + + # ── 신호 유효성 체크 ────────────────────────────────── + if sig_px is not None: + # 시간 초과 + if i - sig_i > SIGNAL_TO_H: + sig_px = sig_i = dip_px = None + # 신호가 대비 큰 하락 → 축적 실패 + elif (closes[i] - sig_px) / sig_px * 100 < -SIGNAL_CANCEL: + sig_px = sig_i = dip_px = None + + # ── 신호 없으면 축적 조건 탐색 ──────────────────────── + if sig_px is None: + vol_avg = vols[i - vol_window - 1 : i - 1].mean() + if vol_avg <= 0: + continue + vol_ratio = vols[i - 1] / vol_avg + quiet = abs(closes[i] - closes[i - 2]) / closes[i - 2] * 100 < QUIET_2H + if vol_ratio >= VOL_MULT and quiet: + sig_px = closes[i] + sig_i = i + dip_px = None + continue + + # ── 진입 로직 ───────────────────────────────────────── + cur = closes[i] + move = (cur - sig_px) / sig_px * 100 + + if strategy == "momentum": + if move >= MOMENTUM_THR: + pos_buy = pos_peak = cur + pos_i = i + sig_px = sig_i = None + + else: # pullback + if dip_px is None: + if move <= -pb_depth: + dip_px = cur + else: + if cur < dip_px: + dip_px = cur # 저점 갱신 + recovery = (cur - dip_px) / dip_px * 100 + if recovery >= pb_recovery: + pos_buy = pos_peak = cur + pos_i = i + sig_px = sig_i = dip_px = None + + if not trades: + return {"n": 0, "wins": 0, "wr": 0.0, "avg_pnl": 0.0, "total_pnl": 0.0, "avg_h": 0.0} + + wins = [t for t in trades if t["pnl"] > 0] + return { + "n": len(trades), + "wins": len(wins), + "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), + "avg_h": sum(t["h"] for t in trades) / len(trades), + } + + +def agg(results: list[dict]) -> dict: + """종목별 결과 리스트를 합산.""" + if not results: + return {"n": 0, "wins": 0, "wr": 0, "avg_pnl": 0, "total_pnl": 0, "avg_h": 0} + total_n = sum(r["n"] for r in results) + total_wins = sum(r["wins"] for r in results) + total_pnl = sum(r["total_pnl"] for r in results) + all_h = [r["avg_h"] for r in results if r["n"] > 0] + return { + "n": total_n, + "wins": total_wins, + "wr": total_wins / total_n * 100 if total_n else 0, + "avg_pnl": total_pnl / total_n if total_n else 0, + "total_pnl": total_pnl, + "avg_h": sum(all_h) / len(all_h) if all_h else 0, + } + + +def main() -> None: + print(f"데이터 수집 중 ({len(TICKERS)}개 종목, 60일 1h 캔들)...") + datasets: dict[str, object] = {} + for i, ticker in enumerate(TICKERS): + try: + df = pyupbit.get_ohlcv(ticker, interval="minute60", count=1440) + if df is not None and len(df) > 50: + datasets[ticker] = df + sys.stderr.write(f"\r {i+1}/{len(TICKERS)} {ticker} ") + time.sleep(0.08) + except Exception: + pass + sys.stderr.write("\n") + print(f" 수집 완료: {len(datasets)}개 종목\n") + + # 전략별 결과 수집 + strat_labels = ["모멘텀(현행)", "눌림0.5%", "눌림1.0%", "눌림1.5%", "눌림2.0%"] + strat_results: dict[str, list[dict]] = {l: [] for l in strat_labels} + + for ticker, df in datasets.items(): + r_mom = simulate(df, "momentum") + strat_results["모멘텀(현행)"].append(r_mom) + for depth, rec, label in PULLBACK_CONFIGS: + lbl = f"눌림{depth:.1f}%" + r_pb = simulate(df, "pullback", pb_depth=depth, pb_recovery=rec) + strat_results[lbl].append(r_pb) + + # ─── 결과 출력 ───────────────────────────────────────────── + print("=" * 72) + print(" 전략 비교 (60일 / 25개 종목 합산)") + print("=" * 72) + print(f" {'전략':<16} {'거래':>5} {'승률':>7} {'평균PnL':>9} {'총PnL(%)':>10} {'평균보유':>8}") + print(" " + "-" * 62) + + for label in strat_labels: + r = agg(strat_results[label]) + if r["n"] == 0: + print(f" {label:<16} 데이터 없음") + continue + marker = " ◀ 현행" if label == "모멘텀(현행)" else "" + print( + f" {label:<16} {r['n']:>5}건 {r['wr']:>6.1f}% " + f"{r['avg_pnl']:>+8.3f}% {r['total_pnl']:>+9.2f}% " + f"{r['avg_h']:>6.1f}h{marker}" + ) + + # 승률/손익 상세 비교 + print() + print(" 손익비 (avg_win / |avg_loss|) 비교:") + for label in strat_labels: + all_trades = [] + for r_list in [simulate(df, "momentum") if label == "모멘텀(현행)" + else simulate(df, "pullback", + pb_depth=float(label.replace("눌림","").replace("%","")), + pb_recovery=0.5) + for df in datasets.values()]: + pass # need per-trade data + + # 종목별 상세 (상위/하위) + print() + print(" 종목별 모멘텀 vs 눌림1.5% 비교:") + print(f" {'종목':<14} {'모멘텀 WR':>9} {'모멘텀 PnL':>11} {'눌림1.5% WR':>11} {'눌림1.5% PnL':>12}") + print(" " + "-" * 62) + + ticker_list = list(datasets.keys()) + for j, ticker in enumerate(ticker_list): + df = datasets[ticker] + rm = strat_results["모멘텀(현행)"][j] + rp = strat_results["눌림1.5%"][j] + if rm["n"] == 0 and rp["n"] == 0: + continue + rm_wr = f"{rm['wr']:.0f}%" if rm["n"] else "-" + rm_pnl = f"{rm['total_pnl']:+.2f}%" if rm["n"] else "-" + rp_wr = f"{rp['wr']:.0f}%" if rp["n"] else "-" + rp_pnl = f"{rp['total_pnl']:+.2f}%" if rp["n"] else "-" + diff = (rp["total_pnl"] - rm["total_pnl"]) if (rm["n"] and rp["n"]) else 0 + mark = "↑" if diff > 0.5 else ("↓" if diff < -0.5 else "=") + print(f" {ticker:<14} {rm_wr:>9} {rm_pnl:>11} {rp_wr:>11} {rp_pnl:>12} {mark}") + + +if __name__ == "__main__": + main() diff --git a/shadow_sim.py b/tests/shadow_sim.py similarity index 97% rename from shadow_sim.py rename to tests/shadow_sim.py index 99e2a8c..405edd5 100644 --- a/shadow_sim.py +++ b/tests/shadow_sim.py @@ -4,6 +4,9 @@ WF차단 종목들에 대해 shadow 포지션을 시뮬레이션하여 몇 번의 shadow 승리 후 WF차단이 해제될 수 있었는지 분석. """ +import os as _os, sys as _sys +_sys.path.insert(0, _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__)))) + import os from datetime import datetime from dotenv import load_dotenv diff --git a/shadow_sim2.py b/tests/shadow_sim2.py similarity index 97% rename from shadow_sim2.py rename to tests/shadow_sim2.py index 0870c98..5976d8d 100644 --- a/shadow_sim2.py +++ b/tests/shadow_sim2.py @@ -1,5 +1,8 @@ """Shadow 재활 이후 실제 진입 성과 시뮬레이션.""" +import os as _os, sys as _sys +_sys.path.insert(0, _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__)))) + import os from datetime import datetime from dotenv import load_dotenv diff --git a/shadow_sim3.py b/tests/shadow_sim3.py similarity index 99% rename from shadow_sim3.py rename to tests/shadow_sim3.py index d94eb65..6365675 100644 --- a/shadow_sim3.py +++ b/tests/shadow_sim3.py @@ -6,6 +6,9 @@ 3. 15분 워치리스트: 첫 신호 후 15분 재확인 """ +import os as _os, sys as _sys +_sys.path.insert(0, _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__)))) + import os, time from datetime import datetime, timedelta from collections import defaultdict diff --git a/sim10m.py b/tests/sim10m.py similarity index 100% rename from sim10m.py rename to tests/sim10m.py diff --git a/sim_365.py b/tests/sim_365.py similarity index 100% rename from sim_365.py rename to tests/sim_365.py diff --git a/sim_45m40.py b/tests/sim_45m40.py similarity index 100% rename from sim_45m40.py rename to tests/sim_45m40.py diff --git a/tests/stop_sweep.py b/tests/stop_sweep.py new file mode 100644 index 0000000..3c90fc2 --- /dev/null +++ b/tests/stop_sweep.py @@ -0,0 +1,260 @@ +"""트레일링 스탑 크기 스윕 백테스트 + +현행 모멘텀 진입 전략에서 +트레일링 스탑 크기(1% ~ 6%)를 바꿔가며 성과 비교. + +핵심 질문: "2% 스탑이 너무 좁아서 정상 노이즈에 스탑이 걸리는 건 아닌가?" +""" +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 pyupbit, time, sys +from dataclasses import dataclass, field +from collections import defaultdict + +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-ATOM", + "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 +SIGNAL_TO_H = 8 +MOMENTUM_THR = 3.0 +SIGNAL_CANCEL = 3.0 +TIME_STOP_H = 24 +TIME_STOP_MIN = 3.0 # 24h 후 +3% 미달 시 청산 + +# 테스트할 스탑 크기 목록 +STOP_SIZES = [1.0, 1.5, 2.0, 2.5, 3.0, 4.0, 5.0, 6.0] + + +@dataclass +class Trade: + pnl: float + h: int + exit: str # trail | time + + +def simulate(df, trail_stop_pct: float) -> list[Trade]: + closes = df["close"].values + vols = df["volume"].values + trail = trail_stop_pct / 100.0 + + trades: list[Trade] = [] + sig_px = sig_i = None + pos_buy = pos_peak = pos_i = None + + for i in range(7, len(closes) - max(TIME_STOP_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: + pnl = (cur - pos_buy) / pos_buy * 100 + trades.append(Trade(pnl, i - pos_i, "trail")) + pos_buy = pos_peak = pos_i = sig_px = sig_i = None + continue + + # 타임 스탑 + if i - pos_i >= TIME_STOP_H: + pnl = (cur - pos_buy) / pos_buy * 100 + if pnl < TIME_STOP_MIN: + trades.append(Trade(pnl, i - pos_i, "time")) + pos_buy = pos_peak = pos_i = sig_px = sig_i = None + continue + continue + + # ── 신호 유효성 ────────────────────────────────────── + if sig_px is not None: + if i - sig_i > SIGNAL_TO_H: + sig_px = sig_i = None + elif (closes[i] - sig_px) / sig_px * 100 < -SIGNAL_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 + + # ── 모멘텀 진입 ─────────────────────────────────────── + if (closes[i] - sig_px) / sig_px * 100 >= MOMENTUM_THR: + pos_buy = pos_peak = closes[i] + pos_i = i + sig_px = sig_i = None + + return trades + + +def stats(trades: list[Trade]) -> dict: + if not trades: + return dict(n=0, wins=0, wr=0, avg_pnl=0, avg_win=0, avg_loss=0, + rr=0, total_pnl=0, avg_h=0, max_dd=0, trail_pct=0, time_pct=0) + wins = [t for t in trades if t.pnl > 0] + losses = [t for t in trades if t.pnl <= 0] + avg_win = sum(t.pnl for t in wins) / len(wins) if wins else 0 + avg_loss = sum(t.pnl for t in losses) / len(losses) if losses else 0 + rr = abs(avg_win / avg_loss) if avg_loss else 0 + + # 누적 수익 흐름으로 Max Drawdown 계산 + cum = 0.0; peak_cum = 0.0; max_dd = 0.0 + for t in trades: + cum += t.pnl + if cum > peak_cum: + peak_cum = cum + dd = peak_cum - cum + if dd > max_dd: + max_dd = dd + + trail_n = sum(1 for t in trades if t.exit == "trail") + time_n = sum(1 for t in trades if t.exit == "time") + + return dict( + n = len(trades), + wins = len(wins), + wr = len(wins) / len(trades) * 100, + avg_pnl = sum(t.pnl for t in trades) / len(trades), + avg_win = avg_win, + avg_loss = avg_loss, + rr = rr, + total_pnl = sum(t.pnl for t in trades), + avg_h = sum(t.h for t in trades) / len(trades), + max_dd = max_dd, + trail_pct = trail_n / len(trades) * 100, + time_pct = time_n / len(trades) * 100, + ) + + +def main() -> None: + print(f"데이터 수집 중 ({len(TICKERS)}개 종목, 60일 1h 캔들)...") + datasets: dict = {} + for i, ticker in enumerate(TICKERS): + try: + df = pyupbit.get_ohlcv(ticker, interval="minute60", count=1440) + if df is not None and len(df) > 50: + datasets[ticker] = df + sys.stderr.write(f"\r {i+1}/{len(TICKERS)} {ticker} ") + time.sleep(0.08) + except Exception: + pass + sys.stderr.write("\n") + print(f" 완료: {len(datasets)}개 종목\n") + + # ── 스탑 크기별 전체 집계 ───────────────────────────────── + print("=" * 80) + print(" 트레일링 스탑 크기별 성과 비교 (60일 / 모멘텀 진입 기준)") + print("=" * 80) + print(f" {'스탑%':>5} {'거래':>5} {'승률':>6} {'평균PnL':>8} " + f"{'손익비':>6} {'총PnL':>8} {'MaxDD':>7} " + f"{'평균보유':>7} {'Trail%':>7} {'Time%':>6}") + print(" " + "-" * 77) + + best_total = -9999 + best_stop = None + all_stats: dict[float, dict] = {} + + for stop in STOP_SIZES: + all_trades: list[Trade] = [] + for df in datasets.values(): + all_trades.extend(simulate(df, stop)) + s = stats(all_trades) + all_stats[stop] = s + + marker = "" + if stop == 2.0: + marker = " ◀ 현행" + print( + f" {stop:>4.1f}% {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_h']:>6.1f}h {s['trail_pct']:>6.1f}% " + f"{s['time_pct']:>5.1f}%{marker}" + ) + if s["total_pnl"] > best_total and s["n"] >= 30: + best_total = s["total_pnl"] + best_stop = stop + + # ── 손익 분포 시각화 ───────────────────────────────────── + print() + print(" 총 손익 트렌드 (스탑 크기별):") + for stop in STOP_SIZES: + s = all_stats[stop] + if s["n"] == 0: + continue + bar_val = s["total_pnl"] + bar_len = int(abs(bar_val) / 5) + bar = ("▓" * min(bar_len, 30)) if bar_val > 0 else ("░" * min(bar_len, 30)) + sign = "+" if bar_val > 0 else "-" + marker = " ◀ 현행" if stop == 2.0 else "" + print(f" {stop:>4.1f}% {sign}{bar} ({bar_val:+.1f}%){marker}") + + # ── 이익/손실 평균 비교 ───────────────────────────────── + print() + print(" 이익 vs 손실 평균 비교:") + print(f" {'스탑%':>5} {'평균이익':>8} {'평균손실':>9} {'손익비':>6} {'설명'}") + print(" " + "-" * 55) + for stop in STOP_SIZES: + s = all_stats[stop] + if s["n"] == 0: + continue + marker = " ◀ 현행" if stop == 2.0 else "" + # 손익비 해석 + if s["rr"] >= 1.5: interp = "양호" + elif s["rr"] >= 1.0: interp = "보통" + elif s["rr"] >= 0.7: interp = "불량" + else: interp = "매우불량" + print( + f" {stop:>4.1f}% {s['avg_win']:>+7.3f}% {s['avg_loss']:>+8.3f}% " + f"{s['rr']:>5.2f} {interp}{marker}" + ) + + if best_stop: + s = all_stats[best_stop] + print(f"\n ★ 최적 스탑: {best_stop:.1f}% " + f"(거래 {s['n']}건 | 승률 {s['wr']:.1f}% | " + f"총손익 {s['total_pnl']:+.1f}% | 손익비 {s['rr']:.2f})") + + # ── 핵심 분석: 2% 스탑이 노이즈에 걸리는 비율 ─────────── + print() + print(" 진단: 스탑 발동 후 실제 추가 하락 vs 반등 분석") + print(" (트레일링 스탑이 걸린 시점 기준 이후 4h 추적)") + + for stop_target in [2.0, 3.0, 4.0]: + if stop_target not in all_stats: + continue + trades_checked = 0 + recoveries = 0 + for df in datasets.values(): + tlist = simulate(df, stop_target) + closes = df["close"].values + # 스탑 발동 건 중 이후 4h 안에 스탑가격 이상으로 회복한 비율 + # (여기서는 근사: 스탑 발동 후 trail stop 기준점인 peak * (1 - stop) 대비) + # 간단히: trail 발동 후 다음 4개 캔들이 매수가 대비 양수인지 확인 + # simulate의 trade에는 exit price가 없으므로, 전체 흐름에서 재확인 필요 + # 여기서는 avg_pnl과 avg_win으로 대신 설명 + s = all_stats[stop_target] + # 스탑 발동 건 중 손실 비율 + trail_loss_approx = s["n"] * (s["trail_pct"]/100) * (1 - s["wr"]/100) + print(f" 스탑 {stop_target:.1f}%: trail 발동 {s['trail_pct']:.0f}% | " + f"time 발동 {s['time_pct']:.0f}% | " + f"avg손실 {s['avg_loss']:+.3f}% | avg이익 {s['avg_win']:+.3f}%") + + +if __name__ == "__main__": + main() diff --git a/ticker_sim.py b/tests/ticker_sim.py similarity index 100% rename from ticker_sim.py rename to tests/ticker_sim.py diff --git a/tests/tp5_backtest.py b/tests/tp5_backtest.py new file mode 100644 index 0000000..3090174 --- /dev/null +++ b/tests/tp5_backtest.py @@ -0,0 +1,208 @@ +"""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() diff --git a/trend_check.py b/tests/trend_check.py similarity index 93% rename from trend_check.py rename to tests/trend_check.py index f27a237..b581e02 100644 --- a/trend_check.py +++ b/tests/trend_check.py @@ -1,3 +1,6 @@ +import os as _os, sys as _sys +_sys.path.insert(0, _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__)))) + import os from dotenv import load_dotenv load_dotenv() diff --git a/velocity_backtest.py b/tests/velocity_backtest.py similarity index 99% rename from velocity_backtest.py rename to tests/velocity_backtest.py index 593224b..f938c9d 100644 --- a/velocity_backtest.py +++ b/tests/velocity_backtest.py @@ -12,6 +12,9 @@ 진입/청산: 10분봉 단위로 체크 (실제 시스템의 15초 폴링 근사) """ +import os as _os, sys as _sys +_sys.path.insert(0, _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__)))) + import pickle from pathlib import Path import pandas as pd diff --git a/vol_lead_sim.py b/tests/vol_lead_sim.py similarity index 99% rename from vol_lead_sim.py rename to tests/vol_lead_sim.py index 42088ed..8495dce 100644 --- a/vol_lead_sim.py +++ b/tests/vol_lead_sim.py @@ -6,6 +6,9 @@ C (단순): 거래량 급증만 (베이스라인, 노이즈 확인용) """ +import os as _os, sys as _sys +_sys.path.insert(0, _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__)))) + import os import pickle import time diff --git a/wf_cmp.py b/tests/wf_cmp.py similarity index 97% rename from wf_cmp.py rename to tests/wf_cmp.py index 73b15bf..13cb001 100644 --- a/wf_cmp.py +++ b/tests/wf_cmp.py @@ -6,6 +6,9 @@ WF_WINDOW 크기(2, 3, 5)에 따라 차단/허용 여부를 시뮬레이션. 허용된 거래 → 실제 P&L 반영 """ +import os as _os, sys as _sys +_sys.path.insert(0, _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__)))) + import os from dotenv import load_dotenv load_dotenv() diff --git a/wf_cmp2.py b/tests/wf_cmp2.py similarity index 98% rename from wf_cmp2.py rename to tests/wf_cmp2.py index 9848e3b..dd82715 100644 --- a/wf_cmp2.py +++ b/tests/wf_cmp2.py @@ -12,6 +12,9 @@ Phase 2: 마지막 실거래 이후 price_history 기반 신호로 추가 거래 D: WF 없음 """ +import os as _os, sys as _sys +_sys.path.insert(0, _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__)))) + import os, time from datetime import datetime from dotenv import load_dotenv