okx-codex-trader/freqtrade/user_data/strategies/EthNextgenMicroSignalStream.py

from __future__ import annotations

import os
from datetime import datetime
from pathlib import Path

import pandas as pd
from freqtrade.strategy import IStrategy


class EthNextgenMicroSignalStream(IStrategy):
    INTERFACE_VERSION = 3

    timeframe = "15m"
    can_short = True
    startup_candle_count = 0
    process_only_new_candles = True

    minimal_roi = {"0": 100.0}
    stoploss = -0.99
    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False

    def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        signal_path = Path(
            os.environ.get(
                "ETH_NEXTGEN_MICRO_SIGNAL_STREAM",
                "reports/eth-exploration/eth-nextgen-micro-signal-stream.csv",
            )
        )
        signals = pd.read_csv(signal_path)
        signals["date"] = pd.to_datetime(signals["time"], utc=True)
        columns = [
            "date",
            "active_engine",
            "selected_entry_count",
            "selected_exit_count",
            "selected_entry_labels",
            "selected_exit_labels",
        ]
        merged = dataframe.merge(signals[columns], on="date", how="left")
        merged["selected_entry_count"] = merged["selected_entry_count"].fillna(0).astype(int)
        merged["selected_exit_count"] = merged["selected_exit_count"].fillna(0).astype(int)
        merged["selected_entry_labels"] = merged["selected_entry_labels"].fillna("")
        merged["selected_exit_labels"] = merged["selected_exit_labels"].fillna("")
        return merged

    def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        has_entry = dataframe["selected_entry_count"] > 0
        short_entry = has_entry & dataframe["selected_entry_labels"].str.contains("short", regex=False)
        long_entry = has_entry & ~short_entry
        dataframe.loc[long_entry, ["enter_long", "enter_tag"]] = (1, "switch_signal_long")
        dataframe.loc[short_entry, ["enter_short", "enter_tag"]] = (1, "switch_signal_short")
        return dataframe

    def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        has_exit = dataframe["selected_exit_count"] > 0
        dataframe.loc[has_exit, ["exit_long", "exit_tag"]] = (1, "switch_signal_exit")
        dataframe.loc[has_exit, ["exit_short", "exit_tag"]] = (1, "switch_signal_exit")
        return dataframe

    def leverage(
        self,
        pair: str,
        current_time: datetime,
        current_rate: float,
        proposed_leverage: float,
        max_leverage: float,
        entry_tag: str | None,
        side: str,
        **kwargs,
    ) -> float:
        return min(3.0, max_leverage)