"""Configuration loader for micro-scanner. Reads /app/config/micro.yaml, then applies environment variable overrides. Validates scoring_weights sum == 1.0 at startup โ€” aborts if not. """ from __future__ import annotations import dataclasses import os import sys from dataclasses import dataclass, field from pathlib import Path import structlog import yaml log = structlog.get_logger(__name__) _DEFAULT_CONFIG_PATH = "/app/config/micro.yaml" _WEIGHTS_TOLERANCE = 1e-9 @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 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 = 15.0 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" # "debug" | "info" 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 # --- Breakout detector + paper trading (Unit A, 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 paper_stop_loss_pct: float = 5.0 paper_starting_balance: float = 1000.0 paper_enabled: bool = True momentum_accel_min: float = ( 0.0 # minimum momentum acceleration; 0.0 preserves current behaviour ) # --- Zombie-coin filter (Layer A2) --- alive_coin_min_volume_24h: float = 500_000.0 alive_coin_min_price_range_24h_pct: float = 3.0 # --- Big-Mover feature (WU5, 2026-04-21) --- # Narrative boost categories for universe scoring narrative_boost_categories: list = field( default_factory=lambda: ["layer-1", "defi", "layer-2"] ) # Watchlist filter thresholds watchlist_filter_min_mc: float = 8_000_000.0 watchlist_filter_max_mc: float = 120_000_000.0 watchlist_filter_vol_mc_ratio: float = 0.15 watchlist_filter_circ_supply_ratio: float = 0.35 watchlist_filter_mc_bypass_threshold: float = 20_000_000.0 # Gate thresholds gate_volmc_spot_ratio_threshold: float = 0.15 gate_volmc_prior_avg_max: float = 0.05 gate_squeeze_min_candles: int = 4 gate_squeeze_vol_pct: float = 0.40 gate_depth_slope_min: float = 0.01 gate_derivatives_futures_spot_ratio: float = 3.0 gate_derivatives_oi_pct_threshold: float = 15.0 gate_derivatives_funding_threshold: float = -0.0005 gate_derivatives_score_bonus: float = 0.15 gate_tvl_change_threshold: float = 30.0 gate_tvl_score_bonus: float = 0.10 gate_ath_risk_exit_multiplier: float = 0.75 # Regime thresholds regime_fear_greed_pause: float = 25.0 regime_btc_price_change_halt: float = 5.0 regime_eth_btc_slope_reduced: float = 0.0 regime_btc_dom_advisory: float = 60.0 regime_reduced_score_threshold: float = 0.75 def _load_yaml(path: Path) -> dict: """Load YAML file, returning empty dict if missing.""" if path.exists(): with open(path) as f: return yaml.safe_load(f) or {} return {} def _validate_scoring_weights(weights: dict) -> None: """Abort if scoring weights do not sum to 1.0.""" total = sum(weights.values()) if abs(total - 1.0) > _WEIGHTS_TOLERANCE: log.critical( "scoring_weights_invalid", weights=weights, total=total, expected=1.0, ) sys.exit(1) KNOWN_KEYS: frozenset[str] = frozenset(f.name for f in dataclasses.fields(MicroConfig)) def load_micro_config( config_path: str = _DEFAULT_CONFIG_PATH, variant_name: str = "default", ) -> MicroConfig: """Load MicroConfig from YAML file + environment variable overrides. Environment variables take precedence over YAML. scoring_weights sum is validated; process aborts if invalid. Unknown YAML keys emit a warning (never silently dropped without notice). """ yaml_data = _load_yaml(Path(config_path)) # Flatten nested YAML sections (regime, watchlist_filter, gates) into the # top-level flat MicroConfig namespace. These sections are purely # organisational in YAML โ€” the loader maps them to the flat field names. _regime = yaml_data.pop("regime", {}) or {} if _regime.get("fear_greed_pause") is not None: yaml_data.setdefault("regime_fear_greed_pause", _regime["fear_greed_pause"]) if _regime.get("btc_price_change_halt_pct") is not None: yaml_data.setdefault( "regime_btc_price_change_halt", _regime["btc_price_change_halt_pct"] ) elif _regime.get("btc_vol_halt_pct") is not None: # Legacy YAML key preserved for back-compat; mapped to the new field. yaml_data.setdefault( "regime_btc_price_change_halt", _regime["btc_vol_halt_pct"] ) if _regime.get("btc_dom_advisory") is not None: yaml_data.setdefault("regime_btc_dom_advisory", _regime["btc_dom_advisory"]) if _regime.get("eth_btc_slope_reduced") is not None: yaml_data.setdefault( "regime_eth_btc_slope_reduced", _regime["eth_btc_slope_reduced"] ) if _regime.get("reduced_score_threshold") is not None: yaml_data.setdefault( "regime_reduced_score_threshold", _regime["reduced_score_threshold"] ) _wf = yaml_data.pop("watchlist_filter", {}) or {} if _wf.get("min_mc") is not None: yaml_data.setdefault("watchlist_filter_min_mc", _wf["min_mc"]) if _wf.get("max_mc") is not None: yaml_data.setdefault("watchlist_filter_max_mc", _wf["max_mc"]) if _wf.get("vol_mc_ratio") is not None: yaml_data.setdefault("watchlist_filter_vol_mc_ratio", _wf["vol_mc_ratio"]) if _wf.get("circ_supply_ratio") is not None: yaml_data.setdefault( "watchlist_filter_circ_supply_ratio", _wf["circ_supply_ratio"] ) if _wf.get("mc_bypass_threshold") is not None: yaml_data.setdefault( "watchlist_filter_mc_bypass_threshold", _wf["mc_bypass_threshold"] ) _gates = yaml_data.pop("gates", {}) or {} _gate_map = { "volmc_spot_ratio_threshold": "gate_volmc_spot_ratio_threshold", "volmc_prior_avg_max": "gate_volmc_prior_avg_max", "squeeze_min_candles": "gate_squeeze_min_candles", "squeeze_vol_pct": "gate_squeeze_vol_pct", "depth_slope_min": "gate_depth_slope_min", "derivatives_futures_spot_ratio": "gate_derivatives_futures_spot_ratio", "derivatives_oi_pct_threshold": "gate_derivatives_oi_pct_threshold", "derivatives_funding_threshold": "gate_derivatives_funding_threshold", "derivatives_score_bonus": "gate_derivatives_score_bonus", "tvl_change_threshold": "gate_tvl_change_threshold", "tvl_score_bonus": "gate_tvl_score_bonus", "ath_risk_exit_multiplier": "gate_ath_risk_exit_multiplier", } for yaml_key, field_name in _gate_map.items(): if _gates.get(yaml_key) is not None: yaml_data.setdefault(field_name, _gates[yaml_key]) # Warn on unknown keys before filtering โ€” catches typos in overlay YAML. unknown = set(yaml_data.keys()) - KNOWN_KEYS if unknown: log.warning( "unknown_config_keys", keys=sorted(unknown), variant=variant_name, ) # Build config from YAML, filtering to known fields known = {k: v for k, v in yaml_data.items() if k in KNOWN_KEYS} config = MicroConfig(**known) # Environment variable overrides (typed coercions) env_overrides: dict[str, object] = {} def _bool(s: str) -> bool: return s.lower() in ("1", "true", "yes") if (v := os.getenv("MICRO_ENABLED")) is not None: env_overrides["enabled"] = _bool(v) if (v := os.getenv("MICRO_UNIVERSE_REFRESH_SECONDS")) is not None: env_overrides["universe_refresh_seconds"] = int(v) if (v := os.getenv("MICRO_WATCHLIST_MAX")) is not None: env_overrides["watchlist_max"] = int(v) if (v := os.getenv("MICRO_SCORE_ENTRY_THRESHOLD")) is not None: env_overrides["score_entry_threshold"] = float(v) if (v := os.getenv("MICRO_SCORE_FULL_POSITION")) is not None: env_overrides["score_full_position"] = float(v) if (v := os.getenv("MICRO_SIGNAL_MAX_AGE_SECONDS")) is not None: env_overrides["signal_max_age_seconds"] = int(v) if (v := os.getenv("MICRO_VOLUME_SURGE_THRESHOLD")) is not None: env_overrides["volume_surge_threshold"] = float(v) if (v := os.getenv("MICRO_SPREAD_MAX_PCT")) is not None: env_overrides["spread_max_pct"] = float(v) if (v := os.getenv("MICRO_MAX_SIMULTANEOUS")) is not None: env_overrides["max_simultaneous"] = int(v) if (v := os.getenv("MICRO_MAX_TOTAL_EXPOSURE_PCT")) is not None: env_overrides["max_total_exposure_pct"] = float(v) if (v := os.getenv("MICRO_SCORE_CACHE_TTL_SECONDS")) is not None: env_overrides["score_cache_ttl_seconds"] = int(v) if (v := os.getenv("MICRO_SCORE_CACHE_MAXSIZE")) is not None: env_overrides["score_cache_maxsize"] = int(v) if (v := os.getenv("MICRO_CIRCUIT_BREAKER_LOSS_PCT")) is not None: env_overrides["circuit_breaker_loss_pct"] = float(v) # v2 env overrides if (v := os.getenv("MICRO_FLOOD_THRESHOLD")) is not None: env_overrides["flood_threshold"] = int(v) if (v := os.getenv("MICRO_FLOOD_PROMOTE_TOP_N")) is not None: env_overrides["flood_promote_top_n"] = int(v) if (v := os.getenv("MICRO_HYSTERESIS_BONUS")) is not None: env_overrides["hysteresis_bonus"] = float(v) if (v := os.getenv("MICRO_DEPTH_GATE_ENABLED")) is not None: env_overrides["depth_gate_enabled"] = _bool(v) if (v := os.getenv("MICRO_ENTRY_REJECTION_LOG_LEVEL")) is not None: env_overrides["entry_rejection_log_level"] = v if (v := os.getenv("MICRO_GATE_ROLLUP_EVERY_N_SCANS")) is not None: env_overrides["gate_rollup_every_n_scans"] = int(v) if (v := os.getenv("MICRO_AUDIT_LOG_REJECTIONS")) is not None: env_overrides["audit_log_rejections"] = _bool(v) if (v := os.getenv("MICRO_DRY_RUN_SIGNALS_ONLY")) is not None: env_overrides["dry_run_signals_only"] = _bool(v) if (v := os.getenv("MICRO_COMPOSITE_OBSERVER_ENABLED")) is not None: env_overrides["composite_observer_enabled"] = _bool(v) if (v := os.getenv("MICRO_COMPOSITE_OBSERVER_THRESHOLD")) is not None: env_overrides["composite_observer_threshold"] = float(v) if (v := os.getenv("MICRO_CONFIG_VERSION")) is not None: env_overrides["config_version"] = int(v) if (v := os.getenv("MICRO_SCAN_DEADLINE_SECONDS")) is not None: env_overrides["scan_deadline_seconds"] = float(v) if (v := os.getenv("MICRO_MAX_REJECTION_LINES_PER_SCAN")) is not None: env_overrides["max_rejection_lines_per_scan"] = int(v) # Breakout detector + paper trading overrides if (v := os.getenv("MICRO_BREAKOUT_1H_MIN_PCT")) is not None: env_overrides["breakout_1h_min_pct"] = float(v) if (v := os.getenv("MICRO_BREAKOUT_1H_MAX_PCT")) is not None: env_overrides["breakout_1h_max_pct"] = float(v) if (v := os.getenv("MICRO_BREAKOUT_2H_MAX_PCT")) is not None: env_overrides["breakout_2h_max_pct"] = float(v) if (v := os.getenv("MICRO_BREAKOUT_RSI_MAX")) is not None: env_overrides["breakout_rsi_max"] = float(v) if (v := os.getenv("MICRO_TRAIL_STOP_PCT")) is not None: env_overrides["trail_stop_pct"] = float(v) if (v := os.getenv("MICRO_RSI_OVERBOUGHT")) is not None: env_overrides["rsi_overbought"] = float(v) if (v := os.getenv("MICRO_RSI_REVERSAL")) is not None: env_overrides["rsi_reversal"] = float(v) if (v := os.getenv("MICRO_MAX_HOLD_HOURS")) is not None: env_overrides["max_hold_hours"] = int(v) if (v := os.getenv("MICRO_PAPER_STOP_LOSS_PCT")) is not None: env_overrides["paper_stop_loss_pct"] = float(v) if (v := os.getenv("MICRO_PAPER_STARTING_BALANCE")) is not None: env_overrides["paper_starting_balance"] = float(v) if (v := os.getenv("MICRO_PAPER_ENABLED")) is not None: env_overrides["paper_enabled"] = _bool(v) if (v := os.getenv("MICRO_MOMENTUM_ACCEL_MIN")) is not None: env_overrides["momentum_accel_min"] = float(v) # Zombie-coin filter env overrides if (v := os.getenv("MICRO_ALIVE_COIN_MIN_VOLUME_24H")) is not None: env_overrides["alive_coin_min_volume_24h"] = float(v) if (v := os.getenv("MICRO_ALIVE_COIN_MIN_PRICE_RANGE_24H_PCT")) is not None: env_overrides["alive_coin_min_price_range_24h_pct"] = float(v) # Big-Mover feature env overrides if (v := os.getenv("MICRO_WATCHLIST_FILTER_MIN_MC")) is not None: env_overrides["watchlist_filter_min_mc"] = float(v) if (v := os.getenv("MICRO_WATCHLIST_FILTER_MAX_MC")) is not None: env_overrides["watchlist_filter_max_mc"] = float(v) if (v := os.getenv("MICRO_WATCHLIST_FILTER_VOL_MC_RATIO")) is not None: env_overrides["watchlist_filter_vol_mc_ratio"] = float(v) if (v := os.getenv("MICRO_WATCHLIST_FILTER_CIRC_SUPPLY_RATIO")) is not None: env_overrides["watchlist_filter_circ_supply_ratio"] = float(v) if (v := os.getenv("MICRO_WATCHLIST_FILTER_MC_BYPASS_THRESHOLD")) is not None: env_overrides["watchlist_filter_mc_bypass_threshold"] = float(v) if (v := os.getenv("MICRO_GATE_VOLMC_SPOT_RATIO_THRESHOLD")) is not None: env_overrides["gate_volmc_spot_ratio_threshold"] = float(v) if (v := os.getenv("MICRO_GATE_VOLMC_PRIOR_AVG_MAX")) is not None: env_overrides["gate_volmc_prior_avg_max"] = float(v) if (v := os.getenv("MICRO_GATE_SQUEEZE_MIN_CANDLES")) is not None: env_overrides["gate_squeeze_min_candles"] = int(v) if (v := os.getenv("MICRO_GATE_SQUEEZE_VOL_PCT")) is not None: env_overrides["gate_squeeze_vol_pct"] = float(v) if (v := os.getenv("MICRO_GATE_DEPTH_SLOPE_MIN")) is not None: env_overrides["gate_depth_slope_min"] = float(v) if (v := os.getenv("MICRO_GATE_DERIVATIVES_FUTURES_SPOT_RATIO")) is not None: env_overrides["gate_derivatives_futures_spot_ratio"] = float(v) if (v := os.getenv("MICRO_GATE_DERIVATIVES_OI_PCT_THRESHOLD")) is not None: env_overrides["gate_derivatives_oi_pct_threshold"] = float(v) if (v := os.getenv("MICRO_GATE_DERIVATIVES_FUNDING_THRESHOLD")) is not None: env_overrides["gate_derivatives_funding_threshold"] = float(v) if (v := os.getenv("MICRO_GATE_DERIVATIVES_SCORE_BONUS")) is not None: env_overrides["gate_derivatives_score_bonus"] = float(v) if (v := os.getenv("MICRO_GATE_TVL_CHANGE_THRESHOLD")) is not None: env_overrides["gate_tvl_change_threshold"] = float(v) if (v := os.getenv("MICRO_GATE_TVL_SCORE_BONUS")) is not None: env_overrides["gate_tvl_score_bonus"] = float(v) if (v := os.getenv("MICRO_GATE_ATH_RISK_EXIT_MULTIPLIER")) is not None: env_overrides["gate_ath_risk_exit_multiplier"] = float(v) if (v := os.getenv("MICRO_REGIME_FEAR_GREED_PAUSE")) is not None: env_overrides["regime_fear_greed_pause"] = float(v) if (v := os.getenv("MICRO_REGIME_BTC_PRICE_CHANGE_HALT")) is not None: env_overrides["regime_btc_price_change_halt"] = float(v) elif (v := os.getenv("MICRO_REGIME_BTC_VOL_HALT")) is not None: # Legacy env var preserved for back-compat. env_overrides["regime_btc_price_change_halt"] = float(v) if (v := os.getenv("MICRO_REGIME_ETH_BTC_SLOPE_REDUCED")) is not None: env_overrides["regime_eth_btc_slope_reduced"] = float(v) if (v := os.getenv("MICRO_REGIME_BTC_DOM_ADVISORY")) is not None: env_overrides["regime_btc_dom_advisory"] = float(v) if (v := os.getenv("MICRO_REGIME_REDUCED_SCORE_THRESHOLD")) is not None: env_overrides["regime_reduced_score_threshold"] = float(v) for k, val in env_overrides.items(): setattr(config, k, val) _validate_scoring_weights(config.scoring_weights) log.info( "micro_config_loaded", config_path=config_path, enabled=config.enabled, watchlist_max=config.watchlist_max, score_full_position=config.score_full_position, # v2 fields (Rule 5: no silent defaults) config_version=config.config_version, depth_gate_enabled=config.depth_gate_enabled, flood_threshold=config.flood_threshold, flood_promote_top_n=config.flood_promote_top_n, hysteresis_bonus=config.hysteresis_bonus, entry_rejection_log_level=config.entry_rejection_log_level, gate_rollup_every_n_scans=config.gate_rollup_every_n_scans, audit_log_rejections=config.audit_log_rejections, dry_run_signals_only=config.dry_run_signals_only, composite_observer_enabled=config.composite_observer_enabled, composite_observer_threshold=config.composite_observer_threshold, scan_deadline_seconds=config.scan_deadline_seconds, max_rejection_lines_per_scan=config.max_rejection_lines_per_scan, breakout_1h_min_pct=config.breakout_1h_min_pct, breakout_1h_max_pct=config.breakout_1h_max_pct, breakout_2h_max_pct=config.breakout_2h_max_pct, breakout_rsi_max=config.breakout_rsi_max, trail_stop_pct=config.trail_stop_pct, rsi_overbought=config.rsi_overbought, rsi_reversal=config.rsi_reversal, max_hold_hours=config.max_hold_hours, paper_stop_loss_pct=config.paper_stop_loss_pct, paper_starting_balance=config.paper_starting_balance, paper_enabled=config.paper_enabled, momentum_accel_min=config.momentum_accel_min, ) return config # Module-level mode: authoritative toggle from environment (Architecture Design ยง6.1) def get_sigil_mode() -> str: """Return SIGIL_MODE from env; authoritative source for draft/live toggle.""" mode = os.getenv("SIGIL_MODE", "draft").strip().lower() if mode not in ("draft", "live"): log.warning("sigil_mode_invalid_defaulting_draft", raw_mode=mode) return "draft" return mode