211 lines
8.3 KiB
Python
211 lines
8.3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
holding_param_search.py — 홀딩 전략 파라미터 자동 탐색 (종목별 Grid Search)
|
|
=============================================================================
|
|
실행:
|
|
cd /home/hoon/kis_bot
|
|
python3 backtest_scalping/holding_param_search.py
|
|
|
|
옵션:
|
|
--code 종목코드 (미지정 시 관심종목 전체)
|
|
--start 시작일 (기본: 2년 전)
|
|
--end 종료일 (기본: 오늘)
|
|
--mode coarse(기본) / fine / full
|
|
--top 상위 N개 출력 (기본: 10)
|
|
--apply 1위 파라미터를 DB에 자동 적용
|
|
|
|
예시:
|
|
python3 backtest_scalping/holding_param_search.py --code 005930 --apply
|
|
python3 backtest_scalping/holding_param_search.py --mode fine --apply
|
|
"""
|
|
|
|
import sys, os, json, time, argparse
|
|
from datetime import datetime, timedelta
|
|
from itertools import product as iproduct
|
|
|
|
import logging
|
|
logging.getLogger("TradeDB").setLevel(logging.WARNING)
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
sys.path.insert(0, ROOT)
|
|
|
|
from database import TradeDB
|
|
import holding_bot as hb
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# 그리드 정의
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
GRIDS = {
|
|
# 빠른 1차 탐색 (~270 조합)
|
|
"coarse": {
|
|
"rsi_buy1": [40, 35, 30],
|
|
"rsi_buy2": [35, 30, 25],
|
|
"rsi_buy3": [30, 25, 20],
|
|
"take_profit_pct": [3.0, 4.0, 5.0, 6.0, 8.0],
|
|
"stop_loss_pct": [7.0, 10.0, 15.0],
|
|
},
|
|
# 세밀 2차 탐색 (~2,400 조합)
|
|
"fine": {
|
|
"rsi_buy1": [45, 40, 35, 30],
|
|
"rsi_buy2": [40, 35, 30, 25],
|
|
"rsi_buy3": [35, 30, 25, 20],
|
|
"take_profit_pct": [2.0, 3.0, 4.0, 5.0, 6.0, 8.0, 10.0],
|
|
"stop_loss_pct": [5.0, 7.0, 10.0, 15.0, 20.0],
|
|
"rsi_sell": [65, 70, 75],
|
|
},
|
|
# 전체 탐색 (~10,000+ 조합)
|
|
"full": {
|
|
"rsi_buy1": [50, 45, 40, 35, 30],
|
|
"rsi_buy2": [45, 40, 35, 30, 25],
|
|
"rsi_buy3": [40, 35, 30, 25, 20, 15],
|
|
"take_profit_pct": [2.0, 3.0, 4.0, 5.0, 6.0, 8.0, 10.0, 12.0],
|
|
"stop_loss_pct": [5.0, 7.0, 10.0, 15.0, 20.0],
|
|
"rsi_sell": [60, 65, 70, 75, 80],
|
|
"rsi_period": [7, 14, 21],
|
|
},
|
|
}
|
|
|
|
FIXED_DEFAULTS = {
|
|
"rsi_sell": 70,
|
|
"rsi_period": 14,
|
|
"buy1_ratio": 30,
|
|
"buy2_ratio": 30,
|
|
"buy3_ratio": 40,
|
|
"slot_money": 3_000_000,
|
|
}
|
|
|
|
|
|
def run_search(code: str, name: str, candles, mode: str,
|
|
top_n: int, min_trades: int, apply_best: bool, db):
|
|
grid = GRIDS[mode]
|
|
keys = list(grid.keys())
|
|
combos = list(iproduct(*[grid[k] for k in keys]))
|
|
valid = [(zip(keys, v)) for v in combos]
|
|
|
|
results = []
|
|
base_cfg = dict(hb.DEFAULT_STOCK_CONFIG)
|
|
base_cfg.update(FIXED_DEFAULTS)
|
|
|
|
t0 = time.time()
|
|
total = len(combos)
|
|
print(f"\n[{code}] {name} | {mode.upper()} | 조합: {total:,}개")
|
|
print("=" * 60)
|
|
|
|
for i, vals in enumerate(combos):
|
|
cfg = dict(base_cfg)
|
|
cfg.update(dict(zip(keys, vals)))
|
|
# rsi_buy 순서 보정
|
|
if cfg["rsi_buy1"] <= cfg["rsi_buy2"] or cfg["rsi_buy2"] <= cfg["rsi_buy3"]:
|
|
continue
|
|
|
|
res = hb.run_backtest(candles, cfg)
|
|
s = res.get("summary", {})
|
|
if s.get("total_trades", 0) < min_trades:
|
|
continue
|
|
|
|
results.append({
|
|
"params": {k: cfg[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_days"],
|
|
"mdd": s["max_drawdown"],
|
|
})
|
|
|
|
elapsed = time.time() - t0
|
|
print(f"완료: {elapsed:.1f}초 | 유효: {len(results)}건")
|
|
|
|
if not results:
|
|
print("⚠️ 유효 결과 없음. 기간을 늘리거나 min_trades를 낮추세요.")
|
|
return
|
|
|
|
results.sort(key=lambda x: x["total_pnl"], reverse=True)
|
|
|
|
col_w = max(len(k) for k in keys) + 2
|
|
hdr = " ".join(f"{k:>{col_w}}" for k in keys)
|
|
print(f"\n{'='*60}")
|
|
print(f" TOP {min(top_n, len(results))}")
|
|
print(f"{'='*60}")
|
|
print(f"{hdr} | {'손익':>12} {'승률':>6} {'거래':>5} {'PF':>5} {'보유(일)':>8}")
|
|
print("-" * (len(hdr) + 50))
|
|
for r in results[:top_n]:
|
|
p = r["params"]
|
|
row = " ".join(f"{p[k]:>{col_w}.4g}" for k in 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"]
|
|
print(f"""
|
|
╔═══════════════════════════════════════════╗
|
|
║ 🏆 [{code}] {name[:10]} 최적 파라미터
|
|
╠═══════════════════════════════════════════╣""")
|
|
for k, v in bp.items():
|
|
print(f"║ {k:<22s} : {v!s:>8} ║")
|
|
print(f"""╠═══════════════════════════════════════════╣
|
|
║ 총 손익 : {best['total_pnl']:>+12,.0f} 원 ║
|
|
║ 승률 : {best['win_rate']:>6.1f}% ║
|
|
║ 거래 : {best['total_trades']:>5} 건 ║
|
|
║ PF : {best['pf']:>5.2f} ║
|
|
║ 보유(일) : {best['avg_hold']:>5.1f} 일 ║
|
|
║ MDD : {best['mdd']:>12,.0f} 원 ║
|
|
╚═══════════════════════════════════════════╝""")
|
|
|
|
# 결과 저장
|
|
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"holding_{code}_{mode}_{ts}.json")
|
|
with open(out_path, "w", encoding="utf-8") as f:
|
|
json.dump({"code": code, "name": name, "mode": mode,
|
|
"top": results[:top_n]}, f, ensure_ascii=False, indent=2)
|
|
print(f"\n💾 결과 저장: {out_path}")
|
|
|
|
if apply_best:
|
|
hb.set_stock_config(db, code, name, bp)
|
|
print(f"✅ DB 파라미터 적용 완료 ({code})")
|
|
|
|
|
|
def main():
|
|
today = datetime.now().strftime("%Y-%m-%d")
|
|
two_ago = (datetime.now() - timedelta(days=730)).strftime("%Y-%m-%d")
|
|
|
|
parser = argparse.ArgumentParser(description="홀딩 전략 파라미터 Grid Search")
|
|
parser.add_argument("--code", default="", help="종목코드 (공백=관심종목 전체)")
|
|
parser.add_argument("--start", default=two_ago, help="시작일")
|
|
parser.add_argument("--end", default=today, help="종료일")
|
|
parser.add_argument("--mode", default="coarse", choices=["coarse", "fine", "full"])
|
|
parser.add_argument("--top", default=10, type=int)
|
|
parser.add_argument("--min_trades", default=2, type=int)
|
|
parser.add_argument("--apply", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
db = TradeDB()
|
|
hb.ensure_holding_tables(db)
|
|
|
|
items = hb.load_watchlist()
|
|
if args.code:
|
|
items = [i for i in items if i["code"] == args.code]
|
|
if not items:
|
|
print(f"종목을 찾을 수 없습니다: {args.code}")
|
|
db.close()
|
|
return
|
|
|
|
for item in items:
|
|
code = item["code"]
|
|
name = item["name"]
|
|
candles = hb.get_stored_candles(db, code, args.start, args.end)
|
|
if len(candles) < 20:
|
|
print(f"⚠️ [{code}] {name}: 봉 부족 ({len(candles)}개) → 스킵. 먼저 캔들을 수집하세요.")
|
|
print(f" 실행: python3 holding_bot.py fetch --start {args.start} --end {args.end}")
|
|
continue
|
|
run_search(code, name, candles, args.mode,
|
|
args.top, args.min_trades, args.apply, db)
|
|
|
|
db.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|