53 lines
2.6 KiB
Python
53 lines
2.6 KiB
Python
from freqtrade.strategy import IStrategy
|
|
from pandas_ta import ema
|
|
import pandas as pd
|
|
import pandas_ta as ta
|
|
import numpy as np
|
|
from datetime import datetime, timedelta
|
|
from freqtrade.persistence import Trade, Order
|
|
# freqtrade backtesting -c ./user_data/Chan/config/ChanLun_BTC_30.json --strategy HammerRsiStrategy --strategy-path ./user_data/Chan/strategies --timerange=20250520-
|
|
|
|
|
|
class HammerRsiStrategy(IStrategy):
|
|
timeframe = "1m" # 1分钟K线
|
|
minimal_roi = {"0": 0.005} # 0.5% 止盈
|
|
stoploss = -0.002 # 0.2% 固定止损
|
|
trailing_stop = True
|
|
trailing_stop_positive = 0.001 # 0.1% 追踪止损
|
|
trailing_stop_positive_offset = 0.002 # 0.2% 触发追踪止损
|
|
startup_candle_count = 20 # 启动K线数
|
|
|
|
def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
|
|
dataframe['rsi'] = ta.rsi(dataframe['close'], length=14)
|
|
dataframe['ema_fast'] = ta.ema(dataframe['close'], length=5)
|
|
dataframe['ema_slow'] = ta.ema(dataframe['close'], length=20)
|
|
dataframe['atr'] = ta.atr(dataframe['high'], dataframe['low'], dataframe['close'], length=14)
|
|
return dataframe
|
|
|
|
def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
|
|
conditions = (
|
|
(dataframe['ema_fast'] > dataframe['ema_slow']) & # 快EMA上穿慢EMA
|
|
(dataframe['rsi'] < 45) # RSI < 45
|
|
)
|
|
print(f"Signal check: ema_fast={dataframe['ema_fast'].iloc[-1]}, ema_slow={dataframe['ema_slow'].iloc[-1]}, rsi={dataframe['rsi'].iloc[-1]}")
|
|
dataframe.loc[conditions, ['enter_long', 'enter_tag']] = (1, 'ema_rsi_entry')
|
|
return dataframe
|
|
|
|
def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
|
|
conditions = (
|
|
(dataframe['ema_fast'] < dataframe['ema_slow']) | # 快EMA下穿慢EMA
|
|
(dataframe['rsi'] > 60) # RSI > 60
|
|
)
|
|
dataframe.loc[conditions, ['exit_long', 'exit_tag']] = (1, 'ema_rsi_exit')
|
|
return dataframe
|
|
|
|
def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime,
|
|
current_rate: float, current_profit: float, **kwargs) -> float:
|
|
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
|
|
atr = dataframe['atr'].iloc[-1]
|
|
return -1.5 * atr / current_rate # 止损为1.5倍ATR
|
|
|
|
def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float,
|
|
proposed_stake: float, min_stake: float, max_stake: float,
|
|
entry_tag: str) -> float:
|
|
return proposed_stake * 0.01 # 1%账户余额 |