# pragma pylint: disable=missing-docstring import logging import re from datetime import datetime from unittest.mock import MagicMock import pytest from jsonschema import validate from telegram import Bot, Update, Message, Chat from freqtrade.main import init, create_trade from freqtrade.misc import update_state, State, get_state, CONF_SCHEMA from freqtrade.persistence import Trade from freqtrade.rpc.telegram import ( _status, _status_table, _profit, _forcesell, _performance, _count, _start, _stop, _balance ) @pytest.fixture def conf(): configuration = { "max_open_trades": 3, "stake_currency": "BTC", "stake_amount": 0.05, "dry_run": True, "minimal_roi": { "2880": 0.005, "720": 0.01, "0": 0.02 }, "bid_strategy": { "ask_last_balance": 0.0 }, "exchange": { "name": "bittrex", "enabled": True, "key": "key", "secret": "secret", "pair_whitelist": [ "BTC_ETH", "BTC_ETC" ] }, "telegram": { "enabled": True, "token": "token", "chat_id": "0" }, "initial_state": "running" } validate(configuration, CONF_SCHEMA) return configuration @pytest.fixture def update(): _update = Update(0) _update.message = Message(0, 0, datetime.utcnow(), Chat(0, 0)) return _update class MagicBot(MagicMock, Bot): pass def test_status_handle(conf, update, mocker): mocker.patch.dict('freqtrade.main._CONF', conf) mocker.patch('freqtrade.main.get_buy_signal', side_effect=lambda _: True) msg_mock = MagicMock() mocker.patch.multiple('freqtrade.main.telegram', _CONF=conf, init=MagicMock(), send_msg=msg_mock) mocker.patch.multiple('freqtrade.main.exchange', validate_pairs=MagicMock(), get_ticker=MagicMock(return_value={ 'bid': 0.07256061, 'ask': 0.072661, 'last': 0.07256061 }), buy=MagicMock(return_value='mocked_order_id')) init(conf, 'sqlite://') # Create some test data trade = create_trade(15.0) assert trade Trade.session.add(trade) Trade.session.flush() # Trigger status while we don't know the open_rate yet _status(bot=MagicBot(), update=update) # Simulate fulfilled LIMIT_BUY order for trade trade.update({ 'id': 'mocked_limit_buy', 'type': 'LIMIT_BUY', 'pair': 'mocked', 'opened': datetime.utcnow(), 'rate': 0.07256060, 'amount': 206.43811673387373, 'remaining': 0.0, 'closed': datetime.utcnow(), }) Trade.session.flush() # Trigger status while we have a fulfilled order for the open trade _status(bot=MagicBot(), update=update) assert msg_mock.call_count == 3 assert '[BTC_ETH]' in msg_mock.call_args_list[-1][0][0] def test_status_table_handle(conf, update, mocker): mocker.patch.dict('freqtrade.main._CONF', conf) mocker.patch('freqtrade.main.get_buy_signal', side_effect=lambda _: True) msg_mock = MagicMock() mocker.patch.multiple('freqtrade.main.telegram', _CONF=conf, init=MagicMock(), send_msg=msg_mock) mocker.patch.multiple('freqtrade.main.exchange', validate_pairs=MagicMock(), get_ticker=MagicMock(return_value={ 'bid': 0.07256061, 'ask': 0.072661, 'last': 0.07256061 }), buy=MagicMock(return_value='mocked_order_id')) init(conf, 'sqlite://') # Create some test data trade = create_trade(15.0) assert trade Trade.session.add(trade) Trade.session.flush() _status_table(bot=MagicBot(), update=update) text = re.sub('<\/?pre>', '', msg_mock.call_args_list[-1][0][0]) line = text.split("\n") fields = re.sub('[ ]+', ' ', line[2].strip()).split(' ') assert int(fields[0]) == 1 assert fields[1] == 'BTC_ETH' assert msg_mock.call_count == 2 def test_profit_handle(conf, update, mocker): mocker.patch.dict('freqtrade.main._CONF', conf) mocker.patch('freqtrade.main.get_buy_signal', side_effect=lambda _: True) msg_mock = MagicMock() mocker.patch.multiple('freqtrade.main.telegram', _CONF=conf, init=MagicMock(), send_msg=msg_mock) mocker.patch.multiple('freqtrade.main.exchange', validate_pairs=MagicMock(), get_ticker=MagicMock(return_value={ 'bid': 0.07256061, 'ask': 0.072661, 'last': 0.07256061 }), buy=MagicMock(return_value='mocked_limit_buy')) init(conf, 'sqlite://') # Create some test data trade = create_trade(15.0) assert trade # Simulate fulfilled LIMIT_BUY order for trade trade.update({ 'id': 'mocked_limit_buy', 'type': 'LIMIT_BUY', 'pair': 'mocked', 'opened': datetime.utcnow(), 'rate': 0.07256061, 'amount': 206.43811673387373, 'remaining': 0.0, 'closed': datetime.utcnow(), }) # Simulate fulfilled LIMIT_SELL order for trade trade.update({ 'id': 'mocked_limit_sell', 'type': 'LIMIT_SELL', 'pair': 'mocked', 'opened': datetime.utcnow(), 'rate': 0.0802134, 'amount': 206.43811673387373, 'remaining': 0.0, 'closed': datetime.utcnow(), }) trade.close_date = datetime.utcnow() trade.open_order_id = None trade.is_open = False Trade.session.add(trade) Trade.session.flush() _profit(bot=MagicBot(), update=update) assert msg_mock.call_count == 2 assert '*ROI:* `1.507013 (10.05%)`' in msg_mock.call_args_list[-1][0][0] assert 'Best Performing:* `BTC_ETH: 10.05%`' in msg_mock.call_args_list[-1][0][0] def test_forcesell_handle(conf, update, mocker): mocker.patch.dict('freqtrade.main._CONF', conf) mocker.patch('freqtrade.main.get_buy_signal', side_effect=lambda _: True) msg_mock = MagicMock() mocker.patch.multiple('freqtrade.main.telegram', _CONF=conf, init=MagicMock(), send_msg=msg_mock) mocker.patch.multiple('freqtrade.main.exchange', validate_pairs=MagicMock(), get_ticker=MagicMock(return_value={ 'bid': 0.07256061, 'ask': 0.072661, 'last': 0.07256061 }), buy=MagicMock(return_value='mocked_order_id')) init(conf, 'sqlite://') # Create some test data trade = create_trade(15.0) assert trade # Simulate fulfilled LIMIT_BUY order for trade trade.update({ 'id': 'mocked_limit_buy', 'type': 'LIMIT_BUY', 'pair': 'mocked', 'opened': datetime.utcnow(), 'rate': 0.07256060, 'amount': 206.43811673387373, 'remaining': 0.0, 'closed': datetime.utcnow(), }) Trade.session.add(trade) Trade.session.flush() update.message.text = '/forcesell 1' _forcesell(bot=MagicBot(), update=update) assert msg_mock.call_count == 2 assert 'Selling [BTC/ETH]' in msg_mock.call_args_list[-1][0][0] assert '0.072561 (profit: ~-0.5%)' in msg_mock.call_args_list[-1][0][0] def test_performance_handle(conf, update, mocker): mocker.patch.dict('freqtrade.main._CONF', conf) mocker.patch('freqtrade.main.get_buy_signal', side_effect=lambda _: True) msg_mock = MagicMock() mocker.patch.multiple('freqtrade.main.telegram', _CONF=conf, init=MagicMock(), send_msg=msg_mock) mocker.patch.multiple('freqtrade.main.exchange', validate_pairs=MagicMock(), get_ticker=MagicMock(return_value={ 'bid': 0.07256061, 'ask': 0.072661, 'last': 0.07256061 }), buy=MagicMock(return_value='mocked_order_id')) init(conf, 'sqlite://') # Create some test data trade = create_trade(15.0) assert trade # Simulate fulfilled LIMIT_BUY order for trade trade.update({ 'id': 'mocked_limit_buy', 'type': 'LIMIT_BUY', 'pair': 'mocked', 'opened': datetime.utcnow(), 'rate': 0.07256061, 'amount': 206.43811673387373, 'remaining': 0.0, 'closed': datetime.utcnow(), }) # Simulate fulfilled LIMIT_SELL order for trade trade.update({ 'id': 'mocked_limit_sell', 'type': 'LIMIT_SELL', 'pair': 'mocked', 'opened': datetime.utcnow(), 'rate': 0.0802134, 'amount': 206.43811673387373, 'remaining': 0.0, 'closed': datetime.utcnow(), }) trade.close_date = datetime.utcnow() trade.is_open = False Trade.session.add(trade) Trade.session.flush() _performance(bot=MagicBot(), update=update) assert msg_mock.call_count == 2 assert 'Performance' in msg_mock.call_args_list[-1][0][0] assert 'BTC_ETH\t10.05%' in msg_mock.call_args_list[-1][0][0] def test_count_handle(conf, update, mocker): mocker.patch.dict('freqtrade.main._CONF', conf) mocker.patch('freqtrade.main.get_buy_signal', side_effect=lambda _: True) msg_mock = MagicMock() mocker.patch.multiple('freqtrade.main.telegram', _CONF=conf, init=MagicMock(), send_msg=msg_mock) mocker.patch.multiple('freqtrade.main.exchange', validate_pairs=MagicMock(), get_ticker=MagicMock(return_value={ 'bid': 0.07256061, 'ask': 0.072661, 'last': 0.07256061 }), buy=MagicMock(return_value='mocked_order_id')) init(conf, 'sqlite://') # Create some test data trade = create_trade(15.0) trade2 = create_trade(15.0) assert trade assert trade2 Trade.session.add(trade) Trade.session.add(trade2) Trade.session.flush() _count(bot=MagicBot(), update=update) line = msg_mock.call_args_list[-1][0][0].split("\n") assert line[2] == '{}/{}'.format(2, conf['max_open_trades']) def test_start_handle(conf, update, mocker): mocker.patch.dict('freqtrade.main._CONF', conf) msg_mock = MagicMock() mocker.patch.multiple('freqtrade.main.telegram', _CONF=conf, init=MagicMock(), send_msg=msg_mock) mocker.patch.multiple('freqtrade.main.exchange', _CONF=conf, init=MagicMock()) init(conf, 'sqlite://') update_state(State.STOPPED) assert get_state() == State.STOPPED _start(bot=MagicBot(), update=update) assert get_state() == State.RUNNING assert msg_mock.call_count == 0 def test_stop_handle(conf, update, mocker): mocker.patch.dict('freqtrade.main._CONF', conf) msg_mock = MagicMock() mocker.patch.multiple('freqtrade.main.telegram', _CONF=conf, init=MagicMock(), send_msg=msg_mock) mocker.patch.multiple('freqtrade.main.exchange', _CONF=conf, init=MagicMock()) init(conf, 'sqlite://') update_state(State.RUNNING) assert get_state() == State.RUNNING _stop(bot=MagicBot(), update=update) assert get_state() == State.STOPPED assert msg_mock.call_count == 1 assert 'Stopping trader' in msg_mock.call_args_list[0][0][0] def test_balance_handle(conf, update, mocker): mock_balance = [{ 'Currency': 'BTC', 'Balance': 10.0, 'Available': 12.0, 'Pending': 0.0, 'CryptoAddress': 'XXXX'}] mocker.patch.dict('freqtrade.main._CONF', conf) msg_mock = MagicMock() mocker.patch.multiple('freqtrade.main.telegram', _CONF=conf, init=MagicMock(), send_msg=msg_mock) mocker.patch.multiple('freqtrade.main.exchange', get_balances=MagicMock(return_value=mock_balance)) _balance(bot=MagicBot(), update=update) assert msg_mock.call_count == 1 assert '*Currency*: BTC' in msg_mock.call_args_list[0][0][0] assert 'Balance' in msg_mock.call_args_list[0][0][0]