Trademind_OpenCode/trademind/config.py

"""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,
    )