Files
Bithumb/deepcoin/matching/simulation_html.py
xavis d385456867 hybrid DD tier와 Option C 2차(+1000%) 검증을 추가하고 실거래 사이징을 정합한다.
인과 GT leg 엔진·drawdown tier·train 캘리브레이션, Phase 2 Go/No-Go 및 시뮬 리포트를 반영한다.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-01 16:09:28 +09:00

607 lines
21 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
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,
simulate_fixed_order_portfolio,
simulate_fixed_order_portfolio_steps,
simulate_sized_portfolio,
sort_fires_chronological,
)
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 "<tr><td colspan='8'>발화 없음</td></tr>"
sorted_fires = fires.sort_values("dt").reset_index(drop=True)
lines: list[str] = []
lines.append(
f"""
<tr class="initial-row">
<td>시작</td><td>-</td><td>-</td><td>-</td><td>-</td>
<td><b>₩{GT_INITIAL_CASH_KRW:,.0f}</b></td>
<td>-</td><td>초기 현금 (1회 ₩{LIVE_ORDER_KRW:,.0f} 가정)</td>
</tr>"""
)
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"<tr>"
f"<td>{str(r['dt'])[:16]}</td>"
f'<td class="{cls}">{mark}</td>'
f"<td>{r['rule_id']}</td>"
f"<td>{kind}</td>"
f'<td class="num">₩{float(r["close"]):,.0f}{ret_s}</td>'
f"<td><b>{total_s}</b>{hold_s}</td>"
f'<td class="{win_cls}">{win}</td>'
f"<td>leg_gt 구간 수익</td>"
f"</tr>"
)
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 "<tr><td colspan='6'>타점 없음</td></tr>"
step_key = {
(s["dt"], s["action"], float(s["price"]), float(s["weight"])): s
for s in steps
}
lines: list[str] = []
lines.append(
f"""
<tr class="initial-row">
<td>시작</td><td>-</td><td>-</td><td>-</td>
<td><b>₩{GT_INITIAL_CASH_KRW:,.0f}</b></td>
<td>초기 현금 (보유 0)</td>
</tr>"""
)
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"<tr>"
f"<td>{t['dt'][:16]}</td>"
f'<td class="{cls}">{mark}</td>'
f"<td>{w*100:.0f}%</td>"
f"<td>₩{t['price']:,.0f}{ret_s}</td>"
f"<td><b>{total_s}</b>{hold_s}</td>"
f"<td>{t.get('memo', '')}</td>"
f"</tr>"
)
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 = "",
sim_causal_gt_pnl: dict[str, Any] | None = None,
sim_causal_hybrid_pnl: dict[str, Any] | None = None,
sim_tier_enhanced_pnl: dict[str, Any] | None = None,
sim_primary_pnl: dict[str, Any] | None = None,
primary_sizing: str = "",
hybrid_go: bool = False,
phase2_go: bool = False,
) -> 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 모델 한 줄 요약.
sim_causal_gt_pnl: 인과 GT leg 엔진 요약.
sim_causal_hybrid_pnl: monitor buy + 인과 sell 하이브리드.
sim_tier_enhanced_pnl: monitor + conviction tier.
sim_primary_pnl: 검증 통과 권장 배분 경로.
primary_sizing: hybrid | causal_tier.
hybrid_go: hybrid tier Go/No-Go.
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 = (
"시뮬·GT tier 복리 (전기간, 상한 없음) — "
f'<span class="{go_cls}">{"GO" if go_flag else "NO-GO"}</span>'
)
sim_fixed_title = f"시뮬·고정 ₩{LIVE_ORDER_KRW:,}/회 (비교)"
primary_block = ""
if sim_primary_pnl and float(sim_primary_pnl.get("pnl_pct") or 0) != 0:
hgo_cls = "go-pass" if hybrid_go else "go-fail"
primary_block = stacked_summary_cards_html(
(
f"권장 primary · {primary_sizing or 'causal_tier'} · "
f'<span class="{hgo_cls}">hybrid {"GO" if hybrid_go else "NO-GO"}</span> · '
f'2차 {"GO" if phase2_go else "NO-GO"}'
),
pnl_cards_html(
sim_primary_pnl,
"Primary",
int(sim_primary_pnl.get("trade_count") or 0),
),
)
causal_block = ""
if sim_causal_gt_pnl and (
float(sim_causal_gt_pnl.get("pnl_pct") or 0) != 0
or sim_causal_gt_pnl.get("trade_count")
):
legs = sim_causal_gt_pnl.get("leg_count", "-")
causal_block += stacked_summary_cards_html(
f"인과 GT leg 엔진 (peak_local·분할매수·causal tier) · leg {legs}",
pnl_cards_html(
sim_causal_gt_pnl,
"인과 GT",
int(sim_causal_gt_pnl.get("trade_count") or 0),
),
)
if sim_causal_hybrid_pnl and (
float(sim_causal_hybrid_pnl.get("pnl_pct") or 0) != 0
or sim_causal_hybrid_pnl.get("trade_count")
):
legs_h = sim_causal_hybrid_pnl.get("leg_count", "-")
causal_block += stacked_summary_cards_html(
f"하이브리드 (monitor + DD tier) · 발화 {sim_causal_hybrid_pnl.get('input_fires', '-')}",
pnl_cards_html(
sim_causal_hybrid_pnl,
"하이브리드",
int(sim_causal_hybrid_pnl.get("trade_count") or 0),
),
)
if sim_tier_enhanced_pnl and (
float(sim_tier_enhanced_pnl.get("pnl_pct") or 0) != 0
or sim_tier_enhanced_pnl.get("trade_count")
):
ast = sim_tier_enhanced_pnl.get("alloc_stats") or {}
causal_block += stacked_summary_cards_html(
f"Enhanced tier (conviction·DD) · large매수 {ast.get('large_tier_buy_count', '-')}",
pnl_cards_html(
sim_tier_enhanced_pnl,
"Enhanced",
int(sim_tier_enhanced_pnl.get("trade_count") or 0),
),
)
return (
'<div class="summary-cards">'
+ 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),
)
+ primary_block
+ causal_block
+ "</div>"
)
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}
sim_fires = pd.DataFrame()
holdout_fires = pd.DataFrame()
if op.is_file():
outcomes = pd.read_csv(op)
outcomes["split"] = _split_train_valid_holdout(outcomes)
sim_fires = outcomes[outcomes["rule_id"].isin(monitor_ids)].copy()
holdout_fires = outcomes[
(outcomes["rule_id"].isin(monitor_ids)) & (outcomes["split"] == "holdout")
].copy()
compound_fires = sort_fires_chronological(sim_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"))
go_hybrid = report.get("go_no_go_hybrid") or {}
hybrid_go_flag = bool(go_hybrid.get("go"))
go_phase2 = report.get("go_no_go_option_c_phase2") or {}
phase2_go_flag = bool(go_phase2.get("go"))
pc = report.get("portfolio_compare") or {}
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(compound_fires, apply_dynamic_sizing=True)
sim_trades_fixed = fires_to_trade_list(compound_fires, 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)
hybrid_checks = go_hybrid.get("checks") or []
if hybrid_checks:
hybrid_rows = "".join(
f"<tr><td>{c.get('name')}</td>"
f"<td>{'PASS' if c.get('pass') else 'FAIL'}</td>"
f"<td>{c.get('value', '-')}</td></tr>"
for c in hybrid_checks
)
hgo_cls = "go-pass" if hybrid_go_flag else "go-fail"
go_table += (
f"<h2>Hybrid tier Go/No-Go · "
f'<span class="{hgo_cls}">{"GO" if hybrid_go_flag else "NO-GO"}</span></h2>'
f"<table><thead><tr><th>검사</th><th>결과</th><th>값</th></tr></thead>"
f"<tbody>{hybrid_rows}</tbody></table>"
)
phase2_checks = go_phase2.get("checks") or []
if phase2_checks:
p2_rows = "".join(
f"<tr><td>{c.get('name')}</td>"
f"<td>{'PASS' if c.get('pass') else 'FAIL'}</td>"
f"<td>{c.get('value', '-')}</td></tr>"
for c in phase2_checks
)
p2_cls = "go-pass" if phase2_go_flag else "go-fail"
go_table += (
f"<h2>Option C 2차 (+1000%) · "
f'<span class="{p2_cls}">{"GO" if phase2_go_flag else "NO-GO"}</span></h2>'
f"<table><thead><tr><th>검사</th><th>결과</th><th>값</th></tr></thead>"
f"<tbody>{p2_rows}</tbody></table>"
)
def _mark_note(price: float) -> str:
if price > 0:
return f" 총보유자산(미청산 포함)은 종가 ₩{price:,.0f} 평가."
return ""
sim_table = f"""
<h2>시뮬 타점 (전기간 {len(sim_fires)}건 → 복리 체결 {len(compound_fires)}건)</h2>
<p class="meta">총자산×GT비중×leg tier·보유현금 한도·전기간 복리(일한도 없음).
가격 열 (+/-) = <b>{label_mode}</b> 구간 수익%.{_mark_note(close_val)}</p>
<div class="table-scroll">
<table>
<thead><tr><th>시각</th><th>구분</th><th>규칙</th><th>유형</th><th>가격</th>
<th>총 평가금액</th><th>승/패</th><th>비고</th></tr></thead>
<tbody>{_sim_fire_table_rows(compound_fires, rules_by_id, sim_steps)}</tbody>
</table>
</div>"""
gt_table = f"""
<h2>정답 타점 (ground_truth)</h2>
<p class="meta">삼각형 크기 = GT 체결 금액. 매수 분할·매도 leg 반영.{_mark_note(close_val)}</p>
<div class="table-scroll">
<table>
<thead><tr><th>시각</th><th>구분</th><th>비중</th><th>가격</th><th>총 평가금액</th><th>해석</th></tr></thead>
<tbody>{_gt_table_rows(gt_trades, gt_steps)}</tbody>
</table>
</div>"""
sections = f"""
{go_table}
<h2>매수·매도 판단 기준 (monitor_rules)</h2>
<p class="meta">04 GT 프로필 + 전구간 EV 선별. 조건 <b>모두</b> 충족 시 3분봉 종가에 신호.</p>
{criteria_blocks}
{sim_table}
{gt_table}
"""
note = (
f"1단계 시뮬 · holdout {report.get('holdout_ratio', 0.15)} · "
f"전기간 발화 {len(sim_fires)}건 / holdout {len(holdout_fires)}건. "
"상단 카드: 초기 금액·총보유자산·초기 대비 증감율·수수료."
)
legend = (
"▲ <b>정답 매수</b> · ▼ <b>정답 매도</b> — 삼각형 = GT 체결 금액.<br>"
"● <b>시뮬</b> — holdout 발화 (차트). 테이블 = 전기간 GT tier 복리 체결."
)
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(compound_fires),
go_flag,
model_note=model_note,
sim_causal_gt_pnl=pc.get("sim_causal_gt"),
sim_causal_hybrid_pnl=pc.get("sim_causal_hybrid"),
sim_tier_enhanced_pnl=pc.get("sim_tier_enhanced"),
sim_primary_pnl=pc.get("sim_primary"),
primary_sizing=str(pc.get("primary_sizing") or ""),
hybrid_go=hybrid_go_flag,
phase2_go=phase2_go_flag,
)
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; 카드·테이블은 전기간 GT tier 복리
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"<p class='note'>{note}</p>",
legend_html=legend,
cards_html=cards,
chart_html=(
"<p class='note'>차트 데이터 없음 — "
"<code>python scripts/01_download.py</code> 후 재생성.</p>"
),
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