From e218a8ea32421e3341ea8d161f18889a76f84cfd Mon Sep 17 00:00:00 2001 From: dsyoon Date: Fri, 29 May 2026 01:20:36 +0900 Subject: [PATCH] =?UTF-8?q?=EC=A0=84=20=EB=B4=89=20BB=C2=B7=EC=9D=BC?= =?UTF-8?q?=EB=AA=A9=20=EC=A1=B0=ED=95=A9=20=EB=B6=84=EC=84=9D=20=EB=B0=8F?= =?UTF-8?q?=20simulation=20=EB=8B=A8=EC=9D=BC=20=EC=8B=A4=ED=96=89?= =?UTF-8?q?=EC=9C=BC=EB=A1=9C=20=ED=86=B5=ED=95=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 9개 간격(1~1440분) BB·일목 위치 특징을 3분 타임라인에 맞춰 분석하고, discover로 매수·매도 규칙을 찾은 뒤 HTML 차트에 해당 체결만 표시한다. simulation_1h.py를 simulation.py로 변경했으며, 파라미터 없이 실행하면 analyze→discover→차트가 한 번에 수행된다. Co-authored-by: Cursor --- .env.example | 4 +- README.md | 84 ++-- candle_features.py | 212 +++++++--- combination_analyzer.py | 222 +++++++++++ config.py | 12 +- discovered_rules.json | 21 +- downloader.py | 11 + indicators.py | 58 +++ monitor.py | 57 ++- monitor_coin.py | 4 +- mtf_bb.py | 2 +- rule_discovery.py | 58 ++- simulation.py | 855 ++++++++++++++++++++++++++++++++++++++++ simulation_1h.py | 671 ------------------------------- strategy.py | 42 +- 15 files changed, 1510 insertions(+), 803 deletions(-) create mode 100644 combination_analyzer.py create mode 100644 indicators.py create mode 100644 simulation.py delete mode 100644 simulation_1h.py 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)