upbit-trader/core/signal.py

"""시그널 감지 + 지표 계산 모듈.

20초봉 데이터에서 양봉 + 거래량 + 사전 필터를 적용하여
매수 시그널 후보를 반환한다.
"""
from __future__ import annotations

import logging
from typing import Optional

log = logging.getLogger(__name__)


def calc_vr(bar_list: list[dict], idx: int, lookback: int = 61) -> float:
    """거래량비(Volume Ratio) 계산. 상위 10% 트리밍.

    Args:
        bar_list: 봉 리스트.
        idx: 현재 봉 인덱스.
        lookback: 기준 봉 수.

    Returns:
        현재 봉 거래량 / trimmed mean 비율.
    """
    start = max(0, idx - lookback)
    end = max(0, idx - 2)
    baseline = sorted(bar_list[i]['volume'] for i in range(start, end))
    if not baseline:
        return 0.0
    trim = max(1, len(baseline) // 10)
    trimmed = baseline[:len(baseline) - trim]
    if not trimmed:
        return 0.0
    avg = sum(trimmed) / len(trimmed)
    return bar_list[idx]['volume'] / avg if avg > 0 else 0.0


def calc_atr(bar_list: list[dict], lookback: int = 28) -> float:
    """ATR(Average True Range) 비율 계산.

    Args:
        bar_list: 봉 리스트.
        lookback: ATR 계산 봉 수.

    Returns:
        ATR / 직전 종가 비율 (0~1 범위).
    """
    if len(bar_list) < lookback + 2:
        return 0.0
    trs = []
    for i in range(-lookback - 1, -1):
        b = bar_list[i]
        bp = bar_list[i - 1]
        tr = max(
            b['high'] - b['low'],
            abs(b['high'] - bp['close']),
            abs(b['low'] - bp['close']),
        )
        trs.append(tr)
    prev_close = bar_list[-2]['close']
    return (sum(trs) / len(trs)) / prev_close if prev_close > 0 else 0.0


def detect_signal(
    ticker: str,
    bar_list: list[dict],
    *,
    vol_min: float = 5.0,
    vol_lookback: int = 61,
    vol_krw_min: float = 5_000_000,
    spread_min: float = 0.3,
) -> Optional[dict]:
    """양봉 + 거래량 + 사전 필터 3종을 적용하여 시그널 후보를 반환.

    Args:
        ticker: 종목 코드.
        bar_list: 봉 리스트 (list로 변환된 deque).
        vol_min: 최소 거래량 배수.
        vol_lookback: 거래량 평균 기준 봉 수.
        vol_krw_min: 최소 거래대금 (원).
        spread_min: 횡보 필터 최소 변동폭 (%).

    Returns:
        시그널 dict 또는 None.
    """
    n = len(bar_list)
    if n < vol_lookback + 5:
        return None

    b = bar_list[-1]
    if b['close'] <= b['open']:
        return None

    vr = calc_vr(bar_list, n - 1, lookback=vol_lookback)
    if vr < vol_min:
        return None

    bar_krw = b['close'] * b['volume']
    if bar_krw < vol_krw_min:
        return None

    # 1) 횡보 필터: 최근 15봉 변동폭 < 0.3%
    recent = bar_list[-15:]
    period_high = max(x['high'] for x in recent)
    period_low = min(x['low'] for x in recent)
    if period_low > 0:
        spread_pct = (period_high - period_low) / period_low * 100
        if spread_pct < spread_min:
            log.debug(f"[필터/횡보] {ticker} 15봉 변동 {spread_pct:.2f}% -> 스킵")
            return None

    # 2) 고점 필터: 30분 구간 90%+ 위치 & 변동 1%+
    long_bars = bar_list[-90:]
    long_high = max(x['high'] for x in long_bars)
    long_low = min(x['low'] for x in long_bars)
    if long_high > long_low:
        pos_in_range = (b['close'] - long_low) / (long_high - long_low)
        move_pct = (long_high - long_low) / long_low * 100
        if pos_in_range > 0.9 and move_pct > 1.0:
            log.debug(f"[필터/고점] {ticker} 구간 {pos_in_range:.0%} 위치, 변동 {move_pct:.1f}% -> 스킵")
            return None

    # 3) 연속 양봉 필터: 직전 2봉 이상 연속 양봉
    prev_greens = 0
    for k in range(len(bar_list) - 2, max(len(bar_list) - 12, 0), -1):
        if bar_list[k]['close'] > bar_list[k]['open']:
            prev_greens += 1
        else:
            break
    if prev_greens < 2:
        log.debug(f"[필터/양봉] {ticker} 직전 연속양봉 {prev_greens}개 < 2 -> 스킵")
        return None

    return {
        'ticker': ticker,
        'price': b['close'],
        'vol_ratio': vr,
        'bar_list': bar_list,
    }