mirror of
https://github.com/Ichinga-Samuel/aiomql.git
synced 2026-07-28 12:47:43 +00:00
Ichinga Samuel
150 lines
6.3 KiB
Python
150 lines
6.3 KiB
Python
from aiomql.contrib import ForexSymbol
|
|
from aiomql.core import MetaBackTester, BackTestEngine, GetData
|
|
from aiomql.lib import Order
|
|
|
|
|
|
async def make_buy_sell_orders():
|
|
sym = ForexSymbol(name="BTCUSD")
|
|
sym_info = await sym.mt5.symbol_info(sym.name)
|
|
dsl = (sym_info.trade_stops_level + sym_info.spread) * 2 * sym_info.point
|
|
sl = sym_info.ask - dsl
|
|
tp = sym_info.ask + dsl
|
|
buy_req = {
|
|
"action": sym.mt5.TRADE_ACTION_DEAL,
|
|
"symbol": sym.name,
|
|
"volume": sym_info.volume_min,
|
|
"type": sym.mt5.ORDER_TYPE_BUY,
|
|
"price": sym_info.ask,
|
|
"sl": sl,
|
|
"tp": tp,
|
|
}
|
|
|
|
sell_req = buy_req.copy()
|
|
sell_req["type"] = sym.mt5.ORDER_TYPE_SELL
|
|
sell_req["price"] = sym_info.bid
|
|
del sell_req["tp"]
|
|
del sell_req["sl"]
|
|
return {"buy": Order(**buy_req), "sell": Order(**sell_req)}
|
|
|
|
|
|
def test_trade_mode(config, backtest_engine, history, positions, order_sell, order_buy, btc_usd):
|
|
assert config.mode == "backtest"
|
|
assert isinstance(backtest_engine, BackTestEngine)
|
|
assert isinstance(history.mt5, MetaBackTester)
|
|
assert isinstance(positions.mt5, MetaBackTester)
|
|
assert isinstance(order_sell.mt5, MetaBackTester)
|
|
assert isinstance(order_buy.mt5, MetaBackTester)
|
|
assert isinstance(btc_usd.mt5, MetaBackTester)
|
|
|
|
|
|
async def test_order_send(backtest_engine, order_sell, order_buy):
|
|
await backtest_engine.setup_account(balance=100)
|
|
so = await backtest_engine.order_send(request=order_sell.request)
|
|
bo = await backtest_engine.order_send(request=order_buy.request)
|
|
assert so.retcode == 10009
|
|
assert bo.retcode == 10009
|
|
backtest_engine.reset(clear_data=True)
|
|
|
|
|
|
async def test_positions(backtest_engine, positions, order_sell, order_buy):
|
|
await backtest_engine.setup_account(balance=100)
|
|
so = await backtest_engine.order_send(request=order_sell.request)
|
|
bo = await backtest_engine.order_send(request=order_buy.request)
|
|
all_positions = await positions.get_positions()
|
|
assert len(all_positions) == 2
|
|
await positions.close_position_by_ticket(ticket=so.order)
|
|
all_positions = await positions.get_positions()
|
|
assert len(all_positions) == 1
|
|
await positions.close_position_by_ticket(ticket=bo.order)
|
|
all_positions = await positions.get_positions()
|
|
assert len(all_positions) == 0
|
|
backtest_engine.reset(clear_data=True)
|
|
|
|
|
|
async def test_history(backtest_engine, history, order_sell, order_buy, positions):
|
|
await backtest_engine.setup_account(balance=100)
|
|
so = await backtest_engine.order_send(request=order_sell.request)
|
|
await backtest_engine.order_send(request=order_buy.request)
|
|
await history.initialize()
|
|
assert len(history.orders) == 2
|
|
assert len(history.deals) == 2
|
|
await positions.close_position_by_ticket(ticket=so.order)
|
|
deals = await history.get_deals()
|
|
assert len(deals) == 3
|
|
backtest_engine.reset(clear_data=True)
|
|
|
|
|
|
async def test_margin(backtest_engine, order_sell, order_buy):
|
|
await backtest_engine.setup_account(balance=100)
|
|
so_margin = await backtest_engine.order_calc_margin(
|
|
action=order_sell.action, volume=order_sell.volume, symbol=order_sell.symbol, price=order_sell.price
|
|
)
|
|
bo_margin = await backtest_engine.order_calc_margin(
|
|
action=order_buy.action, volume=order_buy.volume, symbol=order_buy.symbol, price=order_buy.price
|
|
)
|
|
total_margin = so_margin + bo_margin
|
|
await backtest_engine.order_send(request=order_sell.request)
|
|
await backtest_engine.order_send(request=order_buy.request)
|
|
# noinspection PyTestUnpassedFixture
|
|
assert backtest_engine.positions.margin == total_margin == backtest_engine._account.margin
|
|
backtest_engine.reset(clear_data=True)
|
|
|
|
|
|
async def test_account(backtest_engine, positions):
|
|
await backtest_engine.setup_account(balance=100)
|
|
backtest_engine.fast_forward(steps=100)
|
|
balance = backtest_engine._account.balance
|
|
equity = backtest_engine._account.equity
|
|
orders = await make_buy_sell_orders()
|
|
buy_order = orders["buy"]
|
|
sell_order = orders["sell"]
|
|
so = await backtest_engine.order_send(request=sell_order.request)
|
|
bo = await backtest_engine.order_send(request=buy_order.request)
|
|
backtest_engine.fast_forward(steps=22000)
|
|
all_pos = await positions.get_positions()
|
|
for _ in range(1000):
|
|
backtest_engine.fast_forward(steps=1)
|
|
await backtest_engine.tracker()
|
|
all_pos = await positions.get_positions()
|
|
if len(all_pos) == 1:
|
|
break
|
|
|
|
deal = backtest_engine.deals.history_deals_get(position=bo.order)
|
|
bo_profit = deal[-1].profit
|
|
assert len(all_pos) == 1
|
|
assert backtest_engine.positions.margin == backtest_engine._account.margin == backtest_engine.positions.margins[so.order]
|
|
profit = sum([pos.profit for pos in all_pos])
|
|
n_balance = backtest_engine._account.balance
|
|
n_equity = backtest_engine._account.equity
|
|
assert backtest_engine._account.profit == profit
|
|
assert n_balance == balance + bo_profit
|
|
assert n_equity == equity + bo_profit + profit
|
|
so_pos = await positions.get_position_by_ticket(ticket=so.order)
|
|
gain = so_pos.profit
|
|
await positions.close_position(position=so_pos)
|
|
assert backtest_engine._account.balance == n_balance + gain
|
|
backtest_engine.reset(clear_data=True)
|
|
|
|
|
|
async def test_wrapup(positions, buy_order, sell_order, backtest_engine, config):
|
|
await backtest_engine.setup_account(balance=100)
|
|
backtest_engine.fast_forward(steps=500)
|
|
bo = await backtest_engine.order_send(request=buy_order)
|
|
await backtest_engine.order_send(request=sell_order)
|
|
backtest_engine.fast_forward(steps=5000)
|
|
await backtest_engine.tracker()
|
|
await positions.close_position_by_ticket(ticket=bo.order)
|
|
await backtest_engine.wrap_up()
|
|
last_balance = backtest_engine._account.balance
|
|
last_equity = backtest_engine._account.equity
|
|
last_profit = backtest_engine._account.profit
|
|
tdata = GetData.load_data(name=config.backtest_dir / f"{backtest_engine.name}.pkl")
|
|
new_bte = BackTestEngine(data=tdata, restart=False, assign_to_config=False, preload=False)
|
|
assert new_bte._account.balance == last_balance
|
|
assert new_bte._account.equity == last_equity
|
|
assert new_bte._account.profit == last_profit
|
|
assert new_bte.span == backtest_engine.span
|
|
assert new_bte.range == backtest_engine.range
|
|
assert new_bte.name == backtest_engine.name
|
|
assert new_bte.cursor.time == backtest_engine.cursor.time
|