#!/usr/bin/env python3 """ kis_holding_ver1.py — 홀딩 전략 V1 (RSI 3단계 분할매수 · 횡보장 특화) ======================================================================= holding_bot.py(추세추종)의 대응 버전. 추세 판단 없이 RSI만으로 진입, 가격이 더 빠질수록 비중을 늘려가는 전통적 '물타기(분할매수)' 전략. 전략 개요 --------- 진입 (분할매수): · RSI ≤ rsi_buy1 → 1단계 매수 (slot_money × buy1_ratio) · 보유 중 RSI ≤ rsi_buy2 → 2단계 추가매수 (× buy2_ratio) · 보유 중 RSI ≤ rsi_buy3 → 3단계 추가매수 (× buy3_ratio) · 낙폭 필터(ath/year/w52)로 "충분히 싼 구간"에만 진입 가능 청산: · 평단가 대비 +take_profit_pct% → 익절 · RSI ≥ rsi_sell → 과열 청산 · 평단가 대비 −stop_loss_pct% → 손절 holding_bot.py와 동일한 DB 테이블(holding_candles, holding_stock_config) 사용. 실행 예시: python3 kis_holding_ver1.py --code 005930 \\ --start 2024-01-01 --end 2026-03-06 """ import sys, os, json, argparse from datetime import date as _date, timedelta as _td, datetime from typing import List, Dict, Optional ROOT = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, ROOT) import logging logger = logging.getLogger(__name__) # holding_bot.py와 DB/캔들 공통 함수 공유 from holding_bot import ( ensure_holding_tables, get_stored_candles, _rsi_series, ) from database import TradeDB # ───────────────────────────────────────────────────────────────────────────── # 기본 파라미터 (V1 전용 — MA/추세 관련 파라미터 없음) # ───────────────────────────────────────────────────────────────────────────── DEFAULT_V1_CONFIG: Dict = { "rsi_period": 14.0, # ── 진입 RSI 임계값 (rsi_buy1 > rsi_buy2 > rsi_buy3 이어야 함) ────────── "rsi_buy1": 50.0, # 1단계: RSI ≤ 이 값 → 처음 매수 "rsi_buy2": 40.0, # 2단계: RSI ≤ 이 값 → 추가매수 "rsi_buy3": 30.0, # 3단계: RSI ≤ 이 값 → 최종 추가매수 # ── 청산 기준 ──────────────────────────────────────────────────────────── "rsi_sell": 75.0, # RSI 과열 청산 "take_profit_pct": 15.0, # 평단가 대비 익절 % "stop_loss_pct": 10.0, # 평단가 대비 손절 % # ── 투자금 / 분할 비율 ──────────────────────────────────────────────────── "slot_money": 3_000_000.0, "buy1_ratio": 0.4, # 1단계 투자금 비율 (40%) "buy2_ratio": 0.35, # 2단계 투자금 비율 (35%) "buy3_ratio": 0.25, # 3단계 투자금 비율 (25%) # ── 비용 ──────────────────────────────────────────────────────────────── "fee_rate": 0.0015, # 수수료율 (편도) "sell_tax": 0.0018, # 증권거래세 # ── 낙폭 필터 (0=비활성) ───────────────────────────────────────────────── "ath_drop_min_pct": 0.0, # 역대 최고가 대비 최소 낙폭 % "year_drop_min_pct": 0.0, # 당해연도 고점 대비 최소 낙폭 % "w52_drop_min_pct": 0.0, # 52주 고점 대비 최소 낙폭 % } # ───────────────────────────────────────────────────────────────────────────── # 백테스트 엔진 # ───────────────────────────────────────────────────────────────────────────── def run_backtest_v1(candles: List[Dict], cfg: Dict) -> Dict: """ V1 RSI 분할매수 전략 백테스트. 실행가: 신호봉 다음 봉 시가 (1봉 지연, 실매매와 동일) 수수료: 편도 fee_rate × 2 + 매도 시 sell_tax """ rsi_period = int(cfg.get("rsi_period", DEFAULT_V1_CONFIG["rsi_period"])) rsi_buy1 = float(cfg.get("rsi_buy1", DEFAULT_V1_CONFIG["rsi_buy1"])) rsi_buy2 = float(cfg.get("rsi_buy2", DEFAULT_V1_CONFIG["rsi_buy2"])) rsi_buy3 = float(cfg.get("rsi_buy3", DEFAULT_V1_CONFIG["rsi_buy3"])) rsi_sell = float(cfg.get("rsi_sell", DEFAULT_V1_CONFIG["rsi_sell"])) tp_pct = float(cfg.get("take_profit_pct",DEFAULT_V1_CONFIG["take_profit_pct"])) sl_pct = float(cfg.get("stop_loss_pct", DEFAULT_V1_CONFIG["stop_loss_pct"])) slot_money = float(cfg.get("slot_money", DEFAULT_V1_CONFIG["slot_money"])) buy1_r = float(cfg.get("buy1_ratio", DEFAULT_V1_CONFIG["buy1_ratio"])) buy2_r = float(cfg.get("buy2_ratio", DEFAULT_V1_CONFIG["buy2_ratio"])) buy3_r = float(cfg.get("buy3_ratio", DEFAULT_V1_CONFIG["buy3_ratio"])) fee_rate = float(cfg.get("fee_rate", DEFAULT_V1_CONFIG["fee_rate"])) sell_tax = float(cfg.get("sell_tax", DEFAULT_V1_CONFIG["sell_tax"])) ath_drop_min = float(cfg.get("ath_drop_min_pct", DEFAULT_V1_CONFIG["ath_drop_min_pct"])) year_drop_min = float(cfg.get("year_drop_min_pct", DEFAULT_V1_CONFIG["year_drop_min_pct"])) w52_drop_min = float(cfg.get("w52_drop_min_pct", DEFAULT_V1_CONFIG["w52_drop_min_pct"])) if len(candles) < rsi_period + 5: return {"error": f"봉 부족: {len(candles)}개 (최소 {rsi_period + 5}개 필요)"} closes = [float(c["close"]) for c in candles] highs = [float(c["high"]) for c in candles] lows = [float(c["low"]) for c in candles] opens = [float(c["open"]) for c in candles] dates = [str(c["candle_date"])[:10] for c in candles] rsis = _rsi_series(closes, rsi_period) # ── 컨텍스트 배열 사전 계산 (ATH / 연도 고점 / 52주 고점) ───────────────── ath_arr: List[float] = [] year_arr: List[float] = [] w52_arr: List[float] = [] _year_max: Dict[str, float] = {} _ath_run = 0.0 for i in range(len(candles)): h = highs[i]; y = dates[i][:4] _ath_run = max(_ath_run, h) _year_max[y] = max(_year_max.get(y, 0.0), h) # 52주(약 252거래일) 슬라이딩 윈도우 w52_s = max(0, i - 251) try: cutoff = str(_date.fromisoformat(dates[i]) - _td(days=365)) tmp = i while tmp > 0 and dates[tmp - 1] >= cutoff: tmp -= 1 w52_s = tmp except Exception: pass ath_arr.append(_ath_run) year_arr.append(_year_max[y]) w52_arr.append(max(highs[w52_s:i + 1])) # ── 시뮬레이션 ──────────────────────────────────────────────────────────── position = None # None 또는 Dict trades: List[Dict] = [] equity: List[Dict] = [] cum_pnl = 0.0 start_i = rsi_period + 1 for i in range(start_i, len(candles) - 1): rsi = rsis[i] if rsi is None: continue close = closes[i] next_open = float(opens[i + 1]) if opens[i + 1] > 0 else close if next_open <= 0: continue date_str = dates[i] def _drop(ref: float) -> float: """현재가 기준 고점 대비 낙폭 %""" return (ref - close) / ref * 100 if ref > 0 else 0.0 # ─── 보유 중: 청산 먼저 체크, 그 다음 추가매수 ─────────────────────── if position is not None: avg = position["avg"] qty = position["qty"] stage = position["stage"] profit_pct_now = (close - avg) / avg * 100 if avg > 0 else 0.0 # 청산 조건 sell_reason = None if profit_pct_now >= tp_pct: sell_reason = f"익절(+{profit_pct_now:.1f}%)" elif rsi >= rsi_sell: sell_reason = f"RSI과열({rsi:.1f})" elif profit_pct_now <= -sl_pct: sell_reason = f"손절({profit_pct_now:.1f}%)" if sell_reason: exit_price = next_open fee = exit_price * qty * (fee_rate + sell_tax) pnl = (exit_price - avg) * qty - fee cum_pnl += pnl trades.append({ "buy_date": dates[position["entry_i"]], "sell_date": dates[i + 1], "avg_price": round(avg), "exit_price": round(exit_price), "qty": qty, "pnl": round(pnl), "hold_days": i + 1 - position["entry_i"], "reason": sell_reason, "stage": stage, "ath_drop": position.get("ath_drop", 0), }) equity.append({"date": dates[i + 1], "cum_pnl": round(cum_pnl)}) position = None else: # ─ 추가매수 (분할매수 핵심) ───────────────────────────────── # 2단계: 더 빠져서 rsi_buy2 이하가 됐을 때 추가 if stage == 1 and rsi <= rsi_buy2: add_inv = slot_money * buy2_r add_qty = max(1, int(add_inv / next_open)) add_cost = next_open * add_qty * (1 + fee_rate) new_qty = qty + add_qty position["avg"] = (avg * qty + next_open * add_qty) / new_qty position["qty"] = new_qty position["cost"] += add_cost position["stage"] = 2 # 3단계: 더 빠져서 rsi_buy3 이하가 됐을 때 추가 elif stage == 2 and rsi <= rsi_buy3: add_inv = slot_money * buy3_r add_qty = max(1, int(add_inv / next_open)) add_cost = next_open * add_qty * (1 + fee_rate) new_qty = qty + add_qty position["avg"] = (avg * qty + next_open * add_qty) / new_qty position["qty"] = new_qty position["cost"] += add_cost position["stage"] = 3 continue # 보유 중엔 신규 진입 스킵 # ─── 미보유: 낙폭 필터 → 1단계 진입 ────────────────────────────────── if ath_drop_min > 0 and _drop(ath_arr[i]) < ath_drop_min: continue if year_drop_min > 0 and _drop(year_arr[i]) < year_drop_min: continue if w52_drop_min > 0 and _drop(w52_arr[i]) < w52_drop_min: continue if rsi > rsi_buy1: continue invest = slot_money * buy1_r qty = max(1, int(invest / next_open)) cost = next_open * qty * (1 + fee_rate) position = { "avg": next_open, "qty": qty, "cost": cost, "stage": 1, "entry_i": i + 1, "ath_drop": round(_drop(ath_arr[i]), 1), } # ── 기간 종료 강제 청산 ─────────────────────────────────────────────────── if position is not None and candles: exit_price = closes[-1] avg = position["avg"]; qty = position["qty"] fee = exit_price * qty * (fee_rate + sell_tax) pnl = (exit_price - avg) * qty - fee cum_pnl += pnl trades.append({ "buy_date": dates[position["entry_i"]], "sell_date": dates[-1], "avg_price": round(avg), "exit_price": round(exit_price), "qty": qty, "pnl": round(pnl), "hold_days": len(candles) - 1 - position["entry_i"], "reason": "기간종료", "stage": position["stage"], "ath_drop": position.get("ath_drop", 0), }) equity.append({"date": dates[-1], "cum_pnl": round(cum_pnl)}) # ── 요약 통계 ────────────────────────────────────────────────────────────── total = len(trades) wins = [t for t in trades if t["pnl"] > 0] losses = [t for t in trades if t["pnl"] < 0] gross_profit = sum(t["pnl"] for t in wins) gross_loss = abs(sum(t["pnl"] for t in losses)) pf = round(gross_profit / gross_loss, 2) if gross_loss > 0 else 9999.0 peak_eq = 0.0; mdd = 0.0; run_pnl = 0.0 for t in trades: run_pnl += t["pnl"] peak_eq = max(peak_eq, run_pnl) mdd = max(mdd, peak_eq - run_pnl) # Buy & Hold 비교 (첫 진입 시점 기준) bnh_pct = 0.0; bnh_pnl = 0.0; bot_pct = 0.0 if candles and slot_money > 0: first_p = float(candles[start_i]["open"]) last_p = float(candles[-1]["close"]) if first_p > 0: bnh_qty = max(1, int(slot_money / first_p)) bnh_pnl = round((last_p - first_p) * bnh_qty) bnh_pct = round((last_p - first_p) / first_p * 100, 2) total_pnl = sum(t["pnl"] for t in trades) bot_pct = round(total_pnl / slot_money * 100, 2) if slot_money > 0 else 0.0 reason_dist: Dict[str, int] = {} for t in trades: r = t["reason"].split("(")[0] reason_dist[r] = reason_dist.get(r, 0) + 1 avg_hold = round(sum(t["hold_days"] for t in trades) / total, 1) if total else 0 return { "candle_count": len(candles), "summary": { "total_trades": total, "win_rate": round(len(wins) / total * 100, 1) if total else 0, "total_pnl": round(total_pnl), "profit_factor": pf, "max_drawdown": round(mdd), "avg_hold_days": avg_hold, "reason_dist": reason_dist, "bnh_pct": bnh_pct, "bnh_pnl": bnh_pnl, "bot_pct": bot_pct, }, "equity": equity[-200:], "trades": trades[-200:], } # ───────────────────────────────────────────────────────────────────────────── # 파라미터 Grid Search # ───────────────────────────────────────────────────────────────────────────── def run_param_search_v1( candles: List[Dict], grid: Optional[Dict] = None, min_trades: int = 2, base_cfg: Optional[Dict] = None, ) -> List[Dict]: """ V1 전략 파라미터 Grid Search. base_cfg: 탐색 그리드에 없는 파라미터 기준값. 웹 카드에서 전달받은 사용자 설정값 사용. """ from itertools import product as iproduct if grid is None: grid = { # ── 진입 RSI 임계값 ────────────────────────────────────────────── "rsi_buy1": [60, 55, 50, 45], "rsi_buy2": [50, 45, 40, 35], "rsi_buy3": [40, 35, 30, 25], # ── 매도 RSI ──────────────────────────────────────────────────── "rsi_sell": [70, 75, 80], # ── 익절/손절 ──────────────────────────────────────────────────── "take_profit_pct": [10.0, 15.0, 20.0, 30.0], "stop_loss_pct": [7.0, 10.0, 15.0], # ── ATH 낙폭 필터 ───────────────────────────────────────────── "ath_drop_min_pct": [0.0, 20.0], } keys = list(grid.keys()) combos = list(iproduct(*[grid[k] for k in keys])) _base = dict(DEFAULT_V1_CONFIG) if base_cfg: _base.update(base_cfg) results = [] for vals in combos: cfg = dict(_base) cfg.update(dict(zip(keys, vals))) # rsi_buy1 > rsi_buy2 > rsi_buy3 조건 보정 if not (cfg["rsi_buy1"] > cfg["rsi_buy2"] > cfg["rsi_buy3"]): continue res = run_backtest_v1(candles, cfg) if "error" in res: continue 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"], }) results.sort(key=lambda x: x["total_pnl"], reverse=True) return results # ───────────────────────────────────────────────────────────────────────────── # CLI 진입점 # ───────────────────────────────────────────────────────────────────────────── def main(): today = datetime.now().strftime("%Y-%m-%d") year_ago = (datetime.now() - _td(days=365)).strftime("%Y-%m-%d") parser = argparse.ArgumentParser( description="홀딩 V1 (RSI 분할매수) 백테스트 · 파라미터탐색" ) parser.add_argument("--code", required=True, help="종목 코드 (예: 005930)") parser.add_argument("--start", default=year_ago, help="시작일 YYYY-MM-DD") parser.add_argument("--end", default=today, help="종료일 YYYY-MM-DD") parser.add_argument("--search", action="store_true", help="파라미터 탐색 모드") parser.add_argument("--min_trades", default=2, type=int, help="탐색 최소 거래 수") parser.add_argument("--top", default=20, type=int, help="탐색 결과 상위 N개") parser.add_argument("--rsi_buy1", default=None, type=float) parser.add_argument("--rsi_buy2", default=None, type=float) parser.add_argument("--rsi_buy3", default=None, type=float) parser.add_argument("--rsi_sell", default=None, type=float) parser.add_argument("--tp", default=None, type=float, dest="take_profit_pct") parser.add_argument("--sl", default=None, type=float, dest="stop_loss_pct") parser.add_argument("--ath_drop", default=None, type=float, dest="ath_drop_min_pct") args = parser.parse_args() import logging as _lg _lg.getLogger("TradeDB").setLevel(_lg.WARNING) db = TradeDB() ensure_holding_tables(db) candles = get_stored_candles(db, args.code, args.start, args.end) db.close() if not candles: print(f"❌ {args.code} 캔들 없음 (holding_candles 테이블 확인)") return print(f"✅ {args.code} | {len(candles)}봉 ({candles[0]['candle_date']} ~ {candles[-1]['candle_date']})") # CLI 파라미터 오버라이드 override = {} for k in ["rsi_buy1", "rsi_buy2", "rsi_buy3", "rsi_sell", "take_profit_pct", "stop_loss_pct", "ath_drop_min_pct"]: v = getattr(args, k, None) if v is not None: override[k] = v if args.search: print(f"\n🔍 V1 파라미터 탐색 중…") results = run_param_search_v1(candles, min_trades=args.min_trades, base_cfg=override or None) if not results: print("⚠️ 유효 결과 없음") return print(f"\n{'='*70}") print(f" 🏆 RSI 분할매수 V1 — TOP {min(args.top, len(results))}") print(f"{'='*70}") keys = list(results[0]["params"].keys()) hdr = " ".join(f"{k:>14}" for k in keys) print(f"{hdr} | {'손익':>10} {'승률':>6} {'거래':>5} {'PF':>5} {'보유':>6}") print("-" * (len(hdr) + 50)) for r in results[:args.top]: p = r["params"] row = " ".join(f"{p[k]:>14.4g}" for k in keys) print(f"{row} | {r['total_pnl']:>+10,.0f} {r['win_rate']:>5.1f}% " f"{r['total_trades']:>5} {r['pf']:>5.2f} {r['avg_hold']:>5.1f}일") else: cfg = dict(DEFAULT_V1_CONFIG) cfg.update(override) res = run_backtest_v1(candles, cfg) if "error" in res: print(f"❌ {res['error']}") return s = res["summary"] print(f""" ╔══════════════════════════════════════════════╗ ║ 📊 V1 RSI 분할매수 백테스트 결과 ║ ╠══════════════════════════════════════════════╣ ║ 총 거래 : {s['total_trades']:>5}건 ║ ║ 승률 : {s['win_rate']:>5.1f}% ║ ║ 순손익 : {s['total_pnl']:>+12,.0f} 원 ║ ║ PF : {s['profit_factor']:>5.2f} ║ ║ MDD : {s['max_drawdown']:>12,.0f} 원 ║ ║ 평균보유 : {s['avg_hold_days']:>5.1f}일 ║ ╠══════════════════════════════════════════════╣ ║ 봇 수익률 : {s['bot_pct']:>+8.2f}% ║ ║ B&H 수익률 : {s['bnh_pct']:>+8.2f}% ║ ║ 알파 : {round(s['bot_pct']-s['bnh_pct'],2):>+8.2f}%p ║ ╚══════════════════════════════════════════════╝""") for t in res["trades"][-10:]: print(f" {t['buy_date']} → {t['sell_date']} " f"평단:{t['avg_price']:,} 매도:{t['exit_price']:,} " f"수량:{t['qty']} 손익:{t['pnl']:+,} 단계:{t['stage']} {t['reason']}") if __name__ == "__main__": main()