from __future__ import annotations
from html import escape
import smtplib
from decimal import Decimal
from email.message import EmailMessage
from typing import Callable
from .auto_buy import AutoBuyExecutionResult
from .ai_trader import AiTradeBatchExecutionResult, AiTradeExecutionResult, is_ai_trade_executed_status
from .config import Settings
from .futures_ai_trader import FuturesAiBatchResult
class EmailNotificationError(RuntimeError):
pass
def send_ai_trade_email(
settings: Settings,
result: AiTradeExecutionResult,
smtp_factory: Callable[..., object] = smtplib.SMTP,
) -> None:
if not settings.email_notifications_enabled:
return
password = settings.email_sender_password.get_secret_value()
message = EmailMessage()
message["From"] = settings.email_login or ""
message["To"] = settings.email_recipient or ""
message["Subject"] = f"Binance AI trade {result.status} {result.action}"
message.set_content(_body(result))
try:
with smtp_factory(settings.email_smtp_server, settings.email_smtp_port, timeout=20) as smtp:
smtp.login(settings.email_login, password)
smtp.send_message(message)
except Exception as exc:
raise EmailNotificationError("email notification failed") from exc
def send_ai_trade_batch_email(
settings: Settings,
result: AiTradeBatchExecutionResult,
smtp_factory: Callable[..., object] = smtplib.SMTP,
) -> None:
if not settings.email_notifications_enabled:
return
password = settings.email_sender_password.get_secret_value()
message = EmailMessage()
message["From"] = settings.email_login or ""
message["To"] = settings.email_recipient or ""
message["Subject"] = f"Binance AI trade batch {result.status}"
message.set_content(_batch_body(result))
message.add_alternative(_batch_html_body(result), subtype="html")
try:
with smtp_factory(settings.email_smtp_server, settings.email_smtp_port, timeout=20) as smtp:
smtp.login(settings.email_login, password)
smtp.send_message(message)
except Exception as exc:
raise EmailNotificationError("email notification failed") from exc
def send_auto_buy_protection_failure_email(
settings: Settings,
result: AutoBuyExecutionResult,
smtp_factory: Callable[..., object] = smtplib.SMTP,
) -> None:
if not settings.email_notifications_enabled:
return
password = settings.email_sender_password.get_secret_value()
message = EmailMessage()
message["From"] = settings.email_login or ""
message["To"] = settings.email_recipient or ""
message["Subject"] = f"Binance auto-buy protection failure {result.status} {result.symbol or '-'}"
message.set_content(_auto_buy_protection_failure_body(result))
try:
with smtp_factory(settings.email_smtp_server, settings.email_smtp_port, timeout=20) as smtp:
smtp.login(settings.email_login, password)
smtp.send_message(message)
except Exception as exc:
raise EmailNotificationError("email notification failed") from exc
def send_futures_ai_trade_batch_email(
settings: Settings,
result: FuturesAiBatchResult,
smtp_factory: Callable[..., object] = smtplib.SMTP,
) -> None:
if not settings.email_notifications_enabled:
return
password = settings.email_sender_password.get_secret_value()
message = EmailMessage()
message["From"] = settings.email_login or ""
message["To"] = settings.email_recipient or ""
message["Subject"] = f"Binance futures AI trade batch {result.status}"
message.set_content(_futures_batch_body(result))
message.add_alternative(_futures_batch_html_body(result), subtype="html")
try:
with smtp_factory(settings.email_smtp_server, settings.email_smtp_port, timeout=20) as smtp:
smtp.login(settings.email_login, password)
smtp.send_message(message)
except Exception as exc:
raise EmailNotificationError("email notification failed") from exc
def _body(result: AiTradeExecutionResult) -> str:
return "\n".join(
[
"Binance AI trade result",
"",
f"Run ID: {result.run_id if result.run_id is not None else '-'}",
f"Status: {result.status}",
f"Symbol: {result.symbol or '-'}",
f"Action: {result.action}",
f"Confidence: {_format_optional_decimal(result.confidence)}",
f"Reason: {result.reason}",
"",
]
)
def _auto_buy_protection_failure_body(result: AutoBuyExecutionResult) -> str:
return "\n".join(
[
"Binance auto-buy protection failure",
"",
f"Run ID: {result.run_id}",
f"Status: {result.status}",
f"Symbol: {result.symbol or '-'}",
f"Reason: {result.reason}",
"",
]
)
def _batch_body(result: AiTradeBatchExecutionResult) -> str:
action_results, hold_results = _split_hold_results(result.results)
lines = [
"Binance AI trade batch result",
"",
f"Status: {result.status}",
f"Reason: {result.reason}",
f"Max actions: {result.max_actions}",
f"Executed actions: {result.executed_count}",
f"Predictions: {result.prediction_count}",
"",
]
if action_results:
lines.append("Action results:")
lines.extend(_result_table(action_results, include_executed=True))
lines.append("")
if hold_results:
lines.append("HOLD predictions:")
lines.extend(_result_table(hold_results))
lines.append("")
lines.append("")
return "\n".join(lines)
def _batch_html_body(result: AiTradeBatchExecutionResult) -> str:
action_results, hold_results = _split_hold_results(result.results)
sections = [
"",
"",
"",
"Binance AI trade batch result
",
"",
_summary_row("Status", result.status),
_summary_row("Reason", result.reason),
_summary_row("Max actions", str(result.max_actions)),
_summary_row("Executed actions", str(result.executed_count)),
_summary_row("Predictions", str(result.prediction_count)),
"
",
]
if action_results:
sections.extend(
[
"Action results
",
_html_result_table(action_results, include_executed=True),
]
)
if hold_results:
sections.extend(
[
"HOLD predictions
",
_html_result_table(hold_results),
]
)
sections.extend(["", ""])
return "\n".join(sections)
def _futures_batch_body(result: FuturesAiBatchResult) -> str:
action_results, hold_results = _split_hold_results(result.results)
lines = [
"Binance futures AI trade batch result",
"",
f"Status: {result.status}",
f"Reason: {result.reason}",
f"Max actions: {result.max_actions}",
f"Executed actions: {result.executed_count}",
f"Predictions: {result.prediction_count}",
f"Daily net PnL: {_format_compact_decimal(result.daily_realized_pnl)}",
f"Total isolated margin: {_format_compact_decimal(result.total_isolated_margin)}",
"",
]
if action_results:
lines.append("Action results:")
lines.extend(_futures_result_table(action_results))
lines.append("")
if hold_results:
lines.append("HOLD predictions:")
lines.extend(_futures_result_table(hold_results))
lines.append("")
lines.append("")
return "\n".join(lines)
def _futures_batch_html_body(result: FuturesAiBatchResult) -> str:
action_results, hold_results = _split_hold_results(result.results)
sections = [
"",
"",
"",
"Binance futures AI trade batch result
",
"",
_summary_row("Status", result.status),
_summary_row("Reason", result.reason),
_summary_row("Max actions", str(result.max_actions)),
_summary_row("Executed actions", str(result.executed_count)),
_summary_row("Predictions", str(result.prediction_count)),
_summary_row("Daily net PnL", _format_compact_decimal(result.daily_realized_pnl)),
_summary_row("Total isolated margin", _format_compact_decimal(result.total_isolated_margin)),
"
",
]
if action_results:
sections.extend(
[
"Action results
",
_futures_html_result_table(action_results),
]
)
if hold_results:
sections.extend(
[
"HOLD predictions
",
_futures_html_result_table(hold_results),
]
)
sections.extend(["", ""])
return "\n".join(sections)
def _split_hold_results(results):
action_results = tuple(item for item in results if item.action != "HOLD")
hold_results = tuple(item for item in results if item.action == "HOLD")
return action_results, hold_results
def _result_table(results: tuple[AiTradeExecutionResult, ...], *, include_executed: bool = False) -> list[str]:
headers = ["Summary", "Reason"]
rows = []
for item in results:
row = [_result_summary(item, include_executed=include_executed)]
row.append(item.reason)
rows.append(row)
widths = [max(len(header), *(len(row[index]) for row in rows)) for index, header in enumerate(headers)]
separator = "| " + " | ".join("-" * width for width in widths) + " |"
return [_table_row(headers, widths), separator, *(_table_row(row, widths) for row in rows)]
def _table_row(values: list[str], widths: list[int]) -> str:
return "| " + " | ".join(value.ljust(width) for value, width in zip(values, widths)) + " |"
def _summary_row(label: str, value: str) -> str:
return (
""
f"| {escape(label)} | "
f"{escape(value)} | "
"
"
)
def _html_result_table(results: tuple[AiTradeExecutionResult, ...], *, include_executed: bool = False) -> str:
headers = ["Summary", "Reason"]
header_cells = "".join(
f"{escape(header)} | "
for header in headers
)
rows = []
for item in results:
values = [_result_summary(item, include_executed=include_executed)]
values.append(item.reason)
cells = "".join(
f"{escape(value)} | " for value in values
)
rows.append(f"{cells}
")
return (
""
f"{header_cells}
"
f"{''.join(rows)}"
"
"
)
def _futures_result_table(results) -> list[str]:
headers = ["Summary", "Reason"]
rows = [_futures_result_values(item) for item in results]
if not rows:
rows = [["-", "-"]]
widths = [max(len(header), *(len(row[index]) for row in rows)) for index, header in enumerate(headers)]
separator = "| " + " | ".join("-" * width for width in widths) + " |"
return [_table_row(headers, widths), separator, *(_table_row(row, widths) for row in rows)]
def _futures_html_result_table(results) -> str:
headers = ["Summary", "Reason"]
header_cells = "".join(
f"{escape(header)} | "
for header in headers
)
rows = []
result_rows = [_futures_result_values(item) for item in results]
if not result_rows:
result_rows = [["-", "-"]]
for values in result_rows:
cells = "".join(
f"{escape(value)} | " for value in values
)
rows.append(f"{cells}
")
return (
""
f"{header_cells}
"
f"{''.join(rows)}"
"
"
)
def _futures_result_values(item) -> list[str]:
return [
_futures_result_summary(item),
item.reason,
]
def _futures_result_summary(item) -> str:
return (
f"Run ID: {item.run_id if item.run_id is not None else '-'}; "
f"Status: {item.status}; "
f"Symbol: {item.symbol or '-'}; "
f"Action: {item.action}; "
f"Confidence: {_format_compact_decimal(item.confidence)}; "
f"Leverage: {'-' if item.leverage is None else item.leverage}; "
f"Margin: {_format_compact_decimal(item.margin_amount)}; "
f"Close%: {'-' if item.close_percent is None else item.close_percent}"
)
def _result_summary(item: AiTradeExecutionResult, *, include_executed: bool = False) -> str:
summary = (
f"Run ID: {item.run_id if item.run_id is not None else '-'}; "
f"Status: {item.status}; "
f"Symbol: {item.symbol or '-'}; "
f"Action: {item.action}; "
f"Confidence: {_format_optional_decimal(item.confidence)}"
)
if include_executed:
summary += f"; Executed: {_format_ai_trade_executed(item.status)}"
return summary
def _format_optional_decimal(value) -> str:
return "-" if value is None else str(value)
def _format_compact_decimal(value) -> str:
if value is None:
return "-"
if isinstance(value, Decimal):
normalized = value.normalize()
if normalized == 0:
return "0"
return format(normalized, "f")
return str(value)
def _format_ai_trade_executed(status: str) -> str:
return "yes" if is_ai_trade_executed_status(status) else "no"