FreqAIStrategy

"""
FreqAI Strategy - 머신러닝 기반 전략

LightGBM/XGBoost/CatBoost 모델을 사용한 예측 전략
강화학습(RL) 모드도 지원

Generated by TradingView Strategy Research Lab
"""

import logging
from functools import reduce
from typing import Optional

import numpy as np
import pandas as pd
import talib.abstract as ta
from pandas import DataFrame
from technical import qtpylib

from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter


logger = logging.getLogger(__name__)


class FreqAIStrategy(IStrategy):
    """
    FreqAI 머신러닝 전략
    
    지원 모델:
    - LightGBMRegressor (기본, 빠름)
    - XGBoostRegressor (정확도 높음)
    - CatBoostRegressor (범주형 데이터에 강함)
    - ReinforcementLearner (강화학습)
    """
    
    # 전략 설정
    INTERFACE_VERSION = 3
    
    # 타임프레임
    timeframe = '5m'
    
    # FreqAI 사용
    can_short = False
    
    # 리스크 관리
    stoploss = -0.05
    trailing_stop = True
    trailing_stop_positive = 0.01
    trailing_stop_positive_offset = 0.02
    trailing_only_offset_is_reached = True
    
    # ROI
    minimal_roi = {
        "0": 0.1,
        "30": 0.05,
        "60": 0.02,
        "120": 0.01,
    }
    
    # 주문 설정
    order_types = {
        'entry': 'limit',
        'exit': 'limit',
        'stoploss': 'market',
        'stoploss_on_exchange': True
    }
    
    # 최적화 파라미터
    buy_rsi = IntParameter(20, 40, default=30, space='buy', optimize=True)
    sell_rsi = IntParameter(60, 80, default=70, space='sell', optimize=True)
    
    def feature_engineering_expand_all(
        self, dataframe: DataFrame, period: int, metadata: dict, **kwargs
    ) -> DataFrame:
        """
        FreqAI 피처 엔지니어링 - 모든 타임프레임에 적용
        """
        # RSI
        dataframe["%-rsi-period"] = ta.RSI(dataframe, timeperiod=period)
        
        # MFI
        dataframe["%-mfi-period"] = ta.MFI(dataframe, timeperiod=period)
        
        # ADX
        dataframe["%-adx-period"] = ta.ADX(dataframe, timeperiod=period)
        
        # SMA
        dataframe["%-sma-period"] = ta.SMA(dataframe, timeperiod=period)
        
        # EMA
        dataframe["%-ema-period"] = ta.EMA(dataframe, timeperiod=period)
        
        # 볼린저 밴드
        bollinger = qtpylib.bollinger_bands(
            qtpylib.typical_price(dataframe), window=period, stds=2.2
        )
        dataframe["bb_lowerband-period"] = bollinger["lower"]
        dataframe["bb_middleband-period"] = bollinger["mid"]
        dataframe["bb_upperband-period"] = bollinger["upper"]
        
        dataframe["%-bb_width-period"] = (
            dataframe["bb_upperband-period"] - dataframe["bb_lowerband-period"]
        ) / dataframe["bb_middleband-period"]
        
        dataframe["%-close-bb_lower-period"] = (
            dataframe["close"] / dataframe["bb_lowerband-period"]
        )
        
        # MACD
        macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9)
        dataframe["%-macd-period"] = macd["macd"]
        dataframe["%-macdsignal-period"] = macd["macdsignal"]
        dataframe["%-macdhist-period"] = macd["macdhist"]
        
        # ROC (Rate of Change)
        dataframe["%-roc-period"] = ta.ROC(dataframe, timeperiod=period)
        
        # 상대 거래량
        dataframe["%-relative_volume-period"] = (
            dataframe["volume"] / dataframe["volume"].rolling(period).mean()
        )
        
        return dataframe
    
    def feature_engineering_expand_basic(
        self, dataframe: DataFrame, metadata: dict, **kwargs
    ) -> DataFrame:
        """
        FreqAI 피처 엔지니어링 - 기본 타임프레임에만 적용
        """
        # 가격 변화율
        dataframe["%-pct-change"] = dataframe["close"].pct_change()
        
        # 고가-저가 범위
        dataframe["%-high-low"] = (
            dataframe["high"] - dataframe["low"]
        ) / dataframe["close"]
        
        # 시가-종가 범위
        dataframe["%-open-close"] = (
            dataframe["open"] - dataframe["close"]
        ) / dataframe["close"]
        
        # 거래량 변화
        dataframe["%-volume-change"] = dataframe["volume"].pct_change()
        
        return dataframe
    
    def feature_engineering_standard(
        self, dataframe: DataFrame, metadata: dict, **kwargs
    ) -> DataFrame:
        """
        FreqAI 피처 엔지니어링 - 표준 피처
        """
        # 날짜/시간 피처
        dataframe["%-day_of_week"] = dataframe["date"].dt.dayofweek
        dataframe["%-hour_of_day"] = dataframe["date"].dt.hour
        
        return dataframe
    
    def set_freqai_targets(
        self, dataframe: DataFrame, metadata: dict, **kwargs
    ) -> DataFrame:
        """
        FreqAI 타겟 설정 (예측할 값)
        """
        # 미래 수익률 예측 (24캔들 후)
        dataframe["&-s_close"] = (
            dataframe["close"]
            .shift(-self.freqai_info["feature_parameters"]["label_period_candles"])
            .rolling(self.freqai_info["feature_parameters"]["label_period_candles"])
            .mean()
            / dataframe["close"]
            - 1
        )
        
        return dataframe
    
    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        지표 계산
        """
        # FreqAI 예측 수행
        dataframe = self.freqai.start(dataframe, metadata, self)
        
        # 추가 지표 (FreqAI 외)
        dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)
        dataframe['sma_20'] = ta.SMA(dataframe, timeperiod=20)
        dataframe['sma_50'] = ta.SMA(dataframe, timeperiod=50)
        
        return dataframe
    
    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        진입 조건
        """
        enter_long_conditions = [
            # FreqAI 예측이 양수 (상승 예측)
            dataframe["&-s_close"] > 0.01,
            
            # 예측 신뢰도
            dataframe["do_predict"] == 1,
            
            # RSI 필터
            dataframe['rsi'] < self.buy_rsi.value,
            
            # 거래량 필터
            dataframe["volume"] > 0,
        ]
        
        if enter_long_conditions:
            dataframe.loc[
                reduce(lambda x, y: x & y, enter_long_conditions),
                "enter_long"
            ] = 1
        
        return dataframe
    
    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        청산 조건
        """
        exit_long_conditions = [
            # FreqAI 예측이 음수 (하락 예측)
            dataframe["&-s_close"] < -0.005,
            
            # 또는 RSI 과매수
            dataframe['rsi'] > self.sell_rsi.value,
            
            # 거래량 필터
            dataframe["volume"] > 0,
        ]
        
        if exit_long_conditions:
            dataframe.loc[
                reduce(lambda x, y: x & y, exit_long_conditions),
                "exit_long"
            ] = 1
        
        return dataframe
    
    def get_ticker_indicator(self):
        """
        FreqAI 모델 식별자
        """
        return int(self.config["freqai"]["identifier"])