From 16b4c932a266882112c62b738e2dae283f29c76f Mon Sep 17 00:00:00 2001 From: joungmin Date: Sun, 1 Mar 2026 10:51:02 +0900 Subject: [PATCH] feat: bear block, trend-continuation entry, partial TP backtest MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. daemon/runner.py: skip scan entirely in bear regime - calls get_regime() at start of each scan loop - logs bear block with score before sleeping 2. core/strategy.py: trend-continuation entry filter - check_trend_6h(): 6h price trend >= 1% (rejects flash spikes) - 15-min confirmation watchlist (_watchlist dict) - should_buy() adds watchlist to existing 12h+regime+momentum logic - CONFIRM_SECONDS env var (default 900 = 15min) - TREND_6H_MIN_PCT env var (default 1.0%) 3. backtest.py: partial take-profit scenario comparison (--tp-cmp) - simulate(): partial_tp_pct / partial_tp_ratio params - blended pnl = ratio * partial_pnl + (1-ratio) * remaining_pnl - main_tp_cmp(): 3 scenarios A/B/C (none / +5% 50% / +3% 50%) - result: partial TP reduces cumulative return (-56% → -63%) big winners carry the strategy; trimming them hurts expected value Co-Authored-By: Claude Sonnet 4.6 --- backtest.py | 125 ++++++++++++++++++++++++++++++++++++++++++----- core/strategy.py | 84 +++++++++++++++++++++++++++++-- daemon/runner.py | 11 +++++ 3 files changed, 204 insertions(+), 16 deletions(-) diff --git a/backtest.py b/backtest.py index edf287b..290ff29 100644 --- a/backtest.py +++ b/backtest.py @@ -12,6 +12,7 @@ python backtest.py --top50-cmp # 종목수(20vs50) + 거래대금 급증 비교 (2x/3x/4x) python backtest.py --trend-cmp # 추세 상한선 비교 (+5/7/10/15% 상한) python backtest.py --ticker-cmp # 종목 승률 필터 비교 (SQL 즉시 분석) + python backtest.py --tp-cmp # 부분 익절 구조 비교 (없음 / +5% 50% / +3% 50%) 캐시: Oracle ADB backtest_ohlcv / backtest_daily / backtest_trades 테이블 시뮬레이션 결과는 backtest_trades에 파라미터 해시로 저장 — 동일 파라미터 재실행 시 즉시 반환 @@ -72,6 +73,7 @@ COMPARE_TICKER = "--ticker-cmp" in sys.argv # 종목 승률 필터 비 COMPARE_WF = "--walkforward-cmp" in sys.argv # walk-forward 필터 비교 모드 COMPARE_TIMESTOP = "--timestop-cmp" in sys.argv # 타임스탑 조건 비교 모드 COMPARE_COMBO = "--combo-cmp" in sys.argv # 추세+거래량 조합 비교 모드 +COMPARE_TP = "--tp-cmp" in sys.argv # 부분 익절 구조 비교 모드 if "--10m" in sys.argv: DEFAULT_INTERVAL = "minute10" @@ -578,15 +580,19 @@ def simulate( wf_window: int = 5, # walk-forward: 승률 계산 윈도우 크기 time_stop_h: float = TIME_STOP_H, # 타임스탑 기준 시간 (0=비활성) time_stop_gain: float = TIME_STOP_GAIN, # 타임스탑 최소 수익률 + partial_tp_pct: float = 0.0, # 부분 익절 트리거 수익률 (0=비활성) + partial_tp_ratio: float = 0.5, # 부분 익절 비율 (기본 50%) ) -> list[dict]: """단일 종목 전략 시뮬레이션. - stop_loss : 트레일링 스탑 (최고가 대비 하락률) - hard_stop : 하드 스탑 (매수가 대비 하락률). None이면 stop_loss 값 사용. - volume_mult : 진입 조건 — 전일 거래량 > N일 평균 × volume_mult - wf_min_wr : walk-forward 필터 — 직전 wf_window건 승률이 이 값 미만이면 진입 차단 - 윈도우가 채워지기 전(워밍업)에는 필터 미적용 - time_stop_h : 타임스탑 기준 시간 (0 이하 = 비활성) + stop_loss : 트레일링 스탑 (최고가 대비 하락률) + hard_stop : 하드 스탑 (매수가 대비 하락률). None이면 stop_loss 값 사용. + volume_mult : 진입 조건 — 전일 거래량 > N일 평균 × volume_mult + wf_min_wr : walk-forward 필터 — 직전 wf_window건 승률이 이 값 미만이면 진입 차단 + 윈도우가 채워지기 전(워밍업)에는 필터 미적용 + time_stop_h : 타임스탑 기준 시간 (0 이하 = 비활성) + partial_tp_pct : 이 수익률 도달 시 partial_tp_ratio 만큼 익절 (0=비활성) + partial_tp_ratio: 부분 익절 비율 (0~1). 나머지는 트레일링 스탑으로 유지 """ _hard_stop = hard_stop if hard_stop is not None else stop_loss trades: list[dict] = [] @@ -614,6 +620,14 @@ def simulate( pnl = (current - bp) / bp elapsed = (t - pos["entry"]).total_seconds() / 3600 + # ── 부분 익절 트리거 ────────────────────────────────────────────── + if partial_tp_pct > 0 and not pos.get("partial_done") and pnl >= partial_tp_pct: + pos["partial_done"] = True + pos["partial_price"] = current + pos["partial_pnl"] = ( + current * (1 - FEE) - bp * (1 + FEE) + ) / (bp * (1 + FEE)) * 100 + reason = None if drop_pk >= stop_loss: reason = "trailing_stop" @@ -623,9 +637,20 @@ def simulate( reason = "time_stop" if reason: - net_pnl = ( + # 나머지 포지션 청산 pnl + rem_pnl = ( current * (1 - FEE) - bp * (1 + FEE) ) / (bp * (1 + FEE)) * 100 + + # 부분 익절이 실행됐으면 가중 평균 pnl 계산 + if partial_tp_pct > 0 and pos.get("partial_done"): + net_pnl = ( + partial_tp_ratio * pos["partial_pnl"] + + (1 - partial_tp_ratio) * rem_pnl + ) + else: + net_pnl = rem_pnl + trades.append({ "ticker": ticker, "entry": pos["entry"], @@ -686,20 +711,35 @@ def simulate( continue bp = current * (1 + FEE) - pos = {"buy_price": bp, "peak": current, "entry": t} + pos = { + "buy_price": bp, + "peak": current, + "entry": t, + "partial_done": False, + "partial_price": None, + "partial_pnl": None, + } # 기간 종료 시 미청산 포지션 강제 종료 if pos is not None: current = closes[-1] elapsed = (times[-1] - pos["entry"]).total_seconds() / 3600 - net_pnl = ( - current * (1 - FEE) - pos["buy_price"] * (1 + FEE) - ) / (pos["buy_price"] * (1 + FEE)) * 100 + bp = pos["buy_price"] + rem_pnl = ( + current * (1 - FEE) - bp * (1 + FEE) + ) / (bp * (1 + FEE)) * 100 + if partial_tp_pct > 0 and pos.get("partial_done"): + net_pnl = ( + partial_tp_ratio * pos["partial_pnl"] + + (1 - partial_tp_ratio) * rem_pnl + ) + else: + net_pnl = rem_pnl trades.append({ "ticker": ticker, "entry": pos["entry"], "exit": times[-1], - "buy_price": round(pos["buy_price"], 4), + "buy_price": round(bp, 4), "sell_price": round(current, 4), "pnl_pct": round(net_pnl, 3), "elapsed_h": round(elapsed, 1), @@ -858,6 +898,8 @@ def run_scenario( wf_window: int = 5, # walk-forward 윈도우 time_stop_h: float = TIME_STOP_H, # 타임스탑 기준 시간 (0=비활성) time_stop_gain: float = TIME_STOP_GAIN, # 타임스탑 최소 수익률 + partial_tp_pct: float = 0.0, # 부분 익절 트리거 수익률 (0=비활성) + partial_tp_ratio: float = 0.5, # 부분 익절 비율 ) -> list[dict]: """주어진 파라미터로 전체 종목 시뮬레이션 (데이터는 외부에서 주입). @@ -905,6 +947,8 @@ def run_scenario( wf_window=wf_window, time_stop_h=time_stop_h, time_stop_gain=time_stop_gain, + partial_tp_pct=partial_tp_pct, + partial_tp_ratio=partial_tp_ratio, ) all_trades.extend(trades) @@ -1672,8 +1716,63 @@ def main_combo_cmp(interval: str = DEFAULT_INTERVAL) -> None: report(trades, clabel) +def main_tp_cmp(interval: str = DEFAULT_INTERVAL) -> None: + """부분 익절 구조 비교: A(없음) vs B(+5% 50% 익절) vs C(+3% 50% 익절). + + 트레일링 스탑·타임스탑 파라미터는 현재 봇 운영 기준과 동일하게 고정. + """ + cfg = INTERVAL_CONFIG[interval] + label = cfg["label"] + trend = cfg["trend"] + + print(f"\n{'='*60}") + print(f" 부분 익절 구조 비교 백테스트 | {MONTHS}개월 | 상위 {TOP_N}종목 | {label}") + print(f" (트레일링 1.5% | 타임스탑 8h/3% | 모멘텀 {trend*100:.1f}%)") + print(f"{'='*60}") + + print("\n▶ 종목 데이터 로드 중 (DB 캐시 우선)...") + tickers = get_top_tickers(TOP_N) + print(f" → {tickers}") + data = fetch_all_data(tickers, interval) + print(f" 사용 종목: {list(data.keys())}") + daily_features = {t: build_daily_features(f["daily"]) for t, f in data.items()} + + common = dict( + stop_loss=0.015, + trend_min_gain=trend, + interval_cd=interval, + daily_features=daily_features, + time_stop_h=8.0, + time_stop_gain=0.03, + ) + + print("\n▶ A: 부분 익절 없음 (현행) ...") + trades_a = run_scenario(data, partial_tp_pct=0.0, **common) + print(f" 완료 ({len(trades_a)}건)") + + print("\n▶ B: +5% 도달 시 50% 익절 ...") + trades_b = run_scenario(data, partial_tp_pct=0.05, partial_tp_ratio=0.5, **common) + print(f" 완료 ({len(trades_b)}건)") + + print("\n▶ C: +3% 도달 시 50% 익절 ...") + trades_c = run_scenario(data, partial_tp_pct=0.03, partial_tp_ratio=0.5, **common) + print(f" 완료 ({len(trades_c)}건)") + + compare_report([ + ("A: 익절없음", trades_a), + ("B: +5%→50%익절", trades_b), + ("C: +3%→50%익절", trades_c), + ], title=f"부분 익절 구조 비교 ({label})") + + report(trades_a, f"A: 부분 익절 없음 | {label}") + report(trades_b, f"B: +5% 50% 익절 | {label}") + report(trades_c, f"C: +3% 50% 익절 | {label}") + + def main() -> None: - if COMPARE_COMBO: + if COMPARE_TP: + main_tp_cmp(DEFAULT_INTERVAL) + elif COMPARE_COMBO: main_combo_cmp(DEFAULT_INTERVAL) elif COMPARE_TIMESTOP: main_timestop_cmp(DEFAULT_INTERVAL) diff --git a/core/strategy.py b/core/strategy.py index 67fd822..ad4aa4a 100644 --- a/core/strategy.py +++ b/core/strategy.py @@ -1,9 +1,15 @@ -"""Strategy C: 현재 기준 N시간 전 대비 상승 추세(DB) AND 거래량 모멘텀 동시 충족 시 매수 신호.""" +"""Strategy C: 현재 기준 N시간 전 대비 상승 추세(DB) AND 거래량 모멘텀 동시 충족 시 매수 신호. + +추가 필터: + 1. 6h 추세 확인 (단기 급등 아닌 지속 추세) + 2. 15분 확인 워치리스트 (신호 첫 발생 후 15분 내 재확인 시 진입) +""" from __future__ import annotations import logging import os +import time import pyupbit from .market import get_current_price, get_ohlcv @@ -16,11 +22,18 @@ logger = logging.getLogger(__name__) TREND_HOURS = float(os.getenv("TREND_HOURS", "12")) TREND_MIN_GAIN_PCT = float(os.getenv("TREND_MIN_GAIN_PCT", "5")) # 레짐이 없을 때 기본값 +# 6h 단기 추세 최소 상승률 (추세 지속형 필터) +TREND_6H_MIN_PCT = float(os.getenv("TREND_6H_MIN_PCT", "1.0")) + # 모멘텀: MA 기간, 거래량 급증 배수 MA_PERIOD = 20 VOLUME_MULTIPLIER = float(os.getenv("VOLUME_MULTIPLIER", "2.0")) # 레짐이 없을 때 기본값 LOCAL_VOL_HOURS = 5 # 로컬 기준 시간 (h) +# 15분 확인 워치리스트: 신호 첫 발생 시각(unix ts) 기록 +CONFIRM_SECONDS = int(os.getenv("CONFIRM_SECONDS", "900")) # 기본 15분 +_watchlist: dict[str, float] = {} # ticker → first_signal_time (unix timestamp) + def check_trend(ticker: str, min_gain_pct: float) -> bool: """상승 추세 조건: 현재가가 DB에 저장된 N시간 전 가격 대비 +min_gain_pct% 이상.""" @@ -96,12 +109,77 @@ def check_momentum(ticker: str, vol_mult: float) -> bool: return vol_ok +def check_trend_6h(ticker: str) -> bool: + """6h 추세 지속 확인: 6h 전 대비 +TREND_6H_MIN_PCT% 이상 상승 중이어야 진입 허용.""" + past = get_price_n_hours_ago(ticker, 6) + if past is None: + logger.debug(f"[6h추세] {ticker} 6h 전 가격 없음 (수집 중)") + return True # 데이터 없으면 필터 패스 (수집 초기) + + current = get_current_price(ticker) + if not current: + return False + + gain_pct = (current - past) / past * 100 + result = gain_pct >= TREND_6H_MIN_PCT + + if result: + logger.debug( + f"[6h추세↑] {ticker} 6h전={past:,.2f} 현재={current:,.2f} " + f"(+{gain_pct:.1f}% ≥ {TREND_6H_MIN_PCT}%)" + ) + else: + logger.debug( + f"[6h추세✗] {ticker} 6h {gain_pct:+.1f}% (기준={TREND_6H_MIN_PCT:+.1f}%)" + ) + return result + + def should_buy(ticker: str) -> bool: - """Strategy C + 시장 레짐: 레짐별 동적 임계값으로 추세 AND 모멘텀 판단.""" + """Strategy C + 시장 레짐 + 추세 지속형 진입 (6h 추세 + 15분 확인 워치리스트). + + 진입 조건: + 1. 12h 추세 ≥ 레짐별 임계값 (bull 3% / neutral 5% / bear 8%) + 2. 6h 추세 ≥ 1% (단기 급등이 아닌 지속 추세) + 3. 모멘텀 (MA20 초과 + 1h 거래량 급증) + 4. 위 조건 최초 충족 후 15분 경과 시 실제 진입 (확인 필터) + """ regime = get_regime() trend_pct = regime["trend_pct"] vol_mult = regime["vol_mult"] + # 조건 평가 (순서: 가장 빠른 필터 먼저) if not check_trend(ticker, trend_pct): + _watchlist.pop(ticker, None) # 조건 깨지면 워치리스트 초기화 return False - return check_momentum(ticker, vol_mult) + + if not check_trend_6h(ticker): + _watchlist.pop(ticker, None) + return False + + if not check_momentum(ticker, vol_mult): + _watchlist.pop(ticker, None) + return False + + # 모든 조건 충족 — 15분 확인 워치리스트 처리 + now = time.time() + if ticker not in _watchlist: + _watchlist[ticker] = now + logger.info( + f"[워치] {ticker} 신호 첫 발생 → {CONFIRM_SECONDS//60}분 후 진입 예정" + ) + return False + + elapsed = now - _watchlist[ticker] + if elapsed < CONFIRM_SECONDS: + logger.debug( + f"[워치] {ticker} 확인 대기 중 ({elapsed/60:.1f}분 / {CONFIRM_SECONDS//60}분)" + ) + return False + + # 15분 경과 → 진입 확정 + _watchlist.pop(ticker, None) + logger.info( + f"[워치확인] {ticker} {elapsed/60:.1f}분 경과 → 진입 확정" + ) + return True diff --git a/daemon/runner.py b/daemon/runner.py index e213d78..5119a0d 100644 --- a/daemon/runner.py +++ b/daemon/runner.py @@ -5,6 +5,7 @@ import time from core import trader from core.market import get_top_tickers +from core.market_regime import get_regime from core.strategy import should_buy logger = logging.getLogger(__name__) @@ -23,6 +24,16 @@ def run_scanner() -> None: time.sleep(SCAN_INTERVAL) continue + # Bear 레짐 시 신규 매수 완전 차단 + regime = get_regime() + if regime["name"] == "bear": + logger.info( + f"[Bear차단] 레짐={regime['emoji']} BEAR " + f"(score={regime['score']:+.2f}%) — 신규 매수 스킵" + ) + time.sleep(SCAN_INTERVAL) + continue + tickers = get_top_tickers() logger.info(f"스캔 시작: {len(tickers)}개 종목")