import json import subprocess from decimal import Decimal from pathlib import Path from binance_quant import futures_ai_trader from binance_quant.config import Settings from binance_quant.futures_ai_trader import ( FuturesAiBatchResult, FuturesAiPrediction, _prediction_schema, run_live_futures_ai_trade_batch_once, ) from binance_quant.futures_models import FuturesAction from binance_quant.storage import Storage def test_futures_prediction_list_parses_valid_payload(): predictions = FuturesAiPrediction.from_json_list( json.dumps( { "predictions": [ { "symbol": "btcusdt", "action": "OPEN_LONG", "confidence": "0.82", "leverage": 5, "stop_loss_price": "59000", "take_profit_price": "63000", "reason": "BTC trend", }, { "symbol": "ethusdt", "action": "CLOSE", "confidence": "0.77", "close_percent": 50, "reason": "ETH risk", }, ] } ), allowed_symbols=("BTCUSDT", "ETHUSDT"), ) assert [item.symbol for item in predictions] == ["BTCUSDT", "ETHUSDT"] assert predictions[0].action is FuturesAction.OPEN_LONG assert predictions[0].leverage == 5 assert predictions[0].stop_loss_price == Decimal("59000") assert predictions[0].take_profit_price == Decimal("63000") assert predictions[1].action is FuturesAction.CLOSE assert predictions[1].close_percent == 50 def test_futures_prediction_rejects_invalid_close_percent_to_hold(): [prediction] = FuturesAiPrediction.from_json_list( json.dumps( { "predictions": [ { "symbol": "BTCUSDT", "action": "CLOSE", "confidence": "0.88", "close_percent": 75, "reason": "invalid percent", } ] } ), allowed_symbols=("BTCUSDT",), ) assert prediction.action is FuturesAction.HOLD assert prediction.reason == "模型返回的该交易对预测无效" def test_futures_prediction_rejects_fractional_close_percent_to_hold(): [prediction] = FuturesAiPrediction.from_json_list( json.dumps( { "predictions": [ { "symbol": "BTCUSDT", "action": "CLOSE", "confidence": "0.88", "close_percent": 25.9, "reason": "fractional percent", } ] } ), allowed_symbols=("BTCUSDT",), ) assert prediction.action is FuturesAction.HOLD assert prediction.reason == "模型返回的该交易对预测无效" def test_futures_prediction_rejects_non_string_reason_to_hold(): [prediction] = FuturesAiPrediction.from_json_list( json.dumps( { "predictions": [ { "symbol": "BTCUSDT", "action": "OPEN_LONG", "confidence": "0.88", "stop_loss_price": "59000", "take_profit_price": "63000", "reason": None, } ] } ), allowed_symbols=("BTCUSDT",), ) assert prediction.action is FuturesAction.HOLD assert prediction.reason == "模型返回的该交易对预测无效" def test_futures_prediction_rejects_non_finite_protection_price_to_hold(): [prediction] = FuturesAiPrediction.from_json_list( json.dumps( { "predictions": [ { "symbol": "BTCUSDT", "action": "OPEN_LONG", "confidence": "0.88", "stop_loss_price": "NaN", "take_profit_price": "63000", "reason": "bad stop", } ] } ), allowed_symbols=("BTCUSDT",), ) assert prediction.action is FuturesAction.HOLD assert prediction.reason == "模型返回的该交易对预测无效" def test_futures_prediction_schema_uses_futures_actions_and_symbols(): schema = _prediction_schema(("BTCUSDT", "ETHUSDT")) item = schema["properties"]["predictions"]["items"] assert item["properties"]["symbol"]["enum"] == ["BTCUSDT", "ETHUSDT"] assert item["properties"]["action"]["enum"] == ["OPEN_LONG", "OPEN_SHORT", "CLOSE", "HOLD"] assert item["properties"]["close_percent"]["enum"] == [25, 50, 100, None] def test_futures_prediction_schema_requires_nullable_action_fields_for_codex_structured_output(): schema = _prediction_schema(("BTCUSDT",)) item = schema["properties"]["predictions"]["items"] assert item["required"] == [ "symbol", "action", "confidence", "leverage", "stop_loss_price", "take_profit_price", "close_percent", "reason", ] assert item["properties"]["leverage"]["type"] == ["integer", "null"] assert item["properties"]["stop_loss_price"]["type"] == ["string", "null"] assert item["properties"]["take_profit_price"]["type"] == ["string", "null"] assert item["properties"]["close_percent"]["type"] == ["integer", "null"] assert item["properties"]["close_percent"]["enum"] == [25, 50, 100, None] def test_codex_futures_predictor_invokes_codex_with_schema_and_parses_predictions(tmp_path): calls = [] def fake_run(command, **kwargs): calls.append((command, kwargs)) schema_path = Path(command[command.index("--output-schema") + 1]) schema = json.loads(schema_path.read_text(encoding="utf-8")) assert schema["properties"]["predictions"]["minItems"] == 2 assert schema["properties"]["predictions"]["maxItems"] == 2 return subprocess.CompletedProcess( command, 0, stdout=json.dumps( { "predictions": [ { "symbol": "BTCUSDT", "action": "OPEN_LONG", "confidence": "0.86", "leverage": 3, "stop_loss_price": "59000", "take_profit_price": "63000", "reason": "BTC多头趋势增强", }, { "symbol": "ETHUSDT", "action": "CLOSE", "confidence": "0.79", "close_percent": 50, "reason": "ETH持仓风险升高", }, ] } ), stderr="", ) predictor = futures_ai_trader.CodexExecFuturesPredictor( model="gpt-5", timeout_seconds=42, cwd=tmp_path, log_dir=tmp_path / "logs", run_command=fake_run, ) predictions = predictor.predict( {"symbols": ["BTCUSDT", "ETHUSDT"], "risk": {"max_leverage": 3}, "market": []}, allowed_symbols=("BTCUSDT", "ETHUSDT"), ) assert [prediction.symbol for prediction in predictions] == ["BTCUSDT", "ETHUSDT"] assert predictions[0].action is FuturesAction.OPEN_LONG assert predictions[0].confidence == Decimal("0.86") assert predictions[0].leverage == 3 assert predictions[0].stop_loss_price == Decimal("59000") assert predictions[0].take_profit_price == Decimal("63000") assert predictions[1].action is FuturesAction.CLOSE assert predictions[1].close_percent == 50 [(command, kwargs)] = calls assert command[:2] == ["codex", "exec"] assert "--output-schema" in command assert command[command.index("--model") + 1] == "gpt-5" assert kwargs["cwd"] == tmp_path assert kwargs["timeout"] == 42 assert kwargs["capture_output"] is True assert kwargs["text"] is True assert kwargs["encoding"] == "utf-8" assert kwargs["errors"] == "replace" assert kwargs["input"].startswith("You are a futures signal generator") assert "Return one JSON object only" in kwargs["input"] assert "OPEN_LONG, OPEN_SHORT, CLOSE, or HOLD" in kwargs["input"] assert "market[].local_risk" in kwargs["input"] assert "effective_min_stop_distance" in kwargs["input"] assert "Every reason must be written in Chinese" in kwargs["input"] assert "Local risk and execution checks remain authoritative" in kwargs["input"] assert "BTCUSDT" in kwargs["input"] assert "ETHUSDT" in kwargs["input"] assert command[-1] != kwargs["input"] def test_codex_futures_predictor_default_call_log_size_limit_is_20m(tmp_path): predictor = futures_ai_trader.CodexExecFuturesPredictor(cwd=tmp_path) assert predictor.log_max_bytes == 20 * 1024 * 1024 def test_codex_futures_predictor_falls_back_to_hold_on_codex_failure(tmp_path): def fake_run(command, **kwargs): return subprocess.CompletedProcess(command, 1, stdout="", stderr="boom") predictor = futures_ai_trader.CodexExecFuturesPredictor( cwd=tmp_path, log_dir=tmp_path / "logs", run_command=fake_run, ) predictions = predictor.predict({"symbols": ["BTCUSDT", "ETHUSDT"], "market": []}, ("BTCUSDT", "ETHUSDT")) assert [prediction.symbol for prediction in predictions] == ["BTCUSDT", "ETHUSDT"] assert [prediction.action for prediction in predictions] == [FuturesAction.HOLD, FuturesAction.HOLD] assert [prediction.reason for prediction in predictions] == ["codex exec 执行失败", "codex exec 执行失败"] class StaticFuturesPredictor: def __init__(self, predictions): self.predictions = predictions self.contexts = [] def predict(self, context, allowed_symbols): self.contexts.append((context, allowed_symbols)) return self.predictions class FailingFuturesPredictor: def predict(self, context, allowed_symbols): raise RuntimeError("secret api_key=SHOULD_NOT_BE_STORED") class FakeFuturesExchange: def __init__(self): self.calls = [] def current_position_mode(self): self.calls.append(("current_position_mode",)) return False def position_risk(self, symbol=None): self.calls.append(("position_risk", symbol)) return [ { "symbol": symbol or "BTCUSDT", "positionAmt": "0", "entryPrice": "0", "markPrice": "60000", "liquidationPrice": "0", "isolatedMargin": "0", "unRealizedProfit": "0", "leverage": "2", "marginType": "isolated", } ] def exchange_info(self, symbol): self.calls.append(("exchange_info", symbol)) from binance_quant.futures_models import FuturesSymbolMetadata return FuturesSymbolMetadata( symbol, "BTC", "USDT", Decimal("0.001"), Decimal("0.001"), Decimal("0.1"), Decimal("1"), ) def mark_price(self, symbol): self.calls.append(("mark_price", symbol)) return {"markPrice": "60000", "lastFundingRate": "0.0001", "nextFundingTime": 1} def depth(self, symbol, limit=20): return { "bids": [["59999", "1"], ["59998", "2"], ["59997", "3"]], "asks": [["60001", "4"], ["60002", "5"], ["60003", "6"]], } def ticker_24hr(self, symbol): return {"priceChangePercent": "1", "quoteVolume": "100000"} def aggregate_trades(self, symbol, limit=500): return [] def klines(self, symbol, interval, limit): return [[1, "60000", "60100", "59900", "60050", "10", 2, "600500", 1, "5", "300250"]] def open_interest(self, symbol): return {"openInterest": "123"} def open_interest_hist(self, symbol, period="15m", limit=12): return [ {"sumOpenInterest": "100", "sumOpenInterestValue": "6000000", "timestamp": 1}, {"sumOpenInterest": "123", "sumOpenInterestValue": "7380000", "timestamp": 2}, ] def funding_rate_history(self, symbol, limit=8): return [ {"fundingTime": 1, "fundingRate": "0.00005", "markPrice": "59900"}, {"fundingTime": 2, "fundingRate": "0.00010", "markPrice": "60000"}, ] def leverage_bracket(self, symbol): return [ { "symbol": symbol, "brackets": [ { "bracket": 1, "initialLeverage": 20, "notionalCap": "50000", "notionalFloor": "0", "maintMarginRatio": "0.004", } ], } ] def account(self): return { "availableBalance": "100", "totalWalletBalance": "120", "totalMaintMargin": "3", "totalPositionInitialMargin": "30", "totalOpenOrderInitialMargin": "2", "totalUnrealizedProfit": "1.5", } def global_long_short_ratio(self, symbol, period="5m", limit=1): return [{"longShortRatio": "1.2"}] def taker_long_short_ratio(self, symbol, period="5m", limit=1): return [{"buySellRatio": "1.1"}] def all_algo_orders(self, symbol=None): return [] def income(self, start_time_ms, end_time_ms): return [ {"incomeType": "REALIZED_PNL", "income": "-1.5"}, {"incomeType": "COMMISSION", "income": "-0.1"}, {"incomeType": "FUNDING_FEE", "income": "0.05"}, ] class ExecutingExchange(FakeFuturesExchange): def __init__(self): super().__init__() self.orders = [] def change_margin_type(self, symbol, margin_type="ISOLATED"): self.orders.append(("change_margin_type", symbol, margin_type)) return {"code": 200} def change_leverage(self, symbol, leverage): self.orders.append(("change_leverage", symbol, leverage)) return {"leverage": leverage} def exchange_info(self, symbol): self.calls.append(("exchange_info", symbol)) from binance_quant.futures_models import FuturesSymbolMetadata return FuturesSymbolMetadata( symbol, "BTC", "USDT", Decimal("0.000001"), Decimal("0.000001"), Decimal("0.1"), Decimal("1"), ) def create_market_order(self, symbol, side, quantity, client_order_id, *, reduce_only): self.orders.append(("market", symbol, side.value, quantity, reduce_only)) return {"status": "FILLED", "clientOrderId": client_order_id, "executedQty": str(quantity)} def create_conditional_algo_order(self, symbol, side, order_type, trigger_price, quantity, client_algo_id): self.orders.append(("algo", symbol, side.value, order_type, trigger_price, quantity)) return {"status": "NEW", "clientAlgoId": client_algo_id} class MixedExecutionExchange(ExecutingExchange): def exchange_info(self, symbol): self.calls.append(("exchange_info", symbol)) from binance_quant.futures_models import FuturesSymbolMetadata step_size = Decimal("0.001") if symbol == "BTCUSDT" else Decimal("0.000001") return FuturesSymbolMetadata( symbol, "BTC", "USDT", step_size, step_size, Decimal("0.1"), Decimal("1"), ) def futures_settings(tmp_path, **overrides): values = { "mode": "live", "api_key": "key", "api_secret": "secret", "database_path": tmp_path / "audit.sqlite3", "live_futures_ai_trader_enabled": True, "live_futures_confirm_production": True, "futures_rest_base_url": "https://fapi.binance.com", "futures_stream_base_url": "wss://fstream.binance.com/ws", "futures_ai_symbols": ("BTCUSDT",), "futures_margin_amount": Decimal("5"), "futures_default_leverage": 2, "futures_max_leverage": 3, "futures_max_total_margin": Decimal("20"), } values.update(overrides) return Settings(_env_file=None, **values) def trend_test_kline(index, close, close_time_ms=1): close = Decimal(str(close)) return [ index, str(close), str(close + Decimal("100")), str(close - Decimal("100")), str(close), "10", close_time_ms, ] def assert_sanitized_preflight_rejection(result, settings): assert result.status == "rejected" assert result.reason == "futures ai preflight/context failed" assert result.prediction_count == 0 assert result.results[0].status == "rejected" assert result.results[0].reason == "futures ai preflight/context failed" [run] = Storage(settings.database_path).list_futures_ai_trade_runs() assert run["status"] == "rejected" assert run["reason"] == "futures ai preflight/context failed" assert "SHOULD_NOT_BE_STORED" not in run["raw_json"] assert "signedUrl" not in run["raw_json"] def test_futures_batch_rejects_when_live_futures_disabled(tmp_path): settings = futures_settings(tmp_path, live_futures_ai_trader_enabled=False) result = run_live_futures_ai_trade_batch_once( settings, exchange_client=FakeFuturesExchange(), predictor=StaticFuturesPredictor([]), ) assert isinstance(result, FuturesAiBatchResult) assert result.status == "rejected" assert result.reason == "live futures ai trader disabled" def test_futures_batch_rejects_non_live_without_calling_exchange_or_predictor(tmp_path): settings = futures_settings( tmp_path, mode="dry_run", futures_rest_base_url="https://testnet.binancefuture.com", futures_stream_base_url="wss://stream.binancefuture.com/ws", ) exchange = FakeFuturesExchange() predictor = StaticFuturesPredictor([FuturesAiPrediction.hold("hold", "BTCUSDT")]) result = run_live_futures_ai_trade_batch_once(settings, exchange_client=exchange, predictor=predictor) assert result.status == "rejected" assert result.reason == "live mode required" assert result.prediction_count == 0 assert exchange.calls == [] assert predictor.contexts == [] def test_futures_batch_rejects_missing_production_confirmation_without_calling_exchange_or_predictor(tmp_path): settings = futures_settings(tmp_path, live_futures_confirm_production=False) exchange = FakeFuturesExchange() predictor = StaticFuturesPredictor([FuturesAiPrediction.hold("hold", "BTCUSDT")]) result = run_live_futures_ai_trade_batch_once(settings, exchange_client=exchange, predictor=predictor) assert result.status == "rejected" assert result.reason == "live futures production confirmation disabled" assert result.prediction_count == 0 assert exchange.calls == [] assert predictor.contexts == [] def test_futures_batch_records_hold_prediction(tmp_path): prediction = FuturesAiPrediction("BTCUSDT", FuturesAction.HOLD, Decimal("0.33"), "BTC震荡") result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=FakeFuturesExchange(), predictor=StaticFuturesPredictor([prediction]), ) assert result.status == "no_action" assert result.prediction_count == 1 assert result.executed_count == 0 assert result.results[0].status == "no_action" def test_futures_batch_passes_configured_codex_command_to_default_predictor(tmp_path, monkeypatch): created = {} class CapturingPredictor: def __init__(self, **kwargs): created.update(kwargs) def predict(self, context, allowed_symbols): return [FuturesAiPrediction("BTCUSDT", FuturesAction.HOLD, Decimal("0.3"), "BTC震荡")] monkeypatch.setattr("binance_quant.futures_ai_trader.CodexExecFuturesPredictor", CapturingPredictor) settings = futures_settings(tmp_path, ai_trader_codex_command=r"D:\Program Files\nodejs\codex.cmd") result = run_live_futures_ai_trade_batch_once(settings, exchange_client=FakeFuturesExchange()) assert result.status == "no_action" assert created["codex_command"] == r"D:\Program Files\nodejs\codex.cmd" def test_futures_batch_total_margin_includes_unconfigured_positions(tmp_path): class AccountWideMarginExchange(FakeFuturesExchange): def position_risk(self, symbol=None): self.calls.append(("position_risk", symbol)) if symbol is None: return [ { "symbol": "BTCUSDT", "positionAmt": "0.01", "entryPrice": "60000", "markPrice": "60000", "liquidationPrice": "0", "isolatedMargin": "1.25", "unRealizedProfit": "0", "leverage": "2", "marginType": "isolated", }, { "symbol": "ETHUSDT", "positionAmt": "0.10", "entryPrice": "3000", "markPrice": "3000", "liquidationPrice": "0", "isolatedMargin": "7.25", "unRealizedProfit": "0", "leverage": "2", "marginType": "isolated", }, ] return [ { "symbol": symbol, "positionAmt": "0.01", "entryPrice": "60000", "markPrice": "60000", "liquidationPrice": "0", "isolatedMargin": "1.25", "unRealizedProfit": "0", "leverage": "2", "marginType": "isolated", } ] result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=AccountWideMarginExchange(), predictor=StaticFuturesPredictor([FuturesAiPrediction.hold("hold", "BTCUSDT")]), ) assert result.total_isolated_margin == Decimal("8.50") def test_futures_batch_context_includes_futures_market_and_account_features(tmp_path): predictor = StaticFuturesPredictor([FuturesAiPrediction("BTCUSDT", FuturesAction.HOLD, Decimal("0.33"), "BTC震荡")]) result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=FakeFuturesExchange(), predictor=predictor, now_ms=1_700_000_000_000, ) context, allowed_symbols = predictor.contexts[0] symbol_context = context["market"][0] assert allowed_symbols == ("BTCUSDT",) assert result.daily_realized_pnl == Decimal("-1.55") assert result.total_isolated_margin == Decimal("0") assert context["account"]["available_balance"] == "100" assert context["account"]["total_wallet_balance"] == "120" assert context["account"]["total_maint_margin"] == "3" assert context["account"]["total_position_initial_margin"] == "30" assert context["account"]["total_open_order_initial_margin"] == "2" assert context["account"]["total_unrealized_pnl"] == "1.5" assert symbol_context["funding"]["last_funding_rate"] == "0.0001" assert symbol_context["funding"]["history"][0]["fundingRate"] == "0.00005" assert symbol_context["open_interest"] == "123" assert symbol_context["open_interest_history"][-1]["sumOpenInterest"] == "123" assert symbol_context["leverage_bracket"]["brackets"][0]["maintMarginRatio"] == "0.004" assert symbol_context["long_short_ratio"] == "1.2" assert symbol_context["taker_long_short_ratio"] == "1.1" assert symbol_context["order_book"]["spread"] == "2" assert symbol_context["order_book"]["bid_quantity_top_3"] == "6" assert symbol_context["order_book"]["ask_quantity_top_3"] == "15" assert symbol_context["order_book"]["quantity_imbalance_top_3"] == "-0.4285714285714285714285714286" def test_futures_batch_loss_limit_uses_daily_net_pnl_including_fees(tmp_path): class FeeHeavyExchange(FakeFuturesExchange): def income(self, start_time_ms, end_time_ms): return [ {"incomeType": "REALIZED_PNL", "income": "0.25"}, {"incomeType": "COMMISSION", "income": "-0.60"}, {"incomeType": "FUNDING_FEE", "income": "-0.05"}, ] prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_SHORT, Decimal("0.90"), "手续费后净损益触发熔断", leverage=2, stop_loss_price=Decimal("60600"), take_profit_price=Decimal("59000"), ) result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path, futures_daily_realized_loss_limit=Decimal("0.30")), exchange_client=FeeHeavyExchange(), predictor=StaticFuturesPredictor([prediction]), now_ms=1_700_000_000_000, ) assert result.daily_realized_pnl == Decimal("-0.40") assert result.results[0].status == "rejected" assert result.results[0].raw["risk_rejection"]["reason"] == "daily net loss limit reached" assert "当前 -0.40" in result.results[0].reason def test_futures_batch_context_includes_local_stop_distance_constraints(tmp_path): class VolatileContextExchange(FakeFuturesExchange): def klines(self, symbol, interval, limit): if interval == "4h": rows = [] for index in range(16): close = Decimal("60000") + Decimal(index * 10) rows.append( [ index, str(close), str(close + Decimal("100")), str(close - Decimal("100")), str(close), "10", ] ) return rows return super().klines(symbol, interval, limit) predictor = StaticFuturesPredictor([FuturesAiPrediction.hold("hold", "BTCUSDT")]) settings = futures_settings(tmp_path, futures_min_stop_atr_multiple=Decimal("0.8")) run_live_futures_ai_trade_batch_once( settings, exchange_client=VolatileContextExchange(), predictor=predictor, ) context, _allowed_symbols = predictor.contexts[0] risk = context["risk"] symbol_context = context["market"][0] assert risk["min_stop_atr_multiple"] == "0.8" assert risk["min_stop_distance_percent"] == "0.3" assert risk["max_stop_distance_percent"] == "5" assert symbol_context["local_risk"]["atr_4h"] == "200" assert symbol_context["local_risk"]["min_stop_distance_from_atr"] == "160.0" assert symbol_context["local_risk"]["min_stop_distance_price"] == "180.0" assert symbol_context["local_risk"]["effective_min_stop_distance"] == "180.0" assert symbol_context["local_risk"]["max_stop_distance_price"] == "3000" def test_futures_batch_rejects_and_records_sanitized_exchange_context_failure(tmp_path): class FailingMarketExchange(FakeFuturesExchange): def mark_price(self, symbol): raise RuntimeError("signedUrl=https://example.test?signature=SECRET") settings = futures_settings(tmp_path) result = run_live_futures_ai_trade_batch_once( settings, exchange_client=FailingMarketExchange(), predictor=StaticFuturesPredictor([FuturesAiPrediction.hold("hold", "BTCUSDT")]), ) assert_sanitized_preflight_rejection(result, settings) def test_futures_batch_records_preflight_failure_stage_and_sanitized_message(tmp_path): class PositionRiskFailingExchange(FakeFuturesExchange): def position_risk(self, symbol=None): raise RuntimeError("position risk 400 signedUrl=https://example.test?signature=SECRET&api_key=SHOULD_NOT") settings = futures_settings(tmp_path) result = run_live_futures_ai_trade_batch_once( settings, exchange_client=PositionRiskFailingExchange(), predictor=StaticFuturesPredictor([FuturesAiPrediction.hold("hold", "BTCUSDT")]), ) assert_sanitized_preflight_rejection(result, settings) [run] = Storage(settings.database_path).list_futures_ai_trade_runs() raw = json.loads(run["raw_json"]) assert raw["stage"] == "position_risk:BTCUSDT" assert raw["error"]["type"] == "RuntimeError" assert "position risk 400" in raw["error"]["message"] assert "SECRET" not in run["raw_json"] assert "signature" not in run["raw_json"] assert "signedUrl" not in run["raw_json"] def test_futures_batch_rejects_and_records_sanitized_predictor_failure(tmp_path): settings = futures_settings(tmp_path) result = run_live_futures_ai_trade_batch_once( settings, exchange_client=FakeFuturesExchange(), predictor=FailingFuturesPredictor(), ) assert_sanitized_preflight_rejection(result, settings) def test_futures_batch_executes_approved_open_long(tmp_path): prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_LONG, Decimal("0.88"), "BTC转强", leverage=3, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) exchange = ExecutingExchange() result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=exchange, predictor=StaticFuturesPredictor([prediction]), now_ms=1_700_000_000_000, ) assert result.status == "completed" assert result.executed_count == 1 assert result.results[0].status == "protected" assert any(call[0] == "market" for call in exchange.orders) assert [call[3] for call in exchange.orders if call[0] == "algo"] == ["STOP_MARKET", "TAKE_PROFIT_MARKET"] def test_futures_batch_allows_open_long_when_four_hour_pullback_is_neutral(tmp_path): class NeutralPullbackExchange(ExecutingExchange): def klines(self, symbol, interval, limit): if interval == "4h": return [ trend_test_kline(index, 60000 + index * 100) for index in range(28) ] + [ trend_test_kline(28, 62800), trend_test_kline(29, 62750), trend_test_kline(30, 62000, close_time_ms=2_000_000_000_000), ] if interval == "1d": return [ trend_test_kline(index, 58000 + index * 500) for index in range(13) ] + [trend_test_kline(13, 64500)] return super().klines(symbol, interval, limit) prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_LONG, Decimal("0.88"), "small pullback long", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) exchange = NeutralPullbackExchange() result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=exchange, predictor=StaticFuturesPredictor([prediction]), now_ms=1_700_000_000_000, ) assert result.status == "completed" assert result.executed_count == 1 assert result.results[0].status == "protected" assert any(call[0] == "market" for call in exchange.orders) def test_futures_batch_allows_open_long_when_only_four_hour_trend_disagrees(tmp_path): class FallingFourHourRisingDailyExchange(ExecutingExchange): def klines(self, symbol, interval, limit): if interval == "4h": return [ trend_test_kline(index, 60000 + index * 100) for index in range(28) ] + [ trend_test_kline(28, 62800), trend_test_kline(29, 62000), trend_test_kline(30, 61900, close_time_ms=2_000_000_000_000), ] if interval == "1d": return [ trend_test_kline(index, 58000 + index * 500) for index in range(12) ] + [ trend_test_kline(12, 64000), trend_test_kline(13, 64500), ] return super().klines(symbol, interval, limit) prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_LONG, Decimal("0.88"), "daily still supports long", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) exchange = FallingFourHourRisingDailyExchange() result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=exchange, predictor=StaticFuturesPredictor([prediction]), now_ms=1_700_000_000_000, ) assert result.status == "completed" assert result.executed_count == 1 assert result.results[0].status == "protected" assert any(call[0] == "market" for call in exchange.orders) def test_futures_batch_allows_open_long_when_four_hour_drop_is_below_stronger_threshold(tmp_path): class Run1091LikeExchange(ExecutingExchange): def klines(self, symbol, interval, limit): if interval == "4h": return [ trend_test_kline(index, 1640 - index * 3) for index in range(27) ] + [ trend_test_kline(27, "1586.00"), trend_test_kline(28, "1584.75"), trend_test_kline(29, "1583.13"), trend_test_kline(30, "1582.00", close_time_ms=2_000_000_000_000), ] if interval == "1d": return [ trend_test_kline(index, 1500 + index * 5) for index in range(12) ] + [ trend_test_kline(12, "1566.97"), trend_test_kline(13, "1577.86"), ] return super().klines(symbol, interval, limit) prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_LONG, Decimal("0.72"), "run 1091 style borderline long", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) exchange = Run1091LikeExchange() result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=exchange, predictor=StaticFuturesPredictor([prediction]), now_ms=1_700_000_000_000, ) assert result.status == "completed" assert result.executed_count == 1 assert result.results[0].status == "protected" assert any(call[0] == "market" for call in exchange.orders) def test_futures_batch_rejects_open_long_when_four_hour_and_daily_trends_block(tmp_path): class FallingFourHourAndDailyExchange(ExecutingExchange): def klines(self, symbol, interval, limit): if interval == "4h": return [ trend_test_kline(index, 60000 + index * 100) for index in range(28) ] + [ trend_test_kline(28, 62800), trend_test_kline(29, 62000), trend_test_kline(30, 61900, close_time_ms=2_000_000_000_000), ] if interval == "1d": return [ trend_test_kline(index, 65000 - index * 100) for index in range(12) ] + [ trend_test_kline(12, 63800), trend_test_kline(13, 63000), ] return super().klines(symbol, interval, limit) prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_LONG, Decimal("0.88"), "counter trend long", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) exchange = FallingFourHourAndDailyExchange() result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=exchange, predictor=StaticFuturesPredictor([prediction]), ) assert result.status == "no_action" assert result.executed_count == 0 assert result.results[0].status == "rejected" assert ( result.results[0].reason == "AI: counter trend long; 策略: 4h趋势过滤拦截（4h方向 down，变化 -1.273885350318471337579617834%，阈值 0.15%）" ) raw = result.results[0].raw["risk_rejection"] assert raw["reason"] == "trend filter blocked" assert raw["threshold_percent"] == "0.15" assert raw["action"] == "OPEN_LONG" assert raw["primary"] == { "interval": "4h", "prev_close": "62800", "last_close": "62000", "delta_percent": "-1.273885350318471337579617834", "direction": "down", } assert raw["confirmation"] == { "interval": "1d", "prev_close": "63800", "last_close": "63000", "delta_percent": "-1.253918495297805642633228840", "direction": "down", } assert raw["ema"]["fast_period"] == 8 assert raw["ema"]["slow_period"] == 21 assert raw["ema"]["direction"] == "up" assert raw["ema"]["threshold_percent"] == "0.25" assert not any(call[0] == "market" for call in exchange.orders) def test_futures_batch_rejects_open_short_when_four_hour_and_strong_ema_trends_block(tmp_path): class RisingFourHourAndEmaExchange(ExecutingExchange): def klines(self, symbol, interval, limit): if interval == "4h": return [ trend_test_kline(index, 60000 + index * 100) for index in range(28) ] + [ trend_test_kline(28, 62800), trend_test_kline(29, 63600), trend_test_kline(30, 63650, close_time_ms=2_000_000_000_000), ] if interval == "1d": return [ trend_test_kline(index, 66000 - index * 100) for index in range(12) ] + [ trend_test_kline(12, 64800), trend_test_kline(13, 64780), ] return super().klines(symbol, interval, limit) prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_SHORT, Decimal("0.88"), "counter trend short", leverage=2, stop_loss_price=Decimal("63000"), take_profit_price=Decimal("59000"), ) exchange = RisingFourHourAndEmaExchange() result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=exchange, predictor=StaticFuturesPredictor([prediction]), now_ms=1_700_000_000_000, ) assert result.status == "no_action" assert result.executed_count == 0 assert result.results[0].status == "rejected" assert ( result.results[0].reason == "AI: counter trend short; 策略: 4h趋势过滤拦截（4h方向 up，变化 1.273885350318471337579617834%，阈值 0.15%）" ) raw = result.results[0].raw["risk_rejection"] assert raw["reason"] == "trend filter blocked" assert raw["primary"]["interval"] == "4h" assert raw["primary"]["prev_close"] == "62800" assert raw["primary"]["last_close"] == "63600" assert raw["primary"]["direction"] == "up" assert raw["confirmation"]["direction"] == "neutral" assert raw["ema"]["direction"] == "up" assert raw["ema"]["threshold_percent"] == "0.25" assert raw["ema"]["delta_percent"] assert not any(call[0] == "market" for call in exchange.orders) def test_futures_batch_rejects_recent_reverse_open_during_cooldown(tmp_path): now_ms = 1_700_000_000_000 settings = futures_settings(tmp_path) storage = Storage(settings.database_path) storage.initialize() storage.record_futures_ai_trade_run( "BTCUSDT", "CLOSE", Decimal("0.88"), "closed", "closed previous long", raw={"reason": "closed"}, created_at_ms=now_ms - 5 * 60 * 1000, ) prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_SHORT, Decimal("0.88"), "reverse short", leverage=2, stop_loss_price=Decimal("63000"), take_profit_price=Decimal("59000"), ) exchange = ExecutingExchange() result = run_live_futures_ai_trade_batch_once( settings, exchange_client=exchange, predictor=StaticFuturesPredictor([prediction]), now_ms=now_ms, ) assert result.status == "no_action" assert result.executed_count == 0 assert result.results[0].status == "rejected" assert result.results[0].reason == "AI: reverse short; 策略: 反向开仓冷却中（当前冷却窗口内有平仓，冷却 1800 秒）" assert result.results[0].raw["risk_rejection"] == { "reason": "reverse cooldown active", "cooldown_seconds": 1800, "since_ms": 1699998200000, "current_recent_close": True, } assert not any(call[0] == "market" for call in exchange.orders) def test_futures_batch_rejects_stop_loss_below_atr_volatility_floor(tmp_path): class VolatileRisingFourHourExchange(ExecutingExchange): def klines(self, symbol, interval, limit): if interval == "4h": rows = [] for index in range(15): close = Decimal("59800") + Decimal(index * 20) rows.append( [ index, str(close), str(close + Decimal("2000")), str(close - Decimal("2000")), str(close), "10", ] ) return rows return super().klines(symbol, interval, limit) prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_LONG, Decimal("0.88"), "tight stop", leverage=2, stop_loss_price=Decimal("59500"), take_profit_price=Decimal("63000"), ) exchange = VolatileRisingFourHourExchange() result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=exchange, predictor=StaticFuturesPredictor([prediction]), ) assert result.status == "no_action" assert result.executed_count == 0 assert result.results[0].status == "rejected" assert result.results[0].reason == "AI: tight stop; 风控: 止损距离低于波动率底线（当前 500，最小 3200.0）" assert result.results[0].raw["risk_rejection"] == { "reason": "stop distance below volatility floor", "mark_price": "60000", "stop_loss_price": "59500", "stop_distance": "500", "atr_4h": "4000", "min_stop_atr_multiple": "0.8", "min_stop_distance_from_atr": "3200.0", } assert not any(call[0] == "market" for call in exchange.orders) def test_futures_batch_records_position_snapshot_after_successful_open(tmp_path): class SnapshotExchange(ExecutingExchange): def position_risk(self, symbol=None): rows = super().position_risk(symbol) if symbol == "BTCUSDT" and any(call[0] == "market" for call in self.orders): rows[0]["positionAmt"] = "0.001" rows[0]["entryPrice"] = "60000" rows[0]["markPrice"] = "60100" rows[0]["liquidationPrice"] = "45000" rows[0]["isolatedMargin"] = "5" rows[0]["unRealizedProfit"] = "0.1" return rows prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_LONG, Decimal("0.88"), "BTC转强", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) settings = futures_settings(tmp_path) result = run_live_futures_ai_trade_batch_once( settings, exchange_client=SnapshotExchange(), predictor=StaticFuturesPredictor([prediction]), ) assert result.results[0].status == "protected" [snapshot] = Storage(settings.database_path).list_futures_position_snapshots(result.results[0].run_id) assert snapshot["symbol"] == "BTCUSDT" assert snapshot["side"] == "LONG" assert snapshot["quantity"] == "0.001" assert snapshot["mark_price"] == "60100" def test_futures_batch_skips_remaining_actions_after_max_actions(tmp_path): first = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_LONG, Decimal("0.88"), "first", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) second = FuturesAiPrediction( "ETHUSDT", FuturesAction.OPEN_LONG, Decimal("0.87"), "second", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) settings = futures_settings(tmp_path, futures_ai_symbols=("BTCUSDT", "ETHUSDT"), futures_ai_max_actions_per_run=1) result = run_live_futures_ai_trade_batch_once( settings, exchange_client=ExecutingExchange(), predictor=StaticFuturesPredictor([first, second]), ) assert result.executed_count == 1 assert [item.status for item in result.results] == ["protected", "skipped"] def test_futures_batch_does_not_count_local_execution_rejection_against_max_actions(tmp_path): rejected = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_LONG, Decimal("0.91"), "invalid quantity", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) valid = FuturesAiPrediction( "ETHUSDT", FuturesAction.OPEN_LONG, Decimal("0.90"), "valid quantity", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) exchange = MixedExecutionExchange() settings = futures_settings(tmp_path, futures_ai_symbols=("BTCUSDT", "ETHUSDT"), futures_ai_max_actions_per_run=1) result = run_live_futures_ai_trade_batch_once( settings, exchange_client=exchange, predictor=StaticFuturesPredictor([rejected, valid]), ) assert result.executed_count == 1 assert [item.status for item in result.results] == ["rejected", "protected"] assert [call[1] for call in exchange.orders if call[0] == "market"] == ["ETHUSDT"] def test_futures_batch_risk_rejection_reason_keeps_ai_reason(tmp_path): class RiskRejectingExchange(ExecutingExchange): def exchange_info(self, symbol): self.calls.append(("exchange_info", symbol)) from binance_quant.futures_models import FuturesSymbolMetadata if symbol == "BNBUSDT": return FuturesSymbolMetadata( "BTCUSDT", "BTC", "USDT", Decimal("0.000001"), Decimal("0.000001"), Decimal("0.1"), Decimal("1"), ) return FuturesSymbolMetadata( symbol, "BTC", "USDT", Decimal("0.000001"), Decimal("0.000001"), Decimal("0.1"), Decimal("100"), ) bnb_prediction = FuturesAiPrediction( "BNBUSDT", FuturesAction.CLOSE, Decimal("0.75"), "BNB持仓风险升高", close_percent=100, ) btc_prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_SHORT, Decimal("0.74"), "BTC短线下行动能增强", leverage=2, stop_loss_price=Decimal("63000"), take_profit_price=Decimal("59000"), ) settings = futures_settings(tmp_path, futures_ai_symbols=("BNBUSDT", "BTCUSDT")) result = run_live_futures_ai_trade_batch_once( settings, exchange_client=RiskRejectingExchange(), predictor=StaticFuturesPredictor([bnb_prediction, btc_prediction]), ) assert [item.status for item in result.results] == ["rejected", "rejected"] assert result.results[0].reason == "AI: BNB持仓风险升高; 风控: 交易规则与交易对不匹配（预测 BNBUSDT，交易规则 BTCUSDT）" assert result.results[1].reason == "AI: BTC短线下行动能增强; 风控: 名义金额低于最小下单金额（当前 10，最小 100）" assert result.results[0].raw["risk_rejection"] == { "reason": "metadata symbol mismatch", "prediction_symbol": "BNBUSDT", "metadata_symbol": "BTCUSDT", } assert result.results[1].raw["risk_rejection"] == { "reason": "notional below min notional", "margin_amount": "5", "leverage": 2, "notional": "10", "min_notional": "100", } runs_by_symbol = {row["symbol"]: row for row in Storage(settings.database_path).list_futures_ai_trade_runs()} assert runs_by_symbol["BNBUSDT"]["reason"] == "AI: BNB持仓风险升高; 风控: 交易规则与交易对不匹配（预测 BNBUSDT，交易规则 BTCUSDT）" assert runs_by_symbol["BTCUSDT"]["reason"] == "AI: BTC短线下行动能增强; 风控: 名义金额低于最小下单金额（当前 10，最小 100）" def test_futures_batch_low_confidence_rejection_includes_current_and_config_values(tmp_path): prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_LONG, Decimal("0.62"), "信号不足", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) settings = futures_settings(tmp_path, futures_ai_min_confidence=Decimal("0.70")) result = run_live_futures_ai_trade_batch_once( settings, exchange_client=ExecutingExchange(), predictor=StaticFuturesPredictor([prediction]), ) expected_reason = "AI: 信号不足; 风控: AI置信度低于最小阈值（当前 0.62，最小 0.70）" assert result.status == "no_action" assert result.executed_count == 0 assert result.results[0].status == "rejected" assert result.results[0].reason == expected_reason [run] = Storage(settings.database_path).list_futures_ai_trade_runs() assert run["reason"] == expected_reason assert result.results[0].raw["risk_rejection"] == { "reason": "confidence below minimum", "confidence": "0.62", "min_confidence": "0.70", } def test_futures_batch_executes_higher_confidence_candidate_first(tmp_path): lower = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_LONG, Decimal("0.80"), "lower", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) higher = FuturesAiPrediction( "ETHUSDT", FuturesAction.OPEN_LONG, Decimal("0.90"), "higher", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) exchange = ExecutingExchange() settings = futures_settings(tmp_path, futures_ai_symbols=("BTCUSDT", "ETHUSDT"), futures_ai_max_actions_per_run=1) result = run_live_futures_ai_trade_batch_once( settings, exchange_client=exchange, predictor=StaticFuturesPredictor([lower, higher]), ) assert [item.symbol for item in result.results] == ["ETHUSDT", "BTCUSDT"] assert [item.status for item in result.results] == ["protected", "skipped"] assert [call[1] for call in exchange.orders if call[0] == "market"] == ["ETHUSDT"] def test_futures_batch_uses_symbol_order_as_confidence_tiebreaker(tmp_path): later_symbol = FuturesAiPrediction( "ETHUSDT", FuturesAction.OPEN_LONG, Decimal("0.88"), "later", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) earlier_symbol = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_LONG, Decimal("0.88"), "earlier", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) exchange = ExecutingExchange() settings = futures_settings(tmp_path, futures_ai_symbols=("BTCUSDT", "ETHUSDT"), futures_ai_max_actions_per_run=1) result = run_live_futures_ai_trade_batch_once( settings, exchange_client=exchange, predictor=StaticFuturesPredictor([later_symbol, earlier_symbol]), ) assert [item.symbol for item in result.results] == ["BTCUSDT", "ETHUSDT"] assert [item.status for item in result.results] == ["protected", "skipped"] assert [call[1] for call in exchange.orders if call[0] == "market"] == ["BTCUSDT"] def test_futures_batch_rejects_candidate_refresh_failure_without_leaving_predicted_run(tmp_path): class RefreshFailingExchange(ExecutingExchange): def __init__(self): super().__init__() self.exchange_info_calls_by_symbol = {} def exchange_info(self, symbol): count = self.exchange_info_calls_by_symbol.get(symbol, 0) self.exchange_info_calls_by_symbol[symbol] = count + 1 if count > 0: raise RuntimeError("signedUrl=https://example.test?signature=SECRET") return super().exchange_info(symbol) prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_LONG, Decimal("0.88"), "refresh fails", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) settings = futures_settings(tmp_path) result = run_live_futures_ai_trade_batch_once( settings, exchange_client=RefreshFailingExchange(), predictor=StaticFuturesPredictor([prediction]), ) [run] = Storage(settings.database_path).list_futures_ai_trade_runs() assert result.results[0].status == "rejected" assert result.results[0].reason == "AI: refresh fails; 执行: 合约AI执行失败" assert run["status"] == "rejected" assert run["reason"] == "AI: refresh fails; 执行: 合约AI执行失败" assert "predicted" != run["status"] assert "SECRET" not in run["raw_json"] assert "signedUrl" not in run["raw_json"] raw = json.loads(run["raw_json"]) assert raw["execution_error"] == { "stage": "exchange_info", "error": { "type": "RuntimeError", "message": "***REDACTED***", }, } def test_futures_batch_skips_margin_type_change_when_symbol_already_isolated(tmp_path): class AlreadyIsolatedExchange(ExecutingExchange): def change_margin_type(self, symbol, margin_type="ISOLATED"): raise AssertionError("margin type should not be changed when already isolated") prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_LONG, Decimal("0.88"), "already isolated", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=AlreadyIsolatedExchange(), predictor=StaticFuturesPredictor([prediction]), ) assert result.results[0].status == "protected" assert result.executed_count == 1 def test_futures_batch_reports_failed_when_open_rollback_fails(tmp_path): class RollbackFailingExchange(ExecutingExchange): def create_market_order(self, symbol, side, quantity, client_order_id, *, reduce_only): self.orders.append(("market", symbol, side.value, quantity, reduce_only)) if reduce_only: raise RuntimeError("rollback signedUrl=https://example.test?signature=SECRET") return {"status": "FILLED", "clientOrderId": client_order_id, "executedQty": str(quantity)} def create_conditional_algo_order(self, symbol, side, order_type, trigger_price, quantity, client_algo_id): self.orders.append(("algo", symbol, side.value, order_type, trigger_price, quantity)) raise RuntimeError("protection failed") prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_LONG, Decimal("0.88"), "rollback fails", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=RollbackFailingExchange(), predictor=StaticFuturesPredictor([prediction]), ) assert result.status == "failed" assert result.reason == "AI合约批量交易失败" assert result.executed_count == 1 assert result.results[0].status == "rollback_failed" def test_futures_batch_reports_failed_when_open_market_order_status_unknown(tmp_path): class OpenUnknownExchange(ExecutingExchange): def create_market_order(self, symbol, side, quantity, client_order_id, *, reduce_only): self.orders.append(("market", symbol, side.value, quantity, reduce_only)) if not reduce_only: raise RuntimeError("open timeout signedUrl=https://example.test?signature=SECRET") return {"status": "FILLED", "clientOrderId": client_order_id, "executedQty": str(quantity)} first = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_LONG, Decimal("0.91"), "unknown open", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) second = FuturesAiPrediction( "ETHUSDT", FuturesAction.OPEN_LONG, Decimal("0.90"), "should not execute", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) settings = futures_settings(tmp_path, futures_ai_symbols=("BTCUSDT", "ETHUSDT"), futures_ai_max_actions_per_run=1) exchange = OpenUnknownExchange() result = run_live_futures_ai_trade_batch_once( settings, exchange_client=exchange, predictor=StaticFuturesPredictor([first, second]), ) events = Storage(settings.database_path).list_futures_order_events(result.results[0].run_id) assert result.status == "failed" assert result.executed_count == 1 assert [item.status for item in result.results] == ["open_unknown", "skipped"] assert [call[1] for call in exchange.orders if call[0] == "market"] == ["BTCUSDT"] assert events[-1]["event_type"] == "open_unknown" assert "open order status unknown" in events[-1]["raw_json"] assert "SECRET" not in events[-1]["raw_json"] assert "signature" not in events[-1]["raw_json"] assert "signedUrl" not in events[-1]["raw_json"] def test_futures_batch_reports_failed_when_open_rollback_completes(tmp_path): class RollbackCompletingExchange(ExecutingExchange): def create_conditional_algo_order(self, symbol, side, order_type, trigger_price, quantity, client_algo_id): self.orders.append(("algo", symbol, side.value, order_type, trigger_price, quantity)) raise RuntimeError("protection failed") prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_LONG, Decimal("0.88"), "rollback completes", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=RollbackCompletingExchange(), predictor=StaticFuturesPredictor([prediction]), ) assert result.status == "failed" assert result.executed_count == 1 assert result.results[0].status == "rollback_completed" def test_futures_batch_executes_close_for_existing_long_position(tmp_path): class ExistingLongCloseExchange(ExecutingExchange): def position_risk(self, symbol=None): rows = super().position_risk(symbol) if symbol == "BTCUSDT": rows[0]["positionAmt"] = "0.01" rows[0]["entryPrice"] = "60000" rows[0]["isolatedMargin"] = "5" return rows def cancel_algo_order(self, client_algo_id): self.orders.append(("cancel_algo", client_algo_id)) return {"status": "CANCELED", "clientAlgoId": client_algo_id} def all_algo_orders(self, symbol=None): if symbol == "BTCUSDT": return [ {"clientAlgoId": "bqf-stop", "type": "STOP_MARKET", "triggerPrice": "59000"}, {"clientAlgoId": "bqf-tp", "type": "TAKE_PROFIT_MARKET", "triggerPrice": "63000"}, ] return [] prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.CLOSE, Decimal("0.88"), "close half", close_percent=50, ) exchange = ExistingLongCloseExchange() result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=exchange, predictor=StaticFuturesPredictor([prediction]), ) assert result.status == "completed" assert result.executed_count == 1 assert result.results[0].status == "partially_closed" assert ("cancel_algo", "bqf-stop") in exchange.orders assert ("cancel_algo", "bqf-tp") in exchange.orders market_order = next(call for call in exchange.orders if call[0] == "market") assert market_order[2] == "SELL" assert market_order[3] == Decimal("0.005") assert market_order[4] is True def test_futures_batch_reports_failed_when_partial_close_cannot_recreate_protection(tmp_path): class ExistingLongUnprotectedCloseExchange(ExecutingExchange): def position_risk(self, symbol=None): rows = super().position_risk(symbol) if symbol == "BTCUSDT": rows[0]["positionAmt"] = "0.01" rows[0]["entryPrice"] = "60000" rows[0]["isolatedMargin"] = "5" return rows def all_algo_orders(self, symbol=None): return [] prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.CLOSE, Decimal("0.88"), "close half", close_percent=50, ) exchange = ExistingLongUnprotectedCloseExchange() result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=exchange, predictor=StaticFuturesPredictor([prediction]), ) assert result.status == "failed" assert result.executed_count == 1 assert result.results[0].status == "close_protection_failed" market_order = next(call for call in exchange.orders if call[0] == "market") assert market_order[2] == "SELL" assert market_order[4] is True def test_futures_batch_malformed_algo_rows_still_reach_close_path(tmp_path): class ExistingLongMalformedProtectionExchange(ExecutingExchange): def position_risk(self, symbol=None): rows = super().position_risk(symbol) if symbol == "BTCUSDT": rows[0]["positionAmt"] = "0.01" rows[0]["entryPrice"] = "60000" rows[0]["isolatedMargin"] = "5" return rows def all_algo_orders(self, symbol=None): return [ None, "not-a-dict", {"clientAlgoId": "bqf-stop", "status": "NEW", "type": "STOP_MARKET", "triggerPrice": "NaN"}, {"clientAlgoId": "bqf-tp", "status": "NEW", "type": "TAKE_PROFIT_MARKET", "triggerPrice": "-1"}, ] def cancel_algo_order(self, client_algo_id): self.orders.append(("cancel_algo", client_algo_id)) return {"status": "CANCELED", "clientAlgoId": client_algo_id} prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.CLOSE, Decimal("0.88"), "close half", close_percent=50, ) exchange = ExistingLongMalformedProtectionExchange() result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=exchange, predictor=StaticFuturesPredictor([prediction]), ) assert result.status == "failed" assert result.executed_count == 1 assert result.results[0].status == "close_protection_failed" assert any(call[0] == "market" for call in exchange.orders) def test_futures_batch_executes_full_close_for_existing_short_position(tmp_path): class ExistingShortCloseExchange(ExecutingExchange): def position_risk(self, symbol=None): rows = super().position_risk(symbol) if symbol == "BTCUSDT": rows[0]["positionAmt"] = "-0.02" rows[0]["entryPrice"] = "60000" rows[0]["isolatedMargin"] = "5" return rows def cancel_algo_order(self, client_algo_id): self.orders.append(("cancel_algo", client_algo_id)) return {"status": "CANCELED", "clientAlgoId": client_algo_id} def all_algo_orders(self, symbol=None): return [] prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.CLOSE, Decimal("0.88"), "close all", close_percent=100, ) exchange = ExistingShortCloseExchange() result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=exchange, predictor=StaticFuturesPredictor([prediction]), ) assert result.status == "completed" assert result.executed_count == 1 assert result.results[0].status == "closed" market_order = next(call for call in exchange.orders if call[0] == "market") assert market_order[2] == "BUY" assert market_order[3] == Decimal("0.020000") assert market_order[4] is True def test_futures_batch_reverse_signal_does_not_open_in_same_run(tmp_path): class ExistingLongExchange(ExecutingExchange): def position_risk(self, symbol=None): rows = super().position_risk(symbol) if symbol == "BTCUSDT": rows[0]["positionAmt"] = "0.01" rows[0]["isolatedMargin"] = "5" return rows prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_SHORT, Decimal("0.88"), "reverse signal", leverage=2, stop_loss_price=Decimal("63000"), take_profit_price=Decimal("59000"), ) exchange = ExistingLongExchange() result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=exchange, predictor=StaticFuturesPredictor([prediction]), ) assert result.status == "no_action" assert result.executed_count == 0 assert result.results[0].status == "rejected" assert result.results[0].reason == "AI: reverse signal; 风控: 反向持仓需要先平仓（当前持仓 LONG，数量 0.01，动作 OPEN_SHORT）" assert not any(call[0] == "market" for call in exchange.orders) def test_futures_batch_close_ignores_non_bot_algo_orders(tmp_path): class ExistingLongCloseExchange(ExecutingExchange): def position_risk(self, symbol=None): rows = super().position_risk(symbol) if symbol == "BTCUSDT": rows[0]["positionAmt"] = "0.01" rows[0]["isolatedMargin"] = "5" return rows def cancel_algo_order(self, client_algo_id): self.orders.append(("cancel_algo", client_algo_id)) return {"status": "CANCELED", "clientAlgoId": client_algo_id} def all_algo_orders(self, symbol=None): return [ {"clientAlgoId": "manual-stop", "type": "STOP_MARKET", "triggerPrice": "59000"}, {"clientAlgoId": "bqf-stop", "type": "STOP_MARKET", "triggerPrice": "59000"}, {"clientAlgoId": "manual-tp", "type": "TAKE_PROFIT_MARKET", "triggerPrice": "63000"}, {"clientAlgoId": "bqf-tp", "type": "TAKE_PROFIT_MARKET", "triggerPrice": "63000"}, ] prediction = FuturesAiPrediction("BTCUSDT", FuturesAction.CLOSE, Decimal("0.88"), "close", close_percent=100) exchange = ExistingLongCloseExchange() result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=exchange, predictor=StaticFuturesPredictor([prediction]), ) assert result.results[0].status == "closed" assert [call[1] for call in exchange.orders if call[0] == "cancel_algo"] == ["bqf-stop", "bqf-tp"] def test_futures_batch_close_uses_only_active_bot_algo_orders(tmp_path): class ExistingLongCloseExchange(ExecutingExchange): def position_risk(self, symbol=None): rows = super().position_risk(symbol) if symbol == "BTCUSDT": rows[0]["positionAmt"] = "0.01" rows[0]["isolatedMargin"] = "5" return rows def cancel_algo_order(self, client_algo_id): self.orders.append(("cancel_algo", client_algo_id)) return {"status": "CANCELED", "clientAlgoId": client_algo_id} def all_algo_orders(self, symbol=None): return [ {"clientAlgoId": "bqf-old-stop", "status": "CANCELED", "type": "STOP_MARKET", "triggerPrice": "1"}, {"clientAlgoId": "bqf-old-tp", "status": "EXPIRED", "type": "TAKE_PROFIT_MARKET", "triggerPrice": "999999"}, {"clientAlgoId": "bqf-stop", "status": "NEW", "type": "STOP_MARKET", "triggerPrice": "59000"}, {"clientAlgoId": "bqf-tp", "status": "NEW", "type": "TAKE_PROFIT_MARKET", "triggerPrice": "63000"}, {"clientAlgoId": "bqf-triggered", "status": "TRIGGERED", "type": "STOP_MARKET", "triggerPrice": "58000"}, ] prediction = FuturesAiPrediction("BTCUSDT", FuturesAction.CLOSE, Decimal("0.88"), "close", close_percent=50) exchange = ExistingLongCloseExchange() result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=exchange, predictor=StaticFuturesPredictor([prediction]), ) assert result.results[0].status == "partially_closed" assert [call[1] for call in exchange.orders if call[0] == "cancel_algo"] == ["bqf-stop", "bqf-tp"] protection_calls = [call for call in exchange.orders if call[0] == "algo"] assert [call[4] for call in protection_calls] == [Decimal("59000"), Decimal("63000")] def test_futures_batch_close_uses_binance_algo_status_and_order_type_fields(tmp_path): class ExistingLongCloseExchange(ExecutingExchange): def position_risk(self, symbol=None): rows = super().position_risk(symbol) if symbol == "BTCUSDT": rows[0]["positionAmt"] = "0.01" rows[0]["isolatedMargin"] = "5" return rows def cancel_algo_order(self, client_algo_id): if client_algo_id.startswith("bqf-old"): raise AssertionError("stale algo orders must not be canceled") self.orders.append(("cancel_algo", client_algo_id)) return {"status": "CANCELED", "clientAlgoId": client_algo_id} def all_algo_orders(self, symbol=None): return [ { "clientAlgoId": "bqf-stop", "algoStatus": "NEW", "orderType": "STOP_MARKET", "triggerPrice": "59000", }, { "clientAlgoId": "bqf-tp", "algoStatus": "NEW", "orderType": "TAKE_PROFIT_MARKET", "triggerPrice": "63000", }, { "clientAlgoId": "bqf-old-stop", "algoStatus": "FINISHED", "orderType": "STOP_MARKET", "triggerPrice": "58000", }, { "clientAlgoId": "bqf-old-tp", "algoStatus": "CANCELED", "orderType": "TAKE_PROFIT_MARKET", "triggerPrice": "64000", }, ] prediction = FuturesAiPrediction("BTCUSDT", FuturesAction.CLOSE, Decimal("0.88"), "close", close_percent=50) exchange = ExistingLongCloseExchange() result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=exchange, predictor=StaticFuturesPredictor([prediction]), ) assert result.results[0].status == "partially_closed" assert [call[1] for call in exchange.orders if call[0] == "cancel_algo"] == ["bqf-stop", "bqf-tp"] protection_calls = [call for call in exchange.orders if call[0] == "algo"] assert [call[4] for call in protection_calls] == [Decimal("59000"), Decimal("63000")] def test_futures_batch_rejects_close_without_position_and_does_not_count_action(tmp_path): prediction = FuturesAiPrediction("BTCUSDT", FuturesAction.CLOSE, Decimal("0.88"), "close", close_percent=100) exchange = ExecutingExchange() result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=exchange, predictor=StaticFuturesPredictor([prediction]), ) assert result.status == "no_action" assert result.executed_count == 0 assert result.results[0].status == "rejected" assert result.results[0].reason == "AI: close; 风控: 没有可关闭的持仓（当前持仓 FLAT，数量 0）" assert result.results[0].raw["risk_rejection"] == { "reason": "no position to close", "position_side": "FLAT", "position_quantity": "0", } assert not any(call[0] == "market" for call in exchange.orders) def test_futures_batch_risk_rejection_does_not_count_against_max_actions(tmp_path): class ExistingBtcPositionExchange(ExecutingExchange): def position_risk(self, symbol=None): rows = super().position_risk(symbol) if symbol == "BTCUSDT": rows[0]["positionAmt"] = "0.01" rows[0]["isolatedMargin"] = "1" return rows rejected = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_LONG, Decimal("0.90"), "position exists", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) valid = FuturesAiPrediction( "ETHUSDT", FuturesAction.OPEN_LONG, Decimal("0.80"), "valid", leverage=2, stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) exchange = ExistingBtcPositionExchange() settings = futures_settings( tmp_path, futures_ai_symbols=("BTCUSDT", "ETHUSDT"), futures_ai_max_actions_per_run=1, ) result = run_live_futures_ai_trade_batch_once( settings, exchange_client=exchange, predictor=StaticFuturesPredictor([rejected, valid]), ) assert result.executed_count == 1 assert [item.status for item in result.results] == ["rejected", "protected"] assert result.results[0].reason == "AI: position exists; 风控: 已有同方向持仓（当前持仓 LONG，数量 0.01，动作 OPEN_LONG）" assert [call[1] for call in exchange.orders if call[0] == "market"] == ["ETHUSDT"] def test_futures_batch_rejects_open_prediction_with_invalid_execution_quantity(tmp_path): prediction = FuturesAiPrediction( "BTCUSDT", FuturesAction.OPEN_LONG, Decimal("0.91"), "trend", stop_loss_price=Decimal("59000"), take_profit_price=Decimal("63000"), ) result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=FakeFuturesExchange(), predictor=StaticFuturesPredictor([prediction]), ) assert result.status == "no_action" assert result.executed_count == 0 assert result.results[0].status == "rejected" assert result.results[0].reason == "AI: trend; 执行: 执行数量无效" def test_futures_batch_normalizes_injected_invalid_open_prediction_to_hold(tmp_path): prediction = FuturesAiPrediction("BTCUSDT", FuturesAction.OPEN_LONG, Decimal("0.91"), "missing exits") result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=FakeFuturesExchange(), predictor=StaticFuturesPredictor([prediction]), ) assert result.status == "no_action" assert result.prediction_count == 1 assert result.results[0].status == "no_action" assert result.results[0].action == FuturesAction.HOLD.value assert result.results[0].reason == "模型返回的该交易对预测无效" def test_futures_batch_preflight_rejects_hedge_mode(tmp_path): class HedgeExchange(FakeFuturesExchange): def current_position_mode(self): return True result = run_live_futures_ai_trade_batch_once( futures_settings(tmp_path), exchange_client=HedgeExchange(), predictor=StaticFuturesPredictor([]), ) assert result.status == "rejected" assert result.reason == "hedge mode active"