jack/Chan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add more strategies

Browse Source
This commit is contained in:
jackyu66git
2025-05-09 21:10:45 +08:00
parent 65a85823b9
commit 2440e896ea
17 changed files with 887 additions and 271 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ChanKLU.py
+18 -9
View File
@@ -16,6 +16,8 @@ class ChanKLU:
self.ma5 = 0
self.ma10 = 0
self.ma30 = 0
self.ma50 = 0
self.ma200 = 0
self.ma250 = 0
self.rsi = 0
self.volume_ratio = 0
@@ -23,15 +25,20 @@ class ChanKLU:
self.idx = idx
self.index = idx
def set_indicators(self, item):
self.macd = float(item['macd']) if item['macd'] else 0
self.signal = float(item['macdsignal']) if item['macdsignal'] else 0
self.macdhist = float(item['macdhist']) if item['macdhist'] else 0
self.ma5 = float(item['ma5']) if item['ma5'] else 0
self.ma10 = float(item['ma10']) if item['ma10'] else 0
self.ma30 = float(item['ma30']) if item['ma30'] else 0
self.ma250 = float(item['ma250']) if item['ma250'] else 0
self.rsi = float(item['rsi']) if item['rsi'] else 0
self.volume_ratio = float(item['volume_ratio']) if item['volume_ratio'] else 0
self.macd = float(item['macd']) if 'macd' in item and item['macd'] else 0
self.signal = float(item['macdsignal']) if 'macdsignal' in item and item['macdsignal'] else 0
self.macdhist = float(item['macdhist']) if 'macdhist' in item and item['macdhist'] else 0
self.ma5 = float(item['ma5']) if 'ma5' in item and item['ma5'] else 0
self.ma10 = float(item['ma10']) if 'ma10' in item and item['ma10'] else 0
self.ma30 = float(item['ma30']) if 'ma30' in item and item['ma30'] else 0
# 安全检查 ma250、ma50 和 ma200
self.ma250 = float(item['ma250']) if 'ma250' in item and item['ma250'] else 0
self.ma50 = float(item['ma50']) if 'ma50' in item and item['ma50'] else 0
self.ma200 = float(item['ma200']) if 'ma200' in item and item['ma200'] else 0
self.rsi = float(item['rsi']) if 'rsi' in item and item['rsi'] else 0
self.volume_ratio = float(item['volume_ratio']) if 'volume_ratio' in item and item['volume_ratio'] else 0
def get_feature_data(self):
features = dict()
features['klu_close'] = self.close
@@ -46,6 +53,8 @@ class ChanKLU:
features['klu_ma5'] = self.ma5
features['klu_ma10'] = self.ma10
features['klu_ma30'] = self.ma30
features['klu_ma50'] = self.ma50
features['klu_ma200'] = self.ma200
features['klu_ma250'] = self.ma250
features['klu_rsi'] = self.rsi
features['klu_volume_ratio'] = self.volume_ratio
ChanLun.py
+13 -99
View File
@@ -97,6 +97,14 @@ class ChanLun():
state_list = []
if len(klc_list) > 0:
klc_index = 0
# 添加趋势强度判断
dataframe['trend_strength'] = abs(dataframe['close'].pct_change(20))
# 添加波动率判断
dataframe['volatility'] = dataframe['close'].pct_change().rolling(window=20).std()
# 添加成交量趋势
dataframe['volume_trend'] = dataframe['volume'].rolling(window=20).mean()
for index in range(0, len(dataframe)):
if klc_index == len(klc_list):
klc_index = len(klc_list) - 1
@@ -1214,105 +1222,9 @@ class ChanLun():
klc.set_end_klu(klu)
return klc_list
def get_bsp_list1(self, big_df):
big_bi_list = self.get_bi_list(big_df)
big_seg_list = self.get_seg_list(big_bi_list)
big_zs_list = self.get_zs_list(big_bi_list, big_seg_list)
big_bi_macd_div_list = self.get_bi_macd_div_list(big_bi_list, big_df)
big_seg_macd_div_list = self.get_seg_macd_div_list(big_seg_list, big_df)
big_bi_macd_hist_list = self.get_bi_macd_hist_list(big_bi_list, big_df)
big_seg_macd_hist_list = self.get_seg_macd_hist_list(big_seg_list, big_df)
for index in range(0, len(big_seg_list)):
big_seg = big_seg_list[index]
if big_seg.end_bi:
if big_seg.dir == Chan_SEG_DIR.UP:
if big_seg.end_bi.index - big_seg.start_bi.index > 1:
max_high = big_seg.start_bi.high
for bi_index in range(big_seg.start_bi.index + 2, big_seg.end_bi.index + 1):
bi = big_bi_list[bi_index]
#print("MACD DIV: ", big_bi_macd_hist_list[index]/big_bi_macd_hist_list[index - 2])
if bi.is_sure and bi.dir == Chan_BI_DIR.UP:
if bi.high > max_high:
max_high = bi.high
if big_bi_macd_hist_list[bi_index - 2] > 0.0:
bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 2]
if bi_macd_div < 0.01 and len(big_bi_list) - big_seg.start_bi.index > 4 and big_bi_macd_hist_list[bi_index - 4] > 0.0:
bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 4]
else:
if len(big_bi_list) - big_seg.start_bi.index > 4 and big_bi_macd_hist_list[bi_index - 4] > 0.0:
bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 4]
else:
bi_macd_div = 0.0
macd_index = bi.end_klc.end_klu.index
if bi_macd_div < 0.8 and bi_macd_div > 0.01 and big_df['macd'][macd_index] > 0 and big_df['macdsignal'][macd_index] > 0:
print("UP SEG Possible BSP:", bi.start_klc.end_time, bi_macd_div)
else:
if big_seg.end_bi.index - big_seg.start_bi.index > 1:
max_low = big_seg.start_bi.low
for bi_index in range(big_seg.start_bi.index + 2, big_seg.end_bi.index + 1):
bi = big_bi_list[bi_index]
if bi.is_sure and bi.dir == Chan_BI_DIR.DOWN:
#print("DOWN: ", max_low, bi.low)
if bi.low < max_low:
max_low = bi.low
if big_bi_macd_hist_list[bi_index - 2] > 0.0:
bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 2]
if bi_macd_div < 0.01 and len(big_bi_list) - big_seg.start_bi.index > 4 and big_bi_macd_hist_list[bi_index - 4] > 0.0:
bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 4]
else:
if len(big_bi_list) - big_seg.start_bi.index > 4 and big_bi_macd_hist_list[bi_index - 4] > 0.0:
bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 4]
else:
bi_macd_div = 0.0
macd_index = bi.end_klc.end_klu.index
if bi_macd_div < 0.8 and bi_macd_div > 0.01 and big_df['macd'][macd_index] < 0 and big_df['macdsignal'][macd_index] < 0:
print("DOWN SEG Possible BSP:", bi.start_klc.end_time, bi_macd_div)
else:
print("Not completed segment.", len(big_bi_list) - big_seg.start_bi.index, big_seg.dir)
if big_seg.dir == Chan_SEG_DIR.UP:
if len(big_bi_list) - big_seg.start_bi.index > 1:
max_high = big_seg.start_bi.high
for bi_index in range(big_seg.start_bi.index + 2, len(big_bi_list)):
bi = big_bi_list[bi_index]
#print("MACD DIV: ", big_bi_macd_hist_list[index]/big_bi_macd_hist_list[index - 2])
if bi.is_sure and bi.dir == Chan_BI_DIR.UP:
if bi.high > max_high:
max_high = bi.high
if big_bi_macd_hist_list[bi_index - 2] > 0.0:
bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 2]
if bi_macd_div < 0.01 and len(big_bi_list) - big_seg.start_bi.index > 4 and big_bi_macd_hist_list[bi_index - 4] > 0.0:
bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 4]
else:
if len(big_bi_list) - big_seg.start_bi.index > 4 and big_bi_macd_hist_list[bi_index - 4] > 0.0:
bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 4]
else:
bi_macd_div = 0.0
macd_index = len(big_df) - 1
if bi_macd_div < 0.8 and bi_macd_div > 0.01 and big_df['macd'][macd_index] > 0 and big_df['macdsignal'][macd_index] > 0:
print("UP SEG Possible BSP:", bi.start_klc.end_time, bi_macd_div)
else:
if len(big_bi_list) - big_seg.start_bi.index > 1:
max_low = big_seg.start_bi.low
for bi_index in range(big_seg.start_bi.index + 2, len(big_bi_list)):
bi = big_bi_list[bi_index]
#print("MACD DIV: ", big_bi_macd_hist_list[index]/big_bi_macd_hist_list[index - 2])
if bi.is_sure and bi.dir == Chan_BI_DIR.DOWN:
if bi.low < max_low:
max_low = bi.low
if big_bi_macd_hist_list[bi_index - 2] > 0.0:
bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 2]
if bi_macd_div < 0.01 and len(big_bi_list) - big_seg.start_bi.index > 4 and big_bi_macd_hist_list[bi_index - 4] > 0.0:
bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 4]
else:
if len(big_bi_list) - big_seg.start_bi.index > 4 and big_bi_macd_hist_list[bi_index - 4] > 0.0:
bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 4]
else:
bi_macd_div = 0.0
macd_index = len(big_df) - 1
if bi_macd_div < 0.8 and bi_macd_div > 0.01 and big_df['macd'][macd_index] < 0 and big_df['macdsignal'][macd_index] < 0:
print("DOWN SEG Possible BSP:", bi.start_klc.end_time, bi_macd_div)
# ================================================
# 计算BSP列表
# ================================================
def get_bsp_list(self, big_df):
big_bi_list = self.get_bi_list(big_df)
big_seg_list = self.get_seg_list(big_bi_list)
@@ -1501,6 +1413,8 @@ class ChanLun():
print(bsp.bi.end_klc.end_time, bsp.sure_time, bsp.dir, bsp.seg.dir, bsp.bi.macd_div)
return bi_bsp_list
# ================================================
# 第三类买卖点
def find_third_bsp(self, zs_list):
bsp_list = []
zs_count = 0
__pycache__/ChanBI.cpython-312.pyc
BIN
View File
Binary file not shown.
__pycache__/ChanBSP.cpython-312.pyc
BIN
View File
Binary file not shown.
__pycache__/ChanCTime.cpython-312.pyc
BIN
View File
Binary file not shown.
__pycache__/ChanEnum.cpython-312.pyc
BIN
View File
Binary file not shown.
__pycache__/ChanKLC.cpython-312.pyc
BIN
View File
Binary file not shown.
__pycache__/ChanKLU.cpython-312.pyc
BIN
View File
Binary file not shown.
__pycache__/ChanLun.cpython-312.pyc
BIN
View File
Binary file not shown.
__pycache__/ChanSBI.cpython-312.pyc
BIN
View File
Binary file not shown.
__pycache__/ChanSEG.cpython-312.pyc
BIN
View File
Binary file not shown.
__pycache__/ChanZS.cpython-312.pyc
BIN
View File
Binary file not shown.
config/ChanLun_SOL.json
+2 -2
View File
@@ -43,7 +43,7 @@
"ccxt_config": {},
"ccxt_async_config": {},
"pair_whitelist": [
"SOL/USDT:USDT",
"SOL/USDT:USDT"
],
"pair_blacklist": [
"BNB/.*"
@@ -55,7 +55,7 @@
"number_assets": 1,
"sort_key": "quoteVolume",
"min_value": 0,
"refresh_period": 1800,
"refresh_period": 1800
}
],
"telegram": {
config/ChanLun_SOL_Optimized.json
+151
View File
@@ -0,0 +1,151 @@
{
"$schema": "https://schema.freqtrade.io/schema.json",
"max_open_trades": 2,
"stake_currency": "USDT",
"stake_amount": "unlimited",
"tradable_balance_ratio": 0.95,
"fiat_display_currency": "USD",
"dry_run": true,
"db_url": "sqlite:///tradesv3.chanlun_sol_optimized.sqlite",
"dry_run_wallet": 1000,
"cancel_open_orders_on_exit": true,
"trading_mode": "futures",
"margin_mode": "isolated",
"can_short": true,
"timeframe": "5m",
"process_only_new_candles": false,
"unfilledtimeout": {
"entry": 2,
"exit": 2,
"exit_timeout_count": 0,
"unit": "minutes"
},
"entry_pricing": {
"price_side": "same",
"use_order_book": true,
"order_book_top": 1,
"price_last_balance": 0.0,
"check_depth_of_market": {
"enabled": false,
"bids_to_ask_delta": 1
}
},
"exit_pricing":{
"price_side": "other",
"use_order_book": true,
"order_book_top": 1
},
"exchange": {
"name": "binance",
"key": "",
"secret": "",
"ccxt_config": {
"options": {"defaultType": "swap"}
},
"ccxt_async_config": {
"enableRateLimit": true,
"rateLimit": 1000,
"timeout": 30000
},
"pair_whitelist": [
"SOL/USDT:USDT"
],
"pair_blacklist": []
},
"telegram": {
"enabled": false,
"token": "",
"chat_id": ""
},
"api_server": {
"enabled": false,
"listen_ip_address": "0.0.0.0",
"listen_port": 8080,
"verbosity": "error",
"jwt_secret_key": "",
"username": "",
"password": ""
},
"discord": {
"enabled": false,
"webhook": "",
"webhook_avatar": "",
"poll_delay_seconds": 10
},
"notification_settings": {
"status": "on",
"status_inactive_after": 7,
"timeframe_condition_change": "on",
"telegram": { },
"discord": { },
"notify_all": true
},
"bot_name": "SOL_Chan_Optimized",
"initial_state": "running",
"force_entry_enable": false,
"internals": {
"process_throttle_secs": 5
},
"edge": {
"enabled": false,
"process_throttle_secs": 3600,
"calculate_since_number_of_days": 7,
"allowed_risk": 0.01,
"stoploss_range_min": -0.01,
"stoploss_range_max": -0.007,
"stoploss_range_step": 0.001,
"minimum_winrate": 0.60,
"minimum_expectancy": 0.20,
"min_trade_number": 10,
"max_trade_duration_minute": 1440,
"remove_pumps": false
},
"order_types": {
"entry": "limit",
"exit": "market",
"emergency_exit": "market",
"force_exit": "market",
"force_entry": "market",
"stoploss": "market",
"stoploss_on_exchange": false,
"stoploss_on_exchange_interval": 60
},
"order_time_in_force": {
"entry": "GTC",
"exit": "GTC"
},
"strategy_path": "./user_data/Chan/strategies/",
"strategy": "ChanLun_SOL_Optimized",
"minimal_roi": {
"0": 0.012,
"120": 0.010,
"240": 0.007,
"360": 0.005
},
"stoploss": -0.007,
"trailing_stop": true,
"trailing_stop_positive": 0.003,
"trailing_stop_positive_offset": 0.005,
"trailing_only_offset_is_reached": true,
"use_custom_stoploss": true,
"max_open_trades_per_pair": 1,
"dry_run_wallet_refresh_time": 5,
"caches": {
"dataframe": {
"enabled": true,
"refresh_period": 60
},
"strategy": {
"enabled": true,
"refresh_period": 300
}
},
"pairlists": [
{
"method": "StaticPairList",
"config": {
"pairs": ["SOL/USDT:USDT"]
}
}
]
}
strategies/ChanLun_SOL_5.py
+28 -16
View File
@@ -42,17 +42,17 @@ class ChanLun_SOL_5(IStrategy):
}
# 5m and 15m
minimal_roi_1 = {
"0": 0.253,
"60": 0.159,
"120": 0.052,
"0": 0.1,
"60": 0.05,
"120": 0.02,
"240": 0
}
# 15m and 30m
minimal_roi_1 = {
"0": 0.253,
"120": 0.159,
"240": 0.052,
"360": 0
"0": 0.1,
"240": 0.05,
"480": 0.03,
"600": 0
}
minimal_roi_2 = {
"0": 0.10,
@@ -60,8 +60,8 @@ class ChanLun_SOL_5(IStrategy):
"2400": 0.025,
"3600": 0
}
can_short = False
lev = 5.0
can_short = True
lev = 1.0
stoploss = -0.3 * lev
trailing_stop = False
trailing_stop_positive = 0.025
@@ -471,12 +471,16 @@ class ChanLun_SOL_5(IStrategy):
#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:
state_str = 'resample_{}_state'.format(self.get_ticker_indicator()*self.time5)
close_str = 'resample_{}_close'.format(self.get_ticker_indicator()*self.time5)
volume_str = 'resample_{}_volume'.format(self.get_ticker_indicator()*self.time5)
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.time5)] == "99")
(dataframe[state_str].shift(self.time5) == "-10") &
(dataframe[close_str].pct_change().abs() < 0.05) &
(dataframe[close_str] > dataframe[close_str].shift(self.time5)) &
(dataframe[volume_str] > dataframe[volume_str].rolling(window=20).mean())
#(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")
@@ -486,7 +490,10 @@ class ChanLun_SOL_5(IStrategy):
dataframe.loc[
(
#(dataframe['state'] == "30")
(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time5)].shift(self.time5) == "10")
(dataframe[state_str].shift(self.time5) == "10") &
(dataframe[close_str].pct_change().abs() < 0.05) &
(dataframe[close_str] < dataframe[close_str].shift(self.time5)) &
(dataframe[volume_str] > dataframe[volume_str].rolling(window=20).mean())
#(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")
@@ -495,11 +502,14 @@ class ChanLun_SOL_5(IStrategy):
['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.time5)
close_str = 'resample_{}_close'.format(self.get_ticker_indicator()*self.time5)
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.time5)] == "-99")
(dataframe[state_str].shift(self.time5) == "10") |
(dataframe[close_str] < dataframe[close_str].rolling(window=20).min()) |
(dataframe[close_str] > dataframe[close_str].rolling(window=20).max()*1.05)
#(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time30)] == "10") &
#(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time60)] == "10")
),
@@ -507,7 +517,9 @@ class ChanLun_SOL_5(IStrategy):
dataframe.loc[
(
#(dataframe['state'] == "-30")
(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time5)].shift(self.time5) == "-10")
(dataframe[state_str].shift(self.time5) == "-10") |
(dataframe[close_str] > dataframe[close_str].rolling(window=20).max()) |
(dataframe[close_str] < dataframe[close_str].rolling(window=20).min()*0.95)
#(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time30)] == "-10") &
#(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time60)] == "-10")
),
strategies/ChanLun_SOL_Optimized.py
+386
View File
@@ -0,0 +1,386 @@
"""
SOL/USDT 优化交易策略
采用多时间周期分析和缠论技术分析，专注于空头交易
集成了技术指标确认和风险管理功能
"""
# --- 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 technical.util import resample_to_interval, resampled_merge
import talib.abstract as ta
import freqtrade.vendor.qtpylib.indicators as qtpylib
from pandas import DataFrame
from datetime import datetime, timedelta
from freqtrade.persistence import Trade
from typing import Optional
import logging
import numpy as np
logger = logging.getLogger(__name__)
# freqtrade trade -c ./user_data/Chan/config/ChanLun_SOL_Optimized.json --strategy ChanLun_SOL_Optimized --strategy-path ./user_data/Chan/strategies
# freqtrade backtesting -c ./user_data/Chan/config/ChanLun_SOL_Optimized.json --strategy ChanLun_SOL_Optimized --strategy-path ./user_data/Chan/strategies --timerange=20250201-
class ChanLun_SOL_Optimized(IStrategy):
"""
SOL/USDT 优化交易策略 - 专注于空头交易
"""
INTERFACE_VERSION: int = 3
# 优化后的ROI设置，主要针对短期交易
minimal_roi = {
"0": 0.012,
"120": 0.010,
"240": 0.007,
"360": 0.005
}
# 支持做空
can_short = True
only_short = True
# 杠杆设置（谨慎使用）
lev = 1.0
# 止损设置
stoploss = -0.007 * lev
# 追踪止损设置
trailing_stop = True
trailing_stop_positive = 0.003
trailing_stop_positive_offset = 0.005
trailing_only_offset_is_reached = True
# 仓位管理设置
position_adjustment_enable = True
max_entry_position_adjustment = 3
max_dca_multiplier = 4.0
# 策略初始化需要的K线数量
startup_candle_count = 200
# 时间周期定义
timeframe = '5m'
# 时间周期乘数
time5 = 5
time15 = 15
time30 = 30
time60 = 60
time4h = 240
time1d = 1440
# 缠论模块初始化
chan = ChanLun()
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
添加技术指标
"""
# 基础技术指标
for df in [dataframe]:
# 添加MACD指标
macd = ta.MACD(df)
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['ma20'] = ta.MA(df, timeperiod=20)
df['ma30'] = ta.EMA(df, timeperiod=30)
df['ma50'] = ta.MA(df, timeperiod=50)
df['ma200'] = ta.MA(df, timeperiod=200)
# 添加RSI指标
df['rsi'] = ta.RSI(df, timeperiod=14)
df['rsi_slow'] = ta.RSI(df, timeperiod=21)
# 添加ATR（波动率）
df['atr'] = ta.ATR(df, timeperiod=14)
# 计算布林带
bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(df), window=20, stds=2)
df['bb_lowerband'] = bollinger['lower']
df['bb_middleband'] = bollinger['mid']
df['bb_upperband'] = bollinger['upper']
df['bb_width'] = (df['bb_upperband'] - df['bb_lowerband']) / df['bb_middleband']
# 添加ADX指标（趋势强度）
df['adx'] = ta.ADX(df, timeperiod=14)
df['plus_di'] = ta.PLUS_DI(df, timeperiod=14)
df['minus_di'] = ta.MINUS_DI(df, timeperiod=14)
# 添加量比指标
df['volume_ma20'] = df['volume'].rolling(window=20).mean()
df['volume_ratio'] = df['volume'] / df['volume_ma20']
# 计算下降趋势确认指标
dataframe['downtrend'] = (
(dataframe['ma5'] < dataframe['ma10']) &
(dataframe['ma10'] < dataframe['ma30']) &
(dataframe['close'] < dataframe['ma10']) &
(dataframe['close'].shift(1) > dataframe['close'])
)
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
入场信号逻辑 - 放宽条件以产生更多交易信号
"""
# 关闭多头交易
dataframe['enter_long'] = 0
# 空头入场条件 - 条件1：价格下跌趋势
dataframe.loc[
(
# 价格下跌趋势 - 放宽为仅需一根K线下跌
(dataframe['close'] < dataframe['close'].shift(1)) &
# 价格在均线下方 - 使用更短期均线
(dataframe['close'] < dataframe['ma20']) &
# RSI条件放宽 - 只要不是极度超卖
(dataframe['rsi'] > 30) &
# 成交量条件放宽
(dataframe['volume'] > dataframe['volume'].rolling(window=10).mean()) &
# MACD空头
(dataframe['macd'] < dataframe['macdsignal'])
),
['enter_short', 'enter_tag']] = (1, 'short_trend_simple')
# 空头入场条件 - 条件2：突破下降
dataframe.loc[
(
# 价格突破支撑位
(dataframe['close'] < dataframe['low'].shift(1).rolling(window=5).min()) &
# 下降动量增强
(dataframe['close'].pct_change() < -0.005) &
# 非超卖区
(dataframe['rsi'] > 35) &
# 确保不与第一个条件重复
(~dataframe['enter_short'].astype(bool))
),
['enter_short', 'enter_tag']] = (1, 'short_breakdown')
# 空头入场条件 - 条件3：均线死叉
dataframe.loc[
(
# 短期均线下穿长期均线
(qtpylib.crossed_below(dataframe['ma5'], dataframe['ma10'])) &
# 价格已经在中期均线下方
(dataframe['close'] < dataframe['ma20']) &
# 确保不与其他条件重复
(~dataframe['enter_short'].astype(bool))
),
['enter_short', 'enter_tag']] = (1, 'short_ma_cross')
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
出场信号逻辑 - 优化盈利能力和降低风险
"""
# 清除之前的出场条件
dataframe['exit_short'] = 0
dataframe['exit_long'] = 0
# 空头出场条件 - 价格反转
price_reversal = (
# 价格反转
(dataframe['close'] > dataframe['close'].shift(1)) &
(dataframe['close'] > dataframe['open']) & # 收阳
(dataframe['volume'] > dataframe['volume'].rolling(window=10).mean()) # 放量上涨
)
# 空头出场条件 - 超卖反弹
oversold_bounce = (
# RSI超卖
(dataframe['rsi'] < 30) &
(dataframe['rsi'] > dataframe['rsi'].shift(1)) # RSI回升
)
# 空头出场条件 - 盈利保护
profit_protection = (
# 突破下轨后快速回升
(dataframe['close'] < dataframe['bb_lowerband']) &
(dataframe['close'] > dataframe['close'].shift(1)) &
(dataframe['close'].shift(1) > dataframe['close'].shift(2)) # 连续上涨
)
# 空头出场条件 - 趋势转变
trend_change = (
# 价格突破短期均线
(qtpylib.crossed_above(dataframe['close'], dataframe['ma10'])) |
# MACD柱状图由负转正
(dataframe['macdhist'] > 0) &
(dataframe['macdhist'].shift(1) < 0)
)
# 组合所有出场条件
dataframe.loc[price_reversal, ['exit_short', 'exit_tag']] = (1, 'price_reversal')
dataframe.loc[oversold_bounce, ['exit_short', 'exit_tag']] = (1, 'oversold_bounce')
dataframe.loc[profit_protection, ['exit_short', 'exit_tag']] = (1, 'profit_protection')
dataframe.loc[trend_change, ['exit_short', 'exit_tag']] = (1, 'trend_change')
return dataframe
def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime,
current_rate: float, current_profit: float, **kwargs) -> float:
"""
自定义止损逻辑 - 更精细的动态止损
"""
# 获取当前的dataframe
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if len(dataframe) == 0:
return self.stoploss
# 获取最新的K线数据
last_candle = dataframe.iloc[-1].squeeze()
# 计算ATR止损
atr_value = last_candle['atr']
# 根据盈利情况动态调整止损策略
if current_profit >= 0.03:
# 盈利较高，保护大部分利润，使用较紧的止损
return current_profit * 0.6
elif current_profit >= 0.015:
# 中等盈利，保护部分利润
return current_profit * 0.4
elif current_profit >= 0.008:
# 小额盈利，保本为主
return current_profit * 0.15
elif current_profit > 0:
# 微小盈利，保本为主
return 0
else:
# 亏损情况下，判断是否需要立即止损
# 趋势强烈反转，尽快止损
if (last_candle['close'] > last_candle['ma5']) and (last_candle['macd'] > last_candle['macdsignal']):
# 趋势向上反转，立即减小止损
return current_profit * 0.5
# 下跌动量减弱，略微放宽止损
if last_candle['rsi'] < 20 and last_candle['rsi'] > last_candle['rsi_slow']:
# RSI超卖且反弹迹象，提供更多空间
return self.stoploss * 1.3
# 默认返回原始止损设置
return self.stoploss
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:
"""
自定义仓位大小计算
"""
# 为DCA预留资金空间
return proposed_stake / self.max_dca_multiplier
def adjust_trade_position(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]:
"""
动态调整仓位 - 优化加仓策略
"""
# 获取交易数据
dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe)
if len(dataframe) == 0:
return None
filled_entries = trade.select_filled_orders(trade.entry_side)
if not filled_entries:
return None
last_entry = filled_entries[-1]
count_of_entries = trade.nr_of_successful_entries
# 空头加仓逻辑
if last_entry.side == "sell":
# 获取最新K线
last_candle = dataframe.iloc[-1]
prev_candle = dataframe.iloc[-2] if len(dataframe) > 1 else last_candle
# 计算加仓金额 - 基于亏损程度动态调整
stake_amount = filled_entries[0].stake_amount
# 条件1：价格突破新低 + 高阶空头趋势
if (current_profit < -0.005 and
last_candle['close'] < prev_candle['low'] and
last_candle['macd'] < last_candle['macdsignal'] and
count_of_entries < 2):
# 根据亏损程度调整加仓量 - 亏损越多加仓越少
adjustment_factor = max(0.5, 1.0 + current_profit) # 限制最低为0.5
new_stake = stake_amount * adjustment_factor
return new_stake, "short_dca_new_low"
# 条件2：小幅反弹后继续下跌
if (current_profit < -0.003 and
last_candle['close'] < last_candle['open'] and # 阴线
last_candle['close'] < last_candle['ma20'] and # 价格在中期均线下方
prev_candle['close'] > prev_candle['open'] and # 前一根是阳线
count_of_entries < 3):
# 使用标准金额加仓
return stake_amount * 0.8, "short_dca_dip_continuation"
return None
def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float,
time_in_force: str, current_time: datetime, entry_tag: Optional[str],
side: str, **kwargs) -> bool:
"""
在进入交易前进行额外的确认
"""
# 始终允许空头交易，不做额外检查
if side == "sell":
return True
return False
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])
strategies/Chan_SOL_2.py
+289 -145
View File
@@ -5,13 +5,6 @@ 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
@@ -19,174 +12,325 @@ 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
# freqtrade trade -c ./user_data/Chan/config/ChanLun_SOL.json --strategy ChanLun_SOL_2 --strategy-path ./user_data/Chan/strategies
# freqtrade backtesting -c ./user_data/Chan/config/ChanLun_SOL.json --strategy ChanLun_SOL_2 --strategy-path ./user_data/Chan/strategies --timerange=20250309-
# freqtrade download-data -c ./user_data/Chan/config/ChanLun_SOL.json -t 1m --pairs SOL/USDT:USDT --timerange=20250501-
# freqtrade hyperopt --hyperopt-loss SharpeHyperOptLossDaily --spaces roi stoploss --strategy ChanLun_SOL_2 --strategy-path ./user_data/strategies -c ./user_data/ChanLun_SOL.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):
class ChanLun_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"
# 优化的ROI设置 - 更快速获利
minimal_roi = {
"0": 0.253,
"120": 0.159,
"240": 0.052,
"360": 0
"0": 0.012, # 立即获利1.2%
"5": 0.01, # 5分钟后获利1%
"15": 0.007, # 15分钟后获利0.7%
"30": 0.005 # 30分钟后获利0.5%
}
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 = []
stoploss = -0.007 # 降低止损为0.7%
# 追踪止损设置 - 更积极的追踪止损
trailing_stop = True
trailing_stop_positive = 0.003 # 0.3%
trailing_stop_positive_offset = 0.005 # 0.5%
trailing_only_offset_is_reached = True
# 时间周期
timeframe = '5m'
informative_timeframe = '1h'
startup_candle_count = 200
# 只做空头策略
only_short = True
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"),]
informative_pairs = [(pair, self.informative_timeframe) for pair in pairs]
return informative_pairs
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
# 获取更高时间周期的数据
informative = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.informative_timeframe)
# 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)
# === 高时间周期指标 ===
# 三均线系统
informative['ema50'] = ta.EMA(informative, timeperiod=50)
informative['ema100'] = ta.EMA(informative, timeperiod=100)
informative['ema200'] = ta.SMA(informative, timeperiod=200) # 使用SMA作为长期趋势
# 趋势方向
informative['uptrend'] = (
(informative['ema50'] > informative['ema100']) &
(informative['ema100'] > informative['ema200']) &
(informative['close'] > informative['ema50'])
).astype(int)
informative['downtrend'] = (
(informative['ema50'] < informative['ema100']) &
(informative['ema100'] < informative['ema200']) &
(informative['close'] < informative['ema50'])
).astype(int)
# 强下降趋势
informative['strong_downtrend'] = (
(informative['ema50'] < informative['ema100']) &
(informative['ema100'] < informative['ema200']) &
(informative['close'] < informative['ema50']) &
(informative['ema50'].shift(3) < informative['ema50']) # 确认EMA50下降
).astype(int)
# 添加高时间周期的ADX指标
informative['adx'] = ta.ADX(informative, timeperiod=14)
# 添加高时间周期的波动率
informative['atr'] = ta.ATR(informative, timeperiod=14)
informative['atr_percent'] = (informative['atr'] / informative['close']) * 100
# 高时间周期RSI
informative['rsi'] = ta.RSI(informative, timeperiod=14)
# 将informative数据帧中的列重命名，以便在合并后区分
for col in informative.columns:
if col not in ['date', 'open', 'high', 'low', 'close', 'volume']:
informative[f"{col}_{self.informative_timeframe}"] = informative[col]
# 删除原始列，只保留重命名后的列和必要的日期、OHLCV列
for col in list(informative.columns):
if col not in ['date', 'open', 'high', 'low', 'close', 'volume'] and not col.endswith(f"_{self.informative_timeframe}"):
del informative[col]
# 打印列名以便调试
logger.info(f"Informative columns after renaming: {informative.columns.tolist()}")
# 合并数据 - 使用正确的参数
dataframe = resampled_merge(dataframe, informative, self.informative_timeframe)
# 打印合并后的列名以便调试
logger.info(f"Dataframe columns after merge: {dataframe.columns.tolist()}")
# === 主时间周期指标 ===
# 布林带
bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(dataframe), window=20, stds=2)
dataframe['bb_lowerband'] = bollinger['lower']
dataframe['bb_middleband'] = bollinger['mid']
dataframe['bb_upperband'] = bollinger['upper']
dataframe['bb_width'] = ((bollinger['upper'] - bollinger['lower']) / bollinger['mid'])
# 动量指标
dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)
dataframe['mfi'] = ta.MFI(dataframe, timeperiod=14)
# MACD
macd = ta.MACD(dataframe)
dataframe['macd'] = macd['macd']
dataframe['macdsignal'] = macd['macdsignal']
dataframe['macdhist'] = macd['macdhist']
# 均线
dataframe['ema9'] = ta.EMA(dataframe, timeperiod=9)
dataframe['ema21'] = ta.EMA(dataframe, timeperiod=21)
dataframe['ema50'] = ta.EMA(dataframe, timeperiod=50)
dataframe['sma200'] = ta.SMA(dataframe, timeperiod=200)
# 成交量
dataframe['volume_mean'] = dataframe['volume'].rolling(window=20).mean()
dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_mean']
# 波动率
dataframe['atr'] = ta.ATR(dataframe, timeperiod=14)
# ADX - 趋势强度指标
dataframe['adx'] = ta.ADX(dataframe, timeperiod=14)
# 价格突破
dataframe['upper_break'] = (
(dataframe['close'] > dataframe['bb_upperband']) &
(dataframe['close'].shift() <= dataframe['bb_upperband'].shift())
).astype(int)
dataframe['lower_break'] = (
(dataframe['close'] < dataframe['bb_lowerband']) &
(dataframe['close'].shift() >= dataframe['bb_lowerband'].shift())
).astype(int)
# 均线交叉
dataframe['ema_cross_up'] = (
(dataframe['ema9'] > dataframe['ema21']) &
(dataframe['ema9'].shift() <= dataframe['ema21'].shift())
).astype(int)
dataframe['ema_cross_down'] = (
(dataframe['ema9'] < dataframe['ema21']) &
(dataframe['ema9'].shift() >= dataframe['ema21'].shift())
).astype(int)
# 超买超卖区域
dataframe['rsi_oversold'] = (dataframe['rsi'] < 30).astype(int)
dataframe['rsi_overbought'] = (dataframe['rsi'] > 70).astype(int)
# 价格与均线的关系
dataframe['price_above_ema50'] = (dataframe['close'] > dataframe['ema50']).astype(int)
dataframe['price_below_ema50'] = (dataframe['close'] < dataframe['ema50']).astype(int)
# 趋势强度
dataframe['strong_trend'] = (dataframe['adx'] > 25).astype(int)
# 添加蜡烛图形态识别
dataframe['doji'] = ta.CDLDOJI(dataframe['open'], dataframe['high'], dataframe['low'], dataframe['close'])
dataframe['engulfing'] = ta.CDLENGULFING(dataframe['open'], dataframe['high'], dataframe['low'], dataframe['close'])
dataframe['hammer'] = ta.CDLHAMMER(dataframe['open'], dataframe['high'], dataframe['low'], dataframe['close'])
dataframe['shooting_star'] = ta.CDLSHOOTINGSTAR(dataframe['open'], dataframe['high'], dataframe['low'], dataframe['close'])
# 价格动量
dataframe['momentum'] = dataframe['close'] - dataframe['close'].shift(5)
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:
# 检查列名是否存在
downtrend_col = 'resample_60_downtrend_1h'
strong_downtrend_col = 'resample_60_strong_downtrend_1h'
adx_col = 'resample_60_adx_1h'
rsi_col = 'resample_60_rsi_1h'
dataframe.loc[
# 如果列名不存在，使用替代方案
for col, default_value in [
(downtrend_col, 0),
(strong_downtrend_col, 0),
(adx_col, 25),
(rsi_col, 50)
]:
if col not in dataframe.columns:
logger.warning(f"Column {col} not found in dataframe. Creating with default value {default_value}.")
dataframe[col] = default_value
# 禁用多头入场
dataframe['enter_long'] = 0
# 空头入场条件 - 专注于空头策略
short_conditions = (
# 高时间周期处于下降趋势
(dataframe[downtrend_col] > 0) &
# 趋势强度确认
(dataframe[adx_col] > 25) &
# 条件1: 价格突破上轨后回落 + 成交量确认
(
#(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['upper_break'].rolling(window=5).sum() > 0) & # 最近5根K线内有突破上轨
(dataframe['close'] < dataframe['close'].shift(2)) & # 价格开始下跌
(dataframe['close'] < dataframe['ema9']) & # 价格在短期均线下方
(dataframe['volume_ratio'] > 1.3) & # 成交量放大
(dataframe['rsi'] < 70) & # RSI不在极度超买区
(dataframe['rsi'] > 40) & # RSI不在超卖区
(dataframe[rsi_col] < 60) # 高时间周期RSI不过高
) |
# 条件2: 均线死叉 + RSI超买回落 + 趋势确认
(
#(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')
(dataframe['ema_cross_down'] > 0) & # 均线死叉
(dataframe['rsi'] > 55) & # RSI相对较高
(dataframe['rsi'] < dataframe['rsi'].shift(3)) & # RSI下降
(dataframe['volume_ratio'] > 1.2) & # 成交量放大
(dataframe['adx'] > 20) & # ADX显示有一定趋势强度
((dataframe['shooting_star'] > 0) | (dataframe['engulfing'] < 0)) # 流星线或看跌吞没形态
) |
# 条件3: 价格在高点回落 + 强趋势
(
(dataframe['close'] < dataframe['high'].shift()) &
(dataframe['high'].shift() > dataframe['high'].shift(2)) &
(dataframe['close'] < dataframe['ema21']) &
(dataframe['adx'] > 30) &
(dataframe['rsi'] < dataframe['rsi'].shift()) &
(dataframe['rsi'].shift() > 65) &
(dataframe['volume_ratio'] > 1.0)
) |
# 条件4: 强下降趋势确认
(
(dataframe[strong_downtrend_col] > 0) &
(dataframe['close'] < dataframe['ema21']) &
(dataframe['close'] < dataframe['close'].shift(3)) &
(dataframe['momentum'] < 0) &
(dataframe['volume_ratio'] > 1.1) &
(dataframe['adx'] > 25)
)
)
dataframe.loc[short_conditions, 'enter_short'] = 1
dataframe.loc[short_conditions, 'enter_tag'] = 'chan_sol_short'
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
dataframe.loc[
# 禁用多头出场
dataframe['exit_long'] = 0
# 空头出场条件 - 更精确的出场
short_exit_conditions = (
# 条件1: 趋势反转信号
(
#(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['ema_cross_up'] > 0) & # 均线金叉
(dataframe['volume_ratio'] > 1.0) # 成交量确认
) |
# 条件2: 价格突破中期均线
(
#(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')
(dataframe['close'] > dataframe['ema21']) &
(dataframe['close'].shift() < dataframe['ema21'].shift()) & # 确认是刚刚突破
(dataframe['volume_ratio'] > 1.2) # 成交量确认
) |
# 条件3: 超卖信号
(
(dataframe['rsi'] < 30) & # RSI超卖
(dataframe['close'] < dataframe['bb_lowerband']) # 价格突破下轨
) |
# 条件4: 动量减弱
(
(dataframe['rsi'] < 35) &
(dataframe['rsi'] > dataframe['rsi'].shift()) &
(dataframe['rsi'].shift() > dataframe['rsi'].shift(2)) & # RSI连续两根K线上升
(dataframe['momentum'] > 0) # 价格动量转为正
) |
# 条件5: 锤子线形态 (潜在反转信号)
(
(dataframe['hammer'] > 0) &
(dataframe['volume_ratio'] > 1.3)
)
)
dataframe.loc[short_exit_conditions, 'exit_short'] = 1
dataframe.loc[short_exit_conditions, 'exit_tag'] = 'chan_sol_short_exit'
return dataframe
def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float,
time_in_force: str, current_time: datetime, entry_tag: Optional[str],
side: str, **kwargs) -> bool:
"""
在进入交易前进行额外的确认
"""
# 只做空头交易
if side == "sell" and entry_tag == "chan_sol_short":
return True
return False
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: