Fair TurboQuant Core strategy using OHLCV + reconstructed features (tq_xhat_*).
Timeframe
1m
Direction
Long Only
Stoploss
-3.0%
Trailing Stop
Yes
ROI
0m: 2.2%, 45m: 1.0%, 120m: 0.0%
Interface Version
3
Startup Candles
240
Indicators
0
freqtrade/freqtrade-strategies
freqtrade/freqtrade-strategies
this strategy is based around the idea of generating a lot of potentatils buys and make tiny profits on each trade
freqtrade/freqtrade-strategies
this strategy is based around the idea of generating a lot of potentatils buys and make tiny profits on each trade
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_TurboCoreFairStrategy(IStrategy):
"""Fair TurboQuant Core strategy using OHLCV + reconstructed features (tq_xhat_*).
This strategy keeps trade logic aligned with baseline and excludes quality gates.
"""
INTERFACE_VERSION = 3
timeframe = "1m"
can_short = False
process_only_new_candles = True
startup_candle_count = 240
minimal_roi = {
"0": 0.022,
"45": 0.010,
"120": 0.0,
}
stoploss = -0.03
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,
}
]
# Thresholds on reconstructed normalized features (tq_xhat_*)
# Định nghĩa ngưỡng trên các đặc trưng đã tái tạo và chuẩn hóa (tq_xhat_*)
tqx_min_trend = 0.18
tqx_max_volatility = 0.40
tqx_max_atr = 0.40
tqx_exit_trend = -0.01
plot_config = {
"main_plot": {
"tqx_trend": {},
"tqx_momentum": {},
},
"subplots": {
"TurboQuant": {
"tq_xhat_rsi": {},
"tq_xhat_volatility": {},
"tq_xhat_atr": {},
"tqx_trend": {},
"tqx_momentum": {},
},
"OHLCV": {
"volume_ratio": {},
"ohlcv_body": {},
},
},
}
#Hàm khởi tạo để thiết lập các biến cần thiết cho việc quản lý
def __init__(self, config: dict) -> None:
super().__init__(config)
self._tqx_features: DataFrame = pd.DataFrame()
self._tqx_last_mtime_ns: int | None = None
def _dataset_path(self) -> Path:
return Path(__file__).resolve().parents[1] / "freqtrade_dataset.csv"
# Hàm để tải lại các đặc trưng từ dataset nếu có sự thay đổi về thời gian sửa đổi của file dataset hoặc nếu dữ liệu hiện tại rỗng
def _reload_tqx_features_if_needed(self) -> None:
dataset_path = self._dataset_path()
if not dataset_path.exists():
self._tqx_features = pd.DataFrame()
self._tqx_last_mtime_ns = None
return
stat = dataset_path.stat()
if self._tqx_last_mtime_ns == stat.st_mtime_ns and not self._tqx_features.empty:
return
required_cols = [
"time",
"tq_xhat_log_return",
"tq_xhat_return_5",
"tq_xhat_log_volume",
"tq_xhat_candle_body",
"tq_xhat_rsi",
"tq_xhat_macd",
"tq_xhat_macd_signal",
"tq_xhat_volatility",
"tq_xhat_atr",
]
tqx_df = pd.read_csv(dataset_path, usecols=required_cols)
tqx_df["date"] = pd.to_datetime(tqx_df["time"], errors="coerce", utc=True)
tqx_df = tqx_df.dropna(subset=["date"])
for col in required_cols[1:]:
tqx_df[col] = pd.to_numeric(tqx_df[col], errors="coerce")
tqx_df = tqx_df[["date"] + required_cols[1:]]
tqx_df = tqx_df.drop_duplicates(subset=["date"], keep="last").sort_values("date")
self._tqx_features = tqx_df.set_index("date")
self._tqx_last_mtime_ns = stat.st_mtime_ns
# Hàm để hợp nhất các đặc trưng tq_xhat_* vào dataframe chính dựa trên cột "date" đã được chuyển đổi sang định dạng datetime
def _merge_tqx_features(self, dataframe: DataFrame) -> DataFrame:
self._reload_tqx_features_if_needed()
dataframe = dataframe.copy()
dataframe["date"] = pd.to_datetime(dataframe["date"], errors="coerce", utc=True)
tqx_cols = [
"tq_xhat_log_return",
"tq_xhat_return_5",
"tq_xhat_log_volume",
"tq_xhat_candle_body",
"tq_xhat_rsi",
"tq_xhat_macd",
"tq_xhat_macd_signal",
"tq_xhat_volatility",
"tq_xhat_atr",
]
if self._tqx_features.empty:
for col in tqx_cols:
dataframe[col] = 0.0
return dataframe
tqx_aligned = self._tqx_features.reindex(pd.DatetimeIndex(dataframe["date"]))
for col in tqx_cols:
dataframe[col] = tqx_aligned[col].to_numpy()
for col in tqx_cols:
dataframe[col] = dataframe[col].fillna(0.0)
return dataframe
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
dataframe = dataframe.copy()
dataframe["volume_mean"] = dataframe["volume"].rolling(20, min_periods=20).mean()
dataframe["volume_ratio"] = dataframe["volume"] / dataframe["volume_mean"].replace(0, np.nan)
dataframe["ohlcv_body"] = (dataframe["close"] - dataframe["open"]) / dataframe["open"].replace(0, np.nan)
dataframe = self._merge_tqx_features(dataframe)
dataframe["tqx_trend"] = dataframe["tq_xhat_macd"] - dataframe["tq_xhat_macd_signal"]
dataframe["tqx_momentum"] = dataframe["tq_xhat_log_return"] + 0.35 * dataframe["tq_xhat_return_5"]
numeric_cols = [
"volume_mean",
"volume_ratio",
"ohlcv_body",
"tq_xhat_log_return",
"tq_xhat_return_5",
"tq_xhat_rsi",
"tq_xhat_macd",
"tq_xhat_macd_signal",
"tq_xhat_volatility",
"tq_xhat_atr",
"tqx_trend",
"tqx_momentum",
]
dataframe[numeric_cols] = dataframe[numeric_cols].replace([np.inf, -np.inf], np.nan)
dataframe[numeric_cols] = dataframe[numeric_cols].fillna(0.0)
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
conditions = [
dataframe["volume"] > 0,
dataframe["volume_ratio"] > 1.25,
dataframe["tq_xhat_rsi"].between(-0.10, 0.30),
dataframe["tqx_trend"] >= self.tqx_min_trend,
dataframe["tqx_momentum"] > 0.14,
dataframe["tq_xhat_volatility"] <= self.tqx_max_volatility,
dataframe["tq_xhat_atr"] <= self.tqx_max_atr,
dataframe["ohlcv_body"] > -0.005,
]
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["tqx_trend"] <= self.tqx_exit_trend)
| (dataframe["tqx_momentum"] < -0.03)
| (dataframe["tq_xhat_volatility"] > 0.60)
| (dataframe["tq_xhat_atr"] > 0.60)
),
]
dataframe.loc[
pd.concat(conditions, axis=1).all(axis=1),
"exit_long",
] = 1
return dataframe