-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathtrade_bot.py
316 lines (278 loc) · 12.2 KB
/
trade_bot.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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
import logging
import pdb
import datetime
from datetime import datetime
import pandas as pd
import os
from oanda.connect import Connect
from config import CONFIG
from utils import *
from candle.candle import *
from candle.candlelist import CandleList
from candle.candlelist_utils import *
from trade_utils import *
from trade import Trade
# create logger
tb_logger = logging.getLogger(__name__)
tb_logger.setLevel(logging.DEBUG)
class TradeBot(object):
'''
This class represents an automatic Trading bot
Class variables
---------------
start: datetime, Required
Datetime that this Bot will start operating. i.e. 20-03-2017 08:20:00s
end: datetime, Required
Datetime that this Bot will end operating. i.e. 20-03-2020 08:20:00s
pair: str, Required
Currency pair used in the trade. i.e. AUD_USD
timeframe: str, Required
Timeframe used for the trade. Possible values are: D,H12,H10,H8,H4,H1
'''
def __init__(self, start, end, pair, timeframe):
self.start = start
self.end = end
self.pair = pair
self.timeframe = timeframe
def run(self, discard_sat=True):
'''
This function will run the Bot from start to end
one candle at a time
Parameter
---------
discard_sat : Bool
If this is set to True, then the Trade wil not
be taken if IC falls on a Saturday. Default: True
Returns
-------
TradeList object with Trades taken. None if no trades
were taken
'''
tb_logger.info("Running...")
conn = Connect(instrument=self.pair,
granularity=self.timeframe)
ser_dir = None
if CONFIG.has_option('general', 'ser_data_dir'):
ser_dir = CONFIG.get('general', 'ser_data_dir')
delta = nhours = None
if self.timeframe == "D":
nhours = 24
delta = timedelta(hours=24)
else:
p1 = re.compile('^H')
m1 = p1.match(self.timeframe)
if m1:
nhours = int(self.timeframe.replace('H', ''))
delta = timedelta(hours=int(nhours))
# convert to datetime the start and end for this TradeBot
startO = pd.datetime.strptime(self.start, '%Y-%m-%d %H:%M:%S')
endO = pd.datetime.strptime(self.end, '%Y-%m-%d %H:%M:%S')
loop = 0
tlist = []
tend = SRlst = None
# calculate the start datetime for the CList that will be used
# for calculating the S/R areas
delta_period = periodToDelta(CONFIG.getint('trade_bot', 'period_range'),
self.timeframe)
initc_date = startO-delta_period
# Get now a CandleList from 'initc_date' to 'startO' which is the
# total time interval for this TradeBot
res = conn.query(start=initc_date.isoformat(),
end=endO.isoformat(),
indir=ser_dir)
clO = CandleList(res)
while startO <= endO:
if tend is not None:
# this means that there is currently an active trade
if startO <= tend:
startO = startO + delta
loop += 1
continue
else:
tend = None
tb_logger.info("Trade bot - analyzing candle: {0}".format(startO.isoformat()))
sub_clO = clO.slice(initc_date,
startO)
dt_str = startO.strftime("%d_%m_%Y_%H_%M")
if loop == 0:
outfile_txt = "{0}/srareas/{1}.{2}.{3}.halist.txt".format(CONFIG.get("images", "outdir"),
self.pair, self.timeframe, dt_str)
outfile_png = "{0}/srareas/{1}.{2}.{3}.halist.png".format(CONFIG.get("images", "outdir"),
self.pair, self.timeframe, dt_str)
SRlst = calc_SR(sub_clO, outfile=outfile_png)
f = open(outfile_txt, 'w')
res = SRlst.print()
# print SR report to file
f.write(res)
f.close()
tb_logger.info("Identified HAreaList for time {0}:".format(startO.isoformat()))
tb_logger.info("{0}".format(res))
elif loop >= CONFIG.getint('trade_bot',
'period'):
# An entire cycle has occurred. Invoke .calc_SR
outfile_txt = "{0}/srareas/{1}.{2}.{3}.halist.txt".format(CONFIG.get("images", "outdir"),
self.pair, self.timeframe, dt_str)
outfile_png = "{0}/srareas/{1}.{2}.{3}.halist.png".format(CONFIG.get("images", "outdir"),
self.pair, self.timeframe, dt_str)
SRlst = calc_SR(sub_clO, outfile=outfile_png)
f = open(outfile_txt, 'w')
res = SRlst.print()
tb_logger.info("Identified HAreaList for time {0}:".format(startO.isoformat()))
tb_logger.info("{0}".format(res))
# print SR report to file
f.write(res)
f.close()
loop = 0
# fetch candle for current datetime
res = conn.query(start=startO.isoformat(),
count=1,
indir=ser_dir)
# this is the current candle that
# is being checked
c_candle = Candle(dict_data=res['candles'][0])
c_candle.time = datetime.strptime(c_candle.time,
'%Y-%m-%dT%H:%M:%S.%fZ')
# c_candle.time is not equal to startO
# when startO is non-working day, for example
delta1hr = timedelta(hours=1)
if (c_candle.time != startO) and (abs(c_candle.time-startO) > delta1hr):
loop += 1
tb_logger.info("Analysed dt {0} is not the same than APIs returned dt {1}."
" Skipping...".format(startO, c_candle.time))
startO = startO + delta
continue
#check if there is any HArea overlapping with c_candle
HAreaSel, sel_ix = SRlst.onArea(candle=c_candle)
if HAreaSel is not None:
c_candle.set_candle_features()
# guess the if trade is 'long' or 'short'
newCl = clO.slice(start=initc_date, end=c_candle.time)
type = get_trade_type(c_candle.time, newCl)
SL = adjust_SL(type, newCl, CONFIG.getint('trade_bot', 'n_SL'))
prepare = False
if c_candle.indecision_c(ic_perc=CONFIG.getint('general', 'ic_perc')) is True:
prepare = True
elif type == 'short' and c_candle.colour == 'red':
prepare = True
elif type == 'long' and c_candle.colour == 'green':
prepare = True
# discard if IC falls on a Saturday
if c_candle.time.weekday() == 5 and discard_sat is True:
tb_logger.info("Possible trade at {0} falls on Sat. Skipping...".format(c_candle.time))
prepare = False
if prepare is True:
t = prepare_trade(
tb_obj=self,
type=type,
SL=SL,
ic=c_candle,
harea_sel=HAreaSel,
delta=delta,
add_pips=CONFIG.getint('trade', 'add_pips'))
t.tot_SR = len(SRlst.halist)
t.rank_selSR = sel_ix
t.SRlst = SRlst
tlist.append(t)
startO = startO+delta
loop += 1
tb_logger.info("Run done")
if len(tlist) == 0:
return None
else:
return tlist
class TradeDiscover(object):
'''
This class represents an automatic Trading bot to discover new trades
Class variables
---------------
start : datetime
pair: str, Required
Currency pair used in the trade. i.e. AUD_USD
timeframe: str, Required
Timeframe used for the trade. Possible values are: D,H12,H10,H8,H4
'''
def __init__(self, start, pair, timeframe):
self.start = start
self.pair = pair
self.timeframe = timeframe
def run(self):
"""
Run the bot from self.start
Returns
-------
Trade object or none
"""
tb_logger.info("Running...")
conn = Connect(instrument=self.pair,
granularity=self.timeframe)
ser_dir = None
if CONFIG.has_option('general', 'ser_data_dir'):
ser_dir = CONFIG.get('general', 'ser_data_dir')
delta = nhours = None
if self.timeframe == "D":
nhours = 24
delta = timedelta(hours=24)
else:
p1 = re.compile('^H')
m1 = p1.match(self.timeframe)
if m1:
nhours = int(self.timeframe.replace('H', ''))
delta = timedelta(hours=int(nhours))
# calculate the start datetime for the CList that will be used
# for calculating the S/R areas
delta_period = periodToDelta(CONFIG.getint('trade_bot', 'period_range'),
self.timeframe)
initc_date = self.start - delta_period
# Get now a CandleList from 'initc_date' to 'startO' which is the
# total time interval for this TradeBot
res = conn.query(start=initc_date.strftime("%Y-%m-%dT%H:%M:%S"),
end=self.start.strftime("%Y-%m-%dT%H:%M:%S"),
indir=ser_dir)
clO = CandleList(res)
dt_str = self.start.strftime("%d_%m_%Y_%H_%M")
outfile_png = "{0}/srareas/{1}.{2}.{3}.halist.png".format(CONFIG.get("images", "outdir"),
self.pair, self.timeframe, dt_str)
SRlst = calc_SR(clO, outfile=outfile_png)
# fetch candle for current datetime
res = conn.query(start=self.start.strftime("%Y-%m-%dT%H:%M:%S"),
count=1,
indir=ser_dir)
# this is the current candle that
# is being checked
c_candle = Candle(dict_data=res['candles'][0])
c_candle.time = datetime.strptime(c_candle.time,
'%Y-%m-%dT%H:%M:%S.%fZ')
# check if there is any HArea overlapping with c_candle
HAreaSel, sel_ix = SRlst.onArea(candle=c_candle)
if HAreaSel is not None:
c_candle.set_candle_features()
# guess the if trade is 'long' or 'short'
newCl = clO.slice(start=initc_date, end=c_candle.time)
type = get_trade_type(c_candle.time, newCl)
SL = adjust_SL(type, newCl, CONFIG.getint('trade_bot', 'n_SL'))
prepare = False
if c_candle.indecision_c(ic_perc=CONFIG.getint('general', 'ic_perc')) is True:
prepare = True
elif type == 'short' and c_candle.colour == 'red':
prepare = True
elif type == 'long' and c_candle.colour == 'green':
prepare = True
# discard if IC falls on a Saturday
if c_candle.time.weekday() == 5 and discard_sat is True:
tb_logger.info("Possible trade at {0} falls on Sat. Skipping...".format(c_candle.time))
prepare = False
t = None
if prepare is True:
t = prepare_trade(
tb_obj=self,
type=type,
SL=SL,
ic=c_candle,
harea_sel=HAreaSel,
delta=delta,
add_pips=CONFIG.getint('trade', 'add_pips'))
t.tot_SR = len(SRlst.halist)
t.rank_selSR = sel_ix
t.SRlst = SRlst
return t
tb_logger.info("Run done")