feat: bear block, trend-continuation entry, partial TP backtest

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 <noreply@anthropic.com>
This commit is contained in:
joungmin
2026-03-01 10:51:02 +09:00
parent b0f0b3e82a
commit 16b4c932a2
3 changed files with 204 additions and 16 deletions

View File

@@ -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)