-
Notifications
You must be signed in to change notification settings - Fork 6
/
Copy pathnnfx.py
498 lines (411 loc) · 19.1 KB
/
nnfx.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
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
import sys
import datetime # For datetime objects
import os.path # To manage paths
import sys # To find out the script name (in argv[0])
import backtrader as bt
from custom_indicators import *
from custom_functions import *
import BinaryGenerator as BG
import itertools
import time
import os
import glob
class NNFX(bt.Strategy):
params = dict(
base_ind='butter',
base_params=(40,3),
c1_ind='schaff',
c1_params=(20,50,10,0.5),
c2_ind='itrend',
c2_params=(30,),
volume_ind='damiani',
volume_params=(13,20,40,100,1.4,True),
exit_ind='ssl',
exit_params=(20,),
atr_period=14,
sl=1.5,
tp=3.0,
risk=2.0,
leverage=20,
oneplot=False,
verbose=False,
)
def log(self, txt, dt=None):
"""Logging Function"""
dt = dt or self.datas[0].datetime.date(0)
print('%s, %s' % (dt.isoformat(), txt))
def __init__(self):
"""Initialization"""
self.dv = DamianiVolatmeter(self.data)
# Strategy Declarations
self.order = None
self.broker.set_coc=True
self.accn_currency = 'USD'
self.inds = dict()
self.igs = dict()
self.closes = dict()
self.open_orders = dict()
self.data_dict = dict()
for i, d in enumerate(self.datas):
# Data Dictionary
self.data_dict[d._name] = d
# Open Orders
self.open_orders[d] = {
'sl': [],
'tp': [],
'tracker':[]
}
# Closes
self.closes[d] = d.close
# Binary Generator
self.igs[d] = BG.IndicatorGenerator(d)
self.inds[d] = dict()
# Money Management Indicators
self.inds[d]['atr'] = bt.indicators.ATR(d,period=self.p.atr_period, plot=False)
# Generate Strategy Binary Indicators
self.inds[d]['baseline'], self.inds[d]['too_far'] = self.igs[d].baseline_indicator(self.p.base_ind, self.p.base_params, plot=False)
self.inds[d]['c1'] = self.igs[d].entry_indicator(self.p.c1_ind, self.p.c1_params, plot=False)
self.inds[d]['c2'] = self.igs[d].entry_indicator(self.p.c2_ind, self.p.c2_params, plot=False)
self.inds[d]['volume'] = self.igs[d].volume_indicator(self.p.volume_ind, self.p.volume_params, plot=False)
self.inds[d]['exit'] = self.igs[d].exit_indicator(self.p.exit_ind, self.p.exit_params, plot=False)
if i > 0: # Check we are not on the first loop of data feed:
if self.p.oneplot == True:
d.plotinfo.plotmaster = self.datas[0]
# Set Commission and Determine which pairs need live exchange rates
self.manual_commission_pairs = []
for key in list(self.data_dict.keys()):
base = key[0:3]
counter = key[3:]
JPY = True if 'JPY' in [base, counter] else False
if counter == self.accn_currency:
# Set Commission for Pairs where Counter and Account Currency are the Same
comminfo = forexSpreadCommisionScheme(spread=2,method=0,JPY_pair=JPY,leverage=self.p.leverage)
self.broker.addcommissioninfo(comminfo,name=key)
elif base == self.accn_currency:
self.manual_commission_pairs.append(key)
def refresh_conditions(self):
self.trade_conditons = dict()
self.trade_conditons[1.0] = dict()
self.trade_conditons[-1.0] = dict()
for i, d in enumerate(self.datas):
self.trade_conditons[1.0][d] = {
'baseline': self.inds[d]['baseline'] > 0,
'c1': self.inds[d]['c1'] > 0,
'c2': self.inds[d]['c2'] > 0,
'volume': self.inds[d]['volume'] > 0
}
self.trade_conditons[-1.0][d] = {
'baseline': self.inds[d]['baseline'] < 0,
'c1': self.inds[d]['c1'] < 0,
'c2': self.inds[d]['c2'] < 0,
'volume': self.inds[d]['volume'] < 0
}
def decide_trade(self,d):
buy = all(list(self.trade_conditons[1.0][d].values()))
sell = all(list(self.trade_conditons[-1.0][d].values()))
if buy:
return 1.0
elif sell:
return -1.0
else:
return 0.0
def clean_orders(self):
# Clean Open Orders Dictionary
for i, d in enumerate(self.datas):
for o in self.open_orders[d]['sl']:
if not o.alive():
self.open_orders[d]['tracker'].remove(o.ref)
self.open_orders[d]['sl'].remove(o)
for o in self.open_orders[d]['tp']:
if not o.alive():
self.open_orders[d]['tp'].remove(o)
def check_positions(self):
self.position_dict = dict()
for i, d in enumerate(self.datas):
self.position_dict[d._name] = self.getposition(d).size
def notify_order(self, order):
# Call Custom Notification Function to Keep Code Clean
dt, dn = self.datetime.date(), order.data._name
res = notifier(order, dt, self.open_orders[order.data]['tracker'],verbose=self.p.verbose)
# Take Profit Hit, Initialize Trailing Stop Order
if res:
size = res[0]
price = res[1]
if size > 0:
move_stop = self.buy(data=order.data,
size=size,
exectype=bt.Order.StopTrail,
trailamount=self.params.sl*self.inds[order.data]['atr'][0])
elif size < 0:
move_stop = self.sell(size=size,
exectype=bt.Order.StopTrail,
trailamount=self.params.sl*self.inds[order.data]['atr'][0])
self.bar_executed = len(self)
self.order = None
def notify_trade(self, trade):
# Call Custom Notification Function to Keep Code Clean
date = self.data.datetime.datetime().date()
notifier(trade, date,[],verbose=self.p.verbose)
def size_position(self, d, stop_amount, risk):
pair = d._name
base = pair[0:3]
counter = pair[3:]
if counter == self.accn_currency:
method = 0
elif base == self.accn_currency:
method = 1
elif self.accn_currency not in [base, counter]:
method = 2
exchange_pair = self.accn_currency+counter
# Check if Exchange Pair Exists
if exchange_pair in list(self.data_dict.keys()):
exchange_rate = self.closes[self.data_dict[exchange_pair]][0]
# Check if Reverse of Exchange Pair Exists
elif counter+self.accn_currency in list(self.data_dict.keys()):
exchange_rate = self.closes[self.data_dict[counter+self.accn_currency]][0]
exchange_rate = 1/exchange_rate
price = self.closes[d][0]
stop = price - stop_amount
risk = float(risk) / 100.0
if base == 'JPY' or counter == 'JPY':
JPY_pair = True
else:
JPY_pair = False
if JPY_pair == True: # check if a YEN cross and change the multiplier
multiplier = 0.01
else:
multiplier = 0.0001
# Calc how much to risk
acc_value = self.broker.getvalue()
cash_risk = acc_value * risk
stop_pips_int = abs((price - stop) / multiplier)
pip_value = cash_risk / stop_pips_int
if method == 1:
pip_value = pip_value * price
units = pip_value / multiplier
return units
elif method == 2:
pip_value = pip_value * exchange_rate
units = pip_value / multiplier
return units
else: # is method 0
units = pip_value / multiplier
return units
def set_commission(self, d):
# Split Pair into Counter/Base Components
pair = d._name
base = pair[:3]
counter = pair[3:]
# Determine if JPY is in the Pair
JPY = True if 'JPY' in [base, counter] else False
# Case 1: USDXXX
if base == self.accn_currency:
currency_conversion = 1/self.closes[d][0]
comminfo = forexSpreadCommisionScheme(spread=2,
method=1,
JPY_pair=JPY,
mult=currency_conversion,
leverage=self.p.leverage)
# Case 2: XXXYYY
else:
# Look for USDYYY
forward = self.accn_currency+counter
# Look for YYYUSD
reverse = counter+self.accn_currency
if forward in list(self.data_dict.keys()):
exchange_rate=self.closes[self.data_dict[forward]][0]
elif reverse in list(self.data_dict.keys()):
exchange_rate=1.0/self.cloases[self.data_dict[reverse]][0]
currency_conversion = 1.0/exchange_rate
comminfo = forexSpreadCommisionScheme(spread=2,
method=2,
JPY_pair=JPY,
exchange_rate=exchange_rate,
mult=currency_conversion,
leverage=self.p.leverage)
self.broker.addcommissioninfo(comminfo, name=d._name)
def pullback(self,d):
# Baseline Crossing Logic (Did we go too far? Need to Look for a Pullback?)
if self.inds[d]['baseline'] != 0.0 and self.inds[d]['too_far'][0]:
# Too far from baseline - no trading should occur
if self.p.verbose: print('CAUTION BASELINE CROSS TOO FAR - CHECKING FOR PULLBACK')
self.trade_conditons[self.inds[d]['baseline'][0]][d]['baseline'] = False
elif self.inds[d]['baseline'][-1] != 0.0 and self.inds[d]['too_far'][-1] and not self.inds[d]['too_far'][0]:
# Previous Candle was a Baseline Cross more than 1xATR
# We are currently not "too far" from the baseline
if self.p.verbose: print('PULLBACK DETECTED - BASELINE SAYS GO')
self.trade_conditons[1.0][d]['baseline'] = self.inds[d]['baseline'][-1]>0
self.trade_conditons[-1.0][d]['baseline'] = self.inds[d]['baseline'][-1]<0
def continuation(self,d):
# This is Where We Check the Logic for Continuation Trades
# Get last fifty bars of baseline flip data
base_hist = list(self.inds[d]['baseline'].get(size=30))
base_hist.reverse()
# Iterate through base flip history and get index of last flip:
idx = next((i for i, x in enumerate(base_hist) if x != 0.0), None)
if not idx:
return None
base = base_hist[idx]
# Get Confirmation indicator Data since last baseline flip:
c1_hist = list(self.inds[d]['c1'].get(size=idx))
c1_hist.reverse()
# Check if a confirmation flip has been detected
if -1.0 * base in c1_hist:
flips = len(list(itertools.groupby(c1_hist, lambda c1_hist: c1_hist > 0))) - 1
# Check if we have a double confirmation flip
if flips == 2:
# Continuation Opportunity Detected!
# Double Check Confirmation Indicator 2
if self.inds[d]['c2'][0] == c1_hist[0]:
if self.p.verbose: print('CONTINUATION TRADE DETECTED')
# Update the Corresponding Buy/Sell Conditions
for key in self.trade_conditons[self.inds[d]['c1'][0]][d]:
self.trade_conditons[self.inds[d]['c1'][0]][d][key] = True
def bridge_too_far(self,d):
# Determine how long ago last flip in C1 Occurred
c1_hist = list(self.inds[d]['c1'].get(size=10))
c1_hist.reverse()
try:
idx = c1_hist.index(-1.0 * self.inds[d]['baseline']) - 1
except:
idx = 100
trade = self.decide_trade(d)
if idx > 7 and trade != 0.0:
if self.p.verbose: print('BRIDGE TOO FAR, DO NOT TRADE')
self.trade_conditons[trade][d]['c1'] = False
def next(self):
# Make Sure we have the most recent trading conditions:
self.refresh_conditions()
self.clean_orders()
self.check_positions()
# Echo Current Closing Price
if self.p.verbose: self.log('Current Closing Price: %.5f' % self.dataclose[0])
# Check if We have a Pending Order and Exit Function to Avoid Creating Duplicate
for i, d in enumerate(self.datas):
if len(self.open_orders[d]['sl']) > 0 or len(self.open_orders[d]['tp']) > 0:
continue
# Apply Strategy Trading Logic
pos = self.getposition(d).size
# Check if we are in the Market - Only Proceed if we are not
if all(v == 0 for v in list(self.position_dict.values())):
# Check Pullback & Bridge too Far Logic
if self.inds[d]['baseline'] != 0.0:
self.pullback(d)
self.bridge_too_far(d)
# Continuation Trade Logic
elif self.inds[d]['baseline'] == 0.0:
self.continuation(d)
# Check Entry Conditions & Place Trades
if self.decide_trade(d) > 0.0:
# Enter Long Position
# Calculate Risk Profile
if d._name in self.manual_commission_pairs:
#Set Commission for Cross Pair
self.set_commission(d)
size = round(self.size_position(d, self.p.sl * self.inds[d]['atr'], self.p.risk)/2)
price = self.closes[d]
tp = price + self.p.tp * self.inds[d]['atr']
sl = price - self.p.sl * self.inds[d]['atr']
# Generate Bracket Orders
# Bracket 1: To Scale Out at Break Even
main = self.buy(data=d,
size=2*size,
exectype=bt.Order.Market,
transmit=False)
stop_loss = self.sell(data=d,
size=2*size,
exectype=bt.Order.Stop,
price=sl,
parent=main,
transmit=False)
take_profit = self.sell(data=d,
size=size,
exectype=bt.Order.Limit,
price=tp,
parent=main,
transmit=True)
self.open_orders[d]['sl'].append(stop_loss)
self.open_orders[d]['tp'].append(take_profit)
self.open_orders[d]['tracker'].append(stop_loss.ref)
elif self.decide_trade(d) < 0.0:
# Enter Long Position
# Calculate Risk Profile
if d._name in self.manual_commission_pairs:
#Set Commission for Cross Pair
self.set_commission(d)
size = round(self.size_position(d, self.p.sl * self.inds[d]['atr'], self.p.risk)/2)
price = self.closes[d]
tp = price - self.p.tp * self.inds[d]['atr']
sl = price + self.p.sl * self.inds[d]['atr']
# Generate Bracket Orders
# Bracket 1: To Scale Out at Break Even
main = self.sell(data=d,
size=2*size,
exectype=bt.Order.Market,
transmit=False)
stop_loss = self.buy(data=d,
size=2*size,
exectype=bt.Order.Stop,
price=sl,
parent=main,
transmit=False)
take_profit = self.buy(data=d,
size=size,
exectype=bt.Order.Limit,
price=tp,
parent=main,
transmit=True)
self.open_orders[d]['sl'].append(stop_loss)
self.open_orders[d]['tp'].append(take_profit)
self.open_orders[d]['tracker'].append(stop_loss.ref)
else:
buyexit_conds = [self.inds[d]['exit'] > 0, self.inds[d]['baseline'] > 0, self.inds[d]['c1'] > 0, self.inds[d]['c2'] > 0]
sellexit_conds = [self.inds[d]['exit'] < 0, self.inds[d]['baseline'] < 0, self.inds[d]['c1'] < 0, self.inds[d]['c2'] < 0]
if pos > 0:
# Check for Sell Exit Signal
if any(sellexit_conds):
self.order = self.close(data=d)
elif pos < 0:
# Check for Buy Exit Signal
if any(buyexit_conds):
self.order = self.close(data=d)
if __name__ == '__main__':
# Create a cerebro entity
cerebro = bt.Cerebro()
# Enable Slippage on Open Price (Approximately %0.01 percent)
cerebro.broker = bt.brokers.BackBroker(slip_perc=0.0001,slip_open=True)
# Add our strategy
cerebro.addstrategy(NNFX)
# Get Data Files from Data Folder
paths, names = file_browser()
dpath = 'Data/'
datasets = [
(dpath+path,name) for path, name in zip(paths,names)
]
trade_pairs = ['EURUSD']
datasets = [i for i in datasets if i[1] in trade_pairs]
# Create a Data Feeds and Add to Cerebro
for i in range(len(datasets)):
data = bt.feeds.GenericCSVData(dataname=datasets[i][0],
openinterest=-1,
dtformat='%d.%m.%Y %H:%M:%S.000')
cerebro.adddata(data, name=datasets[i][1])
# Set our desired cash start
cerebro.broker.setcash(1000.0)
# Add Analyzers
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name="ta")
cerebro.addanalyzer(bt.analyzers.SQN, _name="sqn")
# Print out the starting conditions
print('Starting Portfolio Value: %.2f' % cerebro.broker.getvalue())
# Run over everything
t0 = time.time()
strategies = cerebro.run()
t1 = time.time()
firstStrat = strategies[0]
# Print out the final result
print('Final Portfolio Value: %.2f' % cerebro.broker.getvalue())
print('Backtest Time: %.2fs' % (t1-t0))
# print the analyzers
printTradeAnalysis(firstStrat.analyzers.ta.get_analysis())
printSQN(firstStrat.analyzers.sqn.get_analysis())
cerebro.plot(style='candle')