ScalperStrategyV2

from freqtrade.strategy.interface import IStrategy
from pandas import DataFrame
import talib.abstract as ta
import numpy as np

class ScalperStrategyV2(IStrategy):
    """
    高频剥头皮策略
    - 时间周期: 1m
    - 信号: EMA + VWAP + 成交量激增
    - 止盈止损: 小幅快速止盈，极紧止损
    """

    timeframe = '1m'
    stoploss = -0.003  # -0.3%
    minimal_roi = {
        "0": 0.004,     # 0.4%
        "2": 0          # 2分钟后不强制退出
    }
    trailing_stop = False

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # EMA
        dataframe['ema_short'] = ta.EMA(dataframe, timeperiod=9)
        dataframe['ema_long'] = ta.EMA(dataframe, timeperiod=21)

        # VWAP
        dataframe['vwap'] = (dataframe['close'] * dataframe['volume']).cumsum() / dataframe['volume'].cumsum()

        # 成交量均值 & 激增
        dataframe['vol_mean'] = dataframe['volume'].rolling(20).mean()
        dataframe['vol_spike'] = dataframe['volume'] > 1.5 * dataframe['vol_mean']

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe.loc[
            (dataframe['ema_short'] > dataframe['ema_long']) &  # 短期趋势向上
            (dataframe['close'] > dataframe['vwap']) &          # 高于 VWAP
            (dataframe['vol_spike']) &                          # 成交量激增
            (dataframe['volume'] > 0),                          # 避免空值
            'enter_long'
        ] = 1

        dataframe.loc[
            (dataframe['ema_short'] < dataframe['ema_long']) &  # 短期趋势向下
            (dataframe['close'] < dataframe['vwap']) &
            (dataframe['vol_spike']) &
            (dataframe['volume'] > 0),
            'enter_short'
        ] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 快速出场：均线反转 或 价格跌破 VWAP
        dataframe.loc[
            (dataframe['ema_short'] < dataframe['ema_long']) |
            (dataframe['close'] < dataframe['vwap']),
            'exit_long'
        ] = 1

        dataframe.loc[
            (dataframe['ema_short'] > dataframe['ema_long']) |
            (dataframe['close'] > dataframe['vwap']),
            'exit_short'
        ] = 1

        return dataframe