"""Configuration loader with HMAC tamper detection.""" from __future__ import annotations import hashlib import hmac import os from dataclasses import dataclass, field, fields from pathlib import Path import yaml @dataclass class ExchangeConfig: exchange_id: str = "binance" api_key: str = "" api_secret: str = "" @dataclass class CoreConfig: symbols: list[str] = field( default_factory=lambda: ["BTCUSDT", "ETHUSDT", "SOLUSDT"] ) max_position_pct: float = 20.0 max_total_exposure_pct: float = 50.0 min_cash_reserve_pct: float = 20.0 stop_loss_pct: float = 8.0 take_profit_pct: float = 15.0 partial_exit_pct: float = 10.0 reentry_cooldown_hours: int = 2 draft_starting_capital: float = 50.0 min_confidence: float = 0.65 high_confidence: float = 0.85 signal_interval_minutes: int = 15 market_trend_interval_minutes: int = 60 reconciliation_interval_minutes: int = 5 # Rules strategy tuning (T1.5 — relaxed for paper trading) rules_rsi_oversold: float = 40.0 # was 30.0 rules_volume_mult: float = 1.2 # was 1.5 rules_buy_min_score: int = 2 # was 3 (all conditions) @dataclass class PumpConfig: universe_size: int = 200 scan_interval_seconds: int = 30 volume_spike_multiplier: float = 5.0 price_rise_pct: float = 15.0 detection_window_minutes: int = 10 min_confidence: float = 0.75 max_position_pct: float = 5.0 max_simultaneous: int = 2 max_total_exposure_pct: float = 10.0 stop_loss_pct: float = 8.0 take_profit_pct: float = 10.0 max_hold_minutes: int = 60 reentry_cooldown_hours: int = 4 @dataclass class MicroConfig: enabled: bool = False universe_refresh_seconds: int = 60 watchlist_max: int = 30 score_entry_threshold: float = 0.55 score_full_position: float = 0.70 signal_max_age_seconds: int = 30 min_consecutive_candles: int = 3 volume_surge_threshold: float = 3.0 spread_max_pct: float = 0.3 rsi_entry_min: float = 40.0 rsi_entry_max: float = 70.0 depth_imbalance_ratio: float = 1.3 resistance_threshold_pct: float = 3.0 antispoof_samples: int = 3 antispoof_window_ms: int = 2000 antispoof_blacklist_minutes: int = 5 rsi_exit_accelerator: float = 70.0 take_profit_pct: float = 5.0 partial_tp_pct: float = 3.5 partial_tp_fraction: float = 0.6 stop_loss_pct: float = 3.0 stop_loss_limit_buffer_pct: float = 0.35 time_exit_minutes: int = 20 dead_man_max_hold_hours: int = 4 order_type: str = "limit_maker" fill_window_seconds: int = 30 cooldown_after_unfilled_minutes: int = 30 reentry_cooldown_hours: float = 2.0 cancel_max_retries: int = 3 cancel_backoff_seconds: int = 2 sl_attachment_max_retries: int = 3 sl_attachment_backoff_seconds: int = 2 max_simultaneous: int = 10 # capped, actual = floor(portfolio/$15) max_total_exposure_pct: float = 15.0 base_position_pct: float = 5.0 half_position_pct: float = 2.5 min_notional_usd: float = 5.0 max_position_pct: float = 10.0 capital_pct: float = 40.0 # % of portfolio reserved for micro ttl_minutes: int = 25 score_cache_ttl_seconds: int = 90 score_cache_maxsize: int = 50 scoring_weights: dict = field( default_factory=lambda: { "volatility": 0.25, "momentum": 0.35, "liquidity": 0.25, "trend": 0.15, } ) circuit_breaker_loss_pct: float = 8.0 # --- v2 fields (Sigil Unhinged rethink, 2026-04-19) --- flood_threshold: int = 20 flood_promote_top_n: int = 5 hysteresis_bonus: float = 0.10 depth_gate_enabled: bool = True entry_rejection_log_level: str = "debug" gate_rollup_every_n_scans: int = 1 audit_log_rejections: bool = False dry_run_signals_only: bool = True composite_observer_enabled: bool = True composite_observer_threshold: float = 0.55 config_version: int = 1 scan_deadline_seconds: float = 50.0 max_rejection_lines_per_scan: int = 20 # --- v3 fields: breakout detection + paper trading (2026-04-20) --- breakout_1h_min_pct: float = 5.0 breakout_1h_max_pct: float = 20.0 breakout_2h_max_pct: float = 35.0 breakout_rsi_max: float = 65.0 trail_stop_pct: float = 10.0 rsi_overbought: float = 72.0 rsi_reversal: float = 68.0 max_hold_hours: int = 6 stop_loss_pct: float = 5.0 paper_stop_loss_pct: float = 5.0 paper_starting_balance: float = 1000.0 paper_enabled: bool = True momentum_accel_min: float = 0.0 def __post_init__(self) -> None: total = sum(self.scoring_weights.values()) if abs(total - 1.0) > 1e-9: raise ValueError( f"MicroConfig.scoring_weights must sum to 1.0, got {total}" ) @dataclass class RiskConfig: daily_loss_limit_pct: float = 5.0 max_drawdown_pct: float = 15.0 max_trades_per_hour: int = 10 min_cash_reserve_pct: float = 20.0 max_total_exposure_pct: float = 80.0 price_staleness_seconds: int = 30 signal_staleness_pct: float = 2.0 api_error_consecutive_limit: int = 5 api_error_cooldown_seconds: int = 60 api_error_max_backoff_seconds: int = 300 recovery_position_scale: float = 0.5 recovery_trades_required: int = 3 kill_switch_path: str = "/opt/app/data/kill_switch.flag" admin_api_bind: str = "###_IP" admin_api_port: int = 8080 @dataclass class LLMConfig: enabled: bool = True max_calls_per_day: int = 50 timeout_seconds: int = 30 sentiment_interval_hours: int = 4 @dataclass class MLConfig: enabled: bool = False model_path: str = "" label_horizon_hours: int = 4 label_buy_threshold: float = 0.015 label_sell_threshold: float = -0.015 @dataclass class SigilConfig: mode: str = "draft" # "draft" or "live" exchange: ExchangeConfig = field(default_factory=ExchangeConfig) core: CoreConfig = field(default_factory=CoreConfig) pump: PumpConfig = field(default_factory=PumpConfig) risk: RiskConfig = field(default_factory=RiskConfig) llm: LLMConfig = field(default_factory=LLMConfig) micro: MicroConfig = field(default_factory=MicroConfig) ml: MLConfig = field(default_factory=MLConfig) admin_api_key: str = "" model_signing_key: str = "" config_hmac_key: str = "" smtp_host: str = "" smtp_port: int = 25 email_to: str = "" email_from: str = "" db_path: str = "/app/data/sigil.db" heartbeat_path: str = "/app/data/bot_heartbeat" data_dir: str = "/app/data" variant: str = "baseline" def _load_yaml(path: Path) -> dict: """Load a YAML config file.""" if path.exists(): with open(path) as f: return yaml.safe_load(f) or {} return {} def load_config(config_dir: str = "config") -> SigilConfig: """Load configuration from YAML files and environment variables. Environment variables override YAML values. Secrets come exclusively from environment (secrets.env loaded by Docker/dotenv). """ config_path = Path(config_dir) core_yaml = _load_yaml(config_path / "core.yaml") # Grid overlay: merge variant-specific config on top of core/pump/micro.yaml # Supports flat keys (applied to CoreConfig) and nested pump:/micro: dicts. overlay_path = os.environ.get("SIGIL_CONFIG_OVERLAY", "") overlay_pump: dict = {} overlay_micro: dict = {} overlay_risk: dict = {} if overlay_path and Path(overlay_path).exists(): overlay = _load_yaml(Path(overlay_path)) overlay_pump = overlay.pop("pump", {}) or {} overlay_micro = overlay.pop("micro", {}) or {} overlay_risk = overlay.pop("risk", {}) or {} core_yaml.update(overlay) pump_yaml = _load_yaml(config_path / "pump.yaml") pump_yaml.update(overlay_pump) risk_yaml = _load_yaml(config_path / "risk.yaml") risk_yaml.update(overlay_risk) micro_yaml = _load_yaml(config_path / "micro.yaml") micro_yaml.update(overlay_micro) # ML config: read from ml.yaml if present, else from core.yaml ml: block (E3 fix) ml_yaml_file = _load_yaml(config_path / "ml.yaml") ml_yaml_core = ( core_yaml.get("ml", {}) if isinstance(core_yaml.get("ml"), dict) else {} ) ml_yaml = {**ml_yaml_core, **ml_yaml_file} def _pick(cls, data: dict) -> dict: """Filter dict keys to only valid dataclass field names.""" valid = {f.name for f in fields(cls)} return {k: v for k, v in data.items() if k in valid} core = CoreConfig(**_pick(CoreConfig, core_yaml)) pump = PumpConfig(**_pick(PumpConfig, pump_yaml)) risk = RiskConfig(**_pick(RiskConfig, risk_yaml)) micro = MicroConfig(**_pick(MicroConfig, micro_yaml)) ml = MLConfig(**_pick(MLConfig, ml_yaml)) mode = os.getenv("SIGIL_MODE", "draft") if mode not in ("draft", "live"): raise ValueError(f"SIGIL_MODE must be 'draft' or 'live', got: {mode!r}") exchange = ExchangeConfig( exchange_id=os.getenv("EXCHANGE_ID", "binance"), api_key=os.getenv("EXCHANGE_API_KEY", ""), api_secret=os.getenv("EXCHANGE_API_SECRET", ""), ) llm = LLMConfig( enabled=os.getenv("LLM_ENABLED", "true").lower() == "true", max_calls_per_day=int(os.getenv("LLM_MAX_CALLS_PER_DAY", "50")), timeout_seconds=int(os.getenv("LLM_TIMEOUT_SECONDS", "30")), ) return SigilConfig( mode=mode, exchange=exchange, core=core, pump=pump, risk=risk, llm=llm, micro=micro, ml=ml, admin_api_key=os.getenv("ADMIN_API_KEY", ""), model_signing_key=os.getenv("MODEL_SIGNING_KEY", ""), config_hmac_key=os.getenv("CONFIG_HMAC_KEY", ""), smtp_host=os.getenv("SMTP_HOST", ""), smtp_port=int(os.getenv("SMTP_PORT", "25")), email_to=os.getenv("EMAIL_TO", ""), email_from=os.getenv("EMAIL_FROM", ""), db_path=os.getenv("SIGIL_DB_PATH", "/app/data/sigil.db"), heartbeat_path=os.getenv("SIGIL_HEARTBEAT_PATH", "/app/data/bot_heartbeat"), data_dir=os.getenv("SIGIL_DATA_DIR", "/app/data"), variant=os.getenv("SIGIL_VARIANT", "baseline"), ) def verify_config_hmac(config_path: Path, hmac_key: str) -> bool: """Verify YAML config files haven't been tampered with.""" if not hmac_key: return True # Skip if no key configured content = b"" for name in sorted(["core.yaml", "micro.yaml", "pump.yaml", "risk.yaml"]): filepath = config_path / name if filepath.exists(): content += filepath.read_bytes() expected_path = config_path / ".config.hmac" if not expected_path.exists(): return False expected = expected_path.read_text().strip() computed = hmac.new(hmac_key.encode(), content, hashlib.sha256).hexdigest() return hmac.compare_digest(computed, expected) def sign_config(config_path: Path, hmac_key: str) -> None: """Generate HMAC for config files.""" content = b"" for name in sorted(["core.yaml", "micro.yaml", "pump.yaml", "risk.yaml"]): filepath = config_path / name if filepath.exists(): content += filepath.read_bytes() computed = hmac.new(hmac_key.encode(), content, hashlib.sha256).hexdigest() (config_path / ".config.hmac").write_text(computed + "\n")