feat: replace trend strategy with volume-lead accumulation strategy
- strategy.py: rewrite should_buy() with volume-lead logic - detect accumulation: vol spike + 2h quiet price → record signal_price - entry: price rises ≥ TREND_AFTER_VOL% from signal_price - signal reset: timeout (8h) or price drops below signal_price - .env: add PRICE_QUIET_PCT=2.0, TREND_AFTER_VOL=4.8, SIGNAL_TIMEOUT_H=8.0 - vol_lead_sim.py: add parameter sweep 0.5~5.0% + fine sweep 4.0~5.0% Backtest result (9 tickers, 2026-01-15~): +4.8% threshold 26 trades | 69% win rate | +73.38% cumulative (vs A 33 trades 45% +24.25%) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
joungmin
226
core/strategy.py
226
core/strategy.py
@@ -1,8 +1,11 @@
|
||||
"""Strategy C: 현재 기준 N시간 전 대비 상승 추세(DB) AND 거래량 모멘텀 동시 충족 시 매수 신호.
|
||||
"""Volume Lead 전략: 거래량 축적(급증+횡보) 감지 후 +TREND_AFTER_VOL% 상승 시 선진입.
|
||||
|
||||
추가 필터:
|
||||
1. 6h 추세 확인 (단기 급등 아닌 지속 추세)
|
||||
2. 15분 확인 워치리스트 (신호 첫 발생 후 15분 내 재확인 시 진입)
|
||||
흐름:
|
||||
1. 직전 1h 거래량 > 로컬 5h 평균 × VOL_MULT AND
|
||||
2h 가격 변동 < PRICE_QUIET_PCT% (횡보 중 축적)
|
||||
→ 신호가(signal_price) 기록
|
||||
2. signal_price 대비 +TREND_AFTER_VOL% 이상 상승 시 진입
|
||||
3. SIGNAL_TIMEOUT_H 내 임계값 미달 또는 신호가 이하 하락 시 신호 초기화
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -12,174 +15,127 @@ import os
|
||||
import time
|
||||
|
||||
import pyupbit
|
||||
from .market import get_current_price, get_ohlcv
|
||||
|
||||
from .market import get_current_price
|
||||
from .market_regime import get_regime
|
||||
from .price_db import get_price_n_hours_ago
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# 추세 판단: 현재 기준 N시간 전 DB 가격 대비 +M% 이상이면 상승 중
|
||||
TREND_HOURS = float(os.getenv("TREND_HOURS", "12"))
|
||||
TREND_MIN_GAIN_PCT = float(os.getenv("TREND_MIN_GAIN_PCT", "5")) # 레짐이 없을 때 기본값
|
||||
# 축적 감지 파라미터
|
||||
PRICE_QUIET_PCT = float(os.getenv("PRICE_QUIET_PCT", "2.0")) # 2h 횡보 기준 (%)
|
||||
TREND_AFTER_VOL = float(os.getenv("TREND_AFTER_VOL", "5.0")) # 진입 임계값 (신호가 대비 %)
|
||||
SIGNAL_TIMEOUT_H = float(os.getenv("SIGNAL_TIMEOUT_H", "8.0")) # 신호 유효 시간 (h)
|
||||
|
||||
# 6h 단기 추세 최소 상승률 (추세 지속형 필터)
|
||||
TREND_6H_MIN_PCT = float(os.getenv("TREND_6H_MIN_PCT", "1.0"))
|
||||
# 거래량 파라미터
|
||||
LOCAL_VOL_HOURS = 5 # 로컬 기준 시간 (h)
|
||||
VOLUME_MULTIPLIER = float(os.getenv("VOLUME_MULTIPLIER", "2.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)
|
||||
# 축적 신호 상태: ticker → {"price": float, "time": float(unix)}
|
||||
_accum_signals: dict[str, dict] = {}
|
||||
|
||||
|
||||
def check_trend(ticker: str, min_gain_pct: float) -> bool:
|
||||
"""상승 추세 조건: 현재가가 DB에 저장된 N시간 전 가격 대비 +min_gain_pct% 이상."""
|
||||
past_price = get_price_n_hours_ago(ticker, TREND_HOURS)
|
||||
if past_price is None:
|
||||
logger.debug(f"[추세] {ticker} {TREND_HOURS:.0f}h 전 가격 없음 (수집 중)")
|
||||
return False
|
||||
|
||||
current = get_current_price(ticker)
|
||||
if not current:
|
||||
return False
|
||||
|
||||
gain_pct = (current - past_price) / past_price * 100
|
||||
result = gain_pct >= min_gain_pct
|
||||
|
||||
if result:
|
||||
logger.info(
|
||||
f"[추세↑] {ticker} {TREND_HOURS:.0f}h 전={past_price:,.2f} "
|
||||
f"현재={current:,.2f} (+{gain_pct:.1f}% ≥ {min_gain_pct}%)"
|
||||
)
|
||||
else:
|
||||
logger.debug(
|
||||
f"[추세✗] {ticker} {gain_pct:+.1f}% (기준={min_gain_pct:+.0f}%)"
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def check_momentum(ticker: str, vol_mult: float) -> bool:
|
||||
"""모멘텀 조건: 현재가 > MA20(일봉) AND 최근 1h 거래량 > 로컬 5h 평균 × vol_mult.
|
||||
|
||||
23h 평균은 낮 시간대 고거래량이 포함돼 새벽에 항상 미달하므로,
|
||||
로컬 5h 평균(같은 시간대 컨텍스트)과 비교한다.
|
||||
"""
|
||||
# MA20: 일봉 기준
|
||||
df_daily = get_ohlcv(ticker, count=MA_PERIOD + 1)
|
||||
if df_daily is None or len(df_daily) < MA_PERIOD + 1:
|
||||
return False
|
||||
|
||||
ma = df_daily["close"].iloc[-MA_PERIOD:].mean()
|
||||
current = get_current_price(ticker)
|
||||
if current is None:
|
||||
return False
|
||||
|
||||
price_ok = current > ma
|
||||
if not price_ok:
|
||||
logger.debug(f"[모멘텀✗] {ticker} 현재={current:,.0f} < MA20={ma:,.0f} (가격 기준 미달)")
|
||||
return False
|
||||
|
||||
# 거래량: 60분봉 기준 (최근 1h vs 이전 LOCAL_VOL_HOURS h 로컬 평균)
|
||||
fetch_count = LOCAL_VOL_HOURS + 3 # 여유 있게 fetch
|
||||
def _check_vol_spike(ticker: str, vol_mult: float) -> bool:
|
||||
"""직전 완성 1h 거래량이 로컬 5h 평균의 vol_mult 배 이상인지 확인."""
|
||||
fetch_count = LOCAL_VOL_HOURS + 3
|
||||
try:
|
||||
df_hour = pyupbit.get_ohlcv(ticker, interval="minute60", count=fetch_count)
|
||||
df = pyupbit.get_ohlcv(ticker, interval="minute60", count=fetch_count)
|
||||
except Exception:
|
||||
return False
|
||||
if df_hour is None or len(df_hour) < LOCAL_VOL_HOURS + 1:
|
||||
if df is None or len(df) < LOCAL_VOL_HOURS + 1:
|
||||
return False
|
||||
|
||||
recent_vol = df_hour["volume"].iloc[-2] # 직전 완성된 1h 봉
|
||||
local_avg = df_hour["volume"].iloc[-(LOCAL_VOL_HOURS + 1):-2].mean() # 이전 LOCAL_VOL_HOURS h 평균
|
||||
vol_ok = local_avg > 0 and recent_vol >= local_avg * vol_mult
|
||||
|
||||
ratio = recent_vol / local_avg if local_avg > 0 else 0
|
||||
if vol_ok:
|
||||
logger.info(
|
||||
f"[모멘텀↑] {ticker} 현재={current:,.0f} MA20={ma:,.0f} "
|
||||
f"1h거래량={recent_vol:.0f} 로컬{LOCAL_VOL_HOURS}h평균={local_avg:.0f} ({ratio:.2f}x ≥ {vol_mult}x)"
|
||||
)
|
||||
else:
|
||||
logger.debug(
|
||||
f"[모멘텀✗] {ticker} 1h거래량={recent_vol:.0f} 로컬{LOCAL_VOL_HOURS}h평균={local_avg:.0f} "
|
||||
f"({ratio:.2f}x < {vol_mult}x)"
|
||||
)
|
||||
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:
|
||||
recent_vol = df["volume"].iloc[-2] # 직전 완성된 1h 봉
|
||||
local_avg = df["volume"].iloc[-(LOCAL_VOL_HOURS + 1):-2].mean() # 이전 5h 평균
|
||||
if local_avg <= 0:
|
||||
return False
|
||||
|
||||
gain_pct = (current - past) / past * 100
|
||||
result = gain_pct >= TREND_6H_MIN_PCT
|
||||
|
||||
ratio = recent_vol / local_avg
|
||||
result = ratio >= vol_mult
|
||||
if result:
|
||||
logger.debug(
|
||||
f"[6h추세↑] {ticker} 6h전={past:,.2f} 현재={current:,.2f} "
|
||||
f"(+{gain_pct:.1f}% ≥ {TREND_6H_MIN_PCT}%)"
|
||||
f"[거래량↑] {ticker} 1h={recent_vol:.0f} / 5h평균={local_avg:.0f} ({ratio:.2f}x ≥ {vol_mult}x)"
|
||||
)
|
||||
else:
|
||||
logger.debug(
|
||||
f"[6h추세✗] {ticker} 6h {gain_pct:+.1f}% (기준={TREND_6H_MIN_PCT:+.1f}%)"
|
||||
f"[거래량✗] {ticker} {ratio:.2f}x < {vol_mult}x"
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def should_buy(ticker: str) -> bool:
|
||||
"""Strategy C + 시장 레짐 + 추세 지속형 진입 (6h 추세 + 15분 확인 워치리스트).
|
||||
"""Volume Lead 전략.
|
||||
|
||||
진입 조건:
|
||||
1. 12h 추세 ≥ 레짐별 임계값 (bull 3% / neutral 5% / bear 8%)
|
||||
2. 6h 추세 ≥ 1% (단기 급등이 아닌 지속 추세)
|
||||
3. 모멘텀 (MA20 초과 + 1h 거래량 급증)
|
||||
4. 위 조건 최초 충족 후 15분 경과 시 실제 진입 (확인 필터)
|
||||
1단계: 거래량 급증 + 2h 횡보 → 신호가 기록
|
||||
2단계: 신호가 대비 +TREND_AFTER_VOL% 상승 확인 시 진입
|
||||
"""
|
||||
regime = get_regime()
|
||||
trend_pct = regime["trend_pct"]
|
||||
vol_mult = regime["vol_mult"]
|
||||
regime = get_regime()
|
||||
vol_mult = regime["vol_mult"]
|
||||
|
||||
# 조건 평가 (순서: 가장 빠른 필터 먼저)
|
||||
if not check_trend(ticker, trend_pct):
|
||||
_watchlist.pop(ticker, None) # 조건 깨지면 워치리스트 초기화
|
||||
current = get_current_price(ticker)
|
||||
if not current:
|
||||
return False
|
||||
|
||||
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
|
||||
|
||||
# ── 기존 신호 유효성 확인 ────────────────────────────────
|
||||
sig = _accum_signals.get(ticker)
|
||||
if sig is not None:
|
||||
age_h = (now - sig["time"]) / 3600
|
||||
if age_h > SIGNAL_TIMEOUT_H:
|
||||
del _accum_signals[ticker]
|
||||
sig = None
|
||||
logger.debug(f"[축적타임아웃] {ticker} {age_h:.1f}h 경과 → 신호 초기화")
|
||||
|
||||
# ── 신호 없음: 축적 조건 체크 ────────────────────────────
|
||||
if sig is None:
|
||||
# 2h 가격 횡보 확인 (DB 가격 활용)
|
||||
price_2h = get_price_n_hours_ago(ticker, 2)
|
||||
if price_2h is None:
|
||||
return False
|
||||
|
||||
quiet = abs(current - price_2h) / price_2h * 100 < PRICE_QUIET_PCT
|
||||
|
||||
if not quiet:
|
||||
logger.debug(
|
||||
f"[횡보✗] {ticker} 2h변동={(current - price_2h) / price_2h * 100:+.1f}% "
|
||||
f"(기준={PRICE_QUIET_PCT}%)"
|
||||
)
|
||||
return False
|
||||
|
||||
# 거래량 급증 확인
|
||||
if not _check_vol_spike(ticker, vol_mult):
|
||||
return False
|
||||
|
||||
# 축적 신호 기록
|
||||
_accum_signals[ticker] = {"price": current, "time": now}
|
||||
logger.info(
|
||||
f"[워치] {ticker} 신호 첫 발생 → {CONFIRM_SECONDS//60}분 후 진입 예정"
|
||||
f"[축적감지] {ticker} 거래량 급증 + 2h 횡보 → 신호가={current:,.2f}원"
|
||||
)
|
||||
return False
|
||||
return False # 신호 첫 발생 시는 진입 안 함
|
||||
|
||||
elapsed = now - _watchlist[ticker]
|
||||
if elapsed < CONFIRM_SECONDS:
|
||||
# ── 신호 있음: 상승 확인 → 진입 ─────────────────────────
|
||||
signal_price = sig["price"]
|
||||
move_pct = (current - signal_price) / signal_price * 100
|
||||
|
||||
if current < signal_price:
|
||||
# 신호가 이하 하락 → 축적 실패
|
||||
del _accum_signals[ticker]
|
||||
logger.debug(
|
||||
f"[워치] {ticker} 확인 대기 중 ({elapsed/60:.1f}분 / {CONFIRM_SECONDS//60}분)"
|
||||
f"[축적실패] {ticker} 신호가={signal_price:,.2f} → 현재={current:,.2f} (하락) → 초기화"
|
||||
)
|
||||
return False
|
||||
|
||||
# 15분 경과 → 진입 확정
|
||||
_watchlist.pop(ticker, None)
|
||||
logger.info(
|
||||
f"[워치확인] {ticker} {elapsed/60:.1f}분 경과 → 진입 확정"
|
||||
if move_pct >= TREND_AFTER_VOL:
|
||||
del _accum_signals[ticker]
|
||||
logger.info(
|
||||
f"[축적진입] {ticker} 신호가={signal_price:,.2f}원 → 현재={current:,.2f}원 "
|
||||
f"(+{move_pct:.1f}% ≥ {TREND_AFTER_VOL}%)"
|
||||
)
|
||||
return True
|
||||
|
||||
logger.debug(
|
||||
f"[축적대기] {ticker} 신호가={signal_price:,.2f} 현재={current:,.2f} "
|
||||
f"(+{move_pct:.1f}% / 목표={TREND_AFTER_VOL}%)"
|
||||
)
|
||||
return True
|
||||
return False
|
||||
|
||||
Reference in New Issue
Block a user