286 lines
11 KiB
Python
286 lines
11 KiB
Python
# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
|
|
import logging
|
|
from functools import reduce
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
import numpy as np
|
|
import talib.abstract as ta
|
|
from pandas import DataFrame
|
|
|
|
from freqtrade.strategy import IStrategy, merge_informative_pair
|
|
from freqtrade.persistence import Trade
|
|
|
|
# freqtrade backtesting -c ./user_data/Chan/config/FreqAI_Test.json --strategy CryptoFuturesAIStrategy --strategy-path ./user_data/Chan/strategies --freqaimodel LightGBMRegressor --timerange=20260201-
|
|
# freqtrade trade -c ./user_data/Chan/config/FreqAI_Test.json --strategy CryptoFuturesAIStrategy --strategy-path ./user_data/Chan/strategies --freqaimodel LightGBMRegressor
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class CryptoFuturesAIStrategy(IStrategy):
|
|
"""
|
|
SOL/USDT 合约 AI 策略 - FreqAI + LightGBM
|
|
|
|
核心思路：
|
|
1. 用 FreqAI 的 LightGBM 回归模型预测未来价格变化方向和幅度
|
|
2. 模型自动在滚动窗口上重新训练，适应市场变化
|
|
3. 结合传统技术指标作为特征输入，让 AI 学习最优组合
|
|
4. 用 z-score 动态阈值代替固定参数，自适应不同市场环境
|
|
5. 保留 trailing_stop 作为风控（这是原策略的盈利核心）
|
|
|
|
相比固定参数策略的优势：
|
|
- 参数自适应：模型每隔一段时间重新训练，适应市场状态变化
|
|
- 特征自动选择：LightGBM 自动学习哪些指标在当前市场最有用
|
|
- 动态阈值：用预测值的统计分布来决定入场，而非固定数值
|
|
- 多维度输入：同时考虑价格、成交量、波动率、时间等多维信息
|
|
"""
|
|
INTERFACE_VERSION = 3
|
|
timeframe = '5m' # FreqAI 用5分钟作为基础时间框架，更稳定
|
|
can_short = True
|
|
|
|
# === 风控参数（保留原策略的盈利核心） ===
|
|
stoploss = -0.025
|
|
trailing_stop = True
|
|
trailing_stop_positive = 0.008
|
|
trailing_stop_positive_offset = 0.015
|
|
trailing_only_offset_is_reached = True
|
|
|
|
use_custom_stoploss = False
|
|
use_exit_signal = False # 禁用exit_signal，让trailing_stop管理退出
|
|
|
|
process_only_new_candles = True
|
|
startup_candle_count: int = 100 # 需要足够的历史数据计算指标
|
|
|
|
# =====================================================
|
|
# FreqAI 特征工程函数
|
|
# =====================================================
|
|
|
|
def feature_engineering_expand_all(
|
|
self, dataframe: DataFrame, period: int, metadata: dict, **kwargs
|
|
) -> DataFrame:
|
|
"""
|
|
自动扩展特征 - 精简版，减少特征数量防止内存溢出
|
|
"""
|
|
# 核心动量指标
|
|
dataframe["%-rsi-period"] = ta.RSI(dataframe, timeperiod=period)
|
|
dataframe["%-adx-period"] = ta.ADX(dataframe, timeperiod=period)
|
|
dataframe["%-ema-period"] = ta.EMA(dataframe, timeperiod=period)
|
|
dataframe["%-roc-period"] = ta.ROC(dataframe, timeperiod=period)
|
|
|
|
# 相对成交量
|
|
dataframe["%-relative_volume-period"] = (
|
|
dataframe["volume"] / dataframe["volume"].rolling(period).mean()
|
|
)
|
|
|
|
return dataframe
|
|
|
|
def feature_engineering_expand_basic(
|
|
self, dataframe: DataFrame, metadata: dict, **kwargs
|
|
) -> DataFrame:
|
|
"""
|
|
基础特征 - 在所有时间框架上展开，但不按周期展开
|
|
"""
|
|
# 价格变化率
|
|
dataframe["%-pct-change"] = dataframe["close"].pct_change()
|
|
dataframe["%-raw_volume"] = dataframe["volume"]
|
|
dataframe["%-raw_price"] = dataframe["close"]
|
|
|
|
# K线形态特征
|
|
dataframe["%-candle_body"] = (
|
|
(dataframe["close"] - dataframe["open"]) / dataframe["open"]
|
|
)
|
|
dataframe["%-upper_shadow"] = (
|
|
(dataframe["high"] - dataframe[["open", "close"]].max(axis=1))
|
|
/ dataframe["close"]
|
|
)
|
|
dataframe["%-lower_shadow"] = (
|
|
(dataframe[["open", "close"]].min(axis=1) - dataframe["low"])
|
|
/ dataframe["close"]
|
|
)
|
|
|
|
# 价格与高低点的关系
|
|
dataframe["%-high_low_range"] = (
|
|
(dataframe["high"] - dataframe["low"]) / dataframe["close"]
|
|
)
|
|
|
|
return dataframe
|
|
|
|
def feature_engineering_standard(
|
|
self, dataframe: DataFrame, metadata: dict, **kwargs
|
|
) -> DataFrame:
|
|
"""
|
|
标准特征 - 不自动展开，只在基础时间框架上计算一次
|
|
适合放时间特征等不需要跨时间框架的特征
|
|
"""
|
|
# 时间特征（让模型学习时间规律）
|
|
dataframe["%-day_of_week"] = dataframe["date"].dt.dayofweek
|
|
dataframe["%-hour_of_day"] = dataframe["date"].dt.hour
|
|
dataframe["%-minute_of_hour"] = dataframe["date"].dt.minute
|
|
|
|
# 是否是高波动时段（美国开市等）
|
|
hour = dataframe["date"].dt.hour
|
|
dataframe["%-is_us_session"] = (
|
|
((hour >= 13) & (hour <= 21)) # UTC 13-21 = 美东 8am-4pm
|
|
).astype(int)
|
|
dataframe["%-is_asia_session"] = (
|
|
((hour >= 0) & (hour <= 8)) # UTC 0-8 = 亚洲时段
|
|
).astype(int)
|
|
|
|
# 连续涨跌统计
|
|
pct = dataframe["close"].pct_change()
|
|
dataframe["%-consec_up"] = (pct > 0).astype(int)
|
|
dataframe["%-consec_up"] = dataframe["%-consec_up"].groupby(
|
|
(dataframe["%-consec_up"] != dataframe["%-consec_up"].shift()).cumsum()
|
|
).cumcount() + 1
|
|
dataframe["%-consec_up"] = dataframe["%-consec_up"] * (pct > 0).astype(int)
|
|
|
|
dataframe["%-consec_down"] = (pct < 0).astype(int)
|
|
dataframe["%-consec_down"] = dataframe["%-consec_down"].groupby(
|
|
(dataframe["%-consec_down"] != dataframe["%-consec_down"].shift()).cumsum()
|
|
).cumcount() + 1
|
|
dataframe["%-consec_down"] = dataframe["%-consec_down"] * (pct < 0).astype(int)
|
|
|
|
# 近期波动率变化
|
|
dataframe["%-vol_change_5"] = (
|
|
dataframe["volume"].rolling(5).mean()
|
|
/ dataframe["volume"].rolling(20).mean()
|
|
)
|
|
|
|
# 价格距离近期高低点
|
|
dataframe["%-dist_high_20"] = (
|
|
dataframe["close"] / dataframe["high"].rolling(20).max() - 1
|
|
)
|
|
dataframe["%-dist_low_20"] = (
|
|
dataframe["close"] / dataframe["low"].rolling(20).min() - 1
|
|
)
|
|
|
|
return dataframe
|
|
|
|
def set_freqai_targets(self, dataframe: DataFrame, metadata: dict, **kwargs) -> DataFrame:
|
|
"""
|
|
设置 AI 模型的预测目标
|
|
|
|
目标：预测未来 N 根K线的平均价格变化率
|
|
模型会学习：当前市场状态 → 未来价格走向
|
|
"""
|
|
label_period = self.freqai_info["feature_parameters"]["label_period_candles"]
|
|
|
|
# 回归目标：未来 N 根K线的平均收盘价相对当前的变化率
|
|
dataframe["&-s_close"] = (
|
|
dataframe["close"]
|
|
.shift(-label_period)
|
|
.rolling(label_period)
|
|
.mean()
|
|
/ dataframe["close"]
|
|
- 1
|
|
)
|
|
|
|
return dataframe
|
|
|
|
# =====================================================
|
|
# 策略核心逻辑
|
|
# =====================================================
|
|
|
|
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
"""
|
|
populate_indicators 中调用 FreqAI
|
|
所有指标由 feature_engineering_*() 函数定义
|
|
"""
|
|
# FreqAI 会自动调用所有 feature_engineering_*() 函数
|
|
# 然后训练模型并返回预测结果
|
|
dataframe = self.freqai.start(dataframe, metadata, self)
|
|
|
|
# 计算动态阈值（z-score 方式）
|
|
# &-s_close 是模型预测的未来价格变化
|
|
# &-s_close_mean 和 &-s_close_std 是训练期间的统计值
|
|
# 当预测值超过 mean + factor * std 时，说明模型认为有较强的方向性
|
|
|
|
return dataframe
|
|
|
|
def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
|
|
"""
|
|
入场信号 - 基于 AI 预测
|
|
|
|
核心逻辑：
|
|
1. do_predict == 1：模型认为当前数据在训练分布内（可信）
|
|
2. &-s_close > threshold：预测未来上涨幅度超过阈值
|
|
3. 动态阈值 = mean + 1.0 * std（约 84% 置信度）
|
|
"""
|
|
# 动态阈值：使用训练期间的统计值
|
|
# 当 &-s_close_mean 和 &-s_close_std 可用时，用 z-score
|
|
# 否则用固定阈值
|
|
if "&-s_close_mean" in df.columns and "&-s_close_std" in df.columns:
|
|
long_threshold = df["&-s_close_mean"] + df["&-s_close_std"] * 1.0
|
|
short_threshold = df["&-s_close_mean"] - df["&-s_close_std"] * 1.0
|
|
else:
|
|
long_threshold = 0.005
|
|
short_threshold = -0.005
|
|
|
|
# 做多条件
|
|
enter_long_conditions = [
|
|
df["do_predict"] == 1, # 模型预测可信
|
|
df["&-s_close"] > long_threshold, # 预测超过动态阈值
|
|
]
|
|
|
|
if enter_long_conditions:
|
|
df.loc[
|
|
reduce(lambda x, y: x & y, enter_long_conditions),
|
|
["enter_long", "enter_tag"]
|
|
] = (1, "ai_long")
|
|
|
|
# 做空条件
|
|
enter_short_conditions = [
|
|
df["do_predict"] == 1, # 模型预测可信
|
|
df["&-s_close"] < short_threshold, # 预测低于动态阈值
|
|
]
|
|
|
|
if enter_short_conditions:
|
|
df.loc[
|
|
reduce(lambda x, y: x & y, enter_short_conditions),
|
|
["enter_short", "enter_tag"]
|
|
] = (1, "ai_short")
|
|
|
|
return df
|
|
|
|
def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
|
|
"""
|
|
出场信号 - AI 预测方向反转时退出
|
|
"""
|
|
# 多头退出：预测转为下跌
|
|
exit_long_conditions = [
|
|
df["do_predict"] == 1,
|
|
df["&-s_close"] < 0, # 预测未来下跌
|
|
]
|
|
if exit_long_conditions:
|
|
df.loc[reduce(lambda x, y: x & y, exit_long_conditions), "exit_long"] = 1
|
|
|
|
# 空头退出：预测转为上涨
|
|
exit_short_conditions = [
|
|
df["do_predict"] == 1,
|
|
df["&-s_close"] > 0, # 预测未来上涨
|
|
]
|
|
if exit_short_conditions:
|
|
df.loc[reduce(lambda x, y: x & y, exit_short_conditions), "exit_short"] = 1
|
|
|
|
return df
|
|
|
|
def confirm_trade_entry(
|
|
self, pair: str, order_type: str, amount: float, rate: float,
|
|
time_in_force: str, current_time: datetime, entry_tag: Optional[str],
|
|
side: str, **kwargs
|
|
) -> bool:
|
|
"""
|
|
实盘入场确认 - 防止滑点过大
|
|
"""
|
|
df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
|
|
last_candle = df.iloc[-1].squeeze()
|
|
|
|
if side == "long":
|
|
if rate > (last_candle["close"] * (1 + 0.0025)):
|
|
return False
|
|
else:
|
|
if rate < (last_candle["close"] * (1 - 0.0025)):
|
|
return False
|
|
|
|
return True
|