-
Notifications
You must be signed in to change notification settings - Fork 2
/
testLiveTrader.py
126 lines (102 loc) · 3.26 KB
/
testLiveTrader.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
import csv
import argparse
from liveStrategies.firstStrategy import TestStrategy
###Arg parsing
def parse_args():
global crypto
global timeframe
global file_name
global output_file_name
crypto = "ltc"
timeframe = "1m"
file_name = "data/ltc1m.csv"
output_file_name = "output_of_test_live_trader.txt"
parser = argparse.ArgumentParser(description='Live trading bot')
parser.add_argument('-c', '--crypto', help='Set crypto currency', required=False)
parser.add_argument('-tf', '--timeframe', help='Set time frame', required=False)
parser.add_argument('-f', '--file', help='File name', required=False)
parser.add_argument('-o', '--output', help='Output txt file', required=False)
args = vars(parser.parse_args())
if args['crypto']:
crypto = args['crypto']
if args['timeframe']:
timeframe = args['timeframe']
if args['file']:
file_name = args['file']
if args['output']:
output_file_name = args['output']
###Initializing data
def initialize_data():
global candlesticks
candlesticks = []
with open(file_name, 'r') as f:
reader = csv.reader(f)
for row in reader:
# print(row)
candlesticks.append({
"E": float(row[0]),
"k": {
"o": row[1],
"h": row[2],
"l": row[3],
"c": row[4],
"v": row[5],
"T": row[6],
"t": row[0]
}
})
###Run test
def run_test():
idx = 0
while True:
input_val = input("Enter command: ")
input_val = input_val.lower()
##Plot
if (input_val == "plot"):
strategy.plot()
continue
##Stop
if (input_val == "s" or input_val == "stop"):
print('Done')
break
##Print to txt
if (input_val == "print"):
strategy.log_to_txt(output_file_name)
continue
##If all data has been processed already
if(idx >= len(candlesticks)):
print('All data was processed')
continue
##Go through all the data
if(input_val == "all"):
for i in range(len(candlesticks)):
message = candlesticks[idx]
strategy.add_tick(message)
idx += 1
if (idx >= len(candlesticks)):
print('All data was processed')
break
continue
##Number of messages to process
if (input_val.isnumeric()):
for i in range(int(input_val)):
message = candlesticks[idx]
strategy.add_tick(message)
idx += 1
if (idx >= len(candlesticks)):
print('All data was processed')
break
continue
##For any other input, just process one message
message = candlesticks[idx]
strategy.add_tick(message)
idx +=1
def main():
global strategy
strategy = TestStrategy(timeframe=timeframe, crypto=crypto)
##Initialize data from specified file
initialize_data()
run_test()
if __name__=="__main__":
parse_args()
main()