-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathapp.py
139 lines (121 loc) · 6.06 KB
/
app.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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from matplotlib.dates import DateFormatter
from matplotlib.patches import Rectangle
import matplotlib.dates as mdates
import sys
def main():
inputFile = "eurusd_hour.csv"
index_start = int(sys.argv[1])
index_end = int(sys.argv[2])
app = App(inputFile, index_start, index_end)
app.launch()
class App:
def __init__(self, inputFile, index_start, index_end):
self.inputFile = inputFile
self.index_start = index_start
self.index_end = index_end
def launch(self):
print("app started")
print(self.index_end, self.index_start)
data = self.loadCsv(self.inputFile)
data_cleaned = self.dataCleaning(data)
#self.plotCandleStick(data_cleaned,400,500)
imbalance = self.generateImbalance(data_cleaned)
self.plotCandleStickImbalance(data_cleaned,self.index_start,self.index_end)
return 1
def loadCsv(self, fileLocation):
data = pd.read_csv(fileLocation)
return data
def resetDate(self, data):
data['Date and Time'] = pd.to_datetime(data['Date'] + ' ' + data['Time'], format="%Y-%m-%d %H:%M")
for index, row in data.iterrows():
if index>1:
previousDate = data.iloc[index-1]['Date and Time']
data.at[index, 'Date and Time'] = previousDate + pd.DateOffset(hours=1)
def addCandlestickSign(self, data):
data['isUp'] = True
data.loc[data['BO'] >= data['BC'], 'isUp'] = False
def addHeight(self, data):
data['total_height'] = abs(data["BH"] - data["BL"])
data['height'] = abs(data["BO"] - data["BC"])
def dataCleaning(self, data):
self.resetDate(data)
self.addCandlestickSign(data)
self.addHeight(data)
return data
def plotCandleStick(self, data, start,end):
data = data.iloc[start:end]
up_price_bottom = data.loc[data['isUp']][['BO','BL', 'height', 'Date and Time', 'total_height']]
down_price_bottom = data.loc[~data['isUp']][['BC', 'BL', 'height', 'Date and Time', 'total_height']]
width = 0.02
width_narrow= 0.002
fig, ax = plt.subplots()
ax.bar(up_price_bottom['Date and Time'], up_price_bottom['height'], width=width, bottom=up_price_bottom['BO'], color='green')
ax.bar(up_price_bottom['Date and Time'], up_price_bottom['total_height'], width=width_narrow, bottom=up_price_bottom['BL'], color='green')
ax.bar(down_price_bottom['Date and Time'], down_price_bottom['height'], width=width, bottom=down_price_bottom['BC'], color='red')
ax.bar(down_price_bottom['Date and Time'], down_price_bottom['total_height'], width=width_narrow, bottom=down_price_bottom['BL'], color='red')
formatDate = DateFormatter("%d-%m")
ax.xaxis.set_major_formatter(formatDate)
fig.autofmt_xdate()
plt.show()
def _check_imbalance(self, data, index):
if data.iloc[index, data.columns.get_loc("isUp")]:
high = data.iloc[index, data.columns.get_loc("BH")]
low = data.iloc[index+2, data.columns.get_loc("BL")]
if high < low:
return True
else:
high = data.iloc[index+2, data.columns.get_loc("BH")]
low = data.iloc[index, data.columns.get_loc("BL")]
if low > high:
return True
return False
def generateImbalance(self, data):
data_imbalance =[]
data = data.reset_index(drop=True)
size, col = data.shape
for index, row in data.iterrows():
first_color = row['isUp']
if index<size-1:
second_color = data.iloc[index+1, data.columns.get_loc("isUp")]
if second_color == first_color:
if index<size-2 and data.iloc[index+2, data.columns.get_loc("isUp")] == first_color:
#Three consecutive values
if self._check_imbalance(data, index):
data_imbalance.append(data.iloc[index:index+3])
print(data_imbalance)
return data_imbalance
def plotCandleStickImbalance(self, data, start, end):
data = data.iloc[start:end]
data_imbalance = self.generateImbalance(data)
up_price_bottom = data.loc[data['isUp']][['BO','BL', 'height', 'Date and Time', 'total_height']]
down_price_bottom = data.loc[~data['isUp']][['BC', 'BL', 'height', 'Date and Time', 'total_height']]
width = 0.02
width_narrow= 0.007
fig, ax = plt.subplots()
ax.bar(up_price_bottom['Date and Time'], up_price_bottom['height'], width=width, bottom=up_price_bottom['BO'], color='green')
ax.bar(up_price_bottom['Date and Time'], up_price_bottom['total_height'], width=width_narrow, bottom=up_price_bottom['BL'], color='green')
ax.bar(down_price_bottom['Date and Time'], down_price_bottom['height'], width=width, bottom=down_price_bottom['BC'], color='red')
ax.bar(down_price_bottom['Date and Time'], down_price_bottom['total_height'], width=width_narrow, bottom=down_price_bottom['BL'], color='red')
for imbalance in data_imbalance:
imbalance = imbalance.reset_index(drop = True)
if imbalance.iloc[0]['isUp']:
x_rec = imbalance.loc[0]['Date and Time']
y_rec = imbalance.loc[0]['BH']
rec_height = imbalance.loc[2]['BL']-imbalance.loc[0]['BH']
else:
x_rec = imbalance.loc[0]['Date and Time']
y_rec = imbalance.loc[2]['BH']
rec_height = imbalance.loc[0]['BL']-imbalance.loc[2]['BH']
start = mdates.date2num(x_rec)
end = mdates.date2num(x_rec)
rec_width = 0.2
ax.add_patch(Rectangle((mdates.date2num(x_rec), y_rec), rec_width, rec_height,facecolor='yellow', alpha=0.4,edgecolor='orange',linewidth=0.5))
formatDate = DateFormatter("%d-%m")
ax.xaxis.set_major_formatter(formatDate)
fig.autofmt_xdate()
plt.show()
if __name__ == '__main__':
main()