from __future__ import annotations from functools import lru_cache from pathlib import Path from pydantic import field_validator from pydantic_settings import BaseSettings, SettingsConfigDict REPO_ROOT = Path(__file__).resolve().parents[2] DATA_DIR = REPO_ROOT / "data" class Settings(BaseSettings): model_config = SettingsConfigDict( env_file=(REPO_ROOT / ".env", ".env"), env_file_encoding="utf-8", extra="ignore", ) app_name: str = "Crypto Trader" env: str = "dev" host: str = "0.0.0.0" port: int = 8000 database_url: str = "sqlite:///./data/trader.db" mode: str = "paper" exchange: str = "binance" symbol: str = "BTC/USDT" timeframe: str = "5m" poll_interval: float = 5.0 candle_limit: int = 200 starting_balance: float = 10_000.0 fee_bps: float = 10.0 slippage_bps: float = 5.0 exchange_api_key: str = "" exchange_api_secret: str = "" exchange_sandbox: bool = False # --- Go-live gate --- # Live trading only starts when this is exactly "I UNDERSTAND THE RISK". # It is deliberately not a boolean: setting MODE=live alone is not enough. live_confirmation: str = "" # Refuse to start live below this account size (quote currency). min_live_balance: float = 100.0 # --- Risk controls (0 disables a limit) --- # Absolute loss in quote currency since the start of the UTC day. max_daily_loss: float = 0.0 # Percentage fall from peak equity. max_drawdown_pct: float = 0.0 # Cap on a single position as a fraction of equity (1.0 = all-in). max_position_pct: float = 1.0 # Maximum number of new entries per UTC day. max_trades_per_day: int = 0 # --- Notifications --- telegram_bot_token: str = "" telegram_chat_id: str = "" notify_webhook_url: str = "" # Optional bearer token required on /api requests (defense-in-depth behind # the Traefik Basic auth). Empty disables the check. api_token: str = "" api_token_header: str = "X-API-Token" cors_origins: str = "*" @field_validator("mode") @classmethod def _validate_mode(cls, value: str) -> str: value = value.lower().strip() if value not in {"paper", "live"}: raise ValueError("mode must be 'paper' or 'live'") return value @property def cors_origin_list(self) -> list[str]: return [origin.strip() for origin in self.cors_origins.split(",") if origin.strip()] @property def is_live(self) -> bool: return self.mode == "live" @lru_cache def get_settings() -> Settings: settings = Settings() DATA_DIR.mkdir(parents=True, exist_ok=True) return settings