diff --git a/.env.example b/.env.example
index 3fb85d9..72e048e 100644
--- a/.env.example
+++ b/.env.example
@@ -20,8 +20,10 @@ BB_MIN_WIDTH_PCT=0.8
# downloader.py — coins.db 적재 개월 수
DOWNLOAD_MONTHS=6
+DOWNLOAD_MONTHS_1M=2
+USE_DISCOVERED_LIVE=true
-# simulation_1h.py
+# simulation.py
SIM_INITIAL_CASH_KRW=200000
SIM_MIN_ORDER_KRW=5000
TRADING_FEE_RATE=0.0005
diff --git a/README.md b/README.md
index 8b85caa..ea09da2 100644
--- a/README.md
+++ b/README.md
@@ -1,50 +1,64 @@
-# DeepCoin — WLD 볼린저 MTF
+# DeepCoin — WLD 전봉 BB·일목 조합 매매
-빗썸 KRW-WLD 현물 전용. **모든 봉**에 동일한 BB 규칙을 적용하고, 봉별 상태를 비교해 실행·확인 봉을 정합니다.
+빗썸 KRW-WLD 현물. **1, 3, 5, 10, 15, 30, 60, 240, 1440분** 모든 봉에서
+볼린저 밴드·일목균형표 **캔들 위치**를 분석하고, 봉 조합으로 매수·매도 규칙을 탐색합니다.
-## BB 기본 규칙 (모든 간격 동일)
+## 구조
-| 구분 | 조건 |
-|------|------|
-| 매수 | 이전 종가 ≤ 하단, 현재 종가 > 하단 (하단 **상향 돌파**) |
-| 매도 | 이전 종가 < 상단, 현재 종가 ≥ 상단 (상단 **상향 돌파**) |
-| 손절(선택) | 하단 재이탈 |
+```text
+downloader.py → coins.db (전 간격 증분)
+indicators.py → BB·일목 계산
+candle_features.py → 봉별 위치 특징 → 3분 타임라인 행렬
+combination_analyzer.py → 조합 분석·combination_report.json
+rule_discovery.py → discovered_rules.json
+strategy.py → 실시간 evaluate_discovered_live
+monitor_coin.py → 실거래 루프
+simulation.py → 백테스트·HTML 차트
+```
-**MTF 적용** (`mtf_bb.py`, `ACTIVE_MTF_POLICY` / `mtf_bb_policy.json`)
+## 봉별 분석 항목
-- 실행 봉: 3·10·15·30·60분 중 백테스트 수익률 1위
-- 확인 봉: 60분·일봉 등 상위 봉 상태가 매수/매도에 맞을 때만 체결
-- 하락 추세: 매수 차단 (설정 시)
+### 볼린저
+- 이벤트: `cross_up_lower`, `cross_up_upper`, `inside_band`, `squeeze` …
+- 구간: `bb_zone_bottom` ~ `bb_zone_top` (%B)
-봉별 상태: `inside`, `cross_up_lower`, `cross_up_upper`, `below_lower`, `above_upper`, `squeeze` 등
+### 일목균형표
+- `ichi_above_cloud`, `ichi_below_cloud`, `ichi_in_cloud`
+- `ichi_tk_bull` / `ichi_tk_cross_up`, `ichi_cloud_bull` …
-## 파일
+### 조합
+- 3분 기준 `merge_asof`로 모든 봉 특징을 한 행에 정렬
+- `discover`가 AND/OR 조합으로 매수·매도 규칙 탐색
-| 파일 | 역할 |
-|------|------|
-| `strategy.py` | 신호·금액·매도 비율 |
-| `monitor.py` | MTF 데이터, `process_wld_mtf`, 현물 주문 |
-| `monitor_coin.py` | 실시간 루프 |
-| `downloader.py` | `coins.db` (3분·1시간·일봉) |
-| `mtf_bb.py` | 봉별 BB 비교·정책 추천 |
-| `simulation_1h.py` | 백테스트 차트 |
-
-## 실행
+## 실행 순서
```bash
cp .env.example .env
-python downloader.py
-python simulation_1h.py discover # 모든 봉·캔들 특징 탐색 → discovered_rules.json
-python simulation_1h.py # 탐색 규칙 HTML 차트 (기본)
-python simulation_1h.py compare # 9종 조합 순위
-python simulation_1h.py mtf # 봉별 BB 비교 (실거래 전 참고)
-python monitor_coin.py # 실거래는 HTML 최적화 후 연동 예정
+python downloader.py # 1분봉 2개월, 나머지 6개월
+python simulation.py # analyze → discover → HTML (탐색 매수·매도 규칙 표시)
+python monitor_coin.py # 실거래
```
-`DOWNLOAD_MONTHS=6` — 간격: **3, 10, 15, 30, 60, 240, 1440**분.
-**증분 저장**: DB `MAX(ymdhms)` 이후 봉만 INSERT (재실행 시 전체 삭제 없음).
+HTML 차트에는 `discovered_rules.json` 에서 찾은 **매수·매도 규칙**의 체결만 표시합니다.
+고급: `analyze`, `discover`, `compare`, `mtf`.
-## 환경 변수
+## 설정 (`config.py`)
-`BITHUMB_ACCESS_KEY`, `BITHUMB_SECRET_KEY`, `COIN_TELEGRAM_*`,
-`BUY_COOLDOWN_SEC`(기본 300), `SELL_COOLDOWN_SEC`(180), `DEFAULT_BUY_KRW` 등 — `.env.example` 참고.
+| 항목 | 설명 |
+|------|------|
+| `ALL_INTERVALS` | 1,3,5,10,15,30,60,240,1440 |
+| `ENTRY_INTERVAL` | 조합 행렬 기준 3분 |
+| `DOWNLOAD_MONTHS_1M` | 1분봉 보관 개월 (기본 2) |
+| `USE_DISCOVERED_LIVE` | 실거래에 discovered_rules 사용 |
+
+## 출력 파일
+
+| 파일 | 내용 |
+|------|------|
+| `combination_report.json` | 봉별 최신 위치·매수/매도 힌트 |
+| `discovered_rules.json` | 탐색된 매매 규칙 |
+| `reports/wld_bb_simulation.html` | 시뮬 차트 |
+
+## 면책
+
+실거래 손실 책임은 사용자에게 있습니다.
diff --git a/candle_features.py b/candle_features.py
index c088b07..7366683 100644
--- a/candle_features.py
+++ b/candle_features.py
@@ -1,5 +1,5 @@
"""
-모든 봉 간격에 대해 BB 위치·캔들 형태(몸통/꼬리/높이) 특징을 계산하고
+모든 봉(1~1440분)에 BB·일목 위치·캔들 형태 특징을 계산하고
기준 타임라인(3분)에 맞춰 정렬합니다.
"""
@@ -8,11 +8,14 @@ from __future__ import annotations
import numpy as np
import pandas as pd
-from config import ENTRY_INTERVAL
+from config import ALL_INTERVALS, ENTRY_INTERVAL
+from indicators import add_bollinger, add_ichimoku
from strategy import prepare_entry_df
INTERVAL_LABELS: dict[int, str] = {
+ 1: "m1",
3: "m3",
+ 5: "m5",
10: "m10",
15: "m15",
30: "m30",
@@ -27,9 +30,69 @@ def interval_prefix(interval: int) -> str:
return INTERVAL_LABELS.get(interval, f"m{interval}")
+def interval_display(interval: int) -> str:
+ if interval >= 1440:
+ return "일봉"
+ return f"{interval}분"
+
+
+# BB 위치 (밴드 내 %B 구간)
+BB_ZONE_FEATURES: tuple[str, ...] = (
+ "bb_zone_bottom",
+ "bb_zone_low",
+ "bb_zone_mid",
+ "bb_zone_high",
+ "bb_zone_top",
+)
+
+# 일목 위치
+ICHI_FEATURES: tuple[str, ...] = (
+ "ichi_above_cloud",
+ "ichi_below_cloud",
+ "ichi_in_cloud",
+ "ichi_cloud_bull",
+ "ichi_cloud_bear",
+ "ichi_tk_bull",
+ "ichi_tk_bear",
+ "ichi_price_above_tenkan",
+ "ichi_price_below_kijun",
+ "ichi_tk_cross_up",
+ "ichi_tk_cross_down",
+)
+
+# BB 이벤트·캔들 형태
+BB_EVENT_FEATURES: tuple[str, ...] = (
+ "cross_up_lower",
+ "cross_up_upper",
+ "cross_down_lower",
+ "below_lower",
+ "above_upper",
+ "inside_band",
+ "bb_pos_low",
+ "bb_pos_high",
+ "squeeze",
+)
+
+CANDLE_SHAPE_FEATURES: tuple[str, ...] = (
+ "body_strong",
+ "body_weak",
+ "hammer",
+ "shooting_star",
+ "bullish",
+ "bearish",
+)
+
+FEATURE_BOOL_COLS: tuple[str, ...] = (
+ BB_EVENT_FEATURES
+ + BB_ZONE_FEATURES
+ + ICHI_FEATURES
+ + CANDLE_SHAPE_FEATURES
+)
+
+
def compute_bar_features(df: pd.DataFrame) -> pd.DataFrame:
- """단일 봉 DataFrame에 위치·캔들 높이 특징을 추가합니다."""
- out = prepare_entry_df(df.copy())
+ """단일 봉 DataFrame에 BB·일목·캔들 위치 특징을 추가합니다."""
+ out = add_bollinger(add_ichimoku(prepare_entry_df(df.copy())))
if len(out) < 2:
return out
@@ -42,13 +105,6 @@ def compute_bar_features(df: pd.DataFrame) -> pd.DataFrame:
lower = out["Lower"].astype(float)
prev_upper = upper.shift(1)
prev_lower = lower.shift(1)
- ma = out["MA"].astype(float)
-
- band = (upper - lower).replace(0, np.nan)
- out["bb_pos"] = ((c - lower) / band).clip(0, 1)
- out["bb_width_pct"] = (
- out["BB_Width"] if "BB_Width" in out.columns else (band / ma.replace(0, np.nan) * 100)
- )
rng = (h - l).replace(0, np.nan)
body = (c - o).abs()
@@ -57,8 +113,13 @@ def compute_bar_features(df: pd.DataFrame) -> pd.DataFrame:
out["upper_wick_ratio"] = ((h - np.maximum(o, c)) / rng).fillna(0).clip(0, 1)
out["lower_wick_ratio"] = ((np.minimum(o, c) - l) / rng).fillna(0).clip(0, 1)
out["ret_pct"] = ((c - prev_c) / prev_c.replace(0, np.nan)) * 100
- out["bullish"] = (c > o).astype(int)
- out["bearish"] = (c < o).astype(int)
+
+ pos = out["bb_pos"].astype(float)
+ out["bb_zone_bottom"] = (pos < 0.15).astype(int)
+ out["bb_zone_low"] = ((pos >= 0.15) & (pos < 0.35)).astype(int)
+ out["bb_zone_mid"] = ((pos >= 0.35) & (pos < 0.65)).astype(int)
+ out["bb_zone_high"] = ((pos >= 0.65) & (pos < 0.85)).astype(int)
+ out["bb_zone_top"] = (pos >= 0.85).astype(int)
out["cross_up_lower"] = ((prev_c <= prev_lower) & (c > lower)).astype(int)
out["cross_up_upper"] = ((prev_c < prev_upper) & (c >= upper)).astype(int)
@@ -66,35 +127,83 @@ def compute_bar_features(df: pd.DataFrame) -> pd.DataFrame:
out["below_lower"] = (c < lower).astype(int)
out["above_upper"] = (c > upper).astype(int)
out["inside_band"] = ((c >= lower) & (c <= upper)).astype(int)
+ out["bb_pos_low"] = (pos < 0.2).astype(int)
+ out["bb_pos_high"] = (pos > 0.8).astype(int)
+ out["squeeze"] = (out["BB_Width"] < 0.8).astype(int)
+
+ ct = out["ichi_cloud_top"].astype(float)
+ cb = out["ichi_cloud_bottom"].astype(float)
+ ten = out["ichi_tenkan"].astype(float)
+ kij = out["ichi_kijun"].astype(float)
+ prev_ten = ten.shift(1)
+ prev_kij = kij.shift(1)
+
+ out["ichi_above_cloud"] = (c > ct).astype(int)
+ out["ichi_below_cloud"] = (c < cb).astype(int)
+ out["ichi_in_cloud"] = ((c >= cb) & (c <= ct)).astype(int)
+ out["ichi_cloud_bull"] = (out["ichi_span_a"] > out["ichi_span_b"]).astype(int)
+ out["ichi_cloud_bear"] = (out["ichi_span_a"] < out["ichi_span_b"]).astype(int)
+ out["ichi_tk_bull"] = (ten > kij).astype(int)
+ out["ichi_tk_bear"] = (ten < kij).astype(int)
+ out["ichi_price_above_tenkan"] = (c > ten).astype(int)
+ out["ichi_price_below_kijun"] = (c < kij).astype(int)
+ out["ichi_tk_cross_up"] = ((prev_ten <= prev_kij) & (ten > kij)).astype(int)
+ out["ichi_tk_cross_down"] = ((prev_ten >= prev_kij) & (ten < kij)).astype(int)
- out["bb_pos_low"] = (out["bb_pos"] < 0.2).astype(int)
- out["bb_pos_high"] = (out["bb_pos"] > 0.8).astype(int)
out["body_strong"] = (out["body_ratio"] > 0.55).astype(int)
out["body_weak"] = (out["body_ratio"] < 0.25).astype(int)
out["hammer"] = ((out["lower_wick_ratio"] > 0.45) & (out["body_ratio"] < 0.35)).astype(int)
out["shooting_star"] = ((out["upper_wick_ratio"] > 0.45) & (out["body_ratio"] < 0.35)).astype(int)
- out["squeeze"] = (out["bb_width_pct"] < 0.8).astype(int)
+ out["bullish"] = (c > o).astype(int)
+ out["bearish"] = (c < o).astype(int)
return out
-FEATURE_BOOL_COLS: tuple[str, ...] = (
- "cross_up_lower",
- "cross_up_upper",
- "cross_down_lower",
- "below_lower",
- "above_upper",
- "inside_band",
- "bb_pos_low",
- "bb_pos_high",
- "body_strong",
- "body_weak",
- "hammer",
- "shooting_star",
- "squeeze",
- "bullish",
- "bearish",
-)
+def describe_latest_position(df: pd.DataFrame, interval: int) -> dict:
+ """한 봉의 최신 BB·일목 위치 요약."""
+ feat = compute_bar_features(df)
+ if feat.empty:
+ return {"interval": interval, "label": interval_display(interval)}
+ row = feat.iloc[-1]
+ pos = float(row.get("bb_pos", 0.5))
+ bb_zone = "mid"
+ for z in BB_ZONE_FEATURES:
+ if int(row.get(z, 0)) == 1:
+ bb_zone = z.replace("bb_zone_", "")
+ break
+ ichi_pos = "in_cloud"
+ if int(row.get("ichi_above_cloud", 0)):
+ ichi_pos = "above_cloud"
+ elif int(row.get("ichi_below_cloud", 0)):
+ ichi_pos = "below_cloud"
+
+ return {
+ "interval": interval,
+ "label": interval_display(interval),
+ "close": float(row["Close"]),
+ "bb_pos": round(pos, 3),
+ "bb_zone": bb_zone,
+ "bb_state": _bb_event_label(row),
+ "ichi_position": ichi_pos,
+ "ichi_tk": "bull" if int(row.get("ichi_tk_bull", 0)) else "bear",
+ "ichi_cloud": "bull" if int(row.get("ichi_cloud_bull", 0)) else "bear",
+ }
+
+
+def _bb_event_label(row: pd.Series) -> str:
+ for name in (
+ "cross_up_lower",
+ "cross_up_upper",
+ "cross_down_lower",
+ "below_lower",
+ "above_upper",
+ "squeeze",
+ "inside_band",
+ ):
+ if int(row.get(name, 0)) == 1:
+ return name
+ return "neutral"
def _merge_interval_features(
@@ -104,7 +213,16 @@ def _merge_interval_features(
) -> pd.DataFrame:
"""master_index 길이와 동일한 간격 특징만 반환."""
pick = [c for c in FEATURE_BOOL_COLS if c in feat.columns]
- extra = [c for c in ("bb_pos", "body_ratio", "lower_wick_ratio", "ret_pct") if c in feat.columns]
+ extra = [
+ c
+ for c in ("bb_pos", "body_ratio", "lower_wick_ratio", "ret_pct", "bb_width_pct")
+ if c in feat.columns
+ ]
+ if "bb_width_pct" not in feat.columns and "BB_Width" in feat.columns:
+ feat = feat.copy()
+ feat["bb_width_pct"] = feat["BB_Width"]
+ extra.append("bb_width_pct")
+
sub = feat[pick + extra].copy()
sub.columns = [f"{prefix}_{c}" for c in sub.columns]
@@ -124,33 +242,35 @@ def _merge_interval_features(
def build_master_feature_matrix(frames: dict[int, pd.DataFrame]) -> pd.DataFrame:
- """3분 타임라인에 모든 봉의 위치·캔들 특징을 붙인 행렬 (인덱스 유일)."""
+ """3분 타임라인에 모든 봉의 BB·일목·캔들 특징을 붙인 행렬."""
entry = frames.get(ENTRY_INTERVAL)
if entry is None or entry.empty:
- raise ValueError("ENTRY_INTERVAL 데이터가 없습니다.")
+ raise ValueError(f"{ENTRY_INTERVAL}분봉(ENTRY_INTERVAL) 데이터가 없습니다.")
entry_feat = compute_bar_features(entry)
entry_feat = entry_feat[~entry_feat.index.duplicated(keep="last")].sort_index()
- p3 = interval_prefix(ENTRY_INTERVAL)
+ p0 = interval_prefix(ENTRY_INTERVAL)
ohlc = ["Open", "High", "Low", "Close", "Volume", "Upper", "Lower", "MA"]
master = entry_feat[[c for c in ohlc if c in entry_feat.columns]].copy()
for col in FEATURE_BOOL_COLS:
if col in entry_feat.columns:
- master[f"{p3}_{col}"] = entry_feat[col]
- for col in ("bb_pos", "body_ratio", "lower_wick_ratio", "ret_pct", "bb_width_pct"):
+ master[f"{p0}_{col}"] = entry_feat[col]
+ for col in ("bb_pos", "body_ratio", "lower_wick_ratio", "ret_pct", "bb_width_pct", "BB_Width"):
if col in entry_feat.columns:
- master[f"{p3}_{col}"] = entry_feat[col]
+ master[f"{p0}_{col}"] = entry_feat[col]
- parts = [master]
- for interval, df in sorted(frames.items()):
- if interval == ENTRY_INTERVAL or df is None or df.empty:
+ for interval in ALL_INTERVALS:
+ if interval == ENTRY_INTERVAL:
+ continue
+ df = frames.get(interval)
+ if df is None or df.empty:
continue
feat = compute_bar_features(df)
feat = feat[~feat.index.duplicated(keep="last")].sort_index()
prefix = interval_prefix(interval)
- parts.append(_merge_interval_features(master.index, feat, prefix))
+ merged = _merge_interval_features(master.index, feat, prefix)
+ master = pd.concat([master, merged], axis=1)
- out = pd.concat(parts, axis=1)
- return out.loc[:, ~out.columns.duplicated()]
+ return master.loc[:, ~master.columns.duplicated()]
diff --git a/combination_analyzer.py b/combination_analyzer.py
new file mode 100644
index 0000000..2af7478
--- /dev/null
+++ b/combination_analyzer.py
@@ -0,0 +1,222 @@
+"""
+모든 봉의 BB·일목 위치를 조합해 매수/매도 후보 규칙을 분석합니다.
+
+ python simulation.py analyze
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import asdict, dataclass, field
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+
+from candle_features import (
+ FEATURE_BOOL_COLS,
+ INTERVAL_LABELS,
+ build_master_feature_matrix,
+ describe_latest_position,
+ interval_prefix,
+)
+from config import ALL_INTERVALS, ENTRY_INTERVAL, SYMBOL
+
+REPORT_FILE = Path(__file__).parent / "combination_report.json"
+
+
+@dataclass
+class CombinationReport:
+ """봉 조합 분석 결과."""
+
+ generated_at: str
+ intervals_loaded: list[int]
+ latest_positions: list[dict]
+ buy_recommendations: list[str] = field(default_factory=list)
+ sell_recommendations: list[str] = field(default_factory=list)
+ buy_avoid: list[str] = field(default_factory=list)
+ top_buy_pairs: list[dict] = field(default_factory=list)
+ top_sell_pairs: list[dict] = field(default_factory=list)
+ suggested_rules: dict = field(default_factory=dict)
+
+
+def _forward_return(close: pd.Series, bars: int = 20) -> pd.Series:
+ """N봉 후 수익률 (%)."""
+ future = close.shift(-bars)
+ return (future - close) / close.replace(0, np.nan) * 100
+
+
+def _predicate_keys(intervals: list[int]) -> list[str]:
+ keys: list[str] = []
+ for iv in intervals:
+ pfx = interval_prefix(iv)
+ for feat in FEATURE_BOOL_COLS:
+ keys.append(f"{pfx}:{feat}")
+ return keys
+
+
+def analyze_forward_edge(
+ matrix: pd.DataFrame,
+ forward_bars: int = 20,
+ min_samples: int = 40,
+) -> tuple[list[dict], list[dict], list[dict]]:
+ """
+ 단일 조건·2봉 조합별 미래 수익 통계 (학습용 힌트).
+
+ Returns:
+ (매수 유리 top, 매도/회피 top)
+ """
+ close = matrix["Close"].astype(float)
+ fwd = _forward_return(close, forward_bars)
+ valid = fwd.notna()
+ base_mean = float(fwd[valid].mean()) if valid.any() else 0.0
+
+ singles: list[dict] = []
+ cols = [c for c in matrix.columns if any(c.startswith(f"{interval_prefix(iv)}_") for iv in ALL_INTERVALS)]
+
+ for col in cols:
+ mask = matrix[col].fillna(0).astype(bool) & valid
+ n = int(mask.sum())
+ if n < min_samples:
+ continue
+ avg = float(fwd[mask].mean())
+ singles.append(
+ {
+ "key": _col_to_key(col),
+ "column": col,
+ "count": n,
+ "avg_forward_pct": round(avg, 4),
+ "edge_vs_base": round(avg - base_mean, 4),
+ }
+ )
+
+ singles.sort(key=lambda x: x["edge_vs_base"], reverse=True)
+ buy_top = [s for s in singles if s["edge_vs_base"] > 0][:25]
+ sell_top = sorted(singles, key=lambda x: x["edge_vs_base"])[:15]
+
+ pairs: list[dict] = []
+ buy_cols = [s["column"] for s in buy_top[:12]]
+ for i, c1 in enumerate(buy_cols):
+ for c2 in buy_cols[i + 1 :]:
+ if c1.split("_")[0] == c2.split("_")[0]:
+ continue
+ mask = (
+ matrix[c1].fillna(0).astype(bool)
+ & matrix[c2].fillna(0).astype(bool)
+ & valid
+ )
+ n = int(mask.sum())
+ if n < min_samples // 2:
+ continue
+ avg = float(fwd[mask].mean())
+ pairs.append(
+ {
+ "keys": [_col_to_key(c1), _col_to_key(c2)],
+ "count": n,
+ "avg_forward_pct": round(avg, 4),
+ "edge_vs_base": round(avg - base_mean, 4),
+ }
+ )
+ pairs.sort(key=lambda x: x["edge_vs_base"], reverse=True)
+
+ return buy_top, pairs[:20], sell_top
+
+
+def _col_to_key(col: str) -> str:
+ """m3_cross_up_lower -> m3:cross_up_lower."""
+ for pfx in INTERVAL_LABELS.values():
+ if col.startswith(f"{pfx}_"):
+ return f"{pfx}:{col[len(pfx) + 1:]}"
+ return col
+
+
+def build_recommendations(
+ buy_top: list[dict],
+ pair_top: list[dict],
+ sell_top: list[dict],
+) -> tuple[list[str], list[str], list[str], dict]:
+ """사람이 읽을 수 있는 권장·규칙 초안."""
+ buy_rec: list[str] = []
+ sell_rec: list[str] = []
+ avoid: list[str] = []
+
+ for s in buy_top[:8]:
+ buy_rec.append(
+ f"{s['key']} — {s['count']}회, {s['avg_forward_pct']:+.2f}% ({s['edge_vs_base']:+.2f}%p)"
+ )
+ for p in pair_top[:5]:
+ buy_rec.append(
+ f"조합 {' + '.join(p['keys'])} — {p['count']}회, {p['edge_vs_base']:+.2f}%p"
+ )
+ for s in sell_top[:6]:
+ if s["edge_vs_base"] < -0.05:
+ avoid.append(f"매수 회피: {s['key']} ({s['edge_vs_base']:.2f}%p)")
+
+ for s in sell_top[:5]:
+ if "cross_up_upper" in s["key"] or "above_upper" in s["key"] or "ichi_above" in s["key"]:
+ sell_rec.append(f"매도 후보: {s['key']}")
+
+ suggested: dict = {"buy_all": [], "buy_any": [], "sell_all": [], "sell_stop": []}
+ if pair_top:
+ suggested["buy_all"] = pair_top[0]["keys"]
+ elif buy_top:
+ suggested["buy_all"] = [buy_top[0]["key"]]
+ if sell_top:
+ for s in sell_top:
+ if "cross_up_upper" in s.get("key", ""):
+ suggested["sell_all"] = [s["key"]]
+ break
+
+ return buy_rec, sell_rec, avoid, suggested
+
+
+def analyze_combinations(frames: dict[int, pd.DataFrame]) -> CombinationReport:
+ """전체 봉 BB·일목 위치 분석 + 조합 매매 힌트."""
+ from datetime import datetime
+
+ loaded = sorted(frames.keys())
+ latest = [describe_latest_position(frames[iv], iv) for iv in ALL_INTERVALS if iv in frames]
+
+ print("\n=== 봉별 최신 BB·일목 위치 ===")
+ for p in latest:
+ print(
+ f" {p['label']:>6} | BB {p['bb_zone']:>6} ({p['bb_pos']:.2f}) {p['bb_state']:>16} | "
+ f"일목 {p['ichi_position']:>12} TK={p['ichi_tk']} 구름={p['ichi_cloud']}"
+ )
+
+ matrix = build_master_feature_matrix(frames).iloc[52:].copy()
+ print(f"\n특징 행렬: {len(matrix)}행 × {len(matrix.columns)}열")
+
+ buy_top, pair_top, sell_top = analyze_forward_edge(matrix)
+ buy_rec, sell_rec, avoid, suggested = build_recommendations(buy_top, pair_top, sell_top)
+
+ print("\n=== 매수 유리 조건 (단일·상위) ===")
+ for line in buy_rec[:10]:
+ print(f" {line}")
+ print("\n=== 매수 회피 / 매도 참고 ===")
+ for line in avoid[:6]:
+ print(f" {line}")
+ for line in sell_rec[:5]:
+ print(f" {line}")
+
+ return CombinationReport(
+ generated_at=datetime.now().isoformat(timespec="seconds"),
+ intervals_loaded=loaded,
+ latest_positions=latest,
+ buy_recommendations=buy_rec,
+ sell_recommendations=sell_rec,
+ buy_avoid=avoid,
+ top_buy_pairs=pair_top,
+ suggested_rules=suggested,
+ )
+
+
+def save_report(report: CombinationReport, path: Path = REPORT_FILE) -> None:
+ path.write_text(json.dumps(asdict(report), ensure_ascii=False, indent=2), encoding="utf-8")
+ print(f"\n저장: {path}")
+
+
+def load_frames(monitor) -> dict[int, pd.DataFrame]:
+ from mtf_bb import load_frames_from_db
+
+ return load_frames_from_db(monitor, SYMBOL)
diff --git a/config.py b/config.py
index f39c8cb..e6cd40f 100644
--- a/config.py
+++ b/config.py
@@ -24,7 +24,6 @@ KR_COINS: dict[str, str] = {
}
# --- 타임프레임 (분) ---
-ENTRY_INTERVAL = 3
TREND_INTERVAL_1H = 60
TREND_INTERVAL_1D = 1440
@@ -55,10 +54,19 @@ TRADING_FEE_RATE = float(os.getenv("TRADING_FEE_RATE", "0.0005"))
# --- coins.db (downloader.py 적재 간격, 분) ---
# 빗썸 분봉 API: 1,3,5,10,15,30,60,240 / 일봉 1440
-DOWNLOAD_INTERVALS: tuple[int, ...] = (3, 10, 15, 30, 60, 240, 1440)
+ALL_INTERVALS: tuple[int, ...] = (1, 3, 5, 10, 15, 30, 60, 240, 1440)
+DOWNLOAD_INTERVALS: tuple[int, ...] = ALL_INTERVALS
DOWNLOAD_MONTHS = int(os.getenv("DOWNLOAD_MONTHS", "6"))
+# 1분봉은 용량·API 부담으로 기본 2개월 (환경변수로 조정)
+DOWNLOAD_MONTHS_1M = int(os.getenv("DOWNLOAD_MONTHS_1M", "2"))
DB_PATH = "coins.db"
+# 규칙 탐색·조합 분석 기준 타임라인
+ENTRY_INTERVAL = 3
+
+# 실시간: discovered_rules + 전 봉 BB·일목 조합 (False면 mtf_bb_policy)
+USE_DISCOVERED_LIVE = os.getenv("USE_DISCOVERED_LIVE", "true").lower() in ("1", "true", "yes")
+
# --- 시뮬레이션 ---
SIM_INITIAL_CASH_KRW = int(os.getenv("SIM_INITIAL_CASH_KRW", "200000"))
SIM_MIN_ORDER_KRW = int(os.getenv("SIM_MIN_ORDER_KRW", "5000"))
diff --git a/discovered_rules.json b/discovered_rules.json
index ff08afc..c32b1d0 100644
--- a/discovered_rules.json
+++ b/discovered_rules.json
@@ -1,22 +1,19 @@
{
"name": "discovered_best",
"buy_all": [
- "m3:bb_pos_low",
- "m3:shooting_star",
- "m15:inside_band",
- "m3:inside_band",
- "m3:hammer"
+ "m240:above_upper",
+ "m3:bb_zone_bottom"
],
"buy_any": [],
"sell_all": [
"m3:cross_up_upper",
- "m3:bearish",
- "m10:cross_up_upper",
- "m30:bb_pos_high"
+ "m3:ichi_above_cloud",
+ "m3:ichi_cloud_bull",
+ "m3:ichi_tk_cross_up"
],
"sell_stop": [],
- "train_return_pct": 1.9043499145753595,
- "test_return_pct": 0.6201861088011792,
- "full_return_pct": 2.524536023376539,
- "trade_count": 9
+ "train_return_pct": 2.9835000000000003,
+ "test_return_pct": 1.45272321428571,
+ "full_return_pct": 4.43622321428571,
+ "trade_count": 3
}
\ No newline at end of file
diff --git a/downloader.py b/downloader.py
index 92f3674..879ba9e 100644
--- a/downloader.py
+++ b/downloader.py
@@ -18,6 +18,7 @@ from config import (
DB_PATH,
DOWNLOAD_INTERVALS,
DOWNLOAD_MONTHS,
+ DOWNLOAD_MONTHS_1M,
KR_COINS,
SYMBOL,
)
@@ -71,9 +72,18 @@ def interval_label(interval: int) -> str:
return f"{interval}분봉"
+def months_for_interval(interval: int, default_months: int) -> int:
+ """간격별 DB 보관 개월 수 (1분봉은 별도 상한)."""
+ if interval == 1:
+ return DOWNLOAD_MONTHS_1M
+ return default_months
+
+
def download_jobs() -> list[tuple[int, str]]:
labels = {
+ 1: "1분",
3: "3분",
+ 5: "5분",
10: "10분",
15: "15분",
30: "30분",
@@ -241,6 +251,7 @@ def download_symbol(
months: int,
) -> None:
"""한 간격의 봉을 API로 받아 증분 저장합니다."""
+ months = months_for_interval(interval, months)
label = interval_label(interval)
last_ts = get_last_timestamp(symbol, interval)
existing = get_row_count(symbol, interval)
diff --git a/indicators.py b/indicators.py
new file mode 100644
index 0000000..0169bc0
--- /dev/null
+++ b/indicators.py
@@ -0,0 +1,58 @@
+"""
+볼린저 밴드·일목균형표 계산 (모든 봉 간격 공용).
+"""
+
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+
+from config import BB_PERIOD, BB_STD
+
+
+def add_bollinger(
+ df: pd.DataFrame,
+ period: int = BB_PERIOD,
+ std_mult: float = BB_STD,
+) -> pd.DataFrame:
+ """MA, Upper, Lower, BB_Width, bb_pos 컬럼 추가."""
+ out = df.copy()
+ if "MA" not in out.columns:
+ out["MA"] = out["Close"].rolling(period).mean()
+ if "Upper" not in out.columns or "Lower" not in out.columns:
+ std = out["Close"].rolling(period).std()
+ out["STD"] = std
+ out["Upper"] = out["MA"] + std_mult * std
+ out["Lower"] = out["MA"] - std_mult * std
+ ma = out["MA"].replace(0, np.nan)
+ band = (out["Upper"] - out["Lower"]).replace(0, np.nan)
+ out["bb_pos"] = ((out["Close"] - out["Lower"]) / band).clip(0, 1)
+ out["BB_Width"] = band / ma * 100
+ return out
+
+
+def add_ichimoku(
+ df: pd.DataFrame,
+ tenkan: int = 9,
+ kijun: int = 26,
+ senkou_b_period: int = 52,
+) -> pd.DataFrame:
+ """
+ 일목균형표 라인·구름 위치 컬럼 추가 (해당 봉 시점, 미래 데이터 미사용).
+
+ Returns:
+ ichi_tenkan, ichi_kijun, ichi_span_a, ichi_span_b,
+ ichi_cloud_top, ichi_cloud_bottom
+ """
+ out = df.copy()
+ h = out["High"].astype(float)
+ l = out["Low"].astype(float)
+ c = out["Close"].astype(float)
+
+ out["ichi_tenkan"] = (h.rolling(tenkan).max() + l.rolling(tenkan).min()) / 2
+ out["ichi_kijun"] = (h.rolling(kijun).max() + l.rolling(kijun).min()) / 2
+ out["ichi_span_a"] = (out["ichi_tenkan"] + out["ichi_kijun"]) / 2
+ out["ichi_span_b"] = (h.rolling(senkou_b_period).max() + l.rolling(senkou_b_period).min()) / 2
+ out["ichi_cloud_top"] = np.maximum(out["ichi_span_a"], out["ichi_span_b"])
+ out["ichi_cloud_bottom"] = np.minimum(out["ichi_span_a"], out["ichi_span_b"])
+ return out
diff --git a/monitor.py b/monitor.py
index 6f97497..e704502 100644
--- a/monitor.py
+++ b/monitor.py
@@ -307,11 +307,14 @@ class Monitor(HTS):
def process_wld_mtf(self, symbol: str, balances: dict | None = None) -> None:
"""
- WLD MTF: 모든 봉 BB 상태 비교 후 정책에 따라 매수/매도.
+ WLD: 전 봉(1~1440분) BB·일목 위치 조합 매매.
- mtf_bb_policy.json 이 있으면 해당 정책, 없으면 ACTIVE_MTF_POLICY 사용.
+ USE_DISCOVERED_LIVE=True: discovered_rules.json + combination 특징
+ False: mtf_bb_policy.json BB MTF
"""
+ from config import USE_DISCOVERED_LIVE
from mtf_bb import load_frames_from_db, load_policy, print_latest_states
+ from candle_features import describe_latest_position
try:
frames = load_frames_from_db(self, symbol)
@@ -326,29 +329,41 @@ class Monitor(HTS):
if df_1h is None or df_1h.empty:
df_1h = frames.get(ENTRY_INTERVAL)
- policy = load_policy() or strategy.ACTIVE_MTF_POLICY
- cfg = strategy.ACTIVE_CONFIG
- print_latest_states(frames, cfg)
- print(
- f"MTF 정책: {policy.name} | "
- f"매수={policy.buy_interval}분 | 매도={policy.sell_interval}분 | "
- f"확인={list(policy.buy_confirm_intervals)}"
- )
-
trend = strategy.get_trend(df_1d, df_1h)
print(f"{symbol} 추세: {trend}")
+ print("--- 봉별 BB·일목 위치 ---")
+ for iv in sorted(frames.keys()):
+ pos = describe_latest_position(frames[iv], iv)
+ print(
+ f" {pos['label']:>6} | BB {pos['bb_zone']} {pos['bb_state']:>16} | "
+ f"일목 {pos['ichi_position']} TK={pos['ichi_tk']}"
+ )
- entry = frames.get(ENTRY_INTERVAL)
- trade = strategy.evaluate(
- symbol,
- entry if entry is not None else frames[policy.buy_interval],
- df_1h,
- df_1d,
- config=cfg,
- frames=frames,
- policy=policy,
- )
+ if USE_DISCOVERED_LIVE:
+ print("모드: 전봉 BB·일목 조합 (discovered_rules)")
+ trade = strategy.evaluate_discovered_live(
+ symbol, frames, df_1d, df_1h, balances or {}
+ )
+ else:
+ policy = load_policy() or strategy.ACTIVE_MTF_POLICY
+ cfg = strategy.ACTIVE_CONFIG
+ print_latest_states(frames, cfg)
+ print(
+ f"MTF 정책: {policy.name} | 매수={policy.buy_interval}분 | "
+ f"매도={policy.sell_interval}분"
+ )
+ entry = frames.get(ENTRY_INTERVAL)
+ trade = strategy.evaluate(
+ symbol,
+ entry if entry is not None else frames[policy.buy_interval],
+ df_1h,
+ df_1d,
+ config=cfg,
+ frames=frames,
+ policy=policy,
+ )
if trade is None:
+ print("신호 없음")
return
self.execute_trade_signal(symbol, trade, balances=balances)
except Exception as e:
diff --git a/monitor_coin.py b/monitor_coin.py
index 879de22..87b8f86 100644
--- a/monitor_coin.py
+++ b/monitor_coin.py
@@ -1,7 +1,7 @@
"""
-WLD(월드코인) 실시간 모니터 — 3분 BB MTF (평균회귀 + 돌파).
+WLD(월드코인) 실시간 모니터 — 전 봉 BB·일목 조합 (discovered_rules).
-전략: strategy.py
+전략: strategy.py / rule_discovery.py
"""
from datetime import datetime
diff --git a/mtf_bb.py b/mtf_bb.py
index 5673fed..b2b06a7 100644
--- a/mtf_bb.py
+++ b/mtf_bb.py
@@ -83,7 +83,7 @@ def backtest_interval(
cfg: StrategyConfig,
) -> IntervalBacktestSummary | None:
"""한 간격만으로 BB 매매 백테스트."""
- from simulation_1h import run_backtest
+ from simulation import run_backtest
if interval not in frames:
return None
diff --git a/rule_discovery.py b/rule_discovery.py
index 2074c16..cc16459 100644
--- a/rule_discovery.py
+++ b/rule_discovery.py
@@ -1,7 +1,7 @@
"""
모든 봉·캔들 특징 행렬에서 매수/매도 규칙을 탐색합니다 (인과적 백테스트).
- python simulation_1h.py discover
+ python simulation.py discover
"""
from __future__ import annotations
@@ -46,6 +46,9 @@ NEG_BLOCK_FEATURES: tuple[str, ...] = (
"above_upper",
"cross_down_lower",
"shooting_star",
+ "ichi_above_cloud",
+ "bb_zone_top",
+ "bb_pos_high",
)
# 탐색 규칙 적용 시 항상 매수 차단 (상단 돌파·과열)
@@ -238,7 +241,7 @@ def backtest_rules(
"""
HTML 시뮬과 동일한 run_backtest 로직으로 수익률 계산.
"""
- from simulation_1h import run_backtest
+ from simulation import run_backtest
import strategy as st
df = entry_ohlc.loc[matrix.index].copy()
@@ -261,12 +264,39 @@ def backtest_rules(
def _baseline_rules() -> DiscoveredRules:
+ """다봉 BB 하단 돌파 + 상단 돌파 기준선."""
p3 = interval_prefix(ENTRY_INTERVAL)
return DiscoveredRules(
- name="baseline_bb",
- buy_all=[f"{p3}:cross_up_lower"],
- sell_all=[f"{p3}:cross_up_upper"],
- sell_stop=[],
+ name="baseline_bb_mtf",
+ buy_all=[
+ f"{p3}:cross_up_lower",
+ f"{interval_prefix(60)}:ichi_tk_bull",
+ f"{interval_prefix(1440)}:!ichi_below_cloud",
+ ],
+ sell_all=[
+ f"{p3}:cross_up_upper",
+ f"{interval_prefix(60)}:cross_up_upper",
+ ],
+ sell_stop=[f"{p3}:cross_down_lower"],
+ )
+
+
+def _seed_from_combination_report() -> DiscoveredRules | None:
+ """combination_report.json 제안 규칙."""
+ path = Path(__file__).parent / "combination_report.json"
+ if not path.exists():
+ return None
+ data = json.loads(path.read_text(encoding="utf-8"))
+ sug = data.get("suggested_rules") or {}
+ buy_all = list(sug.get("buy_all") or [])
+ if not buy_all:
+ return None
+ return DiscoveredRules(
+ name="combination_seed",
+ buy_all=buy_all,
+ buy_any=[list(g) for g in (sug.get("buy_any") or []) if g],
+ sell_all=list(sug.get("sell_all") or [f"{interval_prefix(ENTRY_INTERVAL)}:cross_up_upper"]),
+ sell_stop=list(sug.get("sell_stop") or []),
)
@@ -278,8 +308,8 @@ def greedy_search(
df_1d: pd.DataFrame,
df_1h: pd.DataFrame,
entry_ohlc: pd.DataFrame,
- max_buy: int = 5,
- max_sell: int = 4,
+ max_buy: int = 8,
+ max_sell: int = 6,
max_stop: int = 2,
) -> DiscoveredRules:
"""학습 구간 수익률을 올리도록 매수/매도 조건을 탐욕적으로 확장."""
@@ -375,7 +405,14 @@ def try_buy_any_branches(
) -> DiscoveredRules:
"""매수 OR 분기: 다른 봉의 cross_up_lower / hammer 등."""
train = matrix.iloc[:train_end]
- triggers = [p for p in pool if p.endswith(":cross_up_lower") or p.endswith(":hammer")]
+ triggers = [
+ p
+ for p in pool
+ if p.endswith(":cross_up_lower")
+ or p.endswith(":hammer")
+ or p.endswith(":bb_zone_bottom")
+ or p.endswith(":ichi_tk_cross_up")
+ ]
best = DiscoveredRules(
name=base.name,
buy_all=list(base.buy_all),
@@ -476,8 +513,9 @@ def discover_rules(frames: dict[int, pd.DataFrame]) -> DiscoveredRules:
pool = generate_predicate_pool(intervals)
print(f" 샘플 {n}봉 | 학습 {train_end} | predicate 후보 {len(pool)}개")
- baseline = _baseline_rules()
+ baseline = _seed_from_combination_report() or _baseline_rules()
br, bt = backtest_rules(matrix.iloc[:train_end], baseline, df_1d, df_1h, entry_ohlc)
+ print(f" 시드 규칙: {baseline.name}")
bf, _ = backtest_rules(matrix, baseline, df_1d, df_1h, entry_ohlc)
print(f" 기준선(3분 BB만): 학습 {br:+.2f}% | 전체 {bf:+.2f}%")
diff --git a/simulation.py b/simulation.py
new file mode 100644
index 0000000..6f5e4de
--- /dev/null
+++ b/simulation.py
@@ -0,0 +1,855 @@
+"""
+WLD 3분 BB 시뮬레이션.
+
+기본: 하단 상향 돌파 매수, 상단 상향 돌파 매도.
+수수료 반영, 레짐/필터 조합 비교 지원.
+
+ python simulation.py # analyze → discover → HTML (탐색 규칙 매수·매도)
+ python simulation.py analyze # (고급) 조합 분석만
+ python simulation.py discover # (고급) 규칙 탐색만
+ python simulation.py compare # (고급) 9종 프리셋 비교
+ python simulation.py mtf # (고급) 구 MTF BB 정책
+"""
+
+from __future__ import annotations
+
+import sys
+import webbrowser
+from dataclasses import dataclass
+from pathlib import Path
+
+import pandas as pd
+import plotly.graph_objs as go
+from plotly.subplots import make_subplots
+
+from config import (
+ BUY_COOLDOWN_SEC,
+ COIN_NAME,
+ ENTRY_INTERVAL,
+ SELL_COOLDOWN_SEC,
+ SIM_INITIAL_CASH_KRW,
+ SIM_MIN_ORDER_KRW,
+ SYMBOL,
+ TRADING_FEE_RATE,
+ TREND_INTERVAL_1D,
+ TREND_INTERVAL_1H,
+)
+from monitor import Monitor
+import strategy
+
+REPORT_DIR = Path(__file__).resolve().parent / "reports"
+OUTPUT_HTML = REPORT_DIR / "wld_bb_simulation.html"
+
+
+def interval_chart_label(interval_min: int) -> str:
+ """차트 제목용 봉 라벨."""
+ if interval_min >= 1440:
+ return "일봉"
+ return f"{interval_min}분봉"
+
+
+def _add_trade_markers(fig, trades: list["SimTrade"], row: int = 1) -> None:
+ """
+ 매수·매도 마커·라벨 (Scatter trace만 사용, 범례와 함께 토글).
+ simulate_mtf.py 와 동일 스타일.
+ """
+ for action, color, symbol, label, text_pos in [
+ ("매수", "#16a34a", "triangle-up", "매수", "top center"),
+ ("매도", "#dc2626", "triangle-down", "매도", "bottom center"),
+ ]:
+ pts = [t for t in trades if t.action == action]
+ if not pts:
+ continue
+ fig.add_trace(
+ go.Scatter(
+ x=[t.dt for t in pts],
+ y=[t.price for t in pts],
+ mode="markers+text",
+ name=label,
+ legendgroup=label,
+ text=[label] * len(pts),
+ textposition=text_pos,
+ textfont=dict(
+ size=12,
+ color=color,
+ family="Malgun Gothic, Arial, sans-serif",
+ ),
+ marker=dict(
+ symbol=symbol,
+ size=16,
+ color=color,
+ line=dict(width=2, color="#111"),
+ ),
+ hovertext=[
+ f"{label} 체결
{t.signal}
₩{t.price:,.2f}
₩{t.krw:,.0f}"
+ for t in pts
+ ],
+ hovertemplate="%{hovertext}",
+ ),
+ row=row,
+ col=1,
+ )
+
+
+def build_simulation_html(
+ df: pd.DataFrame,
+ result: SimResult,
+ trend: str,
+ interval_min: int = ENTRY_INTERVAL,
+ note: str = "",
+) -> str:
+ """simulate_mtf.py 와 동일 레이아웃의 HTML 리포트."""
+ df = strategy.prepare_entry_df(df.copy())
+ iv_label = interval_chart_label(interval_min)
+ buy_n = sum(1 for t in result.trades if t.action == "매수")
+ sell_n = sum(1 for t in result.trades if t.action == "매도")
+ pnl_krw = result.final_asset - result.initial_cash
+
+ summary = {
+ "config_name": result.config_name,
+ "period_start": str(df.index[0]),
+ "period_end": str(df.index[-1]),
+ "interval_label": iv_label,
+ "trend": trend,
+ "signal_count": len(result.trades),
+ "buy_signal_count": buy_n,
+ "sell_signal_count": sell_n,
+ "total_trades": result.trade_count,
+ "pnl_krw": round(pnl_krw, 0),
+ "pnl_pct": round(result.total_return_pct, 2),
+ "total_fees": round(result.total_fees, 0),
+ "win_count": result.win_count,
+ "note": note,
+ }
+
+ fig = make_subplots(
+ rows=3,
+ cols=1,
+ shared_xaxes=True,
+ vertical_spacing=0.05,
+ row_heights=[0.58, 0.2, 0.22],
+ subplot_titles=(
+ f"{COIN_NAME} ({SYMBOL}) {iv_label} — {result.config_name}",
+ "RSI (14)",
+ "거래량",
+ ),
+ )
+
+ fig.add_trace(
+ go.Candlestick(
+ x=df.index,
+ open=df["Open"],
+ high=df["High"],
+ low=df["Low"],
+ close=df["Close"],
+ name=f"{iv_label} 캔들",
+ increasing_line_color="#ef4444",
+ decreasing_line_color="#3b82f6",
+ ),
+ row=1,
+ col=1,
+ )
+ if "MA" in df.columns:
+ fig.add_trace(
+ go.Scatter(
+ x=df.index,
+ y=df["MA"],
+ name="BB 중심",
+ line=dict(color="#64748b", width=1, dash="dot"),
+ ),
+ row=1,
+ col=1,
+ )
+ if "Upper" in df.columns:
+ fig.add_trace(
+ go.Scatter(
+ x=df.index,
+ y=df["Upper"],
+ name="BB 상단",
+ line=dict(color="#94a3b8", width=1),
+ ),
+ row=1,
+ col=1,
+ )
+ if "Lower" in df.columns:
+ fig.add_trace(
+ go.Scatter(
+ x=df.index,
+ y=df["Lower"],
+ name="BB 하단",
+ line=dict(color="#94a3b8", width=1),
+ ),
+ row=1,
+ col=1,
+ )
+
+ _add_trade_markers(fig, result.trades, row=1)
+
+ if not result.trades:
+ fig.add_annotation(
+ text=(
+ f"이 기간에는 체결 신호가 없습니다.
"
+ f"전략: {result.config_name} | 추세: {trend}"
+ ),
+ xref="paper",
+ yref="paper",
+ x=0.5,
+ y=0.88,
+ showarrow=False,
+ font=dict(size=14, color="#b45309"),
+ bgcolor="#fffbeb",
+ bordercolor="#f59e0b",
+ borderwidth=1,
+ )
+
+ if "RSI" in df.columns:
+ fig.add_trace(
+ go.Scatter(
+ x=df.index,
+ y=df["RSI"],
+ name="RSI",
+ line=dict(color="#7c3aed"),
+ ),
+ row=2,
+ col=1,
+ )
+ fig.add_hline(y=70, line_dash="dot", line_color="#9ca3af", row=2, col=1)
+ fig.add_hline(y=50, line_dash="dot", line_color="#d1d5db", row=2, col=1)
+
+ fig.add_trace(
+ go.Bar(
+ x=df.index,
+ y=df["Volume"],
+ name="Volume",
+ marker_color="#cbd5e1",
+ ),
+ row=3,
+ col=1,
+ )
+
+ fig.update_layout(
+ height=920,
+ template="plotly_white",
+ xaxis_rangeslider_visible=False,
+ legend=dict(orientation="h", y=1.05, x=0),
+ margin=dict(l=60, r=30, t=90, b=40),
+ )
+ fig.update_yaxes(title_text="가격 (KRW)", row=1, col=1)
+ fig.update_yaxes(title_text="RSI", row=2, col=1, range=[0, 100])
+
+ chart_html = fig.to_html(full_html=False, include_plotlyjs="cdn")
+
+ trade_rows = ""
+ for t in result.trades:
+ cls = "buy" if t.action == "매수" else "sell"
+ pnl = f"{t.pnl:+,.0f}" if t.pnl is not None else "-"
+ trade_rows += f"""
+
+ | {t.dt} |
+ {t.action} |
+ 체결 |
+ ₩{t.price:,.2f} |
+ {t.signal} |
+ ₩{t.krw:,.0f} |
+ ₩{t.fee:,.0f} |
+ {pnl} |
+
"""
+ if not trade_rows:
+ trade_rows = '| 신호 없음 |
'
+
+ note_html = f"{summary['note']}
" if summary.get("note") else ""
+ sells = summary["sell_signal_count"]
+ win_rate = (
+ summary["win_count"] / sells * 100 if sells else 0.0
+ )
+
+ return f"""
+
+
+
+ {SYMBOL} BB 시뮬레이션
+
+
+
+ {COIN_NAME} ({SYMBOL}) BB 시뮬레이션
+ 전략: {summary['config_name']} | 추세: {summary['trend']} | 기간: {summary['period_start']} ~ {summary['period_end']}
+ {note_html}
+
+ ▲ 매수 범례 클릭 시 마커·라벨 함께 숨김
+ ▼ 매도 동일
+
+
+
체결{summary['total_trades']} (매수 {summary['buy_signal_count']} / 매도 {summary['sell_signal_count']})
+
손익₩{summary['pnl_krw']:+,.0f} ({summary['pnl_pct']:+.2f}%)
+
수수료₩{summary['total_fees']:,.0f}
+
승률(매도 기준){win_rate:.1f}%
+
{chart_html}
+ 신호·체결 내역
+
+ | 시각 | 구분 | 상태 | 가격 | 신호 | 금액 | 수수료 | 손익 |
|---|
+ {trade_rows}
+
+
+"""
+
+
+@dataclass
+class SimTrade:
+ dt: pd.Timestamp
+ action: str
+ signal: str
+ price: float
+ krw: float
+ fee: float
+ quantity: float
+ pnl: float | None
+ cash_after: float
+ total_asset: float
+
+
+@dataclass
+class SimResult:
+ config_name: str
+ trades: list[SimTrade]
+ initial_cash: float
+ final_cash: float
+ final_coin_qty: float
+ final_price: float
+ realized_pnl: float
+ total_fees: float
+ final_asset: float
+ total_return_pct: float
+ trade_count: int
+ win_count: int
+
+
+def run_backtest(
+ df_3m: pd.DataFrame,
+ df_1d: pd.DataFrame,
+ df_1h: pd.DataFrame,
+ config_name: str = "",
+ initial_cash: float = SIM_INITIAL_CASH_KRW,
+ min_order_krw: float = SIM_MIN_ORDER_KRW,
+ fee_rate: float = TRADING_FEE_RATE,
+) -> SimResult:
+ """신호 순서대로 현물 매수/매도 시뮬레이션 (수수료 차감)."""
+ cash = float(initial_cash)
+ coin_qty = 0.0
+ cost_basis = 0.0
+ realized_pnl = 0.0
+ total_fees = 0.0
+ win_count = 0
+ trades: list[SimTrade] = []
+ last_buy_ts: pd.Timestamp | None = None
+ last_sell_ts: pd.Timestamp | None = None
+
+ signals = df_3m[df_3m["point"] == 1].sort_index()
+
+ for ts, row in signals.iterrows():
+ price = float(row["Close"])
+ action = str(row.get("action", ""))
+ signal_name = str(row.get("signal", ""))
+ if price <= 0:
+ continue
+
+ trend_at = str(row.get("trend", "")) or strategy.get_trend_at(df_1d, df_1h, ts)
+ if trend_at not in ("up", "down", "range"):
+ trend_at = strategy.get_trend_at(df_1d, df_1h, ts)
+
+ if action == "buy":
+ if last_buy_ts is not None:
+ if (ts - last_buy_ts).total_seconds() < BUY_COOLDOWN_SEC:
+ continue
+
+ buy_krw = float(
+ strategy.get_buy_amount(SYMBOL, signal_name, price, trend_at)
+ )
+ buy_krw = max(min_order_krw, min(buy_krw, cash))
+ fee = buy_krw * fee_rate
+ total_cost = buy_krw + fee
+ if buy_krw < min_order_krw or cash < total_cost:
+ continue
+
+ qty = buy_krw / price
+ cash -= total_cost
+ total_fees += fee
+ cost_basis += buy_krw
+ coin_qty += qty
+ last_buy_ts = ts
+
+ trades.append(
+ SimTrade(
+ dt=ts,
+ action="매수",
+ signal=signal_name,
+ price=price,
+ krw=buy_krw,
+ fee=fee,
+ quantity=qty,
+ pnl=None,
+ cash_after=cash,
+ total_asset=cash + coin_qty * price,
+ )
+ )
+ continue
+
+ if action == "sell":
+ if coin_qty <= 0:
+ continue
+ if last_sell_ts is not None:
+ if (ts - last_sell_ts).total_seconds() < SELL_COOLDOWN_SEC:
+ continue
+
+ ratio = strategy.get_sell_ratio(SYMBOL, signal_name)
+ sell_qty = min(coin_qty * ratio, coin_qty)
+ sell_krw = sell_qty * price
+
+ if sell_krw < min_order_krw:
+ if coin_qty * price < min_order_krw:
+ continue
+ sell_qty = coin_qty
+ sell_krw = sell_qty * price
+
+ fee = sell_krw * fee_rate
+ net = sell_krw - fee
+ avg_cost = cost_basis / coin_qty
+ sold_cost = avg_cost * sell_qty
+ pnl = net - sold_cost
+
+ cash += net
+ total_fees += fee
+ cost_basis -= sold_cost
+ coin_qty -= sell_qty
+ realized_pnl += pnl
+ if pnl > 0:
+ win_count += 1
+ if coin_qty < 1e-12:
+ coin_qty = 0.0
+ cost_basis = 0.0
+ last_sell_ts = ts
+
+ trades.append(
+ SimTrade(
+ dt=ts,
+ action="매도",
+ signal=signal_name,
+ price=price,
+ krw=sell_krw,
+ fee=fee,
+ quantity=sell_qty,
+ pnl=pnl,
+ cash_after=cash,
+ total_asset=cash + coin_qty * price,
+ )
+ )
+
+ final_price = float(df_3m["Close"].iloc[-1])
+ final_asset = cash + coin_qty * final_price
+ sell_trades = sum(1 for t in trades if t.action == "매도")
+
+ return SimResult(
+ config_name=config_name,
+ trades=trades,
+ initial_cash=initial_cash,
+ final_cash=cash,
+ final_coin_qty=coin_qty,
+ final_price=final_price,
+ realized_pnl=realized_pnl,
+ total_fees=total_fees,
+ final_asset=final_asset,
+ total_return_pct=(final_asset - initial_cash) / initial_cash * 100
+ if initial_cash > 0
+ else 0.0,
+ trade_count=len(trades),
+ win_count=win_count if sell_trades else 0,
+ )
+
+
+def print_backtest_report(result: SimResult) -> None:
+ fee_pct = TRADING_FEE_RATE * 100
+ print("\n" + "=" * 80)
+ print(
+ f"[{result.config_name}] 시작 {result.initial_cash:,.0f}원 | "
+ f"최소주문 {SIM_MIN_ORDER_KRW:,.0f}원 | 수수료 {fee_pct:.3f}%/쪽"
+ )
+ print("=" * 80)
+ if not result.trades:
+ print("체결 없음")
+ else:
+ print(
+ f"{'일시':<18} {'구분':<4} {'신호':<22} {'가격':>9} {'금액':>10} "
+ f"{'수수료':>8} {'수익':>10}"
+ )
+ print("-" * 80)
+ for t in result.trades:
+ pnl_s = f"{t.pnl:+,.0f}" if t.pnl is not None else "-"
+ print(
+ f"{t.dt.strftime('%Y-%m-%d %H:%M'):<18} {t.action:<4} {t.signal:<22} "
+ f"{t.price:>9,.2f} {t.krw:>10,.0f} {t.fee:>8,.0f} {pnl_s:>10}"
+ )
+ print("-" * 80)
+ sells = sum(1 for t in result.trades if t.action == "매도")
+ win_rate = result.win_count / sells * 100 if sells else 0.0
+ print(f"거래 횟수: {result.trade_count} (매도 {sells}회) | 승률: {win_rate:.1f}%")
+ print(f"수수료 합계: {result.total_fees:,.0f}원")
+ print(f"실현 손익(수수료 반영): {result.realized_pnl:+,.0f}원")
+ print(
+ f"최종 자산: {result.final_asset:,.0f}원 | "
+ f"총수익: {result.final_asset - result.initial_cash:+,.0f}원 "
+ f"({result.total_return_pct:+.2f}%)"
+ )
+ print("=" * 80)
+
+
+def run_comparison(df_1d: pd.DataFrame, df_1h: pd.DataFrame, df_3m: pd.DataFrame) -> None:
+ """기법 조합별 수익률 비교 (수수료 포함)."""
+ print(f"\n{'='*80}")
+ print(f"전략 조합 비교 — {SYMBOL} 3분 | {df_3m.index[0]} ~ {df_3m.index[-1]}")
+ print(f"시작 {SIM_INITIAL_CASH_KRW:,}원 | 수수료 {TRADING_FEE_RATE*100:.3f}%/매수·매도")
+ print(f"{'='*80}")
+ print(
+ f"{'순위':<4} {'조합':<22} {'수익률':>9} {'최종자산':>12} "
+ f"{'거래':>6} {'승률':>7} {'수수료':>10}"
+ )
+ print("-" * 80)
+
+ rows: list[tuple[SimResult, strategy.StrategyConfig]] = []
+ for cfg in strategy.comparison_presets():
+ df_sig = strategy.annotate_signals(
+ SYMBOL,
+ df_3m.copy(),
+ simulation=True,
+ df_1h=df_1h,
+ df_1d=df_1d,
+ config=cfg,
+ )
+ res = run_backtest(df_sig, df_1d, df_1h, config_name=cfg.name)
+ rows.append((res, cfg))
+
+ rows.sort(key=lambda x: x[0].total_return_pct, reverse=True)
+
+ for rank, (res, cfg) in enumerate(rows, 1):
+ sells = sum(1 for t in res.trades if t.action == "매도")
+ wr = res.win_count / sells * 100 if sells else 0.0
+ print(
+ f"{rank:<4} {res.config_name:<22} {res.total_return_pct:>+8.2f}% "
+ f"{res.final_asset:>12,.0f} {res.trade_count:>6} {wr:>6.1f}% "
+ f"{res.total_fees:>10,.0f}"
+ )
+
+ best_res, best_cfg = rows[0]
+ print("-" * 80)
+ print(f"1위: {best_cfg.name} ({best_res.total_return_pct:+.2f}%)")
+ print(
+ "실거래 적용: strategy.ACTIVE_CONFIG 를 1위 조합으로 맞추세요 "
+ "(현재 ACTIVE_CONFIG.name=%s)" % strategy.ACTIVE_CONFIG.name
+ )
+ print(f"{'='*80}\n")
+
+
+class Simulation:
+ def __init__(self) -> None:
+ self.monitor = Monitor(cooldown_file=None)
+
+ def load_mtf(self, symbol: str):
+ df_1d = self.monitor.get_coin_some_data(symbol, TREND_INTERVAL_1D)
+ df_1h = self.monitor.get_coin_some_data(symbol, TREND_INTERVAL_1H)
+ df_3m = self.monitor.get_coin_some_data(symbol, ENTRY_INTERVAL)
+
+ if df_1d is None or df_1d.empty:
+ df_1d = self.monitor.get_coin_more_data(symbol, TREND_INTERVAL_1D, bong_count=500)
+ if df_1h is None or df_1h.empty:
+ df_1h = self.monitor.get_coin_more_data(symbol, TREND_INTERVAL_1H, bong_count=5000)
+ if df_3m is None or df_3m.empty:
+ df_3m = self.monitor.get_coin_more_data(
+ symbol, ENTRY_INTERVAL, bong_count=90000, verbose=True
+ )
+
+ df_1d = self.monitor.calculate_technical_indicators(df_1d)
+ df_1h = self.monitor.calculate_technical_indicators(df_1h)
+ df_3m = self.monitor.calculate_technical_indicators(df_3m)
+ return df_1d, df_1h, df_3m
+
+ def render_plotly(
+ self,
+ df_plot: pd.DataFrame,
+ trend: str,
+ result: SimResult,
+ interval_min: int = ENTRY_INTERVAL,
+ note: str = "",
+ open_browser: bool = True,
+ ) -> Path:
+ """HTML 리포트 저장 (simulate_mtf.py 동일 스타일)."""
+ html = build_simulation_html(
+ df_plot, result, trend, interval_min=interval_min, note=note
+ )
+ REPORT_DIR.mkdir(parents=True, exist_ok=True)
+ OUTPUT_HTML.write_text(html, encoding="utf-8")
+ print(f"HTML: {OUTPUT_HTML}")
+ if open_browser:
+ webbrowser.open(OUTPUT_HTML.resolve().as_uri())
+ return OUTPUT_HTML
+
+ def load_all_frames(self) -> dict[int, pd.DataFrame]:
+ """discovered 규칙용 전 간격 로드."""
+ from mtf_bb import load_frames_from_db
+
+ return load_frames_from_db(self.monitor, SYMBOL)
+
+ def _run_one_strategy(
+ self,
+ name: str,
+ df_1d: pd.DataFrame,
+ df_1h: pd.DataFrame,
+ df_3m: pd.DataFrame,
+ cfg: strategy.StrategyConfig,
+ frames: dict | None = None,
+ ) -> tuple[pd.DataFrame, SimResult, int]:
+ """한 전략으로 신호·백테스트. 반환: (df, result, 신호수)."""
+ df_sig = strategy.annotate_signals(
+ SYMBOL,
+ df_3m.copy(),
+ simulation=True,
+ df_1h=df_1h,
+ df_1d=df_1d,
+ config=cfg,
+ frames=frames,
+ )
+ n_sig = int((df_sig["point"] == 1).sum())
+ res = run_backtest(df_sig, df_1d, df_1h, config_name=name)
+ return df_sig, res, n_sig
+
+ def _frames_to_mtf(
+ self, frames: dict[int, pd.DataFrame]
+ ) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
+ """전 간격 frames에서 1d/1h/3m 추출."""
+ df_3m = frames.get(ENTRY_INTERVAL)
+ if df_3m is None or df_3m.empty:
+ raise ValueError(f"{ENTRY_INTERVAL}분봉 데이터 없음")
+ df_1d = frames.get(TREND_INTERVAL_1D)
+ if df_1d is None or df_1d.empty:
+ df_1d = df_3m
+ df_1h = frames.get(TREND_INTERVAL_1H)
+ if df_1h is None or df_1h.empty:
+ df_1h = df_3m
+ return df_1d, df_1h, df_3m
+
+ def run_discovered_chart(
+ self,
+ frames: dict[int, pd.DataFrame],
+ rules=None,
+ ) -> SimResult:
+ """
+ discovered_rules 매수·매도 규칙만 백테스트하고 HTML에 표시합니다.
+
+ 차트 마커 = 해당 규칙으로 발생한 매수·매도 체결.
+ """
+ from rule_discovery import DiscoveredRules, load_rules, rules_have_buy
+
+ rule_set = rules or load_rules()
+ if rule_set is None or not rules_have_buy(rule_set):
+ raise FileNotFoundError(
+ "discovered_rules.json 이 없거나 매수 규칙이 비어 있습니다."
+ )
+
+ df_1d, df_1h, df_3m = self._frames_to_mtf(frames)
+ trend = strategy.get_trend(df_1d, df_1h)
+ print(f"추세(최신): {trend}")
+ print(f"3분: {df_3m.index[0]} ~ {df_3m.index[-1]} ({len(df_3m)}봉)")
+ print(f"\n[적용 규칙] {rule_set.name}")
+ print(f" 매수 AND: {rule_set.buy_all}")
+ if rule_set.buy_any:
+ print(f" 매수 OR: {rule_set.buy_any}")
+ print(f" 매도 AND: {rule_set.sell_all}")
+ if rule_set.sell_stop:
+ print(f" 손절: {rule_set.sell_stop}")
+
+ df_sig = strategy.annotate_discovered_signals(
+ SYMBOL, frames, df_1d, df_1h, rules=rule_set, data=df_3m
+ )
+ n_sig = int((df_sig["point"] == 1).sum())
+ buy_sig = int((df_sig["action"] == "buy").sum())
+ sell_sig = int((df_sig["action"] == "sell").sum())
+ print(f"\n규칙 신호: {n_sig} (매수 {buy_sig} / 매도 {sell_sig})")
+
+ result = run_backtest(df_sig, df_1d, df_1h, config_name=rule_set.name)
+ print_backtest_report(result)
+
+ note = (
+ f"매수 규칙: {rule_set.buy_all}"
+ + (f" | OR {rule_set.buy_any}" if rule_set.buy_any else "")
+ + f" | 매도: {rule_set.sell_all}"
+ )
+ self.render_plotly(df_sig, trend, result, note=note)
+ return result
+
+
+def run_mtf_analysis() -> None:
+ """봉별 BB 백테스트 비교, 정책 저장, MTF 시뮬 차트."""
+ from mtf_bb import apply_policy, load_frames_from_db, run_interval_comparison, save_policy
+
+ monitor = Monitor()
+ policy, _ = run_interval_comparison(monitor)
+ save_policy(policy)
+ apply_policy(policy)
+
+ frames = load_frames_from_db(monitor, SYMBOL)
+ df_1d = frames.get(TREND_INTERVAL_1D)
+ if df_1d is None or df_1d.empty:
+ df_1d = frames[ENTRY_INTERVAL]
+ df_1h = frames.get(TREND_INTERVAL_1H)
+ if df_1h is None or df_1h.empty:
+ df_1h = frames[ENTRY_INTERVAL]
+
+ cfg = strategy.StrategyConfig(
+ name="MTF_BB",
+ use_mtf=True,
+ use_regime_switch=strategy.ACTIVE_CONFIG.use_regime_switch,
+ use_rsi_filter=False,
+ use_volume_filter=False,
+ use_squeeze_filter=False,
+ use_stop_loss=True,
+ )
+ df_sig = strategy.annotate_mtf_signals(SYMBOL, frames, df_1d, df_1h, policy, cfg)
+ trend = strategy.get_trend(df_1d, df_1h)
+ print(f"\nMTF 시뮬 ({policy.name}) | 추세: {trend}")
+ result = run_backtest(df_sig, df_1d, df_1h, config_name=policy.name)
+ print_backtest_report(result)
+ Simulation().render_plotly(
+ df_sig,
+ trend,
+ result,
+ interval_min=policy.buy_interval,
+ note=f"MTF 정책: 매수 {policy.buy_interval}분 / 확인 {policy.buy_confirm_intervals}",
+ )
+
+
+def _load_all_frames_or_exit() -> dict[int, pd.DataFrame] | None:
+ """coins.db 전 간격 로드. 부족 시 None."""
+ from rule_discovery import load_frames
+
+ monitor = Monitor(cooldown_file=None)
+ frames = load_frames(monitor)
+ if len(frames) < 3:
+ print("coins.db 데이터 부족. python downloader.py 실행 후 재시도.")
+ return None
+ return frames
+
+
+def run_analyze(frames: dict[int, pd.DataFrame] | None = None) -> None:
+ """전 봉 BB·일목 위치 조합 분석."""
+ from combination_analyzer import analyze_combinations, save_report
+
+ if frames is None:
+ print("=== 전 봉 BB·일목 조합 분석 ===")
+ frames = _load_all_frames_or_exit()
+ if frames is None:
+ return
+ report = analyze_combinations(frames)
+ save_report(report)
+
+
+def run_discover(frames: dict[int, pd.DataFrame] | None = None):
+ """모든 봉·BB·일목 특징으로 최적 규칙 탐색 후 JSON 저장."""
+ from rule_discovery import discover_rules, save_rules
+
+ if frames is None:
+ print("=== 규칙 탐색 (discover) ===")
+ frames = _load_all_frames_or_exit()
+ if frames is None:
+ return None
+ rules = discover_rules(frames)
+ save_rules(rules)
+ print(f"\n저장: discovered_rules.json")
+ return rules
+
+
+def run_full_pipeline() -> None:
+ """
+ 일반 사용자용 일괄 실행: analyze → discover → HTML.
+
+ DB 로드는 한 번만 수행합니다.
+ """
+ print("=" * 60)
+ print("전체 파이프라인: analyze → discover → HTML")
+ print("=" * 60)
+ frames = _load_all_frames_or_exit()
+ if frames is None:
+ return
+
+ print("\n[1/3] 조합 분석 (analyze)")
+ run_analyze(frames)
+
+ print("\n[2/3] 규칙 탐색 (discover)")
+ run_discover(frames)
+
+ print("\n[3/3] 백테스트·HTML 차트 (탐색 규칙 매수·매도)")
+ if rules is None:
+ print("규칙 탐색 실패 — HTML 생략")
+ return
+ Simulation().run_discovered_chart(frames, rules=rules)
+ print("\n완료.")
+
+
+def print_usage() -> None:
+ print(
+ """
+DeepCoin simulation.py
+
+ python simulation.py
+ analyze + discover + HTML (차트 = discovered_rules 매수·매도)
+
+ (고급) analyze | discover | compare | mtf
+"""
+ )
+
+
+def main() -> None:
+ sim = Simulation()
+ if len(sys.argv) > 1 and sys.argv[1] in ("-h", "--help", "help"):
+ print_usage()
+ return
+ if len(sys.argv) == 1 or (len(sys.argv) > 1 and sys.argv[1] in ("all", "chart", "html")):
+ if len(sys.argv) > 1 and sys.argv[1] in ("chart", "html"):
+ print("참고: chart/html 옵션은 제거되었습니다. python simulation.py 만 사용하세요.\n")
+ run_full_pipeline()
+ return
+ if len(sys.argv) > 1 and sys.argv[1] == "analyze":
+ run_analyze()
+ return
+ if len(sys.argv) > 1 and sys.argv[1] == "discover":
+ run_discover()
+ return
+ if len(sys.argv) > 1 and sys.argv[1] == "mtf":
+ run_mtf_analysis()
+ return
+ if len(sys.argv) > 1 and sys.argv[1] == "compare":
+ df_1d, df_1h, df_3m = sim.load_mtf(SYMBOL)
+ run_comparison(df_1d, df_1h, df_3m)
+ return
+ print(f"알 수 없는 옵션: {sys.argv[1]}\n")
+ print_usage()
+
+
+if __name__ == "__main__":
+ main()
diff --git a/simulation_1h.py b/simulation_1h.py
deleted file mode 100644
index cf4c2f4..0000000
--- a/simulation_1h.py
+++ /dev/null
@@ -1,671 +0,0 @@
-"""
-WLD 3분 BB 시뮬레이션.
-
-기본: 하단 상향 돌파 매수, 상단 상향 돌파 매도.
-수수료 반영, 레짐/필터 조합 비교 지원.
-
- python simulation_1h.py # discovered_rules HTML 차트 (기본)
- python simulation_1h.py discover # 모든 봉 특징 탐색 → discovered_rules.json
- python simulation_1h.py compare # 9종 조합 수익률 순위
- python simulation_1h.py mtf # 봉별 BB 비교 + MTF 시뮬
-"""
-
-from __future__ import annotations
-
-import sys
-from dataclasses import dataclass
-
-import pandas as pd
-import plotly.graph_objs as go
-import plotly.io as pio
-from datetime import datetime
-from plotly import subplots
-
-pio.renderers.default = "browser"
-
-from config import (
- BUY_COOLDOWN_SEC,
- COIN_NAME,
- ENTRY_INTERVAL,
- SELL_COOLDOWN_SEC,
- SIM_INITIAL_CASH_KRW,
- SIM_MIN_ORDER_KRW,
- SYMBOL,
- TRADING_FEE_RATE,
- TREND_INTERVAL_1D,
- TREND_INTERVAL_1H,
-)
-from monitor import Monitor
-import strategy
-
-
-@dataclass
-class SimTrade:
- dt: pd.Timestamp
- action: str
- signal: str
- price: float
- krw: float
- fee: float
- quantity: float
- pnl: float | None
- cash_after: float
- total_asset: float
-
-
-@dataclass
-class SimResult:
- config_name: str
- trades: list[SimTrade]
- initial_cash: float
- final_cash: float
- final_coin_qty: float
- final_price: float
- realized_pnl: float
- total_fees: float
- final_asset: float
- total_return_pct: float
- trade_count: int
- win_count: int
-
-
-def run_backtest(
- df_3m: pd.DataFrame,
- df_1d: pd.DataFrame,
- df_1h: pd.DataFrame,
- config_name: str = "",
- initial_cash: float = SIM_INITIAL_CASH_KRW,
- min_order_krw: float = SIM_MIN_ORDER_KRW,
- fee_rate: float = TRADING_FEE_RATE,
-) -> SimResult:
- """신호 순서대로 현물 매수/매도 시뮬레이션 (수수료 차감)."""
- cash = float(initial_cash)
- coin_qty = 0.0
- cost_basis = 0.0
- realized_pnl = 0.0
- total_fees = 0.0
- win_count = 0
- trades: list[SimTrade] = []
- last_buy_ts: pd.Timestamp | None = None
- last_sell_ts: pd.Timestamp | None = None
-
- signals = df_3m[df_3m["point"] == 1].sort_index()
-
- for ts, row in signals.iterrows():
- price = float(row["Close"])
- action = str(row.get("action", ""))
- signal_name = str(row.get("signal", ""))
- if price <= 0:
- continue
-
- trend_at = str(row.get("trend", "")) or strategy.get_trend_at(df_1d, df_1h, ts)
- if trend_at not in ("up", "down", "range"):
- trend_at = strategy.get_trend_at(df_1d, df_1h, ts)
-
- if action == "buy":
- if last_buy_ts is not None:
- if (ts - last_buy_ts).total_seconds() < BUY_COOLDOWN_SEC:
- continue
-
- buy_krw = float(
- strategy.get_buy_amount(SYMBOL, signal_name, price, trend_at)
- )
- buy_krw = max(min_order_krw, min(buy_krw, cash))
- fee = buy_krw * fee_rate
- total_cost = buy_krw + fee
- if buy_krw < min_order_krw or cash < total_cost:
- continue
-
- qty = buy_krw / price
- cash -= total_cost
- total_fees += fee
- cost_basis += buy_krw
- coin_qty += qty
- last_buy_ts = ts
-
- trades.append(
- SimTrade(
- dt=ts,
- action="매수",
- signal=signal_name,
- price=price,
- krw=buy_krw,
- fee=fee,
- quantity=qty,
- pnl=None,
- cash_after=cash,
- total_asset=cash + coin_qty * price,
- )
- )
- continue
-
- if action == "sell":
- if coin_qty <= 0:
- continue
- if last_sell_ts is not None:
- if (ts - last_sell_ts).total_seconds() < SELL_COOLDOWN_SEC:
- continue
-
- ratio = strategy.get_sell_ratio(SYMBOL, signal_name)
- sell_qty = min(coin_qty * ratio, coin_qty)
- sell_krw = sell_qty * price
-
- if sell_krw < min_order_krw:
- if coin_qty * price < min_order_krw:
- continue
- sell_qty = coin_qty
- sell_krw = sell_qty * price
-
- fee = sell_krw * fee_rate
- net = sell_krw - fee
- avg_cost = cost_basis / coin_qty
- sold_cost = avg_cost * sell_qty
- pnl = net - sold_cost
-
- cash += net
- total_fees += fee
- cost_basis -= sold_cost
- coin_qty -= sell_qty
- realized_pnl += pnl
- if pnl > 0:
- win_count += 1
- if coin_qty < 1e-12:
- coin_qty = 0.0
- cost_basis = 0.0
- last_sell_ts = ts
-
- trades.append(
- SimTrade(
- dt=ts,
- action="매도",
- signal=signal_name,
- price=price,
- krw=sell_krw,
- fee=fee,
- quantity=sell_qty,
- pnl=pnl,
- cash_after=cash,
- total_asset=cash + coin_qty * price,
- )
- )
-
- final_price = float(df_3m["Close"].iloc[-1])
- final_asset = cash + coin_qty * final_price
- sell_trades = sum(1 for t in trades if t.action == "매도")
-
- return SimResult(
- config_name=config_name,
- trades=trades,
- initial_cash=initial_cash,
- final_cash=cash,
- final_coin_qty=coin_qty,
- final_price=final_price,
- realized_pnl=realized_pnl,
- total_fees=total_fees,
- final_asset=final_asset,
- total_return_pct=(final_asset - initial_cash) / initial_cash * 100
- if initial_cash > 0
- else 0.0,
- trade_count=len(trades),
- win_count=win_count if sell_trades else 0,
- )
-
-
-def print_backtest_report(result: SimResult) -> None:
- fee_pct = TRADING_FEE_RATE * 100
- print("\n" + "=" * 80)
- print(
- f"[{result.config_name}] 시작 {result.initial_cash:,.0f}원 | "
- f"최소주문 {SIM_MIN_ORDER_KRW:,.0f}원 | 수수료 {fee_pct:.3f}%/쪽"
- )
- print("=" * 80)
- if not result.trades:
- print("체결 없음")
- else:
- print(
- f"{'일시':<18} {'구분':<4} {'신호':<22} {'가격':>9} {'금액':>10} "
- f"{'수수료':>8} {'수익':>10}"
- )
- print("-" * 80)
- for t in result.trades:
- pnl_s = f"{t.pnl:+,.0f}" if t.pnl is not None else "-"
- print(
- f"{t.dt.strftime('%Y-%m-%d %H:%M'):<18} {t.action:<4} {t.signal:<22} "
- f"{t.price:>9,.2f} {t.krw:>10,.0f} {t.fee:>8,.0f} {pnl_s:>10}"
- )
- print("-" * 80)
- sells = sum(1 for t in result.trades if t.action == "매도")
- win_rate = result.win_count / sells * 100 if sells else 0.0
- print(f"거래 횟수: {result.trade_count} (매도 {sells}회) | 승률: {win_rate:.1f}%")
- print(f"수수료 합계: {result.total_fees:,.0f}원")
- print(f"실현 손익(수수료 반영): {result.realized_pnl:+,.0f}원")
- print(
- f"최종 자산: {result.final_asset:,.0f}원 | "
- f"총수익: {result.final_asset - result.initial_cash:+,.0f}원 "
- f"({result.total_return_pct:+.2f}%)"
- )
- print("=" * 80)
-
-
-def run_comparison(df_1d: pd.DataFrame, df_1h: pd.DataFrame, df_3m: pd.DataFrame) -> None:
- """기법 조합별 수익률 비교 (수수료 포함)."""
- print(f"\n{'='*80}")
- print(f"전략 조합 비교 — {SYMBOL} 3분 | {df_3m.index[0]} ~ {df_3m.index[-1]}")
- print(f"시작 {SIM_INITIAL_CASH_KRW:,}원 | 수수료 {TRADING_FEE_RATE*100:.3f}%/매수·매도")
- print(f"{'='*80}")
- print(
- f"{'순위':<4} {'조합':<22} {'수익률':>9} {'최종자산':>12} "
- f"{'거래':>6} {'승률':>7} {'수수료':>10}"
- )
- print("-" * 80)
-
- rows: list[tuple[SimResult, strategy.StrategyConfig]] = []
- for cfg in strategy.comparison_presets():
- df_sig = strategy.annotate_signals(
- SYMBOL,
- df_3m.copy(),
- simulation=True,
- df_1h=df_1h,
- df_1d=df_1d,
- config=cfg,
- )
- res = run_backtest(df_sig, df_1d, df_1h, config_name=cfg.name)
- rows.append((res, cfg))
-
- rows.sort(key=lambda x: x[0].total_return_pct, reverse=True)
-
- for rank, (res, cfg) in enumerate(rows, 1):
- sells = sum(1 for t in res.trades if t.action == "매도")
- wr = res.win_count / sells * 100 if sells else 0.0
- print(
- f"{rank:<4} {res.config_name:<22} {res.total_return_pct:>+8.2f}% "
- f"{res.final_asset:>12,.0f} {res.trade_count:>6} {wr:>6.1f}% "
- f"{res.total_fees:>10,.0f}"
- )
-
- best_res, best_cfg = rows[0]
- print("-" * 80)
- print(f"1위: {best_cfg.name} ({best_res.total_return_pct:+.2f}%)")
- print(
- "실거래 적용: strategy.ACTIVE_CONFIG 를 1위 조합으로 맞추세요 "
- "(현재 ACTIVE_CONFIG.name=%s)" % strategy.ACTIVE_CONFIG.name
- )
- print(f"{'='*80}\n")
-
-
-class Simulation:
- def __init__(self) -> None:
- self.monitor = Monitor(cooldown_file=None)
-
- def load_mtf(self, symbol: str):
- df_1d = self.monitor.get_coin_some_data(symbol, TREND_INTERVAL_1D)
- df_1h = self.monitor.get_coin_some_data(symbol, TREND_INTERVAL_1H)
- df_3m = self.monitor.get_coin_some_data(symbol, ENTRY_INTERVAL)
-
- if df_1d is None or df_1d.empty:
- df_1d = self.monitor.get_coin_more_data(symbol, TREND_INTERVAL_1D, bong_count=500)
- if df_1h is None or df_1h.empty:
- df_1h = self.monitor.get_coin_more_data(symbol, TREND_INTERVAL_1H, bong_count=5000)
- if df_3m is None or df_3m.empty:
- df_3m = self.monitor.get_coin_more_data(
- symbol, ENTRY_INTERVAL, bong_count=90000, verbose=True
- )
-
- df_1d = self.monitor.calculate_technical_indicators(df_1d)
- df_1h = self.monitor.calculate_technical_indicators(df_1h)
- df_3m = self.monitor.calculate_technical_indicators(df_3m)
- return df_1d, df_1h, df_3m
-
- def render_plotly(self, df_3m: pd.DataFrame, trend: str, result: SimResult) -> None:
- cfg = strategy.ACTIVE_CONFIG.name
- summary = (
- f"[{cfg}] 시작 {result.initial_cash:,.0f} | 최종 {result.final_asset:,.0f} | "
- f"{result.total_return_pct:+.2f}% | 수수료 {result.total_fees:,.0f}"
- )
- fig = subplots.make_subplots(
- rows=3,
- cols=1,
- subplot_titles=(
- f"{COIN_NAME} 3분 BB — {trend}",
- "RSI / BB폭(%)",
- summary,
- ),
- shared_xaxes=False,
- vertical_spacing=0.06,
- row_heights=[0.5, 0.18, 0.32],
- specs=[[{"type": "xy"}], [{"type": "xy"}], [{"type": "table"}]],
- )
- fig.add_trace(
- go.Candlestick(
- x=df_3m.index,
- open=df_3m["Open"],
- high=df_3m["High"],
- low=df_3m["Low"],
- close=df_3m["Close"],
- name="캔들",
- showlegend=False,
- ),
- row=1,
- col=1,
- )
- for col, color in [("MA", "blue"), ("Upper", "gray"), ("Lower", "gray")]:
- if col in df_3m.columns:
- fig.add_trace(
- go.Scatter(
- x=df_3m.index,
- y=df_3m[col],
- name=col,
- line=dict(color=color, dash="dot" if col != "MA" else "solid"),
- showlegend=False,
- ),
- row=1,
- col=1,
- )
-
- buy_trades = [t for t in result.trades if t.action == "매수"]
- sell_trades = [t for t in result.trades if t.action == "매도"]
- fig.add_trace(
- go.Scatter(
- x=[t.dt for t in buy_trades],
- y=[t.price for t in buy_trades],
- mode="markers",
- name="매수",
- legendgroup="trades",
- showlegend=True,
- marker=dict(
- color="#22c55e",
- size=11,
- symbol="triangle-up",
- line=dict(width=1, color="#166534"),
- ),
- ),
- row=1,
- col=1,
- )
- fig.add_trace(
- go.Scatter(
- x=[t.dt for t in sell_trades],
- y=[t.price for t in sell_trades],
- mode="markers",
- name="매도",
- legendgroup="trades",
- showlegend=True,
- marker=dict(
- color="#ef4444",
- size=11,
- symbol="triangle-down",
- line=dict(width=1, color="#991b1b"),
- ),
- ),
- row=1,
- col=1,
- )
- if "RSI" in df_3m.columns:
- fig.add_trace(
- go.Scatter(
- x=df_3m.index,
- y=df_3m["RSI"],
- name="RSI",
- showlegend=False,
- ),
- row=2,
- col=1,
- )
- if "BB_Width" in df_3m.columns:
- fig.add_trace(
- go.Scatter(
- x=df_3m.index,
- y=df_3m["BB_Width"],
- name="BB폭%",
- showlegend=False,
- ),
- row=2,
- col=1,
- )
- if result.trades:
- cells = [
- [t.dt.strftime("%Y-%m-%d %H:%M") for t in result.trades],
- [t.action for t in result.trades],
- [t.signal for t in result.trades],
- [f"{t.price:,.2f}" for t in result.trades],
- [f"{t.krw:,.0f}" for t in result.trades],
- [f"{t.fee:,.0f}" for t in result.trades],
- [f"{t.pnl:+,.0f}" if t.pnl is not None else "-" for t in result.trades],
- [f"{t.total_asset:,.0f}" for t in result.trades],
- ]
- else:
- cells = [["-"] * 8]
- fig.add_trace(
- go.Table(
- header=dict(
- values=[
- "일시",
- "구분",
- "신호",
- "가격",
- "금액",
- "수수료",
- "수익",
- "총자산",
- ],
- fill_color="#e8e8e8",
- ),
- cells=dict(values=cells),
- ),
- row=3,
- col=1,
- )
- fig.update_layout(
- height=1100,
- title=f"{SYMBOL} BB 타이밍 시뮬 (범례 클릭: 매수/매도 표시 토글)",
- margin=dict(l=50, r=140, t=80, b=40),
- dragmode="zoom",
- legend=dict(
- orientation="v",
- yanchor="top",
- y=0.99,
- xanchor="left",
- x=1.01,
- bgcolor="rgba(255,255,255,0.9)",
- bordercolor="#cccccc",
- borderwidth=1,
- font=dict(size=12),
- title=dict(text="체결 (클릭 토글)", side="top"),
- itemclick="toggle",
- itemdoubleclick="toggleothers",
- ),
- )
- # Y축 고정·rangeslider 해제 → 세로 드래그/박스줌·휠 줌 가능
- fig.update_xaxes(
- rangeslider_visible=False,
- fixedrange=False,
- row=1,
- col=1,
- )
- fig.update_xaxes(fixedrange=False, row=2, col=1)
- fig.update_yaxes(
- title_text="가격 (KRW)",
- fixedrange=False,
- scaleanchor=None,
- scaleratio=None,
- row=1,
- col=1,
- )
- fig.update_yaxes(
- fixedrange=False,
- scaleanchor=None,
- scaleratio=None,
- row=2,
- col=1,
- )
- fig.show(
- config={
- "scrollZoom": True,
- "displaylogo": False,
- "doubleClick": "reset",
- "modeBarButtonsToAdd": ["zoom2d", "pan2d", "resetScale2d"],
- }
- )
-
- def load_all_frames(self) -> dict[int, pd.DataFrame]:
- """discovered 규칙용 전 간격 로드."""
- from mtf_bb import load_frames_from_db
-
- return load_frames_from_db(self.monitor, SYMBOL)
-
- def _run_one_strategy(
- self,
- name: str,
- df_1d: pd.DataFrame,
- df_1h: pd.DataFrame,
- df_3m: pd.DataFrame,
- cfg: strategy.StrategyConfig,
- frames: dict | None = None,
- ) -> tuple[pd.DataFrame, SimResult, int]:
- """한 전략으로 신호·백테스트. 반환: (df, result, 신호수)."""
- df_sig = strategy.annotate_signals(
- SYMBOL,
- df_3m.copy(),
- simulation=True,
- df_1h=df_1h,
- df_1d=df_1d,
- config=cfg,
- frames=frames,
- )
- n_sig = int((df_sig["point"] == 1).sum())
- res = run_backtest(df_sig, df_1d, df_1h, config_name=name)
- return df_sig, res, n_sig
-
- def run(self, config: strategy.StrategyConfig | None = None) -> SimResult:
- """기본 BB vs 탐색 규칙 중 수익률·신호가 있는 쪽을 HTML에 표시."""
- df_1d, df_1h, df_3m = self.load_mtf(SYMBOL)
- trend = strategy.get_trend(df_1d, df_1h)
- print(f"추세(최신): {trend}")
- print(f"3분: {df_3m.index[0]} ~ {df_3m.index[-1]} ({len(df_3m)}봉)")
-
- cfg_base = strategy.StrategyConfig(
- name="01_기본_BB만",
- use_discovered_rules=False,
- use_regime_switch=False,
- use_rsi_filter=False,
- use_volume_filter=False,
- use_squeeze_filter=False,
- use_stop_loss=False,
- )
- df_base, res_base, n_base = self._run_one_strategy(
- cfg_base.name, df_1d, df_1h, df_3m, cfg_base
- )
- print(f"\n[기본 BB] 신호 {n_base} | 수익 {res_base.total_return_pct:+.2f}% | 거래 {res_base.trade_count}")
-
- candidates: list[tuple[str, pd.DataFrame, SimResult, int]] = [
- (cfg_base.name, df_base, res_base, n_base),
- ]
-
- try:
- from rule_discovery import load_rules
-
- rules = load_rules()
- frames = self.load_all_frames()
- if rules and frames:
- cfg_disc = strategy.StrategyConfig(
- name=rules.name,
- use_discovered_rules=True,
- use_regime_switch=False,
- use_rsi_filter=False,
- use_volume_filter=False,
- use_squeeze_filter=False,
- use_stop_loss=False,
- )
- df_disc, res_disc, n_disc = self._run_one_strategy(
- cfg_disc.name, df_1d, df_1h, df_3m, cfg_disc, frames=frames
- )
- print(
- f"[탐색 규칙] 신호 {n_disc} | 수익 {res_disc.total_return_pct:+.2f}% "
- f"| 거래 {res_disc.trade_count}"
- )
- print(f" 매수: {rules.buy_all} | OR: {rules.buy_any}")
- print(f" 매도: {rules.sell_all} | 손절: {rules.sell_stop}")
- if n_disc > 0 and res_disc.trade_count > 0:
- candidates.append((cfg_disc.name, df_disc, res_disc, n_disc))
- except Exception as e:
- print(f"[탐색 규칙] 스킵: {e}")
-
- # 신호·거래 있는 후보 중 수익률 최대
- valid = [c for c in candidates if c[3] > 0 and c[2].trade_count > 0]
- if not valid:
- valid = candidates
- name, df_plot, result, n_sig = max(valid, key=lambda c: c[2].total_return_pct)
-
- print(f"\n>>> HTML 적용: {name} (신호 {n_sig}, 거래 {result.trade_count}, {result.total_return_pct:+.2f}%)")
- sigs = df_plot[df_plot["point"] == 1]
- if len(sigs):
- print(sigs["action"].value_counts().to_string())
-
- print_backtest_report(result)
- self.render_plotly(df_plot, trend, result)
- return result
-
-
-def run_mtf_analysis() -> None:
- """봉별 BB 백테스트 비교, 정책 저장, MTF 시뮬 차트."""
- from mtf_bb import apply_policy, load_frames_from_db, run_interval_comparison, save_policy
-
- monitor = Monitor()
- policy, _ = run_interval_comparison(monitor)
- save_policy(policy)
- apply_policy(policy)
-
- frames = load_frames_from_db(monitor, SYMBOL)
- df_1d = frames.get(TREND_INTERVAL_1D)
- if df_1d is None or df_1d.empty:
- df_1d = frames[ENTRY_INTERVAL]
- df_1h = frames.get(TREND_INTERVAL_1H)
- if df_1h is None or df_1h.empty:
- df_1h = frames[ENTRY_INTERVAL]
-
- cfg = strategy.StrategyConfig(
- name="MTF_BB",
- use_mtf=True,
- use_regime_switch=strategy.ACTIVE_CONFIG.use_regime_switch,
- use_rsi_filter=False,
- use_volume_filter=False,
- use_squeeze_filter=False,
- use_stop_loss=True,
- )
- df_sig = strategy.annotate_mtf_signals(SYMBOL, frames, df_1d, df_1h, policy, cfg)
- trend = strategy.get_trend(df_1d, df_1h)
- print(f"\nMTF 시뮬 ({policy.name}) | 추세: {trend}")
- result = run_backtest(df_sig, df_1d, df_1h, config_name=policy.name)
- print_backtest_report(result)
- Simulation().render_plotly(df_sig, trend, result)
-
-
-def run_discover() -> None:
- """모든 봉·캔들 특징으로 최적 규칙 탐색 후 JSON 저장."""
- from rule_discovery import discover_rules, load_frames, save_rules
-
- monitor = Monitor(cooldown_file=None)
- frames = load_frames(monitor)
- rules = discover_rules(frames)
- save_rules(rules)
- print(f"\n저장: discovered_rules.json")
- print("HTML 차트: python simulation_1h.py")
-
-
-def main() -> None:
- sim = Simulation()
- if len(sys.argv) > 1 and sys.argv[1] == "discover":
- run_discover()
- return
- if len(sys.argv) > 1 and sys.argv[1] == "mtf":
- run_mtf_analysis()
- return
- df_1d, df_1h, df_3m = sim.load_mtf(SYMBOL)
- if len(sys.argv) > 1 and sys.argv[1] == "compare":
- run_comparison(df_1d, df_1h, df_3m)
- return
- sim.run()
-
-
-if __name__ == "__main__":
- main()
diff --git a/strategy.py b/strategy.py
index 152109d..bb28f56 100644
--- a/strategy.py
+++ b/strategy.py
@@ -92,7 +92,7 @@ class StrategyConfig:
use_discovered_rules: bool = False
-# HTML 시뮬: discovered_rules.json (python simulation_1h.py discover)
+# HTML 시뮬: discovered_rules.json (python simulation.py discover)
ACTIVE_CONFIG = StrategyConfig(
name="discovered_best",
use_discovered_rules=True,
@@ -478,6 +478,44 @@ def annotate_mtf_signals(
return df
+def evaluate_discovered_live(
+ symbol: str,
+ frames: dict[int, pd.DataFrame],
+ df_1d: pd.DataFrame,
+ df_1h: pd.DataFrame,
+ balances: dict,
+) -> TradeSignal | None:
+ """
+ 최신 3분 봉 시점에서 discovered_rules + 전 봉 BB·일목 조합으로 신호 1건.
+ """
+ from candle_features import build_master_feature_matrix
+ from rule_discovery import buy_mask, load_rules, rules_have_buy, sell_mask
+
+ rules = load_rules()
+ if rules is None or not rules_have_buy(rules):
+ return None
+
+ matrix = build_master_feature_matrix(frames)
+ if len(matrix) < 22:
+ return None
+ last = matrix.iloc[[-1]]
+ ts = last.index[-1]
+ close = float(last["Close"].iloc[-1])
+ trend = get_trend_at(df_1d, df_1h, ts)
+
+ position = float(balances.get(symbol, {}).get("balance", 0) or 0)
+ if position >= 1.0:
+ if rules.sell_stop and sell_mask(last, rules, stop=True)[0]:
+ return TradeSignal("sell", SIGNAL_SELL_STOP, close, trend)
+ if sell_mask(last, rules, stop=False)[0]:
+ return TradeSignal("sell", SIGNAL_SELL_UPPER, close, trend)
+ return None
+
+ if buy_mask(last, rules)[0]:
+ return TradeSignal("buy", SIGNAL_BUY_LOWER, close, trend)
+ return None
+
+
def annotate_discovered_signals(
symbol: str,
frames: dict[int, pd.DataFrame],
@@ -494,7 +532,7 @@ def annotate_discovered_signals(
if rule_set is None or not rules_have_buy(rule_set):
raise FileNotFoundError(
"discovered_rules.json 없거나 매수 규칙이 비어 있습니다. "
- "python simulation_1h.py discover 실행"
+ "python simulation.py 실행"
)
entry = frames.get(ENTRY_INTERVAL)