DeepCoin/strategy.py

"""
WLD 볼린저밴드 MTF 전략.

기본 타이밍 (모든 봉 동일):
  - 매수: 하단 밴드 상향 돌파 (prev_close <= Lower, close > Lower)
  - 매도: 상단 밴드 상향 돌파 (prev_close < Upper, close >= Upper)

여러 봉(3·10·15·30·60·240·1440분)의 BB 상태를 비교해
실행 봉·확인 봉을 정한 뒤 매매합니다 (MtfBbPolicy).
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Literal

import numpy as np
import pandas as pd

from config import (
    BB_MIN_WIDTH_PCT,
    DEFAULT_BUY_KRW,
    ENTRY_INTERVAL,
    RANGE_BUY_KRW,
    RSI_BUY_MAX,
    RSI_PERIOD,
    TREND_RANGE_MA_GAP_PCT,
    VOLUME_BUY_RATIO,
)

Trend = Literal["up", "down", "range"]
Action = Literal["buy", "sell"]

# 기본 신호
SIGNAL_BUY_LOWER = "bb_base_buy_lower"
SIGNAL_SELL_UPPER = "bb_base_sell_upper"
SIGNAL_SELL_STOP = "bb_base_sell_stop"

BUY_SIGNALS = {SIGNAL_BUY_LOWER}
SELL_SIGNALS = {SIGNAL_SELL_UPPER, SIGNAL_SELL_STOP}

# 봉별 BB 상태 (분석·확인용)
STATE_SQUEEZE = "squeeze"
STATE_CROSS_UP_LOWER = "cross_up_lower"
STATE_CROSS_UP_UPPER = "cross_up_upper"
STATE_CROSS_DOWN_LOWER = "cross_down_lower"
STATE_BELOW_LOWER = "below_lower"
STATE_ABOVE_UPPER = "above_upper"
STATE_INSIDE = "inside"

BUY_TRIGGER_STATE = STATE_CROSS_UP_LOWER
SELL_TRIGGER_STATE = STATE_CROSS_UP_UPPER
STOP_TRIGGER_STATE = STATE_CROSS_DOWN_LOWER

# 확인 봉에서 매수 허용/차단 상태
BUY_CONFIRM_OK = {
    STATE_INSIDE,
    STATE_BELOW_LOWER,
    STATE_CROSS_UP_LOWER,
    STATE_SQUEEZE,
}
BUY_CONFIRM_BLOCK = {
    STATE_CROSS_UP_UPPER,
    STATE_ABOVE_UPPER,
    STATE_CROSS_DOWN_LOWER,
}
SELL_CONFIRM_OK = {
    STATE_INSIDE,
    STATE_ABOVE_UPPER,
    STATE_CROSS_UP_UPPER,
    STATE_CROSS_UP_LOWER,
    STATE_BELOW_LOWER,
}


@dataclass
class StrategyConfig:
    """전략 조합 설정 (시뮬 비교·실거래 공용)."""

    name: str = "default"
    use_mtf: bool = True
    use_regime_switch: bool = True
    use_rsi_filter: bool = True
    rsi_buy_max: float = RSI_BUY_MAX
    use_volume_filter: bool = False
    volume_ratio: float = VOLUME_BUY_RATIO
    use_squeeze_filter: bool = True
    use_stop_loss: bool = True
    allow_buy_in_up: bool = True
    allow_buy_in_range: bool = True
    allow_buy_in_down: bool = False
    use_discovered_rules: bool = False


# HTML 시뮬: discovered_rules.json (python simulation.py discover)
ACTIVE_CONFIG = StrategyConfig(
    name="discovered_best",
    use_discovered_rules=True,
    use_mtf=False,
    use_regime_switch=False,
    use_rsi_filter=False,
    use_volume_filter=False,
    use_squeeze_filter=False,
    use_stop_loss=False,
)


@dataclass
class MtfBbPolicy:
    """
    봉 간격별 BB 비교 후 적용할 매매 정책.

    buy_interval / sell_interval: 실제 주문 트리거 봉
    buy_confirm_intervals: 매수 시 함께 봐야 할 상위(긴) 봉
    """

    buy_interval: int = 3
    sell_interval: int = 3
    buy_confirm_intervals: tuple[int, ...] = (60, 1440)
    sell_confirm_intervals: tuple[int, ...] = (60,)
    name: str = "default_mtf"


ACTIVE_MTF_POLICY = MtfBbPolicy(
    name="3분실행_60·일봉확인",
    buy_interval=3,
    sell_interval=3,
    buy_confirm_intervals=(60, 1440),
    sell_confirm_intervals=(60,),
)


@dataclass
class TradeSignal:
    """실시간 1회 매매 신호."""

    action: Action
    signal: str
    close: float
    trend: Trend


def prepare_entry_df(data: pd.DataFrame) -> pd.DataFrame:
    """3분봉 보조 지표 추가."""
    df = data.copy()
    delta = df["Close"].diff()
    gain = delta.where(delta > 0, 0.0).rolling(RSI_PERIOD).mean()
    loss = (-delta.where(delta < 0, 0.0)).rolling(RSI_PERIOD).mean()
    rs = gain / loss.replace(0, np.nan)
    df["RSI"] = 100 - (100 / (1 + rs))
    df["VolMA5"] = df["Volume"].rolling(5).mean()
    ma = df["MA"].replace(0, np.nan)
    df["BB_Width"] = (df["Upper"] - df["Lower"]) / ma * 100
    return df


def get_trend(df_1d: pd.DataFrame, df_1h: pd.DataFrame) -> Trend:
    """일봉·1시간봉 기준 추세."""
    if len(df_1d) < 20 or len(df_1h) < 40:
        return "range"

    d_close = float(df_1d["Close"].iloc[-1])
    d_ma20 = float(df_1d["MA20"].iloc[-1])
    h_close = float(df_1h["Close"].iloc[-1])
    h_ma20 = float(df_1h["MA20"].iloc[-1])
    h_ma40 = float(df_1h["MA40"].iloc[-1])

    if h_ma40 == 0:
        return "range"

    ma_gap_pct = abs(h_ma20 - h_ma40) / h_ma40 * 100
    if ma_gap_pct < TREND_RANGE_MA_GAP_PCT:
        return "range"

    if d_close > d_ma20 and h_ma20 > h_ma40 and h_close > h_ma20:
        return "up"
    if d_close < d_ma20 and h_ma20 < h_ma40 and h_close < h_ma20:
        return "down"
    return "range"


def get_trend_at(
    df_1d: pd.DataFrame, df_1h: pd.DataFrame, ts: pd.Timestamp
) -> Trend:
    """특정 시점까지의 데이터로 추세 판별 (백테스트용)."""
    d = df_1d[df_1d.index <= ts]
    h = df_1h[df_1h.index <= ts]
    if d.empty or h.empty:
        return "range"
    return get_trend(d, h)


def _bb_squeeze(bb_width: float) -> bool:
    return bb_width < BB_MIN_WIDTH_PCT


def classify_bb_state_at(
    df: pd.DataFrame,
    index_pos: int,
    cfg: StrategyConfig | None = None,
) -> str:
    """
    한 봉의 볼린저밴드 상태를 분류합니다.

    Returns:
        cross_up_lower, cross_up_upper, cross_down_lower, below_lower,
        above_upper, inside, squeeze
    """
    cfg = cfg or ACTIVE_CONFIG
    if index_pos < 0:
        index_pos = len(df) + index_pos
    if index_pos < 1 or index_pos >= len(df):
        return STATE_INSIDE

    close = float(df["Close"].iloc[index_pos])
    prev_close = float(df["Close"].iloc[index_pos - 1])
    lower = float(df["Lower"].iloc[index_pos])
    prev_lower = float(df["Lower"].iloc[index_pos - 1])
    upper = float(df["Upper"].iloc[index_pos])
    prev_upper = float(df["Upper"].iloc[index_pos - 1])
    bb_width = float(df["BB_Width"].iloc[index_pos]) if "BB_Width" in df.columns else 999.0

    if cfg.use_squeeze_filter and _bb_squeeze(bb_width):
        return STATE_SQUEEZE
    if prev_close <= prev_lower and close > lower:
        return STATE_CROSS_UP_LOWER
    if prev_close < prev_upper and close >= upper:
        return STATE_CROSS_UP_UPPER
    if prev_close >= prev_lower and close < lower:
        return STATE_CROSS_DOWN_LOWER
    if close < lower:
        return STATE_BELOW_LOWER
    if close > upper:
        return STATE_ABOVE_UPPER
    return STATE_INSIDE


def get_latest_bb_state(df: pd.DataFrame, cfg: StrategyConfig | None = None) -> str:
    """마지막 봉의 BB 상태."""
    df = prepare_entry_df(df)
    if len(df) < 2:
        return STATE_INSIDE
    return classify_bb_state_at(df, -1, cfg)


def _allow_buy(trend: Trend, cfg: StrategyConfig) -> bool:
    """레짐·MTF에 따른 매수 허용 여부."""
    if cfg.use_regime_switch:
        if trend == "down":
            return cfg.allow_buy_in_down
        if trend == "range":
            return cfg.allow_buy_in_range
        if trend == "up":
            return cfg.allow_buy_in_up
    if cfg.use_mtf and trend == "down":
        return False
    return True


def evaluate_interval_bb(
    symbol: str,
    df_entry: pd.DataFrame,
    config: StrategyConfig | None = None,
) -> TradeSignal | None:
    """
    단일 봉 간격에서 BB 기본 규칙만으로 신호 1건.

    매도 우선 → 매수.
    """
    cfg = config or ACTIVE_CONFIG
    df = prepare_entry_df(df_entry)
    if len(df) < 22:
        return None

    state = classify_bb_state_at(df, -1, cfg)
    close = float(df["Close"].iloc[-1])
    trend: Trend = "range"

    if cfg.use_stop_loss and state == STATE_CROSS_DOWN_LOWER:
        return TradeSignal("sell", SIGNAL_SELL_STOP, close, trend)
    if state == STATE_CROSS_UP_UPPER:
        return TradeSignal("sell", SIGNAL_SELL_UPPER, close, trend)
    if state == STATE_CROSS_UP_LOWER:
        if cfg.use_rsi_filter:
            rsi = float(df["RSI"].iloc[-1])
            if not np.isnan(rsi) and rsi > cfg.rsi_buy_max:
                return None
        if cfg.use_volume_filter:
            vol = float(df["Volume"].iloc[-1])
            vol_ma = float(df["VolMA5"].iloc[-1])
            if vol_ma > 0 and vol < vol_ma * cfg.volume_ratio:
                return None
        return TradeSignal("buy", SIGNAL_BUY_LOWER, close, trend)
    return None


def _confirm_buy(states: dict[int, str], policy: MtfBbPolicy) -> bool:
    """상위 봉 상태가 매수에 적합한지."""
    for iv in policy.buy_confirm_intervals:
        st = states.get(iv)
        if st is None:
            return False
        if st in BUY_CONFIRM_BLOCK:
            return False
        if st not in BUY_CONFIRM_OK and st != STATE_SQUEEZE:
            return False
    return True


def _confirm_sell(states: dict[int, str], policy: MtfBbPolicy) -> bool:
    """상위 봉 상태가 매도에 적합한지."""
    for iv in policy.sell_confirm_intervals:
        st = states.get(iv)
        if st is None:
            continue
        if st in {STATE_CROSS_DOWN_LOWER, STATE_BELOW_LOWER}:
            return False
    return True


def evaluate_mtf_bb(
    symbol: str,
    frames: dict[int, pd.DataFrame],
    df_1d: pd.DataFrame,
    df_1h: pd.DataFrame,
    policy: MtfBbPolicy | None = None,
    config: StrategyConfig | None = None,
) -> TradeSignal | None:
    """
    여러 봉의 BB 상태를 비교해 최종 매수/매도 1건을 결정합니다.

    1) 각 봉 최신 BB 상태 분류
    2) policy.sell_interval 에서 손절/상단돌파 → 매도
    3) policy.buy_interval 에서 하단돌파 + 확인봉 OK → 매수
    4) 일봉·1시간 추세로 하락장 매수 차단 (config)
    """
    cfg = config or ACTIVE_CONFIG
    policy = policy or ACTIVE_MTF_POLICY
    if policy.buy_interval not in frames or policy.sell_interval not in frames:
        return None

    states = {iv: get_latest_bb_state(frames[iv], cfg) for iv in frames}
    trend = get_trend(df_1d, df_1h)

    sell_df = prepare_entry_df(frames[policy.sell_interval])
    sell_close = float(sell_df["Close"].iloc[-1])
    sell_state = states[policy.sell_interval]

    if cfg.use_stop_loss and sell_state == STATE_CROSS_DOWN_LOWER:
        if not policy.sell_confirm_intervals or _confirm_sell(states, policy):
            return TradeSignal("sell", SIGNAL_SELL_STOP, sell_close, trend)

    if sell_state == STATE_CROSS_UP_UPPER:
        if not policy.sell_confirm_intervals or _confirm_sell(states, policy):
            return TradeSignal("sell", SIGNAL_SELL_UPPER, sell_close, trend)

    buy_state = states[policy.buy_interval]
    if buy_state != STATE_CROSS_UP_LOWER:
        return None
    if not _allow_buy(trend, cfg):
        return None
    if not _confirm_buy(states, policy):
        return None

    buy_close = float(frames[policy.buy_interval]["Close"].iloc[-1])
    sig = evaluate_interval_bb(symbol, frames[policy.buy_interval], cfg)
    if sig and sig.action == "buy":
        return TradeSignal("buy", sig.signal, buy_close, trend)
    return None


def evaluate(
    symbol: str,
    df_entry: pd.DataFrame,
    df_1h: pd.DataFrame,
    df_1d: pd.DataFrame,
    config: StrategyConfig | None = None,
    trend_override: Trend | None = None,
    frames: dict[int, pd.DataFrame] | None = None,
    policy: MtfBbPolicy | None = None,
) -> TradeSignal | None:
    """
    매매 신호 1건.

    frames 가 있으면 evaluate_mtf_bb, 없으면 df_entry 단일 봉 BB.
    """
    if frames:
        return evaluate_mtf_bb(symbol, frames, df_1d, df_1h, policy, config)
    cfg = config or ACTIVE_CONFIG
    trend = trend_override if trend_override else get_trend(df_1d, df_1h)
    sig = evaluate_interval_bb(symbol, df_entry, cfg)
    if sig is None:
        return None
    if sig.action == "buy" and not _allow_buy(trend, cfg):
        return None
    return TradeSignal(sig.action, sig.signal, sig.close, trend)


def annotate_interval_signals(
    symbol: str,
    data: pd.DataFrame,
    simulation: bool | None = None,
    config: StrategyConfig | None = None,
) -> pd.DataFrame:
    """단일 봉 간격 전체에 BB 신호 기록 (봉별 백테스트용)."""
    cfg = config or ACTIVE_CONFIG
    df = prepare_entry_df(data)
    df["signal"] = ""
    df["point"] = 0
    df["action"] = ""
    df["bb_state"] = ""

    for i in range(21, len(df)):
        st = classify_bb_state_at(df, i, cfg)
        df.at[df.index[i], "bb_state"] = st
        sig = evaluate_interval_bb(symbol, df.iloc[: i + 1], cfg)
        if sig:
            df.at[df.index[i], "signal"] = sig.signal
            df.at[df.index[i], "point"] = 1
            df.at[df.index[i], "action"] = sig.action

    if not simulation:
        sig = evaluate_interval_bb(symbol, df, cfg)
        if sig:
            df.at[df.index[-1], "signal"] = sig.signal
            df.at[df.index[-1], "point"] = 1
            df.at[df.index[-1], "action"] = sig.action
        df.at[df.index[-1], "bb_state"] = get_latest_bb_state(df, cfg)

    return df


def _slice_frames_at(frames: dict[int, pd.DataFrame], ts: pd.Timestamp) -> dict[int, pd.DataFrame]:
    """시점 ts 이하 봉만 남긴 frames."""
    out: dict[int, pd.DataFrame] = {}
    for iv, df in frames.items():
        part = df[df.index <= ts]
        if len(part) >= 22:
            out[iv] = part
    return out


def annotate_mtf_signals(
    symbol: str,
    frames: dict[int, pd.DataFrame],
    df_1d: pd.DataFrame,
    df_1h: pd.DataFrame,
    policy: MtfBbPolicy | None = None,
    config: StrategyConfig | None = None,
) -> pd.DataFrame:
    """실행 봉(buy_interval) 타임라인에 MTF BB 신호 기록."""
    policy = policy or ACTIVE_MTF_POLICY
    cfg = config or ACTIVE_CONFIG
    if policy.buy_interval not in frames:
        raise ValueError(f"buy_interval {policy.buy_interval} 데이터 없음")

    df = prepare_entry_df(frames[policy.buy_interval].copy())
    df["signal"] = ""
    df["point"] = 0
    df["action"] = ""
    df["trend"] = ""
    df["bb_state"] = ""

    for i in range(21, len(df)):
        ts = df.index[i]
        sliced = _slice_frames_at(frames, ts)
        if policy.buy_interval not in sliced:
            continue
        df.at[ts, "bb_state"] = get_latest_bb_state(sliced[policy.buy_interval], cfg)
        trend_at = get_trend_at(df_1d, df_1h, ts)
        sig = evaluate_mtf_bb(symbol, sliced, df_1d, df_1h, policy, cfg)
        if sig:
            df.at[ts, "signal"] = sig.signal
            df.at[ts, "point"] = 1
            df.at[ts, "action"] = sig.action
            df.at[ts, "trend"] = sig.trend

    return df


def evaluate_discovered_live(
    symbol: str,
    frames: dict[int, pd.DataFrame],
    df_1d: pd.DataFrame,
    df_1h: pd.DataFrame,
    balances: dict,
) -> TradeSignal | None:
    """
    최신 3분 봉 시점에서 discovered_rules + 전 봉 BB·일목 조합으로 신호 1건.
    """
    from candle_features import build_master_feature_matrix
    from rule_discovery import buy_mask, load_rules, rules_have_buy, sell_mask

    rules = load_rules()
    if rules is None or not rules_have_buy(rules):
        return None

    matrix = build_master_feature_matrix(frames)
    if len(matrix) < 22:
        return None
    last = matrix.iloc[[-1]]
    ts = last.index[-1]
    close = float(last["Close"].iloc[-1])
    trend = get_trend_at(df_1d, df_1h, ts)

    position = float(balances.get(symbol, {}).get("balance", 0) or 0)
    if position >= 1.0:
        if rules.sell_stop and sell_mask(last, rules, stop=True)[0]:
            return TradeSignal("sell", SIGNAL_SELL_STOP, close, trend)
        if sell_mask(last, rules, stop=False)[0]:
            return TradeSignal("sell", SIGNAL_SELL_UPPER, close, trend)
        return None

    if buy_mask(last, rules)[0]:
        return TradeSignal("buy", SIGNAL_BUY_LOWER, close, trend)
    return None


def annotate_discovered_signals(
    symbol: str,
    frames: dict[int, pd.DataFrame],
    df_1d: pd.DataFrame,
    df_1h: pd.DataFrame,
    rules=None,
    data: pd.DataFrame | None = None,
) -> pd.DataFrame:
    """탐색된 다봉·캔들 규칙으로 3분 타임라인(전체 봉)에 신호 기록."""
    from candle_features import build_master_feature_matrix
    from rule_discovery import generate_trade_events, load_rules, rules_have_buy

    rule_set = rules or load_rules()
    if rule_set is None or not rules_have_buy(rule_set):
        raise FileNotFoundError(
            "discovered_rules.json 없거나 매수 규칙이 비어 있습니다. "
            "python simulation.py 실행"
        )

    entry = frames.get(ENTRY_INTERVAL)
    if entry is None or entry.empty:
        raise ValueError("3분봉 데이터가 없습니다.")

    matrix = build_master_feature_matrix(frames).iloc[21:].copy()
    df = prepare_entry_df(data if data is not None else entry)
    df["signal"] = ""
    df["point"] = 0
    df["action"] = ""
    df["trend"] = ""

    for ts, action, sig in generate_trade_events(matrix, rule_set):
        if ts not in df.index:
            continue
        trend_at = get_trend_at(df_1d, df_1h, ts)
        df.at[ts, "signal"] = sig
        df.at[ts, "point"] = 1
        df.at[ts, "action"] = action
        df.at[ts, "trend"] = trend_at

    return df


def annotate_signals(
    symbol: str,
    data: pd.DataFrame,
    simulation: bool | None = None,
    df_1h: pd.DataFrame | None = None,
    df_1d: pd.DataFrame | None = None,
    config: StrategyConfig | None = None,
    frames: dict[int, pd.DataFrame] | None = None,
) -> pd.DataFrame:
    """3분봉 구간 전체에 signal/point/action/trend 기록."""
    cfg = config or ACTIVE_CONFIG
    if cfg.use_discovered_rules and frames:
        h1d = df_1d if df_1d is not None and not df_1d.empty else data
        h1h = df_1h if df_1h is not None and not df_1h.empty else data
        return annotate_discovered_signals(symbol, frames, h1d, h1h, data=data)
    df = prepare_entry_df(data)
    df["signal"] = ""
    df["point"] = 0
    df["action"] = ""
    df["trend"] = ""

    htf_1h = df_1h if df_1h is not None else df
    htf_1d = df_1d if df_1d is not None else df

    for i in range(21, len(df)):
        ts = df.index[i]
        trend_at = get_trend_at(htf_1d, htf_1h, ts) if simulation else get_trend(htf_1d, htf_1h)
        sig = evaluate(
            symbol,
            df.iloc[: i + 1],
            htf_1h,
            htf_1d,
            config=cfg,
            trend_override=trend_at if simulation else None,
        )
        if sig:
            df.at[df.index[i], "signal"] = sig.signal
            df.at[df.index[i], "point"] = 1
            df.at[df.index[i], "action"] = sig.action
            df.at[df.index[i], "trend"] = sig.trend

    if not simulation:
        live = evaluate(symbol, df, htf_1h, htf_1d, config=cfg)
        if live:
            df.at[df.index[-1], "signal"] = live.signal
            df.at[df.index[-1], "point"] = 1
            df.at[df.index[-1], "action"] = live.action
            df.at[df.index[-1], "trend"] = live.trend

    return df


def comparison_presets() -> list[StrategyConfig]:
    """기법 조합 비교용 프리셋."""
    return [
        StrategyConfig(name="01_기본_BB만", use_mtf=False, use_regime_switch=False,
                       use_rsi_filter=False, use_squeeze_filter=False, use_stop_loss=False),
        StrategyConfig(name="02_기본+손절", use_mtf=False, use_regime_switch=False,
                       use_rsi_filter=False, use_squeeze_filter=False, use_stop_loss=True),
        StrategyConfig(name="03_기본+MTF", use_mtf=True, use_regime_switch=False,
                       use_rsi_filter=False, use_squeeze_filter=False, use_stop_loss=True),
        StrategyConfig(name="04_기본+MTF+스퀴즈", use_mtf=True, use_regime_switch=False,
                       use_rsi_filter=False, use_squeeze_filter=True, use_stop_loss=True),
        StrategyConfig(name="05_기본+MTF+RSI", use_mtf=True, use_regime_switch=False,
                       use_rsi_filter=True, use_squeeze_filter=False, use_stop_loss=True),
        StrategyConfig(name="06_기본+MTF+거래량", use_mtf=True, use_regime_switch=False,
                       use_rsi_filter=False, use_volume_filter=True, volume_ratio=1.1,
                       use_squeeze_filter=False, use_stop_loss=True),
        StrategyConfig(name="07_레짐스위치", use_mtf=True, use_regime_switch=True,
                       use_rsi_filter=False, use_squeeze_filter=False, use_stop_loss=True),
        StrategyConfig(name="08_레짐+RSI+스퀴즈", use_mtf=True, use_regime_switch=True,
                       use_rsi_filter=True, use_squeeze_filter=True, use_stop_loss=True),
        StrategyConfig(name="09_풀필터", use_mtf=True, use_regime_switch=True,
                       use_rsi_filter=True, use_volume_filter=True, volume_ratio=1.1,
                       use_squeeze_filter=True, use_stop_loss=True),
    ]


def get_signal_action(signal: str) -> Action | None:
    if signal in BUY_SIGNALS:
        return "buy"
    if signal in SELL_SIGNALS:
        return "sell"
    return None


def get_buy_amount(symbol: str, signal: str, close: float, trend: Trend = "up") -> int:
    if trend == "range":
        return RANGE_BUY_KRW
    return DEFAULT_BUY_KRW


def get_sell_ratio(symbol: str, signal: str) -> float:
    return 1.0


def allowed_inverse_sell_signals() -> set[str]:
    return set()


def should_double_buy(symbol: str, signal: str, data: pd.DataFrame) -> bool:
    return False