import json from decimal import Decimal from pathlib import Path import pytest from pydantic import SecretStr import binance_quant.auto_buy as auto_buy from binance_quant.auto_buy import run_live_auto_buy_once from binance_quant.config import Settings from binance_quant.models import AccountSnapshot, Balance, SymbolMetadata, TradingMode from binance_quant.storage import Storage def read_audit_events(tmp_path: Path) -> list[dict]: events = [] for block in (tmp_path / "logs" / "audit.log").read_text(encoding="utf-8").split("=" * 80): if "\nJSON:\n" not in block: continue raw_json = block.split("\nJSON:\n", 1)[1].splitlines()[0] events.append(json.loads(raw_json)) return events class FakeExchange: def __init__(self): self.bought = False self.calls = [] def exchange_info(self, symbol): self.calls.append(("exchange_info", symbol)) base = symbol.removesuffix("USDT") return SymbolMetadata( symbol, base, "USDT", Decimal("5"), Decimal("0.00001"), Decimal("0.00001"), Decimal("0.01"), Decimal("0.01"), ) def account(self): self.calls.append(("account",)) return AccountSnapshot( { "USDT": Balance("USDT", Decimal("50")), "BTC": Balance("BTC", Decimal("0")), } ) def symbol_order_book_ticker(self, symbol): self.calls.append(("symbol_order_book_ticker", symbol)) return {"symbol": symbol, "bidPrice": "104.99", "askPrice": "105"} def ticker_24hr(self, symbol): self.calls.append(("ticker_24hr", symbol)) return {"symbol": symbol, "priceChangePercent": "1.94"} def klines(self, symbol, interval="1h", limit=6): self.calls.append(("klines", symbol, interval, limit)) return [ [1, "100", "101", "99", "100", "10", 2], [2, "100", "102", "100", "101", "10", 3], [3, "101", "103", "101", "102", "10", 4], [4, "102", "104", "102", "103", "10", 5], [5, "103", "105", "103", "104", "10", 6], [6, "104", "106", "104", "105", "12", 7], ] def open_orders(self, symbol): self.calls.append(("open_orders", symbol)) return [] def create_market_order(self, symbol, side, quote_amount, client_order_id): self.calls.append(("create_market_order", symbol, side, quote_amount, client_order_id)) self.bought = True return { "symbol": symbol, "side": "BUY", "type": "MARKET", "status": "FILLED", "clientOrderId": client_order_id, "orderId": 12345, "orderListId": -1, "executedQty": "0.04761", "cummulativeQuoteQty": str(quote_amount), "transactTime": 1_700_000_000_000, "fills": [ { "tradeId": 67890, "price": "105", "qty": "0.04761", "commission": "0.00004761", "commissionAsset": "BTC", } ], } def create_oco_sell( self, symbol, quantity, take_profit_price, stop_price, stop_limit_price, list_client_order_id, take_profit_client_order_id, stop_client_order_id, ): self.calls.append( ( "create_oco_sell", symbol, quantity, take_profit_price, stop_price, stop_limit_price, list_client_order_id, take_profit_client_order_id, stop_client_order_id, ) ) return { "symbol": symbol, "contingencyType": "OCO", "listStatusType": "EXEC_STARTED", "listOrderStatus": "EXECUTING", "listClientOrderId": list_client_order_id, } def create_market_sell_quantity(self, symbol, quantity, client_order_id): self.calls.append(("create_market_sell_quantity", symbol, quantity, client_order_id)) return { "symbol": symbol, "side": "SELL", "type": "MARKET", "status": "FILLED", "clientOrderId": client_order_id, "executedQty": str(quantity), } class FalsyExchange(FakeExchange): def __bool__(self): return False class NoExchangeCalls: bought = False def __getattr__(self, name): raise AssertionError(f"exchange method {name} should not be called") def settings(tmp_path: Path, enabled=True, mode=TradingMode.LIVE, daily_limit=5, **overrides) -> Settings: return Settings( _env_file=None, mode=mode, api_key="key", api_secret=SecretStr("secret"), database_path=tmp_path / "audit.sqlite3", live_auto_buyer_enabled=enabled, auto_buy_symbols=("BTCUSDT",), auto_buy_min_score=Decimal("4"), auto_buy_daily_limit=daily_limit, **overrides, ) def test_run_live_auto_buy_once_places_protected_buy(tmp_path): fake = FakeExchange() result = run_live_auto_buy_once(settings(tmp_path), exchange_client=fake) assert result.status == "protected" assert result.symbol == "BTCUSDT" assert fake.bought audit = Storage(tmp_path / "audit.sqlite3") [run] = audit.list_auto_buy_runs() assert run["id"] == result.run_id assert run["status"] == "protected" assert [event["run_id"] for event in audit.list_auto_buy_events(result.run_id)] == [result.run_id, result.run_id] def test_run_live_auto_buy_once_records_filled_trade_for_cost_basis(tmp_path): result = run_live_auto_buy_once(settings(tmp_path), exchange_client=FakeExchange()) audit = Storage(tmp_path / "audit.sqlite3") [fill] = audit.list_spot_trade_fills("BTCUSDT") assert result.status == "protected" assert fill["trade_id"] == 67890 assert fill["side"] == "BUY" assert fill["price"] == "105" assert fill["quantity"] == "0.04761" assert fill["quote_quantity"] == "5" assert fill["commission_asset"] == "BTC" def test_run_live_auto_buy_once_writes_scan_risk_and_execution_audit_events(tmp_path): fake = FakeExchange() result = run_live_auto_buy_once(settings(tmp_path), exchange_client=fake) assert result.status == "protected" events_by_name = {event["event"]: event for event in read_audit_events(tmp_path)} scan_event = events_by_name["auto_buy.scan"] assert scan_event["params"]["symbols"] == ["BTCUSDT"] assert scan_event["formula"]["spread_bps"] == "(ask - bid) / ((ask + bid) / 2) * 10000" assert scan_event["formula"]["score"] == "momentum flags + volume confirmation + positive 4h momentum percent" assert scan_event["result"]["candidate"]["symbol"] == "BTCUSDT" assert scan_event["result"]["candidate"]["score"] == "7.9604" risk_event = events_by_name["auto_buy.risk"] assert risk_event["params"]["symbol"] == "BTCUSDT" assert risk_event["result"]["approved"] is True assert risk_event["result"]["reason"] == "approved" assert risk_event["result"]["quote_amount"] == "5" execution_event = events_by_name["auto_buy.execution"] assert execution_event["params"] == {"symbol": "BTCUSDT", "run_id": result.run_id} assert execution_event["result"]["status"] == "protected" assert execution_event["result"]["reason"] == "protective oco accepted" def test_run_live_auto_buy_once_records_disabled_rejection(tmp_path): fake = FakeExchange() result = run_live_auto_buy_once(settings(tmp_path, enabled=False), exchange_client=fake) assert result.status == "rejected" assert result.reason == "live auto buyer disabled" assert not fake.bought [run] = Storage(tmp_path / "audit.sqlite3").list_auto_buy_runs() assert run["id"] == result.run_id assert run["status"] == "rejected" assert run["reason"] == "live auto buyer disabled" @pytest.mark.parametrize( ("mode", "enabled", "reason"), [ (TradingMode.LIVE, False, "live auto buyer disabled"), (TradingMode.DRY_RUN, True, "live mode required"), ], ) def test_run_live_auto_buy_once_safety_gates_short_circuit_before_exchange_calls(tmp_path, mode, enabled, reason): result = run_live_auto_buy_once(settings(tmp_path, enabled=enabled, mode=mode), exchange_client=NoExchangeCalls()) assert result.status == "rejected" assert result.reason == reason def test_run_live_auto_buy_once_disabled_short_circuits_before_constructing_exchange(tmp_path, monkeypatch): def fail_construct(*args, **kwargs): raise AssertionError("exchange client should not be constructed") monkeypatch.setattr(auto_buy, "BinanceSpotClient", fail_construct) result = run_live_auto_buy_once(settings(tmp_path, enabled=False)) assert result.status == "rejected" assert result.reason == "live auto buyer disabled" def test_run_live_auto_buy_once_passes_proxy_to_exchange_client(tmp_path, monkeypatch): captured = {} class ConstructedExchange(FakeExchange): def __init__(self, **kwargs): super().__init__() captured.update(kwargs) monkeypatch.setattr(auto_buy, "BinanceSpotClient", ConstructedExchange) result = run_live_auto_buy_once(settings(tmp_path, proxy_url="http://192.168.1.3:28090")) assert result.status == "protected" assert captured["proxy_url"] == "http://192.168.1.3:28090" assert captured["base_urls"] == ( "https://api.binance.com/api", "https://api1.binance.com/api", "https://api2.binance.com/api", "https://api3.binance.com/api", "https://api4.binance.com/api", ) def test_run_live_auto_buy_once_uses_falsy_injected_exchange(tmp_path, monkeypatch): def fail_construct(*args, **kwargs): raise AssertionError("exchange client should not be constructed") monkeypatch.setattr(auto_buy, "BinanceSpotClient", fail_construct) fake = FalsyExchange() result = run_live_auto_buy_once(settings(tmp_path), exchange_client=fake) assert result.status == "protected" assert fake.bought def test_run_live_auto_buy_once_records_pre_buy_failure_without_order(tmp_path): class BadTickerExchange(FakeExchange): def symbol_order_book_ticker(self, symbol): self.calls.append(("symbol_order_book_ticker", symbol)) return {"symbol": symbol, "bidPrice": "104.99"} fake = BadTickerExchange() result = run_live_auto_buy_once(settings(tmp_path), exchange_client=fake) assert result.status == "rejected" assert result.symbol == "" assert result.reason == "pre-buy data error" assert not fake.bought assert not any(call[0] == "create_market_order" for call in fake.calls) [run] = Storage(tmp_path / "audit.sqlite3").list_auto_buy_runs() assert run["id"] == result.run_id assert run["status"] == "rejected" assert run["reason"] == "pre-buy data error" def test_run_live_auto_buy_once_rejects_candidate_after_sharp_24h_rally(tmp_path): class RallyExchange(FakeExchange): def ticker_24hr(self, symbol): self.calls.append(("ticker_24hr", symbol)) return {"symbol": symbol, "priceChangePercent": "10.00"} fake = RallyExchange() result = run_live_auto_buy_once(settings(tmp_path), exchange_client=fake) assert result.status == "rejected" assert result.symbol == "BTCUSDT" assert result.reason == "24h rally above max" assert not fake.bought assert not any(call[0] == "create_market_order" for call in fake.calls) [run] = Storage(tmp_path / "audit.sqlite3").list_auto_buy_runs() assert run["id"] == result.run_id assert run["status"] == "rejected" assert run["reason"] == "24h rally above max" [risk_event] = [event for event in read_audit_events(tmp_path) if event["event"] == "auto_buy.risk"] assert risk_event["result"]["price_change_percent_24h"] == "10.00" assert risk_event["result"]["max_24h_rally_percent"] == "10" def test_run_live_auto_buy_once_rechecks_daily_limit_with_reservation(tmp_path): audit = Storage(tmp_path / "audit.sqlite3") audit.initialize() audit.record_auto_buy_run("ETHUSDT", "reserved", Decimal("4"), "pending", Decimal("5")) fake = FakeExchange() result = run_live_auto_buy_once( settings(tmp_path, daily_limit=1), exchange_client=fake, now_ms=1_700_000_000_000, ) assert result.status == "rejected" assert result.reason == "daily auto-buy limit reached" assert not fake.bought assert not any(call[0] == "create_market_order" for call in fake.calls)