#!/usr/bin/env python3 """ upbit_tail_param_search.py — 업비트 꼬리잡기 백테스트 파라미터 자동 탐색 (Grid Search) ====================================================================================== 실행: cd /home/hoon/upbit_bot python3 upbit_tail_param_search.py 옵션: --start 시작일 (기본: 오늘-7일) --end 종료일 (기본: 오늘) --mode 탐색 모드: coarse(기본) / fine / full --timeframe 분봉 단위 (기본: 3) --top 상위 N개 출력 (기본: 20) --min_trades 최소 거래 건수 필터 (기본: 3) --apply 결과 1위를 DB env_config에 자동 적용 (기본: False) 예시: python3 upbit_tail_param_search.py --start 2026-03-01 --end 2026-03-09 python3 upbit_tail_param_search.py --mode fine --timeframe 3 --apply python3 upbit_tail_param_search.py --mode full --min_trades 5 --top 30 """ import sys, os, json, time, argparse, pymysql, pymysql.cursors from datetime import datetime, timedelta from itertools import product # upbit_backtest_web 임포트 (같은 디렉토리) SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, SCRIPT_DIR) import logging logging.basicConfig(level=logging.WARNING, format="[%(asctime)s] %(message)s", datefmt="%H:%M:%S") from upbit_backtest_web import app # Flask 앱 # ──────────────────────────────────────────────────────────────────────────── # DB 연결 설정 # ──────────────────────────────────────────────────────────────────────────── _DB_CFG = dict( host="192.168.0.141", port=3306, user="jae", password="1234", database="upbit_quant_db", charset="utf8mb4", autocommit=True, cursorclass=pymysql.cursors.DictCursor, connect_timeout=10, ) # ──────────────────────────────────────────────────────────────────────────── # 파라미터 그리드 (코인 시장 특성 반영: 낙폭 크고 수수료 낮음) # ──────────────────────────────────────────────────────────────────────────── GRIDS = { # 빠른 1차 탐색 (약 480 조합, 2~5분) "coarse": { "min_drop_rate": [2.0, 3.0, 4.0, 5.0, 7.0], # 당일 낙폭 최소 (%) "min_recovery_ratio": [30, 40, 50, 60], # 당일 회복률 최소 (%) "sl_pct": [2.0, 3.0, 5.0], # 손절 (%) — 코인은 변동성 크므로 넓게 "tp_pct": [3.0, 5.0, 7.0, 10.0], # 익절 (%) — ATR 기반 목표가 "tail_ratio_min": [1.0, 1.5, 2.0], # 꼬리/몸통 비율 }, # 세밀 2차 탐색 (약 3,240 조합, 10~20분) "fine": { "min_drop_rate": [2.0, 3.0, 4.0, 5.0, 7.0], "min_recovery_ratio": [30, 40, 50, 60], "sl_pct": [1.5, 2.0, 3.0, 5.0], "tp_pct": [3.0, 5.0, 7.0, 10.0, 15.0], "tail_ratio_min": [0.8, 1.0, 1.5, 2.0], "shoulder_cut_pct": [2.0, 3.0, 5.0], }, # 전체 탐색 (약 25,920 조합, 1시간 이상) "full": { "min_drop_rate": [1.5, 2.0, 3.0, 4.0, 5.0, 7.0, 10.0], "min_recovery_ratio": [20, 30, 40, 50, 60, 70], "sl_pct": [1.5, 2.0, 3.0, 5.0, 7.0], "tp_pct": [3.0, 5.0, 7.0, 10.0, 15.0, 20.0], "tail_ratio_min": [0.5, 0.8, 1.0, 1.5, 2.0], "shoulder_cut_pct": [2.0, 3.0, 5.0, 7.0], "rsi_threshold": [72, 75, 78, 82], }, } # 고정값 (탐색에서 제외된 파라미터 — 업비트 코인 특성 반영) FIXED_DEFAULTS = dict( rsi_period = 14, rsi_threshold = 78, # RSI 과열 기준 max_rec = 80, # 최대 회복 위치 (%) tail_pct_min = 0.3, # 꼬리 최소 % (%) shoulder_min_high = 1.5, # 어깨 발동 최소 이익 (%) shoulder_cut_pct = 3.0, # 어깨 컷 (%) high_chase_thr = 96.0, # 고점 추격 방지 (%) slot_money = 100_000, # 코인당 투자금 (원) — 업비트 단위 fee_rate = 0.05, # 업비트 수수료 0.05% (편도) cooldown_min = 30, # 쿨다운 (분) max_hold_hours = 24, # 코인: 장 마감 없음 → 24시간 강제청산 max_daily = 5, # 코인당 일일 최대 거래횟수 (코인 특성상 주식보다 많게) ) # ──────────────────────────────────────────────────────────────────────────── # 탐색 실행 # ──────────────────────────────────────────────────────────────────────────── def run_search(start: str, end: str, mode: str, timeframe: int, top_n: int, min_trades: int, apply_best: bool): grid = GRIDS[mode] keys = list(grid.keys()) combos = list(product(*[grid[k] for k in keys])) total = len(combos) print(f"\n[{mode.upper()} 모드 — 업비트 꼬리잡기] 탐색 조합: {total:,}개 | 기간: {start} ~ {end} | 분봉: {timeframe}분") print("=" * 72) results = [] t0 = time.time() with app.test_client() as c: for i, vals in enumerate(combos): params = dict(FIXED_DEFAULTS) params.update(dict(zip(keys, vals))) params["start"] = start params["end"] = end params["timeframe"] = timeframe qs = "&".join(f"{k}={v}" for k, v in params.items()) try: r = c.get(f"/api/backtest/tail?{qs}") if r.status_code != 200: continue d = json.loads(r.data) s = d.get("summary", {}) except Exception: continue if s.get("total_trades", 0) < min_trades: continue results.append({ "params": {k: params[k] for k in keys}, "total_pnl": s["total_pnl"], "win_rate": s["win_rate"], "total_trades": s["total_trades"], "pf": s["profit_factor"], "avg_hold": s["avg_hold_min"], "mdd": s["max_drawdown"], }) if (i + 1) % 100 == 0: elapsed = time.time() - t0 eta = elapsed / (i + 1) * (total - i - 1) print(f" 진행: {i+1:>5}/{total} 경과: {elapsed:>5.0f}s 남은: {eta:>5.0f}s", flush=True) elapsed = time.time() - t0 print(f"\n완료: {elapsed:.1f}초 | 유효 결과: {len(results)}건") if not results: print("⚠️ 유효 조합을 찾지 못했습니다.") print(" - 기간을 늘리거나 (--start 2026-01-01)") print(" - min_trades를 낮추거나 (--min_trades 1)") print(" - upbit_candles DB에 데이터가 있는지 확인하세요.") return results.sort(key=lambda x: x["total_pnl"], reverse=True) # ── 결과 출력 ────────────────────────────────────────────────────────── hdr_keys = list(keys) col_w = max(len(k) for k in hdr_keys) + 2 print(f"\n{'='*72}") print(f" 🏆 수익 TOP {min(top_n, len(results))} (투자금 {FIXED_DEFAULTS['slot_money']:,}원 기준, 수수료 0.05%)") print(f"{'='*72}") hdr = " ".join(f"{k:>{col_w}}" for k in hdr_keys) print(f"{hdr} | {'손익(원)':>12} {'승률':>6} {'거래':>5} {'PF':>5} {'보유(분)':>8}") print("-" * (len(hdr) + 57)) for r in results[:top_n]: p = r["params"] row = " ".join(f"{p[k]:>{col_w}.4g}" for k in hdr_keys) print(f"{row} | {r['total_pnl']:>+12,.0f} {r['win_rate']:>5.1f}% " f"{r['total_trades']:>5} {r['pf']:>5.2f} {r['avg_hold']:>7.1f}분") best = results[0] bp = best["params"] # 파라미터명 → DB 컬럼명 표시 _disp_map = { "min_drop_rate": "MIN_DROP_RATE (÷100)", "min_recovery_ratio": "MIN_RECOVERY_RATIO (÷100)", "sl_pct": "STOP_LOSS_PCT (음수÷100)", "tp_pct": "TAKE_PROFIT_PCT (÷100)", "tail_ratio_min": "TAIL_RATIO_MIN", "shoulder_cut_pct": "SHOULDER_CUT_PCT (÷100)", "rsi_threshold": "RSI_OVERHEAT_THRESHOLD", } print(f""" ╔══════════════════════════════════════════════╗ ║ 🏆 업비트 꼬리잡기 최적 파라미터 ║ ╠══════════════════════════════════════════════╣""") for k, v in bp.items(): label = _disp_map.get(k, k) print(f"║ {label:<36s}: {v!s:>6} ║") print(f"""╠══════════════════════════════════════════════╣ ║ 총 손익 : {best['total_pnl']:>+12,.0f} 원 ║ ║ 승률 : {best['win_rate']:>6.1f}% ║ ║ 총 거래 : {best['total_trades']:>5} 건 ║ ║ Profit Factor : {best['pf']:>5.2f} ║ ║ 평균 보유 : {best['avg_hold']:>7.1f} 분 ║ ║ 최대 낙폭(MDD): {best['mdd']:>12,.0f} 원 ║ ╚══════════════════════════════════════════════╝""") # ── JSON 저장 ────────────────────────────────────────────────────────── out_dir = os.path.join(SCRIPT_DIR, "param_search_results") os.makedirs(out_dir, exist_ok=True) ts = datetime.now().strftime("%Y%m%d_%H%M%S") out_path = os.path.join(out_dir, f"upbit_tail_search_{mode}_{ts}.json") with open(out_path, "w", encoding="utf-8") as f: json.dump({ "mode": mode, "timeframe": timeframe, "start": start, "end": end, "top": results[:top_n], }, f, ensure_ascii=False, indent=2) print(f"\n💾 결과 저장: {out_path}") if apply_best: _apply_to_db(bp) # ──────────────────────────────────────────────────────────────────────────── # DB 자동 적용 # ──────────────────────────────────────────────────────────────────────────── def _apply_to_db(best_params: dict): """ 1위 파라미터를 upbit_quant_db env_config에 자동 반영 - 비율(%) 값을 소수점으로 변환하여 봇이 직접 사용 가능한 형태로 저장 """ field_map = { "min_drop_rate": ("MIN_DROP_RATE", lambda v: str(v / 100)), "min_recovery_ratio": ("MIN_RECOVERY_RATIO", lambda v: str(v / 100)), "sl_pct": ("STOP_LOSS_PCT", lambda v: str(-v / 100)), # 음수 저장 "tp_pct": ("TAKE_PROFIT_PCT", lambda v: str(v / 100)), "tail_ratio_min": ("TAIL_RATIO_MIN", lambda v: str(v)), "shoulder_cut_pct": ("SHOULDER_CUT_PCT", lambda v: str(v / 100)), "rsi_threshold": ("RSI_OVERHEAT_THRESHOLD", lambda v: str(v)), } sets, vals = [], [] for param_k, val in best_params.items(): if param_k not in field_map: continue db_col, fmt = field_map[param_k] sets.append(f"`{db_col}` = %s") vals.append(fmt(val)) if not sets: print("DB 적용할 파라미터가 없습니다.") return try: conn = pymysql.connect(**_DB_CFG) cur = conn.cursor() # 최신 env_config row 업데이트 cur.execute("SELECT id FROM env_config ORDER BY id DESC LIMIT 1") row = cur.fetchone() if row: vals.append(row["id"]) cur.execute(f"UPDATE env_config SET {', '.join(sets)} WHERE id = %s", vals) print("\n✅ DB env_config 자동 적용 완료:") else: # row가 없으면 새로 생성 cols = ", ".join(f"`{sets[i].split('`')[1]}`" for i in range(len(sets))) marks = ", ".join(["%s"] * len(sets)) cur.execute(f"INSERT INTO env_config ({cols}) VALUES ({marks})", vals) print("\n✅ DB env_config 새 row 생성 완료:") for param_k, val in best_params.items(): if param_k in field_map: db_col, fmt = field_map[param_k] print(f" {db_col:<35s} = {fmt(val)}") conn.close() except Exception as e: print(f"❌ DB 적용 실패: {e}") # ──────────────────────────────────────────────────────────────────────────── # CLI 진입점 # ──────────────────────────────────────────────────────────────────────────── def main(): today = datetime.now().strftime("%Y-%m-%d") week_ago = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d") parser = argparse.ArgumentParser( description="업비트 꼬리잡기 백테스트 파라미터 Grid Search", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 예시: python3 upbit_tail_param_search.py python3 upbit_tail_param_search.py --start 2026-01-01 --end 2026-03-09 python3 upbit_tail_param_search.py --mode fine --timeframe 3 --apply python3 upbit_tail_param_search.py --mode full --min_trades 5 --top 30 탐색 전 캔들 수집 필수: → http://localhost:6060 → [캔들 수집] 탭에서 마켓/기간 설정 후 수집 """, ) parser.add_argument("--start", default=week_ago, help="시작일 (YYYY-MM-DD)") parser.add_argument("--end", default=today, help="종료일 (YYYY-MM-DD)") parser.add_argument("--mode", default="coarse", choices=["coarse", "fine", "full"], help="탐색 모드 (coarse=빠름/fine=세밀/full=전체)") parser.add_argument("--timeframe", default=3, type=int, help="분봉 단위 (3/5/15/60, 기본: 3)") parser.add_argument("--top", default=20, type=int, help="상위 N개 출력") parser.add_argument("--min_trades", default=3, type=int, help="최소 거래 건수 필터") parser.add_argument("--apply", action="store_true", help="1위 결과를 DB env_config에 자동 적용") args = parser.parse_args() # 사전 확인: DB에 캔들 데이터가 있는지 체크 try: conn = pymysql.connect(**_DB_CFG) cur = conn.cursor() cur.execute( "SELECT COUNT(*) as cnt FROM upbit_candles WHERE timeframe=%s " "AND candle_time >= %s AND candle_time <= %s", [args.timeframe, args.start.replace("-","") + "0000", args.end.replace("-","") + "2359"] ) cnt = (cur.fetchone() or {}).get("cnt", 0) conn.close() if cnt == 0: print(f"\n⚠️ [경고] upbit_candles에 {args.timeframe}분봉 데이터가 없습니다!") print(f" 기간: {args.start} ~ {args.end}") print(" → http://localhost:6060 → [캔들 수집] 탭에서 먼저 데이터를 수집하세요.\n") sys.exit(1) else: print(f"✅ DB 캔들 확인: {cnt:,}봉 ({args.timeframe}분봉, {args.start}~{args.end})") except Exception as e: print(f"⚠️ DB 연결 확인 실패: {e}") run_search( start = args.start, end = args.end, mode = args.mode, timeframe = args.timeframe, top_n = args.top, min_trades = args.min_trades, apply_best = args.apply, ) if __name__ == "__main__": main()