From f62b2ed2378d44d53281e631854f6c0de4bbe393 Mon Sep 17 00:00:00 2001 From: Hwang Date: Wed, 11 Mar 2026 15:25:32 +0900 Subject: [PATCH] =?UTF-8?q?=EC=98=A4=ED=94=88=EB=9D=BC=EC=9A=B0=ED=84=B0?= =?UTF-8?q?=20=ED=95=98=EA=B8=B0=EC=A0=84=EC=A0=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backtest_web.py | 3287 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 3287 insertions(+) create mode 100644 backtest_web.py diff --git a/backtest_web.py b/backtest_web.py new file mode 100644 index 0000000..3583ab7 --- /dev/null +++ b/backtest_web.py @@ -0,0 +1,3287 @@ +#!/usr/bin/env python3 +""" +backtest_web.py — 매매 성과 분석 & 백테스트 웹 대시보드 +========================================================== +실행: python3 backtest_web.py +접속: http://localhost:5050 + +탭1. 실거래 분석 → trade_history 기반 실제 매매 결과 +탭2. 스캘핑 백테스트 → ws_candles 1분봉 가격 재현(Price-Replay) 백테스트 +""" + +import sys, os, math, json, logging, threading, uuid +from datetime import datetime, timedelta +from typing import List, Dict, Optional, Tuple + +sys.path.insert(0, os.path.dirname(__file__)) +from database import TradeDB +import holding_bot as hb +import kis_holding_ver1 as hv1 # V1: RSI 3단계 분할매수 (횡보장 전략) + +from flask import Flask, jsonify, request, render_template_string + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger("backtest_web") + +# TradeDB 초기화/종료 반복 로그 억제 (백테스트 루프에서 수백 번 찍히는 것 방지) +logging.getLogger("TradeDB").setLevel(logging.WARNING) +app = Flask(__name__) + +# 60분봉 수집 백그라운드 job 상태 저장소 +_min_fetch_jobs: Dict[str, Dict] = {} + +# ──────────────────────────────────────────────────────────────────────────── +# 헬퍼 함수 +# ──────────────────────────────────────────────────────────────────────────── + +def _db() -> TradeDB: + return TradeDB() + + +def _get_fee_defaults() -> dict: + """ + DB env_config 에서 수수료/세금 기본값 로드. + 백테스트 URL 파라미터 기본값으로 사용 → DB에 값이 있으면 그걸 따름. + """ + try: + db = _db() + row = db.conn.execute( + "SELECT * FROM env_config ORDER BY id DESC LIMIT 1" + ).fetchone() + db.close() + if row: + r = dict(row) + return { + "fee_rate": float(r.get("FEE_RATE_PCT") or 0.015), + "sell_tax": float(r.get("SELL_TAX_RATE_PCT") or 0.18), + } + except Exception: + pass + return {"fee_rate": 0.015, "sell_tax": 0.18} + + +def _compute_rsi_series(closes: list, period: int = 3) -> list: + """RSI 시리즈 계산 (Wilder 스무딩)""" + rsi_list = [None] * len(closes) + if len(closes) < period + 1: + return rsi_list + deltas = [closes[i] - closes[i - 1] for i in range(1, len(closes))] + gains = [max(d, 0) for d in deltas] + losses = [max(-d, 0) for d in deltas] + avg_gain = sum(gains[:period]) / period + avg_loss = sum(losses[:period]) / period + for i in range(period, len(closes)): + idx = i - 1 # delta 배열 기준 + if i > period: + avg_gain = (avg_gain * (period - 1) + gains[idx]) / period + avg_loss = (avg_loss * (period - 1) + losses[idx]) / period + rs = avg_gain / avg_loss if avg_loss > 0 else float('inf') + rsi_val = 100 - (100 / (1 + rs)) if avg_loss > 0 else 100.0 + rsi_list[i] = rsi_val + return rsi_list + + +# ──────────────────────────────────────────────────────────────────────────── +# API: 실거래 분석 +# ──────────────────────────────────────────────────────────────────────────── + +@app.route("/api/actual", methods=["GET"]) +def api_actual(): + strategy = request.args.get("strategy", "SCALP_RSI_REVERSAL") + start = request.args.get("start", "") + end = request.args.get("end", "") + + db = _db() + try: + params = [strategy] + sql = "SELECT * FROM trade_history WHERE strategy = %s" + if start: + sql += " AND sell_date >= %s" + params.append(start + " 00:00:00") + if end: + sql += " AND sell_date <= %s" + params.append(end + " 23:59:59") + sql += " ORDER BY sell_date ASC" + rows = db.conn.execute(sql, params).fetchall() + trades = [dict(r) for r in rows] + + # 날짜 직렬화 + for t in trades: + for k in ("buy_date", "sell_date"): + if t.get(k): + t[k] = str(t[k]) + + # 누적 손익 계산 + equity = [] + cum_pnl = 0.0 + for t in trades: + cum_pnl += float(t.get("realized_pnl") or 0) + equity.append({ + "date": t["sell_date"][:10] if t.get("sell_date") else "", + "cum_pnl": round(cum_pnl), + "pnl": round(float(t.get("realized_pnl") or 0)), + }) + + # 요약 통계 + total = len(trades) + wins = [t for t in trades if float(t.get("realized_pnl") or 0) > 0] + losses = [t for t in trades if float(t.get("realized_pnl") or 0) < 0] + total_pnl = sum(float(t.get("realized_pnl") or 0) for t in trades) + avg_hold = (sum(float(t.get("hold_minutes") or 0) for t in trades) / total) if total else 0 + + win_pnl = sum(float(t.get("realized_pnl") or 0) for t in wins) + loss_pnl = sum(float(t.get("realized_pnl") or 0) for t in losses) + profit_factor = round(abs(win_pnl / loss_pnl), 2) if loss_pnl != 0 else 9999.0 + + # 최대 낙폭(MDD) + peak, mdd = 0.0, 0.0 + cum = 0.0 + for t in trades: + cum += float(t.get("realized_pnl") or 0) + if cum > peak: + peak = cum + dd = peak - cum + if dd > mdd: + mdd = dd + + # 매도 이유별 집계 + reasons: Dict[str, int] = {} + for t in trades: + r = t.get("sell_reason") or "기타" + reasons[r] = reasons.get(r, 0) + 1 + + # 일별 P&L + daily: Dict[str, float] = {} + for t in trades: + day = (t.get("sell_date") or "")[:10] + if day: + daily[day] = daily.get(day, 0) + float(t.get("realized_pnl") or 0) + daily_list = [{"date": d, "pnl": round(v)} for d, v in sorted(daily.items())] + + # 종목별 상위 손익 + code_pnl: Dict[str, float] = {} + code_name: Dict[str, str] = {} + for t in trades: + c = t["code"] + code_pnl[c] = code_pnl.get(c, 0) + float(t.get("realized_pnl") or 0) + code_name[c] = t.get("name") or c + top_codes = sorted(code_pnl.items(), key=lambda x: x[1], reverse=True)[:10] + top_list = [{"code": c, "name": code_name[c], "pnl": round(v)} for c, v in top_codes] + + return jsonify({ + "summary": { + "total_trades": total, + "win_trades": len(wins), + "loss_trades": len(losses), + "win_rate": round(len(wins) / total * 100, 1) if total else 0, + "total_pnl": round(total_pnl), + "avg_hold_min": round(avg_hold, 1), + "profit_factor": round(profit_factor, 2), + "max_drawdown": round(mdd), + }, + "equity": equity, + "daily": daily_list, + "reasons": reasons, + "top_codes": top_list, + "trades": trades[-200:], # 최근 200건 + }) + finally: + db.close() + + +# ──────────────────────────────────────────────────────────────────────────── +# API: 스캘핑 가격 재현 백테스트 (ws_candles 기반) +# ──────────────────────────────────────────────────────────────────────────── + +def _t2dt(t: str) -> datetime: + return datetime.strptime(t, "%Y%m%d%H%M") + +@app.route("/api/backtest/scalping", methods=["GET"]) +def api_backtest_scalping(): + start = request.args.get("start", "") + end = request.args.get("end", "") + rsi_period = int(request.args.get("rsi_period", 3)) + rsi_oversold = float(request.args.get("rsi_oversold", 25)) + sl_pct = float(request.args.get("sl_pct", 1.5)) / 100 # 손절 % + tp_pct = float(request.args.get("tp_pct", 1.5)) / 100 # 익절 % + drop_rate = float(request.args.get("drop_rate", 1.5)) / 100 # 최소 낙폭(당일 시가→저가) + slot_money = float(request.args.get("slot_money", 1_000_000)) # 1회 투자금 + _fee_d = _get_fee_defaults() + fee_rate = float(request.args.get("fee_rate", _fee_d["fee_rate"])) / 100 + sell_tax = float(request.args.get("sell_tax", _fee_d["sell_tax"])) / 100 + cooldown_min = int(request.args.get("cooldown_min", 10)) # 손익 후 재진입 금지(분) + vol_mult = float(request.args.get("vol_mult", 0)) # 거래량 필터(0=비활성) + trail_trigger = float(request.args.get("trail_trigger", 0.7)) / 100 # 트레일링 발동 수익률 + trail_stop = float(request.args.get("trail_stop", 0.4)) / 100 # 트레일링 추적폭 + time_start_hm = int(request.args.get("time_start", 900)) # 매수 가능 시작(HHMM) + time_end_hm = int(request.args.get("time_end", 1400)) # 매수 가능 종료(HHMM) + max_daily = int(request.args.get("max_daily", 3)) # 종목당 일일 최대 거래횟수 + + db = _db() + try: + # 날짜 → candle_time 형식 (YYYYMMDDHHMI) 변환 + start_key = (start.replace("-", "") + "0000") if start else "20260101" + end_key = (end.replace("-", "") + "2359") if end else "99991231" + + codes_raw = db.conn.execute( + "SELECT DISTINCT code FROM ws_candles WHERE timeframe=1 " + "AND candle_time >= %s AND candle_time <= %s ORDER BY code", + [start_key, end_key] + ).fetchall() + codes = [r['code'] for r in codes_raw] + + all_virtual_trades: List[Dict] = [] + + for code in codes: + rows = db.conn.execute( + "SELECT candle_time, open, high, low, close, volume " + "FROM ws_candles " + "WHERE timeframe=1 AND code=%s " + "AND candle_time >= %s AND candle_time <= %s " + "AND is_confirmed=1 " + "ORDER BY candle_time ASC", + [code, start_key, end_key] + ).fetchall() + if len(rows) < rsi_period + 5: + continue + + candles = [dict(r) for r in rows] + closes = [float(c["close"]) for c in candles] + volumes = [float(c["volume"]) for c in candles] + rsis = _compute_rsi_series(closes, rsi_period) + + position = None + last_exit_dt: Dict[str, datetime] = {} # day → 마지막 청산 시각 (쿨다운) + daily_cnt: Dict[str, int] = {} # day → 당일 거래횟수 + + # 선행 편향 없는 당일 OHLC 누적값 + cur_day = None + running_open = 0.0 + running_low = 0.0 + + for i in range(rsi_period + 1, len(candles)): + c = candles[i] + day = c["candle_time"][:8] + hm = int(c["candle_time"][8:12]) # HHMM 정수 + cl = float(c["close"]) + lo = float(c["low"]) + hi = float(c["high"]) + vol = volumes[i] + + # ── 당일 누적값 갱신 (look-ahead 없이 실시간 계산) ──────── + if day != cur_day: + cur_day = day + running_open = float(c["open"]) + running_low = lo + else: + running_low = min(running_low, lo) + + # ── 당일 마지막 봉 여부 ──────────────────────────────────── + is_eod = (i == len(candles) - 1) or (candles[i + 1]["candle_time"][:8] != day) + + # ───────────────────────────────────────────────────────── + # 포지션 보유 중: 청산 체크 + # ───────────────────────────────────────────────────────── + if position is not None: + max_price = max(position["max_price"], hi) + position["max_price"] = max_price + + reason = None + exit_price = cl + + # 손절: 캔들 저가가 손절선 이하 + if lo <= position["stop"]: + reason = "손절" + exit_price = position["stop"] # limit-stop 가정 + # 익절: 캔들 고가가 익절선 이상 + elif hi >= position["target"]: + reason = "익절" + exit_price = position["target"] + # 트레일링스탑: 고점 대비 하락 + elif trail_trigger > 0 and max_price >= position["entry_price"] * (1 + trail_trigger): + ts = max_price * (1 - trail_stop) + if cl <= ts: + reason = "트레일링스탑" + exit_price = cl + # 장마감 강제청산 (당일 마지막 봉) + if reason is None and is_eod: + reason = "장마감청산" + exit_price = cl + + if reason: + qty = position["qty"] + buy_amt = position["entry_price"] * qty + sell_amt = exit_price * qty + pnl = (sell_amt - buy_amt + - buy_amt * fee_rate + - sell_amt * fee_rate + - sell_amt * sell_tax) + hold_min = int((_t2dt(c["candle_time"]) - + _t2dt(position["entry_time"])).total_seconds() / 60) + all_virtual_trades.append({ + "code": code, + "buy_time": position["entry_time"], + "sell_time": c["candle_time"], + "buy_price": position["entry_price"], + "sell_price": round(exit_price, 2), + "qty": qty, + "pnl": round(pnl), + "profit_rate": round((exit_price - position["entry_price"]) + / position["entry_price"] * 100, 2), + "hold_min": hold_min, + "sell_reason": reason, + "rsi_entry": round(position["rsi"], 1), + }) + last_exit_dt[day] = _t2dt(c["candle_time"]) + position = None + continue # 포지션 보유 중이면 신규 진입 건너뜀 + + # ───────────────────────────────────────────────────────── + # 포지션 없음: 매수 신호 체크 + # ───────────────────────────────────────────────────────── + + # 거래 가능 시간 필터 + if hm < time_start_hm or hm >= time_end_hm: + continue + + # 쿨다운 체크 (당일 마지막 청산 기준) + if day in last_exit_dt: + elapsed = (_t2dt(c["candle_time"]) - last_exit_dt[day]).total_seconds() / 60 + if elapsed < cooldown_min: + continue + + # 일일 최대 거래횟수 초과 + if daily_cnt.get(day, 0) >= max_daily: + continue + + # RSI 과매도 조건 + rsi = rsis[i] + if rsi is None or rsi > rsi_oversold: + continue + + # 반전 캔들 패턴: 이전봉 음봉 + 현재봉 양봉 + prev_c = candles[i - 1] + prev_bear = float(prev_c["close"]) < float(prev_c["open"]) + curr_bull = cl > float(c["open"]) + if not (prev_bear and curr_bull): + continue + + # 당일 낙폭 필터 (look-ahead 없는 running_low 사용) + if running_open <= 0: + continue + dr = (running_open - running_low) / running_open + if dr < drop_rate: + continue + + # 거래량 급증 필터 + if vol_mult > 0: + win = max(1, min(20, i)) + vol_avg = sum(volumes[i - win:i]) / win + if vol_avg > 0 and vol < vol_avg * vol_mult: + continue + + # ─── 진입 가격: 신호봉 다음 봉 시가 (현실적 체결 가정) ─── + if i + 1 >= len(candles): + continue + next_c = candles[i + 1] + if next_c["candle_time"][:8] != day: # 다음봉이 익일이면 스킵 + continue + + entry_price = float(next_c["open"]) + if entry_price <= 0: + continue + qty = max(1, int(slot_money / entry_price)) + stop = entry_price * (1 - sl_pct) + target = entry_price * (1 + tp_pct) + position = { + "entry_price": entry_price, + "entry_time": next_c["candle_time"], + "qty": qty, + "stop": stop, + "target": target, + "max_price": entry_price, + "rsi": rsi, + } + daily_cnt[day] = daily_cnt.get(day, 0) + 1 + + # ───────────────────────────────────────────────────────────────── + # 결과 집계 + # ───────────────────────────────────────────────────────────────── + all_virtual_trades.sort(key=lambda x: x["sell_time"]) + + equity = [] + cum = 0.0 + for t in all_virtual_trades: + cum += t["pnl"] + equity.append({"date": t["sell_time"][:8], "cum_pnl": round(cum), "pnl": t["pnl"]}) + + total = len(all_virtual_trades) + wins = [t for t in all_virtual_trades if t["pnl"] > 0] + losses = [t for t in all_virtual_trades if t["pnl"] < 0] + total_pnl = sum(t["pnl"] for t in all_virtual_trades) + avg_hold = (sum(t["hold_min"] for t in all_virtual_trades) / total) if total else 0 + win_pnl = sum(t["pnl"] for t in wins) + loss_pnl = sum(t["pnl"] for t in losses) + pf = round(abs(win_pnl / loss_pnl), 2) if loss_pnl != 0 else 9999.0 + + peak, mdd, cum = 0.0, 0.0, 0.0 + for t in all_virtual_trades: + cum += t["pnl"] + if cum > peak: peak = cum + dd = peak - cum + if dd > mdd: mdd = dd + + reasons: Dict[str, int] = {} + for t in all_virtual_trades: + reasons[t["sell_reason"]] = reasons.get(t["sell_reason"], 0) + 1 + + daily: Dict[str, float] = {} + for t in all_virtual_trades: + d8 = t["sell_time"][:8] + daily[d8] = daily.get(d8, 0) + t["pnl"] + daily_list = [{"date": d[:4]+"-"+d[4:6]+"-"+d[6:], "pnl": round(v)} + for d, v in sorted(daily.items())] + + return jsonify({ + "params": { + "rsi_period": rsi_period, + "rsi_oversold": rsi_oversold, + "sl_pct": sl_pct * 100, + "tp_pct": tp_pct * 100, + "drop_rate": drop_rate * 100, + "slot_money": slot_money, + "cooldown_min": cooldown_min, + "vol_mult": vol_mult, + "trail_trigger": trail_trigger * 100, + "trail_stop": trail_stop * 100, + "time_window": f"{time_start_hm:04d}-{time_end_hm:04d}", + "max_daily": max_daily, + "codes_analyzed": len(codes), + }, + "summary": { + "total_trades": total, + "win_trades": len(wins), + "loss_trades": len(losses), + "win_rate": round(len(wins) / total * 100, 1) if total else 0, + "total_pnl": round(total_pnl), + "avg_hold_min": round(avg_hold, 1), + "profit_factor": round(pf, 2), + "max_drawdown": round(mdd), + }, + "equity": equity, + "daily": daily_list, + "reasons": reasons, + "trades": all_virtual_trades[-200:], + }) + finally: + db.close() + + +# ──────────────────────────────────────────────────────────────────────────── +# API: 꼬리잡기 가격 재현 백테스트 (ws_candles 3분봉 기반) +# ──────────────────────────────────────────────────────────────────────────── + +@app.route("/api/backtest/tail", methods=["GET"]) +def api_backtest_tail(): + """ + 꼬리잡기 전략 가격 재현 백테스트. + entry 조건: 당일 낙폭(drop_rate) + 회복률(recovery_ratio) + 망치봉 꼬리 + RSI + exit 조건: 손절 / 익절 / 어깨 컷(trailing) / 장 마감 강제 청산 + """ + start = request.args.get("start", "") + end = request.args.get("end", "") + rsi_period = int( request.args.get("rsi_period", 14)) + rsi_threshold = float(request.args.get("rsi_threshold", 78)) # RSI 과열 기준 + min_drop_rate = float(request.args.get("min_drop_rate", 3.0)) / 100 # 최소 당일 낙폭 + min_recovery_ratio = float(request.args.get("min_recovery_ratio",50)) / 100 # 최소 당일 회복률 + max_rec_3m = float(request.args.get("max_rec_3m", 80)) / 100 # 3분봉 최대 회복위치 + tail_ratio_min = float(request.args.get("tail_ratio_min", 1.5)) # 꼬리/몸통 비율 + tail_pct_min = float(request.args.get("tail_pct_min", 0.3)) / 100 # 꼬리 최소 % + sl_pct = float(request.args.get("sl_pct", 3.0)) / 100 # 손절 % + tp_pct = float(request.args.get("tp_pct", 5.0)) / 100 # 익절 % + shoulder_min_high = float(request.args.get("shoulder_min_high", 1.5)) / 100 # 어깨 발동 최소 이익 + shoulder_cut_pct = float(request.args.get("shoulder_cut_pct", 3.0)) / 100 # 고점 대비 하락 시 매도 + high_chase_thr = float(request.args.get("high_chase_thr", 96.0)) / 100 # 고점 추격 방지 + slot_money = float(request.args.get("slot_money", 1_000_000)) + _fee_d = _get_fee_defaults() + fee_rate = float(request.args.get("fee_rate", _fee_d["fee_rate"])) / 100 + sell_tax = float(request.args.get("sell_tax", _fee_d["sell_tax"])) / 100 + cooldown_min = int( request.args.get("cooldown_min", 30)) + time_start_hm = int( request.args.get("time_start", 930)) + time_end_hm = int( request.args.get("time_end", 1500)) + max_daily = int( request.args.get("max_daily", 3)) + + db = _db() + try: + start_key = (start.replace("-", "") + "0000") if start else "20260101" + end_key = (end.replace("-", "") + "2359") if end else "99991231" + + codes_raw = db.conn.execute( + "SELECT DISTINCT code FROM ws_candles WHERE timeframe=3 " + "AND candle_time >= %s AND candle_time <= %s ORDER BY code", + [start_key, end_key] + ).fetchall() + codes = [r["code"] for r in codes_raw] + + all_trades: List[Dict] = [] + + for code in codes: + rows = db.conn.execute( + "SELECT candle_time, open, high, low, close, volume " + "FROM ws_candles " + "WHERE timeframe=3 AND code=%s " + "AND candle_time >= %s AND candle_time <= %s " + "AND is_confirmed=1 " + "ORDER BY candle_time ASC", + [code, start_key, end_key] + ).fetchall() + if len(rows) < rsi_period + 5: + continue + + candles = [dict(r) for r in rows] + closes = [float(c["close"]) for c in candles] + rsis = _compute_rsi_series(closes, rsi_period) + + position = None + last_exit_dt: Dict[str, datetime] = {} + daily_cnt: Dict[str, int] = {} + + # look-ahead 없는 당일 누적 OHLC + cur_day = None + running_open = 0.0 + running_high = 0.0 + running_low = 0.0 + + for i in range(rsi_period + 1, len(candles)): + c = candles[i] + day = c["candle_time"][:8] + hm = int(c["candle_time"][8:12]) + op = float(c["open"]) + hi = float(c["high"]) + lo = float(c["low"]) + cl = float(c["close"]) + + # ── 당일 누적 OHLC 갱신 (선행 편향 없음) ───────────────── + if day != cur_day: + cur_day = day + running_open = op + running_high = hi + running_low = lo if lo > 0 else hi + else: + running_high = max(running_high, hi) + if lo > 0: + running_low = min(running_low, lo) + + # ── 마지막 봉 여부 ──────────────────────────────────────── + is_eod = (i == len(candles) - 1) or (candles[i + 1]["candle_time"][:8] != day) + + # ───────────────────────────────────────────────────────── + # 포지션 보유 중: 청산 체크 + # ───────────────────────────────────────────────────────── + if position is not None: + max_p = max(position["max_price"], hi) + position["max_price"] = max_p + + reason = None + exit_price = cl + + # 손절: 저가가 손절선 이하 + if lo > 0 and lo <= position["stop"]: + reason = "손절" + exit_price = position["stop"] + # 익절: 고가가 익절선 이상 + elif hi >= position["target"]: + reason = "익절" + exit_price = position["target"] + # 어깨 컷(trailing): 충분히 오른 뒤 고점에서 일정 이상 하락 + elif (max_p >= position["entry_price"] * (1 + shoulder_min_high) + and cl <= max_p * (1 - shoulder_cut_pct)): + reason = "어깨컷" + exit_price = cl + # 장 마감 강제 청산 + elif is_eod: + reason = "장마감" + exit_price = cl + + if reason: + ep = position["entry_price"] + qty = max(1, int(slot_money / ep)) + fee = (ep + exit_price) * qty * fee_rate + tax = exit_price * qty * sell_tax + pnl = (exit_price - ep) * qty - fee - tax + hold = round(( + _t2dt(c["candle_time"]) - + _t2dt(position["entry_time"]) + ).total_seconds() / 60, 1) + + all_trades.append({ + "code": code, + "entry_time": position["entry_time"], + "exit_time": c["candle_time"], + "entry": round(ep), + "exit": round(exit_price), + "pnl": round(pnl), + "reason": reason, + "hold_min": hold, + }) + last_exit_dt[day] = _t2dt(c["candle_time"]) + daily_cnt[day] = daily_cnt.get(day, 0) + 1 + position = None + continue # 다음 봉으로 + + # ───────────────────────────────────────────────────────── + # 포지션 없음: 매수 조건 체크 + # ───────────────────────────────────────────────────────── + if cl <= 0 or running_open <= 0: + continue + if hm < time_start_hm or hm > time_end_hm: + continue + if daily_cnt.get(day, 0) >= max_daily: + continue + + # 쿨다운: 마지막 청산 후 N분 이내 재진입 금지 + if day in last_exit_dt: + elapsed = (_t2dt(c["candle_time"]) - last_exit_dt[day]).total_seconds() / 60 + if elapsed < cooldown_min: + continue + + # ── 1. 당일 낙폭 ───────────────────────────────────────── + drop = (running_open - running_low) / running_open + if drop < min_drop_rate: + continue + + # ── 2. 당일 회복률 ───────────────────────────────────── + day_range = running_high - running_low + rec_day = (cl - running_low) / day_range if day_range > 0 else 0 + if rec_day < min_recovery_ratio: + continue + + # ── 3. 망치봉 꼬리 비율 계산 ──────────────────────────── + body_top = max(op, cl) + body_bot = min(op, cl) + body_len = body_top - body_bot if body_top > body_bot else 1.0 + tail_len = body_bot - lo if lo > 0 else 0.0 + # 꼬리 없는 봉이면 이전 봉에서 재탐색 (최대 3봉 전) + if tail_len <= 0: + for j in range(i - 1, max(i - 4, rsi_period), -1): + prev = candles[j] + o2, h2, l2, c2 = float(prev["open"]), float(prev["high"]), float(prev["low"]), float(prev["close"]) + if l2 <= 0: + continue + bt2, bb2 = max(o2, c2), min(o2, c2) + bl2 = bt2 - bb2 if bt2 > bb2 else 1.0 + tl2 = bb2 - l2 + if tl2 > 0: + tail_len = tl2 + body_len = bl2 + lo = l2 + break + + tail_ratio = tail_len / body_len + tail_pct = tail_len / lo if lo > 0 and tail_len > 0 else 0.0 + + if tail_ratio < tail_ratio_min or tail_pct < tail_pct_min: + continue + + # ── 4. 3분봉 내 회복 위치 (무릎~어깨) ────────────────── + c_range = float(c["high"]) - float(c["low"]) + rec_3m = (cl - float(c["low"])) / c_range if c_range > 0 else 0 + if not (min_recovery_ratio <= rec_3m <= max_rec_3m): + continue + + # ── 5. RSI 과열 방지 ──────────────────────────────────── + rsi_val = rsis[i] + if rsi_val is None or rsi_val >= rsi_threshold: + continue + + # ── 6. 피뢰침 방지: 고점 근접 추격 금지 ──────────────── + if cl >= running_high * high_chase_thr: + continue + + # ── 매수 실행: 다음 봉 시가 진입 ─────────────────────── + if i + 1 >= len(candles): + continue + next_c = candles[i + 1] + if next_c["candle_time"][:8] != day: + continue # 장 마감 직전 봉이면 다음날 시가 = 갭위험 → skip + + entry_price = float(next_c["open"]) + if entry_price <= 0: + entry_price = cl + + position = { + "entry_price": entry_price, + "entry_time": next_c["candle_time"], + "stop": entry_price * (1 - sl_pct), + "target": entry_price * (1 + tp_pct), + "max_price": entry_price, + } + # 진입 봉을 이미 처리했으므로 다음 인덱스로 이동 + i += 1 + + # ── 통계 집계 ────────────────────────────────────────────────── + total = len(all_trades) + wins = [t for t in all_trades if t["pnl"] > 0] + losses = [t for t in all_trades if t["pnl"] < 0] + total_pnl = sum(t["pnl"] for t in all_trades) + avg_hold = (sum(t["hold_min"] for t in all_trades) / total) if total else 0 + + win_pnl = sum(t["pnl"] for t in wins) + loss_pnl = sum(t["pnl"] for t in losses) + pf = round(abs(win_pnl / loss_pnl), 2) if loss_pnl != 0 else 9999.0 + + # MDD + peak, mdd, cum = 0.0, 0.0, 0.0 + equity, daily_map = [], {} + for t in sorted(all_trades, key=lambda x: x["exit_time"]): + cum += t["pnl"] + if cum > peak: + peak = cum + dd = peak - cum + if dd > mdd: + mdd = dd + day = t["exit_time"][:8] + equity.append({"date": f"{day[:4]}-{day[4:6]}-{day[6:]}", "cum_pnl": round(cum)}) + daily_map[day] = daily_map.get(day, 0) + t["pnl"] + + daily_list = [{"date": f"{d[:4]}-{d[4:6]}-{d[6:]}", "pnl": round(v)} + for d, v in sorted(daily_map.items())] + + reasons: Dict[str, int] = {} + for t in all_trades: + reasons[t["reason"]] = reasons.get(t["reason"], 0) + 1 + + return jsonify({ + "params": { + "start": start, "end": end, + "rsi_period": rsi_period, "rsi_threshold": rsi_threshold, + "min_drop_rate": min_drop_rate * 100, + "min_recovery_ratio": min_recovery_ratio * 100, + "max_rec_3m": max_rec_3m * 100, + "tail_ratio_min": tail_ratio_min, + "tail_pct_min": tail_pct_min * 100, + "sl_pct": sl_pct * 100, + "tp_pct": tp_pct * 100, + "shoulder_min_high": shoulder_min_high * 100, + "shoulder_cut_pct": shoulder_cut_pct * 100, + "slot_money": slot_money, + "cooldown_min": cooldown_min, + "codes_analyzed": len(codes), + }, + "summary": { + "total_trades": total, + "win_trades": len(wins), + "loss_trades": len(losses), + "win_rate": round(len(wins) / total * 100, 1) if total else 0, + "total_pnl": round(total_pnl), + "avg_hold_min": round(avg_hold, 1), + "profit_factor": round(pf, 2), + "max_drawdown": round(mdd), + }, + "equity": equity, + "daily": daily_list, + "reasons": reasons, + "trades": all_trades[-200:], + }) + finally: + db.close() + + +# ──────────────────────────────────────────────────────────────────────────── +# API: 홀딩 전략 — 관심종목 + 종목별 파라미터 + 캔들 수집 + 백테스트 + 파라미터 탐색 +# ──────────────────────────────────────────────────────────────────────────── + +def _holding_db() -> TradeDB: + db = _db() + hb.ensure_holding_tables(db) + return db + + +@app.route("/api/holding/stocks", methods=["GET"]) +def api_holding_stocks(): + """관심종목 목록 + 종목별 현재 파라미터 + 보유 봉수 반환""" + db = _holding_db() + try: + items = hb.load_watchlist() + result = [] + for item in items: + code = item["code"] + cfg = hb.get_stock_config(db, code) + cfg["name"] = item.get("name", cfg.get("name", "")) + # 보유 봉수 + row = db.conn.execute( + "SELECT COUNT(*) as cnt, MIN(candle_date) as mn, MAX(candle_date) as mx " + "FROM holding_candles WHERE code=%s", [code] + ).fetchone() + cfg["candle_count"] = int(row["cnt"]) if row else 0 + cfg["candle_min"] = str(row["mn"]) if row and row["mn"] else "" + cfg["candle_max"] = str(row["mx"]) if row and row["mx"] else "" + # 60분봉 현황 추가 + ms = hb.get_min_candle_stats(db, code, tf_min=60) + cfg["min60_count"] = ms["count"] + cfg["min60_min"] = ms["min"] + cfg["min60_max"] = ms["max"] + result.append(cfg) + return jsonify(result) + finally: + db.close() + + +@app.route("/api/holding/config/", methods=["GET", "POST"]) +def api_holding_config(code): + """GET: 종목 파라미터 조회 | POST: 파라미터 저장""" + db = _holding_db() + try: + if request.method == "GET": + cfg = hb.get_stock_config(db, code) + return jsonify(cfg) + else: + body = request.get_json(force=True) or {} + name = body.pop("name", "") + hb.set_stock_config(db, code, name, body) + return jsonify({"ok": True}) + finally: + db.close() + + +@app.route("/api/holding/candles/fetch", methods=["POST"]) +def api_holding_fetch_candles(): + """종목 일봉 캔들 KIS API로 수집 후 DB 저장""" + body = request.get_json(force=True) or {} + code = body.get("code", "") + start_date = body.get("start", "2023-01-01") + end_date = body.get("end", datetime.now().strftime("%Y-%m-%d")) + + if not code: + return jsonify({"error": "code 필수"}), 400 + + db = _holding_db() + try: + app_key, app_secret, base_url, mock = hb._get_kis_token(db) + rows = hb.fetch_daily_ohlcv(code, start_date, end_date, app_key, app_secret, base_url, mock=mock) + saved = hb.store_candles(db, code, rows) + return jsonify({"ok": True, "fetched": len(rows), "saved": saved}) + except Exception as e: + logger.error(f"캔들 수집 오류 ({code}): {e}") + return jsonify({"error": str(e)}), 500 + finally: + db.close() + + +@app.route("/api/holding/backtest", methods=["GET"]) +def api_holding_backtest(): + """홀딩 전략 백테스트 (종목별 파라미터 사용 or 요청 파라미터 오버라이드)""" + code = request.args.get("code", "") + start_date = request.args.get("start", "2023-01-01") + end_date = request.args.get("end", datetime.now().strftime("%Y-%m-%d")) + + if not code: + return jsonify({"error": "code 필수"}), 400 + + db = _holding_db() + try: + candles = hb.get_stored_candles(db, code, start_date, end_date) + if len(candles) < 20: + # 친절한 에러: DB 전체 보유 봉 수와 기간도 함께 안내 + all_candles = hb.get_stored_candles(db, code) + if not all_candles: + hint = "📥 먼저 [캔들 수집] 버튼으로 데이터를 수집하세요." + else: + first = str(all_candles[0]["candle_date"])[:10] + last = str(all_candles[-1]["candle_date"])[:10] + hint = ( + f"DB에 {len(all_candles)}봉 있음 ({first} ~ {last})\n" + f"👉 백테스트 시작일을 '{first}' 이후로 설정하세요." + ) + return jsonify({ + "error": f"봉 부족: {len(candles)}개 (최소 20개 필요)\n{hint}" + }), 400 + + # 요청 파라미터로 DB 설정 오버라이드 가능 + cfg = hb.get_stock_config(db, code) + for ck in hb.DEFAULT_STOCK_CONFIG: + v = request.args.get(ck) + if v is not None: + cfg[ck] = float(v) + + result = hb.run_backtest(candles, cfg) + result["code"] = code + result["candle_count"] = len(candles) + result["params"] = {k: cfg[k] for k in hb.DEFAULT_STOCK_CONFIG} + return jsonify(result) + finally: + db.close() + + +@app.route("/api/holding/v1/backtest", methods=["GET"]) +def api_holding_v1_backtest(): + """홀딩 V1 (RSI 분할매수) 백테스트""" + code = request.args.get("code", "") + start_date = request.args.get("start", "") + end_date = request.args.get("end", datetime.now().strftime("%Y-%m-%d")) + if not code: + return jsonify({"error": "code 필수"}), 400 + + # 카드 UI 파라미터 수집 (DEFAULT_V1_CONFIG 키 기준) + cfg = {} + for key in hv1.DEFAULT_V1_CONFIG.keys(): + val = request.args.get(key) + if val is not None: + try: + cfg[key] = float(val) + except (ValueError, TypeError): + pass + + db = _holding_db() + try: + candles = hb.get_stored_candles(db, code, start_date, end_date) + if len(candles) < 10: + return jsonify({"error": f"봉 부족: {len(candles)}개"}), 400 + res = hv1.run_backtest_v1(candles, cfg) + return jsonify(res) + finally: + db.close() + + +@app.route("/api/holding/v1/param_search", methods=["GET"]) +def api_holding_v1_param_search(): + """홀딩 V1 (RSI 분할매수) 파라미터 Grid Search""" + code = request.args.get("code", "") + start_date = request.args.get("start", "") + end_date = request.args.get("end", datetime.now().strftime("%Y-%m-%d")) + min_trades = int(request.args.get("min_trades", 2)) + if not code: + return jsonify({"error": "code 필수"}), 400 + + # 카드 UI 파라미터를 base_cfg로 (그리드 외 파라미터 고정) + base_cfg = {} + for key in hv1.DEFAULT_V1_CONFIG.keys(): + val = request.args.get(key) + if val is not None: + try: + base_cfg[key] = float(val) + except (ValueError, TypeError): + pass + + db = _holding_db() + try: + candles = hb.get_stored_candles(db, code, start_date, end_date) + if len(candles) < 20: + return jsonify({"error": f"봉 부족: {len(candles)}개"}), 400 + results = hv1.run_param_search_v1(candles, min_trades=min_trades, + base_cfg=base_cfg if base_cfg else None) + return jsonify({"code": code, "top": results[:30]}) + finally: + db.close() + + +@app.route("/api/holding/param_search", methods=["GET"]) +def api_holding_param_search(): + """홀딩 전략 파라미터 Grid Search (단일 종목)""" + code = request.args.get("code", "") + start_date = request.args.get("start", "2023-01-01") + end_date = request.args.get("end", datetime.now().strftime("%Y-%m-%d")) + min_trades = int(request.args.get("min_trades", 2)) + + if not code: + return jsonify({"error": "code 필수"}), 400 + + # 카드에서 전달된 현재 파라미터를 base_cfg로 사용 + # (탐색 그리드에 없는 rsi_sell, rsi_period, rsi_buy3 등이 사용자 값으로 고정됨) + base_cfg = {} + for key in hb.DEFAULT_STOCK_CONFIG.keys(): + val = request.args.get(key) + if val is not None: + try: + base_cfg[key] = float(val) + except (ValueError, TypeError): + pass + + db = _holding_db() + try: + candles = hb.get_stored_candles(db, code, start_date, end_date) + if len(candles) < 20: + return jsonify({"error": f"봉 부족: {len(candles)}개"}), 400 + + results = hb.run_param_search(candles, min_trades=min_trades, + base_cfg=base_cfg if base_cfg else None) + return jsonify({"code": code, "top": results[:30]}) + finally: + db.close() + + +# ──────────────────────────────────────────────────────────────────────────── +# 메인 페이지 +# ──────────────────────────────────────────────────────────────────────────── + +@app.route("/api/holding/min_candles/fetch", methods=["POST"]) +def api_holding_min_candles_fetch(): + """60분봉 수집 API: 백그라운드 스레드로 실행, 즉시 job_id 반환""" + body = request.get_json(force=True, silent=True) or {} + code = body.get("code", "") + start = body.get("start", "") + end = body.get("end", datetime.now().strftime("%Y-%m-%d")) + tf = int(body.get("tf", 60)) + + if not code or not start: + return jsonify({"error": "code, start 필수"}), 400 + + job_id = uuid.uuid4().hex[:8] + _min_fetch_jobs[job_id] = { + "status": "running", + "code": code, + "fetched": 0, # 수집한 1분봉 수 + "saved": 0, # DB에 저장된 60분봉 수 + "current_date": "", # 현재 처리 중인 날짜 + "error": None, + } + + def _run(): + db = _holding_db() + try: + app_key, app_secret, base_url, mock = hb._get_kis_token(db) + hb.fetch_and_store_min_candles( + db, code, start, end, + app_key, app_secret, base_url, + tf_min=tf, mock=mock, + progress=_min_fetch_jobs[job_id], + ) + _min_fetch_jobs[job_id]["status"] = "done" + except Exception as e: + logger.error(f"60분봉 수집 오류 ({code}): {e}") + _min_fetch_jobs[job_id]["status"] = "error" + _min_fetch_jobs[job_id]["error"] = str(e) + finally: + db.close() + + threading.Thread(target=_run, daemon=True).start() + return jsonify({"job_id": job_id, "status": "started"}) + + +@app.route("/api/holding/min_candles/status/") +def api_holding_min_candles_status(job_id: str): + """60분봉 수집 진행상황 폴링 엔드포인트""" + job = _min_fetch_jobs.get(job_id) + if not job: + return jsonify({"error": "없는 job_id"}), 404 + return jsonify(job) + + +@app.route("/api/holding/min_candles/fetch_kiwoom", methods=["POST"]) +def api_holding_min_candles_fetch_kiwoom(): + """ + 키움 REST API (ka10080) 로 60분봉 수집 → holding_min_candles 저장. + KIS와 달리 1회 호출에 900봉(약 128영업일), 연속조회로 2년치 수집 가능. + + 필요 DB env_config 키: KIWOOM_APP_KEY, KIWOOM_APP_SECRET + """ + body = request.get_json(force=True, silent=True) or {} + code = body.get("code", "") + start = body.get("start", "") + end = body.get("end", datetime.now().strftime("%Y-%m-%d")) + + if not code or not start: + return jsonify({"error": "code, start 필수"}), 400 + + job_id = uuid.uuid4().hex[:8] + _min_fetch_jobs[job_id] = { + "status": "running", + "code": code, + "source": "kiwoom", + "fetched": 0, + "saved": 0, + "current_date": "", + "error": None, + } + + def _run(): + db = _holding_db() + try: + row = db.conn.execute( + "SELECT * FROM env_config ORDER BY id DESC LIMIT 1" + ).fetchone() + if not row: + raise RuntimeError("env_config 없음") + r = dict(row) + is_mock = str(r.get("KIS_MOCK", "true")).lower() in ("true", "1", "yes") + + # KIS_MOCK 설정에 따라 키움 실전/모의 키 자동 선택 + # 우선순위: REAL/MOCK 전용 키 → 레거시 KIWOOM_APP_KEY + if is_mock: + kiwoom_key = str(r.get("KIWOOM_APP_KEY_MOCK", "") or "").strip() + kiwoom_secret = str(r.get("KIWOOM_APP_SECRET_MOCK", "") or "").strip() + mode_label = "모의" + else: + kiwoom_key = str(r.get("KIWOOM_APP_KEY_REAL", "") or "").strip() + kiwoom_secret = str(r.get("KIWOOM_APP_SECRET_REAL", "") or "").strip() + mode_label = "실전" + + # 전용 키 없으면 레거시 키로 폴백 + if not kiwoom_key or not kiwoom_secret: + kiwoom_key = str(r.get("KIWOOM_APP_KEY", "") or "").strip() + kiwoom_secret = str(r.get("KIWOOM_APP_SECRET", "") or "").strip() + mode_label += "(레거시키 사용)" + + if not kiwoom_key or not kiwoom_secret: + raise RuntimeError( + f"키움 {mode_label} API 키 미설정.\n" + "DB env_config에 KIWOOM_APP_KEY_REAL(또는 KIWOOM_APP_KEY_MOCK) / " + "KIWOOM_APP_SECRET_REAL(또는 KIWOOM_APP_SECRET_MOCK) 추가 필요" + ) + + logger.info(f"키움 60분봉 수집: {code} [{mode_label}] {start}~{end}") + rows = hb.fetch_60min_via_kiwoom( + code, start, end, kiwoom_key, kiwoom_secret, is_mock=is_mock + ) + _min_fetch_jobs[job_id]["fetched"] = len(rows) + + if not rows: + _min_fetch_jobs[job_id]["status"] = "done" + _min_fetch_jobs[job_id]["saved"] = 0 + return + + saved = 0 + logger.info(f"키움 60분봉 DB 저장 시작: {code} {len(rows)}봉") + err_sample = None + for row_data in rows: + try: + # MariaDB 문법: ON DUPLICATE KEY UPDATE (SQLite의 ON CONFLICT 아님) + db.conn.execute( + """ + INSERT INTO holding_min_candles + (code, candle_dt, tf, open, high, low, close, volume) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s) + ON DUPLICATE KEY UPDATE + open=VALUES(open), high=VALUES(high), + low=VALUES(low), close=VALUES(close), volume=VALUES(volume) + """, + ( + code, + row_data["candle_date"], + 60, + row_data["open"], + row_data["high"], + row_data["low"], + row_data["close"], + row_data["volume"], + ), + ) + saved += 1 + except Exception as row_err: + if err_sample is None: + err_sample = str(row_err) # 첫 번째 오류만 샘플 보존 + db.conn.commit() + if err_sample: + logger.warning(f"⚠️ 키움 60분봉 일부 저장 실패 ({code}): {err_sample}") + _min_fetch_jobs[job_id]["saved"] = saved + _min_fetch_jobs[job_id]["current_date"] = rows[-1]["candle_date"] if rows else "" + _min_fetch_jobs[job_id]["status"] = "done" + logger.info(f"✅ 키움 60분봉 저장 완료: {code} {saved}/{len(rows)}봉") + except Exception as e: + logger.error(f"❌ 키움 60분봉 수집 오류 ({code}): {e}", exc_info=True) + _min_fetch_jobs[job_id]["status"] = "error" + _min_fetch_jobs[job_id]["error"] = str(e) + finally: + db.close() + + threading.Thread(target=_run, daemon=True).start() + return jsonify({"job_id": job_id, "status": "started"}) + + +@app.route("/api/holding/min_backtest", methods=["GET"]) +def api_holding_min_backtest(): + """60분봉 기반 백테스트 (run_backtest 재사용, candle_date=candle_dt 로 호환)""" + code = request.args.get("code", "") + start_date = request.args.get("start", "") + end_date = request.args.get("end", datetime.now().strftime("%Y-%m-%d")) + tf = int(request.args.get("tf", 60)) + + if not code: + return jsonify({"error": "code 필수"}), 400 + + db = _holding_db() + try: + candles = hb.get_stored_min_candles(db, code, start_date, end_date, tf_min=tf) + if not candles: + stats = hb.get_min_candle_stats(db, code, tf_min=tf) + if stats["count"] == 0: + hint = f"📥 먼저 [60분봉 수집] 버튼으로 데이터를 수집하세요." + else: + hint = (f"DB에 {stats['count']}봉 있음 ({stats['min']} ~ {stats['max']})\n" + f"👉 백테스트 시작일을 '{stats['min'][:10]}' 이후로 설정하세요.") + return jsonify({"error": f"60분봉 없음\n{hint}"}), 400 + + if len(candles) < 20: + stats = hb.get_min_candle_stats(db, code, tf_min=tf) + hint = (f"DB에 {stats['count']}봉 있음 ({stats['min']} ~ {stats['max']})\n" + f"👉 백테스트 날짜 범위를 넓혀 최소 20봉 이상 포함하세요.") + return jsonify({ + "error": f"봉 부족: {len(candles)}개 (최소 20개 필요)\n{hint}" + }), 400 + + # 파라미터 오버라이드 (일봉 백테스트와 동일 방식) + cfg = hb.get_stock_config(db, code) + overrides = ["rsi_period","rsi_buy1","rsi_buy2","rsi_buy3","rsi_sell", + "take_profit_pct","stop_loss_pct","buy1_ratio","buy2_ratio", + "buy3_ratio","slot_money"] + for k in overrides: + v = request.args.get(k) + if v is not None: + cfg[k] = float(v) + + # tf_min=60 전달 → 52주 윈도우를 날짜 기반으로 정확히 계산 + result = hb.run_backtest(candles, cfg, tf_min=tf) + result["code"] = code + result["candle_count"] = len(candles) + result["tf"] = tf + result["params"] = {k: cfg[k] for k in hb.DEFAULT_STOCK_CONFIG} + return jsonify(result) + finally: + db.close() + + +@app.route("/api/holding/min_stats", methods=["GET"]) +def api_holding_min_stats(): + """60분봉 보유 현황 (종목 카드에서 표시용)""" + code = request.args.get("code", "") + tf = int(request.args.get("tf", 60)) + if not code: + return jsonify({"error": "code 필수"}), 400 + db = _holding_db() + try: + return jsonify(hb.get_min_candle_stats(db, code, tf_min=tf)) + finally: + db.close() + + +# ───────────────────────────────────────────────────────────────────────────── +# 스캘핑 / 꼬리잡기 파라미터 탐색 (Web 전용 coarse 그리드) +# ───────────────────────────────────────────────────────────────────────────── + +# Web용 coarse 그리드 (256 조합 내외 → 약 5~10초) +_SCALP_WEB_GRID = { + "rsi_oversold": [15, 18, 20, 25], + "sl_pct": [0.8, 1.0, 1.5, 2.0], + "tp_pct": [1.5, 2.0, 2.5, 3.5], + "drop_rate": [1.0, 1.5, 2.0, 3.0], +} +_TAIL_WEB_GRID = { + "min_drop_rate": [2.0, 3.0, 4.0, 5.0], + "min_recovery_ratio": [30.0, 40.0, 50.0, 60.0], + "sl_pct": [2.0, 3.0, 5.0], + "tp_pct": [3.0, 5.0, 7.0, 10.0], +} + + +@app.route("/api/backtest/scalping/param_search", methods=["GET"]) +def api_scalping_param_search(): + """스캘핑 파라미터 Grid Search (Web coarse, ~256 조합)""" + start = request.args.get("start", "") + end = request.args.get("end", datetime.now().strftime("%Y-%m-%d")) + top_n = int(request.args.get("top", 10)) + min_tr = int(request.args.get("min_trades", 3)) + + # 그리드 외 파라미터: JS가 현재 UI 값을 전달하면 사용, 없으면 Python 기본값 + _SCALP_NON_GRID_KEYS = [ + "rsi_period", "slot_money", "cooldown_min", "vol_mult", + "trail_trigger", "trail_stop", "time_start", "time_end", "max_daily", + ] + base_params = {"start": start, "end": end} + for k in _SCALP_NON_GRID_KEYS: + v = request.args.get(k) + if v is not None: + base_params[k] = v + + from itertools import product as iproduct + keys = list(_SCALP_WEB_GRID.keys()) + combos = list(iproduct(*[_SCALP_WEB_GRID[k] for k in keys])) + + results = [] + with app.test_client() as c: + for vals in combos: + p = dict(base_params) # base(UI 값) 먼저 + p.update(dict(zip(keys, vals))) # 그리드 값으로 덮어쓰기 + r = c.get("/api/backtest/scalping?" + "&".join(f"{k}={v}" for k, v in p.items())) + if r.status_code != 200: + continue + s = json.loads(r.data).get("summary", {}) + if s.get("total_trades", 0) < min_tr: + continue + results.append({ + "params": p, + "total_pnl": s.get("total_pnl", 0), + "win_rate": s.get("win_rate", 0), + "total_trades": s.get("total_trades", 0), + "pf": s.get("profit_factor", 0), + "mdd": s.get("max_drawdown", 0), + }) + + results.sort(key=lambda x: x["total_pnl"], reverse=True) + return jsonify({"results": results[:top_n], "total_combos": len(combos)}) + + +@app.route("/api/backtest/scalping/save_config", methods=["POST"]) +def api_scalping_save_config(): + """스캘핑 파라미터를 env_config에 저장 (봇이 즉시 반영)""" + body = request.get_json(force=True, silent=True) or {} + db = _db() + try: + latest = db.get_latest_env() + snap = dict(latest["snapshot"]) if latest else {} + # UI % 단위 → 봇이 사용하는 소수 단위로 변환 + if "rsi_oversold" in body: + snap["SCALP_RSI_OVERSOLD"] = str(float(body["rsi_oversold"])) + if "sl_pct" in body: + snap["SCALP_STOP_LOSS_PCT"] = str(float(body["sl_pct"]) / 100) + if "tp_pct" in body: + snap["SCALP_TAKE_PROFIT_PCT"]= str(float(body["tp_pct"]) / 100) + if "drop_rate" in body: + snap["SCALP_MIN_DROP_RATE"] = str(float(body["drop_rate"]) / 100) + if "trail_trigger" in body: + snap["SCALP_ATR_UP_MULT"] = str(float(body["trail_trigger"])) + if "cooldown_min" in body: + snap["SCALP_COOLDOWN_SEC"] = str(int(float(body["cooldown_min"])) * 60) + env_id = db.insert_env_snapshot(snap) + return jsonify({"ok": True, "env_id": env_id, "saved_keys": [ + "SCALP_RSI_OVERSOLD","SCALP_STOP_LOSS_PCT","SCALP_TAKE_PROFIT_PCT","SCALP_MIN_DROP_RATE" + ]}) + except Exception as e: + logger.error(f"스캘핑 설정저장 오류: {e}") + return jsonify({"error": str(e)}), 500 + finally: + db.close() + + +@app.route("/api/backtest/tail/param_search", methods=["GET"]) +def api_tail_param_search(): + """꼬리잡기 파라미터 Grid Search (Web coarse, ~192 조합)""" + start = request.args.get("start", "") + end = request.args.get("end", datetime.now().strftime("%Y-%m-%d")) + top_n = int(request.args.get("top", 10)) + min_tr = int(request.args.get("min_trades", 3)) + + # 그리드 외 파라미터: JS가 현재 UI 값을 전달하면 사용, 없으면 Python 기본값 + _TAIL_NON_GRID_KEYS = [ + "slot_money", "tail_ratio_min", "shoulder_min_high", "shoulder_cut_pct", + "rsi_threshold", "cooldown_min", "time_start", "time_end", "max_daily", + ] + base_params = {"start": start, "end": end} + for k in _TAIL_NON_GRID_KEYS: + v = request.args.get(k) + if v is not None: + base_params[k] = v + + from itertools import product as iproduct + keys = list(_TAIL_WEB_GRID.keys()) + combos = list(iproduct(*[_TAIL_WEB_GRID[k] for k in keys])) + + results = [] + with app.test_client() as c: + for vals in combos: + p = dict(base_params) # base(UI 값) 먼저 + p.update(dict(zip(keys, vals))) # 그리드 값으로 덮어쓰기 + r = c.get("/api/backtest/tail?" + "&".join(f"{k}={v}" for k, v in p.items())) + if r.status_code != 200: + continue + s = json.loads(r.data).get("summary", {}) + if s.get("total_trades", 0) < min_tr: + continue + results.append({ + "params": p, + "total_pnl": s.get("total_pnl", 0), + "win_rate": s.get("win_rate", 0), + "total_trades": s.get("total_trades", 0), + "pf": s.get("profit_factor", 0), + "mdd": s.get("max_drawdown", 0), + }) + + results.sort(key=lambda x: x["total_pnl"], reverse=True) + return jsonify({"results": results[:top_n], "total_combos": len(combos)}) + + +@app.route("/api/backtest/tail/save_config", methods=["POST"]) +def api_tail_save_config(): + """꼬리잡기 파라미터를 env_config에 저장 (봇이 즉시 반영)""" + body = request.get_json(force=True, silent=True) or {} + db = _db() + try: + latest = db.get_latest_env() + snap = dict(latest["snapshot"]) if latest else {} + if "min_drop_rate" in body: + snap["MIN_DROP_RATE"] = str(float(body["min_drop_rate"]) / 100) + if "min_recovery_ratio" in body: + snap["MIN_RECOVERY_RATIO_SHORT"] = str(float(body["min_recovery_ratio"]) / 100) + if "tail_ratio_min" in body: + snap["TAIL_RATIO_MIN"] = str(float(body["tail_ratio_min"])) + if "sl_pct" in body: + # 꼬리잡기 STOP_LOSS_PCT는 봇에서 음수 저장 + snap["STOP_LOSS_PCT"] = str(-abs(float(body["sl_pct"])) / 100) + if "tp_pct" in body: + snap["TAKE_PROFIT_PCT"] = str(float(body["tp_pct"]) / 100) + if "shoulder_cut_pct" in body: + snap["SHOULDER_CUT_PCT"] = str(float(body["shoulder_cut_pct"]) / 100) + if "shoulder_min_high" in body: + snap["SHOULDER_MIN_HIGH_PCT"] = str(float(body["shoulder_min_high"]) / 100) + if "cooldown_min" in body: + snap["REENTRY_COOLDOWN_SEC"] = str(int(float(body["cooldown_min"])) * 60) + env_id = db.insert_env_snapshot(snap) + return jsonify({"ok": True, "env_id": env_id, "saved_keys": [ + "MIN_DROP_RATE","MIN_RECOVERY_RATIO_SHORT","STOP_LOSS_PCT", + "TAKE_PROFIT_PCT","TAIL_RATIO_MIN","SHOULDER_CUT_PCT" + ]}) + except Exception as e: + logger.error(f"꼬리잡기 설정저장 오류: {e}") + return jsonify({"error": str(e)}), 500 + finally: + db.close() + + +@app.route("/api/env/params", methods=["GET"]) +def api_env_params(): + """ + env_config 최신 스냅샷에서 스캘핑·꼬리잡기 UI 초기값 반환. + % 단위 변환 및 음수 → 양수 변환까지 수행해 JS가 바로 input.value에 넣을 수 있도록 함. + 값이 DB에 없으면 null 반환 → JS에서 기존 HTML 기본값 유지. + """ + db = _db() + try: + latest = db.get_latest_env() + snap = (latest or {}).get("snapshot", {}) + + def fv(key): + """DB 값을 float으로 파싱, 없으면 None""" + v = snap.get(key) + if v is None or v == "": + return None + try: + return float(v) + except (ValueError, TypeError): + return None + + def smart_pct(key): + """ + DB 저장 형식이 소수(0.03) 또는 퍼센트(3.0) 중 어느 쪽이든 + UI에 항상 퍼센트 단위(3.0)로 반환. + - 절댓값 < 0.5 → 소수 형식 → ×100 + - 절댓값 >= 0.5 → 이미 퍼센트 형식 → 그대로 + - STOP_LOSS_PCT처럼 음수 저장된 경우 → 양수 변환 + """ + v = fv(key) + if v is None: + return None + av = abs(v) + if av == 0: + return 0.0 + result = av if av >= 0.5 else round(av * 100, 3) + return round(result, 3) + + def sec_to_min(key): + """초 → 분, None 유지""" + v = fv(key) + return round(v / 60) if v is not None else None + + return jsonify({ + "scalp": { + "rsi_oversold": fv("SCALP_RSI_OVERSOLD"), # 과매도 임계값 (숫자 그대로) + "sl_pct": smart_pct("SCALP_STOP_LOSS_PCT"), # 소수/퍼센트 자동감지 + "tp_pct": smart_pct("SCALP_TAKE_PROFIT_PCT"), + "drop_rate": smart_pct("SCALP_MIN_DROP_RATE"), + "trail_trigger": fv("SCALP_ATR_UP_MULT"), # 배수 그대로 + "cooldown_min": sec_to_min("SCALP_COOLDOWN_SEC"), # 스캘핑 전용 (꼬리잡기 REENTRY_COOLDOWN_SEC 와 분리) + "slot_money": fv("SLOT_MONEY_DEFAULT"), + }, + "tail": { + "drop": smart_pct("MIN_DROP_RATE"), + "rec": smart_pct("MIN_RECOVERY_RATIO_SHORT"), + "tail_ratio": fv("TAIL_RATIO_MIN"), + "sl_pct": smart_pct("STOP_LOSS_PCT"), # 음수 → 양수도 자동처리 + "tp_pct": smart_pct("TAKE_PROFIT_PCT"), + "smin": smart_pct("SHOULDER_MIN_HIGH_PCT"), + "scut": smart_pct("SHOULDER_CUT_PCT"), + "cool": sec_to_min("REENTRY_COOLDOWN_SEC"), + "rsi": fv("RSI_OVERHEAT_THRESHOLD"), + }, + }) + finally: + db.close() + + +HTML_TEMPLATE = r""" + + + + +📈 KIS Quant 백테스트 대시보드 + + + + + + + +
+ +
+ + + +
+ + + + + +
+
+
+
+
+
+ + +
+
+
+ + +
+
+ + +
+
+ +
+
+
+ + +
+
+
총 거래
-
+
+
+
승률
-
+
+
+
순손익
-
+
+
+
Profit Factor
-
+
+
+
최대 낙폭(MDD)
-
+
+
+
평균 보유(분)
-
+
+
+ + + +
+ +
+
+
누적 손익 곡선
+ +
+
+ +
+
+
매도 이유 분포
+ +
+
+
종목별 손익 TOP10
+ +
+
+
+ + +
+
일별 손익
+ +
+ + +
거래 내역
+
+
+ + + + + + +
매도일시종목매수가매도가수량손익(원)수익률보유(분)매도사유
+
+
+
+ + + + + + + + + + +
+ + + +""" + + +@app.route("/") +def index(): + return render_template_string(HTML_TEMPLATE) + + +if __name__ == "__main__": + app.run(host="0.0.0.0", port=5050, debug=False)