jack/Chan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
6e1c335d2837f2b99f76fb64041d19bb6dd0b55c
Chan/strategies/ChanLun_BTC_30.py
T
jackyu66git 6e1c335d28 Change some thing
2025-06-19 18:39:02 +08:00

330 lines
14 KiB
Python
Raw Blame History

# --- Do not remove these libs ---
from statistics import median
from freqtrade.strategy import IStrategy
import sys
import os
# 添加父目录到系统路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from ChanLun import ChanLun
from ChanLun_Classifier import ChanLunClassifier
from ChanEnum import Chan_FX_TYPE, Chan_KLC_FX, Chan_BI_DIR, Chan_KLC_FX
from ChanPY import ChanPY
# --------------------------------
from technical.util import resample_to_interval, resampled_merge
import talib.abstract as ta
from pandas import DataFrame
from datetime import datetime, timedelta
from freqtrade.persistence import Trade, Order
from typing import Optional
import logging
logger = logging.getLogger(__name__)
### Now you can use logger.info('asfd') to log
# freqtrade plot-dataframe --strategy ChanLun_BTC_30 --datadir user_data/data/binance -c ./user_data/ChanLun_SOL_30.json --timerange=20250309-
# freqtrade trade -c ./user_data/Chan/config/ChanLun_BTC_30.json --strategy ChanLun_BTC_30 --strategy-path ./user_data/Chan/strategies
# freqtrade backtesting -c ./user_data/Chan/config/ChanLun_BTC_30.json --strategy ChanLun_BTC_30 --strategy-path ./user_data/Chan/strategies --timerange=20250520-
# freqtrade download-data -c ./user_data/Chan/config/ChanLun_BTC_30.json -t 1m --pairs BTC/USDT:USDT --timerange=20250405-
# freqtrade hyperopt --hyperopt-loss SharpeHyperOptLossDaily --spaces roi stoploss --strategy ChanLun_BTC_30 --strategy-path ./user_data/Chan/strategies -c ./user_data/Chan/config/ChanLun_BTC_30.json -e 200 --timerange=20250201-20250401
# sudo docker compose run --rm chan_btc backtesting -c ./user_data/Chan/config/ChanLun_BTC_30.json --strategy ChanLun_BTC_30 --strategy-path ./user_data/Chan/strategies --timerange=20250101-
# sudo docker compose run --rm chan_btc download-data -c ./user_data/Chan/config/ChanLun_BTC_30.json --pairs BTC/USDT:USDT -t 1m --timerange 20240101-
# sudo docker compose run --rm chan_btc trade -c ./user_data/Chan/config/ChanLun_BTC_30.json --strategy ChanLun_BTC_30 --strategy-path ./user_data/Chan/strategies
class ChanLun_BTC_30(IStrategy):
INTERFACE_VERSION: int = 3
# Minimal ROI designed for the strategy.
# This attribute will be overridden if the config file contains "minimal_roi"
# 30m and 1h
minimal_roi = {
"0": 0.60,
"360": 0.2,
"640": 0.1,
"1200": 0
}
# 5m and 15m
minimal_roi = {
"0": 0.1,
"60": 0.05,
"120": 0.02,
"240": 0
}
# 15m and 30m
minimal_roi_1 = {
"0": 0.1,
"240": 0.05,
"480": 0.03,
"600": 0
}
minimal_roi_1 = {
"0": 0.10,
"1200": 0.05,
"2400": 0.025,
"3600": 0
}
can_short = False
lev = 2.0
stoploss = -0.5
trailing_stop = False
trailing_stop_positive = 0.025
trailing_stop_positive_offset = 0.045
trailing_only_offset_is_reached = False
position_adjustment_enable = True
startup_candle_count = 600
time5 = 5
time15 = 15
time30 = 30
time60 = 60
time4h = 240
last_time = datetime.now()
chan = ChanLun()
chanpy = ChanPY()
classifier = ChanLunClassifier(None)
last_trade = None
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
# resample our dataframes
dataframe_5 = resample_to_interval(dataframe, self.get_ticker_indicator() * 5)
dataframe_15 = resample_to_interval(dataframe, self.get_ticker_indicator() * 15)
dataframe_30 = resample_to_interval(dataframe, self.get_ticker_indicator() * 30)
dataframe_60 = resample_to_interval(dataframe, self.get_ticker_indicator() * 60)
dataframe_4h = resample_to_interval(dataframe, self.get_ticker_indicator() * 240)
#dataframe_1d = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe='1d')
#dataframe_1w = resample_to_interval(dataframe_1d, self.get_ticker_indicator() * 10080)
#dataframe_1m = resample_to_interval(dataframe_1d, self.get_ticker_indicator() * 43200)
dataframe_1d = resample_to_interval(dataframe, self.get_ticker_indicator() * 1440)
#dataframe_1w = resample_to_interval(dataframe, self.get_ticker_indicator() * 10080)
#dataframe_1m = resample_to_interval(dataframe, self.get_ticker_indicator() * 43200)
dataframe = self.add_indicators(dataframe)
dataframe_5 = self.add_indicators(dataframe_5)
dataframe_30 = self.add_indicators(dataframe_30)
dataframe_60 = self.add_indicators(dataframe_60)
dataframe_4h = self.add_indicators(dataframe_4h)
dataframe_1d = self.add_indicators(dataframe_1d)
#self.chan.plot_dual(dataframe_5, dataframe_30)
chanpy_state = self.chanpy.get_bsp_state(dataframe_5)
dataframe_5['chanpy_state'] = chanpy_state
state_list, fx_list = self.chan.get_klc_strength_list(dataframe_30)
dataframe_30['state'] = state_list
dataframe_30['fx'] = fx_list
#bi_list_1 = self.chan.get_bi_list(dataframe)
#bi_list_5 = self.chan.get_bi_list(dataframe_5)
#bi_list_15 = self.chan.get_bi_list(dataframe_15)
#bi_list_30 = self.chan.get_bi_list(dataframe_30)
#bi_list_60 = self.chan.get_bi_list(dataframe_60)
if self.last_time + timedelta(minutes=1) < datetime.now():
#self.print_bi(bi_list_1)
#self.print_bi(bi_list_5)
#self.print_bi(bi_list_15)
#self.print_bi(bi_list_30)
#self.print_bi(bi_list_60)
self.print_seg(dataframe_5)
print("-------------------------------------------------------------------------------")
self.last_time = datetime.now()
dataframe = resampled_merge(dataframe, dataframe_5)
dataframe = resampled_merge(dataframe, dataframe_30)
#dataframe = resampled_merge(dataframe, dataframe_30)
#dataframe = resampled_merge(dataframe, dataframe_60)
#dataframe = resampled_merge(dataframe, dataframe_4h)
return dataframe
def print_seg(self, dataframe):
klc_list = self.chan.get_klc_list(dataframe)
bi_list = self.chan.cal_bi_list(klc_list)
seg_list = self.chan.get_seg_list(bi_list)
seg = seg_list[-1]
bi = bi_list[-1]
print(seg.start_time, seg.dir, bi.start_time, bi.dir)
def print_bi(self, bi_list):
if bi_list and len(bi_list) > 2:
bi1 = bi_list[-1]
bi2 = bi_list[-2]
print(bi1.start_time, bi1.end_time, bi1.dir, bi2.start_time, bi2.end_time, bi2.dir)
def add_indicators(self, df):
fast = 8
slow = 16
period = 6
macd = ta.MACD(df, fastperiod=fast, slowperiod=slow, signalperiod=period)
df['macd'] = macd['macd']
df['macdsignal'] = macd['macdsignal']
df['macdhist'] = macd['macdhist']
df['ma5'] = ta.MA(df, timeperiod=5)
df['ma10'] = ta.MA(df, timeperiod=10)
df['ma30'] = ta.EMA(df, timeperiod=30)
df['ma250'] = ta.MA(df, timeperiod=250)
df['rsi'] = ta.RSI(df, timeperiod=14)
df['volume_ratio'] = self.cal_volume_ratio(df)
return df
def cal_volume_ratio(self, dataframe, window=10):
df = dataframe.copy()
# 计算过去N根K线的平均成交量
df['avg_volume'] = df['volume'].rolling(window=window).mean()
# 计算量比
df['volume_ratio'] = df['volume'] / df['avg_volume']
# 填充缺失值（前N根K线）
df['volume_ratio'] = df['volume_ratio'].fillna(1.0)
return df['volume_ratio']
def custom_entry_price(self, pair: str, trade: Trade | None, current_time: datetime, proposed_rate: float,
entry_tag: str | None, side: str, **kwargs) -> float:
new_entryprice = proposed_rate
if trade:
if trade.is_short:
new_entryprice = proposed_rate - 50
else:
new_entryprice = proposed_rate + 50
return new_entryprice
def custom_exit_price(self, pair: str, trade: Trade,
current_time: datetime, proposed_rate: float,
current_profit: float, exit_tag: str | None, **kwargs) -> float:
new_exitprice = proposed_rate
if trade:
if trade.is_short:
new_exitprice = proposed_rate + 50
else:
new_exitprice = proposed_rate - 50
return new_exitprice
def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float,
time_in_force: str, current_time: datetime, entry_tag: str | None,
side: str, **kwargs) -> bool:
if self.last_trade:
if self.last_trade.is_short:
if side == 'short':
if self.last_trade.open_date + timedelta(minutes=30) > current_time:
return False
else:
return True
else:
if side == 'long':
if self.last_trade.open_date + timedelta(minutes=30) > current_time:
return True
else:
return False
#if self.last_trade:
#print(self.last_trade.open_date, current_time, self.last_trade.open_date + timedelta(minutes=self.time5))
return True
def custom_exit1(self, pair: str, trade: Trade, current_time: datetime, current_rate: float,
current_profit: float, **kwargs):
#dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
#last_candle = dataframe.iloc[-1].squeeze()
"""
# Above 20% profit, sell when rsi < 80
if current_profit > 0.2:
if last_candle["rsi"] < 80:
return "rsi_below_80"
# Between 2% and 10%, sell if EMA-long above EMA-short
if 0.02 < current_profit < 0.1:
if last_candle["emalong"] > last_candle["emashort"]:
return "ema_long_below_80"
# Sell any positions at a loss if they are held for more than one day.
if current_profit < 0.0 and (current_time - trade.open_date_utc).days >= 1:
return "unclog"
"""
if trade.is_short:
last_high = trade.get_custom_data(key="entry_candle_high")
if current_rate > last_high:
#print(trade.open_date, last_high, current_rate, "Relay Top FX exit")
return "Relay Top FX exit"
else:
last_low = trade.get_custom_data(key="entry_candle_low")
if current_rate < last_low:
#print(trade.open_date, last_low, current_rate, "Relay Bottom FX exit")
return "Relay Bottom FX exit"
def order_filled(self, pair: str, trade: Trade, order: Order, current_time: datetime, **kwargs) -> None:
"""
Called right after an order fills.
Will be called for all order types (entry, exit, stoploss, position adjustment).
:param pair: Pair for trade
:param trade: trade object.
:param order: Order object.
:param current_time: datetime object, containing the current datetime
:param **kwargs: Ensure to keep this here so updates to this won't break your strategy.
"""
# Obtain pair dataframe (just to show how to access it)
dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe)
#last_candle = dataframe.iloc[-1].squeeze()
klc_list = self.chan.get_klc_list(resample_to_interval(dataframe, self.get_ticker_indicator() * 30))
bi_list = self.chan.cal_bi_list(klc_list)
last_high = klc_list[-2].high
last_low = klc_list[-2].low
if trade.is_short:
if (trade.nr_of_successful_entries == 1) and (order.ft_order_side == trade.entry_side):
trade.set_custom_data(key="entry_candle_high", value=last_high)
else:
if (trade.nr_of_successful_entries == 1) and (order.ft_order_side == trade.entry_side):
trade.set_custom_data(key="entry_candle_low", value=last_low)
#print(trade.open_date, trade.close_date, last_high, last_low, order.ft_order_side, klc_list[-2].start_time, klc_list[-2].end_time)
self.last_trade = trade
return None
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
state_str = 'resample_{}_state'.format(self.get_ticker_indicator()*self.time30)
fx_str = 'resample_{}_fx'.format(self.get_ticker_indicator()*self.time30)
#chanpy_state_str = 'resample_{}_chanpy_state'.format(self.get_ticker_indicator()*self.time5)
shift_time = self.time30
strength = 0.9
dataframe.loc[
(
#(dataframe['state'] == "-30")
(dataframe[state_str].shift(shift_time) > strength) &
(dataframe[fx_str].shift(shift_time) == -1)
#(dataframe[chanpy_state_str].shift(shift_time+30) == 1)
#(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time5)].shift(self.time5) == "-10") &
#(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time30)] == "-10") &
#(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time5)].shift(self.time5) == "-10")
#(qtpylib.crossed_above(dataframe['macd'], dataframe['macdsignal']))
),
['enter_long', 'enter_tag']] = (1, 'long_signal_chan')
dataframe.loc[
(
#(dataframe['state'] == "-30")
(dataframe[state_str].shift(shift_time) > strength) &
(dataframe[fx_str].shift(shift_time) == 1)
#(dataframe[chanpy_state_str].shift(shift_time+30) == -1)
#(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time5)].shift(self.time5) == "-10") &
#(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time30)] == "-10") &
#(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time5)].shift(self.time5) == "-10")
#(qtpylib.crossed_above(dataframe['macd'], dataframe['macdsignal']))
),
['enter_short', 'enter_tag']] = (1, 'short_signal_chan')
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
state_str = 'resample_{}_state'.format(self.get_ticker_indicator()*self.time30)
fx_str = 'resample_{}_fx'.format(self.get_ticker_indicator()*self.time30)
#chanpy_state_str = 'resample_{}_chanpy_state'.format(self.get_ticker_indicator()*self.time5)
shift_time = self.time30
strength = 0.9
dataframe.loc[
(
#(dataframe['state']== "30")
(dataframe[state_str].shift(shift_time) > strength) &
(dataframe[fx_str].shift(shift_time) == 1)
#(dataframe[chanpy_state_str].shift(shift_time+30) == -1)
#(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time30)] == "10") &
#(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time60)] == "10")
),
['exit_long', 'exit_tag']] = (1, 'long_close_signal_chan')
dataframe.loc[
(
#(dataframe['state']== "30")
(dataframe[state_str].shift(shift_time) > strength) &
(dataframe[fx_str].shift(shift_time) == -1)
#(dataframe[chanpy_state_str].shift(shift_time+30) == 1)
#(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time30)] == "10") &
#(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time60)] == "10")
),
['exit_short', 'exit_tag']] = (1, 'short_close_signal_chan')
return dataframe
def leverage(self, pair: str, current_time: datetime, current_rate: float,
proposed_leverage: float, max_leverage: float, entry_tag: Optional[str], side: str,
**kwargs) -> float:
return self.lev
def get_ticker_indicator(self):
return int(self.timeframe[:-1])
Reference in New Issue View Git Blame Copy Permalink