"""
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,
order_trades_chronological,
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,
simulate_sized_portfolio,
)
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,
stacked_summary_cards_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 "
| 발화 없음 |
"
sorted_fires = fires.sort_values("dt").reset_index(drop=True)
lines: list[str] = []
lines.append(
f"""
| 시작 | - | - | - | - |
₩{GT_INITIAL_CASH_KRW:,.0f} |
- | 초기 현금 (1회 ₩{LIVE_ORDER_KRW:,.0f} 가정) |
"""
)
for i in range(len(sorted_fires)):
r = sorted_fires.iloc[i]
side = r["side"]
cls = "buy" if side == "buy" else "sell"
mark = "매수" if side == "buy" else "매도"
ret = float(r["forward_ret_pct"])
ret_s = f" (+{ret:.2f}%)" if ret > 0 else f" ({ret:.2f}%)"
win = "승" if int(r.get("win", 0)) else "패"
win_cls = "pass" if int(r.get("win", 0)) else "fail"
kind = rules_by_id.get(r["rule_id"], {}).get("kind", "")
step = steps[i] if i < len(steps) else None
if step:
total_s = f"₩{step['total_asset_krw']:,.0f}"
hold_s = (
f" (현금 ₩{step['cash_krw']:,.0f} + 코인 {step['holding_qty']:,.2f}개)"
)
else:
total_s = "-"
hold_s = ""
lines.append(
f""
f"| {str(r['dt'])[:16]} | "
f'{mark} | '
f"{r['rule_id']} | "
f"{kind} | "
f'₩{float(r["close"]):,.0f}{ret_s} | '
f"{total_s}{hold_s} | "
f'{win} | '
f"leg_gt 구간 수익 | "
f"
"
)
return "\n".join(lines)
def _gt_table_rows(trades: list[dict[str, Any]], steps: list[dict[str, Any]]) -> str:
"""
GT 타점 테이블 tbody (ground_truth 차트와 동일).
Args:
trades: ground_truth trades.
steps: simulate_truth_portfolio_steps.
Returns:
tr HTML.
"""
if not trades:
return "| 타점 없음 |
"
step_key = {
(s["dt"], s["action"], float(s["price"]), float(s["weight"])): s
for s in steps
}
lines: list[str] = []
lines.append(
f"""
| 시작 | - | - | - |
₩{GT_INITIAL_CASH_KRW:,.0f} |
초기 현금 (보유 0) |
"""
)
for t in sorted(trades, key=lambda x: x["dt"]):
cls = "buy" if t["action"] == "buy" else "sell"
mark = "매수" if t["action"] == "buy" else "매도"
ret = t.get("forward_return_pct")
ret_s = f" (+{ret}%)" if ret is not None else ""
w = float(t.get("weight", 1.0))
key = (t["dt"], t["action"], float(t["price"]), w)
step = step_key.get(key)
if step:
total_s = f"₩{step['total_asset_krw']:,.0f}"
hold_s = (
f" (현금 ₩{step['cash_krw']:,.0f} + 코인 {step['holding_qty']:,.2f}개)"
)
else:
total_s = "-"
hold_s = ""
lines.append(
f""
f"| {t['dt'][:16]} | "
f'{mark} | '
f"{w*100:.0f}% | "
f"₩{t['price']:,.0f}{ret_s} | "
f"{total_s}{hold_s} | "
f"{t.get('memo', '')} | "
f"
"
)
return "\n".join(lines)
def _summary_cards_html(
close_last: float,
bb_txt: str,
gt_trades: list[dict[str, Any]],
gt_pnl: dict[str, Any],
sim_sized_pnl: dict[str, Any],
sim_fixed_pnl: dict[str, Any],
sim_trade_count: int,
go_flag: bool,
model_note: str = "",
) -> str:
"""
ground_truth HTML과 동일 구성의 상단 카드 (GT + 시뮬 2줄).
Args:
close_last: 종가.
bb_txt: BB %B.
gt_trades: GT trades.
gt_pnl: GT 포트폴리오 요약.
sim_sized_pnl: 총자산%·EV/WF·leg 시뮬 요약.
sim_fixed_pnl: 고정 ₩/회 baseline.
sim_trade_count: 체결 가정 발화 수.
go_flag: Go/No-Go.
model_note: GT 모델 한 줄 요약.
Returns:
cards HTML.
"""
go_cls = "go-pass" if go_flag else "go-fail"
gt_sub = (
"저점 분할매수(1/price 비중) · 고점 65/35% 매도 · "
"총자산×비중×leg티어 · 시각순 복리"
)
if model_note:
gt_sub = model_note
gt_cards = market_cards_html(close_last, bb_txt) + pnl_cards_html(
gt_pnl, "정답 GT", len(gt_trades)
)
sim_sized_title = (
"시뮬·총자산% (EV/WF·leg상위) — "
f'{"GO" if go_flag else "NO-GO"}'
)
sim_fixed_title = f"시뮬·고정 ₩{LIVE_ORDER_KRW:,}/회 (비교)"
return (
''
+ stacked_summary_cards_html(gt_sub, gt_cards)
+ stacked_summary_cards_html(
sim_sized_title,
pnl_cards_html(sim_sized_pnl, "시뮬(비율)", sim_trade_count),
)
+ stacked_summary_cards_html(
sim_fixed_title,
pnl_cards_html(sim_fixed_pnl, "시뮬(고정)", sim_trade_count),
)
+ "
"
)
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_sized = fires_to_trade_list(capped, apply_dynamic_sizing=True)
sim_trades_fixed = fires_to_trade_list(capped, apply_dynamic_sizing=False)
gt_pnl: dict[str, Any] = {}
if gt_trades:
gt_chron = order_trades_chronological(gt_trades)
gt_pnl = simulate_truth_portfolio(
gt_chron,
initial_cash=GT_INITIAL_CASH_KRW,
fee_rate=TRADING_FEE_RATE,
last_price=close_val if close_val else None,
)
mark = close_val if close_val else None
sim_sized_pnl = simulate_sized_portfolio(
sim_trades_sized,
initial_cash=GT_INITIAL_CASH_KRW,
fee_rate=TRADING_FEE_RATE,
last_price=mark,
)
sim_fixed_pnl = simulate_fixed_order_portfolio(
sim_trades_fixed,
order_krw=LIVE_ORDER_KRW,
initial_cash=GT_INITIAL_CASH_KRW,
fee_rate=TRADING_FEE_RATE,
last_price=mark,
sizing_mode="fixed",
)
sim_steps = simulate_fixed_order_portfolio_steps(
sim_trades_sized,
order_krw=LIVE_ORDER_KRW,
initial_cash=GT_INITIAL_CASH_KRW,
fee_rate=TRADING_FEE_RATE,
sizing_mode="amount_krw",
)
gt_steps = (
simulate_truth_portfolio_steps(
order_trades_chronological(gt_trades),
initial_cash=GT_INITIAL_CASH_KRW,
fee_rate=TRADING_FEE_RATE,
)
if gt_trades
else []
)
model = gt_data.get("model") or {}
model_note = (
f"mode={model.get('selection_mode', 'split_buy_peak_sell')} · "
f"매수비중=1/price · 매도=65/35%"
if model
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"""
시뮬 타점 (holdout {len(holdout_fires)}건 → 체결 가정 {len(capped)}건)
총자산×최적비중·현금한도·EV/WF통과·leg상위 대형 매수. 일한도·최대 거래수 적용.
가격 열 (+/-) = {label_mode} 구간 수익%.{_mark_note(close_val)}
"""
gt_table = f"""
정답 타점 (ground_truth)
삼각형 크기 = GT 체결 금액. 매수 분할·매도 leg 반영.{_mark_note(close_val)}
"""
sections = f"""
{go_table}
매수·매도 판단 기준 (monitor_rules)
04 GT 프로필 + 전구간 EV 선별. 조건 모두 충족 시 3분봉 종가에 신호.
{criteria_blocks}
{sim_table}
{gt_table}
"""
note = (
f"1단계 시뮬 · holdout {report.get('holdout_ratio', 0.15)} · "
f"발화 {len(holdout_fires)}건 / 체결가정 {len(capped)}건. "
"상단 카드: 초기 금액·총보유자산·초기 대비 증감율·수수료."
)
legend = (
"▲ 정답 매수 · ▼ 정답 매도 — 삼각형 = GT 체결 금액.
"
"● 시뮬 — 원 = holdout 발화 (차트). 테이블 = 일한도 적용 체결 순서."
)
if frames is not None:
meta_line = (
f"추세(참고): {trend} | 기간: {df_chart.index[0]} ~ {df_chart.index[-1]} "
f"| 봉 {len(df_chart)}"
)
else:
meta_line = (
f"추세·{SYMBOL} | lookback {CHART_LOOKBACK_DAYS}일 | "
f"초기 ₩{GT_INITIAL_CASH_KRW:,.0f}"
)
cards = _summary_cards_html(
close_val,
bb_txt,
gt_trades,
gt_pnl,
sim_sized_pnl,
sim_fixed_pnl,
len(capped),
go_flag,
model_note=model_note,
)
if frames is not None:
return build_chart_html(
df_chart,
trend,
note=note,
truth_trades=gt_trades,
sim_trades=_fires_to_chart_trades(holdout_fires),
# 차트 마커는 holdout 전체; 카드·테이블은 일한도 capped
title_suffix="1단계 시뮬레이션 (monitor · holdout)",
legend_html=legend,
footer_sections=sections,
cards_html=cards,
)
return wrap_chart_report_page(
page_title=f"{SYMBOL} 시뮬레이션",
heading=f"{COIN_NAME} ({SYMBOL}) 1단계 시뮬레이션",
meta_line=meta_line,
note_html=f"{note}
",
legend_html=legend,
cards_html=cards,
chart_html=(
"차트 데이터 없음 — "
"python scripts/01_download.py 후 재생성.
"
),
sections_html=sections,
)
def write_simulation_report_html(
report: dict[str, Any],
out_path: Path,
outcomes_path: Path | None = None,
matched_path: Path | None = None,
) -> Path:
"""
simulation_report.html 저장 (ground_truth 동일 스타일).
Args:
report: build_simulation_report 결과.
out_path: HTML 경로.
outcomes_path: fire_outcomes.csv.
matched_path: matched_rules.json.
Returns:
out_path.
"""
html = build_simulation_page_html(report, outcomes_path, matched_path)
out_path.parent.mkdir(parents=True, exist_ok=True)
out_path.write_text(html, encoding="utf-8")
return out_path