phyer/myTestFreqAI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

趋势函数独立出来, 95分以上不退出

Browse Source
This commit is contained in:
zhangkun9038@dingtalk.com 2025-08-30 15:27:20 +08:00
parent 7aa9ebe55b
commit 2bb31da591
1 changed files with 53 additions and 127 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

180
freqtrade/templates/freqaiprimer.py
View File

@ -2,9 +2,8 @@ import warnings
warnings.filterwarnings("ignore", category=UserWarning, module="pandas_ta")
import logging
import pandas as pd
from freqtrade.strategy import IStrategy
from pandas import DataFrame, Series
from pandas import DataFrame
import pandas_ta as ta
from freqtrade.persistence import Trade
import numpy as np
@ -121,128 +120,52 @@ class FreqaiPrimer(IStrategy):
(dataframe['open'] < dataframe['close'].shift(1))
)
# 计算市场评分并更新到dataframe
market_score = self.calculate_market_score(dataframe, metadata)
# 计算各时间框架的趋势状态（牛/熊）
# 3m时间框架：ema50下穿ema200为熊，上穿为牛
dataframe['trend_3m'] = np.where(dataframe['ema_50_3m'] > dataframe['ema_200_3m'], 1, 0)
# 15m时间框架：ema50下穿ema200为熊，上穿为牛
dataframe['trend_15m'] = np.where(dataframe['ema_50_15m'] > dataframe['ema_200_15m'], 1, 0)
# 1h时间框架：ema50下穿ema200为熊，上穿为牛
dataframe['trend_1h_ema'] = np.where(dataframe['ema_50_1h'] > dataframe['ema_200_1h'], 1, 0)
# 计算熊牛得分（0-100）
# 权重：3m熊牛权重10，15m熊牛权重35，1h熊牛权重65
# 计算加权得分
dataframe['market_score'] = (
dataframe['trend_3m'] * 10 +
dataframe['trend_15m'] * 35 +
dataframe['trend_1h_ema'] * 65
)
# 确保得分在0-100范围内
dataframe['market_score'] = dataframe['market_score'].clip(lower=0, upper=100)
# 根据得分分类市场状态
dataframe['market_state'] = 'neutral'
dataframe.loc[dataframe['market_score'] > 70, 'market_state'] = 'strong_bull'
dataframe.loc[(dataframe['market_score'] > 50) & (dataframe['market_score'] <= 70), 'market_state'] = 'weak_bull'
dataframe.loc[(dataframe['market_score'] >= 30) & (dataframe['market_score'] <= 50), 'market_state'] = 'neutral'
dataframe.loc[(dataframe['market_score'] > 10) & (dataframe['market_score'] < 30), 'market_state'] = 'weak_bear'
dataframe.loc[dataframe['market_score'] <= 10, 'market_state'] = 'strong_bear'
# 记录当前的市场状态
if len(dataframe) > 0:
current_score = dataframe['market_score'].iloc[-1]
current_state = dataframe['market_state'].iloc[-1]
logger.info(f"[{metadata['pair']}] 熊牛得分: {current_score:.1f}, 市场状态: {current_state}")
logger.info(f"[{metadata['pair']}] 各时间框架趋势: 3m={'牛' if dataframe['trend_3m'].iloc[-1] == 1 else '熊'}, \
15m={'牛' if dataframe['trend_15m'].iloc[-1] == 1 else '熊'}, \
1h={'牛' if dataframe['trend_1h_ema'].iloc[-1] == 1 else '熊'}")
# 调试：打印指标值（最后 5 行）
print(f"Pair: {metadata['pair']}, Last 5 rows:")
print(dataframe[['date', 'close', 'bb_lower_3m', 'rsi_3m', 'rsi_15m', 'rsi_1h', 'trend_1h',
'trend_3m', 'trend_15m', 'market_score', 'market_state',
'trend_3m', 'trend_15m', 'trend_1h_ema', 'market_score', 'market_state',
'bullish_engulfing', 'volume', 'volume_ma']].tail(5))
return dataframe
def calculate_market_score(self, dataframe: DataFrame, metadata: dict) -> int:
"""
计算市场评分（0-100的整数）
参数:
- dataframe: 数据框
- metadata: 元数据
返回:
- int: 市场评分(0-100)
"""
# 确保数据框不为空
if len(dataframe) == 0:
return 50 # 默认返回中性分数
# 计算各时间框架的趋势状态（0-100归一化）
dataframe = self.calculate_timeframe_score(dataframe, '3m', 'ema_50_3m', 'ema_200_3m')
dataframe = self.calculate_timeframe_score(dataframe, '15m', 'ema_50_15m', 'ema_200_15m')
dataframe = self.calculate_timeframe_score(dataframe, '1h', 'ema_50_1h', 'ema_200_1h')
# 计算加权综合得分（权重：3m=10，15m=35，1h=65）
market_score = (
dataframe['trend_score_3m'].iloc[-1] * 10 +
dataframe['trend_score_15m'].iloc[-1] * 35 +
dataframe['trend_score_1h'].iloc[-1] * 65
)
# 确保得分在0-100范围内并转为整数
market_score = int(max(0, min(100, market_score)))
# 更新dataframe中的market_score和market_state列
dataframe['market_score'] = market_score
dataframe['market_state'] = self.classify_market_state(pd.Series([market_score]))
# 记录当前的市场状态
current_state = dataframe['market_state'].iloc[-1]
logger.info(f"[{metadata['pair']}] 熊牛得分: {market_score}, 市场状态: {current_state}")
# 获取各时间框架的趋势状态
trend_3m = 1 if dataframe['ema_50_3m'].iloc[-1] > dataframe['ema_200_3m'].iloc[-1] else 0
trend_15m = 1 if dataframe['ema_50_15m'].iloc[-1] > dataframe['ema_200_15m'].iloc[-1] else 0
trend_1h_ema = 1 if dataframe['ema_50_1h'].iloc[-1] > dataframe['ema_200_1h'].iloc[-1] else 0
logger.info(f"[{metadata['pair']}] 各时间框架趋势: 3m={'牛' if trend_3m == 1 else '熊'}, \
15m={'牛' if trend_15m == 1 else '熊'}, \
1h={'牛' if trend_1h_ema == 1 else '熊'}")
return market_score
def get_market_score(self, dataframe: DataFrame, metadata: dict) -> int:
"""
返回市场评分（0-100的整数）
参数:
- dataframe: 数据框
- metadata: 元数据
返回:
- int: 市场评分(0-100)
"""
# 确保数据框不为空
if len(dataframe) == 0:
return 50 # 默认返回中性分数
# 计算各时间框架的趋势
trend_3m = 1 if dataframe['ema_50_3m'].iloc[-1] > dataframe['ema_200_3m'].iloc[-1] else 0
trend_15m = 1 if dataframe['ema_50_15m'].iloc[-1] > dataframe['ema_200_15m'].iloc[-1] else 0
trend_1h_ema = 1 if dataframe['ema_50_1h'].iloc[-1] > dataframe['ema_200_1h'].iloc[-1] else 0
# 计算加权综合得分（权重：3m=10，15m=35，1h=65）
market_score = (
trend_3m * 10 +
trend_15m * 35 +
trend_1h_ema * 65
)
# 确保得分在0-100范围内并转为整数
market_score = int(max(0, min(100, market_score)))
return market_score
def calculate_timeframe_score(self, dataframe: DataFrame, timeframe: str, ema_short: str, ema_long: str) -> DataFrame:
"""
计算单个时间框架的趋势得分(0-100)
"""
# 基本趋势判断（1为牛，0为熊）
column_name = f'trend_{timeframe}'
score_column = f'trend_score_{timeframe}'
dataframe[column_name] = np.where(dataframe[ema_short] > dataframe[ema_long], 1, 0)
# 将趋势值归一化为0-100分
dataframe[score_column] = dataframe[column_name] * 100
return dataframe
def classify_market_state(self, market_score: Series) -> Series:
"""
根据市场得分分类市场状态
"""
market_state = np.where(
market_score > 70, 'strong_bull',
np.where(
market_score > 50, 'weak_bull',
np.where(
market_score >= 30, 'neutral',
np.where(
market_score > 10, 'weak_bear',
'strong_bear'
)
)
)
)
return pd.Series(market_state, index=market_score.index)
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
# 做多条件（放宽以增加入场信号）
@ -305,14 +228,13 @@ class FreqaiPrimer(IStrategy):
)
# 强劲趋势条件：1h趋势向上 + 熊牛得分>70
# 使用EMA交叉来判断趋势，而不是不存在的trend_1h_ema列
strong_trend = (dataframe['ema_50_1h'] > dataframe['ema_200_1h']) & (dataframe['market_score'] > 70)
strong_trend = (dataframe['trend_1h_ema'] == 1) & (dataframe['market_score'] > 70)
# 一般趋势条件：熊牛得分50-70
normal_trend = (dataframe['market_score'] >= 50) & (dataframe['market_score'] <= 70)
# 极强劲趋势条件：熊牛得分>95，此时不给退出信号
extreme_strong_trend = dataframe['market_score'] > 95
# 极端强劲趋势条件：熊牛得分>95
extreme_strong_trend = (dataframe['market_score'] > 95)
# 获取15m数据进行趋势确认
df_15m = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe='15m')
@ -351,17 +273,16 @@ class FreqaiPrimer(IStrategy):
trend_reversal = merged_data['rsi_15m'] > 75
# 动态调整退出条件 - 提高ATR倍数以增加单笔盈利潜力
# 强劲趋势中：大幅提高ATR倍数到5.5倍，充分利用趋势利润，但当趋势极强时不给退出信号
# 强劲趋势中：大幅提高ATR倍数到5.5倍，充分利用趋势利润
# 但在极端强劲趋势(market_score>95)下不给出退出信号
dataframe.loc[strong_trend & ~extreme_strong_trend & ((dataframe['close'] > dataframe['open'] + dataframe['atr'] * 5.5) | (basic_exit & trend_reversal)), 'exit_long'] = 1
# 一般趋势中：提高到4倍ATR退出
dataframe.loc[normal_trend & ((dataframe['close'] > dataframe['open'] + dataframe['atr'] * 4) | basic_exit), 'exit_long'] = 1
# 记录极端强劲趋势阻止退出的情况
if len(dataframe[extreme_strong_trend]) > 0:
logger.info(f"[{metadata['pair']}] 熊牛得分>95，处于极端强劲趋势，不给出退出信号")
# 但在极端强劲趋势(market_score>95)下不给出退出信号
dataframe.loc[normal_trend & ~extreme_strong_trend & ((dataframe['close'] > dataframe['open'] + dataframe['atr'] * 4) | basic_exit), 'exit_long'] = 1
# 非趋势或弱势：保持2倍ATR退出
# 但在极端强劲趋势(market_score>95)下不给出退出信号
# 记录退出决策日志
if len(dataframe[dataframe['exit_long'] == 1]) > 0:
@ -369,6 +290,11 @@ class FreqaiPrimer(IStrategy):
current_state = dataframe['market_state'].iloc[-1]
logger.info(f"[{metadata['pair']}] 触发退出信号，市场状态: {current_state}, 价格: {last_exit['close']:.2f}")
# 记录因极端强劲趋势阻止的退出信号
extreme_strong_prevented = dataframe[extreme_strong_trend & ((strong_trend & ((dataframe['close'] > dataframe['open'] + dataframe['atr'] * 5.5) | (basic_exit & trend_reversal))) | (normal_trend & ((dataframe['close'] > dataframe['open'] + dataframe['atr'] * 4) | basic_exit)))]
if len(extreme_strong_prevented) > 0:
logger.info(f"[{metadata['pair']}] 由于极端强劲趋势(market_score>95)，阻止了 {len(extreme_strong_prevented)} 个潜在的退出信号")
return dataframe
def detect_h1_rapid_rise(self, pair: str, dataframe: DataFrame, metadata: dict) -> tuple[bool, float]: