wiki/quant/multi_asset_channel_breakout_v5_stock.py

import pandas as pd
import talib.abstract as ta
from functools import reduce
from datetime import datetime
from freqtrade.strategy import (IStrategy, IntParameter, DecimalParameter)
from freqtrade.persistence import Trade


class MultiAssetChannelBreakoutV5Stock(IStrategy):
    """
    多股票通道突破策略 V5-Stock（OKX 代币化美股版）。

    基于 MultiAssetChannelBreakoutV5 改造，适配 OKX 现货代币化股票：
    - spot 模式（1x，无杠杆）
    - 增加美股交易时段过滤，排除盘前/盘后/周末的薄流动性噪声
    - 参数面向股票市场特性调整（ADX 阈值降低、滑点容忍放宽）
    - stoploss / custom_stoploss 逻辑不变（V5 已是 1x 价格维度，spot 直接适用）

    前置条件：
    - freqtrade spot 模式实例（trading_mode 留空，非 futures）
    - pair_whitelist 使用 X 前缀代币化股票（如 XAAPL/USDT）
    - 已验证 OKX 账户可交易 X 前缀标的（小额实单测试）
    """

    INTERFACE_VERSION = 3

    # --- 1) 基础交易设置 ---
    minimal_roi = {"0": 100}
    stoploss = -0.08
    trailing_stop = False
    use_custom_stoploss = True
    timeframe = "15m"
    can_short = False
    startup_candle_count = 600

    # --- 2) 可优化参数（默认值面向股票市场调整） ---
    up_line_span = IntParameter(100, 400, default=340, space="buy", optimize=True)
    buy_stop_profit_span = IntParameter(50, 200, default=141, space="buy", optimize=True)
    ma_span_long_days = IntParameter(1, 5, default=2, space="buy", optimize=True)
    # 降低默认值：股票 ADX 普遍低于加密货币（BTC/ETH 15m 常态 ADX 30-50，股票 15m 常态 15-30）
    adx_threshold = IntParameter(15, 40, default=25, space="buy", optimize=True)
    up_line_offset = DecimalParameter(-0.2, 0.4, default=-0.18, decimals=2, space="buy", optimize=True)
    buy_stop_profit_offset = DecimalParameter(-0.2, 0.2, default=0.19, decimals=2, space="sell", optimize=True)
    # 滑点容忍：代币化股票流动性薄，从 V5 的 1% 放宽到 1.5%
    max_slippage = DecimalParameter(0.005, 0.03, default=0.015, decimals=3, space="buy", optimize=True)

    def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        dataframe["n_value"] = ta.ATR(dataframe, timeperiod=10)

        ma_length = self.ma_span_long_days.value * 96
        dataframe["ma_long"] = ta.SMA(dataframe, timeperiod=ma_length)

        base_up_line = dataframe["high"].rolling(window=self.up_line_span.value).max().shift(1)
        dataframe["up_line"] = base_up_line + (dataframe["n_value"] * self.up_line_offset.value)

        dataframe["bottom_line"] = dataframe["low"].rolling(window=self.buy_stop_profit_span.value).min().shift(1)
        dataframe["adx"] = ta.ADX(dataframe, timeperiod=14)

        # --- 美股交易时段标记 ---
        # OKX 代币化股票 24/7 有 K 线，但非美股时段流动性极薄、价格漂移。
        # 只在美股正常交易时段（周一至周五 9:30-16:00 美东）允许入场。
        if "date" in dataframe.columns:
            df_date = pd.to_datetime(dataframe["date"], utc=True)
            et_time = df_date.dt.tz_convert("US/Eastern")
            et_hour = et_time.dt.hour
            et_minute = et_time.dt.minute
            et_weekday = et_time.dt.weekday  # 0=Mon, 6=Sun

            is_market_hours = (
                (et_weekday < 5)
                & (
                    (et_hour > 9)
                    | ((et_hour == 9) & (et_minute >= 30))
                )
                & (et_hour < 16)
            )
            dataframe["is_market_hours"] = is_market_hours
        else:
            dataframe["is_market_hours"] = True

        return dataframe

    def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        conditions = []
        conditions.append(dataframe["n_value"].notnull())
        conditions.append(dataframe["up_line"].notnull())
        conditions.append(dataframe["close"] > dataframe["ma_long"])
        conditions.append(dataframe["adx"] > self.adx_threshold.value)
        conditions.append(dataframe["close"] > dataframe["up_line"])
        conditions.append(dataframe["is_market_hours"])

        if conditions:
            is_entry = reduce(lambda x, y: x & y, conditions)
            dataframe.loc[is_entry, "enter_long"] = 1
            dataframe.loc[is_entry, "enter_tag"] = "trend_breakout"

        return dataframe

    def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        offset = self.buy_stop_profit_offset.value
        exit_line = dataframe["bottom_line"] + (dataframe["n_value"] * offset)

        exit_condition = dataframe["close"] < exit_line
        dataframe.loc[exit_condition, "exit_long"] = 1
        dataframe.loc[exit_condition, "exit_tag"] = "channel_exit"
        return dataframe

    def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime,
                        current_rate: float, current_profit: float, **kwargs) -> float:
        if current_profit > 0.30:
            return 0.10
        if current_profit > 0.15:
            return 0.10
        if current_profit > 0.05:
            breakeven_target = trade.open_rate * 1.005
            return (current_rate - breakeven_target) / current_rate
        return 1.0

    def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float,
                            time_in_force: str, current_time: datetime, entry_tag: str,
                            side: str, **kwargs) -> bool:
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)

        if len(dataframe) < 3:
            return False

        signal_candle = None
        for i in range(-1, -4, -1):
            if dataframe["enter_long"].iloc[i] == 1:
                signal_candle = dataframe.iloc[i]
                break

        if signal_candle is None:
            return False

        if rate <= signal_candle["ma_long"]:
            return False
        if rate <= signal_candle["up_line"]:
            return False

        if rate > signal_candle["close"] * (1 + self.max_slippage.value):
            return False

        return True