from decimal import Decimal from binance_quant import email_notify from binance_quant.ai_trader import AiTradeBatchExecutionResult, AiTradeExecutionResult from binance_quant.config import Settings from binance_quant.email_notify import ( EmailNotificationError, send_auto_buy_protection_failure_email, send_ai_trade_batch_email, send_ai_trade_email, ) from binance_quant.auto_buy import AutoBuyExecutionResult from binance_quant.futures_ai_trader import FuturesAiBatchResult from binance_quant.futures_models import FuturesExecutionResult class FakeSMTP: instances = [] def __init__(self, server, port, timeout): self.server = server self.port = port self.timeout = timeout self.started_tls = False self.login_args = None self.messages = [] FakeSMTP.instances.append(self) def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def starttls(self): self.started_tls = True def login(self, login, password): self.login_args = (login, password) def send_message(self, message): self.messages.append(message) def test_send_ai_trade_email_sends_result_summary(): FakeSMTP.instances = [] settings = Settings( _env_file=None, email_notifications_enabled=True, email_smtp_server="smtp.163.com", email_smtp_port=25, email_login="19176654177@163.com", email_sender_password="secret-token", email_recipient="xia_qingbo@163.com", ) result = AiTradeExecutionResult(33, "rejected", "BNBUSDT", "BUY", "weak signal", confidence="0.60") send_ai_trade_email(settings, result, smtp_factory=FakeSMTP) [smtp] = FakeSMTP.instances assert smtp.server == "smtp.163.com" assert smtp.port == 25 assert smtp.timeout == 20 assert smtp.login_args == ("19176654177@163.com", "secret-token") [message] = smtp.messages assert message["From"] == "19176654177@163.com" assert message["To"] == "xia_qingbo@163.com" assert "Binance AI trade rejected BUY" in message["Subject"] body = message.get_body(preferencelist=("plain",)).get_content() assert "Run ID: 33" in body assert "Status: rejected" in body assert "Symbol: BNBUSDT" in body assert "Action: BUY" in body assert "Confidence: 0.60" in body assert "Reason: weak signal" in body assert "secret-token" not in body def test_send_ai_trade_email_noops_when_disabled(): FakeSMTP.instances = [] settings = Settings(_env_file=None) send_ai_trade_email(settings, AiTradeExecutionResult(1, "no_action", "", "HOLD", "mixed"), smtp_factory=FakeSMTP) assert FakeSMTP.instances == [] def test_send_ai_trade_email_sanitizes_failures(): class FailingSMTP(FakeSMTP): def login(self, login, password): raise RuntimeError(f"auth failed for {password}") settings = Settings( _env_file=None, email_notifications_enabled=True, email_smtp_server="smtp.163.com", email_smtp_port=25, email_login="19176654177@163.com", email_sender_password="secret-token", email_recipient="xia_qingbo@163.com", ) try: send_ai_trade_email(settings, AiTradeExecutionResult(1, "no_action", "", "HOLD", "mixed"), smtp_factory=FailingSMTP) except EmailNotificationError as exc: message = str(exc) else: raise AssertionError("EmailNotificationError was not raised") assert "email notification failed" in message assert "secret-token" not in message def test_send_auto_buy_protection_failure_email_sends_result_summary(): FakeSMTP.instances = [] settings = Settings( _env_file=None, email_notifications_enabled=True, email_smtp_server="smtp.163.com", email_smtp_port=25, email_login="19176654177@163.com", email_sender_password="secret-token", email_recipient="xia_qingbo@163.com", ) result = AutoBuyExecutionResult(44, "rollback_failed", "BTCUSDT", "protective stop failed; rollback failed") send_auto_buy_protection_failure_email(settings, result, smtp_factory=FakeSMTP) [smtp] = FakeSMTP.instances assert smtp.login_args == ("19176654177@163.com", "secret-token") [message] = smtp.messages assert message["From"] == "19176654177@163.com" assert message["To"] == "xia_qingbo@163.com" assert "Binance auto-buy protection failure rollback_failed BTCUSDT" in message["Subject"] body = message.get_body(preferencelist=("plain",)).get_content() assert "Binance auto-buy protection failure" in body assert "Run ID: 44" in body assert "Status: rollback_failed" in body assert "Symbol: BTCUSDT" in body assert "Reason: protective stop failed; rollback failed" in body assert "secret-token" not in body def test_send_ai_trade_batch_email_sends_result_summary(): FakeSMTP.instances = [] settings = Settings( _env_file=None, email_notifications_enabled=True, email_smtp_server="smtp.163.com", email_smtp_port=25, email_login="19176654177@163.com", email_sender_password="secret-token", email_recipient="xia_qingbo@163.com", ) result = AiTradeBatchExecutionResult( status="completed", reason="AI批量交易完成", max_actions=5, prediction_count=3, executed_count=2, results=( AiTradeExecutionResult(1, "protected", "BTCUSDT", "BUY", "done", Decimal("0.93")), AiTradeExecutionResult(2, "sold", "ETHUSDT", "SELL", "done", Decimal("0.91")), AiTradeExecutionResult(3, "rejected", "BNBUSDT", "BUY", "risk rejected", Decimal("0.88")), ), ) send_ai_trade_batch_email(settings, result, smtp_factory=FakeSMTP) [smtp] = FakeSMTP.instances assert smtp.login_args == ("19176654177@163.com", "secret-token") [message] = smtp.messages assert message["From"] == "19176654177@163.com" assert message["To"] == "xia_qingbo@163.com" assert "Binance AI trade batch completed" in message["Subject"] body = message.get_body(preferencelist=("plain",)).get_content() assert "Status: completed" in body assert "Reason: AI批量交易完成" in body assert "Max actions: 5" in body assert "Executed actions: 2" in body assert "Predictions: 3" in body assert "| Run ID | Status" not in body assert "| Summary" in body assert "| Executed |" not in body assert "| Reason" in body assert "Run ID: 1; Status: protected; Symbol: BTCUSDT; Action: BUY; Confidence: 0.93; Executed: yes" in body assert "Run ID: 2; Status: sold; Symbol: ETHUSDT; Action: SELL; Confidence: 0.91" in body assert "Run ID: 3; Status: rejected; Symbol: BNBUSDT; Action: BUY; Confidence: 0.88; Executed: no" in body assert "secret-token" not in body assert message.is_multipart() html = message.get_body(preferencelist=("html",)).get_content() assert "Run ID" not in html assert ">Summary" in html assert ">Executed" not in html assert ">Reason" in html assert "Run ID: 1; Status: protected; Symbol: BTCUSDT; Action: BUY; Confidence: 0.93; Executed: yes" in html assert "Run ID: 3; Status: rejected; Symbol: BNBUSDT; Action: BUY; Confidence: 0.88; Executed: no" in html assert ">done" in html assert "secret-token" not in html def test_send_ai_trade_batch_email_places_hold_results_at_bottom(): FakeSMTP.instances = [] settings = Settings( _env_file=None, email_notifications_enabled=True, email_smtp_server="smtp.163.com", email_smtp_port=25, email_login="19176654177@163.com", email_sender_password="secret-token", email_recipient="xia_qingbo@163.com", ) result = AiTradeBatchExecutionResult( status="completed", reason="AI批量交易完成", max_actions=5, prediction_count=3, executed_count=1, results=( AiTradeExecutionResult(1, "no_action", "BTCUSDT", "HOLD", "BTC震荡", Decimal("0.33")), AiTradeExecutionResult(2, "sold", "ETHUSDT", "SELL", "ETH转弱", Decimal("0.91")), AiTradeExecutionResult(3, "no_action", "BNBUSDT", "HOLD", "BNB观望", Decimal("0.42")), ), ) send_ai_trade_batch_email(settings, result, smtp_factory=FakeSMTP) [smtp] = FakeSMTP.instances [message] = smtp.messages body = message.get_body(preferencelist=("plain",)).get_content() html = message.get_body(preferencelist=("html",)).get_content() action_index = body.index("Action results:") eth_index = body.index("Run ID: 2; Status: sold; Symbol: ETHUSDT; Action: SELL; Confidence: 0.91") hold_index = body.index("HOLD predictions:") btc_index = body.index("Run ID: 1; Status: no_action; Symbol: BTCUSDT; Action: HOLD; Confidence: 0.33") bnb_index = body.index("Run ID: 3; Status: no_action; Symbol: BNBUSDT; Action: HOLD; Confidence: 0.42") assert action_index < eth_index < hold_index < btc_index < bnb_index html_action_index = html.index("Action results") html_eth_index = html.index("Run ID: 2; Status: sold; Symbol: ETHUSDT; Action: SELL; Confidence: 0.91") html_hold_index = html.index("HOLD predictions") html_btc_index = html.index("Run ID: 1; Status: no_action; Symbol: BTCUSDT; Action: HOLD; Confidence: 0.33") html_bnb_index = html.index("Run ID: 3; Status: no_action; Symbol: BNBUSDT; Action: HOLD; Confidence: 0.42") assert html_action_index < html_eth_index < html_hold_index < html_btc_index < html_bnb_index def test_send_futures_ai_trade_batch_email_sends_summary_and_result_table(): FakeSMTP.instances = [] settings = Settings( _env_file=None, email_notifications_enabled=True, email_smtp_server="smtp.163.com", email_smtp_port=25, email_login="19176654177@163.com", email_sender_password="secret-token", email_recipient="xia_qingbo@163.com", ) result = FuturesAiBatchResult( status="completed", reason="AI合约批量交易完成", max_actions=5, prediction_count=3, executed_count=1, daily_realized_pnl=Decimal("-1.5"), total_isolated_margin=Decimal("5"), results=( FuturesExecutionResult( 11, "protected", "BTCUSDT", "OPEN_LONG", "AI: BTC趋势增强; 执行: protected", Decimal("0.88"), leverage=3, margin_amount=Decimal("5"), ), FuturesExecutionResult( 13, "rejected", "BNBUSDT", "CLOSE", "AI: BNB持仓风险升高; 风控: 交易规则与交易对不匹配", Decimal("0.75"), close_percent=100, ), FuturesExecutionResult( 12, "no_action", "ETHUSDT", "HOLD", "ETH震荡", Decimal("0.30"), ), ), ) email_notify.send_futures_ai_trade_batch_email(settings, result, smtp_factory=FakeSMTP) [smtp] = FakeSMTP.instances assert smtp.login_args == ("19176654177@163.com", "secret-token") [message] = smtp.messages assert "Binance futures AI trade batch completed" in message["Subject"] body = message.get_body(preferencelist=("plain",)).get_content() assert "Binance futures AI trade batch result" in body assert "Status: completed" in body assert "Daily net PnL: -1.5" in body assert "Total isolated margin: 5" in body assert "Action results:" in body assert "HOLD predictions:" in body assert body.index("Action results:") < body.index("HOLD predictions:") assert "| Summary" in body assert "| Reason" in body assert "| Run | Status" not in body assert "| Conf |" not in body assert "Run ID: 11; Status: protected; Symbol: BTCUSDT; Action: OPEN_LONG; Confidence: 0.88; Leverage: 3; Margin: 5; Close%: -" in body assert "protected" in body assert "BTCUSDT" in body assert "OPEN_LONG" in body assert "0.88" in body assert "Run ID: 13; Status: rejected; Symbol: BNBUSDT; Action: CLOSE; Confidence: 0.75; Leverage: -; Margin: -; Close%: 100" in body assert "交易规则与交易对不匹配" in body assert "metadata symbol mismatch" not in body assert "Run ID: 12; Status: no_action; Symbol: ETHUSDT; Action: HOLD; Confidence: 0.3; Leverage: -; Margin: -; Close%: -" in body assert "no_action" in body assert "ETHUSDT" in body assert "HOLD" in body assert "0.3" in body assert "secret-token" not in body html = message.get_body(preferencelist=("html",)).get_content() assert "Daily net PnL" in html assert ">Summary" in html assert ">Reason" in html assert ">Run" not in html assert ">Close%" not in html assert "Action results" in html assert "HOLD predictions" in html assert ">AI: BTC趋势增强; 执行: protected" in html assert ">AI: BNB持仓风险升高; 风控: 交易规则与交易对不匹配" in html assert "metadata symbol mismatch" not in html assert "secret-token" not in html def test_send_futures_ai_trade_batch_email_noops_when_disabled(): FakeSMTP.instances = [] settings = Settings(_env_file=None) result = FuturesAiBatchResult( status="no_action", reason="无合约动作", max_actions=5, prediction_count=0, executed_count=0, daily_realized_pnl=Decimal("0"), total_isolated_margin=Decimal("0"), results=(), ) email_notify.send_futures_ai_trade_batch_email(settings, result, smtp_factory=FakeSMTP) assert FakeSMTP.instances == [] def test_send_futures_ai_trade_batch_email_formats_decimal_without_scientific_notation(): FakeSMTP.instances = [] settings = Settings( _env_file=None, email_notifications_enabled=True, email_smtp_server="smtp.163.com", email_smtp_port=25, email_login="19176654177@163.com", email_sender_password="secret-token", email_recipient="xia_qingbo@163.com", ) result = FuturesAiBatchResult( status="no_action", reason="无合约动作", max_actions=5, prediction_count=1, executed_count=0, daily_realized_pnl=Decimal("0E-8"), total_isolated_margin=Decimal("0E-8"), results=( FuturesExecutionResult( 21, "no_action", "BTCUSDT", "HOLD", "BTC震荡", Decimal("0.3000"), margin_amount=Decimal("0E-8"), ), ), ) email_notify.send_futures_ai_trade_batch_email(settings, result, smtp_factory=FakeSMTP) [smtp] = FakeSMTP.instances [message] = smtp.messages body = message.get_body(preferencelist=("plain",)).get_content() html = message.get_body(preferencelist=("html",)).get_content() assert "Daily net PnL: 0" in body assert "Total isolated margin: 0" in body assert "0E-8" not in body assert "0.3000" not in body assert "0.3" in body assert ">0" in html assert "0E-8" not in html