from __future__ import annotations from pathlib import Path import numpy as np import pandas as pd from pandas import DataFrame from freqtrade.strategy import IStrategy class FQ_BaselineFairStrategy(IStrategy): """Fair baseline strategy using OHLCV + normalized raw features (n_*). This is the benchmark strategy for comparison against TurboQuant Core. """ INTERFACE_VERSION = 3 timeframe = "1m" can_short = False process_only_new_candles = True startup_candle_count = 240 # Lịch ROI theo thời gian giữ lệnh minimal_roi = { "0": 0.022, "45": 0.010, "120": 0.0, } # Cắt lỗ tối đa 3% stoploss = -0.03 # Cấu hình trailing stop để bảo vệ lợi nhuận sẽ kích hoạt khi đạt lợi nhuận trailing_stop = True trailing_stop_positive = 0.007 trailing_stop_positive_offset = 0.015 trailing_only_offset_is_reached = True use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False protections = [ { "method": "CooldownPeriod", "stop_duration_candles": 30, } ] plot_config = { "main_plot": { "n_macd": {}, "n_macd_signal": {}, }, "subplots": { "Normalized": { "n_rsi": {}, "n_return_5": {}, "n_volatility": {}, "n_atr": {}, "n_trend": {}, }, "OHLCV": { "ohlcv_spread": {}, "ohlcv_body": {}, }, }, } def __init__(self, config: dict) -> None: super().__init__(config) self._n_features: DataFrame = pd.DataFrame() self._n_last_mtime_ns: int | None = None def _dataset_path(self) -> Path: return Path(__file__).resolve().parents[1] / "freqtrade_dataset.csv" # Tải lại các đặc trưng đã chuẩn hóa nếu dataset đã được cập nhật def _reload_n_features_if_needed(self) -> None: dataset_path = self._dataset_path() if not dataset_path.exists(): self._n_features = pd.DataFrame() self._n_last_mtime_ns = None return stat = dataset_path.stat() if self._n_last_mtime_ns == stat.st_mtime_ns and not self._n_features.empty: return required_cols = [ "time", "n_log_return", "n_return_5", "n_log_volume", "n_candle_body", "n_rsi", "n_macd", "n_macd_signal", "n_volatility", "n_atr", ] n_df = pd.read_csv(dataset_path, usecols=required_cols) n_df["date"] = pd.to_datetime(n_df["time"], errors="coerce", utc=True) n_df = n_df.dropna(subset=["date"]) for col in required_cols[1:]: n_df[col] = pd.to_numeric(n_df[col], errors="coerce") n_df = n_df[["date"] + required_cols[1:]] n_df = n_df.drop_duplicates(subset=["date"], keep="last").sort_values("date") self._n_features = n_df.set_index("date") self._n_last_mtime_ns = stat.st_mtime_ns # Kết hợp các đặc trưng đã chuẩn hóa vào dataframe chính, đảm bảo căn chỉnh theo thời gian def _merge_n_features(self, dataframe: DataFrame) -> DataFrame: self._reload_n_features_if_needed() dataframe = dataframe.copy() dataframe["date"] = pd.to_datetime(dataframe["date"], errors="coerce", utc=True) n_cols = [ "n_log_return", "n_return_5", "n_log_volume", "n_candle_body", "n_rsi", "n_macd", "n_macd_signal", "n_volatility", "n_atr", ] if self._n_features.empty: for col in n_cols: dataframe[col] = 0.0 #không bị crash hệ thống nếu dataset chưa sẵn sàng return dataframe n_aligned = self._n_features.reindex(pd.DatetimeIndex(dataframe["date"])) for col in n_cols: dataframe[col] = n_aligned[col].to_numpy() dataframe[col] = dataframe[col].fillna(0.0) return dataframe # Tính toán các chỉ báo kỹ thuật và đặc trưng đã chuẩn hóa, đảm bảo xử lý các giá trị vô hạn và NaN def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe = dataframe.copy() dataframe = self._merge_n_features(dataframe) dataframe["n_trend"] = dataframe["n_macd"] - dataframe["n_macd_signal"] dataframe["n_trend_slow"] = dataframe["n_trend"].rolling(5, min_periods=5).mean() dataframe["volume_mean"] = dataframe["volume"].rolling(30, min_periods=30).mean() dataframe["volume_ratio"] = dataframe["volume"] / dataframe["volume_mean"].replace(0, np.nan) dataframe["ohlcv_spread"] = (dataframe["high"] - dataframe["low"]) / dataframe["close"].replace(0, np.nan) dataframe["ohlcv_body"] = (dataframe["close"] - dataframe["open"]) / dataframe["open"].replace(0, np.nan) dataframe["ohlcv_green"] = (dataframe["close"] > dataframe["open"]).astype(int) dataframe[[ "n_log_return", "n_return_5", "n_log_volume", "n_candle_body", "n_rsi", "n_macd", "n_macd_signal", "n_volatility", "n_atr", "n_trend", "n_trend_slow", "volume_mean", "volume_ratio", "ohlcv_spread", "ohlcv_body", ]] = dataframe[[ "n_log_return", "n_return_5", "n_log_volume", "n_candle_body", "n_rsi", "n_macd", "n_macd_signal", "n_volatility", "n_atr", "n_trend", "n_trend_slow", "volume_mean", "volume_ratio", "ohlcv_spread", "ohlcv_body", ]].replace([float("inf"), float("-inf")], np.nan) dataframe[["volume_ratio", "n_trend_slow"]] = dataframe[["volume_ratio", "n_trend_slow"]].fillna(0.0) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: conditions = [ dataframe["volume"] > 0, dataframe["volume_ratio"] >= 1.20, dataframe["n_rsi"].between(-0.35, 0.45), dataframe["n_macd"] > dataframe["n_macd_signal"], dataframe["n_trend"] > 0.03, dataframe["n_trend_slow"] > 0.02, dataframe["n_return_5"] > 0.08, dataframe["n_volatility"].between(-0.30, 0.55), dataframe["n_atr"].between(-0.30, 0.70), dataframe["n_log_return"] > 0.0, dataframe["ohlcv_green"] == 1, dataframe["ohlcv_spread"].between(0.00005, 0.020), dataframe["ohlcv_body"] > -0.001, ] dataframe.loc[ pd.concat(conditions, axis=1).all(axis=1), "enter_long", ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: conditions = [ dataframe["volume"] > 0, ( (dataframe["n_macd"] < dataframe["n_macd_signal"]) | (dataframe["n_trend"] < -0.12) | (dataframe["n_return_5"] < -0.15) | (dataframe["n_volatility"] > 0.85) | (dataframe["n_atr"] > 0.90) | (dataframe["n_rsi"] < -0.65) | (dataframe["n_rsi"] > 0.85) | (dataframe["volume_ratio"] < 0.70) ), ] dataframe.loc[ pd.concat(conditions, axis=1).all(axis=1), "exit_long", ] = 1 return dataframe