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) summary = storage.closed_trade_summary() assert summary["trades"] == 1 assert summary["net_pnl"] == round(trade.net_pnl, 6) 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, strategy_mode="torch_forecast", 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=6, 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_raises_small_signal_to_exchange_min_notional(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("XRPUSDT", 1.0, 0.999, 1.001, 10_000_000, 100, 0) instrument = Instrument("XRPUSDT", "XRP", "USDT", "Trading", 0.0001, 0.01, 0.01, 5) position = broker.buy( Signal("XRPUSDT", "BUY", 0.8, "small rebound", {"position_notional_usdt": 1.5}), ticker, instrument, {"XRPUSDT": 1.0}, ) assert position is not None assert position.notional_usdt >= instrument.min_notional_value assert position.notional_usdt <= settings.max_position_usdt def test_paper_broker_rounds_small_order_up_to_exchange_qty_step(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("HYPEUSDT", 39.6, 39.59, 39.61, 10_000_000, 100, 0) instrument = Instrument("HYPEUSDT", "HYPE", "USDT", "Trading", 0.001, 0.01, 0.01, 1) position = broker.buy( Signal("HYPEUSDT", "BUY", 0.8, "small torch edge", {"position_notional_usdt": 1.05}), ticker, instrument, {"HYPEUSDT": 39.6}, ) assert position is not None assert position.qty == 0.03 assert position.notional_usdt >= instrument.min_notional_value assert position.notional_usdt <= settings.max_position_usdt 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_allows_dynamic_entries_until_total_limit(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}) 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 def test_torch_forecast_broker_respects_configured_symbol_position_limit(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=20, max_positions_per_symbol=2, 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}) assert first is not None assert second is not None assert third is None assert len(broker.open_positions()) == 2 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"]