-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtechnical_indicators_calculator.py
116 lines (92 loc) · 3.67 KB
/
technical_indicators_calculator.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
115
116
from ta.momentum import RSIIndicator
from ta.trend import MACD
import numpy as np
import pandas
from ta.volatility import BollingerBands
class Company:
def __init__(self, symbol):
self.symbol = symbol
self.technical_indicators = None
self.prices = None
self.macd_diff = None
self.rsi = None
# def generate_buy_sell_signals(condition_buy, condition_sell, dataframe, strategy):
# last_signal = None
# indicators = []
# buy = []
# sell = []
# for i in range(0, len(dataframe)):
# # if buy condition is true and last signal was not Buy
# if condition_buy(i, dataframe) and last_signal != 'Buy':
# last_signal = 'Buy'
# indicators.append(last_signal)
# buy.append(dataframe['Close'].iloc[i])
# sell.append(np.nan)
# # if sell condition is true and last signal was Buy
# elif condition_sell(i, dataframe) and last_signal == 'Buy':
# last_signal = 'Sell'
# indicators.append(last_signal)
# buy.append(np.nan)
# sell.append(dataframe['Close'].iloc[i])
# else:
# indicators.append(last_signal)
# buy.append(np.nan)
# sell.append(np.nan)
#
# dataframe[f'{strategy}_Last_Signal'] = np.array(last_signal)
# dataframe[f'{strategy}_Indicator'] = np.array(indicators)
# dataframe[f'{strategy}_Buy'] = np.array(buy)
# dataframe[f'{strategy}_Sell'] = np.array(sell)
# def set_technical_indicators(company):
# company.technical_indicators = pandas.DataFrame()
# company.technical_indicators['Close'] = company.prices
#
# get_macd(company)
# get_rsi(company)
# get_bollinger_bands(config, company)
def get_macd(company):
close_prices = company.prices
dataframe = company.technical_indicators
window_slow = 26
signal = 9
window_fast = 12
macd = MACD(company.prices, window_slow, window_fast, signal)
dataframe['MACD'] = macd.macd()
dataframe['MACD_Histogram'] = macd.macd_diff()
dataframe['MACD_Signal'] = macd.macd_signal()
# generate_buy_sell_signals(
# lambda x, dataframe: dataframe['MACD'].values[x] < dataframe['MACD_Signal'].iloc[x],
# lambda x, dataframe: dataframe['MACD'].values[x] > dataframe['MACD_Signal'].iloc[x],
# dataframe,
# 'MACD')
# return dataframe
return macd.macd_diff()
def get_rsi(company):
close_prices = company.prices
dataframe = company.technical_indicators
rsi_time_period = 14
rsi_indicator = RSIIndicator(close_prices, rsi_time_period)
dataframe['RSI'] = rsi_indicator.rsi()
# print(rsi_indicator.rsi())
low_rsi = 40
high_rsi = 60
# generate_buy_sell_signals(
# lambda x, dataframe: dataframe['RSI'].values[x] < low_rsi,
# lambda x, dataframe: dataframe['RSI'].values[x] > high_rsi,
# dataframe, 'RSI')
# return dataframe
return rsi_indicator.rsi()
def get_bollinger_bands(company):
close_prices = company.prices
dataframe = company.technical_indicators
window = 20
indicator_bb = BollingerBands(close=close_prices, window=window, window_dev=2)
# Add Bollinger Bands features
dataframe['Bollinger_Bands_Middle'] = indicator_bb.bollinger_mavg()
dataframe['Bollinger_Bands_Upper'] = indicator_bb.bollinger_hband()
dataframe['Bollinger_Bands_Lower'] = indicator_bb.bollinger_lband()
# generate_buy_sell_signals(
# lambda x, signal: signal['Close'].values[x] < signal['Bollinger_Bands_Lower'].values[x],
# lambda x, signal: signal['Close'].values[x] > signal['Bollinger_Bands_Upper'].values[x],
# dataframe, 'Bollinger_Bands')
return dataframe