"""Configuration and secret loading for CoinHunter V2.""" from __future__ import annotations import os from pathlib import Path from typing import Any from .runtime import RuntimePaths, ensure_runtime_dirs, get_runtime_paths try: import tomllib except ModuleNotFoundError: # pragma: no cover import tomli as tomllib DEFAULT_CONFIG = """[runtime] timezone = "Asia/Shanghai" log_dir = "logs" output_format = "tui" [binance] spot_base_url = "https://api.binance.com" recv_window = 5000 [market] default_quote = "USDT" universe_allowlist = [] universe_denylist = [] [trading] spot_enabled = true dry_run_default = false dust_usdt_threshold = 10.0 [opportunity] min_quote_volume = 1000000.0 top_n = 10 scan_limit = 50 ignore_dust = true lookback_intervals = ["1h", "4h", "1d"] [opportunity.weights] trend = 1.0 momentum = 1.0 breakout = 0.8 volume = 0.7 volatility_penalty = 0.5 position_concentration_penalty = 0.6 """ DEFAULT_ENV = "BINANCE_API_KEY=\nBINANCE_API_SECRET=\n" def _permission_denied_message(paths: RuntimePaths, exc: PermissionError) -> RuntimeError: return RuntimeError( "Unable to initialize CoinHunter runtime files because the target directory is not writable: " f"{paths.root}. Set COINHUNTER_HOME to a writable directory or rerun with permissions that can write there. " f"Original error: {exc}" ) def ensure_init_files(paths: RuntimePaths | None = None, *, force: bool = False) -> dict[str, Any]: paths = paths or get_runtime_paths() try: ensure_runtime_dirs(paths) except PermissionError as exc: raise _permission_denied_message(paths, exc) from exc created: list[str] = [] updated: list[str] = [] for path, content in ((paths.config_file, DEFAULT_CONFIG), (paths.env_file, DEFAULT_ENV)): if force or not path.exists(): try: path.write_text(content, encoding="utf-8") except PermissionError as exc: raise _permission_denied_message(paths, exc) from exc (updated if force and path.exists() else created).append(str(path)) return { "root": str(paths.root), "config_file": str(paths.config_file), "env_file": str(paths.env_file), "logs_dir": str(paths.logs_dir), "created_or_updated": created + updated, "force": force, } def load_config(paths: RuntimePaths | None = None) -> dict[str, Any]: paths = paths or get_runtime_paths() if not paths.config_file.exists(): raise RuntimeError(f"Missing config file at {paths.config_file}. Run `coinhunter init` first.") return tomllib.loads(paths.config_file.read_text(encoding="utf-8")) # type: ignore[no-any-return] def load_env_file(paths: RuntimePaths | None = None) -> dict[str, str]: paths = paths or get_runtime_paths() loaded: dict[str, str] = {} if not paths.env_file.exists(): return loaded for raw_line in paths.env_file.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) key = key.strip() value = value.strip() loaded[key] = value os.environ[key] = value return loaded def get_binance_credentials(paths: RuntimePaths | None = None) -> dict[str, str]: load_env_file(paths) api_key = os.getenv("BINANCE_API_KEY", "").strip() api_secret = os.getenv("BINANCE_API_SECRET", "").strip() if not api_key or not api_secret: runtime_paths = paths or get_runtime_paths() raise RuntimeError( "Missing BINANCE_API_KEY or BINANCE_API_SECRET. " f"Populate {runtime_paths.env_file} or export them in the environment." ) return {"api_key": api_key, "api_secret": api_secret} def resolve_log_dir(config: dict[str, Any], paths: RuntimePaths | None = None) -> Path: paths = paths or get_runtime_paths() raw = config.get("runtime", {}).get("log_dir", "logs") value = Path(raw).expanduser() return value if value.is_absolute() else paths.root / value