379 lines
16 KiB
Python
379 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
param_search.py — 스캘핑 백테스트 파라미터 자동 탐색 (Grid Search)
|
|
=====================================================================
|
|
실행:
|
|
cd /home/hoon/kis_bot
|
|
python3 backtest_scalping/param_search.py
|
|
|
|
옵션:
|
|
--start 시작일 (기본: 오늘-7일)
|
|
--end 종료일 (기본: 오늘)
|
|
--mode 탐색 모드: coarse(기본) / fine / full
|
|
--top 상위 N개 출력 (기본: 20)
|
|
--min_trades 최소 거래 건수 필터 (기본: 5)
|
|
--apply 결과 1위를 DB에 자동 적용 (기본: False)
|
|
|
|
예시:
|
|
python3 backtest_scalping/param_search.py --start 2026-03-01 --end 2026-03-06 --mode fine --apply
|
|
"""
|
|
"""
|
|
# 오늘 하루 기준, 빠른 탐색 (약 720 조합, 2분 소요)
|
|
python3 backtest_scalping/param_search.py --start 2026-03-06 --end 2026-03-06
|
|
|
|
# 지난 1주일, 세밀 탐색 (약 5,400 조합, 15분 소요)
|
|
python3 backtest_scalping/param_search.py --start 2026-03-01 --end 2026-03-06 --mode fine
|
|
|
|
# 탐색 후 1위 파라미터를 DB에 자동 적용
|
|
python3 backtest_scalping/param_search.py --start 2026-11-01 --end 2026-03-06 --mode fine --apply
|
|
|
|
# 전체 탐색 (51,840 조합, 2시간 이상, 주말에 돌려두기)
|
|
python3 backtest_scalping/param_search.py --start 2026-02-01 --end 2026-03-06 --mode full --apply
|
|
|
|
결과는 backtest_scalping/results/search_coarse_YYYYMMDD_HHMMSS.json에 자동 저장됩니다.
|
|
|
|
핵심 팁: 데이터가 많을수록 신뢰도가 높으니, ws_candles 데이터가 2주 이상 쌓인 후에 --mode fine으로 돌리는 게 좋습니다.
|
|
"""
|
|
|
|
import sys, os, json, time, argparse
|
|
from datetime import datetime, timedelta
|
|
from itertools import product
|
|
from typing import Optional
|
|
|
|
MIN_WIN_RATE_DEFAULT = 52.0
|
|
|
|
# kis_bot 루트를 경로에 추가
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
sys.path.insert(0, ROOT)
|
|
|
|
import logging
|
|
logging.getLogger("TradeDB").setLevel(logging.WARNING) # 반복 초기화 로그 억제
|
|
|
|
from backtest_web import app
|
|
from database import TradeDB
|
|
import scalping_engine as se
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# 스캘핑 기본값 = 엔진에서 DB 로드 (백테스트 API와 동일 단일 소스, 실매매와 동기화)
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
def _fixed_defaults():
|
|
"""엔진 get_scalping_defaults_from_db() 사용. API 쿼리용으로 percent 변환."""
|
|
_d = se.get_scalping_defaults_from_db()
|
|
return {
|
|
"rsi_period": _d["rsi_period"],
|
|
"slot_money": _d["slot_money"],
|
|
"vol_mult": _d["vol_mult"],
|
|
"trail_trigger": _d["trail_trigger"] * 100,
|
|
"trail_stop": _d["trail_stop"] * 100,
|
|
"cooldown_min": _d["cooldown_min"],
|
|
"time_start": _d["time_start"],
|
|
"time_end": _d["time_end"],
|
|
"max_daily": _d["max_daily"],
|
|
"fee_rate": _d["fee_rate"] * 100,
|
|
"sell_tax": _d["sell_tax"] * 100,
|
|
}
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# 파라미터 그리드 정의
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
GRIDS = {
|
|
# 빠른 1차 탐색 (약 720 조합)
|
|
"coarse": {
|
|
"rsi_oversold": [10, 12, 15, 18, 20, 25],
|
|
"sl_pct": [0.8, 1.0, 1.5, 2.0],
|
|
"tp_pct": [1.5, 2.0, 2.5, 3.0, 4.0],
|
|
"drop_rate": [0.5, 1.0, 1.5, 2.0, 2.5, 3.0],
|
|
# 아래는 고정 (coarse 모드)
|
|
},
|
|
# 세밀 2차 탐색 (약 5,400 조합, 시간 오래 걸림)
|
|
"fine": {
|
|
"rsi_oversold": [15, 16, 17, 18, 19, 20],
|
|
"sl_pct": [0.7, 0.8, 1.0, 1.2, 1.5],
|
|
"tp_pct": [2.0, 2.5, 3.0, 3.5, 4.0],
|
|
"drop_rate": [1.5, 2.0, 2.5, 3.0],
|
|
"trail_trigger": [0, 0.5, 0.8],
|
|
"trail_stop": [0.3, 0.5],
|
|
"cooldown_min": [5, 10, 15],
|
|
},
|
|
# 전체 탐색 (약 51,840 조합, 매우 오래 걸림)
|
|
"full": {
|
|
"rsi_oversold": [10, 12, 15, 18, 20, 25],
|
|
"sl_pct": [0.8, 1.0, 1.5, 2.0],
|
|
"tp_pct": [1.5, 2.0, 2.5, 3.0, 4.0],
|
|
"drop_rate": [0.5, 1.0, 1.5, 2.0, 2.5, 3.0],
|
|
"trail_trigger": [0, 0.5, 0.8, 1.0],
|
|
"trail_stop": [0.3, 0.5, 0.8],
|
|
"cooldown_min": [5, 10, 20],
|
|
"max_daily": [2, 3],
|
|
},
|
|
}
|
|
|
|
def _get_scalp_field_map():
|
|
"""스캘핑 파라미터 → env_config 컬럼 매핑 (apply / db_snapshot 공용)."""
|
|
return {
|
|
"rsi_oversold": ("SCALP_RSI_OVERSOLD", lambda v: str(int(v))),
|
|
"sl_pct": ("SCALP_STOP_LOSS_PCT", lambda v: str(float(v) / 100)),
|
|
"tp_pct": ("SCALP_TAKE_PROFIT_PCT", lambda v: str(float(v) / 100)),
|
|
"drop_rate": ("SCALP_MIN_DROP_RATE", lambda v: str(float(v) / 100)),
|
|
}
|
|
|
|
|
|
def _params_to_db_snapshot(params: dict) -> dict:
|
|
"""그리드 params(표시 단위) → env_config 컬럼명:값 문자열 dict."""
|
|
field_map = _get_scalp_field_map()
|
|
return {
|
|
db_col: fmt(params[param_k])
|
|
for param_k, (db_col, fmt) in field_map.items()
|
|
if param_k in params
|
|
}
|
|
|
|
|
|
def _apply_from_latest_json(rank: int):
|
|
"""최근 search_*.json에서 rank번째(1-based) 항목의 merged_params를 DB에 적용."""
|
|
out_dir = os.path.join(ROOT, "backtest_scalping", "results")
|
|
if not os.path.isdir(out_dir):
|
|
print("⚠️ results 디렉터리가 없습니다.")
|
|
return
|
|
jsons = [f for f in os.listdir(out_dir) if f.startswith("search_") and f.endswith(".json")]
|
|
if not jsons:
|
|
print("⚠️ search_*.json 파일이 없습니다.")
|
|
return
|
|
jsons.sort(key=lambda f: os.path.getmtime(os.path.join(out_dir, f)), reverse=True)
|
|
latest_path = os.path.join(out_dir, jsons[0])
|
|
with open(latest_path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
top = data.get("top") or []
|
|
if rank < 1 or rank > len(top):
|
|
print(f"⚠️ 순번 {rank}이(가) 유효하지 않습니다. (1~{len(top)})")
|
|
return
|
|
item = top[rank - 1]
|
|
merged = item.get("merged_params")
|
|
if not merged:
|
|
merged = item.get("params")
|
|
if not merged:
|
|
print("⚠️ 해당 항목에 merged_params/params가 없습니다.")
|
|
return
|
|
print(f"📂 {latest_path} 에서 {rank}번째 적용합니다.")
|
|
_apply_to_db(merged)
|
|
|
|
|
|
def run_search(start: str, end: str, mode: str, top_n: int,
|
|
min_trades: int, min_win_rate: float, apply_rank: Optional[int], from_file_only: bool):
|
|
if from_file_only and apply_rank is not None and apply_rank >= 1:
|
|
_apply_from_latest_json(apply_rank)
|
|
return
|
|
|
|
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}")
|
|
print("=" * 70)
|
|
|
|
results = []
|
|
t0 = time.time()
|
|
|
|
FIXED_DEFAULTS = _fixed_defaults()
|
|
with app.test_client() as c:
|
|
for i, vals in enumerate(combos):
|
|
params = dict(FIXED_DEFAULTS) # 고정값 먼저 (DB 쿨다운/장시간 반영)
|
|
params.update(dict(zip(keys, vals))) # 그리드값으로 덮어쓰기
|
|
params["start"] = start
|
|
params["end"] = end
|
|
|
|
qs = "&".join(f"{k}={v}" for k, v in params.items())
|
|
r = c.get(f"/api/backtest/scalping?{qs}")
|
|
if r.status_code != 200:
|
|
continue
|
|
d = json.loads(r.data)
|
|
s = d.get("summary", {})
|
|
|
|
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"],
|
|
})
|
|
|
|
# 진행률 출력 (200건마다)
|
|
if (i + 1) % 200 == 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("⚠️ 수익 조합을 찾지 못했습니다. 기간을 늘리거나 min_trades를 낮춰보세요.")
|
|
return
|
|
|
|
# 총 손익 기준 정렬 후, 승률 min_win_rate 이상만 우선; 없으면 차악(손익 1위) 유지
|
|
results.sort(key=lambda x: x["total_pnl"], reverse=True)
|
|
results_52 = [r for r in results if r["win_rate"] >= min_win_rate]
|
|
if results_52:
|
|
results = results_52
|
|
results.sort(key=lambda x: x["total_pnl"], reverse=True)
|
|
print(f"✅ 승률 {min_win_rate}% 이상 {len(results)}건 중 손익순으로 1위 선정")
|
|
else:
|
|
print(f"⚠️ 승률 {min_win_rate}% 이상 없음 → 차악(손익 1위) 적용")
|
|
|
|
# ── 결과 출력 ──
|
|
hdr_keys = [k for k in keys]
|
|
col_w = max(len(k) for k in hdr_keys) + 2
|
|
|
|
print(f"\n{'='*70}")
|
|
print(f" 🏆 수익 TOP {min(top_n, len(results))} (투자금 1,000,000원 기준)")
|
|
print(f"{'='*70}")
|
|
|
|
# 헤더
|
|
hdr = " ".join(f"{k:>{col_w}}" for k in hdr_keys)
|
|
print(f"{hdr} | {'손익(원)':>12} {'승률':>6} {'거래':>5} {'PF':>5} {'보유':>6}")
|
|
print("-" * (len(hdr) + 55))
|
|
|
|
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']:>5.1f}분")
|
|
|
|
best = results[0]
|
|
bp = best["params"]
|
|
print(f"""
|
|
╔══════════════════════════════════════════╗
|
|
║ 🏆 1위 최적 파라미터 ║
|
|
╠══════════════════════════════════════════╣""")
|
|
# 결과 파라미터명 → DB 컬럼명 표시
|
|
_disp_map = {
|
|
"rsi_oversold": "SCALP_RSI_OVERSOLD",
|
|
"sl_pct": "SCALP_STOP_LOSS_PCT (÷100)",
|
|
"tp_pct": "SCALP_TAKE_PROFIT_PCT(÷100)",
|
|
"drop_rate": "SCALP_MIN_DROP_RATE (÷100)",
|
|
"trail_trigger": "(백테스트전용-미저장)",
|
|
"trail_stop": "(백테스트전용-미저장)",
|
|
"cooldown_min": "SCALP_COOLDOWN_SEC(÷60=분)",
|
|
}
|
|
for k, v in bp.items():
|
|
label = _disp_map.get(k, k)
|
|
print(f"║ {label:<30s} : {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']:>5.1f} 분 ║
|
|
║ 최대 낙폭(MDD): {best['mdd']:>12,.0f} 원 ║
|
|
╚══════════════════════════════════════════╝""")
|
|
|
|
# ── 각 결과에 순번·merged_params·db_snapshot 부여 (JSON 및 apply N 지원) ──
|
|
top_list = []
|
|
for idx, r in enumerate(results[:top_n]):
|
|
p = r["params"]
|
|
top_list.append({
|
|
"rank": idx + 1,
|
|
"params": p,
|
|
"merged_params": p,
|
|
"db_snapshot": _params_to_db_snapshot(p),
|
|
"total_pnl": r["total_pnl"],
|
|
"win_rate": r["win_rate"],
|
|
"total_trades": r["total_trades"],
|
|
"pf": r["pf"],
|
|
"avg_hold": r["avg_hold"],
|
|
"mdd": r["mdd"],
|
|
})
|
|
|
|
# ── JSON 결과 저장 ──
|
|
out_dir = os.path.join(ROOT, "backtest_scalping", "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"search_{mode}_{ts}.json")
|
|
with open(out_path, "w", encoding="utf-8") as f:
|
|
json.dump({
|
|
"mode": mode,
|
|
"start": start,
|
|
"end": end,
|
|
"min_win_rate": min_win_rate,
|
|
"top": top_list,
|
|
}, f, ensure_ascii=False, indent=2)
|
|
print(f"\n💾 결과 저장: {out_path}")
|
|
|
|
# ── DB 자동 적용 ──
|
|
if apply_rank is not None and apply_rank >= 1 and apply_rank <= len(top_list):
|
|
_apply_to_db(top_list[apply_rank - 1]["merged_params"])
|
|
print(f"✅ {apply_rank}번째 결과 적용 완료")
|
|
|
|
|
|
def _apply_to_db(best_params: dict):
|
|
"""1위 파라미터를 env_config DB에 자동 반영.
|
|
trail_trigger / trail_stop / cooldown_min 은 env_config 컬럼이 없으므로 저장 제외.
|
|
"""
|
|
field_map = _get_scalp_field_map()
|
|
|
|
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
|
|
|
|
db = TradeDB()
|
|
db.conn.execute(f"UPDATE env_config SET {', '.join(sets)}", vals)
|
|
db.conn.commit()
|
|
db.close()
|
|
print("\n✅ DB env_config 자동 적용 완료:")
|
|
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:<30s} = {fmt(val)}")
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# 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")
|
|
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("--top", default=20, type=int, help="상위 N개 출력")
|
|
parser.add_argument("--min_trades", default=5, type=int, help="최소 거래 건수")
|
|
parser.add_argument("--min_win_rate", default=MIN_WIN_RATE_DEFAULT, type=float, help="승률 하한 (%%). 이 이상만 손익순 1위")
|
|
parser.add_argument("--apply", nargs="?", const=1, type=int, default=None, metavar="N",
|
|
help="N번째 결과를 DB에 적용 (기본 1). --from-file 시 최근 JSON에서 적용")
|
|
parser.add_argument("--from-file", action="store_true", help="--apply N 과 함께 사용 시, 최근 결과 JSON에서만 적용 (탐색 생략)")
|
|
args = parser.parse_args()
|
|
|
|
run_search(
|
|
start = args.start,
|
|
end = args.end,
|
|
mode = args.mode,
|
|
top_n = args.top,
|
|
min_trades = args.min_trades,
|
|
min_win_rate = args.min_win_rate,
|
|
apply_rank = args.apply,
|
|
from_file_only = args.from_file,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|