-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_worker.py
75 lines (63 loc) · 2.66 KB
/
test_worker.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
from pyalgotrade.optimizer import worker
from pyalgotrade import strategy
from pyalgotrade.technical import ma
from pyalgotrade.technical import rsi
import sys
class MyStrategy(strategy.BacktestingStrategy):
def __init__(self, feed, entrySMA, exitSMA, rsiPeriod, overBoughtThreshold, overSoldThreshold):
strategy.BacktestingStrategy.__init__(self, feed, 5000)
ds = feed[instrument].getCloseDataSeries()
self.__entrySMA = ma.SMA(ds, entrySMA)
self.__exitSMA = ma.SMA(ds, exitSMA)
self.__rsi = rsi.RSI(ds, rsiPeriod)
self.__overBoughtThreshold = overBoughtThreshold
self.__overSoldThreshold = overSoldThreshold
self.__longPos = None
self.__shortPos = None
def onEnterOk(self, position):
pass
def onEnterCanceled(self, position):
if self.__longPos == position:
self.__longPos = None
elif self.__shortPos == position:
self.__shortPos = None
else:
assert(False)
def onExitOk(self, position):
if self.__longPos == position:
self.__longPos = None
elif self.__shortPos == position:
self.__shortPos = None
else:
assert(False)
def onExitCanceled(self, position):
# If the exit was canceled, re-submit it.
position.exit()
def onBars(self, bars):
# Wait for enough bars to be available to calculate SMA and RSI.
if self.__exitSMA[-1] is None or self.__entrySMA[-1] is None or self.__rsi[-1] is None:
return
bar = bars[instrument]
if self.__longPos != None:
if self.exitLongSignal(bar):
self.__longPos.exit()
elif self.__shortPos != None:
if self.exitShortSignal(bar):
self.__shortPos.exit()
else:
if self.enterLongSignal(bar):
self.__longPos = self.enterLong(instrument, 10, True)
elif self.enterShortSignal(bar):
self.__shortPos = self.enterShort(instrument, 10, True)
def enterLongSignal(self, bar):
return bar.getClose() > self.__entrySMA[-1] and self.__rsi[-1] <= self.__overSoldThreshold
def exitLongSignal(self, bar):
return bar.getClose() > self.__exitSMA[-1]
def enterShortSignal(self, bar):
return bar.getClose() < self.__entrySMA[-1] and self.__rsi[-1] >= self.__overBoughtThreshold
def exitShortSignal(self, bar):
return bar.getClose() < self.__exitSMA[-1]
# The if __name__ == '__main__' part is necessary if running on Windows.
instrument = sys.argv[1]
if __name__ == '__main__':
worker.run(MyStrategy, "localhost", 5000, workerName="localworker")