jack/Chan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
f5e76663e62269d350341d85c5ebe3c52c4f4215
Chan/strategies/Chan_SOL_2.py
T

196 lines
10 KiB
Python
Raw Blame History

# --- Do not remove these libs ---
from freqtrade.strategy import IStrategy
from typing import Dict, List
from functools import reduce
from pandas import DataFrame, pandas
import freqtrade.vendor.qtpylib.indicators as qtpylib
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 technical.util import resample_to_interval, resampled_merge
import talib.abstract as ta
import freqtrade.vendor.qtpylib.indicators as qtpylib
from datetime import datetime, timedelta, timezone
from freqtrade.persistence import Trade, Order
from typing import Optional
from ChanPY import ChanPY
import logging
logger = logging.getLogger(__name__)
### Now you can use logger.info('asfd') to log
# freqtrade trade -c ./user_data/Chan.json --strategy Chan_SOL_2 --strategy-path ./user_data/strategies
# freqtrade backtesting -c ./user_data/Chan.json --strategy Chan_SOL_2 --strategy-path ./user_data/strategies --timerange=20250309-
# freqtrade download-data -c ./user_data/Chan.json -t 1m --pairs SOL/USDT:USDT --timerange=20240101-
# freqtrade hyperopt --hyperopt-loss SharpeHyperOptLossDaily --spaces roi stoploss --strategy Chan_SOL_2 --strategy-path ./user_data/strategies -c ./user_data/Chan.json -e 200 --timerange=20250101-20250215
# sudo docker compose run --rm chan_btc backtesting -c ./user_data/Chan.json --strategy Chan_SOL_2 --strategy-path ./user_data/strategies --timerange=20250101-
# sudo docker compose run --rm chan_btc download-data -c ./user_data/Chan.json --pairs SOL/USDT:USDT -t 1m --timerange 20240101-
# sudo docker compose run --rm chan_btc trade -c ./user_data/Chan.json --strategy Chan_SOL_2 --strategy-path ./user_data/strategies
class Chan_SOL_2(IStrategy):
INTERFACE_VERSION: int = 3
# Minimal ROI designed for the strategy.
# This attribute will be overridden if the config file contains "minimal_roi"
minimal_roi = {
"0": 0.253,
"120": 0.159,
"240": 0.052,
"360": 0
}
can_short = True
# Optimal stoploss designed for the strategy
# This attribute will be overridden if the config file contains "stoploss"
stoploss = -0.21
trailing_stop = False
trailing_stop_positive = 0.015
trailing_stop_positive_offset = 0.043
trailing_only_offset_is_reached = False
# Optimal timeframe for the strategy
# timeframe = '15m'
startup_candle_count = 600
time5 = 5
time15 = 15
time30 = 30
time60 = 60
time240 = 240
last_time = datetime.now()
big_size = 0
big_state = "00"
big_state_list = []
chanpy = ChanPY()
chan = ChanLun()
small_size = 0
small_state = "00"
small_state_list = []
def informative_pairs(self):
# get access to all pairs available in whitelist.
pairs = self.dp.current_whitelist()
# Assign tf to each pair so they can be downloaded and cached for strategy.
informative_pairs = [(pair, '1h') for pair in pairs]
# Optionally Add additional "static" pairs
#informative_pairs += [("ETH/USDT:USDT", "5m"),("ETH/USDT:USDT", "15m"),]
return informative_pairs
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)
self.local_print(dataframe_5)
dataframe_5['state'] = self.chan.resample_klc_list(dataframe_5)
#dataframe_15['state'] = self.chan.resample_klc_list(dataframe_15)
#dataframe_30['state'] = self.chan.resample_klc_list(dataframe_30)
dataframe_60['state'] = self.chan.resample_klc_list(dataframe_60)
#dataframe_4h['state'] = self.chan.resample_klc_list(dataframe_4h)
#dataframe_5['bsps'], dataframe_5['updown'], dataframe_5['bi_sure'] = self.chanpy.get_bsps(dataframe_5)
print("===================================================")
#print(dataframe_60['high'].rolling(window).max())
#print(dataframe_60['low'].rolling(window).min())
#for index in range(0, len(dataframe_5)):
#print(dataframe_5[dataframe_5['bi_sure'] == 1][dataframe_5['state'] == "-10"])
#print(dataframe_5[dataframe_5['bi_sure'] == 1][dataframe_5['state'] == "10"])
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
def print_df(self, df):
for index in range(0, len(df)):
print(df['date'][index], df['rsi'][index], df['state'][index])
def print_resample_df(self, df, time):
for index in range(0, len(df)):
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)
print(df[cn1][index], df[cn2][index], df[cn3][index])
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'].shift(1) == "-10") &
#((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.time60)].shift(self.time60) == "11") &
(dataframe['resample_{}_bsps'.format(self.get_ticker_indicator()*self.time5)].shift(self.time5) > 0) &
(dataframe['resample_{}_bsps'.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'].shift(1) == "10") &
#((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.time60)].shift(self.time60) == "-11") &
(dataframe['resample_{}_bsps'.format(self.get_ticker_indicator()*self.time5)].shift(self.time5) < 0)
#(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'].shift(1) == "10") &
(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time60)].shift(self.time60) == "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'].shift(1) == "-10") &
(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time60)].shift(self.time60) == "-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])
Reference in New Issue View Git Blame Copy Permalink