365 lines
17 KiB
Python
365 lines
17 KiB
Python
#!/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()
|