-
Notifications
You must be signed in to change notification settings - Fork 88
/
Copy pathCCIStrategy.py
114 lines (96 loc) · 4.24 KB
/
CCIStrategy.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
# --- Do not remove these libs ---
from freqtrade.strategy.interface import IStrategy
from typing import Dict, List
from functools import reduce
from pandas import DataFrame, Series, DatetimeIndex, merge
# --------------------------------
import talib.abstract as ta
import freqtrade.vendor.qtpylib.indicators as qtpylib
class CCIStrategy(IStrategy):
# Minimal ROI designed for the strategy.
# This attribute will be overridden if the config file contains "minimal_roi"
minimal_roi = {"0": 0.1}
# Optimal stoploss designed for the strategy
# This attribute will be overridden if the config file contains "stoploss"
stoploss = -0.02
# Optimal timeframe for the strategy
timeframe = "1m"
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
dataframe = self.resample(dataframe, self.timeframe, 5)
dataframe["cci_one"] = ta.CCI(dataframe, timeperiod=170)
dataframe["cci_two"] = ta.CCI(dataframe, timeperiod=34)
dataframe["rsi"] = ta.RSI(dataframe)
dataframe["mfi"] = ta.MFI(dataframe)
dataframe["cmf"] = self.chaikin_mf(dataframe)
# required for graphing
bollinger = qtpylib.bollinger_bands(dataframe["close"], window=20, stds=2)
dataframe["bb_lowerband"] = bollinger["lower"]
dataframe["bb_upperband"] = bollinger["upper"]
dataframe["bb_middleband"] = bollinger["mid"]
return dataframe
def populate_buy_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Based on TA indicators, populates the buy signal for the given dataframe
:param dataframe: DataFrame
:return: DataFrame with buy column
"""
dataframe.loc[
(
(dataframe["cci_one"] < -100)
& (dataframe["cci_two"] < -100)
& (dataframe["cmf"] < -0.1)
& (dataframe["mfi"] < 25)
# insurance
& (dataframe["resample_medium"] > dataframe["resample_short"])
& (dataframe["resample_long"] < dataframe["close"])
),
"buy",
] = 1
return dataframe
def populate_sell_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Based on TA indicators, populates the sell signal for the given dataframe
:param dataframe: DataFrame
:return: DataFrame with buy column
"""
dataframe.loc[
(
(dataframe["cci_one"] > 100)
& (dataframe["cci_two"] > 100)
& (dataframe["cmf"] > 0.3)
& (dataframe["resample_sma"] < dataframe["resample_medium"])
& (dataframe["resample_medium"] < dataframe["resample_short"])
),
"sell",
] = 1
return dataframe
def chaikin_mf(self, df, periods=20):
close = df["close"]
low = df["low"]
high = df["high"]
volume = df["volume"]
mfv = ((close - low) - (high - close)) / (high - low)
mfv = mfv.fillna(0.0) # float division by zero
mfv *= volume
cmf = mfv.rolling(periods).sum() / volume.rolling(periods).sum()
return Series(cmf, name="cmf")
def resample(self, dataframe, interval, factor):
# defines the reinforcement logic
# resampled dataframe to establish if we are in an uptrend, downtrend or sideways trend
df = dataframe.copy()
df = df.set_index(DatetimeIndex(df["date"]))
ohlc_dict = {"open": "first", "high": "max", "low": "min", "close": "last"}
df = df.resample(str(int(interval[:-1]) * factor) + "min", label="right").agg(
ohlc_dict
)
df["resample_sma"] = ta.SMA(df, timeperiod=100, price="close")
df["resample_medium"] = ta.SMA(df, timeperiod=50, price="close")
df["resample_short"] = ta.SMA(df, timeperiod=25, price="close")
df["resample_long"] = ta.SMA(df, timeperiod=200, price="close")
df = df.drop(columns=["open", "high", "low", "close"])
df = df.resample(interval[:-1] + "min")
df = df.interpolate(method="time")
df["date"] = df.index
df.index = range(len(df))
dataframe = merge(dataframe, df, on="date", how="left")
return dataframe