"""Lädt und validiert die YAML-Konfiguration (inkl. API-Keys aus Env).""" from __future__ import annotations import os from dataclasses import dataclass, field from typing import Any, Dict, List, Optional import yaml SUPPORTED_EXCHANGES = ( "binance", "kraken", "coinbase", "kucoin", "bitmex", "okx", "bybit", ) def _env(value: Optional[str]) -> str: """Ersetzt ${ENV_VAR} Referenzen durch den jeweiligen Umgebungsvariable-Wert.""" if value and value.startswith("${") and value.endswith("}"): return os.environ.get(value[2:-1], "") return value or "" @dataclass class ExchangeConfig: name: str api_key: str = "" api_secret: str = "" password: str = "" # zB. binance passphrase / kucoin passkey sandbox: bool = True @classmethod def from_dict(cls, name: str, data: Dict[str, Any]) -> "ExchangeConfig": data = data or {} return cls( name=name, api_key=_env(str(data.get("api_key", ""))), api_secret=_env(str(data.get("api_secret", ""))), password=_env(str(data.get("password", ""))), sandbox=bool(data.get("sandbox", True)), ) @dataclass class TradingConfig: quote_currency: str = "USDT" base_currency: str = "BTC" initial_balance: float = 10_000.0 position_size_pct: float = 0.10 # Anteil des Portfolios pro Trade max_open_positions: int = 1 fee_pct: float = 0.001 # 0.1 % pro Order (Spread/fee) slippage_pct: float = 0.0005 timeframe: str = "1h" candles: int = 500 dry_run: bool = True # True = Simulation / Paper-Trading @classmethod def from_dict(cls, data: Dict[str, Any]) -> "TradingConfig": data = data or {} return cls( quote_currency=data.get("quote_currency", "USDT"), base_currency=data.get("base_currency", "BTC"), initial_balance=float(data.get("initial_balance", 10_000.0)), position_size_pct=float(data.get("position_size_pct", 0.10)), max_open_positions=int(data.get("max_open_positions", 1)), fee_pct=float(data.get("fee_pct", 0.001)), slippage_pct=float(data.get("slippage_pct", 0.0005)), timeframe=data.get("timeframe", "1h"), candles=int(data.get("candles", 500)), dry_run=bool(data.get("dry_run", True)), ) @dataclass class StrategyConfig: fast_period: int = 12 slow_period: int = 26 signal_period: int = 9 rsi_period: int = 14 rsi_overbought: float = 70.0 rsi_oversold: float = 30.0 atr_period: int = 14 atr_stop_mult: float = 2.5 allow_long: bool = True allow_short: bool = False @classmethod def from_dict(cls, data: Dict[str, Any]) -> "StrategyConfig": data = data or {} return cls( fast_period=int(data.get("fast_period", 12)), slow_period=int(data.get("slow_period", 26)), signal_period=int(data.get("signal_period", 9)), rsi_period=int(data.get("rsi_period", 14)), rsi_overbought=float(data.get("rsi_overbought", 70.0)), rsi_oversold=float(data.get("rsi_oversold", 30.0)), atr_period=int(data.get("atr_period", 14)), atr_stop_mult=float(data.get("atr_stop_mult", 2.5)), allow_long=bool(data.get("allow_long", True)), allow_short=bool(data.get("allow_short", False)), ) @dataclass class TrainingConfig: mode: str = "walk-forward" # walk-forward | full train_ratio: float = 0.7 generations: int = 20 population: int = 40 mutation_rate: float = 0.2 crossover_rate: float = 0.4 fitness_weight_return: float = 0.6 fitness_weight_sharpe: float = 0.3 fitness_weight_drawdown: float = 0.1 seed: int = 42 state_file: str = "state/weights.json" @classmethod def from_dict(cls, data: Dict[str, Any]) -> "TrainingConfig": data = data or {} return cls( mode=data.get("mode", "walk-forward"), train_ratio=float(data.get("train_ratio", 0.7)), generations=int(data.get("generations", 20)), population=int(data.get("population", 40)), mutation_rate=float(data.get("mutation_rate", 0.2)), crossover_rate=float(data.get("crossover_rate", 0.4)), fitness_weight_return=float(data.get("fitness_weight_return", 0.6)), fitness_weight_sharpe=float(data.get("fitness_weight_sharpe", 0.3)), fitness_weight_drawdown=float(data.get("fitness_weight_drawdown", 0.1)), seed=int(data.get("seed", 42)), state_file=data.get("state_file", "state/weights.json"), ) @dataclass class ServerConfig: """HTTP-API/Dashboard (Befehl `trademind serve`).""" host: str = "0.0.0.0" port: int = 8080 public_base_url: str = "" # optional, zB. Reverse-Proxy-URL für den Dashboard-Link @classmethod def from_dict(cls, data: Dict[str, Any]) -> "ServerConfig": data = data or {} return cls( host=data.get("host", "0.0.0.0"), port=int(data.get("port", 8080)), public_base_url=data.get("public_base_url", "") or "", ) @dataclass class Config: trading: TradingConfig strategy: StrategyConfig training: TrainingConfig server: ServerConfig = field(default_factory=ServerConfig) exchanges: Dict[str, ExchangeConfig] = field(default_factory=dict) def active_exchange(self) -> Optional[ExchangeConfig]: """Erste Konfiguration mit aktiver API-Anbindung (oder der ersten).""" if not self.exchanges: return None for cfg in self.exchanges.values(): if cfg.api_key and cfg.api_secret: return cfg return next(iter(self.exchanges.values())) def load(path: str) -> Config: with open(path, "r", encoding="utf-8") as fh: raw = yaml.safe_load(fh) or {} trading = TradingConfig.from_dict(raw.get("trading", {})) strategy = StrategyConfig.from_dict(raw.get("strategy", {})) training = TrainingConfig.from_dict(raw.get("training", {})) server = ServerConfig.from_dict(raw.get("server", {})) exchanges: Dict[str, ExchangeConfig] = {} for name, data in (raw.get("exchanges") or {}).items(): name = name.lower() if name not in SUPPORTED_EXCHANGES: raise ValueError( f"Unbekannte Exchange '{name}'. Erlaubt: {', '.join(SUPPORTED_EXCHANGES)}" ) exchanges[name] = ExchangeConfig.from_dict(name, data) return Config( trading=trading, strategy=strategy, training=training, server=server, exchanges=exchanges, )