410 lines
19 KiB
Python
410 lines
19 KiB
Python
# --- Do not remove these libs ---
|
|
from freqtrade.strategy import IStrategy
|
|
import sys
|
|
import os
|
|
#sys.setrecursionlimit(1000000) #例如这里设置为一百万
|
|
#sys.path.append(os.path.abspath("/freqtrade/user_data/Chan"))
|
|
#sys.path.append(os.path.abspath("/Users/jack/Project/freqtrade/user_data/Chan"))
|
|
sys.path.append(os.path.abspath("/Users/jack/Documents/GitHub/freqtrade/user_data/Chan"))
|
|
from ChanLun import ChanLun
|
|
from ChanLun_Classifier import ChanLunClassifier
|
|
from ChanEnum import Chan_FX_TYPE
|
|
# --------------------------------
|
|
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
|
|
from typing import Optional
|
|
import logging
|
|
logger = logging.getLogger(__name__)
|
|
### Now you can use logger.info('asfd') to log
|
|
# freqtrade plot-dataframe --strategy ChanLun_SOL_5 --datadir user_data/data/binance -c ./user_data/ChanLun_SOL.json --timerange=20250309-
|
|
|
|
# freqtrade trade -c ./user_data/ChanLun_SOL.json --strategy ChanLun_SOL_5 --strategy-path ./user_data/strategies
|
|
# freqtrade backtesting -c ./user_data/ChanLun_SOL.json --strategy ChanLun_SOL_5 --strategy-path ./user_data/strategies --timerange=20250416-
|
|
# freqtrade download-data -c ./user_data/ChanLun_SOL.json -t 1m --pairs SOL/USDT:USDT --timerange=20250405-
|
|
# freqtrade hyperopt --hyperopt-loss SharpeHyperOptLossDaily --spaces roi stoploss --strategy ChanLun_SOL_5 --strategy-path ./user_data/strategies -c ./user_data/ChanLun_SOL.json -e 200 --timerange=20250201-20250401
|
|
|
|
# sudo docker compose run --rm chan_btc backtesting -c ./user_data/ChanLun_SOL.json --strategy ChanLun_SOL --strategy-path ./user_data/strategies --timerange=20250101-
|
|
# sudo docker compose run --rm chan_btc download-data -c ./user_data/ChanLun_SOL.json --pairs SOL/USDT:USDT -t 1m --timerange 20240101-
|
|
# sudo docker compose run --rm chan_btc trade -c ./user_data/ChanLun_SOL.json --strategy ChanLun_SOL --strategy-path ./user_data/strategies
|
|
|
|
class ChanLun_SOL_5(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.10,
|
|
"360": 0.05,
|
|
"640": 0.025,
|
|
"1200": 0
|
|
}
|
|
# 5m and 15m
|
|
minimal_roi_1 = {
|
|
"0": 0.253,
|
|
"60": 0.159,
|
|
"120": 0.052,
|
|
"240": 0
|
|
}
|
|
# 15m and 30m
|
|
minimal_roi_1 = {
|
|
"0": 0.253,
|
|
"120": 0.159,
|
|
"240": 0.052,
|
|
"360": 0
|
|
}
|
|
can_short = True
|
|
stoploss = -0.20
|
|
trailing_stop = False
|
|
trailing_stop_positive = 0.015
|
|
trailing_stop_positive_offset = 0.043
|
|
trailing_only_offset_is_reached = False
|
|
|
|
position_adjustment_enable = True
|
|
max_entry_position_adjustment = 3
|
|
max_dca_multiplier = 5.5
|
|
startup_candle_count = 600
|
|
|
|
time5 = 5
|
|
time15 = 15
|
|
time30 = 30
|
|
time60 = 60
|
|
time4h = 240
|
|
time5 = 30
|
|
last_time = datetime.now()
|
|
big_size = 0
|
|
big_state = "00"
|
|
big_state_list = []
|
|
chan = ChanLun()
|
|
small_size = 0
|
|
small_state = "00"
|
|
small_state_list = []
|
|
classifier = ChanLunClassifier(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)
|
|
dataframe['state'] = self.chan.plot_dataframe(dataframe)
|
|
#dataframe_5['state'] = self.chan.plot_dataframe(dataframe_5)
|
|
#self.chan.print_data(dataframe_5)
|
|
#self.chan.cal_qjt(dataframe, dataframe_5)
|
|
#self.classifier.train_model(dataframe_4h, model_name="4h_model")
|
|
"""
|
|
self.classifier.train_model(dataframe_5, model_name="5m_model")
|
|
self.classifier.train_model(dataframe_30, model_name="30m_model")
|
|
self.classifier.train_model(dataframe_60, model_name="1h_model")
|
|
self.classifier.train_model(dataframe_4h, model_name="4h_model")
|
|
self.classifier.train_model(dataframe, model_name="1m_model")
|
|
self.classifier.train_model(dataframe_1d, model_name="1d_model")
|
|
"""
|
|
|
|
"""
|
|
if self.classifier.model is None:
|
|
self.classifier.train_model(dataframe_30, model_name="30m_model")
|
|
self.classifier.load_model(model_name="30m_model")
|
|
klc_list = self.chan.get_klc_list(dataframe_30)
|
|
top_avg = 0
|
|
bottom_avg = 0
|
|
top_count = 0
|
|
bottom_count = 0
|
|
for index in range(int(len(klc_list) * 0.8), len(klc_list)):
|
|
klc = klc_list[index]
|
|
if self.classifier.predict(klc) > 0.01 and klc.fx == Chan_FX_TYPE.BOTTOM:
|
|
features = klc.get_feature_data()
|
|
print(klc.end_time, klc.fx, self.classifier.predict(klc), features['klc_macd'], features['klc_macd_hist'], features['klc_rsi'], features['klc_macd_signal'])
|
|
bottom_avg += self.classifier.predict(klc)
|
|
bottom_count += 1
|
|
if self.classifier.predict(klc) > 0.05 and klc.fx == Chan_FX_TYPE.TOP:
|
|
features = klc.get_feature_data()
|
|
print(klc.end_time, klc.fx, self.classifier.predict(klc), features['klc_macd'], features['klc_macd_hist'], features['klc_rsi'], features['klc_macd_signal'])
|
|
top_avg += self.classifier.predict(klc)
|
|
top_count += 1
|
|
bottom_avg /= bottom_count
|
|
top_avg /= top_count
|
|
print(bottom_avg, top_avg)
|
|
print("-------------------------------------------------------------------------------")
|
|
"""
|
|
"""
|
|
self.print_xgb(dataframe, "1m_model")
|
|
self.print_xgb(dataframe_5, "5m_model")
|
|
self.print_xgb(dataframe_30, "30m_model")
|
|
self.print_xgb(dataframe_60, "1h_model")
|
|
self.print_xgb(dataframe_4h, "4h_model")
|
|
print("-------------------------------------------------------------------------------")
|
|
"""
|
|
|
|
#classifier.find_best_params(dataframe)
|
|
#classifier.train_model(use_cv=False)
|
|
#classifier.validate_model(dataframe)
|
|
#self.chan.get_bsp_list(dataframe)
|
|
#dataframe_5['state'] = self.chan.cal_klu_state(dataframe_5)
|
|
#dataframe_15['state'] = self.chan.cal_klu_state(dataframe_15)
|
|
#dataframe_30['state'] = self.chan.cal_klu_state(dataframe_30)
|
|
#dataframe_60['state'] = self.chan.cal_klu_state(dataframe_60)
|
|
#dataframe_4h['state'] = self.chan.resample_klc_list(dataframe_4h)
|
|
|
|
self.chan.plot_dual(dataframe_30, dataframe_60)
|
|
dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)
|
|
#self.print_macd_div_list(dataframe)
|
|
#self.print_resample_df(dataframe, 1, 50)
|
|
self.chan.get_bi_list(dataframe_30)
|
|
if self.last_time + timedelta(minutes=1) < datetime.now():
|
|
#print(informative.iloc[-1])
|
|
#self.log_macd_div_list(dataframe)
|
|
#self.print_xgb(dataframe, "1m_model")
|
|
#self.print_xgb(dataframe_5, "5m_model")
|
|
#self.print_xgb(dataframe_30, "30m_model")
|
|
#self.print_xgb(dataframe_60, "1h_model")
|
|
#self.print_xgb(dataframe_4h, "4h_model")
|
|
print("-------------------------------------------------------------------------------")
|
|
self.last_time = datetime.now()
|
|
#dataframe = resampled_merge(dataframe, dataframe_5)
|
|
#dataframe = resampled_merge(dataframe, dataframe_15)
|
|
#dataframe = resampled_merge(dataframe, dataframe_30)
|
|
#dataframe = resampled_merge(dataframe, dataframe_60)
|
|
#dataframe = resampled_merge(dataframe, dataframe_4h)
|
|
return dataframe
|
|
# This is called when placing the initial order (opening trade)
|
|
def print_xgb(self, dataframe, model_name):
|
|
self.classifier.load_model(model_name)
|
|
klc_list = self.chan.get_klc_list(dataframe)
|
|
klc1 = klc_list[-1]
|
|
klc2 = klc_list[-2]
|
|
klc3 = klc_list[-3]
|
|
if klc1.end_time == klc2.start_time:
|
|
print(model_name, klc1.end_time, klc1.fx, self.classifier.predict(klc1))
|
|
else:
|
|
print(model_name, klc1.start_time, klc1.fx, self.classifier.predict(klc1))
|
|
print(model_name, klc2.end_time, klc2.fx, self.classifier.predict(klc2))
|
|
print(model_name, klc3.end_time, klc3.fx, self.classifier.predict(klc3))
|
|
def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float,
|
|
proposed_stake: float, min_stake: float | None, max_stake: float,
|
|
leverage: float, entry_tag: str | None, side: str,
|
|
**kwargs) -> float:
|
|
|
|
# We need to leave most of the funds for possible further DCA orders
|
|
# This also applies to fixed stakes
|
|
return proposed_stake / self.max_dca_multiplier
|
|
def adjust_trade_position1(self, trade: Trade, current_time: datetime,
|
|
current_rate: float, current_profit: float,
|
|
min_stake: float | None, max_stake: float,
|
|
current_entry_rate: float, current_exit_rate: float,
|
|
current_entry_profit: float, current_exit_profit: float,
|
|
**kwargs
|
|
) -> float | None | tuple[float | None, str | None]:
|
|
"""
|
|
Custom trade adjustment logic, returning the stake amount that a trade should be
|
|
increased or decreased.
|
|
This means extra entry or exit orders with additional fees.
|
|
Only called when `position_adjustment_enable` is set to True.
|
|
|
|
For full documentation please go to https://www.freqtrade.io/en/latest/strategy-advanced/
|
|
|
|
When not implemented by a strategy, returns None
|
|
|
|
:param trade: trade object.
|
|
:param current_time: datetime object, containing the current datetime
|
|
:param current_rate: Current entry rate (same as current_entry_profit)
|
|
:param current_profit: Current profit (as ratio), calculated based on current_rate
|
|
(same as current_entry_profit).
|
|
:param min_stake: Minimal stake size allowed by exchange (for both entries and exits)
|
|
:param max_stake: Maximum stake allowed (either through balance, or by exchange limits).
|
|
:param current_entry_rate: Current rate using entry pricing.
|
|
:param current_exit_rate: Current rate using exit pricing.
|
|
:param current_entry_profit: Current profit using entry pricing.
|
|
:param current_exit_profit: Current profit using exit pricing.
|
|
:param **kwargs: Ensure to keep this here so updates to this won't break your strategy.
|
|
:return float: Stake amount to adjust your trade,
|
|
Positive values to increase position, Negative values to decrease position.
|
|
Return None for no action.
|
|
Optionally, return a tuple with a 2nd element with an order reason
|
|
"""
|
|
#if trade.has_open_orders:
|
|
# Only act if no orders are open
|
|
#return
|
|
|
|
#if current_profit > 0.05 and trade.nr_of_successful_exits == 0:
|
|
# Take half of the profit at +5%
|
|
#return -(trade.stake_amount / 2), "half_profit_5%"
|
|
|
|
#if current_profit > -0.05:
|
|
#return None
|
|
|
|
# Obtain pair dataframe (just to show how to access it)
|
|
dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe)
|
|
# Only buy when not actively falling price.
|
|
#last_candle = dataframe.iloc[-1].squeeze()
|
|
#previous_candle = dataframe.iloc[-2].squeeze()
|
|
#if last_candle["close"] < previous_candle["close"]:
|
|
#return None
|
|
filled_entries = trade.select_filled_orders(trade.entry_side)
|
|
last_entry = filled_entries[-1]
|
|
count_of_entries = trade.nr_of_successful_entries
|
|
# Allow up to 3 additional increasingly larger buys (4 in total)
|
|
# Initial buy is 1x
|
|
# If that falls to -5% profit, we buy 1.25x more, average profit should increase to roughly -2.2%
|
|
# If that falls down to -5% again, we buy 1.5x more
|
|
# If that falls once again down to -5%, we buy 1.75x more
|
|
# Total stake for this trade would be 1 + 1.25 + 1.5 + 1.75 = 5.5x of the initial allowed stake.
|
|
# That is why max_dca_multiplier is 5.5
|
|
# Hope you have a deep wallet!
|
|
# This returns first order stake size
|
|
#print(dataframe.iloc[-1]['resample_{}_state'.format(self.get_ticker_indicator()*self.time60)])
|
|
|
|
# This returns first order stake size
|
|
stake_amount = filled_entries[0].stake_amount
|
|
# This then calculates current safety order size
|
|
stake_amount = stake_amount * (1 + (count_of_entries * 0.5))
|
|
dataframe_date = dataframe.iloc[-1]['date']
|
|
#print(stake_amount, "---------------------------------------------------")
|
|
if last_entry.order_filled_utc + timedelta(minutes=self.time5) < dataframe_date:
|
|
if dataframe.iloc[-self.time5]['resample_{}_state'.format(self.get_ticker_indicator()*self.time5)] == "-10" and last_entry.side == "buy":
|
|
#print(dataframe.iloc[-self.time5])
|
|
#print(stake_amount)
|
|
return stake_amount, "1/3rd_increase"
|
|
if last_entry.order_filled_utc + timedelta(minutes=self.time5) < dataframe_date:
|
|
if dataframe.iloc[-self.time5]['resample_{}_state'.format(self.get_ticker_indicator()*self.time5)] == "10" and last_entry.side == "sell":
|
|
#print(dataframe.iloc[-self.time5])
|
|
#print(stake_amount)
|
|
return stake_amount, "1/3rd_increase"
|
|
return None
|
|
def log_macd_div_list(self, dataframe):
|
|
bi_macd_div_list, bi_list, seg_macd_div_list, seg_list = self.chan.get_macd_div_list(dataframe)
|
|
logger.info(f"BI MACD DIV LIST")
|
|
for index in range(len(bi_list)-5, len(bi_list)):
|
|
bi = bi_list[index]
|
|
logger.info(f'{bi.start_time}, {bi.high}, {bi.low}, {bi.dir}, {bi.macd_div}')
|
|
logger.info(f"SEG MACD DIV LIST")
|
|
for index in range(len(seg_list)-5, len(seg_list)):
|
|
seg = seg_list[index]
|
|
logger.info(f'{seg.start_bi.start_time}, {seg.high}, {seg.low}, {seg.dir}, {seg.macd_div}')
|
|
def print_fx(self, df, label=5):
|
|
fx_list = self.chan.get_bsp_list(self.chan.get_klc_list(df))
|
|
fx1 = fx_list[-1]
|
|
fx2 = fx_list[-2]
|
|
fx3 = fx_list[-3]
|
|
fx4 = fx_list[-4]
|
|
fx5 = fx_list[-5]
|
|
#print(fx1.end_time, fx1.fx, fx2.end_time, fx2.fx, fx3.end_time, fx3.fx)
|
|
logger.info(f'\n{fx5.end_time} {fx5.state} {fx4.end_time} {fx4.state} {fx3.end_time} {fx3.state} {fx2.end_time} {fx2.state} {fx1.end_time} {fx1.state} TF: {label}')
|
|
def print_fx_list(self, df):
|
|
bsp_list = self.chan.get_bsp_list(self.chan.get_klc_list(df))
|
|
for bsp in bsp_list:
|
|
logger.info(f'{bsp.start_time}, {bsp.end_time}, {bsp.fx}')
|
|
def print_df(self, df):
|
|
for index in range(0, len(df)):
|
|
state = 'state'
|
|
rsi = 'rsi'
|
|
logger.info(f'{df[state][index]}, {df[rsi][index]}, {df[state][index]}')
|
|
def print_resample_df(self, dataframe, time, limit=10):
|
|
df = dataframe.tail(limit)
|
|
if limit > 0:
|
|
if time == 1:
|
|
for index in range(len(dataframe) - limit, len(dataframe)):
|
|
cn1 = 'date'
|
|
cn2 = 'rsi'
|
|
cn3 = 'state'
|
|
logger.info(f'{df[cn1][index]}, {df[cn2][index]}, {df[cn3][index]}')
|
|
else:
|
|
for index in range(0, limit):
|
|
cn1 = 'resample_{}_date'.format(self.get_ticker_indicator()*time)
|
|
cn2 = 'resample_{}_rsi'.format(self.get_ticker_indicator()*time)
|
|
cn3 = 'resample_{}_state'.format(self.get_ticker_indicator()*time)
|
|
logger.info(f'{df[cn1][index]}, {df[cn2][index]}, {df[cn3][index]}')
|
|
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)
|
|
return df
|
|
def local_print(self, df):
|
|
fast = 7
|
|
slow = 14
|
|
macd = ta.MACD(df, fast=fast, slow=slow)
|
|
df['macd'] = macd['macd']
|
|
df['macdsignal'] = macd['macdsignal']
|
|
df['macdhist'] = macd['macdhist']
|
|
df['ema26'] = ta.EMA(df, timeperiod=26)
|
|
df['ema52'] = ta.EMA(df, timeperiod=52)
|
|
df['ma5'] = ta.MA(df, timeperiod=5)
|
|
df['ma10'] = ta.MA(df, timeperiod=10)
|
|
df['masub'] = df['ma5'].subtract(df['ma10'])
|
|
#for index in range(0, len(df)):
|
|
#print(df['date'][index], df['macdhist'][index], df['macd'][index], df['macdsignal'][index], df['masub'][index])
|
|
# (1,1) = 1, (1,0) = 2, (-1,1) = 3, (-1, 0) = 4, (0,0) = 0
|
|
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
|
|
dataframe.loc[
|
|
(
|
|
(dataframe['state'] == "-30")
|
|
#((dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time5)] == "-11") |
|
|
#(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['resample_{}_state'.format(self.get_ticker_indicator()*self.time5)] == "11") |
|
|
#(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:
|
|
dataframe.loc[
|
|
(
|
|
(dataframe['state']== "30")
|
|
#(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.time60)] == "10")
|
|
),
|
|
['exit_long', 'exit_tag']] = (1, 'long_close_signal_chan')
|
|
dataframe.loc[
|
|
(
|
|
(dataframe['state'] == "-30")
|
|
#(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.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 1.0
|
|
|
|
def get_ticker_indicator(self):
|
|
return int(self.timeframe[:-1]) |