from __future__ import annotations from collections.abc import Iterable from decimal import Decimal from pathlib import Path from typing import Any from urllib.parse import urlparse import yaml from pydantic import Field, SecretStr, field_validator, model_validator from pydantic_settings import BaseSettings, SettingsConfigDict from .models import TradingMode TESTNET_REST_BASE_URL = "https://testnet.binance.vision/api" TESTNET_STREAM_BASE_URL = "wss://stream.testnet.binance.vision/ws" LIVE_REST_BASE_URL = "https://api.binance.com/api" LIVE_REST_BASE_URLS = ( "https://api.binance.com/api", "https://api1.binance.com/api", "https://api2.binance.com/api", "https://api3.binance.com/api", "https://api4.binance.com/api", ) LIVE_STREAM_BASE_URL = "wss://stream.binance.com:9443/ws" TESTNET_REST_HOSTS = {"testnet.binance.vision"} TESTNET_STREAM_HOSTS = {"stream.testnet.binance.vision"} LIVE_REST_HOSTS = {"api.binance.com", "api1.binance.com", "api2.binance.com", "api3.binance.com", "api4.binance.com"} LIVE_STREAM_HOSTS = {"stream.binance.com", "data-stream.binance.vision"} FUTURES_TESTNET_REST_BASE_URL = "https://testnet.binancefuture.com" FUTURES_TESTNET_STREAM_BASE_URL = "wss://stream.binancefuture.com/ws" FUTURES_LIVE_REST_BASE_URL = "https://fapi.binance.com" FUTURES_LIVE_REST_BASE_URLS = ( "https://fapi.binance.com", "https://fapi1.binance.com", "https://fapi2.binance.com", "https://fapi3.binance.com", ) FUTURES_LIVE_STREAM_BASE_URL = "wss://fstream.binance.com/ws" FUTURES_TESTNET_REST_HOSTS = {"testnet.binancefuture.com"} FUTURES_TESTNET_STREAM_HOSTS = {"stream.binancefuture.com"} FUTURES_LIVE_REST_HOSTS = {"fapi.binance.com", "fapi1.binance.com", "fapi2.binance.com", "fapi3.binance.com"} FUTURES_LIVE_STREAM_HOSTS = {"fstream.binance.com"} DEFAULT_TRADING_SYMBOLS = ( "BTCUSDT", "ETHUSDT", "BNBUSDT", "SOLUSDT", "SPCXBUSDT", "TSLABUSDT", "MUBUSDT", "CRCLBUSDT", "SNDKBUSDT", "NVDABUSDT", ) DEFAULT_AI_TRADER_KLINES = (("1h", 24), ("4h", 30), ("1d", 14)) class Settings(BaseSettings): model_config = SettingsConfigDict(env_prefix="BINANCE_", env_file=".env", extra="ignore") api_key: str = Field(default="", repr=False) api_secret: SecretStr = SecretStr("") mode: TradingMode = TradingMode.DRY_RUN symbol: str = "BTCUSDT" quote_amount: Decimal = Decimal("25") max_quote_amount: Decimal = Decimal("50") fast_window: int = Field(default=5, gt=0) slow_window: int = Field(default=20, gt=0) cooldown_seconds: int = 60 database_path: Path = Path("binance_quant.sqlite3") log_level: str = "INFO" rest_base_url: str | None = None rest_base_urls: tuple[str, ...] | str | None = None stream_base_url: str | None = None proxy_url: str | None = None prevent_repeated_side: bool = True recv_window: int = Field(default=5000, ge=1, le=60000) live_auto_buyer_enabled: bool = False auto_buy_symbols: tuple[str, ...] | str = DEFAULT_TRADING_SYMBOLS auto_buy_quote_amount: Decimal = Decimal("5") auto_buy_existing_base_value_limit: Decimal | None = None auto_buy_daily_limit: int = 5 auto_buy_take_profit_percent: Decimal = Decimal("3") auto_buy_stop_loss_percent: Decimal = Decimal("2") auto_buy_stop_limit_buffer_percent: Decimal = Decimal("0.2") auto_buy_max_spread_bps: Decimal = Decimal("10") auto_buy_max_24h_rally_percent: Decimal = Decimal("10") auto_buy_cooldown_seconds: int = Field(default=3600, ge=0) auto_buy_min_score: Decimal = Decimal("4") live_auto_seller_enabled: bool = False auto_sell_symbols: tuple[str, ...] | str = DEFAULT_TRADING_SYMBOLS auto_sell_daily_limit: int = 5 auto_sell_cooldown_seconds: int = Field(default=3600, ge=0) auto_sell_min_score: Decimal = Decimal("4") auto_sell_max_spread_bps: Decimal = Decimal("10") auto_sell_require_profit: bool = True auto_sell_min_hold_seconds: int = Field(default=21600, ge=0) live_ai_trader_enabled: bool = False ai_trader_symbols: tuple[str, ...] | str = DEFAULT_TRADING_SYMBOLS ai_trader_model: str | None = None ai_trader_min_confidence: Decimal = Decimal("0.70") ai_trader_codex_command: str = "codex" ai_trader_codex_timeout_seconds: int = Field(default=60, ge=1) ai_trader_max_actions_per_run: int = 5 ai_trader_klines: tuple[tuple[str, int], ...] | str = DEFAULT_AI_TRADER_KLINES ai_trader_agg_trades_limit: int = Field(default=500, ge=1, le=1000) ai_trader_ws_enabled: bool = True ai_trader_ws_sample_seconds: int = Field(default=15, ge=1, le=60) live_futures_ai_trader_enabled: bool = False live_futures_confirm_production: bool = False futures_ai_symbols: tuple[str, ...] | str = DEFAULT_TRADING_SYMBOLS futures_margin_amount: Decimal = Decimal("5") futures_default_leverage: int = Field(default=2, ge=1) futures_max_leverage: int = Field(default=3, ge=1) futures_max_total_margin: Decimal = Decimal("20") futures_ai_min_confidence: Decimal = Decimal("0.70") futures_ai_max_actions_per_run: int = Field(default=5, ge=1) futures_daily_realized_loss_limit: Decimal = Decimal("10") futures_max_margin_loss_percent: Decimal = Decimal("50") futures_min_liquidation_buffer_percent: Decimal = Decimal("2") futures_min_stop_distance_percent: Decimal = Decimal("0.3") futures_max_stop_distance_percent: Decimal = Decimal("5") futures_min_take_profit_distance_percent: Decimal = Decimal("0.3") futures_max_take_profit_distance_percent: Decimal = Decimal("10") futures_trend_filter_enabled: bool = True futures_reverse_cooldown_seconds: int = Field(default=1800, ge=0) futures_min_stop_atr_multiple: Decimal = Decimal("0.8") futures_rest_base_url: str | None = None futures_rest_base_urls: tuple[str, ...] | str | None = None futures_stream_base_url: str | None = None email_notifications_enabled: bool = False email_smtp_server: str | None = None email_smtp_port: int = Field(default=25, ge=1, le=65535) email_login: str | None = None email_sender_password: SecretStr = SecretStr("") email_recipient: str | None = None @field_validator("symbol") @classmethod def normalize_symbol(cls, value: str) -> str: value = value.strip().upper() if not value: raise ValueError("symbol must not be empty") return value @field_validator("quote_amount", "max_quote_amount") @classmethod def positive_decimal(cls, value: Decimal) -> Decimal: if value <= 0: raise ValueError("amounts must be positive") return value @field_validator("rest_base_url") @classmethod def validate_rest_base_url(cls, value: str | None) -> str | None: if value is None: return None return _normalize_base_url(value, label="REST", expected_scheme="https") @field_validator("rest_base_urls", mode="before") @classmethod def parse_rest_base_urls(cls, value: Any) -> tuple[str, ...] | None: if value is None: return None if isinstance(value, str): urls = tuple(item.strip() for item in value.split(",") if item.strip()) else: if not isinstance(value, Iterable): raise ValueError("rest_base_urls must be a string or iterable collection of URLs") urls = tuple(str(item).strip() for item in value if str(item).strip()) if not urls: raise ValueError("rest_base_urls must contain at least one URL") return urls @field_validator("rest_base_urls") @classmethod def validate_rest_base_urls(cls, value: tuple[str, ...] | None) -> tuple[str, ...] | None: if value is None: return None return tuple(_normalize_base_url(item, label="REST", expected_scheme="https") for item in value) @field_validator("stream_base_url") @classmethod def validate_stream_base_url(cls, value: str | None) -> str | None: if value is None: return None return _normalize_base_url(value, label="stream", expected_scheme="wss") @field_validator("futures_rest_base_url") @classmethod def validate_futures_rest_base_url(cls, value: str | None) -> str | None: if value is None: return None return _normalize_base_url(value, label="futures REST", expected_scheme="https") @field_validator("futures_rest_base_urls", mode="before") @classmethod def parse_futures_rest_base_urls(cls, value: Any) -> tuple[str, ...] | None: if value is None: return None if isinstance(value, str): urls = tuple(item.strip() for item in value.split(",") if item.strip()) else: if not isinstance(value, Iterable): raise ValueError("futures_rest_base_urls must be a string or iterable collection of URLs") urls = tuple(str(item).strip() for item in value if str(item).strip()) if not urls: raise ValueError("futures_rest_base_urls must contain at least one URL") return urls @field_validator("futures_rest_base_urls") @classmethod def validate_futures_rest_base_urls(cls, value: tuple[str, ...] | None) -> tuple[str, ...] | None: if value is None: return None return tuple(_normalize_base_url(item, label="futures REST", expected_scheme="https") for item in value) @field_validator("futures_stream_base_url") @classmethod def validate_futures_stream_base_url(cls, value: str | None) -> str | None: if value is None: return None return _normalize_base_url(value, label="futures stream", expected_scheme="wss") @field_validator("proxy_url") @classmethod def validate_proxy_url(cls, value: str | None) -> str | None: if value is None or not value.strip(): return None return _normalize_proxy_url(value) @field_validator("auto_buy_symbols", mode="before") @classmethod def parse_auto_buy_symbols(cls, value: Any) -> tuple[str, ...]: if isinstance(value, str): symbols = tuple(item.strip().upper() for item in value.split(",") if item.strip()) else: if not isinstance(value, Iterable): raise ValueError("auto_buy_symbols must be a string or iterable collection of symbols") symbols = tuple(str(item).strip().upper() for item in value if str(item).strip()) if not symbols: raise ValueError("auto_buy_symbols must contain at least one symbol") if any(not symbol.endswith("USDT") for symbol in symbols): raise ValueError("auto_buy_symbols must be USDT spot symbols") return symbols @field_validator("auto_sell_symbols", mode="before") @classmethod def parse_auto_sell_symbols(cls, value: Any) -> tuple[str, ...]: if isinstance(value, str): symbols = tuple(item.strip().upper() for item in value.split(",") if item.strip()) else: if not isinstance(value, Iterable): raise ValueError("auto_sell_symbols must be a string or iterable collection of symbols") symbols = tuple(str(item).strip().upper() for item in value if str(item).strip()) if not symbols: raise ValueError("auto_sell_symbols must contain at least one symbol") if any(not symbol.endswith("USDT") for symbol in symbols): raise ValueError("auto_sell_symbols must be USDT spot symbols") return symbols @field_validator("ai_trader_symbols", mode="before") @classmethod def parse_ai_trader_symbols(cls, value: Any) -> tuple[str, ...]: if isinstance(value, str): symbols = tuple(item.strip().upper() for item in value.split(",") if item.strip()) else: if not isinstance(value, Iterable): raise ValueError("ai_trader_symbols must be a string or iterable collection of symbols") symbols = tuple(str(item).strip().upper() for item in value if str(item).strip()) if not symbols: raise ValueError("ai_trader_symbols must contain at least one symbol") if any(not symbol.endswith("USDT") for symbol in symbols): raise ValueError("ai_trader_symbols must be USDT spot symbols") return symbols @field_validator("futures_ai_symbols", mode="before") @classmethod def parse_futures_ai_symbols(cls, value: Any) -> tuple[str, ...]: if isinstance(value, str): symbols = tuple(item.strip().upper() for item in value.split(",") if item.strip()) else: if not isinstance(value, Iterable): raise ValueError("futures_ai_symbols must be a string or iterable collection of symbols") symbols = tuple(str(item).strip().upper() for item in value if str(item).strip()) if not symbols: raise ValueError("futures_ai_symbols must contain at least one symbol") if any(not symbol.endswith("USDT") for symbol in symbols): raise ValueError("futures_ai_symbols must be USDT futures symbols") return symbols @field_validator("ai_trader_model") @classmethod def normalize_ai_trader_model(cls, value: str | None) -> str | None: if value is None: return None normalized = value.strip() return normalized or None @field_validator("ai_trader_codex_command") @classmethod def normalize_ai_trader_codex_command(cls, value: str) -> str: normalized = value.strip() if not normalized: raise ValueError("ai_trader_codex_command must not be empty") return normalized @field_validator("ai_trader_klines", mode="before") @classmethod def parse_ai_trader_klines(cls, value: Any) -> tuple[tuple[str, int], ...]: if isinstance(value, str): items = tuple(item.strip() for item in value.split(",") if item.strip()) if not items: raise ValueError("ai_trader_klines must contain at least one interval") parsed: list[tuple[str, int]] = [] for item in items: if ":" not in item: raise ValueError("ai_trader_klines items must use interval:limit") interval, raw_limit = (part.strip() for part in item.split(":", 1)) if not interval or not raw_limit: raise ValueError("ai_trader_klines items must use interval:limit") try: limit = int(raw_limit) except ValueError: raise ValueError("ai_trader_klines limit must be positive") from None parsed.append((interval, limit)) else: if not isinstance(value, Iterable): raise ValueError("ai_trader_klines must be a string or iterable of interval/limit pairs") parsed = [] for item in value: if not isinstance(item, Iterable) or isinstance(item, (str, bytes)): raise ValueError("ai_trader_klines items must use interval:limit") pair = tuple(item) if len(pair) != 2: raise ValueError("ai_trader_klines items must use interval:limit") interval = str(pair[0]).strip() try: limit = int(pair[1]) except (TypeError, ValueError): raise ValueError("ai_trader_klines limit must be positive") from None parsed.append((interval, limit)) if not parsed: raise ValueError("ai_trader_klines must contain at least one interval") intervals = [interval for interval, _ in parsed] if any(not interval for interval in intervals): raise ValueError("ai_trader_klines items must use interval:limit") if any(limit <= 0 for _, limit in parsed): raise ValueError("ai_trader_klines limit must be positive") normalized_intervals = [interval.lower() for interval in intervals] if len(set(normalized_intervals)) != len(normalized_intervals): raise ValueError("ai_trader_klines intervals must be unique") return tuple((interval, limit) for interval, limit in parsed) @field_validator("email_smtp_server", "email_login", "email_recipient") @classmethod def normalize_optional_email_text(cls, value: str | None) -> str | None: if value is None: return None normalized = value.strip() return normalized or None @field_validator( "auto_buy_quote_amount", "auto_buy_existing_base_value_limit", "auto_buy_take_profit_percent", "auto_buy_stop_loss_percent", "auto_buy_stop_limit_buffer_percent", "auto_buy_max_spread_bps", "auto_buy_max_24h_rally_percent", "auto_buy_min_score", ) @classmethod def positive_auto_buy_decimal(cls, value: Decimal | None, info) -> Decimal | None: if value is None: return None if value <= 0: raise ValueError(f"{info.field_name} must be positive") return value @field_validator("auto_buy_daily_limit") @classmethod def positive_auto_buy_daily_limit(cls, value: int) -> int: if value <= 0: raise ValueError("auto_buy_daily_limit must be positive") return value @field_validator("auto_sell_min_score", "auto_sell_max_spread_bps") @classmethod def positive_auto_sell_decimal(cls, value: Decimal, info) -> Decimal: if value <= 0: raise ValueError(f"{info.field_name} must be positive") return value @field_validator("auto_sell_daily_limit") @classmethod def positive_auto_sell_daily_limit(cls, value: int) -> int: if value <= 0: raise ValueError("auto_sell_daily_limit must be positive") return value @field_validator("ai_trader_min_confidence") @classmethod def valid_ai_trader_min_confidence(cls, value: Decimal) -> Decimal: if value <= 0 or value > 1: raise ValueError("ai_trader_min_confidence must be greater than 0 and less than or equal to 1") return value @field_validator("ai_trader_max_actions_per_run") @classmethod def positive_ai_trader_max_actions_per_run(cls, value: int) -> int: if value <= 0: raise ValueError("ai_trader_max_actions_per_run must be positive") return value @field_validator( "futures_margin_amount", "futures_max_total_margin", "futures_daily_realized_loss_limit", "futures_max_margin_loss_percent", "futures_min_liquidation_buffer_percent", "futures_min_stop_distance_percent", "futures_max_stop_distance_percent", "futures_min_take_profit_distance_percent", "futures_max_take_profit_distance_percent", "futures_min_stop_atr_multiple", ) @classmethod def positive_futures_decimal(cls, value: Decimal, info) -> Decimal: if value <= 0: raise ValueError(f"{info.field_name} must be positive") return value @field_validator("futures_default_leverage", "futures_max_leverage", mode="before") @classmethod def positive_futures_leverage(cls, value: Any, info) -> Any: try: parsed = int(value) except (TypeError, ValueError): return value if parsed <= 0: raise ValueError(f"{info.field_name} must be positive") return value @field_validator("futures_ai_min_confidence") @classmethod def valid_futures_ai_min_confidence(cls, value: Decimal) -> Decimal: if value <= 0 or value > 1: raise ValueError("futures_ai_min_confidence must be greater than 0 and less than or equal to 1") return value @field_validator("futures_ai_max_actions_per_run", mode="before") @classmethod def positive_futures_ai_max_actions_per_run(cls, value: Any) -> Any: try: parsed = int(value) except (TypeError, ValueError): return value if parsed <= 0: raise ValueError("futures_ai_max_actions_per_run must be positive") return value @model_validator(mode="after") def validate_windows_and_credentials(self) -> "Settings": if self.fast_window >= self.slow_window: raise ValueError("fast_window must be less than slow_window") if self.quote_amount > self.max_quote_amount: raise ValueError("quote_amount must be less than or equal to max_quote_amount") if self.auto_buy_existing_base_value_limit is None: self.auto_buy_existing_base_value_limit = self.auto_buy_quote_amount if self.email_notifications_enabled and ( not self.email_smtp_server or not self.email_login or not self.email_sender_password.get_secret_value() or not self.email_recipient ): raise ValueError("email notifications require SMTP server, login, sender password, and recipient") if self.mode is TradingMode.LIVE: if self.rest_base_urls is None: if self.rest_base_url is None: self.rest_base_urls = LIVE_REST_BASE_URLS else: self.rest_base_urls = (self.rest_base_url,) if self.rest_base_url is None: self.rest_base_url = self.rest_base_urls[0] if self.stream_base_url is None: self.stream_base_url = LIVE_STREAM_BASE_URL if not self.api_key or not self.api_secret.get_secret_value(): raise ValueError("live mode requires BINANCE_API_KEY and BINANCE_API_SECRET") for rest_base_url in self.rest_base_urls: _validate_url_host(rest_base_url, allowed_hosts=LIVE_REST_HOSTS, label="REST", environment="production") _validate_url_host( self.stream_base_url, allowed_hosts=LIVE_STREAM_HOSTS, label="stream", environment="production", ) else: if self.rest_base_urls is None: if self.rest_base_url is None: self.rest_base_urls = (TESTNET_REST_BASE_URL,) else: self.rest_base_urls = (self.rest_base_url,) if self.rest_base_url is None: self.rest_base_url = self.rest_base_urls[0] if self.stream_base_url is None: self.stream_base_url = TESTNET_STREAM_BASE_URL if self.mode is TradingMode.TESTNET_LIVE and ( not self.api_key or not self.api_secret.get_secret_value() ): raise ValueError("Testnet live mode requires BINANCE_API_KEY and BINANCE_API_SECRET") for rest_base_url in self.rest_base_urls: _validate_url_host(rest_base_url, allowed_hosts=TESTNET_REST_HOSTS, label="REST", environment="Testnet") _validate_url_host( self.stream_base_url, allowed_hosts=TESTNET_STREAM_HOSTS, label="stream", environment="Testnet", ) if self.futures_default_leverage > self.futures_max_leverage: raise ValueError("futures_default_leverage must be less than or equal to futures_max_leverage") if self.futures_min_stop_distance_percent > self.futures_max_stop_distance_percent: raise ValueError( "futures_min_stop_distance_percent must be less than or equal to futures_max_stop_distance_percent" ) if self.futures_min_take_profit_distance_percent > self.futures_max_take_profit_distance_percent: raise ValueError( "futures_min_take_profit_distance_percent must be less than or equal to " "futures_max_take_profit_distance_percent" ) if self.mode is TradingMode.LIVE: if self.futures_rest_base_urls is None: if self.futures_rest_base_url is None: self.futures_rest_base_urls = FUTURES_LIVE_REST_BASE_URLS else: self.futures_rest_base_urls = (self.futures_rest_base_url,) if self.futures_rest_base_url is None: self.futures_rest_base_url = self.futures_rest_base_urls[0] if self.futures_stream_base_url is None: self.futures_stream_base_url = FUTURES_LIVE_STREAM_BASE_URL for rest_base_url in self.futures_rest_base_urls: _validate_url_host( rest_base_url, allowed_hosts=FUTURES_LIVE_REST_HOSTS, label="futures REST", environment="production", ) _validate_url_host( self.futures_stream_base_url, allowed_hosts=FUTURES_LIVE_STREAM_HOSTS, label="futures stream", environment="production", ) else: if self.futures_rest_base_urls is None: if self.futures_rest_base_url is None: self.futures_rest_base_urls = (FUTURES_TESTNET_REST_BASE_URL,) else: self.futures_rest_base_urls = (self.futures_rest_base_url,) if self.futures_rest_base_url is None: self.futures_rest_base_url = self.futures_rest_base_urls[0] if self.futures_stream_base_url is None: self.futures_stream_base_url = FUTURES_TESTNET_STREAM_BASE_URL for rest_base_url in self.futures_rest_base_urls: _validate_url_host( rest_base_url, allowed_hosts=FUTURES_TESTNET_REST_HOSTS, label="futures REST", environment="Testnet", ) _validate_url_host( self.futures_stream_base_url, allowed_hosts=FUTURES_TESTNET_STREAM_HOSTS, label="futures stream", environment="Testnet", ) return self def load_settings(config_path: str | Path | None = None) -> Settings: values: dict[str, Any] = {} if config_path: path = Path(config_path) if path.exists(): loaded = yaml.safe_load(path.read_text(encoding="utf-8")) or {} if not isinstance(loaded, dict): raise ValueError("YAML config must contain a mapping") values.update(loaded) return Settings(**values) def _normalize_base_url(value: str, *, label: str, expected_scheme: str) -> str: normalized = value.strip().rstrip("/") parsed = urlparse(normalized) if not parsed.scheme or not parsed.hostname: raise ValueError(f"{label} base URL must include a scheme and host") if parsed.scheme != expected_scheme: raise ValueError(f"{label} base URL scheme must be {expected_scheme}") return normalized def _validate_url_host(value: str, *, allowed_hosts: set[str], label: str, environment: str) -> None: parsed = urlparse(value) if not parsed.hostname or parsed.hostname.lower() not in allowed_hosts: raise ValueError(f"{label} base URL must be a Binance {environment} URL") def _normalize_proxy_url(value: str) -> str: normalized = value.strip() if "://" not in normalized: normalized = f"http://{normalized}" parsed = urlparse(normalized) if parsed.scheme not in {"http", "https", "socks5", "socks5h"} or not parsed.hostname: raise ValueError("proxy_url must be an HTTP or SOCKS URL") return normalized