-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest.py
155 lines (133 loc) · 6 KB
/
test.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
import MetaTrader5 as mt5
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime
import talib
import pandas as pd
from fractal_indicator import *
from get_signals import *
from time import time, sleep
# connect to mt5
if not mt5.initialize():
print("Unable to connect to MT5")
mt5.shutdown()
# you code here
#
# Connect to account
account = 36363563
authorized = mt5.login(account)
if authorized:
print(f"Connected to account {account}")
else:
print(f"Failed to connect to account {account}")
# symbol --argparse ???
symbol = "GBPUSD" # "EURCHF" "EURUSD"
nb_bars = 100 # 1_000
lots = 0.01
timeframe = mt5.TIMEFRAME_M5
# Wait until next bar before launching
while time() % 300 > 1:
sleep(1)
# Set launching and start app
running = True
while running:
# Get dataset of previous trades recorded in order_dataset.csv
full_dataset = pd.read_csv("order_dataset.csv")
if len(full_dataset) == 0:
closing = False
else:
closing = True
if time() % 300 <= 10:
pd_current = get_datas(symbol, timeframe, nb_bars)
# Check to open positon
# Get signals
signalFrac, signalCCI = signals(pd_current)
# Open BUY
if signalCCI == -1 and signalFrac:
result, ticket = send_order(lot=lots, price=mt5.symbol_info_tick(symbol).ask, symbol=symbol, type=mt5.ORDER_TYPE_BUY)
if result:
# calculate sl and tp
R1, S1 = pivot(pd_current["high"].iloc[-1], pd_current["low"].iloc[-1], pd_current["close"].iloc[-1])
# Put type,symbole,ticket,stoploss,takeprofit,lots to csv
order_dataset = pd.DataFrame([["BUY", symbol, ticket, S1, R1, lots]], columns=["type", "symbole", "ticket", "stoploss", "takeprofit", "lots"])
# Refresh inital dataset
full_dataset = pd.read_csv("order_dataset.csv")
full_dataset = full_dataset.append(order_dataset, ignore_index=True)
# Copy to csv
full_dataset.to_csv("order_dataset.csv", index=False)
closing = True
# Open SELL
if signalCCI == 1 and signalFrac:
result, ticket = send_order(lot=lots, price=mt5.symbol_info_tick(symbol).bid, symbol=symbol, type=mt5.ORDER_TYPE_SELL)
if result:
# calculate sl and tp
R1, S1 = pivot(pd_current["high"].iloc[-1], pd_current["low"].iloc[-1], pd_current["close"].iloc[-1])
# Put type,symbole,ticket,stoploss,takeprofit,lots to csv
order_dataset = pd.DataFrame([["SELL", symbol, ticket, R1, S1, lots]], columns=["type", "symbole", "ticket", "stoploss", "takeprofit", "lots"])
# Refresh inital dataset
full_dataset = pd.read_csv("order_dataset.csv")
full_dataset = full_dataset.append(order_dataset, ignore_index=True)
# Copy to csv
full_dataset.to_csv("order_dataset.csv", index=False)
closing = True
# Check for closing position
if closing:
# Get positions from csv
full_dataset = pd.read_csv("order_dataset.csv")
current_datas = mt5.copy_rates_from_pos(symbol, timeframe, 0, 2)
# Iter positions
for i in range(len(full_dataset)):
# If BUY or SELL
# If current close price above or under TP recalculate TP
# Elif current close under or above SL, close position and delete line from csv
if full_dataset["type"].iloc[i] == "BUY":
if full_dataset["takeprofit"].iloc[i] <= current_datas[0][4]:
full_dataset["takeprofit"].iloc[i], full_dataset["stoploss"].iloc[i] = pivot(current_datas[1][2],
current_datas[1][3], current_datas[1][4])
full_dataset.to_csv("order_dataset.csv", index=False)
elif full_dataset["stoploss"].iloc[i] >= current_datas[0][4] and current_datas[0][6] <= 20:
close_position(lots, mt5.symbol_info_tick(symbol).bid, symbol, mt5.ORDER_TYPE_SELL, full_dataset["ticket"].iloc[i])
full_dataset = full_dataset.drop([i])
full_dataset.to_csv("order_dataset.csv", index=False)
elif full_dataset["type"].iloc[i] == "SELL":
if full_dataset["takeprofit"].iloc[i] >= current_datas[0][4]:
full_dataset["stoploss"].iloc[i], full_dataset["takeprofit"].iloc[i] = pivot(current_datas[1][2],
current_datas[1][3], current_datas[1][4])
full_dataset.to_csv("order_dataset.csv", index=False)
elif full_dataset["stoploss"].iloc[i] <= current_datas[0][4] and current_datas[0][6] <= 20:
close_position(lots, mt5.symbol_info_tick(symbol).bid, symbol, mt5.ORDER_TYPE_SELL, full_dataset["ticket"].iloc[i])
full_dataset = full_dataset.drop([i])
full_dataset.to_csv("order_dataset.csv", index=False)
sleep(10)
else:
# Check for closing position
# Test for closing position
if closing:
position_id=result.order
price=mt5.symbol_info_tick(symbol).bid
deviation=20
request={
"action": mt5.TRADE_ACTION_DEAL,
"symbol": symbol,
"volume": lot,
"type": mt5.ORDER_TYPE_SELL,
"position": position_id,
"price": price,
"deviation": deviation,
"magic": 234000,
"comment": "python script close",
"type_time": mt5.ORDER_TIME_GTC,
"type_filling": mt5.ORDER_FILLING_RETURN,
}
# send a trading request
result=mt5.order_send(request)
# check the execution result
print("3. close position #{}: sell {} {} lots at {} with deviation={} points".format(position_id,symbol,lot,price,deviation));
if result.retcode != mt5.TRADE_RETCODE_DONE:
print("4. order_send failed, retcode={}".format(result.retcode))
print(" result",result)
else:
print("4. position #{} closed, {}".format(position_id,result))
running = False
sleep(10)
mt5.shutdown()