gavindiaz/aiomql
1
0
Fork 0
mirror of https://github.com/Ichinga-Samuel/aiomql.git synced 2026-08-22 00:08:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

v4

Browse Source
This commit is contained in:
Ichinga Samuel
2024-11-04 00:32:51 +01:00
parent 2c46e18528
commit d65a8db1a2
85 changed files with 5071 additions and 2512 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/live/integration/test_bot.py
+8 -6
View File
@@ -6,17 +6,19 @@ from aiomql.contrib.symbols import ForexSymbol
async def test_bot():
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
syms = ['BTCUSD', 'SOLUSD', 'ETHUSD']
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
syms = ["BTCUSD", "SOLUSD", "ETHUSD"]
symbols = [ForexSymbol(name=sym) for sym in syms]
stgs = [Chaos(symbol=symbol, name='test_chaos') for symbol in symbols]
strategies = [Chaos(symbol=symbol, name="test_chaos") for symbol in symbols]
bot = Bot()
assert bot.config.shutdown is False
bot.executor.timeout = 30
bot.add_strategies(strategies=stgs)
bot.executor.timeout = 10
bot.add_strategies(strategies=strategies)
await bot.initialize()
await bot.executor.execute()
assert bot.executor.no_of_running_strategies == 3
assert len(bot.executor.coroutines) == 1
assert len(bot.executor.coroutine_threads) == 1
assert bot.config.shutdown is True
tests/live/integration/test_results_records.py
+49 -21
View File
@@ -15,35 +15,59 @@ class TestRecordsAndResults:
def setup_class(cls):
cls.trade_records = TradeRecords()
@pytest.fixture(scope='class')
@pytest.fixture(scope="class")
async def buy(self, mt):
sym = 'BTCUSD'
sym = "BTCUSD"
sym_info = await mt.symbol_info(sym)
dsl = (sym_info.trade_stops_level + sym_info.spread) * 2 * sym_info.point
sl = sym_info.ask - dsl
tp = sym_info.ask + dsl
return {'action': mt.TRADE_ACTION_DEAL, 'symbol': sym, 'volume': sym_info.volume_min,
'type': mt.ORDER_TYPE_BUY, 'price': sym_info.ask, 'sl': sl, 'tp': tp}
return {
"action": mt.TRADE_ACTION_DEAL,
"symbol": sym,
"volume": sym_info.volume_min,
"type": mt.ORDER_TYPE_BUY,
"price": sym_info.ask,
"sl": sl,
"tp": tp,
}
@pytest.fixture(scope='class')
@pytest.fixture(scope="class")
async def sell(self, mt):
sym = 'BTCUSD'
sym = "BTCUSD"
sym_info = await mt.symbol_info(sym)
return {'action': mt.TRADE_ACTION_DEAL, 'symbol': sym, 'volume': sym_info.volume_min,
'type': mt.ORDER_TYPE_SELL, 'price': sym_info.bid}
return {
"action": mt.TRADE_ACTION_DEAL,
"symbol": sym,
"volume": sym_info.volume_min,
"type": mt.ORDER_TYPE_SELL,
"price": sym_info.bid,
}
@pytest.fixture(scope='class', autouse=True)
@pytest.fixture(scope="class", autouse=True)
async def setup(self, sell, buy, mt):
buy_res = await mt.order_send(buy)
buy_res_2 = await mt.order_send(buy)
sell_res = await mt.order_send(sell)
sell_res_2 = await mt.order_send(sell)
buy_res = Result(result=OrderSendResult(**buy_res._asdict()), name='test_result')
sell_res = Result(result=OrderSendResult(**sell_res._asdict()), name='test_result')
sell_res_2 = Result(result=OrderSendResult(**sell_res_2._asdict()), name='test_result')
buy_res_2 = Result(result=OrderSendResult(**buy_res_2._asdict()), name='test_result')
await asyncio.gather(buy_res.save(), sell_res.save(), buy_res_2.save(trade_record_mode='json'),
sell_res_2.save(trade_record_mode='json'))
buy_res = Result(
result=OrderSendResult(**buy_res._asdict()), name="test_result"
)
sell_res = Result(
result=OrderSendResult(**sell_res._asdict()), name="test_result"
)
sell_res_2 = Result(
result=OrderSendResult(**sell_res_2._asdict()), name="test_result"
)
buy_res_2 = Result(
result=OrderSendResult(**buy_res_2._asdict()), name="test_result"
)
await asyncio.gather(
buy_res.save(),
sell_res.save(),
buy_res_2.save(trade_record_mode="json"),
sell_res_2.save(trade_record_mode="json"),
)
await Positions().close_all()
def test_records_dir(self):
@@ -66,29 +90,33 @@ class TestRecordsAndResults:
async def test_json_records(self):
json_records = self.trade_records.get_json_records()
matched_recs = [record for record in json_records if record.match('test_result.json')]
matched_recs = [
record for record in json_records if record.match("test_result.json")
]
assert len(matched_recs) == 1
record = matched_recs[0]
record_data = json.load(record.open())
assert isinstance(record_data, list)
assert len(record_data) == 2
is_open = [data['closed'] is False for data in record_data]
is_open = [data["closed"] is False for data in record_data]
assert all(is_open)
await self.trade_records.update_json_records()
is_close = [data['closed'] is True for data in record_data]
is_close = [data["closed"] is True for data in record_data]
assert len(is_close) == 2
async def test_csv_records(self):
csv_records = self.trade_records.get_csv_records()
matched_recs = [record for record in csv_records if record.match('test_result.csv')]
matched_recs = [
record for record in csv_records if record.match("test_result.csv")
]
assert len(matched_recs) == 1
record = matched_recs[0]
record_data = DictReader(record.open())
record_data = [row for row in record_data]
assert isinstance(record_data, list)
assert len(record_data) == 2
is_open = [data['closed'].title() == 'False' for data in record_data]
is_open = [data["closed"].title() == "False" for data in record_data]
assert all(is_open)
await self.trade_records.update_json_records()
is_close = [data['closed'].title() == 'True' for data in record_data]
is_close = [data["closed"].title() == "True" for data in record_data]
assert len(is_close) == 2