import json import subprocess from decimal import Decimal from pathlib import Path from pydantic import SecretStr from binance_quant.ai_trader import ( AiTradePrediction, CodexExecPredictor, _build_prompt, _indicator_calculation_limit, _kline_to_candle, _timeframe_features, run_live_ai_trade_batch_once, run_live_ai_trade_once, ) from binance_quant.config import Settings from binance_quant.models import AccountSnapshot, Balance, Candle, SymbolMetadata, TradingMode from binance_quant.storage import Storage def read_audit_events(tmp_path: Path) -> list[dict]: audit_path = tmp_path / "logs" / "audit.log" events = [] for block in audit_path.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 def test_prediction_from_valid_json_normalizes_fields(): prediction = AiTradePrediction.from_json( '{"symbol":"btcusdt","action":"buy","confidence":"0.82","reason":"upward momentum"}', allowed_symbols=("BTCUSDT", "ETHUSDT"), ) assert prediction.symbol == "BTCUSDT" assert prediction.action == "BUY" assert prediction.confidence == Decimal("0.82") assert prediction.reason == "upward momentum" def test_prediction_invalid_output_falls_back_to_hold(): prediction = AiTradePrediction.from_json( '{"symbol":"ETHBTC","action":"BUY","confidence":"1.20","reason":""}', allowed_symbols=("BTCUSDT",), ) assert prediction.symbol == "" assert prediction.action == "HOLD" assert prediction.confidence == Decimal("0") assert prediction.reason == "invalid model response" def test_prediction_list_from_json_returns_one_prediction_per_allowed_symbol(): predictions = AiTradePrediction.from_json_list( json.dumps( { "predictions": [ { "symbol": "btcusdt", "action": "hold", "confidence": "0.30", "reason": "BTC走势不清晰", }, { "symbol": "ethusdt", "action": "sell", "confidence": "0.82", "reason": "ETH短线走弱", }, ] } ), allowed_symbols=("BTCUSDT", "ETHUSDT"), ) assert [prediction.symbol for prediction in predictions] == ["BTCUSDT", "ETHUSDT"] assert [prediction.action for prediction in predictions] == ["HOLD", "SELL"] assert [prediction.confidence for prediction in predictions] == [Decimal("0.30"), Decimal("0.82")] assert predictions[1].reason == "ETH短线走弱" def test_prediction_list_from_json_fills_missing_symbols_with_hold(): predictions = AiTradePrediction.from_json_list( json.dumps( { "predictions": [ { "symbol": "BTCUSDT", "action": "BUY", "confidence": "0.81", "reason": "BTC上行动能较强", } ] } ), allowed_symbols=("BTCUSDT", "ETHUSDT"), ) assert [prediction.symbol for prediction in predictions] == ["BTCUSDT", "ETHUSDT"] assert [prediction.action for prediction in predictions] == ["BUY", "HOLD"] assert predictions[1].confidence == Decimal("0") assert predictions[1].reason == "模型未返回该交易对预测" def test_ai_trader_kline_to_candle_captures_full_binance_fields(): candle = _kline_to_candle( "btcusdt", [ 1, "100", "110", "90", "105", "12.5", 2, "1312.5", 44, "6.25", "656.25", "0", ], ) assert candle.symbol == "BTCUSDT" assert candle.open == Decimal("100") assert candle.high == Decimal("110") assert candle.low == Decimal("90") assert candle.close == Decimal("105") assert candle.volume == Decimal("12.5") assert candle.quote_volume == Decimal("1312.5") assert candle.trade_count == 44 assert candle.taker_buy_base_volume == Decimal("6.25") assert candle.taker_buy_quote_volume == Decimal("656.25") def test_indicator_calculation_limit_caps_binance_request_size(): assert _indicator_calculation_limit(24) == 74 assert _indicator_calculation_limit(980) == 1000 def test_codex_exec_predictor_builds_command_and_parses_response(tmp_path): calls = [] def fake_run(command, **kwargs): calls.append((command, kwargs)) return subprocess.CompletedProcess( command, 0, stdout=json.dumps( { "predictions": [ { "symbol": "BTCUSDT", "action": "HOLD", "confidence": "0.33", "reason": "BTC方向不明确", }, { "symbol": "ETHUSDT", "action": "SELL", "confidence": "0.77", "reason": "ETH下行风险增加", }, ] } ), stderr="", ) predictor = CodexExecPredictor( model="gpt-5", timeout_seconds=42, cwd=tmp_path, run_command=fake_run, ) predictions = predictor.predict( {"symbols": ["BTCUSDT", "ETHUSDT"], "market": []}, allowed_symbols=("BTCUSDT", "ETHUSDT"), ) assert [prediction.symbol for prediction in predictions] == ["BTCUSDT", "ETHUSDT"] assert [prediction.action for prediction in predictions] == ["HOLD", "SELL"] assert predictions[1].confidence == Decimal("0.77") [(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 market-direction classifier") assert "Return one JSON object only" in kwargs["input"] assert "exactly one item for every symbol" in kwargs["input"] assert "reason must be written in Chinese" in kwargs["input"] assert "at least three independent evidence groups support the same direction" in kwargs["input"] assert "Reserve 0.85 or above for broad agreement" in kwargs["input"] assert "Use liquidity, order-book imbalance, and aggregate trades only as secondary confirmation" in kwargs["input"] assert "Do not consider account balances" not in kwargs["input"] assert "market[].execution" in kwargs["input"] assert "confidence at or above 0.70" in kwargs["input"] assert "BTCUSDT" in kwargs["input"] assert "ETHUSDT" in kwargs["input"] assert command[-1] != kwargs["input"] def test_codex_exec_predictor_default_call_log_size_limit_is_20m(tmp_path): predictor = CodexExecPredictor(cwd=tmp_path) assert predictor.log_max_bytes == 20 * 1024 * 1024 def test_prediction_schema_is_bound_to_allowed_symbols_and_prediction_count(tmp_path): captured_schema = {} def fake_run(command, **kwargs): schema_path = Path(command[command.index("--output-schema") + 1]) captured_schema.update(json.loads(schema_path.read_text(encoding="utf-8"))) return subprocess.CompletedProcess( command, 0, stdout=json.dumps( { "predictions": [ { "symbol": "BTCUSDT", "action": "HOLD", "confidence": "0.30", "reason": "BTC震荡", }, { "symbol": "ETHUSDT", "action": "HOLD", "confidence": "0.31", "reason": "ETH震荡", }, ] } ), stderr="", ) predictor = CodexExecPredictor(cwd=tmp_path, run_command=fake_run) predictor.predict({"symbols": ["BTCUSDT", "ETHUSDT"], "market": []}, ("BTCUSDT", "ETHUSDT")) predictions_schema = captured_schema["properties"]["predictions"] assert predictions_schema["minItems"] == 2 assert predictions_schema["maxItems"] == 2 item_properties = predictions_schema["items"]["properties"] assert item_properties["symbol"]["enum"] == ["BTCUSDT", "ETHUSDT"] assert "maxLength" not in item_properties["reason"] def test_build_prompt_focuses_on_market_direction_and_data_quality_rules(): prompt = _build_prompt( { "symbols": ["BTCUSDT"], "market": [ { "symbol": "BTCUSDT", "summary": {"trend": {"primary_direction": "mixed"}}, "execution": { "quote_free": "50", "base_free": "0.1", "base_locked": "0", }, } ], } ) assert "Use market[].summary first" in prompt assert "Visible timeframe arrays contain the recent display window" in prompt assert "indicators may use additional warm-up candles" in prompt assert "Do not consider account balances" not in prompt assert "Use market[].execution" in prompt assert "Choose BUY when price trend and momentum are bullish" in prompt assert "Choose SELL when price trend and momentum are bearish" in prompt assert "Choose HOLD when evidence is mixed, weak, stale, or only supported by one isolated signal" in prompt assert "at least three independent evidence groups support the same direction" in prompt assert "Reserve 0.85 or above for broad agreement" in prompt assert "Base the direction mainly on trend, momentum, volatility, volume confirmation, breakout or breakdown, and 24h context" in prompt assert "Use liquidity, order-book imbalance, and aggregate trades only as secondary confirmation" in prompt assert "When realtime.status is unavailable" in prompt assert "reason must describe evidence in natural Chinese" in prompt assert "Do not write raw JSON field names" in prompt assert "volume_ratio_20" not in prompt assert "depth_imbalance" not in prompt def test_codex_exec_predictor_sends_prompt_on_stdin_to_avoid_long_windows_command_lines(tmp_path): calls = [] def fake_run(command, **kwargs): calls.append((command, kwargs)) return subprocess.CompletedProcess( command, 0, stdout=json.dumps( { "predictions": [ { "symbol": "BTCUSDT", "action": "HOLD", "confidence": "0.30", "reason": "BTC震荡", } ] } ), stderr="", ) predictor = CodexExecPredictor(cwd=tmp_path, run_command=fake_run) long_context = {"symbols": ["BTCUSDT"], "market": [{"symbol": "BTCUSDT", "payload": "x" * 60000}]} predictor.predict(long_context, allowed_symbols=("BTCUSDT",)) [(command, kwargs)] = calls assert command[:2] == ["codex", "exec"] assert "--output-schema" in command assert all("x" * 1000 not in part for part in command) assert "x" * 1000 in kwargs["input"] def test_codex_exec_predictor_truncates_verbose_stderr_in_call_log(tmp_path): verbose_stderr = "prompt echo " + ("x" * 5000) + " final upstream error" def fake_run(command, **kwargs): return subprocess.CompletedProcess( command, 1, stdout="", stderr=verbose_stderr, ) predictor = CodexExecPredictor(cwd=tmp_path, log_dir=tmp_path / "logs", run_command=fake_run) predictions = predictor.predict({"symbols": ["BTCUSDT"], "market": []}, allowed_symbols=("BTCUSDT",)) assert predictions[0].action == "HOLD" log_text = (tmp_path / "logs" / "ai_trader_llm.log").read_text(encoding="utf-8") assert "final upstream error" in log_text assert "x" * 3000 not in log_text assert "truncated" in log_text def test_codex_exec_predictor_uses_configured_codex_command(tmp_path): calls = [] def fake_run(command, **kwargs): calls.append((command, kwargs)) return subprocess.CompletedProcess( command, 0, stdout=json.dumps( { "predictions": [ { "symbol": "BTCUSDT", "action": "HOLD", "confidence": "0.30", "reason": "BTC震荡", } ] } ), stderr="", ) predictor = CodexExecPredictor( codex_command=r"D:\Program Files\nodejs\codex.cmd", timeout_seconds=42, cwd=tmp_path, run_command=fake_run, ) predictor.predict({"symbols": ["BTCUSDT"], "market": []}, allowed_symbols=("BTCUSDT",)) [(command, _)] = calls assert command[:2] == [r"D:\Program Files\nodejs\codex.cmd", "exec"] def test_codex_exec_predictor_writes_readable_call_log(tmp_path): def fake_run(command, **kwargs): return subprocess.CompletedProcess( command, 0, stdout=json.dumps( { "predictions": [ { "symbol": "BTCUSDT", "action": "HOLD", "confidence": "0.30", "reason": "BTC震荡", } ] } ), stderr="model stderr", ) predictor = CodexExecPredictor( model="gpt-5", timeout_seconds=42, cwd=tmp_path, run_command=fake_run, ) predictor.predict({"symbols": ["BTCUSDT"], "market": []}, allowed_symbols=("BTCUSDT",)) log_text = (tmp_path / "logs" / "ai_trader_llm.log").read_text(encoding="utf-8") assert "Codex Exec AI Trader Call" in log_text assert "Model: gpt-5" in log_text assert "Allowed symbols: BTCUSDT" in log_text assert "Timeout seconds: 42" in log_text assert "Status: completed" in log_text assert "Return code: 0" in log_text assert "Duration ms:" in log_text assert "Prompt:" in log_text assert "Market context:" in log_text assert '"symbols": ["BTCUSDT"]' in log_text assert "Stdout:" in log_text assert "BTC震荡" in log_text assert "Stderr:" in log_text assert "model stderr" in log_text def test_codex_exec_predictor_writes_structured_audit_summary(tmp_path): def fake_run(command, **kwargs): return subprocess.CompletedProcess( command, 0, stdout=json.dumps( { "predictions": [ { "symbol": "BTCUSDT", "action": "HOLD", "confidence": "0.30", "reason": "BTC震荡", } ] } ), stderr="", ) predictor = CodexExecPredictor( model="gpt-5", timeout_seconds=42, cwd=tmp_path, run_command=fake_run, ) predictor.predict({"symbols": ["BTCUSDT"], "market": []}, allowed_symbols=("BTCUSDT",)) [event] = read_audit_events(tmp_path) assert event["event"] == "ai_trader.llm_call" assert event["params"] == {"model": "gpt-5", "allowed_symbols": ["BTCUSDT"], "timeout_seconds": 42} assert event["result"]["status"] == "completed" assert event["result"]["returncode"] == 0 assert event["result"]["prediction_count"] == 1 assert event["result"]["stdout_bytes"] > 0 def test_codex_exec_predictor_timeout_falls_back_to_hold(tmp_path): def fake_run(command, **kwargs): raise subprocess.TimeoutExpired(command, timeout=kwargs["timeout"]) predictor = CodexExecPredictor(timeout_seconds=1, cwd=tmp_path, run_command=fake_run) predictions = predictor.predict({"symbols": ["BTCUSDT"], "market": []}, allowed_symbols=("BTCUSDT",)) assert [prediction.symbol for prediction in predictions] == ["BTCUSDT"] assert predictions[0].action == "HOLD" assert predictions[0].reason == "codex exec 超时" def test_codex_exec_predictor_logs_timeout(tmp_path): def fake_run(command, **kwargs): raise subprocess.TimeoutExpired(command, timeout=kwargs["timeout"]) predictor = CodexExecPredictor(timeout_seconds=1, cwd=tmp_path, run_command=fake_run) predictor.predict({"symbols": ["BTCUSDT"], "market": []}, allowed_symbols=("BTCUSDT",)) log_text = (tmp_path / "logs" / "ai_trader_llm.log").read_text(encoding="utf-8") assert "Codex Exec AI Trader Call" in log_text assert "Allowed symbols: BTCUSDT" in log_text assert "Timeout seconds: 1" in log_text assert "Status: timeout" in log_text assert "Return code: -" in log_text assert "Error:" in log_text assert "codex exec timed out" in log_text def test_codex_exec_predictor_rotates_call_log_by_size(tmp_path): call_count = 0 def fake_run(command, **kwargs): nonlocal call_count call_count += 1 return subprocess.CompletedProcess( command, 0, stdout=json.dumps( { "predictions": [ { "symbol": "BTCUSDT", "action": "HOLD", "confidence": "0.30", "reason": f"第{call_count}次调用", } ] } ), stderr="", ) predictor = CodexExecPredictor( timeout_seconds=1, cwd=tmp_path, run_command=fake_run, log_max_bytes=1, log_backup_count=1, ) predictor.predict({"symbols": ["BTCUSDT"], "market": []}, allowed_symbols=("BTCUSDT",)) predictor.predict({"symbols": ["BTCUSDT"], "market": []}, allowed_symbols=("BTCUSDT",)) log_path = tmp_path / "logs" / "ai_trader_llm.log" backup_path = tmp_path / "logs" / "ai_trader_llm.log.1" assert "第2次调用" in log_path.read_text(encoding="utf-8") assert "第1次调用" in backup_path.read_text(encoding="utf-8") assert not (tmp_path / "logs" / "ai_trader_llm.log.2").exists() class FakeAiExchange: def __init__(self, base_free=Decimal("0"), balances=None, locked_balances=None): self.balances = {"BTC": base_free} if balances: self.balances.update(balances) self.locked_balances = locked_balances or {} self.calls = [] self.bought = False self.sold = False def exchange_info(self, symbol): self.calls.append(("exchange_info", symbol)) return SymbolMetadata( symbol=symbol, base_asset=symbol.removesuffix("USDT"), quote_asset="USDT", min_notional=Decimal("5"), min_qty=Decimal("0.00001"), step_size=Decimal("0.00001"), tick_size=Decimal("0.01"), min_price=Decimal("0.01"), ) def account(self): self.calls.append(("account",)) balances = {"USDT": Balance("USDT", Decimal("50"))} balances.update( { asset: Balance(asset, free, self.locked_balances.get(asset, Decimal("0"))) for asset, free in self.balances.items() } ) return AccountSnapshot( balances ) def symbol_order_book_ticker(self, symbol): self.calls.append(("symbol_order_book_ticker", symbol)) return { "symbol": symbol, "bidPrice": "104.99", "bidQty": "200", "askPrice": "105", "askQty": "180", } def depth(self, symbol, limit=20): self.calls.append(("depth", symbol, limit)) return { "lastUpdateId": 1, "bids": [["104", "10"], ["103", "20"]], "asks": [["106", "8"], ["107", "12"]], } def ticker_24hr(self, symbol): self.calls.append(("ticker_24hr", symbol)) return { "symbol": symbol, "priceChange": "2", "priceChangePercent": "1.94", "weightedAvgPrice": "103", "highPrice": "110", "lowPrice": "95", "volume": "1234", "quoteVolume": "127102", "count": 456, } def avg_price(self, symbol): self.calls.append(("avg_price", symbol)) return {"mins": 5, "price": "104.5"} def aggregate_trades(self, symbol, limit=500): self.calls.append(("aggregate_trades", symbol, limit)) return [ {"a": 1, "p": "100", "q": "2", "T": 1000, "m": False}, {"a": 2, "p": "110", "q": "1", "T": 2000, "m": True}, {"a": 3, "p": "120", "q": "0.5", "T": 3000, "m": False}, ] def klines(self, symbol, interval="1h", limit=6): self.calls.append(("klines", symbol, interval, limit)) close_offsets = {"1h": 0, "4h": 100, "1d": 200} offset = close_offsets.get(interval, 300) return [ [index, "100", "101", "99", str(offset + index), str(index * 10), index + 1] for index in range(1, limit + 1) ] 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, "executedQty": "0.04761", "cummulativeQuoteQty": str(quote_amount), } 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)) self.sold = True return { "symbol": symbol, "side": "SELL", "type": "MARKET", "status": "FILLED", "clientOrderId": client_order_id, "executedQty": str(quantity), } class LockedProtectiveStopAiExchange(FakeAiExchange): def __init__(self): super().__init__(base_free=Decimal("0"), locked_balances={"BTC": Decimal("0.123456")}) self.stop_client_order_id = "bqab-stop-existing" def open_orders(self, symbol): self.calls.append(("open_orders", symbol)) return [ { "symbol": symbol, "side": "SELL", "type": "STOP_LOSS_LIMIT", "status": "NEW", "clientOrderId": self.stop_client_order_id, } ] def cancel_order(self, symbol, orig_client_order_id): self.calls.append(("cancel_order", symbol, orig_client_order_id)) if orig_client_order_id == self.stop_client_order_id: self.balances["BTC"] = self.locked_balances["BTC"] self.locked_balances["BTC"] = Decimal("0") return { "symbol": symbol, "side": "SELL", "type": "STOP_LOSS_LIMIT", "status": "CANCELED", "clientOrderId": orig_client_order_id, } class LockedProtectiveOcoAiExchange(LockedProtectiveStopAiExchange): def __init__(self): super().__init__() self.take_profit_client_order_id = "bqab-tp-existing" self._canceled_ids: set[str] = set() def open_orders(self, symbol): self.calls.append(("open_orders", symbol)) orders = [] if self.take_profit_client_order_id not in self._canceled_ids: orders.append( { "symbol": symbol, "side": "SELL", "type": "LIMIT_MAKER", "status": "NEW", "clientOrderId": self.take_profit_client_order_id, } ) if self.stop_client_order_id not in self._canceled_ids: orders.append( { "symbol": symbol, "side": "SELL", "type": "STOP_LOSS_LIMIT", "status": "NEW", "clientOrderId": self.stop_client_order_id, } ) return orders def cancel_order(self, symbol, orig_client_order_id): self.calls.append(("cancel_order", symbol, orig_client_order_id)) self._canceled_ids.add(orig_client_order_id) if {self.take_profit_client_order_id, self.stop_client_order_id}.issubset(self._canceled_ids): self.balances["BTC"] = self.locked_balances["BTC"] self.locked_balances["BTC"] = Decimal("0") return { "symbol": symbol, "side": "SELL", "status": "CANCELED", "clientOrderId": orig_client_order_id, } class FailingCancelProtectiveStopAiExchange(LockedProtectiveStopAiExchange): def cancel_order(self, symbol, orig_client_order_id): self.calls.append(("cancel_order", symbol, orig_client_order_id)) raise RuntimeError("cancel rejected") class RallyAiExchange(FakeAiExchange): def ticker_24hr(self, symbol): self.calls.append(("ticker_24hr", symbol)) return { "symbol": symbol, "priceChange": "10.5", "priceChangePercent": "10.00", "weightedAvgPrice": "100", "highPrice": "115", "lowPrice": "95", "volume": "1234", "quoteVolume": "127102", "count": 456, } class ManualOpenSellOrderAiExchange(FakeAiExchange): def open_orders(self, symbol): self.calls.append(("open_orders", symbol)) return [ { "symbol": symbol, "side": "SELL", "type": "LIMIT", "status": "NEW", "clientOrderId": "manual-sell", } ] def cancel_order(self, symbol, orig_client_order_id): self.calls.append(("cancel_order", symbol, orig_client_order_id)) raise AssertionError("manual sell orders must not be canceled by AI sell") class WideSpreadAiExchange(FakeAiExchange): def symbol_order_book_ticker(self, symbol): self.calls.append(("symbol_order_book_ticker", symbol)) return { "symbol": symbol, "bidPrice": "99.9", "bidQty": "200", "askPrice": "100.1", "askQty": "180", } class ProtectedOrderAiExchange(FakeAiExchange): def open_orders(self, symbol): self.calls.append(("open_orders", symbol)) return [ { "symbol": symbol, "side": "SELL", "type": "STOP_LOSS_LIMIT", "status": "NEW", } ] def rich_candle(index: int, close: Decimal, volume: Decimal = Decimal("100")) -> Candle: return Candle( symbol="BTCUSDT", open_time=index, close_time=index + 1, open=close - Decimal("1"), high=close + Decimal("2"), low=close - Decimal("2"), close=close, volume=volume, quote_volume=volume * close, trade_count=index + 10, taker_buy_base_volume=volume / Decimal("2"), taker_buy_quote_volume=(volume * close) / Decimal("2"), ) def expected_wilder_rsi(values: list[Decimal], period: int = 14) -> Decimal: changes = [current - previous for previous, current in zip(values[:-1], values[1:])] gains = [max(change, Decimal("0")) for change in changes] losses = [abs(min(change, Decimal("0"))) for change in changes] average_gain = sum(gains[:period], Decimal("0")) / Decimal(period) average_loss = sum(losses[:period], Decimal("0")) / Decimal(period) for gain, loss in zip(gains[period:], losses[period:]): average_gain = ((average_gain * Decimal(period - 1)) + gain) / Decimal(period) average_loss = ((average_loss * Decimal(period - 1)) + loss) / Decimal(period) if average_loss == 0: return Decimal("100") relative_strength = average_gain / average_loss return Decimal("100") - (Decimal("100") / (Decimal("1") + relative_strength)) def expected_wilder_atr(candles: list[Candle], period: int = 14) -> Decimal: true_ranges = [ max( current.high - current.low, abs(current.high - previous.close), abs(current.low - previous.close), ) for previous, current in zip(candles[:-1], candles[1:]) ] atr = sum(true_ranges[:period], Decimal("0")) / Decimal(period) for true_range in true_ranges[period:]: atr = ((atr * Decimal(period - 1)) + true_range) / Decimal(period) return atr def test_timeframe_features_include_ohlcv_indicators_volume_and_breakout(): candles = [ rich_candle(index, Decimal(index), Decimal("100") + Decimal(index)) for index in range(1, 31) ] features = _timeframe_features(candles) assert features["opens"][0] == "0" assert features["highs"][-1] == "32" assert features["lows"][-1] == "28" assert features["closes"][-1] == "30" assert features["volumes"][-1] == "130" assert features["quote_volumes"][-1] == "3900" assert features["trade_counts"][-1] == 40 assert features["taker_buy_base_volumes"][-1] == "65" assert features["taker_buy_quote_volumes"][-1] == "1950" assert features["indicators"]["sma_7"] == "27" assert features["indicators"]["sma_20"] == "20.5" assert features["indicators"]["ema_12"] is not None assert features["indicators"]["ema_26"] is not None assert features["indicators"]["rsi_14"] == "100" assert features["indicators"]["macd"] is not None assert features["indicators"]["macd_signal"] is not None assert features["indicators"]["macd_histogram"] is not None assert features["indicators"]["atr_14"] == "4" assert features["volume"]["latest_volume"] == "130" assert features["volume"]["average_volume_20"] == "119.5" assert features["volume"]["volume_ratio_20"] is not None assert features["range"]["high_20"] == "32" assert features["range"]["low_20"] == "9" assert features["range"]["breakout_up_20"] is False assert features["range"]["breakdown_down_20"] is False def test_timeframe_features_use_wilder_rsi_and_atr(): closes = [ Decimal("44"), Decimal("45"), Decimal("44.5"), Decimal("46"), Decimal("45.25"), Decimal("47"), Decimal("46.2"), Decimal("48.1"), Decimal("47.4"), Decimal("49.5"), Decimal("48.8"), Decimal("50.2"), Decimal("49.7"), Decimal("51.3"), Decimal("50.6"), Decimal("52.4"), Decimal("51.9"), Decimal("53.5"), ] candles = [ Candle( symbol="BTCUSDT", open_time=index, close_time=index + 1, open=close - Decimal("0.3"), high=close + (Decimal("1.2") if index % 2 else Decimal("2.1")), low=close - (Decimal("0.8") if index % 3 else Decimal("1.7")), close=close, volume=Decimal("100") + Decimal(index), quote_volume=(Decimal("100") + Decimal(index)) * close, trade_count=index, taker_buy_base_volume=Decimal("50"), taker_buy_quote_volume=Decimal("50") * close, ) for index, close in enumerate(closes, start=1) ] features = _timeframe_features(candles) assert Decimal(features["indicators"]["rsi_14"]) == expected_wilder_rsi(closes) assert Decimal(features["indicators"]["atr_14"]) == expected_wilder_atr(candles) def test_timeframe_features_return_nulls_for_insufficient_history(): features = _timeframe_features([rich_candle(1, Decimal("10"))]) assert features["indicators"]["sma_7"] is None assert features["indicators"]["rsi_14"] is None assert features["indicators"]["macd"] is None assert features["indicators"]["atr_14"] is None assert features["range"]["breakout_up_20"] is False assert features["range"]["breakdown_down_20"] is False class StaticPredictor: def __init__(self, prediction): self.prediction = prediction self.contexts = [] def predict(self, context, allowed_symbols): self.contexts.append((context, allowed_symbols)) return self.prediction class NoExchangeCalls: def __getattr__(self, name): raise AssertionError(f"exchange method {name} should not be called") def ai_settings(tmp_path: Path, enabled=True, mode=TradingMode.LIVE, **overrides) -> Settings: values = { "_env_file": None, "mode": mode, "api_key": "key", "api_secret": SecretStr("secret"), "database_path": tmp_path / "audit.sqlite3", "live_ai_trader_enabled": enabled, "ai_trader_symbols": ("BTCUSDT",), "ai_trader_min_confidence": Decimal("0.70"), "ai_trader_ws_enabled": False, "auto_buy_symbols": ("BTCUSDT",), "auto_buy_quote_amount": Decimal("5"), "auto_buy_daily_limit": 5, "auto_buy_min_score": Decimal("4"), "auto_sell_symbols": ("BTCUSDT",), "auto_sell_daily_limit": 5, "auto_sell_min_score": Decimal("4"), } values.update(overrides) return Settings(**values) def test_run_live_ai_trade_once_safety_gates_short_circuit_before_exchange_calls(tmp_path): result = run_live_ai_trade_once( ai_settings(tmp_path, enabled=False), exchange_client=NoExchangeCalls(), predictor=StaticPredictor([AiTradePrediction("BTCUSDT", "BUY", Decimal("0.9"), "up")]), ) assert result.status == "rejected" assert result.action == "HOLD" assert result.reason == "AI实盘交易开关未开启" [run] = Storage(tmp_path / "audit.sqlite3").list_ai_trade_runs() assert run["id"] == result.run_id assert run["status"] == "rejected" assert run["action"] == "HOLD" assert run["reason"] == "AI实盘交易开关未开启" def test_run_live_ai_trade_once_hold_prediction_records_no_action(tmp_path): fake = FakeAiExchange() predictor = StaticPredictor([AiTradePrediction("BTCUSDT", "HOLD", Decimal("0.3"), "BTC震荡")]) result = run_live_ai_trade_once(ai_settings(tmp_path), exchange_client=fake, predictor=predictor) assert result.status == "no_action" assert result.symbol == "" assert result.action == "HOLD" assert result.reason == "所有交易对均为 HOLD" assert result.run_id == 1 assert not fake.bought assert not fake.sold assert predictor.contexts[0][1] == ("BTCUSDT",) [run] = Storage(tmp_path / "audit.sqlite3").list_ai_trade_runs() assert run["id"] == result.run_id assert run["status"] == "no_action" assert run["action"] == "HOLD" assert run["confidence"] == "0.3" assert run["reason"] == "BTC震荡" assert json.loads(run["raw_json"]) == { "action": "HOLD", "confidence": "0.3", "reason": "BTC震荡", "symbol": "BTCUSDT", } def test_run_live_ai_trade_batch_once_sends_configured_timeframes_to_predictor(tmp_path): fake = FakeAiExchange() predictor = StaticPredictor([AiTradePrediction("BTCUSDT", "HOLD", Decimal("0.3"), "BTC震荡")]) settings = ai_settings(tmp_path, ai_trader_klines=(("1h", 24), ("4h", 30), ("1d", 14))) result = run_live_ai_trade_batch_once(settings, exchange_client=fake, predictor=predictor) assert result.status == "no_action" context, allowed_symbols = predictor.contexts[0] assert allowed_symbols == ("BTCUSDT",) assert context["symbols"] == ["BTCUSDT"] assert "balances" not in context [market] = context["market"] assert market["symbol"] == "BTCUSDT" assert market["bid"] == "104.99" assert market["ask"] == "105" assert "open_sell_orders" not in market assert market["execution"]["quote_free"] == "50" assert market["execution"]["base_free"] == "0" assert set(market["timeframes"]) == {"1h", "4h", "1d"} assert market["timeframes"]["1h"]["closes"] == [str(index) for index in range(51, 75)] assert market["timeframes"]["1h"]["volumes"] == [str(index * 10) for index in range(51, 75)] assert market["timeframes"]["4h"]["closes"] == [str(100 + index) for index in range(51, 81)] assert market["timeframes"]["4h"]["volumes"] == [str(index * 10) for index in range(51, 81)] assert market["timeframes"]["1d"]["closes"] == [str(200 + index) for index in range(51, 65)] assert market["timeframes"]["1d"]["volumes"] == [str(index * 10) for index in range(51, 65)] assert market["timeframes"]["1h"]["indicator_metadata"] == { "display_candle_count": 24, "calculation_candle_count": 74, "warmup_candle_count": 50, "warmup_requested": 50, "rsi_method": "wilder", "atr_method": "wilder", "ema_warmup_applied": True, } assert market["timeframes"]["1d"]["indicator_metadata"] == { "display_candle_count": 14, "calculation_candle_count": 64, "warmup_candle_count": 50, "warmup_requested": 50, "rsi_method": "wilder", "atr_method": "wilder", "ema_warmup_applied": True, } assert ("klines", "BTCUSDT", "1h", 74) in fake.calls assert ("klines", "BTCUSDT", "4h", 80) in fake.calls assert ("klines", "BTCUSDT", "1d", 64) in fake.calls def test_run_live_ai_trade_once_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 [AiTradePrediction("BTCUSDT", "HOLD", Decimal("0.3"), "BTC震荡")] monkeypatch.setattr("binance_quant.ai_trader.CodexExecPredictor", CapturingPredictor) settings = ai_settings(tmp_path, ai_trader_codex_command=r"D:\Program Files\nodejs\codex.cmd") result = run_live_ai_trade_once(settings, exchange_client=FakeAiExchange()) assert result.status == "no_action" assert created["codex_command"] == r"D:\Program Files\nodejs\codex.cmd" def test_run_live_ai_trade_once_passes_enriched_market_features_to_ai_context(tmp_path): fake = FakeAiExchange() predictor = StaticPredictor([AiTradePrediction("BTCUSDT", "HOLD", Decimal("0.3"), "BTC震荡")]) run_live_ai_trade_once(ai_settings(tmp_path), exchange_client=fake, predictor=predictor, now_ms=3500) context = predictor.contexts[0][0] assert "balances" not in context [market] = context["market"] hourly = market["timeframes"]["1h"] assert hourly["opens"] assert hourly["highs"] assert hourly["lows"] assert hourly["closes"] assert hourly["quote_volumes"] assert hourly["trade_counts"] assert hourly["indicators"]["sma_7"] is not None assert "volume_ratio_20" in hourly["volume"] assert "distance_from_high_20_bps" in hourly["range"] assert "breakout_up_20" in hourly["range"] assert market["ticker_24h"]["price_change_percent"] == "1.94" assert market["average_price"]["price"] == "104.5" assert market["average_price"]["mins"] == 5 assert market["liquidity"]["best_bid_qty"] == "200" assert market["liquidity"]["best_ask_qty"] == "180" assert market["liquidity"]["depth_limit"] == 20 assert market["liquidity"]["depth_bid_notional"] == "3100" assert market["liquidity"]["depth_ask_notional"] == "2132" assert market["liquidity"]["spread_rating"] == "tight" assert market["liquidity"]["liquidity_rating"] == "poor" assert market["summary"]["timeframes"]["1h"]["latest_close"] == "74" assert market["summary"]["timeframes"]["1h"]["rsi_14"] is not None assert market["execution"] == { "quote_asset": "USDT", "base_asset": "BTC", "quote_free": "50", "base_free": "0", "base_locked": "0", "base_notional_bid": "0", "base_notional_ask": "0", "min_notional": "5", "min_qty": "0.00001", "auto_buy_quote_amount": "5", "auto_buy_existing_base_value_limit": "5", "max_buy_spread_bps": "10", "max_sell_spread_bps": "10", "open_sell_orders": 0, "protective_stop_orders": 0, "non_protective_sell_orders": 0, } assert "open_sell_orders" not in market assert ("depth", "BTCUSDT", 20) in fake.calls assert ("ticker_24hr", "BTCUSDT") in fake.calls assert ("avg_price", "BTCUSDT") in fake.calls def test_run_live_ai_trade_once_passes_aggregate_trade_features_to_ai_context(tmp_path): fake = FakeAiExchange() predictor = StaticPredictor([AiTradePrediction("BTCUSDT", "HOLD", Decimal("0.3"), "BTC震荡")]) settings = ai_settings(tmp_path, ai_trader_agg_trades_limit=3) run_live_ai_trade_once(settings, exchange_client=fake, predictor=predictor, now_ms=3500) context = predictor.contexts[0][0] [market] = context["market"] aggregate_trades = market["aggregate_trades"] assert aggregate_trades["trade_count"] == 3 assert aggregate_trades["first_trade_id"] == 1 assert aggregate_trades["last_trade_id"] == 3 assert aggregate_trades["first_trade_time"] == 1000 assert aggregate_trades["last_trade_time"] == 3000 assert aggregate_trades["base_volume"] == "3.5" assert aggregate_trades["quote_volume"] == "370" assert aggregate_trades["buy_base_volume"] == "2.5" assert aggregate_trades["buy_quote_volume"] == "260" assert aggregate_trades["sell_base_volume"] == "1" assert aggregate_trades["sell_quote_volume"] == "110" assert aggregate_trades["taker_buy_sell_ratio"] == "2.363636363636363636363636364" assert aggregate_trades["vwap"] == "105.7142857142857142857142857" assert aggregate_trades["large_trade_count"] == 1 assert aggregate_trades["large_trade_quote_sum"] == "200" assert aggregate_trades["largest_trade_quote"] == "200" assert aggregate_trades["status"] == "ok" assert aggregate_trades["last_trade_age_ms"] == 500 assert ("aggregate_trades", "BTCUSDT", 3) in fake.calls def test_run_live_ai_trade_once_marks_unavailable_aggregate_trade_features(tmp_path): class AggregateTradeFailureExchange(FakeAiExchange): def aggregate_trades(self, symbol, limit=500): self.calls.append(("aggregate_trades", symbol, limit)) raise RuntimeError("temporary market data error") fake = AggregateTradeFailureExchange() predictor = StaticPredictor([AiTradePrediction("BTCUSDT", "HOLD", Decimal("0.3"), "BTC震荡")]) run_live_ai_trade_once(ai_settings(tmp_path), exchange_client=fake, predictor=predictor) context = predictor.contexts[0][0] [market] = context["market"] assert market["aggregate_trades"]["status"] == "unavailable" assert market["aggregate_trades"]["trade_count"] == 0 def test_run_live_ai_trade_once_passes_realtime_features_to_ai_context(tmp_path, monkeypatch): fake = FakeAiExchange() predictor = StaticPredictor([AiTradePrediction("BTCUSDT", "HOLD", Decimal("0.3"), "BTC震荡")]) async def fake_sample_realtime_market(symbols, *, stream_base_url, sample_seconds): assert symbols == ("BTCUSDT",) assert stream_base_url == "wss://stream.binance.com:9443/ws" assert sample_seconds == 2 return { "BTCUSDT": { "status": "ok", "sample_seconds": 2, "bookticker_updates": 1, "depth_updates": 1, "agg_trade_updates": 2, "spread_avg_bps": "1", } } monkeypatch.setattr("binance_quant.ai_trader.sample_realtime_market", fake_sample_realtime_market) settings = ai_settings( tmp_path, stream_base_url="wss://stream.binance.com:9443/ws", ai_trader_ws_enabled=True, ai_trader_ws_sample_seconds=2, ) run_live_ai_trade_once(settings, exchange_client=fake, predictor=predictor) context = predictor.contexts[0][0] [market] = context["market"] assert market["realtime"]["status"] == "ok" assert market["realtime"]["sample_seconds"] == 2 assert market["realtime"]["agg_trade_updates"] == 2 assert market["realtime"]["spread_avg_bps"] == "1" def test_run_live_ai_trade_once_marks_realtime_disabled_when_configured(tmp_path, monkeypatch): fake = FakeAiExchange() predictor = StaticPredictor([AiTradePrediction("BTCUSDT", "HOLD", Decimal("0.3"), "BTC震荡")]) async def fail_if_called(*args, **kwargs): raise AssertionError("websocket sampler should not run when disabled") monkeypatch.setattr("binance_quant.ai_trader.sample_realtime_market", fail_if_called) settings = ai_settings(tmp_path, ai_trader_ws_enabled=False) run_live_ai_trade_once(settings, exchange_client=fake, predictor=predictor) context = predictor.contexts[0][0] [market] = context["market"] assert market["realtime"]["status"] == "disabled" assert market["realtime"]["bookticker_updates"] == 0 def test_run_live_ai_trade_once_rejects_low_confidence_prediction(tmp_path): fake = FakeAiExchange() predictor = StaticPredictor([AiTradePrediction("BTCUSDT", "BUY", Decimal("0.69"), "BTC上涨但置信度不足")]) result = run_live_ai_trade_once(ai_settings(tmp_path), exchange_client=fake, predictor=predictor) assert result.status == "rejected" assert result.action == "BUY" assert result.confidence == Decimal("0.69") assert result.reason == "AI: BTC上涨但置信度不足; 风控: AI置信度低于最小阈值（当前 0.69，最小 0.70）" assert result.run_id == 1 assert not fake.bought assert not fake.sold [run] = Storage(tmp_path / "audit.sqlite3").list_ai_trade_runs() assert run["id"] == result.run_id assert run["symbol"] == "BTCUSDT" assert run["action"] == "BUY" assert run["confidence"] == "0.69" assert run["status"] == "rejected" assert run["reason"] == "AI: BTC上涨但置信度不足; 风控: AI置信度低于最小阈值（当前 0.69，最小 0.70）" def test_run_live_ai_trade_once_rejected_buy_reason_includes_risk_values(tmp_path): fake = WideSpreadAiExchange(base_free=Decimal("0")) predictor = StaticPredictor([AiTradePrediction("BTCUSDT", "BUY", Decimal("0.88"), "BTC上行动能增强")]) result = run_live_ai_trade_once(ai_settings(tmp_path), exchange_client=fake, predictor=predictor) assert result.status == "rejected" assert result.action == "BUY" assert result.reason == "AI: BTC上行动能增强; 风控: 买卖价差超过上限（当前 20，最大 10）" assert not fake.bought [run] = Storage(tmp_path / "audit.sqlite3").list_ai_trade_runs() assert run["reason"] == "AI: BTC上行动能增强; 风控: 买卖价差超过上限（当前 20，最大 10）" def test_run_live_ai_trade_once_rejects_buy_after_sharp_24h_rally(tmp_path): fake = RallyAiExchange(base_free=Decimal("0")) predictor = StaticPredictor([AiTradePrediction("BTCUSDT", "BUY", Decimal("0.88"), "BTC上行动能增强")]) result = run_live_ai_trade_once(ai_settings(tmp_path), exchange_client=fake, predictor=predictor) assert result.status == "rejected" assert result.action == "BUY" assert result.reason == "AI: BTC上行动能增强; 风控: 24小时涨幅过高（当前 10.00%，上限 10%）" assert not fake.bought assert not any(call[0] == "create_market_order" for call in fake.calls) [run] = Storage(tmp_path / "audit.sqlite3").list_ai_trade_runs() assert run["reason"] == "AI: BTC上行动能增强; 风控: 24小时涨幅过高（当前 10.00%，上限 10%）" def test_run_live_ai_trade_once_existing_base_reason_uses_configured_limit(tmp_path): fake = FakeAiExchange(base_free=Decimal("0.28540565")) predictor = StaticPredictor([AiTradePrediction("BTCUSDT", "BUY", Decimal("0.88"), "BTC上行动能增强")]) settings = ai_settings( tmp_path, auto_buy_quote_amount=Decimal("5"), auto_buy_existing_base_value_limit=Decimal("20"), ) result = run_live_ai_trade_once(settings, exchange_client=fake, predictor=predictor) assert result.status == "rejected" assert result.action == "BUY" assert result.reason == "AI: BTC上行动能增强; 风控: 已持有基础资产（当前价值 29.96759325，限制 20）" assert not fake.bought audit = Storage(tmp_path / "audit.sqlite3") [run] = audit.list_ai_trade_runs() assert run["reason"] == "AI: BTC上行动能增强; 风控: 已持有基础资产（当前价值 29.96759325，限制 20）" [risk_event] = [event for event in read_audit_events(tmp_path) if event["event"] == "ai_trader.risk"] assert risk_event["result"]["quote_amount"] == "5" assert risk_event["result"]["existing_base_value_limit"] == "20" def test_run_live_ai_trade_once_buy_ignores_existing_protective_order(tmp_path): fake = ProtectedOrderAiExchange(base_free=Decimal("0")) predictor = StaticPredictor([AiTradePrediction("BTCUSDT", "BUY", Decimal("0.88"), "BTC上行动能增强")]) result = run_live_ai_trade_once(ai_settings(tmp_path), exchange_client=fake, predictor=predictor) assert result.status == "protected" assert result.action == "BUY" assert result.reason == "AI: BTC上行动能增强; 保护性OCO止盈止损单已接受" assert fake.bought audit = Storage(tmp_path / "audit.sqlite3") [run] = audit.list_ai_trade_runs() assert run["status"] == "protected" assert run["reason"] == "AI: BTC上行动能增强; 保护性OCO止盈止损单已接受" [auto_buy_run] = audit.list_auto_buy_runs() assert auto_buy_run["status"] == "protected" assert auto_buy_run["reason"] == "AI: BTC上行动能增强; 保护性OCO止盈止损单已接受" def test_run_live_ai_trade_once_buy_prediction_places_protected_buy(tmp_path): fake = FakeAiExchange(base_free=Decimal("0")) predictor = StaticPredictor([AiTradePrediction("BTCUSDT", "BUY", Decimal("0.88"), "BTC上行动能增强")]) result = run_live_ai_trade_once(ai_settings(tmp_path), exchange_client=fake, predictor=predictor) assert result.status == "protected" assert result.symbol == "BTCUSDT" assert result.action == "BUY" assert fake.bought audit = Storage(tmp_path / "audit.sqlite3") [run] = audit.list_auto_buy_runs() assert run["status"] == "protected" assert "AI: BTC上行动能增强" in run["reason"] [ai_run] = audit.list_ai_trade_runs() assert ai_run["id"] == result.run_id assert ai_run["status"] == "protected" assert ai_run["symbol"] == "BTCUSDT" assert ai_run["action"] == "BUY" assert ai_run["auto_buy_run_id"] == run["id"] assert ai_run["auto_sell_run_id"] is None assert "保护性OCO止盈止损单已接受" in ai_run["reason"] def test_run_live_ai_trade_once_writes_context_prediction_risk_and_execution_audit_events(tmp_path): fake = FakeAiExchange(base_free=Decimal("0")) predictor = StaticPredictor([AiTradePrediction("BTCUSDT", "BUY", Decimal("0.88"), "BTC上行动能增强")]) result = run_live_ai_trade_once(ai_settings(tmp_path), exchange_client=fake, predictor=predictor) assert result.status == "protected" events = read_audit_events(tmp_path) events_by_name = {event["event"]: event for event in events} assert events_by_name["ai_trader.context"]["params"] == {"symbols": ["BTCUSDT"]} context_result = events_by_name["ai_trader.context"]["result"] assert context_result["market"][0]["symbol"] == "BTCUSDT" assert context_result["market"][0]["spread_bps"] == "0.9524" assert "balances" not in context_result assert "execution" not in context_result["market"][0] assert "open_sell_orders" not in context_result["market"][0] predictions_result = events_by_name["ai_trader.predictions"]["result"] assert predictions_result == [ {"symbol": "BTCUSDT", "action": "BUY", "confidence": "0.88", "reason": "BTC上行动能增强"} ] risk_event = events_by_name["ai_trader.risk"] assert risk_event["params"]["action"] == "BUY" assert risk_event["params"]["symbol"] == "BTCUSDT" assert risk_event["formula"]["spread_bps"] == "(ask - bid) / ((ask + bid) / 2) * 10000" 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["ai_trader.execution"] assert execution_event["params"] == {"action": "BUY", "symbol": "BTCUSDT", "ai_run_id": result.run_id} assert execution_event["result"]["status"] == "protected" assert execution_event["result"]["reason"] == "AI: BTC上行动能增强; 保护性OCO止盈止损单已接受" def test_run_live_ai_trade_once_sell_prediction_sells_holding(tmp_path): fake = FakeAiExchange(base_free=Decimal("0.123456")) predictor = StaticPredictor([AiTradePrediction("BTCUSDT", "SELL", Decimal("0.91"), "BTC下行动能增强")]) result = run_live_ai_trade_once(ai_settings(tmp_path), exchange_client=fake, predictor=predictor) assert result.status == "sold" assert result.symbol == "BTCUSDT" assert result.action == "SELL" assert fake.sold sell_call = next(call for call in fake.calls if call[0] == "create_market_sell_quantity") assert sell_call[1:3] == ("BTCUSDT", Decimal("0.12345")) audit = Storage(tmp_path / "audit.sqlite3") [run] = audit.list_auto_sell_runs() assert run["status"] == "sold" assert "AI: BTC下行动能增强" in run["reason"] [ai_run] = audit.list_ai_trade_runs() assert ai_run["id"] == result.run_id assert ai_run["status"] == "sold" assert ai_run["symbol"] == "BTCUSDT" assert ai_run["action"] == "SELL" assert ai_run["auto_buy_run_id"] is None assert ai_run["auto_sell_run_id"] == run["id"] assert "市价卖出已成交" in ai_run["reason"] def test_run_live_ai_trade_once_rejects_below_cost_sell_before_minimum_hold(tmp_path): audit = Storage(tmp_path / "audit.sqlite3") audit.initialize() audit.record_spot_trade_fill( "BTCUSDT", _spot_buy(1, price="110", qty="0.12345", quote_qty="13.5795", time_ms=1_700_000_000_000), ) fake = FakeAiExchange(base_free=Decimal("0.123456")) predictor = StaticPredictor([AiTradePrediction("BTCUSDT", "SELL", Decimal("0.91"), "BTC下行动能增强")]) result = run_live_ai_trade_once( ai_settings(tmp_path, auto_sell_min_hold_seconds=3600), exchange_client=fake, predictor=predictor, now_ms=1_700_000_600_000, ) assert result.status == "rejected" assert result.symbol == "BTCUSDT" assert result.action == "SELL" assert result.reason == "AI: BTC下行动能增强; 风控: sell below cost before minimum hold" assert not fake.sold assert not any(call[0] == "create_market_sell_quantity" for call in fake.calls) ai_run = audit.list_ai_trade_runs()[0] assert ai_run["status"] == "rejected" [auto_sell_run] = audit.list_auto_sell_runs() assert auto_sell_run["status"] == "rejected" assert auto_sell_run["reason"] == "AI: BTC下行动能增强; 风控: sell below cost before minimum hold" def test_run_live_ai_trade_once_sell_cancels_bot_protective_stop_before_market_sell(tmp_path): fake = LockedProtectiveStopAiExchange() predictor = StaticPredictor([AiTradePrediction("BTCUSDT", "SELL", Decimal("0.91"), "BTC下行动能增强")]) result = run_live_ai_trade_once(ai_settings(tmp_path), exchange_client=fake, predictor=predictor) assert result.status == "sold" assert result.symbol == "BTCUSDT" assert result.action == "SELL" assert ("cancel_order", "BTCUSDT", "bqab-stop-existing") in fake.calls cancel_index = fake.calls.index(("cancel_order", "BTCUSDT", "bqab-stop-existing")) refreshed_account_index = next(index for index, call in enumerate(fake.calls[cancel_index + 1:], start=cancel_index + 1) if call == ("account",)) sell_index = next(index for index, call in enumerate(fake.calls) if call[0] == "create_market_sell_quantity") assert cancel_index < refreshed_account_index < sell_index sell_call = fake.calls[sell_index] assert sell_call[1:3] == ("BTCUSDT", Decimal("0.12345")) def test_run_live_ai_trade_once_sell_cancels_bot_protective_oco_before_market_sell(tmp_path): fake = LockedProtectiveOcoAiExchange() predictor = StaticPredictor([AiTradePrediction("BTCUSDT", "SELL", Decimal("0.91"), "BTC下行动能增强")]) result = run_live_ai_trade_once(ai_settings(tmp_path), exchange_client=fake, predictor=predictor) assert result.status == "sold" assert result.symbol == "BTCUSDT" assert result.action == "SELL" assert ("cancel_order", "BTCUSDT", "bqab-tp-existing") in fake.calls assert ("cancel_order", "BTCUSDT", "bqab-stop-existing") in fake.calls sell_index = next(index for index, call in enumerate(fake.calls) if call[0] == "create_market_sell_quantity") assert fake.calls.index(("cancel_order", "BTCUSDT", "bqab-tp-existing")) < sell_index assert fake.calls.index(("cancel_order", "BTCUSDT", "bqab-stop-existing")) < sell_index sell_call = fake.calls[sell_index] assert sell_call[1:3] == ("BTCUSDT", Decimal("0.12345")) def test_run_live_ai_trade_once_sell_still_rejects_manual_open_sell_order(tmp_path): fake = ManualOpenSellOrderAiExchange(base_free=Decimal("0.123456")) predictor = StaticPredictor([AiTradePrediction("BTCUSDT", "SELL", Decimal("0.91"), "BTC下行动能增强")]) result = run_live_ai_trade_once(ai_settings(tmp_path), exchange_client=fake, predictor=predictor) assert result.status == "rejected" assert result.action == "SELL" assert result.reason == "AI: BTC下行动能增强; 风控: 已有卖出挂单（当前 1）" assert not fake.sold assert not any(call[0] == "cancel_order" for call in fake.calls) def test_run_live_ai_trade_once_sell_does_not_submit_market_sell_when_protective_stop_cancel_fails(tmp_path): fake = FailingCancelProtectiveStopAiExchange() predictor = StaticPredictor([AiTradePrediction("BTCUSDT", "SELL", Decimal("0.91"), "BTC下行动能增强")]) result = run_live_ai_trade_once(ai_settings(tmp_path), exchange_client=fake, predictor=predictor) assert result.status == "rejected" assert result.action == "SELL" assert ("cancel_order", "BTCUSDT", "bqab-stop-existing") in fake.calls assert not fake.sold assert not any(call[0] == "create_market_sell_quantity" for call in fake.calls) def test_run_live_ai_trade_once_executes_best_action_from_all_configured_symbols(tmp_path): fake = FakeAiExchange( balances={ "BTC": Decimal("0"), "ETH": Decimal("0.23456"), } ) predictor = StaticPredictor( [ AiTradePrediction("BTCUSDT", "HOLD", Decimal("0.20"), "BTC方向不明确"), AiTradePrediction("ETHUSDT", "SELL", Decimal("0.91"), "ETH短线转弱"), ] ) settings = ai_settings( tmp_path, ai_trader_symbols=("BTCUSDT", "ETHUSDT"), auto_buy_symbols=("BTCUSDT", "ETHUSDT"), auto_sell_symbols=("BTCUSDT", "ETHUSDT"), ) result = run_live_ai_trade_once(settings, exchange_client=fake, predictor=predictor) assert result.status == "sold" assert result.symbol == "ETHUSDT" assert result.action == "SELL" assert fake.sold assert predictor.contexts[0][1] == ("BTCUSDT", "ETHUSDT") assert ("exchange_info", "BTCUSDT") in fake.calls assert ("exchange_info", "ETHUSDT") in fake.calls sell_call = next(call for call in fake.calls if call[0] == "create_market_sell_quantity") assert sell_call[1:3] == ("ETHUSDT", Decimal("0.23456")) ai_runs = Storage(tmp_path / "audit.sqlite3").list_ai_trade_runs() assert len(ai_runs) == 2 runs_by_symbol = {run["symbol"]: run for run in ai_runs} assert runs_by_symbol["BTCUSDT"]["status"] == "no_action" assert runs_by_symbol["BTCUSDT"]["action"] == "HOLD" assert runs_by_symbol["ETHUSDT"]["status"] == "sold" assert runs_by_symbol["ETHUSDT"]["action"] == "SELL" def test_run_live_ai_trade_batch_once_executes_multiple_approved_actions(tmp_path): fake = FakeAiExchange( balances={ "BTC": Decimal("0"), "ETH": Decimal("0.23456"), } ) predictor = StaticPredictor( [ AiTradePrediction("BTCUSDT", "BUY", Decimal("0.93"), "BTC转强"), AiTradePrediction("ETHUSDT", "SELL", Decimal("0.91"), "ETH转弱"), ] ) settings = ai_settings( tmp_path, ai_trader_symbols=("BTCUSDT", "ETHUSDT"), auto_buy_symbols=("BTCUSDT", "ETHUSDT"), auto_sell_symbols=("BTCUSDT", "ETHUSDT"), ai_trader_max_actions_per_run=5, ) result = run_live_ai_trade_batch_once(settings, exchange_client=fake, predictor=predictor) assert result.status == "completed" assert result.reason == "AI批量交易完成" assert result.prediction_count == 2 assert result.max_actions == 5 assert result.executed_count == 2 assert [(item.symbol, item.action, item.status) for item in result.results] == [ ("BTCUSDT", "BUY", "protected"), ("ETHUSDT", "SELL", "sold"), ] audit = Storage(tmp_path / "audit.sqlite3") ai_runs = audit.list_ai_trade_runs() assert len(ai_runs) == 2 runs_by_symbol = {run["symbol"]: run for run in ai_runs} assert runs_by_symbol["BTCUSDT"]["status"] == "protected" assert runs_by_symbol["ETHUSDT"]["status"] == "sold" assert len(audit.list_auto_buy_runs()) == 1 assert len(audit.list_auto_sell_runs()) == 1 def test_run_live_ai_trade_batch_once_returns_hold_predictions_at_bottom(tmp_path): fake = FakeAiExchange( balances={ "BTC": Decimal("0"), "ETH": Decimal("0.23456"), "BNB": Decimal("0.5"), } ) predictor = StaticPredictor( [ AiTradePrediction("BTCUSDT", "HOLD", Decimal("0.33"), "BTC震荡"), AiTradePrediction("ETHUSDT", "SELL", Decimal("0.91"), "ETH转弱"), AiTradePrediction("BNBUSDT", "HOLD", Decimal("0.42"), "BNB观望"), ] ) settings = ai_settings( tmp_path, ai_trader_symbols=("BTCUSDT", "ETHUSDT", "BNBUSDT"), auto_buy_symbols=("BTCUSDT", "ETHUSDT", "BNBUSDT"), auto_sell_symbols=("BTCUSDT", "ETHUSDT", "BNBUSDT"), ai_trader_max_actions_per_run=5, ) result = run_live_ai_trade_batch_once(settings, exchange_client=fake, predictor=predictor) assert result.status == "completed" assert result.prediction_count == 3 assert result.executed_count == 1 assert [(item.symbol, item.action, item.status) for item in result.results] == [ ("ETHUSDT", "SELL", "sold"), ("BTCUSDT", "HOLD", "no_action"), ("BNBUSDT", "HOLD", "no_action"), ] assert result.results[1].reason == "BTC震荡" assert result.results[2].reason == "BNB观望" def test_run_live_ai_trade_batch_once_skips_after_action_limit(tmp_path): fake = FakeAiExchange( balances={ "BTC": Decimal("0"), "ETH": Decimal("0.23456"), "BNB": Decimal("0.5"), } ) predictor = StaticPredictor( [ AiTradePrediction("BTCUSDT", "BUY", Decimal("0.93"), "BTC转强"), AiTradePrediction("ETHUSDT", "SELL", Decimal("0.91"), "ETH转弱"), AiTradePrediction("BNBUSDT", "SELL", Decimal("0.90"), "BNB转弱"), ] ) settings = ai_settings( tmp_path, ai_trader_symbols=("BTCUSDT", "ETHUSDT", "BNBUSDT"), auto_buy_symbols=("BTCUSDT", "ETHUSDT", "BNBUSDT"), auto_sell_symbols=("BTCUSDT", "ETHUSDT", "BNBUSDT"), ai_trader_max_actions_per_run=1, ) result = run_live_ai_trade_batch_once(settings, exchange_client=fake, predictor=predictor) assert result.status == "completed" assert result.executed_count == 1 assert [(item.symbol, item.action, item.status) for item in result.results] == [ ("BTCUSDT", "BUY", "protected"), ] ai_runs = Storage(tmp_path / "audit.sqlite3").list_ai_trade_runs() runs_by_symbol = {run["symbol"]: run for run in ai_runs} assert runs_by_symbol["BTCUSDT"]["status"] == "protected" assert runs_by_symbol["ETHUSDT"]["status"] == "skipped" assert runs_by_symbol["BNBUSDT"]["status"] == "skipped" assert "AI批量交易次数上限" in runs_by_symbol["ETHUSDT"]["reason"] assert "AI批量交易次数上限" in runs_by_symbol["BNBUSDT"]["reason"] def test_run_live_ai_trade_batch_once_refreshes_account_for_each_candidate(tmp_path): fake = FakeAiExchange( balances={ "BTC": Decimal("0"), "ETH": Decimal("0.23456"), } ) predictor = StaticPredictor( [ AiTradePrediction("BTCUSDT", "BUY", Decimal("0.93"), "BTC转强"), AiTradePrediction("ETHUSDT", "SELL", Decimal("0.91"), "ETH转弱"), ] ) settings = ai_settings( tmp_path, ai_trader_symbols=("BTCUSDT", "ETHUSDT"), auto_buy_symbols=("BTCUSDT", "ETHUSDT"), auto_sell_symbols=("BTCUSDT", "ETHUSDT"), ai_trader_max_actions_per_run=5, ) run_live_ai_trade_batch_once(settings, exchange_client=fake, predictor=predictor) assert sum(1 for call in fake.calls if call[0] == "account") >= 3 assert fake.calls.count(("exchange_info", "BTCUSDT")) >= 2 assert fake.calls.count(("exchange_info", "ETHUSDT")) >= 2 def _spot_buy(trade_id: int, *, price: str, qty: str, quote_qty: str, time_ms: int) -> dict: return { "id": trade_id, "orderId": trade_id + 100, "orderListId": -1, "price": price, "qty": qty, "quoteQty": quote_qty, "commission": "0", "commissionAsset": "BNB", "time": time_ms, "isBuyer": True, "isMaker": False, "isBestMatch": True, }