189 lines
6.8 KiB
Python
189 lines
6.8 KiB
Python
# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
|
||
# flake8: noqa: F401
|
||
# isort: skip_file
|
||
# --- Do not remove these imports ---
|
||
from dingtalk import DingTalkBot
|
||
import numpy as np
|
||
import pandas as pd
|
||
from datetime import datetime, timedelta, timezone
|
||
from pandas import DataFrame
|
||
from typing import Optional, Union
|
||
from freqtrade.strategy import (
|
||
IStrategy,
|
||
Trade,
|
||
Order,
|
||
PairLocks,
|
||
informative, # @informative decorator
|
||
# Hyperopt Parameters
|
||
BooleanParameter,
|
||
CategoricalParameter,
|
||
DecimalParameter,
|
||
IntParameter,
|
||
RealParameter,
|
||
# timeframe helpers
|
||
timeframe_to_minutes,
|
||
timeframe_to_next_date,
|
||
timeframe_to_prev_date,
|
||
# Strategy helper functions
|
||
merge_informative_pair,
|
||
stoploss_from_absolute,
|
||
stoploss_from_open,
|
||
)
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
# --------------------------------
|
||
# Add your lib to import here
|
||
import talib.abstract as ta
|
||
from technical import qtpylib
|
||
|
||
|
||
webhook = "https://oapi.dingtalk.com/robot/send?access_token=21de667159edadd33172c6ec414a2addf9c6359189350ffd36819d2a20e8a0f4" # 填写你的webhook
|
||
secret = "SEC43a0fa0b29717f98637a119b92a0bd5f7b2b6da671bdd2bd1279ed8323454d5e" # 填写你的加签token（如果有），否则留空
|
||
|
||
# CTA 群机器人
|
||
# webhook = "https://oapi.dingtalk.com/robot/send?access_token=87c7abfcdd69b699c32da4e4f5981cd2ca6b0445474fc6ffb36f2ed0f6262fbb"
|
||
# secret = "SECf3d6b43f2f8a87ab91feffd052e71ec314fbf57a1842e483fe07af3c0a0e5aa6"
|
||
dingtalk = DingTalkBot(webhook, secret)
|
||
|
||
|
||
class CCIMultiTimeframeSpotStrategy(IStrategy):
|
||
# 策略参数
|
||
INTERFACE_VERSION = 3
|
||
|
||
minimal_roi = {"0": 100}
|
||
stoploss = -1
|
||
use_custom_stoploss = False
|
||
trailing_stop = False
|
||
timeframe = '4h'
|
||
|
||
use_exit_signal = True
|
||
exit_profit_only = False
|
||
ignore_roi_if_entry_signal = False
|
||
|
||
def order_filled(self, pair: str, trade: Trade, order: Order, current_time: datetime, **kwargs) -> None:
|
||
# 交易对
|
||
trading_pair = pair
|
||
|
||
# 时间
|
||
fill_time = order.order_filled_date or current_time
|
||
|
||
# 价格
|
||
fill_price = order.average or order.price
|
||
|
||
# 买入还是卖出
|
||
side = order.ft_order_side # 'buy' 或 'sell'
|
||
|
||
# 仓位(当前持仓数量)
|
||
position = trade.amount
|
||
|
||
# 或者使用日志
|
||
logger.info(
|
||
f"订单成交 - 交易对: {trading_pair}, "
|
||
f"时间: {fill_time}, "
|
||
f"价格: {fill_price}, "
|
||
f"方向: {side}, "
|
||
f"仓位: {position}"
|
||
)
|
||
# if self.config["runmode"].value in ("live", "dry_run"):
|
||
logger.info(f"11111111{self.config['runmode']}")
|
||
# dingtalk.send_text(
|
||
# content=f"订单成交 - 交易对: {trading_pair}, 时间: {fill_time}, 价格: {fill_price}, 方向: {side}, 仓位: {position}")
|
||
|
||
return None
|
||
|
||
def TD(self, dataframe:DataFrame):
|
||
close = dataframe['close'].to_list()
|
||
td = [0,0,0,0]
|
||
up = 0
|
||
down = 0
|
||
for i in range(4, len(close)):
|
||
if close[i] > close[i-4]:
|
||
up += 1
|
||
down = 0
|
||
td.append(up)
|
||
else:
|
||
down -= 1
|
||
up = 0
|
||
td.append(down)
|
||
return td
|
||
|
||
@informative('1d')
|
||
def populate_indicators_1d(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
||
dataframe['cci'] = ta.CCI(dataframe, timeperiod=26)
|
||
dataframe["adx"] = ta.ADX(dataframe)
|
||
macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9,)
|
||
dataframe["macd"] = macd["macd"]
|
||
dataframe["macdsignal"] = macd["macdsignal"]
|
||
dataframe["macdhist"] = macd["macdhist"]
|
||
logger.info(dataframe.tail())
|
||
return dataframe
|
||
|
||
@informative('1w')
|
||
def populate_indicators_1w(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
||
dataframe['cci'] = ta.CCI(dataframe, timeperiod=26)
|
||
logger.info(dataframe.tail())
|
||
return dataframe
|
||
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
||
# 计算4h CCI
|
||
dataframe['cci_4h'] = ta.CCI(dataframe, timeperiod=26)
|
||
dataframe['sma12_4h'] = ta.SMA(dataframe, timeperiod=12)
|
||
dataframe['sma26_4h'] = ta.SMA(dataframe, timeperiod=26)
|
||
dataframe["adx_4h"] = ta.ADX(dataframe)
|
||
|
||
macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9,)
|
||
dataframe["macd_4h"] = macd["macd"]
|
||
dataframe["macdsignal_4h"] = macd["macdsignal"]
|
||
dataframe["macdhist_4h"] = macd["macdhist"]
|
||
|
||
dataframe['cci_hist_4h'] = ta.CCI(dataframe['macdhist_1d'], dataframe['macdhist_1d'], dataframe['macdhist_1d'], timeperiod=26)
|
||
|
||
# dataframe['adx_hist_4h'] = ta.ADX(dataframe['macdhist_4h'])
|
||
dataframe['TD'] = self.TD(dataframe)
|
||
dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)
|
||
|
||
logger.info(dataframe.tail())
|
||
return dataframe
|
||
|
||
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
||
# 入场：1d与4h CCI < -100，且4h CCI上升（当前 > 前一根）
|
||
dataframe.loc[
|
||
(
|
||
(dataframe['cci_4h'] < -100) &
|
||
(dataframe['cci_1d'] < -100) &
|
||
(dataframe['cci_4h'] > dataframe['cci_4h'].shift(1)) &
|
||
# (dataframe['macdhist_1d'] < dataframe['macdhist_1d'].shift(1)) &
|
||
(dataframe['volume'] > 0)
|
||
),
|
||
'enter_long',
|
||
] = 1
|
||
|
||
return dataframe
|
||
|
||
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
||
# 离场：1d与4h CCI > 100，且4h CCI下降（当前 < 前一根）
|
||
dataframe.loc[
|
||
(
|
||
(dataframe['cci_4h'] > 100) &
|
||
(dataframe['cci_1d'] > 100) &
|
||
(dataframe['cci_4h'] < dataframe['cci_4h'].shift(1)) &
|
||
# (dataframe['macdhist_1d'] > dataframe['macdhist_1d'].shift(1)) &
|
||
(dataframe['volume'] > 0)
|
||
),
|
||
'exit_long',
|
||
] = 1
|
||
|
||
return dataframe
|
||
|
||
def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime,
|
||
current_rate: float, current_profit: float, after_fill: bool,
|
||
**kwargs) -> float | None:
|
||
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
|
||
candle = dataframe.iloc[-1].squeeze()
|
||
side = 1 if trade.is_short else -1
|
||
return stoploss_from_absolute(current_rate + (side * candle["atr"] * 3),
|
||
current_rate=current_rate,
|
||
is_short=trade.is_short,
|
||
leverage=trade.leverage)
|
||
|