#!/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") import scalping_engine as se @app.route("/api/backtest/scalping", methods=["GET"]) def api_backtest_scalping(): # 기본값 = DB(엔진 단일 소스) → 백테스트/param_search/실매매 동일 값 _def = se.get_scalping_defaults_from_db() start = request.args.get("start", "") end = request.args.get("end", "") rsi_period = int(request.args.get("rsi_period", _def["rsi_period"])) 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", _def["slot_money"])) # 1회 투자금 _fee_rate = request.args.get("fee_rate") fee_rate = float(_fee_rate) / 100 if _fee_rate not in (None, "") else _def["fee_rate"] _sell_tax = request.args.get("sell_tax") sell_tax = float(_sell_tax) / 100 if _sell_tax not in (None, "") else _def["sell_tax"] _cooldown = request.args.get("cooldown_min") cooldown_min = float(_cooldown) if _cooldown not in (None, "") else _def["cooldown_min"] vol_mult = float(request.args.get("vol_mult", _def["vol_mult"])) # 거래량 필터(0=비활성) _tr_trigger = request.args.get("trail_trigger") trail_trigger = float(_tr_trigger) / 100 if _tr_trigger not in (None, "") else _def["trail_trigger"] _tr_stop = request.args.get("trail_stop") trail_stop = float(_tr_stop) / 100 if _tr_stop not in (None, "") else _def["trail_stop"] _time_start = request.args.get("time_start") time_start_hm = int(_time_start) if _time_start not in (None, "") else _def["time_start_hm"] _time_end = request.args.get("time_end") time_end_hm = int(_time_end) if _time_end not in (None, "") else _def["time_end_hm"] max_daily = int(request.args.get("max_daily", _def["max_daily"])) # 종목당 일일 최대 거래횟수 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] codes_candles: Dict[str, 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 codes_candles[code] = [dict(r) for r in rows] params = { "rsi_period": rsi_period, "rsi_oversold": rsi_oversold, "sl_pct": sl_pct, "tp_pct": tp_pct, "drop_rate": drop_rate, "slot_money": slot_money, "fee_rate": fee_rate, "sell_tax": sell_tax, "cooldown_min": cooldown_min, "trail_trigger": trail_trigger, "trail_stop": trail_stop, "time_start_hm": time_start_hm, "time_end_hm": time_end_hm, "max_daily": max_daily, "vol_mult": vol_mult, } all_virtual_trades = se.run_scalping_backtest(codes_candles, params) # ───────────────────────────────────────────────────────────────── # 결과 집계 # ───────────────────────────────────────────────────────────────── 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분봉 기반, tail_engine 공통 로직 사용) # ──────────────────────────────────────────────────────────────────────────── try: import tail_engine as te _TAIL_ENGINE_AVAILABLE = True except ImportError: _TAIL_ENGINE_AVAILABLE = False def _get_tail_defaults_for_backtest(): """꼬리잡기 백테스트 기본값: DB 단일 소스. 엔진 없으면 빈 dict.""" if not _TAIL_ENGINE_AVAILABLE: return {} return te.get_tail_defaults_from_db() @app.route("/api/backtest/tail", methods=["GET"]) def api_backtest_tail(): """ 꼬리잡기 전략 가격 재현 백테스트. entry 조건: 당일 낙폭(drop_rate) + 회복률(recovery_ratio) + 망치봉 꼬리 + RSI exit 조건: 손절 / 익절 / 어깨 컷(trailing) / 장 마감 강제 청산 기본값 = DB(env_config) → tail_engine.get_tail_defaults_from_db(), 요청으로 덮어쓰기. """ _def = _get_tail_defaults_for_backtest() start = request.args.get("start", "") end = request.args.get("end", "") rsi_period = int( request.args.get("rsi_period", _def.get("rsi_period", 14))) rsi_threshold = float(request.args.get("rsi_threshold", _def.get("rsi_threshold", 78))) min_drop_rate = float(request.args.get("min_drop_rate", _def.get("min_drop_rate", 0.03) * 100)) / 100 min_recovery_ratio = float(request.args.get("min_recovery_ratio", _def.get("min_recovery_ratio", 0.5) * 100)) / 100 # max_rec_3m / high_chase_thr: 폼에서 80·96(퍼센트) 또는 0.8·0.96(비율) 전달 가능 → 엔진은 항상 비율(0~1) _max_rec_raw = float(request.args.get("max_rec_3m", _def.get("max_rec_3m", 0.8))) max_rec_3m = _max_rec_raw if 0 < _max_rec_raw <= 1 else _max_rec_raw / 100 tail_ratio_min = float(request.args.get("tail_ratio_min", _def.get("tail_ratio_min", 1.5))) tail_pct_min = float(request.args.get("tail_pct_min", _def.get("tail_pct_min", 0.003) * 100)) / 100 sl_pct = float(request.args.get("sl_pct", _def.get("sl_pct", 0.03) * 100)) / 100 tp_pct = float(request.args.get("tp_pct", _def.get("tp_pct", 0.05) * 100)) / 100 shoulder_min_high = float(request.args.get("shoulder_min_high", _def.get("shoulder_min_high", 0.015) * 100)) / 100 shoulder_cut_pct = float(request.args.get("shoulder_cut_pct", _def.get("shoulder_cut_pct", 0.03) * 100)) / 100 _high_chase_raw = float(request.args.get("high_chase_thr", _def.get("high_chase_thr", 0.96))) high_chase_thr = _high_chase_raw if 0 < _high_chase_raw <= 1 else _high_chase_raw / 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", _def.get("cooldown_min", 15))) time_start_hm = int( request.args.get("time_start", _def.get("time_start_hm", 930))) time_end_hm = int( request.args.get("time_end", _def.get("time_end_hm", 1500))) max_daily = int( request.args.get("max_daily", _def.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] use_engine = _TAIL_ENGINE_AVAILABLE and request.args.get("use_engine", "1") == "1" all_trades: List[Dict] = [] if use_engine: # tail_engine 공통 로직 사용 (실매매 ver3과 동일 계산식) candles_by_code = {} 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_by_code[code] = [dict(r) for r in rows] params = { "min_drop_rate": min_drop_rate, "min_recovery_ratio": min_recovery_ratio, "max_rec_3m": max_rec_3m, "tail_ratio_min": tail_ratio_min, "tail_pct_min": tail_pct_min, "sl_pct": sl_pct, "tp_pct": tp_pct, "shoulder_min_high": shoulder_min_high, "shoulder_cut_pct": shoulder_cut_pct, "rsi_period": rsi_period, "rsi_threshold": rsi_threshold, "high_chase_thr": high_chase_thr, "time_start_hm": time_start_hm, "time_end_hm": time_end_hm, "cooldown_min": cooldown_min, "max_daily": max_daily, } all_trades = te.run_tail_backtest(candles_by_code, params) for t in all_trades: qty = max(1, int(slot_money / t["entry"])) fee = (t["entry"] + t["exit"]) * qty * fee_rate tax = t["exit"] * qty * sell_tax t["pnl"] = round((t["exit"] - t["entry"]) * qty - fee - tax) t["hold_min"] = round((_t2dt(t["exit_time"]) - _t2dt(t["entry_time"])).total_seconds() / 60, 1) else: 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, "time_start_hm": time_start_hm, "time_end_hm": time_end_hm, "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_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) if "rsi_threshold" in body: snap["RSI_OVERHEAT_THRESHOLD"] = str(float(body["rsi_threshold"])) if "rsi_period" in body: snap["RSI_PERIOD"] = str(int(float(body["rsi_period"]))) if "time_start" in body: snap["TIME_START"] = str(int(float(body["time_start"]))) if "time_end" in body: snap["TIME_END"] = str(int(float(body["time_end"]))) if "max_daily" in body: snap["MAX_STOCKS"] = str(int(float(body["max_daily"]))) if "tail_pct_min" in body: snap["TAIL_PCT_MIN"] = str(float(body["tail_pct_min"]) / 100) if "max_rec_3m" in body: snap["MAX_RECOVERY_RATIO_3M"] = str(float(body["max_rec_3m"]) / 100) if "high_chase_thr" in body: snap["HIGH_PRICE_CHASE_THRESHOLD"] = str(float(body["high_chase_thr"]) / 100) 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","TAIL_PCT_MIN","SHOULDER_CUT_PCT", "REENTRY_COOLDOWN_SEC","RSI_OVERHEAT_THRESHOLD","RSI_PERIOD", "TIME_START","TIME_END","MAX_STOCKS","MAX_RECOVERY_RATIO_3M","HIGH_PRICE_CHASE_THRESHOLD" ]}) 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 def _ratio_to_pct(val, default): """비율(0~1)을 폼 퍼센트 표시용(80, 96 등)으로. DB 0.8 → 80 반환.""" if val is None: return default try: v = float(val) if 0 < v <= 1: return round(v * 100, 2) if v > 1: return round(v, 2) except (ValueError, TypeError): pass return default 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"), "tail_pct_min": smart_pct("TAIL_PCT_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"), "rsi_period": int(fv("RSI_PERIOD") or 14), # RSI 기간 (실매·파라서치와 동일) "time_start": int(fv("TIME_START") or 930), # 매수시작 HHMM "time_end": int(fv("TIME_END") or 1500), # 매수종료 HHMM "max_daily": int(fv("MAX_STOCKS") or 3), # 일일최대매수 (실매 MAX_STOCKS) # 비율(0~1) 저장값을 폼 퍼센트 표시용으로 80·96 형태로 반환 (백테 API는 80→0.8로 변환) "max_rec_3m": _ratio_to_pct(fv("MAX_RECOVERY_RATIO_3M"), 80), "high_chase": _ratio_to_pct(fv("HIGH_PRICE_CHASE_THRESHOLD"), 96), }, }) 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)