"""Ground Truth 차트 HTML 생성 (전체 기간 지원).""" from __future__ import annotations import bisect import json from pathlib import Path from typing import Any import pandas as pd from deepcoin.data.candle_loader import load_candles # 0이면 제한 없이 전체 봉 표시 DEFAULT_MAX_CANDLES = 0 def _data_js_path(html_path: Path) -> Path: """HTML과 짝을 이루는 데이터 JS 경로 (file:// 프로토콜 호환). 예: ground_truth_chart_v3.html → ground_truth_chart_v3_data.js """ return html_path.with_name(f"{html_path.stem}_data.js") def _to_unix_seconds(dt_series: pd.Series) -> list[int]: """datetime Series를 uPlot/LWC용 unix 초 리스트로 변환한다. pandas datetime64[ns/us/ms] 단위 차이에 관계없이 올바른 초 단위를 반환한다. Args: dt_series: datetime 컬럼. Returns: unix epoch 초 리스트. """ parsed = pd.to_datetime(dt_series) seconds = (parsed - pd.Timestamp("1970-01-01")) / pd.Timedelta(seconds=1) return seconds.astype(int).tolist() def _close_at_timestamp(times: list[int], closes: list[float], ts: int) -> float: """차트 종가 배열에서 시각에 해당하는 종가를 반환한다. Args: times: unix 초 리스트. closes: 종가 리스트. ts: 조회 시각(unix 초). Returns: 해당 봉 종가. 정확히 일치하는 봉이 없으면 가장 가까운 봉 종가. """ if not times: return 0.0 idx = bisect.bisect_left(times, ts) if idx >= len(times): return closes[-1] if idx > 0 and times[idx] != ts: if abs(times[idx - 1] - ts) <= abs(times[idx] - ts): idx -= 1 return closes[idx] def _enrich_markers_chart_price( markers: list[dict[str, Any]], times: list[int], closes: list[float], ) -> list[dict[str, Any]]: """마커에 종가 선(chart) 위치용 chart_price를 추가한다.""" return [ {**marker, "chart_price": _close_at_timestamp(times, closes, marker["time"])} for marker in markers ] def _markers_from_executed_trades( sim_pnl: dict[str, Any], ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: """시뮬에서 실제 체결된 매수·매도만 마커로 변환한다.""" buy_markers: list[dict[str, Any]] = [] sell_markers: list[dict[str, Any]] = [] for trade in sim_pnl.get("trades") or []: if trade.get("skipped"): continue side = trade["side"] signal_type = trade.get("signal_type") or ( "swing_low" if side == "buy" else "swing_high" ) marker = { "time": int(pd.Timestamp(trade["datetime"]).timestamp()), "price": trade["price"], "marker_id": trade.get("marker_id") or trade.get("trade_id"), "signal_type": signal_type, } if side == "buy": buy_markers.append(marker) else: sell_markers.append(marker) return buy_markers, sell_markers def _markers_from_gt_signals( gt_result: dict[str, Any], sim_period_from_ts: int | None = None, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: """GT 신호에서 마커를 구성한다 (1단계 차트용).""" buy_markers: list[dict[str, Any]] = [] sell_markers: list[dict[str, Any]] = [] for sig in gt_result.get("signals") or []: ts = int(pd.Timestamp(sig["datetime"]).timestamp()) if sim_period_from_ts is not None and ts < sim_period_from_ts: continue marker = { "time": ts, "price": sig["price"], "marker_id": sig.get("marker_id", sig.get("leg_id")), "signal_type": sig.get( "signal_type", "swing_low" if sig["side"] == "buy" else "swing_high", ), } if sig["side"] == "buy": buy_markers.append(marker) else: sell_markers.append(marker) return buy_markers, sell_markers def _sim_start_marker( df: pd.DataFrame, sim_pnl: dict[str, Any], ) -> dict[str, Any] | None: """1년 시뮬 매매 시작 시점 마커를 구성한다.""" period_from = sim_pnl.get("period_from") if not period_from: return None start_ts = pd.Timestamp(period_from) parsed = pd.to_datetime(df["datetime"]) idx = int(parsed.searchsorted(start_ts, side="left")) if idx >= len(df): idx = len(df) - 1 row = df.iloc[idx] dt_str = str(row["datetime"]) return { "time": int(pd.Timestamp(dt_str).timestamp()), "price": float(row["close"]), "datetime": dt_str, "label": "거래시작", } def _build_chart_payload( df: pd.DataFrame, gt_result: dict[str, Any], chart_days: int, gt_lookback_days: int, sim_pnl: dict[str, Any] | None = None, ) -> dict[str, Any]: """차트 HTML용 JSON payload를 구성한다.""" times = _to_unix_seconds(df["datetime"]) closes = df["close"].astype(float).tolist() if sim_pnl is not None: buy_markers, sell_markers = _markers_from_executed_trades(sim_pnl) else: buy_markers, sell_markers = _markers_from_gt_signals(gt_result) buy_markers = _enrich_markers_chart_price(buy_markers, times, closes) sell_markers = _enrich_markers_chart_price(sell_markers, times, closes) chart_meta = { **gt_result["meta"], "chart_lookback_days": chart_days, "gt_lookback_days": gt_lookback_days, "chart_data_from": str(df["datetime"].min()), "chart_data_to": str(df["datetime"].max()), "gt_marker_count": len(buy_markers), } if sim_pnl is not None: chart_meta["sim_period_from"] = sim_pnl.get("period_from") chart_meta["sim_period_to"] = sim_pnl.get("period_to") chart_meta["sim_lookback_days"] = sim_pnl.get("sim_lookback_days") payload: dict[str, Any] = { "times": times, "open": df["open"].astype(float).tolist(), "high": df["high"].astype(float).tolist(), "low": df["low"].astype(float).tolist(), "close": closes, "buy_markers": buy_markers, "sell_markers": sell_markers, "meta": chart_meta, "bar_count": len(df), } if sim_pnl is not None: payload["sim_pnl"] = sim_pnl start_marker = _sim_start_marker(df, sim_pnl) if start_marker is not None: payload["sim_start_marker"] = start_marker return payload def render_ground_truth_chart( db_path: Path, symbol: str, gt_result: dict[str, Any], output_path: Path, chart_lookback_days: int | None = None, max_candles: int = DEFAULT_MAX_CANDLES, ) -> Path: """GT 타점이 표시된 HTML 차트를 생성한다. 대용량(3분봉 2년 등)은 종가 라인으로 전체 기간을 표시하고, JSON 데이터는 별도 파일로 분리한다. Args: db_path: SQLite 경로. symbol: 코인 심볼. gt_result: build_ground_truth 결과. output_path: HTML 출력 경로. chart_lookback_days: 차트에 표시할 일수. None이면 GT lookback과 동일. max_candles: 0이면 전체, 양수면 최근 N봉만. Returns: HTML 저장 경로. """ interval_min = gt_result["meta"]["interval_min"] gt_lookback_days = gt_result["meta"]["lookback_days"] chart_days = chart_lookback_days if chart_lookback_days is not None else gt_lookback_days df = load_candles(db_path, symbol, interval_min, lookback_days=chart_days) if max_candles > 0 and len(df) > max_candles: df = df.iloc[-max_candles:].reset_index(drop=True) payload = _build_chart_payload(df, gt_result, chart_days, gt_lookback_days) output_path.parent.mkdir(parents=True, exist_ok=True) data_path = _data_js_path(output_path) with data_path.open("w", encoding="utf-8") as fp: fp.write("window.CHART_DATA=") json.dump(payload, fp, ensure_ascii=False, separators=(",", ":")) fp.write(";") data_js_name = data_path.name output_path.write_text(_html_template(data_js_name), encoding="utf-8") return output_path _HTML_TEMPLATE = """
| 시각 | 구분 | 유형 | 가격 | 주문금액 | 수수료 | 현금 | 코인 | 비고 |
|---|