""" 1단계 시뮬레이션 HTML — ground_truth 차트와 동일 스타일·타점·수익률·규칙 기준. """ from __future__ import annotations import json from pathlib import Path from typing import Any import pandas as pd from config import ( CHART_LOOKBACK_DAYS, COIN_NAME, GT_INITIAL_CASH_KRW, LIVE_ORDER_KRW, SYMBOL, TRADING_FEE_RATE, ) from deepcoin.ground_truth.ground_truth import ( load_ground_truth, simulate_truth_portfolio, simulate_truth_portfolio_steps, ) from deepcoin.matching.portfolio_sim import ( fires_to_trade_list, select_capped_fires, simulate_fixed_order_portfolio, simulate_fixed_order_portfolio_steps, ) from deepcoin.matching.select_rules import _split_train_valid_holdout from deepcoin.ops.chart_report import ( card_html, go_no_go_table_html, market_cards_html, pnl_cards_html, rule_criteria_html, wrap_chart_report_page, ) from deepcoin.ops.simulation import build_chart_html, load_chart_frames, _frames_to_mtf from deepcoin.common.indicators import apply_bar_indicators, get_trend from deepcoin.paths import ( MATCHING_FIRE_OUTCOMES, MATCHING_MATCHED_RULES, MATCHING_SIMULATION_HTML, resolve_ground_truth_file, ) def _fires_to_chart_trades(fires: pd.DataFrame) -> list[dict[str, Any]]: """ fire_outcomes 행을 차트 마커용 dict 리스트로 변환. Args: fires: monitor holdout 발화. Returns: build_chart_html sim_trades 인자. """ rows: list[dict[str, Any]] = [] for _, r in fires.iterrows(): rows.append( { "dt": str(r["dt"]), "action": r["side"], "price": float(r["close"]), "forward_ret_pct": float(r["forward_ret_pct"]), "rule_id": r["rule_id"], } ) return rows def _sim_fire_table_rows( fires: pd.DataFrame, rules_by_id: dict[str, dict], steps: list[dict[str, Any]], ) -> str: """ 시뮬 발화 테이블 tbody (GT와 동일하게 총 평가금액 포함). Args: fires: holdout 체결 발화. rules_by_id: rule_id → rule dict. steps: simulate_fixed_order_portfolio_steps 결과. Returns: tr HTML. """ if fires.empty: return "
정답 (ground_truth) — 분할 비중·leg 체결
' + market_cards_html(close_last, bb_txt) + pnl_cards_html(gt_pnl, "정답 타점", len(gt_trades)) ) sim_row = ( '시뮬 (monitor_rules · holdout · ' f"1회 ₩{LIVE_ORDER_KRW:,.0f}·일한도) — " f'{"GO" if go_flag else "NO-GO"}
' + pnl_cards_html(sim_pnl, "시뮬 체결", sim_trade_count) ) return gt_row + sim_row def build_simulation_page_html( report: dict[str, Any], outcomes_path: Path | None = None, matched_path: Path | None = None, gt_path: Path | None = None, close_last: float | None = None, ) -> str: """ 시뮬 리포트 전체 HTML (차트 + Go/No-Go + 규칙 기준 + 타점 테이블). Args: report: build_simulation_report 결과. outcomes_path: fire_outcomes.csv. matched_path: matched_rules.json. gt_path: ground_truth JSON. close_last: 미청산 평가 종가 (None이면 DB 종가). Returns: HTML 문자열. """ op = outcomes_path or MATCHING_FIRE_OUTCOMES mp = matched_path or MATCHING_MATCHED_RULES matched: dict[str, Any] = {} if mp.is_file(): matched = json.loads(mp.read_text(encoding="utf-8")) monitor_rules = matched.get("monitor_rules") or report.get("monitor_rules") or [] monitor_ids = {r["rule_id"] for r in monitor_rules} rules_by_id = {r["rule_id"]: r for r in monitor_rules} holdout_fires = pd.DataFrame() if op.is_file(): outcomes = pd.read_csv(op) outcomes["split"] = _split_train_valid_holdout(outcomes) holdout_fires = outcomes[ (outcomes["rule_id"].isin(monitor_ids)) & (outcomes["split"] == "holdout") ].copy() capped = select_capped_fires(holdout_fires) gt_data = load_ground_truth(gt_path or resolve_ground_truth_file()) or {} gt_trades = gt_data.get("trades") or [] gt_summary = gt_data.get("summary") or {} go = report.get("go_no_go", {}) go_flag = bool(go.get("go")) label_mode = report.get("label_mode", "leg_gt") frames = load_chart_frames() bb_txt = "-" trend = "-" close_val = float(close_last or 0) if frames is not None: df_1d, df_1h, df_3m = _frames_to_mtf(frames) trend = get_trend(df_1d, df_1h) df_chart = apply_bar_indicators(df_3m) close_val = float(df_chart["Close"].iloc[-1]) bb_pos = ( float(df_chart["bb_pos"].iloc[-1]) if "bb_pos" in df_chart.columns and pd.notna(df_chart["bb_pos"].iloc[-1]) else None ) bb_txt = f"{bb_pos:.2f}" if bb_pos is not None else "-" elif gt_summary.get("mark_price"): close_val = float(gt_summary["mark_price"]) sim_trades = fires_to_trade_list(capped) gt_pnl = {} gt_summary_pnl = gt_data.get("summary") or {} if gt_summary_pnl.get("pnl_krw") is not None and gt_summary_pnl.get( "execution_order" ) == "leg_block": gt_pnl = { k: gt_summary_pnl[k] for k in ( "initial_cash_krw", "final_asset_krw", "pnl_pct", "total_fees_krw", "holding_qty", "holding_value_krw", "mark_price", "cash_krw", ) if k in gt_summary_pnl } elif gt_trades: gt_pnl = simulate_truth_portfolio( gt_trades, initial_cash=GT_INITIAL_CASH_KRW, fee_rate=TRADING_FEE_RATE, last_price=close_val if close_val else None, ) sim_pnl = simulate_fixed_order_portfolio( sim_trades, order_krw=LIVE_ORDER_KRW, initial_cash=GT_INITIAL_CASH_KRW, fee_rate=TRADING_FEE_RATE, last_price=close_val if close_val else None, ) sim_steps = simulate_fixed_order_portfolio_steps( sim_trades, order_krw=LIVE_ORDER_KRW, initial_cash=GT_INITIAL_CASH_KRW, fee_rate=TRADING_FEE_RATE, ) gt_steps = ( simulate_truth_portfolio_steps( gt_trades, initial_cash=GT_INITIAL_CASH_KRW, fee_rate=TRADING_FEE_RATE, ) if gt_trades else [] ) criteria_blocks = "".join(rule_criteria_html(r) for r in monitor_rules) go_table = go_no_go_table_html(go.get("checks", []), go_flag) def _mark_note(price: float) -> str: if price > 0: return f" 총보유자산(미청산 포함)은 종가 ₩{price:,.0f} 평가." return "" sim_table = f"""| 시각 | 구분 | 규칙 | 유형 | 가격 | 총 평가금액 | 승/패 | 비고 |
|---|
| 시각 | 구분 | 비중 | 가격 | 총 평가금액 | 해석 |
|---|
{note}
", legend_html=legend, cards_html=cards, chart_html=( "차트 데이터 없음 — "
"python scripts/01_download.py 후 재생성.