TradeBot/tests/test_execution.py

from __future__ import annotations

from types import SimpleNamespace

from crypto_spot_bot.bybit import Instrument
from crypto_spot_bot.bot import CryptoSpotBot
from crypto_spot_bot.execution import PaperBroker
from crypto_spot_bot.models import Signal, Ticker
from crypto_spot_bot.storage import Storage
from crypto_spot_bot.strategy import SpotStrategy


def test_paper_broker_buy_and_sell_records_trade(make_settings, tmp_path) -> None:
    settings = make_settings(tmp_path)
    storage = Storage(settings.database_path)
    broker = PaperBroker(settings, storage)
    ticker = Ticker("BTCUSDT", 100, 99.9, 100.1, 10_000_000, 100, 0)
    instrument = Instrument("BTCUSDT", "BTC", "USDT", "Trading", 0.01, 0.000001, 0.000001, 5)
    signal = Signal("BTCUSDT", "BUY", 0.8, "test")

    position = broker.buy(signal, ticker, instrument, {"BTCUSDT": 100})
    assert position is not None
    assert broker.cash < settings.starting_balance_usdt
    assert len(broker.open_positions()) == 1

    trade = broker.sell(position, ticker, "test exit")
    assert trade.side == "SELL"
    assert len(broker.open_positions()) == 0
    assert storage.recent_trades(limit=10)


def test_paper_broker_limits_fast_entries_per_minute(make_settings, tmp_path) -> None:
    settings = make_settings(
        tmp_path,
        max_entries_per_minute=1,
        max_open_positions=3,
        max_positions_per_symbol=3,
        max_total_exposure_usdt=90,
    )
    storage = Storage(settings.database_path)
    broker = PaperBroker(settings, storage)
    ticker = Ticker("BTCUSDT", 100, 99.9, 100.1, 10_000_000, 100, 0)
    instrument = Instrument("BTCUSDT", "BTC", "USDT", "Trading", 0.01, 0.000001, 0.000001, 5)

    first = broker.buy(Signal("BTCUSDT", "BUY", 0.8, "first"), ticker, instrument, {"BTCUSDT": 100})
    second = broker.buy(Signal("BTCUSDT", "BUY", 0.8, "second"), ticker, instrument, {"BTCUSDT": 100})

    assert first is not None
    assert second is None
    assert len(broker.open_positions()) == 1
    assert "лимит новых входов" in storage.recent_events(limit=1)[0]["message"]


def test_paper_broker_uses_signal_notional_and_pair_exposure(make_settings, tmp_path) -> None:
    settings = make_settings(
        tmp_path,
        min_position_usdt=1,
        max_position_usdt=20,
        max_symbol_exposure_usdt=6,
        max_total_exposure_usdt=50,
        max_open_positions=20,
        max_positions_per_symbol=1,
        max_entries_per_minute=0,
    )
    storage = Storage(settings.database_path)
    broker = PaperBroker(settings, storage)
    ticker = Ticker("BTCUSDT", 100, 99.9, 100.1, 10_000_000, 100, 0)
    instrument = Instrument("BTCUSDT", "BTC", "USDT", "Trading", 0.01, 0.000001, 0.000001, 1)

    first = broker.buy(
        Signal("BTCUSDT", "BUY", 0.8, "first", {"position_notional_usdt": 2}),
        ticker,
        instrument,
        {"BTCUSDT": 100},
    )
    second = broker.buy(
        Signal("BTCUSDT", "BUY", 0.8, "second", {"position_notional_usdt": 2}),
        ticker,
        instrument,
        {"BTCUSDT": 100},
    )
    third = broker.buy(
        Signal("BTCUSDT", "BUY", 0.8, "third", {"position_notional_usdt": 2}),
        ticker,
        instrument,
        {"BTCUSDT": 100},
    )
    fourth = broker.buy(
        Signal("BTCUSDT", "BUY", 0.8, "fourth", {"position_notional_usdt": 2}),
        ticker,
        instrument,
        {"BTCUSDT": 100},
    )

    assert first is not None
    assert second is not None
    assert third is not None
    assert fourth is None
    assert len(broker.open_positions()) == 3
    assert 5.5 <= broker.symbol_exposure("BTCUSDT") <= 6.0


def test_paper_broker_respects_adaptive_exposure_target(make_settings, tmp_path) -> None:
    settings = make_settings(
        tmp_path,
        min_position_usdt=1,
        max_position_usdt=20,
        max_symbol_exposure_usdt=20,
        max_total_exposure_usdt=80,
        max_open_positions=20,
        max_positions_per_symbol=20,
        max_entries_per_minute=0,
    )
    storage = Storage(settings.database_path)
    broker = PaperBroker(settings, storage)
    ticker = Ticker("BTCUSDT", 100, 99.9, 100.1, 10_000_000, 100, 0)
    instrument = Instrument("BTCUSDT", "BTC", "USDT", "Trading", 0.01, 0.000001, 0.000001, 1)
    capped_signal = Signal(
        "BTCUSDT",
        "BUY",
        0.8,
        "adaptive cap",
        {
            "position_notional_usdt": 10,
            "adaptive_rules": {
                "target_total_exposure_usdt": 0,
                "target_symbol_exposure_usdt": 0,
            },
        },
    )

    position = broker.buy(capped_signal, ticker, instrument, {"BTCUSDT": 100})

    assert position is None
    assert broker.open_positions() == []


def test_trend_macd_broker_blocks_dca_for_same_symbol(make_settings, tmp_path) -> None:
    settings = make_settings(
        tmp_path,
        strategy_mode="trend_macd",
        min_position_usdt=1,
        max_position_usdt=20,
        max_symbol_exposure_usdt=20,
        max_total_exposure_usdt=80,
        max_open_positions=3,
        max_positions_per_symbol=20,
        max_entries_per_minute=0,
    )
    storage = Storage(settings.database_path)
    broker = PaperBroker(settings, storage)
    ticker = Ticker("BTCUSDT", 100, 99.9, 100.1, 10_000_000, 100, 0)
    instrument = Instrument("BTCUSDT", "BTC", "USDT", "Trading", 0.01, 0.000001, 0.000001, 1)

    first = broker.buy(Signal("BTCUSDT", "BUY", 0.8, "first", {"position_notional_usdt": 2}), ticker, instrument, {"BTCUSDT": 100})
    second = broker.buy(Signal("BTCUSDT", "BUY", 0.8, "second", {"position_notional_usdt": 2}), ticker, instrument, {"BTCUSDT": 100})

    assert first is not None
    assert second is None
    assert len(broker.open_positions()) == 1


def test_torch_forecast_broker_blocks_dca_for_same_symbol(make_settings, tmp_path) -> None:
    settings = make_settings(
        tmp_path,
        strategy_mode="torch_forecast",
        min_position_usdt=1,
        max_position_usdt=20,
        max_symbol_exposure_usdt=20,
        max_total_exposure_usdt=80,
        max_open_positions=3,
        max_positions_per_symbol=20,
        max_entries_per_minute=0,
    )
    storage = Storage(settings.database_path)
    broker = PaperBroker(settings, storage)
    ticker = Ticker("BTCUSDT", 100, 99.9, 100.1, 10_000_000, 100, 0)
    instrument = Instrument("BTCUSDT", "BTC", "USDT", "Trading", 0.01, 0.000001, 0.000001, 1)

    first = broker.buy(Signal("BTCUSDT", "BUY", 0.8, "first", {"position_notional_usdt": 2}), ticker, instrument, {"BTCUSDT": 100})
    second = broker.buy(Signal("BTCUSDT", "BUY", 0.8, "second", {"position_notional_usdt": 2}), ticker, instrument, {"BTCUSDT": 100})

    assert first is not None
    assert second is None
    assert len(broker.open_positions()) == 1


def test_trend_macd_closes_old_paper_positions_outside_symbol_universe(make_settings, tmp_path) -> None:
    settings = make_settings(
        tmp_path,
        strategy_mode="trend_macd",
        trading_mode="paper",
        symbols=("BTCUSDT", "ETHUSDT", "SOLUSDT"),
        min_position_usdt=1,
        max_position_usdt=20,
        max_symbol_exposure_usdt=20,
        max_total_exposure_usdt=80,
        max_open_positions=3,
        max_entries_per_minute=0,
    )
    storage = Storage(settings.database_path)
    broker = PaperBroker(settings, storage)
    ticker = Ticker("HYPEUSDT", 10, 9.99, 10.01, 10_000_000, 100, 0)
    instrument = Instrument("HYPEUSDT", "HYPE", "USDT", "Trading", 0.001, 0.001, 0.001, 1)
    position = broker.buy(
        Signal("HYPEUSDT", "BUY", 0.8, "old", {"position_notional_usdt": 5}),
        ticker,
        instrument,
        {"HYPEUSDT": 10},
    )
    assert position is not None

    bot = CryptoSpotBot(
        settings,
        storage,
        SimpleNamespace(symbols=["BTCUSDT", "ETHUSDT", "SOLUSDT"]),
        broker,
        SpotStrategy(settings),
        object(),
        object(),
    )
    bot._close_paper_positions_outside_symbol_universe()

    assert broker.open_positions() == []
    assert storage.open_positions() == []
    trade = storage.recent_trades(limit=1)[0]
    assert trade["side"] == "SELL"
    assert trade["symbol"] == "HYPEUSDT"
    assert "trend_macd" in trade["reason"]