from decimal import Decimal import pytest from binance_quant.auto_buy import AutoBuyCandidate, AutoBuyRiskContext, AutoBuyRiskEngine, AutoBuyRiskSettings from binance_quant.models import AccountSnapshot, Balance, SymbolMetadata, TradingMode METADATA = SymbolMetadata( "BTCUSDT", "BTC", "USDT", Decimal("5"), Decimal("0.00001"), Decimal("0.00001"), Decimal("0.01"), Decimal("0.01"), ) CANDIDATE = AutoBuyCandidate("BTCUSDT", Decimal("4.5"), Decimal("50000"), Decimal("5"), ("test",)) class BrokenMetadata: symbol = "BTCUSDT" base_asset = "BTC" quote_asset = "USDT" min_notional = Decimal("5") min_qty = Decimal("0.00001") step_size = Decimal("0.00001") tick_size = Decimal("0.01") class MissingSymbolMetadata: base_asset = "BTC" quote_asset = "USDT" min_notional = Decimal("5") min_qty = Decimal("0.00001") step_size = Decimal("0.00001") tick_size = Decimal("0.01") def account(usdt: str = "20", btc: str = "0", usdt_locked: str = "0", btc_locked: str = "0") -> AccountSnapshot: return AccountSnapshot( { "USDT": Balance("USDT", Decimal(usdt), Decimal(usdt_locked)), "BTC": Balance("BTC", Decimal(btc), Decimal(btc_locked)), } ) def context(**overrides) -> AutoBuyRiskContext: values = { "mode": TradingMode.LIVE, "enabled": True, "candidate": CANDIDATE, "metadata": METADATA, "account": account(), "daily_attempts": 0, "has_recent_symbol_attempt": False, "quote_amount": Decimal("5"), "price_change_percent_24h": Decimal("0"), } values.update(overrides) return AutoBuyRiskContext(**values) def assert_rejected(decision, reason: str) -> None: assert decision.approved is False assert decision.reason == reason def test_auto_buy_risk_settings_validate_limits(): with pytest.raises(ValueError, match="daily_limit must be positive"): AutoBuyRiskSettings(daily_limit=0) with pytest.raises(ValueError, match="daily_limit must be finite"): AutoBuyRiskSettings(daily_limit=Decimal("NaN")) with pytest.raises(ValueError, match="daily_limit must be finite"): AutoBuyRiskSettings(daily_limit=Decimal("Infinity")) with pytest.raises(ValueError, match="daily_limit must be an integer"): AutoBuyRiskSettings(daily_limit=Decimal("0.5")) with pytest.raises(ValueError, match="max_spread_bps must be positive"): AutoBuyRiskSettings(daily_limit=5, max_spread_bps=Decimal("0")) with pytest.raises(ValueError, match="quote_reserve must be non-negative"): AutoBuyRiskSettings(daily_limit=5, quote_reserve=Decimal("-0.01")) with pytest.raises(ValueError, match="max_spread_bps must be finite"): AutoBuyRiskSettings(daily_limit=5, max_spread_bps=Decimal("NaN")) with pytest.raises(ValueError, match="max_spread_bps must be a Decimal"): AutoBuyRiskSettings(daily_limit=5, max_spread_bps=float("inf")) with pytest.raises(ValueError, match="max_spread_bps must be a Decimal"): AutoBuyRiskSettings(daily_limit=5, max_spread_bps=None) with pytest.raises(ValueError, match="quote_reserve must be finite"): AutoBuyRiskSettings(daily_limit=5, quote_reserve=Decimal("Infinity")) with pytest.raises(ValueError, match="quote_reserve must be a Decimal"): AutoBuyRiskSettings(daily_limit=5, quote_reserve=float("nan")) with pytest.raises(ValueError, match="quote_reserve must be a Decimal"): AutoBuyRiskSettings(daily_limit=5, quote_reserve=1.0) with pytest.raises(ValueError, match="existing_base_value_limit must be positive"): AutoBuyRiskSettings(daily_limit=5, existing_base_value_limit=Decimal("0")) with pytest.raises(ValueError, match="existing_base_value_limit must be finite"): AutoBuyRiskSettings(daily_limit=5, existing_base_value_limit=Decimal("NaN")) with pytest.raises(ValueError, match="existing_base_value_limit must be a Decimal"): AutoBuyRiskSettings(daily_limit=5, existing_base_value_limit=20.0) with pytest.raises(ValueError, match="max_24h_rally_percent must be positive"): AutoBuyRiskSettings(daily_limit=5, max_24h_rally_percent=Decimal("0")) with pytest.raises(ValueError, match="max_24h_rally_percent must be finite"): AutoBuyRiskSettings(daily_limit=5, max_24h_rally_percent=Decimal("NaN")) with pytest.raises(ValueError, match="max_24h_rally_percent must be a Decimal"): AutoBuyRiskSettings(daily_limit=5, max_24h_rally_percent=10.0) def test_auto_buy_risk_approves_valid_candidate(): decision = AutoBuyRiskEngine(AutoBuyRiskSettings(daily_limit=5, quote_reserve=Decimal("0"))).evaluate(context()) assert decision.approved assert decision.reason == "approved" assert decision.candidate == CANDIDATE @pytest.mark.parametrize( ("btc", "btc_locked"), [ ("0.00009", "0"), ("0", "0.00009"), ("0.00004", "0.00005"), ], ) def test_auto_buy_risk_allows_existing_base_value_below_quote_amount(btc, btc_locked): engine = AutoBuyRiskEngine(AutoBuyRiskSettings(daily_limit=5, quote_reserve=Decimal("0"))) decision = engine.evaluate(context(account=account(btc=btc, btc_locked=btc_locked))) assert decision.approved assert decision.reason == "approved" def test_auto_buy_risk_uses_separate_existing_base_value_limit(): engine = AutoBuyRiskEngine( AutoBuyRiskSettings( daily_limit=5, quote_reserve=Decimal("0"), existing_base_value_limit=Decimal("20"), ) ) allowed = engine.evaluate(context(account=account(btc="0.0001"))) rejected = engine.evaluate(context(account=account(btc="0.0004"))) assert allowed.approved assert allowed.reason == "approved" assert_rejected(rejected, "existing base balance") def test_auto_buy_risk_requires_live_mode_and_enable_flag(): engine = AutoBuyRiskEngine(AutoBuyRiskSettings(daily_limit=5)) assert_rejected(engine.evaluate(context(mode=TradingMode.DRY_RUN)), "live mode required") assert_rejected(engine.evaluate(context(enabled=False)), "live auto buyer disabled") assert_rejected(engine.evaluate(context(enabled="false")), "invalid risk input") assert_rejected(engine.evaluate(context(enabled=1)), "invalid risk input") def test_auto_buy_risk_rejects_daily_cap_and_recent_symbol_attempt(): engine = AutoBuyRiskEngine(AutoBuyRiskSettings(daily_limit=5)) assert_rejected(engine.evaluate(context(daily_attempts=5)), "daily auto-buy limit reached") assert_rejected(engine.evaluate(context(has_recent_symbol_attempt=True)), "symbol cooldown active") assert_rejected(engine.evaluate(context(has_recent_symbol_attempt=0)), "invalid risk input") def test_auto_buy_risk_rejects_metadata_mismatch_and_missing_filters(): engine = AutoBuyRiskEngine(AutoBuyRiskSettings(daily_limit=5)) wrong_symbol = SymbolMetadata( "ETHUSDT", "ETH", "USDT", Decimal("5"), Decimal("0.00001"), Decimal("0.00001"), Decimal("0.01"), Decimal("0.01"), ) missing_step_size = SymbolMetadata( "BTCUSDT", "BTC", "USDT", Decimal("5"), Decimal("0.00001"), Decimal("0"), Decimal("0.01"), Decimal("0.01"), ) missing_tick_size = SymbolMetadata( "BTCUSDT", "BTC", "USDT", Decimal("5"), Decimal("0.00001"), Decimal("0.00001"), Decimal("0"), Decimal("0.01"), ) assert_rejected(engine.evaluate(context(metadata=wrong_symbol)), "metadata symbol mismatch") assert_rejected(engine.evaluate(context(metadata=missing_step_size)), "missing exchange filters") assert_rejected(engine.evaluate(context(metadata=missing_tick_size)), "missing exchange filters") def test_auto_buy_risk_rejects_structurally_malformed_metadata(): engine = AutoBuyRiskEngine(AutoBuyRiskSettings(daily_limit=5)) missing_symbol = MissingSymbolMetadata() bad_quote_asset = BrokenMetadata() bad_quote_asset.quote_asset = None assert_rejected(engine.evaluate(context(metadata=None)), "invalid risk input") assert_rejected(engine.evaluate(context(metadata=missing_symbol)), "invalid risk input") assert_rejected(engine.evaluate(context(metadata=bad_quote_asset)), "invalid risk input") def test_auto_buy_risk_rejects_exchange_and_balance_limits(): engine = AutoBuyRiskEngine(AutoBuyRiskSettings(daily_limit=5, quote_reserve=Decimal("1"))) high_min_notional = SymbolMetadata( "BTCUSDT", "BTC", "USDT", Decimal("10"), Decimal("0.00001"), Decimal("0.00001"), Decimal("0.01"), Decimal("0.01"), ) assert_rejected(engine.evaluate(context(metadata=high_min_notional)), "quote amount below min notional") assert_rejected(engine.evaluate(context(account=account(usdt="5"))), "insufficient quote balance") def test_auto_buy_risk_rejects_existing_exposure_and_wide_spread(): engine = AutoBuyRiskEngine(AutoBuyRiskSettings(daily_limit=5)) wide = AutoBuyCandidate("BTCUSDT", Decimal("4.5"), Decimal("50000"), Decimal("20"), ("test",)) assert_rejected(engine.evaluate(context(account=account(btc="0.001"))), "existing base balance") assert_rejected(engine.evaluate(context(account=account(btc_locked="0.001"))), "existing base balance") assert_rejected(engine.evaluate(context(candidate=wide)), "spread above max") def test_auto_buy_risk_rejects_24h_rally_at_configured_threshold(): engine = AutoBuyRiskEngine(AutoBuyRiskSettings(daily_limit=5, max_24h_rally_percent=Decimal("10"))) decision = engine.evaluate(context(price_change_percent_24h=Decimal("10.00"))) assert_rejected(decision, "24h rally above max") @pytest.mark.parametrize( ("overrides", "reason"), [ ({"candidate": AutoBuyCandidate("BTCUSDT", Decimal("4.5"), Decimal("50000"), Decimal("NaN"), ("test",))}, "invalid risk input"), ({"candidate": AutoBuyCandidate("BTCUSDT", Decimal("NaN"), Decimal("50000"), Decimal("5"), ("test",))}, "invalid risk input"), ({"candidate": AutoBuyCandidate("BTCUSDT", Decimal("4.5"), Decimal("Infinity"), Decimal("5"), ("test",))}, "invalid risk input"), ({"candidate": AutoBuyCandidate("BTCUSDT", Decimal("-1"), Decimal("50000"), Decimal("5"), ("test",))}, "invalid risk input"), ({"candidate": AutoBuyCandidate("BTCUSDT", Decimal("4.5"), Decimal("-1"), Decimal("5"), ("test",))}, "invalid risk input"), ({"candidate": AutoBuyCandidate("BTCUSDT", Decimal("4.5"), Decimal("50000"), Decimal("-1"), ("test",))}, "invalid risk input"), ({"quote_amount": Decimal("NaN")}, "invalid risk input"), ({"quote_amount": Decimal("0")}, "invalid risk input"), ({"quote_amount": Decimal("-1")}, "invalid risk input"), ({"quote_amount": 5.0}, "invalid risk input"), ({"price_change_percent_24h": Decimal("NaN")}, "invalid risk input"), ({"price_change_percent_24h": Decimal("Infinity")}, "invalid risk input"), ({"price_change_percent_24h": 1.0}, "invalid risk input"), ({"daily_attempts": Decimal("NaN")}, "invalid risk input"), ({"daily_attempts": Decimal("Infinity")}, "invalid risk input"), ({"daily_attempts": Decimal("0.5")}, "invalid risk input"), ({"account": account(usdt="NaN")}, "invalid risk input"), ({"account": account(btc="NaN")}, "invalid risk input"), ({"account": account(usdt="-1")}, "invalid risk input"), ({"account": account(usdt_locked="-1")}, "invalid risk input"), ({"account": account(btc="-0.001")}, "invalid risk input"), ({"account": account(btc_locked="-0.001")}, "invalid risk input"), ( { "account": AccountSnapshot( { "USDT": Balance("USDT", 20.0), "BTC": Balance("BTC", Decimal("0")), } ) }, "invalid risk input", ), ( { "account": AccountSnapshot( { "USDT": Balance("USDT", Decimal("20")), "BTC": Balance("BTC", Decimal("0"), 1.0), } ) }, "invalid risk input", ), ( { "metadata": SymbolMetadata( "BTCUSDT", "BTC", "USDT", Decimal("NaN"), Decimal("0.00001"), Decimal("0.00001"), Decimal("0.01"), Decimal("0.01"), ) }, "invalid risk input", ), ( { "metadata": SymbolMetadata( "BTCUSDT", "BTC", "USDT", Decimal("0"), Decimal("0.00001"), Decimal("0.00001"), Decimal("0.01"), Decimal("0.01"), ) }, "invalid risk input", ), ( { "metadata": SymbolMetadata( "BTCUSDT", "BTC", "USDT", Decimal("-1"), Decimal("0.00001"), Decimal("0.00001"), Decimal("0.01"), Decimal("0.01"), ) }, "invalid risk input", ), ( { "metadata": SymbolMetadata( "BTCUSDT", "BTC", "USDT", Decimal("5"), Decimal("NaN"), Decimal("0.00001"), Decimal("0.01"), Decimal("0.01"), ) }, "invalid risk input", ), ( { "metadata": SymbolMetadata( "BTCUSDT", "BTC", "USDT", Decimal("5"), Decimal("0.00001"), Decimal("Infinity"), Decimal("0.01"), Decimal("0.01"), ) }, "invalid risk input", ), ( { "metadata": SymbolMetadata( "BTCUSDT", "BTC", "USDT", Decimal("5"), Decimal("0.00001"), Decimal("0.00001"), Decimal("-Infinity"), Decimal("0.01"), ) }, "invalid risk input", ), ], ) def test_auto_buy_risk_rejects_non_finite_runtime_inputs(overrides, reason): decision = AutoBuyRiskEngine(AutoBuyRiskSettings(daily_limit=5)).evaluate(context(**overrides)) assert_rejected(decision, reason) def test_auto_buy_risk_rejection_order_keeps_spread_before_min_notional(): engine = AutoBuyRiskEngine(AutoBuyRiskSettings(daily_limit=5)) wide = AutoBuyCandidate("BTCUSDT", Decimal("4.5"), Decimal("50000"), Decimal("20"), ("test",)) high_min_notional = SymbolMetadata( "BTCUSDT", "BTC", "USDT", Decimal("10"), Decimal("0.00001"), Decimal("0.00001"), Decimal("0.01"), Decimal("0.01"), ) decision = engine.evaluate(context(candidate=wide, metadata=high_min_notional)) assert_rejected(decision, "spread above max")