from __future__ import annotations import os from dataclasses import dataclass from pathlib import Path FIXED_SPOT_SYMBOLS = ( "BTCUSDT", "ETHUSDT", "HYPEUSDT", "SOLUSDT", "XRPUSDT", "XPLUSDT", "WLDUSDT", "MNTUSDT", "HUSDT", "XAUTUSDT", "IPUSDT", "AAVEUSDT", ) STRATEGY_MODES = {"legacy", "trend_macd", "torch_forecast"} def _load_dotenv(path: Path) -> None: if not path.exists(): return for raw in path.read_text(encoding="utf-8").splitlines(): line = raw.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) key = key.strip() value = value.strip().strip('"').strip("'") os.environ.setdefault(key, value) def _bool_env(name: str, default: bool) -> bool: raw = os.getenv(name) if raw is None or raw == "": return default return raw.strip().lower() in {"1", "true", "yes", "y", "on"} def _float_env(name: str, default: float) -> float: raw = os.getenv(name) return default if raw in (None, "") else float(raw) def _int_env(name: str, default: int) -> int: raw = os.getenv(name) return default if raw in (None, "") else int(raw) def _symbols_env(name: str) -> tuple[str, ...]: raw = os.getenv(name, "") return tuple(part.strip().upper() for part in raw.split(",") if part.strip()) @dataclass class Settings: trading_mode: str host: str port: int bybit_testnet: bool bybit_api_key: str bybit_api_secret: str starting_balance_usdt: float auto_select_symbols: bool top_symbols_count: int symbols: tuple[str, ...] strategy_mode: str base_interval: str kline_limit: int trend_interval: str trend_kline_limit: int loop_interval_seconds: int fast_trading_enabled: bool fast_loop_interval_seconds: float fast_entry_cooldown_seconds: int max_entries_per_minute: int websocket_enabled: bool min_signal_confidence: float max_spread_percent: float min_24h_turnover_usdt: float pattern_analysis_enabled: bool pattern_score_weight: float learning_enabled: bool learning_lookback_trades: int learning_min_samples: int learning_max_adjustment: float learning_max_position_multiplier: float llm_advisor_enabled: bool ollama_base_url: str ollama_model: str llm_advisor_min_interval_seconds: int llm_advisor_timeout_seconds: int llm_advisor_max_adjustment: float min_position_usdt: float max_position_usdt: float max_symbol_exposure_usdt: float max_total_exposure_usdt: float max_open_positions: int max_positions_per_symbol: int grid_trading_enabled: bool grid_entry_confidence: float grid_buy_zone: float grid_max_position_usdt: float rebound_trading_enabled: bool rebound_entry_confidence: float rebound_min_probability: float rebound_max_position_usdt: float kelly_sizing_enabled: bool kelly_fraction: float kelly_max_fraction: float risk_per_trade_percent: float risk_guard_enabled: bool risk_symbol_guard_enabled: bool risk_recent_trade_window: int risk_max_consecutive_losses: int risk_min_recent_profit_factor: float risk_reduce_multiplier: float atr_trailing_multiplier: float trend_rsi_min: float trend_rsi_max: float time_series_forecast_enabled: bool time_series_min_candles: int time_series_forecast_horizon: int time_series_min_edge_percent: float time_series_min_probability_up: float time_series_min_confidence: float time_series_max_adjustment: float time_series_lstm_enabled: bool time_series_lstm_model_path: Path time_series_probe_enabled: bool time_series_probe_min_edge_percent: float time_series_probe_min_probability_up: float time_series_probe_size_multiplier: float time_series_rebound_fallback_enabled: bool stop_loss_percent: float stop_loss_exit_enabled: bool take_profit_percent: float trailing_stop_percent: float min_hold_seconds: int min_exit_net_percent: float entry_cooldown_seconds: int max_daily_drawdown_usdt: float min_cash_reserve_usdt: float taker_fee_rate: float slippage_rate: float enable_live_trading: bool live_trading_confirm: str live_order_max_usdt: float database_path: Path log_path: Path env_file_path: Path @property def rest_base_url(self) -> str: return "https://api-testnet.bybit.com" if self.bybit_testnet else "https://api.bybit.com" @property def websocket_url(self) -> str: return ( "wss://stream-testnet.bybit.com/v5/public/spot" if self.bybit_testnet else "wss://stream.bybit.com/v5/public/spot" ) @property def live_ready(self) -> bool: return ( self.trading_mode == "live" and self.enable_live_trading and self.live_trading_confirm == "I_ACCEPT_REAL_RISK" and bool(self.bybit_api_key) and bool(self.bybit_api_secret) ) @property def effective_loop_interval_seconds(self) -> float: interval = ( self.fast_loop_interval_seconds if self.fast_trading_enabled else float(self.loop_interval_seconds) ) return max(0.25, interval) @property def effective_entry_cooldown_seconds(self) -> int: return ( max(0, self.fast_entry_cooldown_seconds) if self.fast_trading_enabled else max(0, self.entry_cooldown_seconds) ) def load_settings(env_file: str | Path | None = None) -> Settings: root = Path.cwd() env_path = Path(env_file) if env_file else root / ".env" _load_dotenv(env_path) mode = os.getenv("TRADING_MODE", "paper").strip().lower() if mode not in {"paper", "live"}: raise ValueError("TRADING_MODE must be paper or live") strategy_mode = os.getenv("STRATEGY_MODE", "torch_forecast").strip().lower() if strategy_mode not in STRATEGY_MODES: raise ValueError("STRATEGY_MODE must be legacy, trend_macd or torch_forecast") auto_select_symbols = _bool_env("AUTO_SELECT_SYMBOLS", False) top_symbols_count = _int_env("TOP_SYMBOLS_COUNT", len(FIXED_SPOT_SYMBOLS)) requested_symbols = _symbols_env("SYMBOLS") symbols = requested_symbols if requested_symbols else (() if auto_select_symbols else FIXED_SPOT_SYMBOLS) forecast_enabled_default = strategy_mode == "torch_forecast" min_signal_confidence = _float_env("MIN_SIGNAL_CONFIDENCE", 0.64) settings = Settings( trading_mode=mode, host=os.getenv("HOST", "127.0.0.1"), port=_int_env("PORT", 8787), bybit_testnet=_bool_env("BYBIT_TESTNET", False), bybit_api_key=os.getenv("BYBIT_API_KEY", ""), bybit_api_secret=os.getenv("BYBIT_API_SECRET", ""), starting_balance_usdt=_float_env("STARTING_BALANCE_USDT", 100.0), auto_select_symbols=auto_select_symbols, top_symbols_count=top_symbols_count, symbols=symbols, strategy_mode=strategy_mode, base_interval=os.getenv("BASE_INTERVAL", "60"), kline_limit=_int_env("KLINE_LIMIT", 240), trend_interval=os.getenv("TREND_INTERVAL", "D"), trend_kline_limit=_int_env("TREND_KLINE_LIMIT", 260), loop_interval_seconds=_int_env("LOOP_INTERVAL_SECONDS", 5), fast_trading_enabled=_bool_env("FAST_TRADING_ENABLED", False), fast_loop_interval_seconds=_float_env("FAST_LOOP_INTERVAL_SECONDS", 1.0), fast_entry_cooldown_seconds=_int_env("FAST_ENTRY_COOLDOWN_SECONDS", 20), max_entries_per_minute=_int_env("MAX_ENTRIES_PER_MINUTE", 12), websocket_enabled=_bool_env("WEBSOCKET_ENABLED", True), min_signal_confidence=min_signal_confidence, max_spread_percent=_float_env("MAX_SPREAD_PERCENT", 0.18), min_24h_turnover_usdt=_float_env("MIN_24H_TURNOVER_USDT", 1000000.0), pattern_analysis_enabled=_bool_env("PATTERN_ANALYSIS_ENABLED", False), pattern_score_weight=_float_env("PATTERN_SCORE_WEIGHT", 0.18), learning_enabled=_bool_env("LEARNING_ENABLED", False), learning_lookback_trades=_int_env("LEARNING_LOOKBACK_TRADES", 120), learning_min_samples=_int_env("LEARNING_MIN_SAMPLES", 3), learning_max_adjustment=_float_env("LEARNING_MAX_ADJUSTMENT", 0.12), learning_max_position_multiplier=_float_env("LEARNING_MAX_POSITION_MULTIPLIER", 1.6), llm_advisor_enabled=_bool_env("LLM_ADVISOR_ENABLED", False), ollama_base_url=os.getenv("OLLAMA_BASE_URL", "http://192.168.0.210:11434").rstrip("/"), ollama_model=os.getenv("OLLAMA_MODEL", "gemma4:e4b"), llm_advisor_min_interval_seconds=_int_env("LLM_ADVISOR_MIN_INTERVAL_SECONDS", 180), llm_advisor_timeout_seconds=_int_env("LLM_ADVISOR_TIMEOUT_SECONDS", 45), llm_advisor_max_adjustment=_float_env("LLM_ADVISOR_MAX_ADJUSTMENT", 0.06), min_position_usdt=_float_env("MIN_POSITION_USDT", 1.0), max_position_usdt=_float_env("MAX_POSITION_USDT", 25.0), max_symbol_exposure_usdt=_float_env("MAX_SYMBOL_EXPOSURE_USDT", 25.0), max_total_exposure_usdt=_float_env("MAX_TOTAL_EXPOSURE_USDT", 75.0), max_open_positions=_int_env("MAX_OPEN_POSITIONS", len(FIXED_SPOT_SYMBOLS)), max_positions_per_symbol=_int_env("MAX_POSITIONS_PER_SYMBOL", 1), grid_trading_enabled=_bool_env("GRID_TRADING_ENABLED", False), grid_entry_confidence=_float_env("GRID_ENTRY_CONFIDENCE", 0.58), grid_buy_zone=_float_env("GRID_BUY_ZONE", 0.45), grid_max_position_usdt=_float_env("GRID_MAX_POSITION_USDT", 8.0), rebound_trading_enabled=_bool_env("REBOUND_TRADING_ENABLED", False), rebound_entry_confidence=_float_env("REBOUND_ENTRY_CONFIDENCE", 0.58), rebound_min_probability=_float_env("REBOUND_MIN_PROBABILITY", 0.58), rebound_max_position_usdt=_float_env("REBOUND_MAX_POSITION_USDT", 6.0), kelly_sizing_enabled=_bool_env("KELLY_SIZING_ENABLED", False), kelly_fraction=_float_env("KELLY_FRACTION", 0.25), kelly_max_fraction=_float_env("KELLY_MAX_FRACTION", 0.20), risk_per_trade_percent=_float_env("RISK_PER_TRADE_PERCENT", 0.01), risk_guard_enabled=_bool_env("RISK_GUARD_ENABLED", True), risk_symbol_guard_enabled=_bool_env("RISK_SYMBOL_GUARD_ENABLED", True), risk_recent_trade_window=_int_env("RISK_RECENT_TRADE_WINDOW", 20), risk_max_consecutive_losses=_int_env("RISK_MAX_CONSECUTIVE_LOSSES", 4), risk_min_recent_profit_factor=_float_env("RISK_MIN_RECENT_PROFIT_FACTOR", 0.85), risk_reduce_multiplier=_float_env("RISK_REDUCE_MULTIPLIER", 0.50), atr_trailing_multiplier=_float_env("ATR_TRAILING_MULTIPLIER", 2.2), trend_rsi_min=_float_env("TREND_RSI_MIN", 45.0), trend_rsi_max=_float_env("TREND_RSI_MAX", 65.0), time_series_forecast_enabled=_bool_env("TIME_SERIES_FORECAST_ENABLED", forecast_enabled_default), time_series_min_candles=_int_env("TIME_SERIES_MIN_CANDLES", 120), time_series_forecast_horizon=_int_env("TIME_SERIES_FORECAST_HORIZON", 3), time_series_min_edge_percent=_float_env("TIME_SERIES_MIN_EDGE_PERCENT", 0.08), time_series_min_probability_up=_float_env("TIME_SERIES_MIN_PROBABILITY_UP", 0.58), time_series_min_confidence=_float_env("TIME_SERIES_MIN_CONFIDENCE", 0.4), time_series_max_adjustment=_float_env("TIME_SERIES_MAX_ADJUSTMENT", 0.08), time_series_lstm_enabled=_bool_env("TIME_SERIES_LSTM_ENABLED", True), time_series_lstm_model_path=Path(os.getenv("TIME_SERIES_LSTM_MODEL_PATH", "runtime/lstm_forecaster.json")), time_series_probe_enabled=_bool_env("TIME_SERIES_PROBE_ENABLED", True), time_series_probe_min_edge_percent=_float_env("TIME_SERIES_PROBE_MIN_EDGE_PERCENT", 0.02), time_series_probe_min_probability_up=_float_env("TIME_SERIES_PROBE_MIN_PROBABILITY_UP", 0.55), time_series_probe_size_multiplier=_float_env("TIME_SERIES_PROBE_SIZE_MULTIPLIER", 0.40), time_series_rebound_fallback_enabled=_bool_env("TIME_SERIES_REBOUND_FALLBACK_ENABLED", True), stop_loss_percent=_float_env("STOP_LOSS_PERCENT", 0.04), stop_loss_exit_enabled=_bool_env("STOP_LOSS_EXIT_ENABLED", True), take_profit_percent=_float_env("TAKE_PROFIT_PERCENT", 0.035), trailing_stop_percent=_float_env("TRAILING_STOP_PERCENT", 0.015), min_hold_seconds=_int_env("MIN_HOLD_SECONDS", 180), min_exit_net_percent=_float_env("MIN_EXIT_NET_PERCENT", 0.20), entry_cooldown_seconds=_int_env("ENTRY_COOLDOWN_SECONDS", 180), max_daily_drawdown_usdt=_float_env("MAX_DAILY_DRAWDOWN_USDT", 6.0), min_cash_reserve_usdt=_float_env("MIN_CASH_RESERVE_USDT", 5.0), taker_fee_rate=_float_env("TAKER_FEE_RATE", 0.001), slippage_rate=_float_env("SLIPPAGE_RATE", 0.0003), enable_live_trading=_bool_env("ENABLE_LIVE_TRADING", False), live_trading_confirm=os.getenv("LIVE_TRADING_CONFIRM", ""), live_order_max_usdt=_float_env("LIVE_ORDER_MAX_USDT", 10.0), database_path=Path(os.getenv("DATABASE_PATH", "runtime/tradebot.sqlite3")), log_path=Path(os.getenv("LOG_PATH", "runtime/tradebot.log")), env_file_path=env_path, ) if settings.trading_mode == "live" and not settings.live_ready: raise ValueError( "Live mode is locked. Set ENABLE_LIVE_TRADING=true, " "LIVE_TRADING_CONFIRM=I_ACCEPT_REAL_RISK and Bybit keys." ) return settings def update_env_value(path: Path, key: str, value: str) -> None: lines = path.read_text(encoding="utf-8").splitlines() if path.exists() else [] output: list[str] = [] replaced = False for line in lines: stripped = line.strip() if stripped and not stripped.startswith("#") and "=" in stripped: current_key = stripped.split("=", 1)[0].strip() if current_key == key: output.append(f"{key}={value}") replaced = True continue output.append(line) if not replaced: output.append(f"{key}={value}") path.parent.mkdir(parents=True, exist_ok=True) path.write_text("\n".join(output).rstrip() + "\n", encoding="utf-8")