diff --git a/.env.example b/.env.example
index 825df4d..41a98dc 100644
--- a/.env.example
+++ b/.env.example
@@ -10,10 +10,14 @@ CHART_LOOKBACK_DAYS=365
# 02 Ground Truth
GT_MIN_ORDER_KRW=5000
-GT_MAX_BUY_ORDER_KRW=100000
GT_BUY_PCT_LARGE_LEG=1.0
GT_BUY_PCT_SMALL_LEG=0.05
GT_LARGE_LEG_TOP_PCT=0.2
+# 시뮬·스캔: 1=인과적(운영 정합), 0=사후 ZigZag(정답 라벨용)
+GT_SIGNAL_CAUSAL=1
+SIM_CAUSAL_TIER=1
+GT_BUY_WEIGHT_RULE=inverse_price_normalized
+GT_SELL_SPLIT_WEIGHTS=0.65,0.35
# 04 매칭
MATCH_LABEL_MODE=leg_gt
@@ -28,7 +32,7 @@ SIM_FEE_STRESS_MULT=2.0
MONITOR_ALERT_COOLDOWN_MIN=180
MONITOR_ALERT_KRW_AMOUNT=100000
-# 3 실거래 (오픈 시에만 1)
+# 3 실거래 (오픈 시에만 1) — 아래 한도는 06_live 전용, 시뮬(04)은 GT tier 복리 사용
LIVE_TRADING_ENABLED=0
LIVE_ORDER_KRW=100000
LIVE_BUY_PCT_LARGE=1.0
diff --git a/config.py b/config.py
index f6c7e42..f56545a 100644
--- a/config.py
+++ b/config.py
@@ -199,14 +199,30 @@ GT_BUY_MIN_BARS = _getenv_int("GT_BUY_MIN_BARS", "24")
GT_MAX_BUYS_PER_LEG = _getenv_int("GT_MAX_BUYS_PER_LEG", "12")
GT_MAX_SELLS_PER_LEG = _getenv_int("GT_MAX_SELLS_PER_LEG", "2")
GT_SELL_SPLIT_GAP_PCT = _getenv_float("GT_SELL_SPLIT_GAP_PCT", "2.5")
+GT_SELL_SPLIT_WEIGHTS: tuple[float, ...] = tuple(
+ float(x.strip())
+ for x in _getenv("GT_SELL_SPLIT_WEIGHTS", "0.65,0.35").split(",")
+ if x.strip()
+) or (0.65, 0.35)
+GT_BUY_WEIGHT_RULE = _getenv("GT_BUY_WEIGHT_RULE", "inverse_price_normalized")
GT_MARKER_SIZE_MIN = _getenv_int("GT_MARKER_SIZE_MIN", "10")
GT_MARKER_SIZE_MAX = _getenv_int("GT_MARKER_SIZE_MAX", "32")
GT_INITIAL_CASH_KRW = _getenv_int("GT_INITIAL_CASH_KRW", "1000000")
GT_MIN_ORDER_KRW = _getenv_int("GT_MIN_ORDER_KRW", "5000")
-GT_MAX_BUY_ORDER_KRW = _getenv_int("GT_MAX_BUY_ORDER_KRW", "100000")
GT_BUY_PCT_LARGE_LEG = _getenv_float("GT_BUY_PCT_LARGE_LEG", "1.0")
GT_BUY_PCT_SMALL_LEG = _getenv_float("GT_BUY_PCT_SMALL_LEG", "0.05")
GT_LARGE_LEG_TOP_PCT = _getenv_float("GT_LARGE_LEG_TOP_PCT", "0.2")
+# 시뮬·스캔: 1=인과적 신호·tier (미래 데이터 미사용, 운영 정합)
+GT_SIGNAL_CAUSAL = _getenv("GT_SIGNAL_CAUSAL", "1").strip().lower() in (
+ "1",
+ "true",
+ "yes",
+)
+SIM_CAUSAL_TIER = _getenv("SIM_CAUSAL_TIER", "1").strip().lower() in (
+ "1",
+ "true",
+ "yes",
+)
TRADING_FEE_RATE = _getenv_float("TRADING_FEE_RATE", "0.0005")
# --- 모니터 / API 수집 ---
diff --git a/data/ground_truth/ground_truth_trades.json b/data/ground_truth/ground_truth_trades.json
index 933f057..ac8add6 100644
--- a/data/ground_truth/ground_truth_trades.json
+++ b/data/ground_truth/ground_truth_trades.json
@@ -16,7 +16,7 @@
"pivot": "peak",
"price": "High",
"weight_rule": "fixed_split_or_full",
- "weights_two_sell": [
+ "weights_split": [
0.65,
0.35
],
@@ -24,13 +24,13 @@
"max_sells_per_leg": 2
},
"capital": {
- "buy": "min(total_asset * w_share * tier_scale, cash/(1+fee))",
+ "buy": "target = total_asset * (weight/remaining_weights) * tier_scale; amount = min(target, available_cash/(1+fee))",
"optimal_buy_rate": "weight / sum(remaining_buy_weights_in_leg)",
"large_leg_top_pct": 0.2,
"pct_large_leg": 1.0,
"pct_small_leg": 0.05,
"min_order_krw": 5000.0,
- "sell": "leg_qty * sell_weight * price (last sell = full leg_qty)"
+ "sell": "sell_base_qty * sell_weight * price (last sell = full leg_qty)"
},
"execution": {
"chrono": "amount_krw 배분·summary.pnl_pct = 시각순 체결(매도 후 현금 → 다음 매수 반영)",
@@ -42,7 +42,7 @@
"interval_min": 3,
"lookback_days": 365,
"period_start": "2025-06-04 03:57:00",
- "period_end": "2026-05-31 15:58:00",
+ "period_end": "2026-05-31 19:36:00",
"trend_at_end": "range",
"params": {
"min_swing_pct": 4.0,
@@ -55,40 +55,39 @@
"max_sells_per_leg": 2
},
"summary": {
- "pivot_candidates": 382,
+ "pivot_candidates": 383,
"sell_peaks": 75,
- "trade_count": 454,
- "buy_count": 304,
+ "trade_count": 456,
+ "buy_count": 306,
"sell_count": 150,
"round_trips": 76,
- "sum_sell_leg_return_pct": 1634.3,
+ "sum_sell_leg_return_pct": 1631.75,
"initial_cash_krw": 1000000,
- "final_asset_krw": 43963648.0,
- "pnl_krw": 42963648.0,
- "pnl_pct": 4296.36,
- "total_fees_krw": 238798.0,
- "cash_krw": 43963648.0,
+ "final_asset_krw": 43913513.0,
+ "pnl_krw": 42913513.0,
+ "pnl_pct": 4291.35,
+ "total_fees_krw": 240578.0,
+ "cash_krw": 43913513.0,
"holding_qty": 0.0,
"holding_value_krw": 0.0,
- "mark_price": 503.0,
+ "mark_price": 487.0,
"fee_rate": 0.0005,
- "realized_final_asset_krw": 43963648.0,
- "realized_pnl_krw": 42963648.0,
- "realized_pnl_pct": 4296.36,
+ "realized_final_asset_krw": 43913513.0,
+ "realized_pnl_krw": 42913513.0,
+ "realized_pnl_pct": 4291.35,
"unrealized_pnl_krw": 0.0,
"execution_order": "chronological",
"order_amount_min_krw": 5000,
- "order_amount_max_buy_krw": 100000,
"buy_pct_large_leg": 1.0,
"buy_pct_small_leg": 0.05,
"large_leg_top_pct": 0.2,
- "buy_executed": 295,
+ "buy_executed": 297,
"buy_skipped": 9,
"sell_executed": 150,
"sell_skipped": 0,
- "buy_total_krw": 217196361.0,
+ "buy_total_krw": 219000974.0,
"large_leg_count": 16,
- "buy_amount_avg_krw": 736259.0,
+ "buy_amount_avg_krw": 737377.0,
"buy_amount_min_krw": 5000,
"buy_amount_max_krw": 14200590.0
},
@@ -5974,27 +5973,53 @@
"dt": "2026-05-31 09:36:00",
"action": "buy",
"price": 495.0,
- "memo": "저점 분할 매수 · 비중 100% · leg#75(기간말)",
- "weight": 1.0,
- "amount_krw": 2196518.0,
+ "memo": "저점 분할 매수 · 비중 33% · leg#75(기간말)",
+ "weight": 0.331,
+ "amount_krw": 727048.0,
"leg_id": 75,
"bb_pos": 0.0,
"rsi": 33.3,
"pivot_kind": "trough",
- "forward_return_pct": 1.62
+ "forward_return_pct": -1.62
},
{
- "dt": "2026-05-31 15:58:00",
+ "dt": "2026-05-31 17:39:00",
+ "action": "buy",
+ "price": 498.0,
+ "memo": "저점 분할 매수 · 비중 33% · leg#75(기간말)",
+ "weight": 0.329,
+ "amount_krw": 1080300.0,
+ "leg_id": 75,
+ "bb_pos": 0.028,
+ "rsi": 38.5,
+ "pivot_kind": "trough",
+ "forward_return_pct": -2.21
+ },
+ {
+ "dt": "2026-05-31 19:33:00",
+ "action": "buy",
+ "price": 482.0,
+ "memo": "저점 분할 매수 · 비중 34% · leg#75(기간말)",
+ "weight": 0.34,
+ "amount_krw": 2193783.0,
+ "leg_id": 75,
+ "bb_pos": 0.0,
+ "rsi": 16.7,
+ "pivot_kind": "trough",
+ "forward_return_pct": 1.04
+ },
+ {
+ "dt": "2026-05-31 19:36:00",
"action": "sell",
- "price": 503.0,
+ "price": 487.0,
"memo": "기간말 잔여 청산 · leg#75",
"weight": 1.0,
- "amount_krw": 2232017.0,
+ "amount_krw": 3988276.0,
"leg_id": 75,
- "bb_pos": 0.091,
- "rsi": 35.7,
+ "bb_pos": 0.042,
+ "rsi": 16.7,
"pivot_kind": "peak",
- "forward_return_pct": 1.62
+ "forward_return_pct": -0.93
}
]
}
\ No newline at end of file
diff --git a/deepcoin/ground_truth/ground_truth.py b/deepcoin/ground_truth/ground_truth.py
index 401a81e..f716a88 100644
--- a/deepcoin/ground_truth/ground_truth.py
+++ b/deepcoin/ground_truth/ground_truth.py
@@ -33,7 +33,6 @@ from config import (
GT_INITIAL_CASH_KRW,
GT_LARGE_LEG_TOP_PCT,
GT_MIN_ORDER_KRW,
- GT_MAX_BUY_ORDER_KRW,
GT_MAX_BUYS_PER_LEG,
GT_MAX_ROUND_TRIPS,
TRADING_FEE_RATE,
@@ -49,6 +48,16 @@ from config import (
from deepcoin.common.indicators import apply_bar_indicators, get_trend
from deepcoin.data.mtf_bb import load_frames_from_db
+from deepcoin.ground_truth.gt_allocation import (
+ allocate_order_amounts_chronological,
+ resolve_sell_qty as _resolve_sell_qty,
+)
+from deepcoin.ground_truth.gt_model import (
+ compute_entry_weights,
+ leg_entry_weights,
+ leg_exit_weights,
+ sell_split_weights,
+)
from deepcoin.paths import resolve_ground_truth_file
DEFAULT_OUTPUT = resolve_ground_truth_file()
@@ -476,15 +485,6 @@ def _row_at_ts(df: pd.DataFrame, ts: pd.Timestamp) -> pd.Series:
return row
-def _normalize_weights(scores: list[float]) -> list[float]:
- """비중 점수를 합 1로 정규화."""
- total = sum(scores)
- if total <= 0:
- n = len(scores)
- return [1.0 / n] * n if n else []
- return [s / total for s in scores]
-
-
def _collect_buy_troughs(
df: pd.DataFrame,
buy_pivots: list[Pivot],
@@ -525,7 +525,6 @@ def _collect_buy_troughs(
filtered.append(p)
if len(filtered) > max_buys:
- # 가격이 낮은(저점) 순으로 max_buys만 유지 후 시간순
filtered.sort(key=lambda x: x.price)
filtered = sorted(filtered[:max_buys], key=lambda x: x.ts)
return filtered
@@ -570,7 +569,8 @@ def _peak_sell_points(
second = max(sub_peaks, key=lambda x: x.price)
if second.ts == main.ts:
return [(main, 1.0)]
- return [(main, 0.65), (second, 0.35)]
+ w = sell_split_weights(2)
+ return [(main, w[0]), (second, w[1])]
def build_split_buy_peak_sell_trades(
@@ -603,8 +603,11 @@ def build_split_buy_peak_sell_trades(
for leg_id, peak in enumerate(sell_peaks):
troughs = _collect_buy_troughs(df, buy_pivots, prev_sell_ts, peak.ts, buy_min_bars)
if troughs:
- scores = [1.0 / max(t.price, 1e-9) for t in troughs]
- weights = _normalize_weights(scores)
+ prices = [
+ float(_row_at_ts(df, t.ts)["Low"]) if "Low" in _row_at_ts(df, t.ts) else t.price
+ for t in troughs
+ ]
+ weights = leg_entry_weights(prices)
for t, w in zip(troughs, weights):
row = _row_at_ts(df, t.ts)
bb_pos, rsi, disp = _bb_context(row)
@@ -672,7 +675,11 @@ def build_split_buy_peak_sell_trades(
)
leg_id = len(sell_peaks)
if troughs:
- weights = _normalize_weights([1.0 / max(t.price, 1e-9) for t in troughs])
+ prices = [
+ float(_row_at_ts(df, t.ts)["Low"]) if "Low" in _row_at_ts(df, t.ts) else t.price
+ for t in troughs
+ ]
+ weights = leg_entry_weights(prices)
leg_buys: list[TradePoint] = []
for t, w in zip(troughs, weights):
row = _row_at_ts(df, t.ts)
@@ -916,7 +923,6 @@ def generate_ground_truth(
else "leg_block"
),
"order_amount_min_krw": GT_MIN_ORDER_KRW,
- "order_amount_max_buy_krw": GT_MAX_BUY_ORDER_KRW,
"buy_pct_large_leg": GT_BUY_PCT_LARGE_LEG,
"buy_pct_small_leg": GT_BUY_PCT_SMALL_LEG,
"large_leg_top_pct": GT_LARGE_LEG_TOP_PCT,
@@ -963,13 +969,12 @@ def allocate_gt_order_amounts(
trades: list[dict[str, Any]],
initial_cash: float = GT_INITIAL_CASH_KRW,
min_order_krw: float = GT_MIN_ORDER_KRW,
- max_buy_krw: float = GT_MAX_BUY_ORDER_KRW,
fee_rate: float = TRADING_FEE_RATE,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
"""
GT 각 타점에 amount_krw를 시각순·총자산·비중(최적 매수율)으로 배분합니다.
- 매수: 총보유자산 × (leg 비중 share × 티어 스케일), 상한=가용 현금.
+ 매수: 목표=총보유자산×(leg 비중 share×티어 스케일), 체결=min(목표, 보유현금/(1+fee)).
leg 상위 GT_LARGE_LEG_TOP_PCT는 GT_BUY_PCT_LARGE_LEG, 그 외는 GT_BUY_PCT_SMALL_LEG.
매도 후 현금 증가분은 다음 매수부터 자동 반영(시각순 복리).
@@ -977,159 +982,18 @@ def allocate_gt_order_amounts(
trades: trade dict 리스트(시각순 정렬 전).
initial_cash: 초기 현금.
min_order_krw: 매수·매도 최소 원화 금액.
- max_buy_krw: 매수 1회 상한(가용 현금·비중 배분 후 캡).
fee_rate: 수수료율.
Returns:
(동일 dict 참조, amount_krw 채움), alloc_stats 요약.
"""
- from deepcoin.matching.position_sizing import (
- compute_buy_amount_krw,
- leg_asset_pct_scale,
- top_leg_ids_by_forward_return,
+ return allocate_order_amounts_chronological(
+ trades,
+ initial_cash=initial_cash,
+ min_order_krw=min_order_krw,
+ fee_rate=fee_rate,
)
- chron = sorted(trades, key=lambda x: x["dt"])
- large_legs = top_leg_ids_by_forward_return(chron)
- leg_buy_idxs: dict[int, list[int]] = {}
- leg_sell_idxs: dict[int, list[int]] = {}
- for i, t in enumerate(chron):
- lid = int(t.get("leg_id", 0))
- if t["action"] == "buy":
- leg_buy_idxs.setdefault(lid, []).append(i)
- elif t["action"] == "sell":
- leg_sell_idxs.setdefault(lid, []).append(i)
-
- cash = float(initial_cash)
- qty = 0.0
- qty_by_leg: dict[int, float] = {}
- sell_leg: int | None = None
- sell_base_qty = 0.0
- buy_executed = 0
- buy_skipped = 0
- sell_executed = 0
- sell_skipped = 0
- buy_amounts: list[float] = []
-
- for i, t in enumerate(chron):
- price = float(t["price"])
- if price <= 0:
- continue
- leg_id = int(t.get("leg_id", 0))
- weight = float(t.get("weight", 1.0))
-
- if t["action"] == "buy":
- rem = [j for j in leg_buy_idxs.get(leg_id, []) if j >= i]
- w_sum = sum(float(chron[j].get("weight", 1.0)) for j in rem)
- w_share = (
- weight / w_sum if w_sum > 0 else 1.0 / max(len(rem), 1)
- )
- scale = leg_asset_pct_scale(leg_id, large_legs)
- amount = compute_buy_amount_krw(
- cash,
- qty,
- price,
- weight,
- w_sum,
- asset_pct_scale=scale,
- min_order_krw=min_order_krw,
- fee_rate=fee_rate,
- )
- if amount <= 0:
- t["amount_krw"] = 0
- buy_skipped += 1
- continue
- t["amount_krw"] = amount
- fee = amount * fee_rate
- cash -= amount + fee
- bought_qty = amount / price
- qty += bought_qty
- qty_by_leg[leg_id] = qty_by_leg.get(leg_id, 0.0) + bought_qty
- buy_executed += 1
- buy_amounts.append(amount)
- sell_leg = None
-
- elif t["action"] == "sell":
- leg_qty = qty_by_leg.get(leg_id, 0.0)
- if leg_qty <= 1e-12:
- sell_skipped += 1
- continue
- if sell_leg != leg_id:
- sell_leg = leg_id
- sell_base_qty = leg_qty
- rem_sells = [j for j in leg_sell_idxs.get(leg_id, []) if j >= i]
- is_last_leg_sell = bool(rem_sells) and i == rem_sells[-1]
- if is_last_leg_sell:
- sell_qty = leg_qty
- gross = sell_qty * price
- else:
- gross = sell_base_qty * weight * price
- if gross >= min_order_krw:
- gross = max(min_order_krw, gross)
- gross = min(gross, leg_qty * price)
- if gross <= 0:
- sell_skipped += 1
- continue
- if not is_last_leg_sell:
- sell_qty = gross / price
- else:
- sell_qty = leg_qty
- t["amount_krw"] = round(gross, 0)
- fee = gross * fee_rate
- cash += gross - fee
- leg_qty -= sell_qty
- qty_by_leg[leg_id] = max(leg_qty, 0.0)
- qty = max(qty - sell_qty, 0.0)
- if qty < 1e-12:
- qty = 0.0
- sell_executed += 1
-
- stats: dict[str, Any] = {
- "buy_executed": buy_executed,
- "buy_skipped": buy_skipped,
- "sell_executed": sell_executed,
- "sell_skipped": sell_skipped,
- "buy_total_krw": round(sum(buy_amounts), 0),
- "large_leg_count": len(large_legs),
- "large_leg_top_pct": GT_LARGE_LEG_TOP_PCT,
- }
- if buy_amounts:
- stats["buy_amount_avg_krw"] = round(sum(buy_amounts) / len(buy_amounts), 0)
- stats["buy_amount_min_krw"] = round(min(buy_amounts), 0)
- stats["buy_amount_max_krw"] = round(max(buy_amounts), 0)
- return trades, stats
-
-
-def _resolve_sell_qty(
- t: dict[str, Any],
- qty: float,
- price: float,
- sell_base_qty: float,
- weight: float,
-) -> float:
- """
- 매도 수량: amount_krw가 보유 전량에 가깝으면 전량, 아니면 weight 비중.
-
- Args:
- t: trade dict.
- qty: 현재 보유 수량.
- price: 체결가.
- sell_base_qty: leg 첫 매도 시점 보유량.
- weight: 매도 비중.
-
- Returns:
- 매도 수량.
- """
- if qty <= 0 or price <= 0:
- return 0.0
- ak = t.get("amount_krw")
- if ak is not None and float(ak) > 0:
- gross_cap = float(ak)
- if gross_cap >= qty * price * 0.999:
- return qty
- return min(qty, gross_cap / price)
- return min(sell_base_qty * weight, qty)
-
def _trade_buy_amount(
t: dict[str, Any],
diff --git a/deepcoin/ground_truth/gt_allocation.py b/deepcoin/ground_truth/gt_allocation.py
new file mode 100644
index 0000000..6f61cee
--- /dev/null
+++ b/deepcoin/ground_truth/gt_allocation.py
@@ -0,0 +1,402 @@
+"""
+GT 공통 자본 배분·포트폴리오 시뮬 엔진.
+
+ground_truth.allocate_gt_order_amounts · simulate_truth_portfolio ·
+matching/portfolio_sim 이 동일 규칙을 공유합니다.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Callable
+
+from config import (
+ GT_INITIAL_CASH_KRW,
+ GT_MIN_ORDER_KRW,
+ TRADING_FEE_RATE,
+)
+from deepcoin.ground_truth.gt_model import remaining_weight_sum
+
+
+def resolve_sell_qty(
+ t: dict[str, Any],
+ qty: float,
+ price: float,
+ sell_base_qty: float,
+ weight: float,
+) -> float:
+ """
+ 매도 수량: amount_krw 우선, 없으면 sell_base_qty × weight.
+
+ Args:
+ t: trade dict.
+ qty: 현재 보유 수량.
+ price: 체결가.
+ sell_base_qty: leg 첫 매도 시점 보유량.
+ weight: 매도 비중.
+
+ Returns:
+ 매도 수량.
+ """
+ if qty <= 0 or price <= 0:
+ return 0.0
+ ak = t.get("amount_krw")
+ if ak is not None and float(ak) > 0:
+ gross_cap = float(ak)
+ if gross_cap >= qty * price * 0.999:
+ return qty
+ return min(qty, gross_cap / price)
+ return min(sell_base_qty * weight, qty)
+
+
+def allocate_order_amounts_chronological(
+ trades: list[dict[str, Any]],
+ *,
+ initial_cash: float = GT_INITIAL_CASH_KRW,
+ min_order_krw: float = GT_MIN_ORDER_KRW,
+ fee_rate: float = TRADING_FEE_RATE,
+ large_legs: set[int] | None = None,
+ asset_pct_scale_fn: Callable[[dict[str, Any]], float] | None = None,
+ causal_tier: bool = False,
+) -> tuple[list[dict[str, Any]], dict[str, Any]]:
+ """
+ 시각순·leg 비중·티어 스케일로 amount_krw를 배분합니다.
+
+ causal_tier=True: 청산 완료 leg의 realized return 만으로 tier 산정 (인과적).
+
+ Args:
+ trades: trade dict (weight·leg_id·action·price).
+ initial_cash: 초기 현금.
+ min_order_krw: 최소 체결 원화.
+ fee_rate: 수수료율.
+ large_legs: 대형 leg. None이면 GT trades에서 산출(비인과).
+ asset_pct_scale_fn: 매수 trade별 tier scale.
+ causal_tier: 과거 청산 leg 수익률만으로 tier.
+
+ Returns:
+ (amount_krw 채워진 trades, alloc_stats).
+ """
+ from config import GT_LARGE_LEG_TOP_PCT
+
+ from deepcoin.matching.position_sizing import (
+ compute_buy_amount_krw,
+ large_leg_ids_from_past_returns,
+ leg_asset_pct_scale,
+ top_leg_ids_by_forward_return,
+ )
+
+ chron = sorted(trades, key=lambda x: x["dt"])
+ if large_legs is None and not causal_tier:
+ large_legs = top_leg_ids_by_forward_return(chron)
+ elif large_legs is None:
+ large_legs = set()
+
+ leg_buy_idxs: dict[int, list[int]] = {}
+ leg_sell_idxs: dict[int, list[int]] = {}
+ for i, t in enumerate(chron):
+ lid = int(t.get("leg_id", 0))
+ if t["action"] == "buy":
+ leg_buy_idxs.setdefault(lid, []).append(i)
+ elif t["action"] == "sell":
+ leg_sell_idxs.setdefault(lid, []).append(i)
+
+ cash = float(initial_cash)
+ qty = 0.0
+ qty_by_leg: dict[int, float] = {}
+ sell_leg: int | None = None
+ sell_base_qty = 0.0
+ buy_executed = 0
+ buy_skipped = 0
+ sell_executed = 0
+ sell_skipped = 0
+ buy_amounts: list[float] = []
+ completed_leg_ret: dict[int, float] = {}
+ leg_cost_krw: dict[int, float] = {}
+ leg_proceeds_krw: dict[int, float] = {}
+
+ for i, t in enumerate(chron):
+ price = float(t["price"])
+ if price <= 0:
+ continue
+ leg_id = int(t.get("leg_id", 0))
+ weight = float(t.get("weight", 1.0))
+
+ if t["action"] == "buy":
+ w_sum = remaining_weight_sum(chron, leg_id, i)
+ if causal_tier:
+ large_now = large_leg_ids_from_past_returns(
+ completed_leg_ret, GT_LARGE_LEG_TOP_PCT
+ )
+ scale = leg_asset_pct_scale(leg_id, large_now)
+ elif asset_pct_scale_fn is not None:
+ scale = asset_pct_scale_fn(t)
+ else:
+ scale = leg_asset_pct_scale(leg_id, large_legs)
+ amount = compute_buy_amount_krw(
+ cash,
+ qty,
+ price,
+ weight,
+ w_sum,
+ asset_pct_scale=scale,
+ min_order_krw=min_order_krw,
+ fee_rate=fee_rate,
+ )
+ if amount <= 0:
+ t["amount_krw"] = 0
+ buy_skipped += 1
+ continue
+ t["amount_krw"] = amount
+ fee = amount * fee_rate
+ cash -= amount + fee
+ bought_qty = amount / price
+ qty += bought_qty
+ qty_by_leg[leg_id] = qty_by_leg.get(leg_id, 0.0) + bought_qty
+ leg_cost_krw[leg_id] = leg_cost_krw.get(leg_id, 0.0) + amount + fee
+ buy_executed += 1
+ buy_amounts.append(amount)
+ sell_leg = None
+
+ elif t["action"] == "sell":
+ leg_qty = qty_by_leg.get(leg_id, 0.0)
+ if leg_qty <= 1e-12:
+ sell_skipped += 1
+ continue
+ if sell_leg != leg_id:
+ sell_leg = leg_id
+ sell_base_qty = leg_qty
+ rem_sells = [j for j in leg_sell_idxs.get(leg_id, []) if j >= i]
+ is_last_leg_sell = bool(rem_sells) and i == rem_sells[-1]
+ if is_last_leg_sell:
+ sell_qty = leg_qty
+ gross = sell_qty * price
+ else:
+ gross = sell_base_qty * weight * price
+ if gross >= min_order_krw:
+ gross = max(min_order_krw, gross)
+ gross = min(gross, leg_qty * price)
+ if gross <= 0:
+ sell_skipped += 1
+ continue
+ sell_qty = leg_qty if is_last_leg_sell else gross / price
+ t["amount_krw"] = round(gross, 0)
+ fee = gross * fee_rate
+ cash += gross - fee
+ leg_proceeds_krw[leg_id] = leg_proceeds_krw.get(leg_id, 0.0) + (gross - fee)
+ leg_qty -= sell_qty
+ qty_by_leg[leg_id] = max(leg_qty, 0.0)
+ qty = max(qty - sell_qty, 0.0)
+ if qty < 1e-12:
+ qty = 0.0
+ sell_executed += 1
+ if causal_tier and leg_qty <= 1e-12:
+ cost = leg_cost_krw.pop(leg_id, 0.0)
+ proceeds = leg_proceeds_krw.pop(leg_id, 0.0)
+ if cost > 0:
+ completed_leg_ret[leg_id] = (proceeds - cost) / cost * 100.0
+
+ stats: dict[str, Any] = {
+ "buy_executed": buy_executed,
+ "buy_skipped": buy_skipped,
+ "sell_executed": sell_executed,
+ "sell_skipped": sell_skipped,
+ "buy_total_krw": round(sum(buy_amounts), 0),
+ "large_leg_count": len(large_legs),
+ }
+ if buy_amounts:
+ stats["buy_amount_avg_krw"] = round(sum(buy_amounts) / len(buy_amounts), 0)
+ stats["buy_amount_min_krw"] = round(min(buy_amounts), 0)
+ stats["buy_amount_max_krw"] = round(max(buy_amounts), 0)
+ return trades, stats
+
+
+def simulate_portfolio_steps(
+ trades: list[dict[str, Any]],
+ *,
+ initial_cash: float = GT_INITIAL_CASH_KRW,
+ fee_rate: float = TRADING_FEE_RATE,
+ use_amount_krw: bool = True,
+) -> list[dict[str, Any]]:
+ """
+ 체결마다 현금·보유·총평가 스냅샷.
+
+ Args:
+ trades: 시각순 trade dict (amount_krw·weight·leg_id).
+ initial_cash: 시작 현금.
+ fee_rate: 수수료율.
+ use_amount_krw: True면 amount_krw 기준 체결.
+
+ Returns:
+ step dict 리스트.
+ """
+ rows = sorted(trades, key=lambda x: x["dt"])
+ cash = float(initial_cash)
+ qty = 0.0
+ qty_by_leg: dict[int, float] = {}
+ sell_leg: int | None = None
+ sell_base_qty = 0.0
+ leg_budget = 0.0
+ current_leg: int | None = None
+ steps: list[dict[str, Any]] = []
+
+ for t in rows:
+ action = t.get("action", t.get("side", ""))
+ price = float(t["price"])
+ if price <= 0:
+ continue
+ weight = float(t.get("weight", 1.0))
+ leg_id = int(t.get("leg_id", 0))
+
+ if action == "buy":
+ if use_amount_krw and t.get("amount_krw") is not None and float(t["amount_krw"]) > 0:
+ amount = min(float(t["amount_krw"]), max(cash / (1.0 + fee_rate), 0.0))
+ else:
+ if leg_id != current_leg:
+ current_leg = leg_id
+ leg_budget = cash
+ amount = min(leg_budget * weight, max(cash / (1.0 + fee_rate), 0.0))
+ if amount <= 0:
+ continue
+ fee = amount * fee_rate
+ cash -= amount + fee
+ bought = amount / price
+ qty += bought
+ qty_by_leg[leg_id] = qty_by_leg.get(leg_id, 0.0) + bought
+ sell_leg = None
+
+ elif action == "sell" and qty > 0:
+ leg_qty = qty_by_leg.get(leg_id, qty)
+ if sell_leg != leg_id:
+ sell_leg = leg_id
+ sell_base_qty = leg_qty
+ sell_qty = resolve_sell_qty(t, leg_qty, price, sell_base_qty, weight)
+ if sell_qty <= 0:
+ continue
+ gross = sell_qty * price
+ fee = gross * fee_rate
+ cash += gross - fee
+ leg_qty -= sell_qty
+ qty_by_leg[leg_id] = max(leg_qty, 0.0)
+ qty -= sell_qty
+ if qty < 1e-12:
+ qty = 0.0
+
+ steps.append(
+ {
+ "dt": t["dt"],
+ "action": action,
+ "price": price,
+ "weight": weight,
+ "leg_id": leg_id,
+ "amount_krw": t.get("amount_krw"),
+ "cash_krw": round(cash, 0),
+ "holding_qty": round(qty, 6),
+ "total_asset_krw": round(cash + qty * price, 0),
+ }
+ )
+ return steps
+
+
+def compute_drawdown_metrics(steps: list[dict[str, Any]]) -> dict[str, Any]:
+ """
+ equity curve 기준 최대 낙폭·고점 대비 하락.
+
+ Args:
+ steps: simulate_portfolio_steps 결과.
+
+ Returns:
+ max_drawdown_pct, peak_asset_krw, trough_after_peak_krw.
+ """
+ if not steps:
+ return {
+ "max_drawdown_pct": 0.0,
+ "peak_asset_krw": 0.0,
+ "trough_asset_krw": 0.0,
+ }
+ assets = [float(s["total_asset_krw"]) for s in steps]
+ peak = assets[0]
+ max_dd = 0.0
+ peak_at = assets[0]
+ trough_at = assets[0]
+ for a in assets:
+ if a > peak:
+ peak = a
+ dd = (peak - a) / peak * 100.0 if peak > 0 else 0.0
+ if dd > max_dd:
+ max_dd = dd
+ peak_at = peak
+ trough_at = a
+ return {
+ "max_drawdown_pct": round(max_dd, 2),
+ "peak_asset_krw": round(peak_at, 0),
+ "trough_asset_krw": round(trough_at, 0),
+ }
+
+
+def simulate_portfolio_summary(
+ trades: list[dict[str, Any]],
+ *,
+ initial_cash: float = GT_INITIAL_CASH_KRW,
+ fee_rate: float = TRADING_FEE_RATE,
+ last_price: float | None = None,
+ use_amount_krw: bool = True,
+) -> dict[str, Any]:
+ """
+ 포트폴리오 시뮬 요약 + MDD.
+
+ Args:
+ trades: trade dict 리스트.
+ initial_cash: 시작 현금.
+ fee_rate: 수수료율.
+ last_price: 미청산 평가가.
+ use_amount_krw: amount_krw 체결 사용.
+
+ Returns:
+ pnl·fee·MDD 포함 dict.
+ """
+ steps = simulate_portfolio_steps(
+ trades,
+ initial_cash=initial_cash,
+ fee_rate=fee_rate,
+ use_amount_krw=use_amount_krw,
+ )
+ if not steps:
+ return {
+ "initial_cash_krw": round(initial_cash, 0),
+ "final_asset_krw": round(initial_cash, 0),
+ "pnl_krw": 0.0,
+ "pnl_pct": 0.0,
+ "trade_count": len(trades),
+ "max_drawdown_pct": 0.0,
+ }
+
+ last_step = steps[-1]
+ cash = float(last_step["cash_krw"])
+ qty = float(last_step["holding_qty"])
+ mark = float(last_price if last_price is not None else last_step["price"])
+ holding_value = qty * mark
+ final_asset = cash + holding_value
+ pnl = final_asset - initial_cash
+ pnl_pct = pnl / initial_cash * 100.0 if initial_cash else 0.0
+
+ fees = 0.0
+ for t in sorted(trades, key=lambda x: x["dt"]):
+ ak = float(t.get("amount_krw") or 0)
+ if ak <= 0:
+ continue
+ fees += ak * fee_rate
+
+ dd = compute_drawdown_metrics(steps)
+ return {
+ "initial_cash_krw": round(initial_cash, 0),
+ "final_asset_krw": round(final_asset, 0),
+ "pnl_krw": round(pnl, 0),
+ "pnl_pct": round(pnl_pct, 2),
+ "total_fees_krw": round(fees, 0),
+ "cash_krw": round(cash, 0),
+ "holding_qty": round(qty, 6),
+ "holding_value_krw": round(holding_value, 0),
+ "mark_price": round(mark, 2),
+ "fee_rate": fee_rate,
+ "trade_count": len(trades),
+ **dd,
+ }
diff --git a/deepcoin/ground_truth/gt_allocation_analysis.py b/deepcoin/ground_truth/gt_allocation_analysis.py
new file mode 100644
index 0000000..836a527
--- /dev/null
+++ b/deepcoin/ground_truth/gt_allocation_analysis.py
@@ -0,0 +1,150 @@
+"""
+GT 체결 amount_krw·총자산 비율 분석 — 시뮬 tier·배분율 최적 추정.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from config import (
+ GT_BUY_PCT_LARGE_LEG,
+ GT_BUY_PCT_SMALL_LEG,
+ GT_INITIAL_CASH_KRW,
+ GT_LARGE_LEG_TOP_PCT,
+ TRADING_FEE_RATE,
+)
+from deepcoin.matching.position_sizing import (
+ leg_asset_pct_scale,
+ optimal_weight_share,
+ portfolio_totals,
+ top_leg_ids_by_forward_return,
+)
+
+
+def analyze_gt_buy_allocation(
+ trades: list[dict[str, Any]],
+ *,
+ initial_cash: float = GT_INITIAL_CASH_KRW,
+ fee_rate: float = TRADING_FEE_RATE,
+) -> dict[str, Any]:
+ """
+ GT 시각순 체결에서 매수별 (실제투입/총자산) 비율을 분석합니다.
+
+ Args:
+ trades: amount_krw·weight·leg_id가 채워진 GT trade dict.
+ initial_cash: 시작 현금.
+ fee_rate: 수수료율.
+
+ Returns:
+ leg tier별·전체 배분 통계 및 권장 pct_large/pct_small.
+ """
+ chron = sorted(trades, key=lambda x: x["dt"])
+ if not chron:
+ return {"note": "체결 없음"}
+
+ large_legs = top_leg_ids_by_forward_return(chron, GT_LARGE_LEG_TOP_PCT)
+ cash = float(initial_cash)
+ qty = 0.0
+
+ ratios_large: list[float] = []
+ ratios_small: list[float] = []
+ ratios_all: list[float] = []
+
+ for i, t in enumerate(chron):
+ price = float(t["price"])
+ if price <= 0:
+ continue
+ leg_id = int(t.get("leg_id", 0))
+ action = t.get("action", "")
+
+ if action == "buy":
+ w = float(t.get("weight", 1.0))
+ rem = sum(
+ float(chron[j].get("weight", 1.0))
+ for j in range(i, len(chron))
+ if int(chron[j].get("leg_id", 0)) == leg_id
+ and chron[j].get("action") == "buy"
+ )
+ opt = optimal_weight_share(w, rem) if rem > 0 else 1.0
+ total_asset, _, _ = portfolio_totals(cash, qty, price)
+ amount = float(t.get("amount_krw") or 0)
+ if total_asset > 0 and amount > 0 and opt > 0:
+ implied = amount / (total_asset * opt)
+ ratios_all.append(implied)
+ if leg_id in large_legs:
+ ratios_large.append(implied)
+ else:
+ ratios_small.append(implied)
+ if amount > 0:
+ fee = amount * fee_rate
+ cash -= amount + fee
+ qty += amount / price
+ elif action == "sell" and qty > 0:
+ gross = float(t.get("amount_krw") or qty * price)
+ cash += gross * (1.0 - fee_rate)
+ qty = 0.0
+
+ def _stats(vals: list[float]) -> dict[str, float]:
+ if not vals:
+ return {}
+ s = sorted(vals)
+ n = len(s)
+ return {
+ "count": n,
+ "mean": round(sum(s) / n, 4),
+ "median": round(s[n // 2], 4),
+ "p25": round(s[max(0, n // 4)], 4),
+ "p75": round(s[min(n - 1, 3 * n // 4)], 4),
+ }
+
+ st_all = _stats(ratios_all)
+ st_large = _stats(ratios_large)
+ st_small = _stats(ratios_small)
+
+ rec_large = st_large.get("median", GT_BUY_PCT_LARGE_LEG)
+ rec_small = st_small.get("median", GT_BUY_PCT_SMALL_LEG)
+ if not rec_large or rec_large <= 0:
+ rec_large = GT_BUY_PCT_LARGE_LEG
+ if not rec_small or rec_small <= 0:
+ rec_small = GT_BUY_PCT_SMALL_LEG
+
+ return {
+ "large_leg_ids": sorted(large_legs),
+ "large_leg_count": len(large_legs),
+ "config_pct_large": GT_BUY_PCT_LARGE_LEG,
+ "config_pct_small": GT_BUY_PCT_SMALL_LEG,
+ "observed_implied_scale": {
+ "all": st_all,
+ "large_leg": st_large,
+ "small_leg": st_small,
+ },
+ "recommended_pct_large_leg": round(rec_large, 4),
+ "recommended_pct_small_leg": round(rec_small, 4),
+ "note": (
+ "implied_scale = amount / (pre_buy_total_asset × weight_share); "
+ "시뮬 tier는 GT 분석 median 사용"
+ ),
+ }
+
+
+def gt_tier_scale_from_analysis(
+ leg_id: int,
+ large_legs: set[int],
+ analysis: dict[str, Any] | None = None,
+) -> float:
+ """
+ GT 분석 권장값 또는 config tier scale.
+
+ Args:
+ leg_id: leg 번호.
+ large_legs: 상위 leg.
+ analysis: analyze_gt_buy_allocation 결과.
+
+ Returns:
+ 총자산 대비 매수 스케일 (0~1).
+ """
+ if analysis and analysis.get("observed_implied_scale", {}).get("all"):
+ if leg_id in large_legs:
+ return float(analysis.get("recommended_pct_large_leg", GT_BUY_PCT_LARGE_LEG))
+ return float(analysis.get("recommended_pct_small_leg", GT_BUY_PCT_SMALL_LEG))
+ return leg_asset_pct_scale(leg_id, large_legs)
diff --git a/deepcoin/ground_truth/gt_model.py b/deepcoin/ground_truth/gt_model.py
index 99edf47..8451a6f 100644
--- a/deepcoin/ground_truth/gt_model.py
+++ b/deepcoin/ground_truth/gt_model.py
@@ -1,13 +1,14 @@
"""
Ground Truth 타점·비중·자본 배분 모델 (일반화 명세).
-타점 생성(ground_truth.py)과 자본 배분(position_sizing.py)의 공통 언어.
+타점 생성(ground_truth.py), 자본 배분(gt_allocation.py),
+시뮬(position_sizing.py)의 공통 언어.
"""
from __future__ import annotations
from dataclasses import dataclass, field
-from typing import Any
+from typing import Any, Callable
from config import (
GT_BUY_BB_MAX,
@@ -15,14 +16,130 @@ from config import (
GT_BUY_MIN_SWING_PCT,
GT_BUY_PCT_LARGE_LEG,
GT_BUY_PCT_SMALL_LEG,
+ GT_BUY_WEIGHT_RULE,
GT_LARGE_LEG_TOP_PCT,
GT_MAX_BUYS_PER_LEG,
GT_MAX_SELLS_PER_LEG,
GT_MIN_ORDER_KRW,
GT_SELL_SPLIT_GAP_PCT,
+ GT_SELL_SPLIT_WEIGHTS,
GT_SELECTION_MODE,
)
+# --- 매수 비중 규칙 (확장 가능) ---
+
+EntryWeightFn = Callable[[list[float]], list[float]]
+
+
+def normalize_weights(scores: list[float]) -> list[float]:
+ """
+ 비중 점수를 합 1로 정규화합니다.
+
+ Args:
+ scores: raw score 리스트.
+
+ Returns:
+ 정규화 weight (합 ≈ 1).
+ """
+ if not scores:
+ return []
+ total = sum(scores)
+ if total <= 0:
+ n = len(scores)
+ return [1.0 / n] * n
+ return [s / total for s in scores]
+
+
+def compute_buy_weights_inverse_price(prices: list[float]) -> list[float]:
+ """
+ 저점 매수 비중: score_i = 1/price_i → 합=1 정규화.
+
+ Args:
+ prices: leg 내 매수 후보 가격.
+
+ Returns:
+ weight 리스트 (합 ≈ 1).
+ """
+ if not prices:
+ return []
+ scores = [1.0 / max(p, 1e-9) for p in prices]
+ return normalize_weights(scores)
+
+
+def compute_buy_weights_equal(_prices: list[float]) -> list[float]:
+ """
+ 균등 분할 매수 비중.
+
+ Args:
+ _prices: leg 내 매수 후보 가격(미사용).
+
+ Returns:
+ weight 리스트 (합 = 1).
+ """
+ n = len(_prices)
+ if n <= 0:
+ return []
+ return [1.0 / n] * n
+
+
+ENTRY_WEIGHT_RULES: dict[str, EntryWeightFn] = {
+ "inverse_price_normalized": compute_buy_weights_inverse_price,
+ "equal": compute_buy_weights_equal,
+}
+
+
+def compute_entry_weights(
+ prices: list[float],
+ rule: str | None = None,
+) -> list[float]:
+ """
+ 매수 타점 비중을 규칙명으로 계산합니다.
+
+ Args:
+ prices: 체결 가격 리스트.
+ rule: `inverse_price_normalized` | `equal`. None이면 config.
+
+ Returns:
+ leg 내 매수 weight (합 ≈ 1).
+ """
+ key = (rule or GT_BUY_WEIGHT_RULE).strip()
+ fn = ENTRY_WEIGHT_RULES.get(key, compute_buy_weights_inverse_price)
+ return fn(prices)
+
+
+def leg_entry_weights(
+ prices: list[float],
+ rule: str | None = None,
+) -> list[float]:
+ """
+ leg 내 매수 타점 비중 (compute_entry_weights 별칭).
+
+ Args:
+ prices: 체결 가격 리스트.
+ rule: 비중 규칙 키.
+
+ Returns:
+ weight 리스트 (합 ≈ 1).
+ """
+ return compute_entry_weights(prices, rule)
+
+
+def leg_exit_weights(
+ n_sells: int,
+ exit_spec: GtExitSpec | None = None,
+) -> list[float]:
+ """
+ leg 매도 분할 비중.
+
+ Args:
+ n_sells: 매도 횟수.
+ exit_spec: 매도 명세.
+
+ Returns:
+ weight 리스트 (합 ≈ 1).
+ """
+ return sell_split_weights(n_sells, exit_spec)
+
@dataclass(frozen=True)
class GtEntrySpec:
@@ -32,14 +149,14 @@ class GtEntrySpec:
Attributes:
pivot_kind: ZigZag 저점(trough).
price_field: 체결가 = 봉 Low.
- weight_rule: 저가일수록 큰 비중 (1/price 정규화).
+ weight_rule: 매수 비중 규칙 키.
max_per_leg: leg당 최대 매수 횟수.
min_bars_gap: 분할 매수 최소 봉 간격.
"""
pivot_kind: str = "trough"
price_field: str = "Low"
- weight_rule: str = "inverse_price_normalized"
+ weight_rule: str = GT_BUY_WEIGHT_RULE
max_per_leg: int = GT_MAX_BUYS_PER_LEG
min_bars_gap: int = GT_BUY_MIN_BARS
bb_filter: str = f"bb_pos <= {GT_BUY_BB_MAX}"
@@ -53,13 +170,13 @@ class GtExitSpec:
Attributes:
pivot_kind: major swing 고점(peak).
price_field: 체결가 = 봉 High.
- split_weights: 2회 분할 시 (65%, 35%).
+ split_weights: N회 분할 시 각 매도 비중 (합=1).
split_gap_pct: 2차 고점 인정 최소 괴리(%).
"""
pivot_kind: str = "peak"
price_field: str = "High"
- split_weights: tuple[float, float] = (0.65, 0.35)
+ split_weights: tuple[float, ...] = GT_SELL_SPLIT_WEIGHTS
split_gap_pct: float = GT_SELL_SPLIT_GAP_PCT
max_per_leg: int = GT_MAX_SELLS_PER_LEG
@@ -78,13 +195,16 @@ class GtCapitalSpec:
min_order_krw: 최소 체결 원화.
"""
- buy_formula: str = "min(total_asset * w_share * tier_scale, cash/(1+fee))"
+ buy_formula: str = (
+ "target = total_asset * (weight/remaining_weights) * tier_scale; "
+ "amount = min(target, available_cash/(1+fee))"
+ )
optimal_buy_rate: str = "weight / sum(remaining_buy_weights_in_leg)"
large_leg_top_pct: float = GT_LARGE_LEG_TOP_PCT
pct_large: float = GT_BUY_PCT_LARGE_LEG
pct_small: float = GT_BUY_PCT_SMALL_LEG
min_order_krw: float = float(GT_MIN_ORDER_KRW)
- sell_formula: str = "leg_qty * sell_weight * price (last sell = full leg_qty)"
+ sell_formula: str = "sell_base_qty * sell_weight * price (last sell = full leg_qty)"
@dataclass
@@ -121,37 +241,74 @@ def default_model() -> GroundTruthModel:
return GroundTruthModel()
-def compute_buy_weights_inverse_price(prices: list[float]) -> list[float]:
+def sell_split_weights(
+ n_sells: int,
+ exit_spec: GtExitSpec | None = None,
+) -> list[float]:
"""
- 저점 매수 비중: score_i = 1/price_i → 합=1 정규화.
+ leg 매도 비중 (1회=100%, N회=split_weights 정규화).
Args:
- prices: leg 내 매수 후보 가격.
+ n_sells: 매도 횟수(1 이상).
+ exit_spec: None이면 default.
Returns:
weight 리스트 (합 ≈ 1).
"""
- if not prices:
- return []
- scores = [1.0 / max(p, 1e-9) for p in prices]
- s = sum(scores)
- return [x / s for x in scores] if s > 0 else [1.0 / len(prices)] * len(prices)
+ spec = exit_spec or GtExitSpec()
+ if n_sells <= 1:
+ return [1.0]
+ weights = list(spec.split_weights[:n_sells])
+ if len(weights) < n_sells:
+ weights.extend([weights[-1]] * (n_sells - len(weights)))
+ return normalize_weights(weights)
-def sell_split_weights(n_sells: int) -> list[float]:
+def pair_peak_sell_weights(
+ n_peaks: int,
+ exit_spec: GtExitSpec | None = None,
+) -> list[tuple[float, float]]:
"""
- leg 매도 비중.
+ 고점 피벗 (피벗, weight) 쌍 — 1회 또는 분할.
Args:
- n_sells: 매도 횟수(1 또는 2).
+ n_peaks: 인정된 고점 수 (1 또는 2+).
+ exit_spec: 매도 명세.
Returns:
- weight 리스트.
+ (weight,) 또는 (w1, w2) 리스트. 호출측에서 피벗과 zip.
"""
- spec = GtExitSpec()
- if n_sells >= 2:
- return list(spec.split_weights)
- return [1.0]
+ if n_peaks <= 1:
+ return [(1.0,)]
+ w = sell_split_weights(2, exit_spec)
+ return [(w[0],), (w[1],)]
+
+
+def remaining_weight_sum(
+ trades: list[dict[str, Any]],
+ leg_id: int,
+ from_index: int,
+) -> float:
+ """
+ leg 내 from_index 이후 남은 매수 weight 합.
+
+ Args:
+ trades: 시각순 trade dict.
+ leg_id: leg 번호.
+ from_index: chron 리스트 인덱스.
+
+ Returns:
+ 남은 weight 합.
+ """
+ total = 0.0
+ for j, t in enumerate(trades):
+ if j < from_index:
+ continue
+ if int(t.get("leg_id", 0)) != leg_id:
+ continue
+ if t.get("action") == "buy":
+ total += float(t.get("weight", 1.0))
+ return total
def model_to_dict(model: GroundTruthModel | None = None) -> dict[str, Any]:
@@ -165,6 +322,12 @@ def model_to_dict(model: GroundTruthModel | None = None) -> dict[str, Any]:
직렬화 dict.
"""
m = model or default_model()
+ w_rule = m.entry.weight_rule
+ w_formula = (
+ "w_i = (1/price_i) / sum(1/price_j)"
+ if w_rule == "inverse_price_normalized"
+ else "w_i = 1 / n"
+ )
return {
"selection_mode": m.selection_mode,
"leg_definition": m.leg_definition,
@@ -172,7 +335,7 @@ def model_to_dict(model: GroundTruthModel | None = None) -> dict[str, Any]:
"pivot": m.entry.pivot_kind,
"price": m.entry.price_field,
"weight_rule": m.entry.weight_rule,
- "weight_formula": "w_i = (1/price_i) / sum(1/price_j)",
+ "weight_formula": w_formula,
"max_buys_per_leg": m.entry.max_per_leg,
"min_bars_between_buys": m.entry.min_bars_gap,
"bb_filter": m.entry.bb_filter,
@@ -181,7 +344,7 @@ def model_to_dict(model: GroundTruthModel | None = None) -> dict[str, Any]:
"pivot": m.exit.pivot_kind,
"price": m.exit.price_field,
"weight_rule": "fixed_split_or_full",
- "weights_two_sell": list(m.exit.split_weights),
+ "weights_split": list(m.exit.split_weights),
"split_gap_pct": m.exit.split_gap_pct,
"max_sells_per_leg": m.exit.max_per_leg,
},
@@ -209,7 +372,7 @@ def summarize_leg_weights(trades: list[dict[str, Any]]) -> dict[str, Any]:
trades: GT trade dict.
Returns:
- leg_id → {buy_sum, sell_sum, n_buy, n_sell}.
+ leg_id → {buy_sum, sell_sum, n_buy, n_sell, valid}.
"""
legs: dict[int, dict[str, Any]] = {}
for t in trades:
@@ -225,4 +388,30 @@ def summarize_leg_weights(trades: list[dict[str, Any]]) -> dict[str, Any]:
elif t.get("action") == "sell":
legs[lid]["sell_sum"] += w
legs[lid]["n_sell"] += 1
+ for lid, info in legs.items():
+ buy_ok = abs(info["buy_sum"] - 1.0) < 0.02 or info["n_buy"] == 0
+ sell_ok = abs(info["sell_sum"] - 1.0) < 0.02 or info["n_sell"] == 0
+ info["valid"] = buy_ok and sell_ok
+ info["buy_sum"] = round(info["buy_sum"], 4)
+ info["sell_sum"] = round(info["sell_sum"], 4)
return legs
+
+
+def weight_policy_summary(model: GroundTruthModel | None = None) -> dict[str, Any]:
+ """
+ 시뮬·리포트용 비중 정책 요약.
+
+ Args:
+ model: GT 모델.
+
+ Returns:
+ entry/exit/capital 요약 dict.
+ """
+ m = model or default_model()
+ return {
+ "entry_weight_rule": m.entry.weight_rule,
+ "exit_split_weights": list(m.exit.split_weights),
+ "capital_large_pct": m.capital.pct_large,
+ "capital_small_pct": m.capital.pct_small,
+ "large_leg_top_pct": m.capital.large_leg_top_pct,
+ }
diff --git a/deepcoin/ground_truth/gt_signal_causal.py b/deepcoin/ground_truth/gt_signal_causal.py
new file mode 100644
index 0000000..bdef526
--- /dev/null
+++ b/deepcoin/ground_truth/gt_signal_causal.py
@@ -0,0 +1,181 @@
+"""
+인과적(미래 미사용) GT 스타일 신호 — t봉 시점에 t 이하 데이터만 사용.
+
+ZigZag/국소극값: pivot bar i-order 는 bar i 에서 확정 (i-order..i 구간만 관측).
+"""
+
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+
+from config import (
+ GT_BUY_BB_MAX,
+ GT_BUY_MIN_SWING_PCT,
+ GT_MIN_SWING_PCT,
+ GT_PIVOT_ORDER,
+)
+
+
+def _confirmed_trough_mask(
+ low: np.ndarray,
+ order: int,
+) -> np.ndarray:
+ """
+ bar i 에서 i-order 봉이 저점임을 확정 (low[i-order:i+1] 만 사용).
+
+ Args:
+ low: Low 가격 배열.
+ order: pivot 반경(봉).
+
+ Returns:
+ 길이 n, i 에 1이면 i 시점 매수 확인 신호.
+ """
+ n = len(low)
+ out = np.zeros(n, dtype=np.int8)
+ for i in range(2 * order, n):
+ p = i - order
+ seg = low[p - order : i + 1]
+ if len(seg) == 0:
+ continue
+ if low[p] <= seg.min() + 1e-12:
+ out[i] = 1
+ return out
+
+
+def _confirmed_peak_mask(
+ high: np.ndarray,
+ order: int,
+) -> np.ndarray:
+ """
+ bar i 에서 i-order 봉이 고점임을 확정.
+
+ Args:
+ high: High 가격 배열.
+ order: pivot 반경.
+
+ Returns:
+ i 시점 매도 확인 신호.
+ """
+ n = len(high)
+ out = np.zeros(n, dtype=np.int8)
+ for i in range(2 * order, n):
+ p = i - order
+ seg = high[p - order : i + 1]
+ if len(seg) == 0:
+ continue
+ if high[p] >= seg.max() - 1e-12:
+ out[i] = 1
+ return out
+
+
+def _zigzag_filter_causal(
+ confirm: np.ndarray,
+ prices: np.ndarray,
+ min_swing_pct: float,
+ kind: str,
+) -> np.ndarray:
+ """
+ 확정 피벗에 ZigZag 최소 스윙% 필터 (인과적, 순차 갱신).
+
+ Args:
+ confirm: bar i 에 확정 플래그.
+ prices: pivot 가격 (i-order 위치의 low/high).
+ pivot_indices: confirm==1 인 bar index.
+ min_swing_pct: 최소 스윙 %.
+ kind: trough | peak.
+
+ Returns:
+ zigzag 통과 시점에 1.
+ """
+ n = len(confirm)
+ out = np.zeros(n, dtype=np.int8)
+ order = GT_PIVOT_ORDER
+ last_kind: str | None = None
+ last_price = 0.0
+ min_ratio = min_swing_pct / 100.0
+
+ for i in range(n):
+ if confirm[i] != 1:
+ continue
+ p = i - order
+ if p < 0:
+ continue
+ price = float(prices[p])
+ if last_kind is None:
+ out[i] = 1
+ last_kind = kind
+ last_price = price
+ continue
+ if kind == last_kind:
+ if kind == "trough" and price < last_price:
+ out[i - 1] = 0
+ out[i] = 1
+ last_price = price
+ elif kind == "peak" and price > last_price:
+ out[i - 1] = 0
+ out[i] = 1
+ last_price = price
+ continue
+ move = abs(price - last_price) / max(last_price, 1e-9)
+ if move >= min_ratio:
+ out[i] = 1
+ last_kind = kind
+ last_price = price
+ return out
+
+
+def enrich_scan_frame_gt_signals_causal(
+ frame: pd.DataFrame,
+ *,
+ pivot_order: int = GT_PIVOT_ORDER,
+ buy_swing_pct: float = GT_BUY_MIN_SWING_PCT,
+ sell_swing_pct: float = GT_MIN_SWING_PCT,
+ bb_max: float = GT_BUY_BB_MAX,
+) -> pd.DataFrame:
+ """
+ 인과적 GT 신호 컬럼 (gt_*). t 시점 신호는 데이터 index<=t 만 사용.
+
+ Args:
+ frame: m3 스캔 프레임.
+ pivot_order: 확정 지연(봉).
+ buy_swing_pct: 매수 ZigZag 스윙%.
+ sell_swing_pct: 매도 ZigZag 스윙%.
+ bb_max: BB 하단 필터.
+
+ Returns:
+ gt_* 컬럼 추가 DataFrame.
+ """
+ out = frame.copy()
+ if "Low" not in out.columns or "High" not in out.columns:
+ return out
+
+ low = out["Low"].astype(float).values
+ high = out["High"].astype(float).values
+ n = len(low)
+
+ trough_conf = _confirmed_trough_mask(low, pivot_order)
+ peak_conf = _confirmed_peak_mask(high, pivot_order)
+
+ trough_z = _zigzag_filter_causal(
+ trough_conf, low, buy_swing_pct, "trough"
+ )
+ peak_z = _zigzag_filter_causal(
+ peak_conf, high, sell_swing_pct, "peak"
+ )
+
+ out["gt_trough_local"] = trough_conf
+ out["gt_peak_local"] = peak_conf
+ out["gt_trough_zigzag"] = trough_z
+ out["gt_peak_zigzag"] = peak_z
+
+ bb_ok = pd.Series(True, index=out.index)
+ if "bb_pos" in out.columns:
+ bb = pd.to_numeric(out["bb_pos"], errors="coerce")
+ bb_ok = bb <= bb_max
+
+ out["gt_buy_signal"] = (pd.Series(trough_z, index=out.index) == 1) & bb_ok
+ out["gt_buy_signal"] = out["gt_buy_signal"].astype(int)
+ out["gt_sell_signal"] = pd.Series(peak_z, index=out.index).astype(int)
+ out["gt_signal_causal"] = 1
+ return out
diff --git a/deepcoin/ground_truth/gt_signal_rules.py b/deepcoin/ground_truth/gt_signal_rules.py
new file mode 100644
index 0000000..4962319
--- /dev/null
+++ b/deepcoin/ground_truth/gt_signal_rules.py
@@ -0,0 +1,197 @@
+"""
+GT 모델(entry/exit)을 규칙 스캔·발화 형식으로 일반화.
+
+ZigZag trough/peak + BB 필터 등 GT 타점 생성 로직과 동일 파라미터를
+rule_eval 스캔 프레임 컬럼(gt_*)으로 노출합니다.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import numpy as np
+import pandas as pd
+
+from config import (
+ GT_BUY_BB_MAX,
+ GT_BUY_MIN_SWING_PCT,
+ GT_MIN_SWING_PCT,
+ GT_PIVOT_ORDER,
+ MATCH_PRIMARY_INTERVAL,
+)
+from deepcoin.ground_truth.ground_truth import build_zigzag_pivots
+
+
+def _local_extrema_mask(
+ series: pd.Series,
+ order: int,
+ kind: str,
+) -> pd.Series:
+ """
+ 국소 극값 boolean 마스크.
+
+ Args:
+ series: 가격 시리즈.
+ order: 좌우 봉 수.
+ kind: min | max.
+
+ Returns:
+ boolean Series (index=series.index).
+ """
+ arr = series.astype(float).values
+ n = len(arr)
+ out = np.zeros(n, dtype=bool)
+ if n < 2 * order + 1:
+ return pd.Series(out, index=series.index)
+ for i in range(order, n - order):
+ window = arr[i - order : i + order + 1]
+ if kind == "min" and arr[i] <= window.min():
+ out[i] = True
+ elif kind == "max" and arr[i] >= window.max():
+ out[i] = True
+ return pd.Series(out, index=series.index)
+
+
+def enrich_scan_frame_gt_signals(
+ frame: pd.DataFrame,
+ *,
+ pivot_order: int = GT_PIVOT_ORDER,
+ buy_swing_pct: float = GT_BUY_MIN_SWING_PCT,
+ sell_swing_pct: float = GT_MIN_SWING_PCT,
+ bb_max: float = GT_BUY_BB_MAX,
+ causal: bool | None = None,
+) -> pd.DataFrame:
+ """
+ 스캔 프레임에 GT 모델 신호 컬럼을 추가합니다.
+
+ GT_SIGNAL_CAUSAL=1 이면 t 시점까지 데이터만 사용 (운영 정합).
+
+ Args:
+ frame: m3 스캔 프레임 (Low, High, bb_pos).
+ pivot_order: 피벗 반경.
+ buy_swing_pct: 매수 ZigZag 스윙%.
+ sell_swing_pct: 매도 ZigZag 스윙%.
+ bb_max: BB 하단 필터.
+ causal: None이면 config GT_SIGNAL_CAUSAL.
+
+ Returns:
+ gt_* 컬럼이 추가된 DataFrame.
+ """
+ from config import GT_SIGNAL_CAUSAL
+
+ use_causal = GT_SIGNAL_CAUSAL if causal is None else causal
+ if use_causal:
+ from deepcoin.ground_truth.gt_signal_causal import (
+ enrich_scan_frame_gt_signals_causal,
+ )
+
+ return enrich_scan_frame_gt_signals_causal(
+ frame,
+ pivot_order=pivot_order,
+ buy_swing_pct=buy_swing_pct,
+ sell_swing_pct=sell_swing_pct,
+ bb_max=bb_max,
+ )
+
+ out = frame.copy()
+ if "Low" not in out.columns or "High" not in out.columns:
+ return out
+
+ low = out["Low"].astype(float)
+ high = out["High"].astype(float)
+ out["gt_trough_local"] = _local_extrema_mask(low, pivot_order, "min").astype(int)
+ out["gt_peak_local"] = _local_extrema_mask(high, pivot_order, "max").astype(int)
+
+ df_ohlc = out[["Low", "High"]].copy()
+ if "close" in out.columns:
+ df_ohlc["close"] = out["close"]
+ df_ohlc.index = out.index
+
+ buy_pivots = build_zigzag_pivots(
+ df_ohlc,
+ min_swing_pct=buy_swing_pct,
+ pivot_order=pivot_order,
+ )
+ sell_pivots = build_zigzag_pivots(
+ df_ohlc,
+ min_swing_pct=sell_swing_pct,
+ pivot_order=pivot_order,
+ )
+
+ trough_z = pd.Series(0, index=out.index, dtype=int)
+ for p in buy_pivots:
+ if p.kind == "trough" and p.ts in trough_z.index:
+ trough_z.loc[p.ts] = 1
+ peak_z = pd.Series(0, index=out.index, dtype=int)
+ for p in sell_pivots:
+ if p.kind == "peak" and p.ts in peak_z.index:
+ peak_z.loc[p.ts] = 1
+
+ out["gt_trough_zigzag"] = trough_z
+ out["gt_peak_zigzag"] = peak_z
+
+ bb_ok = pd.Series(True, index=out.index)
+ if "bb_pos" in out.columns:
+ bb = pd.to_numeric(out["bb_pos"], errors="coerce")
+ bb_ok = bb <= bb_max
+
+ out["gt_buy_signal"] = ((out["gt_trough_zigzag"] == 1) & bb_ok).astype(int)
+ out["gt_sell_signal"] = (out["gt_peak_zigzag"] == 1).astype(int)
+ return out
+
+
+def build_gt_model_rules() -> list[dict[str, Any]]:
+ """
+ GT entry/exit 명세와 동일한 스캔 규칙 후보.
+
+ Returns:
+ rule dict 리스트 (buy 2종 + sell 2종).
+ """
+ return [
+ {
+ "rule_id": "gt_model_buy_zigzag_bb",
+ "side": "buy",
+ "kind": "gt_model",
+ "logic": "and",
+ "conditions": [
+ {"col": "gt_buy_signal", "op": "eq_int", "value": 1},
+ ],
+ "gt_spec": "trough_zigzag + bb_pos <= GT_BUY_BB_MAX",
+ },
+ {
+ "rule_id": "gt_model_buy_trough_local",
+ "side": "buy",
+ "kind": "gt_model",
+ "logic": "and",
+ "conditions": [
+ {"col": "gt_trough_local", "op": "eq_int", "value": 1},
+ {"col": "bb_pos", "op": "lte", "value": GT_BUY_BB_MAX},
+ ],
+ "gt_spec": "local trough + bb filter",
+ },
+ {
+ "rule_id": "gt_model_sell_zigzag_peak",
+ "side": "sell",
+ "kind": "gt_model",
+ "logic": "and",
+ "conditions": [
+ {"col": "gt_sell_signal", "op": "eq_int", "value": 1},
+ ],
+ "gt_spec": "major swing peak (ZigZag)",
+ },
+ {
+ "rule_id": "gt_model_sell_peak_local",
+ "side": "sell",
+ "kind": "gt_model",
+ "logic": "and",
+ "conditions": [
+ {"col": "gt_peak_local", "op": "eq_int", "value": 1},
+ ],
+ "gt_spec": "local high extremum",
+ },
+ ]
+
+
+def gt_signal_rule_ids() -> set[str]:
+ """GT 일반화 규칙 ID 집합."""
+ return {r["rule_id"] for r in build_gt_model_rules()}
diff --git a/deepcoin/matching/portfolio_sim.py b/deepcoin/matching/portfolio_sim.py
index 240c1a1..332c4b3 100644
--- a/deepcoin/matching/portfolio_sim.py
+++ b/deepcoin/matching/portfolio_sim.py
@@ -1,5 +1,5 @@
"""
-규칙 발화 기반 고정 금액 체결 포트폴리오 시뮬 (GT HTML 카드·테이블용).
+규칙 발화 기반 GT 모델 복리 포트폴리오 시뮬.
"""
from __future__ import annotations
@@ -10,14 +10,13 @@ import pandas as pd
from config import (
GT_INITIAL_CASH_KRW,
- LIVE_DAILY_KRW_MAX,
- LIVE_MAX_TRADES_PER_DAY,
+ GT_SIGNAL_CAUSAL,
LIVE_ORDER_KRW,
TRADING_FEE_RATE,
)
+from deepcoin.ground_truth.gt_allocation import simulate_portfolio_summary
from deepcoin.matching.position_sizing import (
- attach_dynamic_buy_amounts,
- load_sizing_context_from_gt,
+ attach_gt_model_amounts,
)
@@ -43,99 +42,95 @@ def _planned_order_krw(
return float(order_krw)
+def sort_fires_chronological(fires: pd.DataFrame) -> pd.DataFrame:
+ """
+ 발화를 시간순 정렬 (일·금액 한도 없음).
+
+ Args:
+ fires: fire_outcomes.
+
+ Returns:
+ 정렬된 DataFrame.
+ """
+ if fires.empty:
+ return fires
+ return fires.sort_values("dt").copy()
+
+
+def simulate_fires_compound(
+ fires: pd.DataFrame,
+ *,
+ initial_cash: float = GT_INITIAL_CASH_KRW,
+ fee_rate: float = TRADING_FEE_RATE,
+) -> tuple[list[dict[str, Any]], dict[str, Any]]:
+ """
+ 발화 → GT tier 복리 amount_krw 배분 (allocate_order_amounts_chronological).
+
+ Args:
+ fires: fire_outcomes.
+ initial_cash: 시작 현금 (이후 체결마다 누적).
+ fee_rate: 수수료율.
+
+ Returns:
+ (amount_krw 채워진 trade dict, stats).
+ """
+ trades = fires_to_trade_list(
+ fires,
+ apply_dynamic_sizing=True,
+ initial_cash=initial_cash,
+ fee_rate=fee_rate,
+ )
+ n_in = int(len(fires))
+ n_out = sum(1 for t in trades if float(t.get("amount_krw") or 0) > 0)
+ return trades, {
+ "input_fires": n_in,
+ "executed": n_out,
+ "skipped": max(n_in - n_out, 0),
+ }
+
+
def select_capped_fires(
fires: pd.DataFrame,
*,
use_dynamic_sizing: bool = True,
) -> pd.DataFrame:
"""
- 일한도·회수 제한으로 체결 가능한 발화만 남깁니다.
+ 시각순 발화 반환 (레거시명; 일·금액 한도 미적용).
Args:
- fires: fire_outcomes (dt, side, close, rule_id …).
+ fires: fire_outcomes.
+ use_dynamic_sizing: 미사용 (하위 호환).
Returns:
- 체결된 발화 DataFrame.
+ 정렬된 발화 DataFrame.
"""
- if fires.empty:
- return fires
- gt_trades, large_legs, approved = load_sizing_context_from_gt()
- df = fires.sort_values("dt").copy()
- df["ts"] = pd.to_datetime(df["dt"])
- df["day"] = df["ts"].dt.date.astype(str)
- cash = float(GT_INITIAL_CASH_KRW)
- qty = 0.0
- taken: list[pd.DataFrame] = []
- for _, day_grp in df.groupby("day", sort=True):
- spent = 0.0
- n_trades = 0
- idxs: list[Any] = []
- for idx, row in day_grp.iterrows():
- if n_trades >= LIVE_MAX_TRADES_PER_DAY:
- break
- side = row["side"]
- price = float(row["close"])
- if side == "buy" and use_dynamic_sizing:
- from deepcoin.matching.position_sizing import (
- compute_buy_amount_krw,
- live_buy_asset_pct_scale,
- )
-
- scale = live_buy_asset_pct_scale(
- str(row["rule_id"]),
- str(row["dt"]),
- gt_trades,
- approved_rules=approved,
- large_legs=large_legs,
- )
- planned = compute_buy_amount_krw(
- cash,
- qty,
- price,
- 1.0,
- 1.0,
- asset_pct_scale=scale,
- )
- else:
- planned = float(LIVE_ORDER_KRW)
- if side == "buy":
- if spent + planned > LIVE_DAILY_KRW_MAX:
- break
- if planned <= 0:
- continue
- fee = planned * TRADING_FEE_RATE
- cash -= planned + fee
- qty += planned / price if price > 0 else 0.0
- spent += planned
- elif side == "sell" and qty > 0:
- gross = qty * price
- cash += gross * (1.0 - TRADING_FEE_RATE)
- qty = 0.0
- n_trades += 1
- idxs.append(idx)
- if idxs:
- taken.append(day_grp.loc[idxs])
- if not taken:
- return df.iloc[0:0]
- return pd.concat(taken, ignore_index=True)
+ _ = use_dynamic_sizing
+ return sort_fires_chronological(fires)
def fires_to_trade_list(
fires: pd.DataFrame,
*,
apply_dynamic_sizing: bool = True,
+ initial_cash: float = GT_INITIAL_CASH_KRW,
+ fee_rate: float = TRADING_FEE_RATE,
) -> list[dict[str, Any]]:
"""
- 발화 DataFrame을 포트폴리오 시뮬용 trade dict 리스트로 변환.
+ 발화 → GT 모델 amount_krw가 채워진 trade dict (복리 배분).
Args:
fires: 체결 대상 발화.
+ apply_dynamic_sizing: True면 GT tier 복리 배분.
+ initial_cash: 시작 현금 (누적 복리).
+ fee_rate: 수수료율.
Returns:
- dt, action, price 키를 가진 dict 리스트.
+ dt, action, price, amount_krw 키 dict 리스트.
"""
+ if fires.empty:
+ return []
rows: list[dict[str, Any]] = []
- for _, r in fires.sort_values("dt").iterrows():
+ for _, r in sort_fires_chronological(fires).iterrows():
rows.append(
{
"dt": str(r["dt"]),
@@ -145,13 +140,11 @@ def fires_to_trade_list(
"forward_ret_pct": float(r.get("forward_ret_pct", 0)),
}
)
- if apply_dynamic_sizing and rows:
- gt_trades, large_legs, approved = load_sizing_context_from_gt()
- attach_dynamic_buy_amounts(
+ if apply_dynamic_sizing:
+ attach_gt_model_amounts(
rows,
- gt_trades=gt_trades,
- approved_rules=approved,
- large_legs=large_legs,
+ initial_cash=initial_cash,
+ fee_rate=fee_rate,
)
return rows
@@ -164,7 +157,7 @@ def simulate_sized_portfolio(
fallback_order_krw: float = LIVE_ORDER_KRW,
) -> dict[str, Any]:
"""
- trade.amount_krw(총자산 비율 배분) 기준 포트폴리오 시뮬.
+ trade.amount_krw(GT 모델·복리 배분) 기준 포트폴리오 시뮬 + MDD.
Args:
trades: 시간순 trade dict (amount_krw 권장).
@@ -174,16 +167,25 @@ def simulate_sized_portfolio(
fallback_order_krw: amount_krw 없을 때 1회 금액.
Returns:
- simulate_truth_portfolio와 동일 키 구조.
+ simulate_truth_portfolio와 동일 키 + max_drawdown_pct.
"""
- return simulate_fixed_order_portfolio(
+ if trades and not any(float(t.get("amount_krw") or 0) > 0 for t in trades):
+ attach_gt_model_amounts(trades, initial_cash=initial_cash, fee_rate=fee_rate)
+ result = simulate_portfolio_summary(
trades,
- order_krw=fallback_order_krw,
initial_cash=initial_cash,
fee_rate=fee_rate,
last_price=last_price,
- sizing_mode="amount_krw",
+ use_amount_krw=True,
)
+ result["sizing_mode"] = (
+ "gt_model_compound_causal" if GT_SIGNAL_CAUSAL else "gt_model_compound"
+ )
+ result["sizing_note"] = (
+ "전기간 복리·GT tier·총자산×비중, 보유현금 한도; "
+ + ("인과적 신호·tier(미래 미사용)" if GT_SIGNAL_CAUSAL else "상한 없음")
+ )
+ return result
def simulate_fixed_order_portfolio(
@@ -203,7 +205,7 @@ def simulate_fixed_order_portfolio(
initial_cash: 시작 현금.
fee_rate: 수수료율.
last_price: 미청산 평가 종가.
- sizing_mode: 'fixed' | 'amount_krw' (없으면 order_krw).
+ sizing_mode: 'fixed' | 'amount_krw'.
Returns:
simulate_truth_portfolio와 동일 키 구조.
diff --git a/deepcoin/matching/position_sizing.py b/deepcoin/matching/position_sizing.py
index d263a61..5159b26 100644
--- a/deepcoin/matching/position_sizing.py
+++ b/deepcoin/matching/position_sizing.py
@@ -1,5 +1,5 @@
"""
-총자산 대비 최적 매수율(비중) · 현금 한도 · leg 상위·EV/WF 통과 대형 매수.
+총자산 대비 GT 모델 매수율(비중) · 보유 현금 한도 · leg tier 배분.
"""
from __future__ import annotations
@@ -15,19 +15,14 @@ from config import (
GT_INITIAL_CASH_KRW,
GT_LARGE_LEG_TOP_PCT,
GT_MIN_ORDER_KRW,
- LIVE_BUY_PCT_LARGE,
- LIVE_BUY_PCT_SMALL,
MATCH_GT_TOLERANCE_MIN,
- SIM_FEE_STRESS_MULT,
- SIM_GO_MIN_HOLDOUT_EV,
- SIM_GO_MIN_HOLDOUT_PF,
- SIM_GO_WF_POSITIVE_RATIO,
- SIM_WALK_FORWARD_MIN_MONTHS,
TRADING_FEE_RATE,
)
from deepcoin.matching.load_rules import load_matched_rules
from deepcoin.paths import MATCHING_FIRE_OUTCOMES, MATCHING_MATCHED_RULES
+_GT_ALLOC_ANALYSIS_CACHE: dict[str, Any] | None = None
+
def portfolio_totals(
cash: float,
@@ -78,10 +73,12 @@ def compute_buy_amount_krw(
fee_rate: float = TRADING_FEE_RATE,
) -> float:
"""
- 총자산 × (최적 매수율 × scale)을 목표로, 가용 현금을 넘지 않게 매수 원화를 산출합니다.
+ 목표=총보유자산×(최적 매수율×scale), 체결=min(목표, 보유현금/(1+fee)) 로 매수 원화를 산출합니다.
+
+ 보유 현금 = 총보유자산 − 코인평가액(cash 인자).
Args:
- cash: 현금.
+ cash: 보유 현금(가용 원화).
qty: 보유 수량.
price: 체결가.
weight: 타점 비중.
@@ -95,8 +92,8 @@ def compute_buy_amount_krw(
"""
if price <= 0:
return 0.0
- total_asset, _, _ = portfolio_totals(cash, qty, price)
- budget = max(cash / (1.0 + fee_rate), 0.0)
+ total_asset, _, available_cash = portfolio_totals(cash, qty, price)
+ budget = max(available_cash / (1.0 + fee_rate), 0.0)
opt_rate = optimal_weight_share(weight, weight_sum_remaining) * asset_pct_scale
target = total_asset * opt_rate
amount = min(target, budget)
@@ -105,6 +102,27 @@ def compute_buy_amount_krw(
return round(max(amount, 0.0), 0)
+def large_leg_ids_from_past_returns(
+ leg_returns: dict[int, float],
+ top_pct: float = GT_LARGE_LEG_TOP_PCT,
+) -> set[int]:
+ """
+ 이미 청산된 leg의 realized return 상위 n% (인과적 tier).
+
+ Args:
+ leg_returns: leg_id → realized return %.
+ top_pct: 상위 비율.
+
+ Returns:
+ large leg id set.
+ """
+ if not leg_returns:
+ return set()
+ ranked = sorted(leg_returns.items(), key=lambda x: x[1], reverse=True)
+ n = max(1, int(len(ranked) * top_pct + 0.999999))
+ return {lid for lid, _ in ranked[:n]}
+
+
def top_leg_ids_by_forward_return(
trades: list[dict[str, Any]],
top_pct: float = GT_LARGE_LEG_TOP_PCT,
@@ -215,6 +233,8 @@ def load_ev_wf_approved_rule_ids(
if _APPROVED_RULES_CACHE is not None:
return set(_APPROVED_RULES_CACHE)
+ from config import SIM_FEE_STRESS_MULT
+
from deepcoin.matching.select_rules import _rule_metrics, _split_train_valid_holdout
from deepcoin.matching.simulation import (
evaluate_go_no_go,
@@ -258,6 +278,72 @@ def load_ev_wf_approved_rule_ids(
return fallback
+def load_gt_allocation_analysis(
+ gt_trades: list[dict[str, Any]] | None = None,
+) -> dict[str, Any]:
+ """
+ GT amount_krw 분석 캐시 (tier 권장 pct).
+
+ Args:
+ gt_trades: GT trades. None이면 파일 로드.
+
+ Returns:
+ analyze_gt_buy_allocation 결과.
+ """
+ global _GT_ALLOC_ANALYSIS_CACHE
+ if _GT_ALLOC_ANALYSIS_CACHE is not None:
+ return _GT_ALLOC_ANALYSIS_CACHE
+ from deepcoin.ground_truth.gt_allocation_analysis import analyze_gt_buy_allocation
+ from deepcoin.paths import resolve_ground_truth_file
+
+ trades = gt_trades
+ if trades is None:
+ p = resolve_ground_truth_file()
+ if p.is_file():
+ trades = json.loads(p.read_text(encoding="utf-8")).get("trades") or []
+ if not trades:
+ _GT_ALLOC_ANALYSIS_CACHE = {}
+ return _GT_ALLOC_ANALYSIS_CACHE
+ chron = sorted(trades, key=lambda x: x["dt"])
+ if not any(float(t.get("amount_krw") or 0) > 0 for t in chron):
+ from deepcoin.ground_truth.ground_truth import allocate_gt_order_amounts
+
+ allocate_gt_order_amounts(chron)
+ _GT_ALLOC_ANALYSIS_CACHE = analyze_gt_buy_allocation(chron)
+ return _GT_ALLOC_ANALYSIS_CACHE
+
+
+def gt_tier_scale_for_trade(
+ trade: dict[str, Any],
+ gt_trades: list[dict[str, Any]],
+ large_legs: set[int],
+ *,
+ analysis: dict[str, Any] | None = None,
+) -> float:
+ """
+ GT leg tier 배분 스케일 (분석 권장값 또는 config).
+
+ 시뮬은 live_buy_asset_pct_scale 대신 GT와 동일 tier 정책을 사용합니다.
+
+ Args:
+ trade: {dt, leg_id?, action, ...}.
+ gt_trades: GT trades (leg 매칭).
+ large_legs: 상위 leg.
+ analysis: analyze_gt_buy_allocation 결과.
+
+ Returns:
+ pct_large 또는 pct_small.
+ """
+ from deepcoin.ground_truth.gt_allocation_analysis import gt_tier_scale_from_analysis
+
+ lid = trade.get("leg_id")
+ if lid is None:
+ lid = nearest_gt_leg_id(str(trade["dt"]), gt_trades)
+ if lid is None:
+ return float(GT_BUY_PCT_SMALL_LEG)
+ return gt_tier_scale_from_analysis(int(lid), large_legs, analysis)
+
+
def live_buy_asset_pct_scale(
rule_id: str,
dt: str,
@@ -267,7 +353,7 @@ def live_buy_asset_pct_scale(
large_legs: set[int],
) -> float:
"""
- 실거래·시뮬 매수: EV/WF 통과 규칙 + leg 상위만 대형 비율.
+ 실거래 전용 매수 tier (EV/WF·leg 상위). 시뮬은 gt_tier_scale_for_trade 사용.
Args:
rule_id: 규칙 ID.
@@ -279,6 +365,8 @@ def live_buy_asset_pct_scale(
Returns:
LIVE_BUY_PCT_LARGE 또는 LIVE_BUY_PCT_SMALL(또는 0에 가까운 소형).
"""
+ from config import LIVE_BUY_PCT_LARGE, LIVE_BUY_PCT_SMALL
+
if rule_id not in approved_rules:
return float(LIVE_BUY_PCT_SMALL)
lid = nearest_gt_leg_id(dt, gt_trades)
@@ -287,10 +375,193 @@ def live_buy_asset_pct_scale(
return float(LIVE_BUY_PCT_SMALL)
+def enrich_sim_trades_with_gt_weights(
+ trades: list[dict[str, Any]],
+ gt_trades: list[dict[str, Any]],
+ *,
+ causal_legs: bool = False,
+) -> list[dict[str, Any]]:
+ """
+ 규칙 발화에 GT leg_id·매수/매도 weight를 부여합니다.
+
+ causal_legs=True: GT leg 매칭 없이 매수~매도 구간 순번 leg_id (인과적).
+
+ Args:
+ trades: {dt, action/side, price, rule_id}.
+ gt_trades: GT trades (leg 매칭, causal_legs=False 일 때).
+ causal_legs: 순차 leg_id.
+
+ Returns:
+ leg_id·weight가 채워진 trade dict.
+ """
+ from deepcoin.ground_truth.gt_model import leg_entry_weights, leg_exit_weights
+
+ rows = sorted(trades, key=lambda x: x["dt"])
+ pos = 0
+ seq_leg = 0
+ while pos < len(rows):
+ action = rows[pos].get("action", rows[pos].get("side", ""))
+ if action != "buy":
+ if causal_legs:
+ rows[pos]["leg_id"] = seq_leg
+ elif "leg_id" not in rows[pos]:
+ rows[pos]["leg_id"] = nearest_gt_leg_id(rows[pos]["dt"], gt_trades) or 0
+ rows[pos]["weight"] = float(rows[pos].get("weight", 1.0))
+ pos += 1
+ continue
+ buy_end = pos
+ while buy_end < len(rows):
+ a = rows[buy_end].get("action", rows[buy_end].get("side", ""))
+ if a != "buy":
+ break
+ buy_end += 1
+ buy_slice = rows[pos:buy_end]
+ sell_slice: list[dict[str, Any]] = []
+ sell_end = buy_end
+ while sell_end < len(rows):
+ a = rows[sell_end].get("action", rows[sell_end].get("side", ""))
+ if a == "buy":
+ break
+ if a == "sell":
+ sell_slice.append(rows[sell_end])
+ sell_end += 1
+
+ if causal_legs:
+ leg_id = seq_leg
+ else:
+ leg_id = nearest_gt_leg_id(buy_slice[0]["dt"], gt_trades) or 0
+ prices = [float(t["price"]) for t in buy_slice]
+ buy_weights = leg_entry_weights(prices)
+ for t, w in zip(buy_slice, buy_weights):
+ t["leg_id"] = leg_id
+ t["weight"] = round(w, 4)
+ if "action" not in t and "side" in t:
+ t["action"] = t["side"]
+
+ if sell_slice:
+ sw = leg_exit_weights(len(sell_slice))
+ for t, w in zip(sell_slice, sw):
+ t["leg_id"] = leg_id
+ t["weight"] = round(w, 4)
+ if "action" not in t and "side" in t:
+ t["action"] = t["side"]
+ if causal_legs and sell_slice:
+ seq_leg += 1
+ pos = sell_end if sell_slice else buy_end
+ return rows
+
+
+def attach_gt_model_amounts(
+ trades: list[dict[str, Any]],
+ *,
+ gt_trades: list[dict[str, Any]] | None = None,
+ approved_rules: set[str] | None = None,
+ large_legs: set[int] | None = None,
+ initial_cash: float = GT_INITIAL_CASH_KRW,
+ fee_rate: float = TRADING_FEE_RATE,
+) -> list[dict[str, Any]]:
+ """
+ GT 모델 비중 + 공통 배분 엔진으로 amount_krw를 채웁니다.
+
+ 시뮬·매칭 전용: leg·tier 모두 인과적(과거 청산 leg 수익만). GT 정답 배분은
+ ground_truth.allocate_gt_order_amounts 를 사용하세요.
+
+ Args:
+ trades: enrich_sim_trades_with_gt_weights 출력 또는 raw fires.
+ gt_trades: GT trades. None이면 파일 로드.
+ approved_rules: EV/WF 통과 rule (live scale용).
+ large_legs: 상위 leg.
+ initial_cash: 초기 현금.
+ fee_rate: 수수료율.
+
+ Returns:
+ amount_krw·weight·leg_id가 채워진 trade dict.
+ """
+ from deepcoin.ground_truth.gt_allocation import allocate_order_amounts_chronological
+
+ if gt_trades is None:
+ gt_trades, _, _ = load_sizing_context_from_gt()
+
+ enriched = enrich_sim_trades_with_gt_weights(
+ list(trades),
+ gt_trades,
+ causal_legs=True,
+ )
+
+ allocate_order_amounts_chronological(
+ enriched,
+ initial_cash=initial_cash,
+ fee_rate=fee_rate,
+ large_legs=None,
+ asset_pct_scale_fn=None,
+ causal_tier=True,
+ )
+ return enriched
+
+
+def plan_open_position_buy(
+ open_buys: list[dict[str, Any]],
+ candidate: dict[str, Any],
+ cash: float,
+ qty: float,
+ gt_trades: list[dict[str, Any]] | None = None,
+ *,
+ large_legs: set[int],
+ analysis: dict[str, Any] | None = None,
+ fee_rate: float = TRADING_FEE_RATE,
+) -> float:
+ """
+ 미청산 포지션 내 다음 매수 원화 (GT tier·보유 현금 한도, 1회 상한 없음).
+
+ Args:
+ open_buys: 현재 포지션에서 이미 체결된 매수 dict.
+ candidate: 이번 매수 후보 {dt, price, rule_id, leg_id?, ...}.
+ cash: 보유 현금.
+ qty: 보유 수량.
+ gt_trades: GT leg 매칭용.
+ large_legs: 상위 leg.
+ analysis: GT 배분 분석.
+ fee_rate: 수수료율.
+
+ Returns:
+ 매수 계획 원화.
+ """
+ from deepcoin.ground_truth.gt_model import leg_entry_weights
+
+ if gt_trades is None:
+ gt_trades, _, _ = load_sizing_context_from_gt()
+ if analysis is None:
+ analysis = load_gt_allocation_analysis(gt_trades)
+
+ prices = [float(t["price"]) for t in open_buys] + [float(candidate["price"])]
+ weights = leg_entry_weights(prices)
+ idx = len(open_buys)
+ w = weights[idx]
+ w_sum = sum(weights[idx:])
+ cand = dict(candidate)
+ if "leg_id" not in cand:
+ cand["leg_id"] = nearest_gt_leg_id(str(candidate["dt"]), gt_trades)
+ scale = gt_tier_scale_for_trade(
+ cand,
+ gt_trades,
+ large_legs,
+ analysis=analysis,
+ )
+ return compute_buy_amount_krw(
+ cash,
+ qty,
+ float(candidate["price"]),
+ w,
+ w_sum,
+ asset_pct_scale=scale,
+ fee_rate=fee_rate,
+ )
+
+
def attach_dynamic_buy_amounts(
trades: list[dict[str, Any]],
*,
- gt_trades: list[dict[str, Any]],
+ gt_trades: list[dict[str, Any]] | None = None,
approved_rules: set[str] | None = None,
large_legs: set[int] | None = None,
initial_cash: float = GT_INITIAL_CASH_KRW,
@@ -298,60 +569,18 @@ def attach_dynamic_buy_amounts(
fee_rate: float = TRADING_FEE_RATE,
) -> list[dict[str, Any]]:
"""
- 시뮬 발화 trade dict에 amount_krw(총자산 비율·현금 한도)를 채웁니다.
+ 시뮬 발화 trade dict에 amount_krw(GT 모델·보유 현금 한도)를 채웁니다.
- Args:
- trades: 시간순 {dt, action, price, rule_id, …}.
- gt_trades: GT leg 매칭용.
- approved_rules: EV/WF 통과 rule. None이면 전 규칙 대형 허용 안 함.
- large_legs: 상위 leg. None이면 GT에서 산출.
- initial_cash: 초기 현금.
- default_weight: 매수 weight 기본값.
- fee_rate: 수수료율.
-
- Returns:
- amount_krw가 채워진 동일 리스트.
+ attach_gt_model_amounts 별칭.
"""
- if large_legs is None:
- large_legs = top_leg_ids_by_forward_return(gt_trades)
- if approved_rules is None:
- approved_rules = set()
-
- cash = float(initial_cash)
- qty = 0.0
- for t in sorted(trades, key=lambda x: x["dt"]):
- action = t.get("action", t.get("side", ""))
- price = float(t["price"])
- if price <= 0:
- continue
- if action == "buy":
- rid = str(t.get("rule_id", ""))
- scale = live_buy_asset_pct_scale(
- rid, t["dt"], gt_trades,
- approved_rules=approved_rules,
- large_legs=large_legs,
- )
- amount = compute_buy_amount_krw(
- cash,
- qty,
- price,
- float(t.get("weight", default_weight)),
- float(t.get("weight", default_weight)),
- asset_pct_scale=scale,
- fee_rate=fee_rate,
- )
- t["amount_krw"] = amount
- if amount > 0:
- fee = amount * fee_rate
- cash -= amount + fee
- qty += amount / price
- elif action == "sell" and qty > 1e-12:
- gross = qty * price
- t["amount_krw"] = round(gross, 0)
- fee = gross * fee_rate
- cash += gross - fee
- qty = 0.0
- return trades
+ return attach_gt_model_amounts(
+ trades,
+ gt_trades=gt_trades,
+ approved_rules=approved_rules,
+ large_legs=large_legs,
+ initial_cash=initial_cash,
+ fee_rate=fee_rate,
+ )
def load_sizing_context_from_gt(
@@ -374,5 +603,4 @@ def load_sizing_context_from_gt(
data = json.loads(p.read_text(encoding="utf-8"))
trades = data.get("trades") or []
large = top_leg_ids_by_forward_return(trades)
- approved = load_ev_wf_approved_rule_ids()
- return trades, large, approved
+ return trades, large, set()
diff --git a/deepcoin/matching/profile_rules.py b/deepcoin/matching/profile_rules.py
index ba5a978..78f38ef 100644
--- a/deepcoin/matching/profile_rules.py
+++ b/deepcoin/matching/profile_rules.py
@@ -380,6 +380,15 @@ def build_rule_candidates(
seen_ids.add(cr["rule_id"])
print(f"[04-1] 캘리브레이션 규칙 병합 → 총 {len(rules)}개")
+ from deepcoin.ground_truth.gt_signal_rules import build_gt_model_rules
+
+ seen_ids = {r["rule_id"] for r in rules}
+ for gr in build_gt_model_rules():
+ if gr["rule_id"] not in seen_ids:
+ rules.append(gr)
+ seen_ids.add(gr["rule_id"])
+ print(f"[04-1] GT 모델 일반화 규칙 추가 → 총 {len(rules)}개")
+
out = {
"source": str(path),
"profile_json": str(ANALYSIS_GT_MTF_PROFILE_JSON),
diff --git a/deepcoin/matching/rule_eval.py b/deepcoin/matching/rule_eval.py
index 9b03014..cc1fc2e 100644
--- a/deepcoin/matching/rule_eval.py
+++ b/deepcoin/matching/rule_eval.py
@@ -188,6 +188,22 @@ def build_mtf_scan_frame(
if align_needed:
out = _add_align_columns_vectorized(out)
+ gt_needed = [c for c in needed_cols if c.startswith("gt_")]
+ bb_in_rules = "bb_pos" in needed_cols
+ if gt_needed or bb_in_rules:
+ ef = enriched[primary]
+ for src in ("Low", "High", "low", "high", "bb_pos", "Open", "Volume"):
+ if src in ef.columns and src not in out.columns:
+ out[src] = ef[src].reindex(out.index)
+ if "Low" not in out.columns and "low" in out.columns:
+ out["Low"] = out["low"]
+ if "High" not in out.columns and "high" in out.columns:
+ out["High"] = out["high"]
+ from deepcoin.ground_truth.gt_signal_rules import enrich_scan_frame_gt_signals
+
+ # 시뮬·live 스캔: 타점 판단은 항상 인과적 (GT 정답 생성은 ground_truth.py 별도)
+ out = enrich_scan_frame_gt_signals(out, causal=True)
+
out = out.loc[:, ~out.columns.duplicated()]
out = out.dropna(subset=["close"])
print(f"[04b] 스캔 프레임: {len(out):,}봉 × {len(out.columns)}열")
diff --git a/deepcoin/matching/simulation.py b/deepcoin/matching/simulation.py
index bc2d428..30558c4 100644
--- a/deepcoin/matching/simulation.py
+++ b/deepcoin/matching/simulation.py
@@ -13,8 +13,6 @@ import pandas as pd
from config import (
GT_INITIAL_CASH_KRW,
- LIVE_DAILY_KRW_MAX,
- LIVE_MAX_TRADES_PER_DAY,
LIVE_ORDER_KRW,
LIVE_SLIPPAGE_PCT,
MATCH_HOLDOUT_RATIO,
@@ -32,13 +30,19 @@ from config import (
from deepcoin.ground_truth.ground_truth import (
load_ground_truth,
order_trades_chronological,
- simulate_truth_portfolio,
+)
+from deepcoin.ground_truth.gt_allocation import simulate_portfolio_summary
+from deepcoin.ground_truth.gt_model import (
+ default_model,
+ model_to_dict,
+ summarize_leg_weights,
+ weight_policy_summary,
)
from deepcoin.matching.portfolio_sim import (
fires_to_trade_list,
- select_capped_fires,
simulate_fixed_order_portfolio,
simulate_sized_portfolio,
+ sort_fires_chronological,
)
from deepcoin.matching.select_rules import _rule_metrics, _split_train_valid_holdout
from deepcoin.paths import resolve_ground_truth_file
@@ -129,7 +133,7 @@ def simulate_live_order_cap(
holdout_only: bool = True,
) -> dict[str, Any]:
"""
- 1회·일 한도·슬리피지 가정으로 체결 가능한 발화만 집계.
+ GT 복리 배분·슬리피지 가정으로 체결 가능한 발화 집계 (일·금액 한도 없음).
Args:
outcomes: fire_outcomes (split 컬럼 있으면 holdout 필터 가능).
@@ -142,71 +146,23 @@ def simulate_live_order_cap(
if outcomes.empty:
return {"rules": {}, "note": "발화 없음"}
- df = outcomes
+ df = outcomes.copy()
if holdout_only and "split" in df.columns:
df = df[df["split"] == "holdout"]
if rule_ids is not None:
df = df[df["rule_id"].isin(rule_ids)]
- df = df.sort_values("dt").copy()
- df["ts"] = pd.to_datetime(df["dt"])
- df["day"] = df["ts"].dt.date.astype(str)
slip = LIVE_SLIPPAGE_PCT
- taken_rows: list[pd.DataFrame] = []
- from deepcoin.matching.position_sizing import (
- compute_buy_amount_krw,
- live_buy_asset_pct_scale,
- load_sizing_context_from_gt,
- )
+ trades = fires_to_trade_list(sort_fires_chronological(df), apply_dynamic_sizing=True)
+ executed_dts = {
+ t["dt"]
+ for t in trades
+ if t.get("action") == "sell" or float(t.get("amount_krw") or 0) > 0
+ }
+ if not executed_dts:
+ return {"rules": {}, "taken_count": 0, "total_count": int(len(df))}
- gt_trades, large_legs, approved = load_sizing_context_from_gt()
- cash = float(GT_INITIAL_CASH_KRW)
- qty = 0.0
-
- for day, day_grp in df.groupby("day", sort=True):
- spent = 0.0
- n_trades = 0
- taken_idx: list[int] = []
- for idx, row in day_grp.iterrows():
- if n_trades >= LIVE_MAX_TRADES_PER_DAY:
- break
- side = row["side"]
- price = float(row["close"])
- if side == "buy":
- scale = live_buy_asset_pct_scale(
- str(row["rule_id"]),
- str(row["dt"]),
- gt_trades,
- approved_rules=approved,
- large_legs=large_legs,
- )
- planned = compute_buy_amount_krw(
- cash, qty, price, 1.0, 1.0, asset_pct_scale=scale
- )
- else:
- planned = float(LIVE_ORDER_KRW)
- if side == "buy":
- if planned <= 0:
- continue
- if spent + planned > LIVE_DAILY_KRW_MAX:
- break
- fee = planned * TRADING_FEE_RATE
- cash -= planned + fee
- qty += planned / price if price > 0 else 0.0
- spent += planned
- elif side == "sell" and qty > 0:
- gross = qty * price
- cash += gross * (1.0 - TRADING_FEE_RATE)
- qty = 0.0
- n_trades += 1
- taken_idx.append(idx)
- if taken_idx:
- taken_rows.append(day_grp.loc[taken_idx])
-
- if not taken_rows:
- return {"rules": {}, "taken_count": 0}
-
- taken = pd.concat(taken_rows, ignore_index=True)
+ taken = df[df["dt"].astype(str).isin(executed_dts)].copy()
taken["adj_ret_pct"] = taken["forward_ret_pct"] - slip
by_rule: dict[str, Any] = {}
@@ -221,10 +177,8 @@ def simulate_live_order_cap(
return {
"assumptions": {
- "order_krw": LIVE_ORDER_KRW,
- "daily_krw_max": LIVE_DAILY_KRW_MAX,
"slippage_pct": slip,
- "sizing": "total_asset_pct_ev_wf_large_leg",
+ "sizing": "gt_model_compound_no_daily_cap",
},
"taken_count": int(len(taken)),
"total_count": int(len(df)),
@@ -338,45 +292,126 @@ def build_simulation_report(
gt_data = load_ground_truth(resolve_ground_truth_file()) or {}
gt_trades = gt_data.get("trades") or []
mark = (gt_data.get("summary") or {}).get("mark_price")
- if gt_trades:
- portfolio_compare["ground_truth_chrono"] = simulate_truth_portfolio(
- order_trades_chronological(gt_trades),
+ gt_chrono = order_trades_chronological(gt_trades) if gt_trades else []
+
+ from deepcoin.ground_truth.gt_signal_rules import gt_signal_rule_ids
+ from config import GT_SIGNAL_CAUSAL, SIM_CAUSAL_TIER
+ from deepcoin.matching.position_sizing import load_gt_allocation_analysis
+
+ gt_alloc_analysis = load_gt_allocation_analysis(gt_trades) if gt_trades else {}
+
+ if gt_chrono:
+ if not any(float(t.get("amount_krw") or 0) > 0 for t in gt_chrono):
+ from deepcoin.ground_truth.ground_truth import allocate_gt_order_amounts
+
+ allocate_gt_order_amounts(gt_chrono)
+ portfolio_compare["ground_truth_chrono"] = simulate_portfolio_summary(
+ gt_chrono,
last_price=float(mark) if mark else None,
+ use_amount_krw=True,
)
- holdout = outcomes[
- outcomes["rule_id"].isin(monitor_ids) & (outcomes["split"] == "holdout")
- ]
- capped = select_capped_fires(holdout)
- if not capped.empty:
+
+ # 전기간 monitor 규칙 — 100만원에서 복리 (holdout만 X)
+ all_monitor = outcomes[outcomes["rule_id"].isin(monitor_ids)]
+ if not all_monitor.empty:
+ sim_trades_full = fires_to_trade_list(sort_fires_chronological(all_monitor))
portfolio_compare["sim_sized"] = simulate_sized_portfolio(
- fires_to_trade_list(capped, apply_dynamic_sizing=True),
+ sim_trades_full,
last_price=float(mark) if mark else None,
)
portfolio_compare["sim_fixed_order"] = simulate_fixed_order_portfolio(
- fires_to_trade_list(capped, apply_dynamic_sizing=False),
+ fires_to_trade_list(all_monitor, apply_dynamic_sizing=False),
last_price=float(mark) if mark else None,
)
+ # GT 모델 일반화 규칙 (ZigZag+BB 매수 / ZigZag 고점 매도)
+ gt_buy_rule = "gt_model_buy_zigzag_bb"
+ gt_sell_rule = "gt_model_sell_zigzag_peak"
+ gt_pair_ids = {gt_buy_rule, gt_sell_rule}
+ if gt_pair_ids.issubset(set(outcomes["rule_id"].unique())):
+ gt_pair_fires = outcomes[outcomes["rule_id"].isin(gt_pair_ids)]
+ gt_pair_trades = fires_to_trade_list(sort_fires_chronological(gt_pair_fires))
+ portfolio_compare["sim_gt_model"] = simulate_sized_portfolio(
+ gt_pair_trades,
+ last_price=float(mark) if mark else None,
+ )
+
+ holdout = outcomes[
+ outcomes["rule_id"].isin(monitor_ids) & (outcomes["split"] == "holdout")
+ ]
+ if not holdout.empty and not all_monitor.empty:
+ full_trades = fires_to_trade_list(sort_fires_chronological(all_monitor))
+ if full_trades:
+ from deepcoin.ground_truth.gt_allocation import simulate_portfolio_steps
+
+ steps = simulate_portfolio_steps(full_trades, use_amount_krw=True)
+ if steps:
+ outcomes_ts = outcomes.copy()
+ outcomes_ts["ts"] = pd.to_datetime(outcomes_ts["dt"])
+ h0 = outcomes_ts["ts"].quantile(1.0 - MATCH_HOLDOUT_RATIO)
+ assets = [(s["dt"], float(s["total_asset_krw"])) for s in steps]
+ pre = [a for d, a in assets if pd.to_datetime(d) < h0]
+ in_h = [a for d, a in assets if pd.to_datetime(d) >= h0]
+ asset_start = pre[-1] if pre else float(GT_INITIAL_CASH_KRW)
+ asset_end = in_h[-1] if in_h else assets[-1][1]
+ ho_pnl_pct = (
+ (asset_end - asset_start) / asset_start * 100.0
+ if asset_start > 0
+ else 0.0
+ )
+ portfolio_compare["sim_sized_holdout"] = {
+ "initial_asset_krw": round(asset_start, 0),
+ "final_asset_krw": round(asset_end, 0),
+ "pnl_krw": round(asset_end - asset_start, 0),
+ "pnl_pct": round(ho_pnl_pct, 2),
+ "note": "전기간 복리 후 holdout 구간 자산 증감 (1M 재시작 아님)",
+ "trade_count": int(len(holdout)),
+ }
+
+ if portfolio_compare.get("sim_sized") and portfolio_compare.get("ground_truth_chrono"):
+ gt_pnl = float(portfolio_compare["ground_truth_chrono"].get("pnl_pct", 0))
+ sim_pnl = float(portfolio_compare["sim_sized"].get("pnl_pct", 0))
+ portfolio_compare["gt_capture_ratio"] = round(
+ sim_pnl / gt_pnl if abs(gt_pnl) > 1e-6 else 0.0,
+ 4,
+ )
+ portfolio_compare["gt_pnl_pct"] = gt_pnl
+ portfolio_compare["sim_sized_pnl_pct"] = sim_pnl
+ if portfolio_compare.get("sim_gt_model"):
+ gtp = float(portfolio_compare["sim_gt_model"].get("pnl_pct", 0))
+ portfolio_compare["gt_model_capture_ratio"] = round(
+ gtp / gt_pnl if abs(gt_pnl) > 1e-6 else 0.0,
+ 4,
+ )
+
+ portfolio_compare["gt_allocation_analysis"] = gt_alloc_analysis
+ portfolio_compare["causal_mode"] = {
+ "gt_signal_causal": GT_SIGNAL_CAUSAL,
+ "sim_causal_tier": SIM_CAUSAL_TIER,
+ "note": "인과적: t 시점까지 데이터만 사용 (운영 정합)",
+ }
+
gt_portfolio: dict[str, Any] = {}
if ANALYSIS_GT_CALIBRATION_JSON.is_file():
cal = json.loads(ANALYSIS_GT_CALIBRATION_JSON.read_text(encoding="utf-8"))
gt_portfolio = cal.get("final", {})
else:
- from deepcoin.ground_truth.ground_truth import load_ground_truth
from deepcoin.matching.gt_asset_calibration import (
portfolio_asset_ratio,
)
- gt_data = load_ground_truth(resolve_ground_truth_file()) or {}
- trades = gt_data.get("trades") or []
- mark = (gt_data.get("summary") or {}).get("mark_price")
+ gt_data_cal = load_ground_truth(resolve_ground_truth_file()) or {}
+ trades = gt_data_cal.get("trades") or []
+ mark_cal = (gt_data_cal.get("summary") or {}).get("mark_price")
if trades:
gt_portfolio = {
- "portfolio": portfolio_asset_ratio(trades, set(), mark),
+ "portfolio": portfolio_asset_ratio(trades, set(), mark_cal),
"note": "캘리브레이션 미실행 — scripts/04_calibrate_gt_assets.py",
}
summaries = matched.get("all_rule_summaries") or matched.get("monitor_rules") or []
+ leg_weight_check = summarize_leg_weights(gt_trades) if gt_trades else {}
+ invalid_legs = [lid for lid, info in leg_weight_check.items() if not info.get("valid", True)]
return {
"label_mode": matched.get("label_mode"),
"train_ratio": MATCH_TRAIN_RATIO,
@@ -389,7 +424,13 @@ def build_simulation_report(
"live_order_cap_sim": live_cap,
"go_no_go": go,
"portfolio_compare": portfolio_compare,
- "gt_model": gt_data.get("model"),
+ "gt_model": gt_data.get("model") or model_to_dict(default_model()),
+ "gt_weight_policy": weight_policy_summary(default_model()),
+ "gt_leg_weight_validation": {
+ "legs": leg_weight_check,
+ "invalid_leg_ids": invalid_legs,
+ "all_valid": len(invalid_legs) == 0,
+ },
"monitor_rules": matched.get("monitor_rules", []),
"gt_portfolio_calibration": gt_portfolio,
"criteria": {
@@ -450,6 +491,24 @@ def run_simulation_report(
)
cal = report.get("gt_portfolio_calibration") or {}
port = cal.get("portfolio") or {}
+ pc = report.get("portfolio_compare") or {}
+ if pc.get("gt_capture_ratio") is not None:
+ print(
+ f"[시뮬] GT 대비 sim_sized(전기간 복리): {pc.get('sim_sized_pnl_pct')}% "
+ f"/ GT {pc.get('gt_pnl_pct')}% "
+ f"(capture={pc.get('gt_capture_ratio'):.2%})"
+ )
+ if pc.get("gt_model_capture_ratio") is not None:
+ print(
+ f"[시뮬] GT 대비 sim_gt_model: "
+ f"{pc.get('sim_gt_model', {}).get('pnl_pct')}% "
+ f"(capture={pc.get('gt_model_capture_ratio'):.2%})"
+ )
+ if pc.get("sim_sized", {}).get("max_drawdown_pct") is not None:
+ print(
+ f"[시뮬] sim_sized MDD: {pc['sim_sized']['max_drawdown_pct']}% "
+ f"(GT MDD: {pc.get('ground_truth_chrono', {}).get('max_drawdown_pct')}%)"
+ )
if port.get("asset_ratio") is not None:
met = cal.get("targets_met", port.get("target_met_90"))
print(
diff --git a/deepcoin/matching/simulation_html.py b/deepcoin/matching/simulation_html.py
index d368416..8b76b22 100644
--- a/deepcoin/matching/simulation_html.py
+++ b/deepcoin/matching/simulation_html.py
@@ -26,10 +26,10 @@ from deepcoin.ground_truth.ground_truth import (
)
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,
+ sort_fires_chronological,
)
from deepcoin.matching.select_rules import _split_train_valid_holdout
from deepcoin.ops.chart_report import (
@@ -231,7 +231,7 @@ def _summary_cards_html(
gt_pnl, "정답 GT", len(gt_trades)
)
sim_sized_title = (
- "시뮬·총자산% (EV/WF·leg상위) — "
+ "시뮬·GT tier 복리 (전기간, 상한 없음) — "
f'{"GO" if go_flag else "NO-GO"}'
)
sim_fixed_title = f"시뮬·고정 ₩{LIVE_ORDER_KRW:,}/회 (비교)"
@@ -280,15 +280,17 @@ def build_simulation_page_html(
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()
- capped = select_capped_fires(holdout_fires)
+ 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 []
@@ -316,8 +318,8 @@ def build_simulation_page_html(
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)
+ 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:
@@ -377,14 +379,14 @@ def build_simulation_page_html(
return ""
sim_table = f"""
- 시뮬 타점 (holdout {len(holdout_fires)}건 → 체결 가정 {len(capped)}건)
- 총자산×최적비중·현금한도·EV/WF통과·leg상위 대형 매수. 일한도·최대 거래수 적용.
+
시뮬 타점 (전기간 {len(sim_fires)}건 → 복리 체결 {len(compound_fires)}건)
+ 총자산×GT비중×leg tier·보유현금 한도·전기간 복리(일한도 없음).
가격 열 (+/-) = {label_mode} 구간 수익%.{_mark_note(close_val)}
"""
@@ -409,12 +411,12 @@ def build_simulation_page_html(
note = (
f"1단계 시뮬 · holdout {report.get('holdout_ratio', 0.15)} · "
- f"발화 {len(holdout_fires)}건 / 체결가정 {len(capped)}건. "
+ f"전기간 발화 {len(sim_fires)}건 / holdout {len(holdout_fires)}건. "
"상단 카드: 초기 금액·총보유자산·초기 대비 증감율·수수료."
)
legend = (
"▲ 정답 매수 · ▼ 정답 매도 — 삼각형 = GT 체결 금액.
"
- "● 시뮬 — 원 = holdout 발화 (차트). 테이블 = 일한도 적용 체결 순서."
+ "● 시뮬 — holdout 발화 (차트). 테이블 = 전기간 GT tier 복리 체결."
)
if frames is not None:
meta_line = (
@@ -434,7 +436,7 @@ def build_simulation_page_html(
gt_pnl,
sim_sized_pnl,
sim_fixed_pnl,
- len(capped),
+ len(compound_fires),
go_flag,
model_note=model_note,
)
@@ -446,7 +448,7 @@ def build_simulation_page_html(
note=note,
truth_trades=gt_trades,
sim_trades=_fires_to_chart_trades(holdout_fires),
- # 차트 마커는 holdout 전체; 카드·테이블은 일한도 capped
+ # 차트 마커는 holdout; 카드·테이블은 전기간 GT tier 복리
title_suffix="1단계 시뮬레이션 (monitor · holdout)",
legend_html=legend,
footer_sections=sections,
diff --git a/deepcoin/ops/simulation.py b/deepcoin/ops/simulation.py
index fc140b3..cefce62 100644
--- a/deepcoin/ops/simulation.py
+++ b/deepcoin/ops/simulation.py
@@ -28,7 +28,6 @@ from config import (
GT_INITIAL_CASH_KRW,
GT_MARKER_SIZE_MAX,
GT_MARKER_SIZE_MIN,
- GT_MAX_BUY_ORDER_KRW,
LIVE_ORDER_KRW,
MACD_FAST,
MACD_SIGNAL,
@@ -108,7 +107,7 @@ def _marker_hover_text(
def _trade_amount_krw(t: dict) -> float:
"""
- 마커 크기·툴팁용 체결 원화. amount_krw 없으면 비중×상한으로 추정.
+ 마커 크기·툴팁용 체결 원화. amount_krw 없으면 비중×초기자본으로 상대 크기만 추정.
Args:
t: trade dict.
@@ -119,7 +118,7 @@ def _trade_amount_krw(t: dict) -> float:
ak = t.get("amount_krw")
if ak is not None and float(ak) > 0:
return float(ak)
- return max(float(t.get("weight", 1.0)), 0.05) * float(GT_MAX_BUY_ORDER_KRW)
+ return max(float(t.get("weight", 1.0)), 0.05) * float(GT_INITIAL_CASH_KRW)
def _marker_sizes(pts: list[dict]) -> list[float]: