dsyoon/Bithumb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Ground Truth 2단계 1년 수익 시뮬 및 sim 차트 추가

Browse Source 
분할 매수/매도 PnL 시뮬, 체결 타점·거래시작 마커, x축 unix 변환 수정.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dsyoon
2026-06-09 10:06:43 +09:00
parent 6f008012c2
commit 75399ce79c
14 changed files with 16989 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
src/deepcoin/config.py
View File

@@ -68,6 +68,10 @@ class Settings:
ground_truth_chart_v1_file: Path
ground_truth_chart_v2_file: Path
ground_truth_chart_v3_file: Path
ground_truth_chart_sim_v1_file: Path
ground_truth_chart_sim_v2_file: Path
ground_truth_chart_sim_v3_file: Path
gt_sim_lookback_days: int
gt_initial_cash_krw: float
gt_trading_fee_rate: float
# Techniques (2단계)
@@ -121,6 +125,24 @@ def load_settings(env_path: Path | None = None) -> Settings:
gt_chart_v3 = _resolve_project_path(
os.getenv("GROUND_TRUTH_CHART_V3_FILE", "docs/02_ground_truth/ground_truth_chart_v3.html")
)
gt_chart_sim_v1 = _resolve_project_path(
os.getenv(
"GROUND_TRUTH_CHART_SIM_V1_FILE",
"docs/02_ground_truth/ground_truth_chart_sim_v1.html",
)
)
gt_chart_sim_v2 = _resolve_project_path(
os.getenv(
"GROUND_TRUTH_CHART_SIM_V2_FILE",
"docs/02_ground_truth/ground_truth_chart_sim_v2.html",
)
)
gt_chart_sim_v3 = _resolve_project_path(
os.getenv(
"GROUND_TRUTH_CHART_SIM_V3_FILE",
"docs/02_ground_truth/ground_truth_chart_sim_v3.html",
)
)
tech_dir_raw = os.getenv("TECHNIQUES_DIR", "data/techniques")
tech_dir = Path(tech_dir_raw)
@@ -165,6 +187,10 @@ def load_settings(env_path: Path | None = None) -> Settings:
ground_truth_chart_v1_file=gt_chart_v1,
ground_truth_chart_v2_file=gt_chart_v2,
ground_truth_chart_v3_file=gt_chart_v3,
ground_truth_chart_sim_v1_file=gt_chart_sim_v1,
ground_truth_chart_sim_v2_file=gt_chart_sim_v2,
ground_truth_chart_sim_v3_file=gt_chart_sim_v3,
gt_sim_lookback_days=int(os.getenv("GT_SIM_LOOKBACK_DAYS", "365")),
gt_initial_cash_krw=float(os.getenv("GT_INITIAL_CASH_KRW", "400000")),
gt_trading_fee_rate=float(os.getenv("GT_TRADING_FEE_RATE", "0.0005")),
techniques_dir=tech_dir,

408
src/deepcoin/ground_truth/chart.py
View File

@@ -22,6 +22,149 @@ def _data_js_path(html_path: Path) -> Path:
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 _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"])
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)
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": df["close"].astype(float).tolist(),
"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,
@@ -54,42 +197,7 @@ def render_ground_truth_chart(
if max_candles > 0 and len(df) > max_candles:
df = df.iloc[-max_candles:].reset_index(drop=True)
times = (pd.to_datetime(df["datetime"]).astype("int64") // 10**9).astype(int).tolist()
buy_markers = []
sell_markers = []
for sig in gt_result.get("signals") or []:
ts = int(pd.Timestamp(sig["datetime"]).timestamp())
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)
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),
}
payload = {
"times": times,
"open": df["open"].astype(float).tolist(),
"high": df["high"].astype(float).tolist(),
"low": df["low"].astype(float).tolist(),
"close": df["close"].astype(float).tolist(),
"buy_markers": buy_markers,
"sell_markers": sell_markers,
"meta": chart_meta,
"bar_count": len(df),
}
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)
@@ -132,6 +240,7 @@ _HTML_TEMPLATE = """<!DOCTYPE html>
#detail-wrap { margin: 0 24px 12px; display: none; }
#detail-wrap h2 { font-size: 15px; margin: 0 0 8px; }
#detail { height: 360px; background: #fff; border: 1px solid #ddd; overflow: visible; }
__EXTRA_STYLES__
</style>
</head>
<body>
@@ -139,6 +248,7 @@ _HTML_TEMPLATE = """<!DOCTYPE html>
<h1 id="title">Ground Truth Chart</h1>
<div class="meta" id="meta"></div>
</header>
__EXTRA_BODY__
<div class="toolbar">
<div class="toolbar-group">
<button id="btn-home" class="home" title="전체 2년 화면으로 복귀">홈</button>
@@ -223,6 +333,7 @@ _HTML_TEMPLATE = """<!DOCTYPE html>
}
const MARKER_FONT = "bold 18px Malgun Gothic, Arial, sans-serif";
const SIM_START_COLOR = "#7b1fa2";
function markerSuffix(signalType) {
if (signalType === "pullback") return "*";
@@ -243,6 +354,37 @@ _HTML_TEMPLATE = """<!DOCTYPE html>
ctx.fillText(label, lx, ly);
}
function drawSimStartMarker(u, marker) {
if (!marker) return;
const ctx = u.ctx;
const x = u.valToPos(marker.time, "x", true);
const y = u.valToPos(marker.price, "y", true);
if (x < u.bbox.left || x > u.bbox.left + u.bbox.width) return;
const color = SIM_START_COLOR;
const s = 10;
const gap = 14;
ctx.fillStyle = color;
ctx.beginPath();
ctx.moveTo(x, y - gap);
ctx.lineTo(x - s, y - gap - 18);
ctx.lineTo(x + s, y - gap - 18);
ctx.closePath();
ctx.fill();
const label = marker.label || "거래시작";
ctx.font = MARKER_FONT;
ctx.textAlign = "center";
ctx.textBaseline = "bottom";
const labelY = y - gap - 18 - 6;
ctx.lineWidth = 3;
ctx.lineJoin = "round";
ctx.strokeStyle = "rgba(255,255,255,0.85)";
ctx.strokeText(label, x, labelY);
ctx.fillStyle = color;
ctx.fillText(label, x, labelY);
ctx.textAlign = "left";
ctx.textBaseline = "alphabetic";
}
function drawMarkers(u, buys, sells) {
if (!showMarkers) return;
const ctx = u.ctx;
@@ -326,7 +468,10 @@ _HTML_TEMPLATE = """<!DOCTYPE html>
],
cursor: { drag: { x: true, y: false, setScale: true } },
hooks: {
draw: [(u) => drawMarkers(u, DATA.buy_markers, DATA.sell_markers)],
draw: [(u) => {
drawSimStartMarker(u, DATA.sim_start_marker);
drawMarkers(u, DATA.buy_markers, DATA.sell_markers);
}],
},
};
overviewPlot = new uPlot(opts, [DATA.times, DATA.close], document.getElementById("overview"));
@@ -381,6 +526,14 @@ _HTML_TEMPLATE = """<!DOCTYPE html>
const t0 = DATA.times[startIdx];
const t1 = DATA.times[end - 1];
const markers = [];
if (DATA.sim_start_marker) {
const sm = DATA.sim_start_marker;
if (sm.time >= t0 && sm.time <= t1) markers.push({
time: sm.time, position: "aboveBar",
color: SIM_START_COLOR, shape: "arrowDown", size: 3,
text: sm.label || "거래시작",
});
}
if (showMarkers) {
DATA.buy_markers.forEach(m => {
if (m.time >= t0 && m.time <= t1) markers.push({
@@ -503,17 +656,21 @@ _HTML_TEMPLATE = """<!DOCTYPE html>
}
}
__EXTRA_SCRIPT__
function init() {
DATA = window.CHART_DATA;
if (!DATA) throw new Error("차트 데이터 JS 없음");
const m = DATA.meta;
const simMode = !!DATA.sim_pnl;
const chartDays = m.chart_lookback_days || m.lookback_days;
const gtDays = m.gt_lookback_days || m.lookback_days;
const chartLabel = chartDays >= 365 ? `${Math.round(chartDays / 365)}년` : `${chartDays}일`;
const gtLabel = gtDays >= 365 ? `${Math.round(gtDays / 365)}년` : `${gtDays}일`;
const tier = m.chart_tier ? ` ${m.chart_tier.toUpperCase()}` : "";
const simSuffix = simMode ? " · 2단계 시뮬" : "";
document.getElementById("title").textContent =
`${m.symbol} Ground Truth${tier} (${m.interval_label}) — 차트 ${chartLabel} / GT ${gtLabel}`;
`${m.symbol} Ground Truth${tier} (${m.interval_label}) — 차트 ${chartLabel} / GT ${gtLabel}${simSuffix}`;
document.getElementById("btn-all").textContent = `전체 ${chartLabel}`;
const chartFrom = m.chart_data_from || m.data_from;
const chartTo = m.chart_data_to || m.data_to;
@@ -523,8 +680,13 @@ _HTML_TEMPLATE = """<!DOCTYPE html>
: tierKey === "v2"
? "B/S=스윙 B*=눌림목"
: "B/S=스윙 B*=눌림 B^=돌파 Bd/Sd=다이버전스";
const markerRange = simMode && m.sim_period_from
? `체결 ${DATA.buy_markers.length}/${DATA.sell_markers.length} · ${m.sim_period_from.slice(0, 16)} ~ ${(m.sim_period_to || chartTo).slice(0, 16)}`
: gtLabel;
const legendExtra = simMode ? " | ▼보라=거래시작" : "";
document.getElementById("meta").textContent =
`차트 ${chartFrom} ~ ${chartTo} (${DATA.bar_count.toLocaleString()}봉) | 매수 ${DATA.buy_markers.length} / 매도 ${DATA.sell_markers.length} (${gtLabel}) | ${legend}`;
`차트 ${chartFrom} ~ ${chartTo} (${DATA.bar_count.toLocaleString()}봉) | 매수 ${DATA.buy_markers.length} / 매도 ${DATA.sell_markers.length} (${markerRange}) | ${legend}${legendExtra}`;
if (simMode) renderSimPanel();
updateLegInfo();
document.getElementById("status").textContent =
`전체 ${DATA.bar_count.toLocaleString()}봉 | 드래그=줌, 더블클릭=리셋`;
@@ -575,5 +737,169 @@ _HTML_TEMPLATE = """<!DOCTYPE html>
def _html_template(data_js_name: str) -> str:
"""차트 HTML 템플릿을 생성한다."""
return _HTML_TEMPLATE.replace("__DATA_JS_NAME__", data_js_name)
"""1단계 GT 차트 HTML 템플릿을 생성한다."""
return _build_html_template(data_js_name, sim_mode=False)
_SIM_EXTRA_STYLES = """
.sim-panel { margin: 12px 24px 0; padding: 16px 20px; background: #fff; border: 1px solid #ddd; border-radius: 4px; }
.sim-panel h2 { margin: 0 0 12px; font-size: 16px; }
.sim-grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(180px, 1fr)); gap: 12px; }
.sim-card { padding: 10px 12px; background: #fafafa; border: 1px solid #eee; border-radius: 4px; }
.sim-card .label { font-size: 12px; color: #777; margin-bottom: 4px; }
.sim-card .value { font-size: 18px; font-weight: bold; }
.sim-card .value.positive { color: #2e7d32; }
.sim-card .value.negative { color: #c62828; }
.sim-note { margin-top: 10px; font-size: 12px; color: #666; line-height: 1.5; }
#trade-table-wrap { margin: 12px 24px 0; background: #fff; border: 1px solid #ddd; border-radius: 4px; overflow: hidden; }
#trade-table-wrap summary { padding: 10px 16px; cursor: pointer; font-size: 14px; background: #fafafa; border-bottom: 1px solid #eee; }
.trade-table { width: 100%; border-collapse: collapse; font-size: 12px; }
.trade-table th, .trade-table td { padding: 6px 10px; border-bottom: 1px solid #eee; text-align: right; }
.trade-table th:first-child, .trade-table td:first-child { text-align: left; }
.trade-table th { background: #f5f5f5; position: sticky; top: 0; }
.trade-table tr.skipped td { color: #999; }
.trade-scroll { max-height: 240px; overflow: auto; }
"""
_SIM_EXTRA_BODY = """
<section class="sim-panel" id="sim-panel">
<h2>2단계 수익 시뮬레이션 (최근 1년 · 초기 40만원)</h2>
<div class="sim-grid" id="sim-grid"></div>
<div class="sim-note" id="sim-note"></div>
</section>
<details id="trade-table-wrap">
<summary>체결 내역 (<span id="trade-count">0</span>건)</summary>
<div class="trade-scroll">
<table class="trade-table">
<thead>
<tr>
<th>시각</th><th>구분</th><th>유형</th><th>가격</th><th>주문금액</th>
<th>수수료</th><th>현금</th><th>코인</th><th>비고</th>
</tr>
</thead>
<tbody id="trade-body"></tbody>
</table>
</div>
</details>
"""
_SIM_EXTRA_SCRIPT = """
function fmtMoney(v) {
return Math.round(v).toLocaleString("ko-KR") + "원";
}
function fmtPct(v) {
const sign = v > 0 ? "+" : "";
return sign + v.toFixed(2) + "%";
}
function renderSimPanel() {
const p = DATA.sim_pnl;
const retClass = p.total_return_pct >= 0 ? "positive" : "negative";
document.getElementById("sim-grid").innerHTML = [
["초기 자본", fmtMoney(p.initial_cash_krw), ""],
["최종 평가액", fmtMoney(p.final_equity_krw), retClass],
["손익", fmtMoney(p.total_pnl_krw), retClass],
["수익률", fmtPct(p.total_return_pct), retClass],
["현금 잔고", fmtMoney(p.final_cash_krw), ""],
["보유 코인", p.final_coin_qty.toFixed(8), ""],
["코인 평가", fmtMoney(p.final_coin_value_krw), ""],
["매수/매도", `${p.buys_executed}/${p.sells_executed}건`, ""],
].map(([label, value, cls]) =>
`<div class="sim-card"><div class="label">${label}</div><div class="value ${cls}">${value}</div></div>`
).join("");
document.getElementById("sim-note").textContent =
`시뮬 기간: ${p.period_from} ~ ${p.period_to} (${p.sim_lookback_days}일) | ` +
`신호 ${p.signals_in_period}건 | 분할매수/매도 클러스터 적용 | ` +
`스킵 매수 ${p.buys_skipped} / 매도 ${p.sells_skipped} | 수수료 ${(p.fee_rate * 100).toFixed(2)}%`;
const tbody = document.getElementById("trade-body");
tbody.innerHTML = "";
(p.trades || []).forEach(t => {
const tr = document.createElement("tr");
if (t.skipped) tr.className = "skipped";
tr.innerHTML = `
<td>${t.datetime}</td>
<td>${t.side === "buy" ? "매수" : "매도"}</td>
<td>${t.signal_type}</td>
<td>${fmtPrice(t.price)}</td>
<td>${t.order_krw ? fmtMoney(t.order_krw) : "-"}</td>
<td>${t.fee_krw ? fmtMoney(t.fee_krw) : "-"}</td>
<td>${fmtMoney(t.cash_after)}</td>
<td>${t.coin_after.toFixed(8)}</td>
<td>${t.skipped ? (t.skip_reason || "스킵") : "분할 " + t.cluster_size}</td>`;
tbody.appendChild(tr);
});
document.getElementById("trade-count").textContent = String((p.trades || []).length);
}
"""
def _build_html_template(data_js_name: str, sim_mode: bool) -> str:
"""GT/시뮬 차트 HTML 템플릿을 생성한다."""
html = _HTML_TEMPLATE.replace("__DATA_JS_NAME__", data_js_name)
if sim_mode:
html = (
html.replace("__EXTRA_STYLES__", _SIM_EXTRA_STYLES)
.replace("__EXTRA_BODY__", _SIM_EXTRA_BODY)
.replace("__EXTRA_SCRIPT__", _SIM_EXTRA_SCRIPT)
)
else:
html = (
html.replace("__EXTRA_STYLES__", "")
.replace("__EXTRA_BODY__", "")
.replace("__EXTRA_SCRIPT__", "")
)
return html
def _sim_html_template(data_js_name: str) -> str:
"""2단계 sim 차트 HTML 템플릿을 생성한다."""
return _build_html_template(data_js_name, sim_mode=True)
def render_ground_truth_sim_chart(
db_path: Path,
symbol: str,
gt_result: dict[str, Any],
sim_pnl: dict[str, Any],
output_path: Path,
chart_lookback_days: int | None = None,
max_candles: int = DEFAULT_MAX_CANDLES,
) -> Path:
"""GT 타점 + 2단계 시뮬 수익 결과가 표시된 HTML 차트를 생성한다.
Args:
db_path: SQLite 경로.
symbol: 코인 심볼.
gt_result: build_ground_truth 결과.
sim_pnl: simulate_gt_signals_pnl 결과.
output_path: HTML 출력 경로.
chart_lookback_days: 차트 표시 일수.
max_candles: 0이면 전체.
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, sim_pnl=sim_pnl
)
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(_sim_html_template(data_js_name), encoding="utf-8")
return output_path

292
src/deepcoin/ground_truth/pnl.py
View File

@@ -3,6 +3,7 @@
from __future__ import annotations
from dataclasses import asdict, dataclass
from datetime import datetime, timedelta
from typing import Any
@@ -22,6 +23,28 @@ class LegPnl:
btc_qty: float
@dataclass
class SignalTrade:
"""신호 1건 실행 기록."""
trade_id: int
side: str
signal_type: str
marker_id: int | None
datetime: str
price: float
cash_before: float
cash_after: float
coin_before: float
coin_after: float
order_krw: float
order_coin: float
fee_krw: float
cluster_size: int
skipped: bool
skip_reason: str | None = None
def simulate_gt_pnl(
legs: list[dict[str, Any]],
initial_cash_krw: float = 400_000.0,
@@ -99,3 +122,272 @@ def simulate_gt_pnl(
"period_to": period_to,
"leg_pnls": [asdict(x) for x in leg_pnls],
}
def _parse_signal_dt(value: str) -> datetime:
"""GT signal datetime 문자열을 파싱한다."""
return datetime.strptime(value, "%Y-%m-%d %H:%M:%S")
def _cluster_signals(signals: list[dict[str, Any]]) -> list[tuple[str, list[dict[str, Any]]]]:
"""연속 동일 side 신호를 클러스터로 묶는다."""
ordered = sorted(signals, key=lambda s: (s["bar_index"], s.get("marker_id", 0)))
clusters: list[tuple[str, list[dict[str, Any]]]] = []
current_side: str | None = None
current: list[dict[str, Any]] = []
for sig in ordered:
side = sig["side"]
if current_side is None:
current_side = side
current = [sig]
continue
if side == current_side:
current.append(sig)
continue
clusters.append((current_side, current))
current_side = side
current = [sig]
if current_side and current:
clusters.append((current_side, current))
return clusters
def simulate_gt_signals_pnl(
signals: list[dict[str, Any]],
initial_cash_krw: float = 400_000.0,
fee_rate: float = 0.0005,
min_order_krw: float = 5_000.0,
sim_lookback_days: int = 365,
data_end: str | None = None,
last_mark_price: float | None = None,
) -> dict[str, Any]:
"""GT 매수·매도 신호를 시간순 실행한 2단계 포트폴리오 시뮬레이션.
- 시뮬 기간: data_end 기준 최근 sim_lookback_days
- 연속 매수: 가용 원화를 매수 신호 수로 균등 분할
- 연속 매도: 보유 코인을 매도 신호 수로 균등 분할
- 원화 부족 시 매수 스킵, 코인 없으면 매도 스킵
Args:
signals: GT signals 리스트.
initial_cash_krw: 시뮬 시작 원화.
fee_rate: 편도 수수료율.
min_order_krw: 최소 주문 금액.
sim_lookback_days: 시뮬 기간(일).
data_end: 데이터 종료 시각 문자열. None이면 마지막 신호 시각.
last_mark_price: 미청산 코인 평가 가격. None이면 마지막 체결가.
Returns:
요약 + 체결/스킵 내역 dict.
"""
if not signals:
return _empty_signal_pnl(initial_cash_krw, fee_rate, sim_lookback_days)
end_dt = _parse_signal_dt(data_end) if data_end else max(
_parse_signal_dt(s["datetime"]) for s in signals
)
start_dt = end_dt - timedelta(days=sim_lookback_days)
start_str = start_dt.strftime("%Y-%m-%d %H:%M:%S")
period_signals = [
s for s in signals if _parse_signal_dt(s["datetime"]) >= start_dt
]
if not period_signals:
return _empty_signal_pnl(
initial_cash_krw,
fee_rate,
sim_lookback_days,
period_from=start_str,
period_to=end_dt.strftime("%Y-%m-%d %H:%M:%S"),
)
cash = float(initial_cash_krw)
coin_qty = 0.0
trades: list[SignalTrade] = []
trade_id = 0
buys_executed = 0
sells_executed = 0
buys_skipped = 0
sells_skipped = 0
mark_price = float(last_mark_price or period_signals[-1]["price"])
for side, cluster in _cluster_signals(period_signals):
cluster_size = len(cluster)
if side == "buy":
budget = cash
per_buy = budget / cluster_size if cluster_size else 0.0
for sig in cluster:
trade_id += 1
price = float(sig["price"])
cash_before = cash
coin_before = coin_qty
order_krw = min(per_buy, cash)
if order_krw < min_order_krw:
buys_skipped += 1
trades.append(
SignalTrade(
trade_id=trade_id,
side="buy",
signal_type=str(sig.get("signal_type", "buy")),
marker_id=sig.get("marker_id"),
datetime=sig["datetime"],
price=price,
cash_before=round(cash_before, 0),
cash_after=round(cash, 0),
coin_before=round(coin_before, 8),
coin_after=round(coin_qty, 8),
order_krw=0.0,
order_coin=0.0,
fee_krw=0.0,
cluster_size=cluster_size,
skipped=True,
skip_reason="원화 부족",
)
)
continue
fee = order_krw * fee_rate
bought = (order_krw - fee) / price
cash -= order_krw
coin_qty += bought
buys_executed += 1
trades.append(
SignalTrade(
trade_id=trade_id,
side="buy",
signal_type=str(sig.get("signal_type", "buy")),
marker_id=sig.get("marker_id"),
datetime=sig["datetime"],
price=price,
cash_before=round(cash_before, 0),
cash_after=round(cash, 0),
coin_before=round(coin_before, 8),
coin_after=round(coin_qty, 8),
order_krw=round(order_krw, 0),
order_coin=round(bought, 8),
fee_krw=round(fee, 0),
cluster_size=cluster_size,
skipped=False,
)
)
else:
budget_coin = coin_qty
per_sell = budget_coin / cluster_size if cluster_size else 0.0
for sig in cluster:
trade_id += 1
price = float(sig["price"])
cash_before = cash
coin_before = coin_qty
order_coin = min(per_sell, coin_qty)
order_krw = order_coin * price
if order_coin <= 0 or order_krw < min_order_krw:
sells_skipped += 1
trades.append(
SignalTrade(
trade_id=trade_id,
side="sell",
signal_type=str(sig.get("signal_type", "sell")),
marker_id=sig.get("marker_id"),
datetime=sig["datetime"],
price=price,
cash_before=round(cash_before, 0),
cash_after=round(cash, 0),
coin_before=round(coin_before, 8),
coin_after=round(coin_qty, 8),
order_krw=0.0,
order_coin=0.0,
fee_krw=0.0,
cluster_size=cluster_size,
skipped=True,
skip_reason="코인 부족",
)
)
continue
gross = order_coin * price
fee = gross * fee_rate
cash += gross - fee
coin_qty -= order_coin
sells_executed += 1
trades.append(
SignalTrade(
trade_id=trade_id,
side="sell",
signal_type=str(sig.get("signal_type", "sell")),
marker_id=sig.get("marker_id"),
datetime=sig["datetime"],
price=price,
cash_before=round(cash_before, 0),
cash_after=round(cash, 0),
coin_before=round(coin_before, 8),
coin_after=round(coin_qty, 8),
order_krw=round(gross, 0),
order_coin=round(order_coin, 8),
fee_krw=round(fee, 0),
cluster_size=cluster_size,
skipped=False,
)
)
coin_value = coin_qty * mark_price
final_equity = cash + coin_value
total_pnl = final_equity - initial_cash_krw
total_return_pct = total_pnl / initial_cash_krw * 100.0
return {
"mode": "signal_split",
"initial_cash_krw": initial_cash_krw,
"final_cash_krw": round(cash, 0),
"final_coin_qty": round(coin_qty, 8),
"final_mark_price": round(mark_price, 2),
"final_coin_value_krw": round(coin_value, 0),
"final_equity_krw": round(final_equity, 0),
"total_pnl_krw": round(total_pnl, 0),
"total_return_pct": round(total_return_pct, 2),
"fee_rate": fee_rate,
"sim_lookback_days": sim_lookback_days,
"period_from": start_str,
"period_to": end_dt.strftime("%Y-%m-%d %H:%M:%S"),
"signals_in_period": len(period_signals),
"buys_executed": buys_executed,
"sells_executed": sells_executed,
"buys_skipped": buys_skipped,
"sells_skipped": sells_skipped,
"trades": [asdict(t) for t in trades],
}
def _empty_signal_pnl(
initial_cash_krw: float,
fee_rate: float,
sim_lookback_days: int,
period_from: str | None = None,
period_to: str | None = None,
) -> dict[str, Any]:
"""신호가 없을 때의 빈 시뮬 결과."""
return {
"mode": "signal_split",
"initial_cash_krw": initial_cash_krw,
"final_cash_krw": initial_cash_krw,
"final_coin_qty": 0.0,
"final_mark_price": 0.0,
"final_coin_value_krw": 0.0,
"final_equity_krw": initial_cash_krw,
"total_pnl_krw": 0.0,
"total_return_pct": 0.0,
"fee_rate": fee_rate,
"sim_lookback_days": sim_lookback_days,
"period_from": period_from,
"period_to": period_to,
"signals_in_period": 0,
"buys_executed": 0,
"sells_executed": 0,
"buys_skipped": 0,
"sells_skipped": 0,
"trades": [],
}