""" config_loader.py Single source of truth for reading config.json and .env credentials. Every other module imports get_config() / get_env() from here instead of touching os.environ or config.json directly. This keeps parameters out of source code, per project requirement. """ import json import os from pathlib import Path from functools import lru_cache BASE_DIR = Path(__file__).resolve().parent CONFIG_PATH = BASE_DIR / "config.json" ENV_PATH = BASE_DIR / ".env" def _load_dotenv(path: Path): """Minimal .env loader (no external dependency required).""" if not path.exists(): return with open(path, "r") as f: for line in f: line = line.strip() if not line or line.startswith("#") or "=" not in line: continue key, _, value = line.partition("=") key = key.strip() value = value.strip().strip('"').strip("'") # Don't overwrite real environment variables if already set os.environ.setdefault(key, value) _load_dotenv(ENV_PATH) @lru_cache(maxsize=1) def get_config() -> dict: with open(CONFIG_PATH, "r") as f: return json.load(f) def reload_config() -> dict: """Force re-read of config.json (useful if tuning live without restart).""" get_config.cache_clear() return get_config() class EnvConfig: """Wraps required Alpaca credentials with clear failure if missing.""" def __init__(self): self.api_key = os.environ.get("ALPACA_API_KEY") self.secret_key = os.environ.get("ALPACA_SECRET_KEY") self.base_url = os.environ.get("ALPACA_BASE_URL", "https://paper-api.alpaca.markets") self.data_url = os.environ.get("ALPACA_DATA_URL", "https://data.alpaca.markets") def validate(self): missing = [] if not self.api_key: missing.append("ALPACA_API_KEY") if not self.secret_key: missing.append("ALPACA_SECRET_KEY") if missing: raise EnvironmentError( f"Missing required environment variables: {', '.join(missing)}. " f"Copy .env.example to .env and fill in your Alpaca credentials." ) @lru_cache(maxsize=1) def get_env() -> EnvConfig: env = EnvConfig() return env def cfg(*keys, default=None): """ Convenience getter for nested config values. Example: cfg("stop", "trailing_distance", default=0.10) """ node = get_config() for k in keys: if not isinstance(node, dict) or k not in node: return default node = node[k] return node class ConfigValidationError(ValueError): """Raised by validate_config() when config.json contains a value that can never be legitimate (e.g. a lookback window of 0 bars, a negative percentage). Collects every problem found, not just the first, so a single startup failure surfaces the whole list instead of a fix-one-rerun-find-the-next loop.""" # [FEATURE 2026-09-01] Deliberately narrow, deliberately not exhaustive. # Per explicit request, this is NOT meant to make entry tuning more # restrictive -- it only catches values that are unambiguously broken # (a typo'd 0 or negative number in a field that is used as a divisor, # a bar-count, a lookback window, or a percentage/ratio that must be a # positive magnitude to mean anything). It deliberately does NOT touch # fields that legitimately use 0 as a real, documented "disabled" # sentinel (e.g. trading.same_symbol_reentry_cooldown_minutes, # entry.min_extension_from_vwap_pct) or fields that can legitimately be # zero by design (e.g. intraday_health.price_slope_negative_credit, # any *_weights entry meant to turn a factor off) -- those are excluded # below on purpose, not missed. It also deliberately does NOT check # cross-field ordering (e.g. min < max) or the time_based_overrides # windows -- those were considered and explicitly scoped out. _STRICTLY_POSITIVE_PATHS = [ ("universe", "price_min"), ("universe", "price_max"), ("universe", "min_avg_daily_volume"), ("candidates", "premarket_candidate_count"), ("candidates", "final_candidate_count"), ("schedule", "poll_interval_seconds_intraday"), ("schedule", "eod_liquidation_max_wait_seconds"), ("premarket_scoring", "min_premarket_volume"), ("premarket_scoring", "min_rvol"), ("premarket_scoring", "vwap_lookback_bars"), ("premarket_scoring", "momentum_lookback_snapshots"), ("premarket_scoring", "acceleration_min_ratio"), ("intraday_health", "eval_interval_seconds"), ("intraday_health", "slope_lookback_bars"), ("intraday_health", "structure_lookback_bars"), ("intraday_health", "atr_period"), ("intraday_health", "healthy_confirm_reads"), ("intraday_health", "unhealthy_confirm_reads"), ("intraday_health", "remove_after_consecutive_unhealthy"), ("intraday_health", "full_rescan_interval_minutes"), ("intraday_health", "full_rescan_pool_size"), ("intraday_health", "full_rescan_lookback_hours"), ("intraday_health", "entry_shortlist_size"), ("intraday_health", "volume_participation_rvol_target"), ("intraday_health", "max_consecutive_confirmation_failures"), ("intraday_health", "confirmation_failure_cooldown_seconds"), ("entry", "vwap_rising_lookback_bars"), ("entry", "opening_volume_expansion_ratio"), ("entry", "breakout_hold_bars"), ("entry", "higher_lows_lookback_bars"), ("entry", "max_spread_pct_entry"), ("entry", "max_extension_from_vwap_pct"), ("entry", "min_confirmation_score"), ("entry", "rsi_period"), ("entry", "cancel_unfilled_after_seconds"), ("risk", "account_risk_pct_per_trade"), ("risk", "max_position_notional_pct_of_equity"), ("risk", "min_shares"), ("risk", "max_daily_loss_pct"), ("risk", "max_consecutive_losses_pause"), ("stop", "initial_distance"), ("stop", "trailing_distance"), ("stop", "percentage_initial_pct"), ("stop", "percentage_trailing_pct"), ("stop", "atr_period"), ("stop", "atr_multiplier_initial"), ("stop", "atr_multiplier_trailing"), ("stop", "min_stop_distance_cents"), ("stop", "max_stop_distance_pct"), ("trading", "max_positions"), ("trading", "simulated_equity_default"), ] # Weight dicts: each entry must be >= 0 (a factor can legitimately be # turned off with 0, but a negative weight would silently invert a # scoring component, which can never be intentional). _NONNEGATIVE_WEIGHT_DICTS = [ ("premarket_scoring", "weights"), ("intraday_health", "weights"), ] def _get_path(cfg: dict, path: tuple): node = cfg for key in path: if not isinstance(node, dict) or key not in node: return None, False node = node[key] return node, True def validate_config(cfg: dict = None) -> None: """ Fail-fast sanity check, run once at startup (see monitor.py's SessionOrchestrator.__init__ and dry_run_check.py). Raises ConfigValidationError listing every problem found if config.json contains a value that can never be legitimate. Deliberately narrow in scope -- see _STRICTLY_POSITIVE_PATHS's comment above for what this does and does not check, and why. """ cfg = cfg if cfg is not None else get_config() problems = [] for path in _STRICTLY_POSITIVE_PATHS: value, present = _get_path(cfg, path) if not present: continue # an absent key is a KeyError elsewhere, not this function's job if not isinstance(value, (int, float)) or isinstance(value, bool): problems.append(f"{'.'.join(path)} = {value!r} is not a number") elif value <= 0: problems.append(f"{'.'.join(path)} = {value!r} must be > 0") for path in _NONNEGATIVE_WEIGHT_DICTS: node, present = _get_path(cfg, path) if not present or not isinstance(node, dict): continue for key, value in node.items(): full_path = ".".join(path) + f".{key}" if not isinstance(value, (int, float)) or isinstance(value, bool): problems.append(f"{full_path} = {value!r} is not a number") elif value < 0: problems.append(f"{full_path} = {value!r} must be >= 0") if problems: raise ConfigValidationError( f"config.json has {len(problems)} invalid value(s):\n " + "\n ".join(problems) )