""" 봉 간격별 볼린저밴드 상태 분석 및 최적 매수/매도 봉 추천. 기본 규칙(모든 봉 동일): - 매수: 하단 밴드 상향 돌파 - 매도: 상단 밴드 상향 돌파 - 손절(선택): 하단 재이탈 """ from __future__ import annotations import json from dataclasses import asdict, dataclass from pathlib import Path import pandas as pd from config import DOWNLOAD_INTERVALS, SYMBOL from strategy import ( ACTIVE_CONFIG, StrategyConfig, annotate_interval_signals, get_latest_bb_state, MtfBbPolicy, ) POLICY_FILE = Path(__file__).parent / "mtf_bb_policy.json" # 매수/매도 트리거 실행 후보 (긴 봉은 확인용만) EXECUTION_INTERVAL_CANDIDATES: tuple[int, ...] = (3, 10, 15, 30, 60) DEFAULT_CONFIRM_INTERVALS: tuple[int, ...] = (60, 1440) @dataclass class IntervalBacktestSummary: """봉 간격별 백테스트 요약.""" interval: int label: str return_pct: float trade_count: int buy_count: int sell_count: int final_asset: float def interval_label(interval: int) -> str: if interval >= 1440: return "일봉" return f"{interval}분" def load_frames_from_db(monitor, symbol: str) -> dict[int, pd.DataFrame]: """coins.db에서 DOWNLOAD_INTERVALS 전부 로드·지표 계산.""" frames: dict[int, pd.DataFrame] = {} for iv in DOWNLOAD_INTERVALS: df = monitor.get_coin_some_data(symbol, iv) if df is None or df.empty: print(f" [{interval_label(iv)}] DB/API 데이터 없음 — 스킵") continue df = monitor.calculate_technical_indicators(df) frames[iv] = df print(f" [{interval_label(iv)}] {len(df)}봉 {df.index[0]} ~ {df.index[-1]}") return frames def print_latest_states(frames: dict[int, pd.DataFrame], cfg: StrategyConfig) -> None: """각 봉의 최신 BB 상태 출력.""" print("\n--- 봉별 최신 BB 상태 ---") for iv in sorted(frames.keys()): st = get_latest_bb_state(frames[iv], cfg) df = frames[iv] close = df["Close"].iloc[-1] print( f" {interval_label(iv):>6} ({iv:>4}분) | {st:20} | 종가 {close:,.2f} | " f"L={df['Lower'].iloc[-1]:,.2f} U={df['Upper'].iloc[-1]:,.2f}" ) def backtest_interval( frames: dict[int, pd.DataFrame], interval: int, cfg: StrategyConfig, ) -> IntervalBacktestSummary | None: """한 간격만으로 BB 매매 백테스트.""" from simulation_1h import run_backtest if interval not in frames: return None df = annotate_interval_signals(SYMBOL, frames[interval].copy(), config=cfg) # 단일 봉 비교이므로 추세 필터 없이 동일 df를 HTF로 전달 res = run_backtest(df, frames[interval], frames[interval], config_name=f"{interval}분") buys = sum(1 for t in res.trades if t.action == "매수") sells = sum(1 for t in res.trades if t.action == "매도") return IntervalBacktestSummary( interval=interval, label=interval_label(interval), return_pct=res.total_return_pct, trade_count=res.trade_count, buy_count=buys, sell_count=sells, final_asset=res.final_asset, ) def recommend_policy( summaries: list[IntervalBacktestSummary], frames: dict[int, pd.DataFrame], ) -> MtfBbPolicy: """ 백테스트 결과로 매수/매도 실행 봉과 확인용 상위 봉을 추천합니다. - 매수/매도 실행: 수익률 1위 간격 (동일하면 더 긴 봉 우선) - 확인 봉: 실행 봉보다 긴 간격 중 가장 가까운 2개 """ if not summaries: return MtfBbPolicy() exec_pool = [s for s in summaries if s.interval in EXECUTION_INTERVAL_CANDIDATES] if not exec_pool: exec_pool = list(summaries) ranked = sorted( exec_pool, key=lambda s: (s.return_pct, s.trade_count, -s.interval), reverse=True, ) best = ranked[0] buy_iv = sell_iv = best.interval if best.trade_count == 0: buy_iv = sell_iv = min( (iv for iv in frames if iv in EXECUTION_INTERVAL_CANDIDATES), default=min(frames.keys()), ) longer = sorted([iv for iv in frames if iv > buy_iv]) confirm = tuple(iv for iv in DEFAULT_CONFIRM_INTERVALS if iv in frames and iv > buy_iv) if not confirm: confirm = tuple(longer[-2:]) if len(longer) >= 2 else tuple(longer) if not confirm and len(longer) == 1: confirm = (longer[0],) return MtfBbPolicy( buy_interval=buy_iv, sell_interval=sell_iv, buy_confirm_intervals=confirm, sell_confirm_intervals=confirm[:1] if confirm else (), name=f"auto_{interval_label(buy_iv)}_buy_{interval_label(sell_iv)}_sell", ) def run_interval_comparison(monitor) -> tuple[MtfBbPolicy, list[IntervalBacktestSummary]]: """모든 봉 간격 BB 백테스트 후 정책 추천.""" cfg = StrategyConfig( name="봉별_BB_기본", use_mtf=False, use_regime_switch=False, use_rsi_filter=False, use_volume_filter=False, use_squeeze_filter=False, use_stop_loss=True, ) print(f"\n{'='*72}") print("봉 간격별 BB 매매 비교 (하단↑매수 / 상단↑매도 / 수수료 반영)") print(f"{'='*72}") frames = load_frames_from_db(monitor, SYMBOL) if not frames: raise RuntimeError("로드된 봉 데이터가 없습니다. downloader.py 먼저 실행하세요.") print_latest_states(frames, cfg) summaries: list[IntervalBacktestSummary] = [] for iv in sorted(frames.keys()): s = backtest_interval(frames, iv, cfg) if s: summaries.append(s) summaries.sort(key=lambda x: x.return_pct, reverse=True) print(f"\n{'순위':<4} {'봉':>8} {'수익률':>9} {'거래':>6} {'매수':>5} {'매도':>5}") print("-" * 45) for i, s in enumerate(summaries, 1): print( f"{i:<4} {s.label:>8} {s.return_pct:>+8.2f}% " f"{s.trade_count:>6} {s.buy_count:>5} {s.sell_count:>5}" ) policy = recommend_policy(summaries, frames) print(f"\n추천 정책:") print(f" 매수 실행 봉: {interval_label(policy.buy_interval)}") print(f" 매도 실행 봉: {interval_label(policy.sell_interval)}") print(f" 매수 확인 봉: {[interval_label(i) for i in policy.buy_confirm_intervals]}") print(f" 매도 확인 봉: {[interval_label(i) for i in policy.sell_confirm_intervals]}") print(f"{'='*72}\n") return policy, summaries def save_policy(policy: MtfBbPolicy, path: Path = POLICY_FILE) -> None: """추천 정책을 JSON으로 저장.""" path.write_text(json.dumps(asdict(policy), ensure_ascii=False, indent=2), encoding="utf-8") def load_policy(path: Path = POLICY_FILE) -> MtfBbPolicy | None: """저장된 정책 로드.""" if not path.exists(): return None data = json.loads(path.read_text(encoding="utf-8")) return MtfBbPolicy( buy_interval=int(data["buy_interval"]), sell_interval=int(data["sell_interval"]), buy_confirm_intervals=tuple(data.get("buy_confirm_intervals", [])), sell_confirm_intervals=tuple(data.get("sell_confirm_intervals", [])), name=data.get("name", "loaded"), ) def apply_policy(policy: MtfBbPolicy) -> None: """strategy.ACTIVE_MTF_POLICY 에 반영.""" import strategy as st st.ACTIVE_MTF_POLICY = policy print(f"ACTIVE_MTF_POLICY 적용: {policy.name}")