| """
|
| config.py — Single source of truth for all configuration.
|
| Validates .env on import. Exits on any failure with a clear message.
|
| Run `python config.py migrate` to update .env with new keys.
|
| """
|
| from __future__ import annotations
|
|
|
| import logging
|
| import os
|
| import re
|
| import sys
|
| from pathlib import Path
|
|
|
| from dotenv import dotenv_values, load_dotenv, set_key
|
|
|
| from contracts import ConfigError
|
|
|
| logger = logging.getLogger("trading_system.config")
|
|
|
|
|
|
|
| EXPECTED_SCHEMA_VERSION = 2
|
| BASE_DIR = Path(__file__).resolve().parent
|
| ENV_FILE = BASE_DIR / ".env"
|
| ENV_EXAMPLE = BASE_DIR / ".env.example"
|
|
|
|
|
|
|
| load_dotenv(ENV_FILE, override=True)
|
|
|
|
|
| def _get(key: str, cast=str, default=None):
|
| """Get env var with type casting."""
|
| val = os.getenv(key)
|
| if val is None or val.strip() == "":
|
| if default is not None:
|
| return default
|
| return None
|
| val = val.strip()
|
| if cast is bool:
|
| return val.lower() in ("true", "1", "yes")
|
| return cast(val)
|
|
|
|
|
| def _require(key: str, cast=str):
|
| """Get required env var or exit."""
|
| val = _get(key, cast)
|
| if val is None:
|
| print(f"MISSING ENV KEY: {key}")
|
| sys.exit(1)
|
| return val
|
|
|
|
|
|
|
|
|
| def _parse_example_keys() -> set[str]:
|
| """Extract all KEY names from .env.example."""
|
| keys = set()
|
| if not ENV_EXAMPLE.exists():
|
| return keys
|
| for line in ENV_EXAMPLE.read_text(encoding="utf-8").splitlines():
|
| line = line.strip()
|
| if not line or line.startswith("#"):
|
| continue
|
| match = re.match(r"^([A-Z_][A-Z0-9_]*)=", line)
|
| if match:
|
| keys.add(match.group(1))
|
| return keys
|
|
|
|
|
|
|
|
|
| def _validate_all_keys_present():
|
| """Check 1: Every key in .env.example must exist in .env."""
|
| example_keys = _parse_example_keys()
|
| loaded = dotenv_values(ENV_FILE) if ENV_FILE.exists() else {}
|
| for key in example_keys:
|
| if key not in loaded and os.getenv(key) is None:
|
| print(f"MISSING ENV KEY: {key}")
|
| sys.exit(1)
|
|
|
|
|
| def _validate_schema_version():
|
| """Check 2: ENV_SCHEMA_VERSION must match expected."""
|
| version = _get("ENV_SCHEMA_VERSION", int, 0)
|
| if version != EXPECTED_SCHEMA_VERSION:
|
| print(
|
| f"ENV schema mismatch: expected v{EXPECTED_SCHEMA_VERSION}, got v{version}.\n"
|
| f"Run `python config.py migrate` to update your .env."
|
| )
|
| sys.exit(1)
|
|
|
|
|
| def _validate():
|
| """Run all validation checks."""
|
| _validate_all_keys_present()
|
| _validate_schema_version()
|
|
|
|
|
| risk = _get("RISK_PER_TRADE_PCT", float, 1.0)
|
| if risk > 3.0:
|
| raise ConfigError(
|
| f"RISK_PER_TRADE_PCT={risk} exceeds hard cap of 3.0. "
|
| "Refusing to start. Lower your risk or accept the cap."
|
| )
|
|
|
|
|
| tw = _get("TECHNICAL_WEIGHT", float, 0.65)
|
| sw = _get("SENTIMENT_WEIGHT", float, 0.35)
|
| if abs((tw + sw) - 1.0) > 0.001:
|
| raise ConfigError(
|
| f"TECHNICAL_WEIGHT ({tw}) + SENTIMENT_WEIGHT ({sw}) = {tw + sw}, "
|
| "must equal 1.0 (±0.001)."
|
| )
|
|
|
|
|
| hdd = _get("HISTORY_DAILY_DAYS", int, 500)
|
| if hdd < 250:
|
| logger.warning(
|
| "HISTORY_DAILY_DAYS overridden to 250 "
|
| "(EMA200 requires minimum 250 daily bars)."
|
| )
|
| os.environ["HISTORY_DAILY_DAYS"] = "250"
|
|
|
|
|
| allow_short = _get("ALLOW_SHORT", bool, False)
|
| acct_type = _get("ALPACA_ACCOUNT_TYPE", str, "margin")
|
| if allow_short and acct_type == "cash":
|
| raise ConfigError("Short selling requires a margin account.")
|
|
|
|
|
| mode = _get("TRADING_MODE", str, "paper")
|
| alloc = _get("ALLOCATED_CAPITAL", float, 0.0)
|
| dry = _get("DRY_RUN", bool, True)
|
| if mode == "live" and not dry and alloc <= 0:
|
| raise ConfigError(
|
| "ALLOCATED_CAPITAL must be > 0 for live trading. "
|
| "Set ALLOCATED_CAPITAL to your maximum bot allocation (e.g., 300). "
|
| "This prevents the bot from using your full account balance."
|
| )
|
|
|
|
|
| allow_overnight = _get("ALLOW_OVERNIGHT_POSITIONS", bool, False)
|
| if allow_overnight:
|
| logger.warning(
|
| "Overnight positions enabled. Using wider stops (2x daily ATR) "
|
| "to absorb gap risk. Max %d positions.",
|
| _get("MAX_OPEN_POSITIONS", int, 3),
|
| )
|
|
|
|
|
|
|
|
|
| try:
|
| _validate()
|
| except ConfigError as e:
|
| print(f"CONFIG ERROR: {e}")
|
| sys.exit(1)
|
|
|
|
|
|
|
| ALPACA_API_KEY: str = _get("ALPACA_API_KEY", str, "")
|
| ALPACA_SECRET_KEY: str = _get("ALPACA_SECRET_KEY", str, "")
|
| ALPACA_BASE_URL: str = _get("ALPACA_BASE_URL", str, "https://paper-api.alpaca.markets")
|
| ALPACA_DATA_URL: str = _get("ALPACA_DATA_URL", str, "https://data.alpaca.markets")
|
| ALPACA_ACCOUNT_TYPE: str = _get("ALPACA_ACCOUNT_TYPE", str, "margin")
|
|
|
| NEWS_API_KEY: str = _get("NEWS_API_KEY", str, "")
|
| GNEWS_API_KEY: str = _get("GNEWS_API_KEY", str, "")
|
|
|
| TELEGRAM_BOT_TOKEN: str = _get("TELEGRAM_BOT_TOKEN", str, "")
|
| TELEGRAM_CHAT_ID: str = _get("TELEGRAM_CHAT_ID", str, "")
|
|
|
| TRADING_MODE: str = _get("TRADING_MODE", str, "paper")
|
| DRY_RUN: bool = _get("DRY_RUN", bool, True)
|
| TARGET_DAILY_PROFIT_USD: float = _get("TARGET_DAILY_PROFIT_USD", float, 20.0)
|
| MAX_DAILY_LOSS_USD: float = _get("MAX_DAILY_LOSS_USD", float, 40.0)
|
| MAX_LOSS_PER_POSITION_USD: float = _get("MAX_LOSS_PER_POSITION_USD", float, 2.0)
|
| RISK_PER_TRADE_PCT: float = _get("RISK_PER_TRADE_PCT", float, 1.0)
|
| MAX_OPEN_POSITIONS: int = _get("MAX_OPEN_POSITIONS", int, 3)
|
| MAX_DAILY_TRADES: int = _get("MAX_DAILY_TRADES", int, 10)
|
| ALLOW_SHORT: bool = _get("ALLOW_SHORT", bool, False)
|
| ALLOW_CRYPTO: bool = _get("ALLOW_CRYPTO", bool, False)
|
|
|
|
|
| _UNIVERSE_A_RAW: str = _get("UNIVERSE_A", str, "")
|
| _UNIVERSE_B_RAW: str = _get("UNIVERSE_B", str, "")
|
| _UNIVERSE_RAW: str = _get("UNIVERSE", str, "")
|
|
|
| if _UNIVERSE_A_RAW and _UNIVERSE_B_RAW:
|
| UNIVERSE_A: list[str] = [s.strip() for s in _UNIVERSE_A_RAW.split(",") if s.strip()]
|
| UNIVERSE_B: list[str] = [s.strip() for s in _UNIVERSE_B_RAW.split(",") if s.strip()]
|
|
|
| from datetime import datetime, timezone
|
| _today_doy = datetime.now(timezone.utc).timetuple().tm_yday
|
| UNIVERSE: list[str] = UNIVERSE_A if (_today_doy % 2 == 1) else UNIVERSE_B
|
| UNIVERSE_LABEL: str = "A" if (_today_doy % 2 == 1) else "B"
|
| logger.info("Universe rotation: Day %d → Universe %s (%d tickers)", _today_doy, UNIVERSE_LABEL, len(UNIVERSE))
|
| else:
|
| UNIVERSE_A = []
|
| UNIVERSE_B = []
|
| UNIVERSE: list[str] = [s.strip() for s in _UNIVERSE_RAW.split(",") if s.strip()] if _UNIVERSE_RAW and _UNIVERSE_RAW != "auto" else ["AAPL", "MSFT", "NVDA", "SPY", "QQQ", "TSLA"]
|
| UNIVERSE_LABEL: str = "single"
|
|
|
| GPU_AVAILABLE: bool = _get("GPU_AVAILABLE", bool, True)
|
|
|
| MAX_SPREAD_PCT: float = _get("MAX_SPREAD_PCT", float, 0.50)
|
| LIMIT_ORDER_TIMEOUT_SEC: int = _get("LIMIT_ORDER_TIMEOUT_SEC", int, 60)
|
| ALLOW_MARKET_ORDERS: bool = _get("ALLOW_MARKET_ORDERS", bool, False)
|
| AGGRESSIVE_ENTRY: bool = _get("AGGRESSIVE_ENTRY", bool, False)
|
| LOCK_PROFITS_AT_TARGET: bool = _get("LOCK_PROFITS_AT_TARGET", bool, True)
|
| ALLOW_OVERNIGHT_POSITIONS: bool = _get("ALLOW_OVERNIGHT_POSITIONS", bool, False)
|
| ALLOCATED_CAPITAL: float = _get("ALLOCATED_CAPITAL", float, 0.0)
|
| MIN_POSITION_NOTIONAL_PCT: float = _get("MIN_POSITION_NOTIONAL_PCT", float, 0.05)
|
| MAX_SECTOR_POSITIONS: int = _get("MAX_SECTOR_POSITIONS", int, 2)
|
| STRATEGY_MODE: str = _get("STRATEGY_MODE", str, "auto")
|
|
|
| TECHNICAL_WEIGHT: float = _get("TECHNICAL_WEIGHT", float, 0.65)
|
| SENTIMENT_WEIGHT: float = _get("SENTIMENT_WEIGHT", float, 0.35)
|
| SIGNAL_BUY_THRESHOLD: float = _get("SIGNAL_BUY_THRESHOLD", float, 0.55)
|
| SIGNAL_SELL_THRESHOLD: float = _get("SIGNAL_SELL_THRESHOLD", float, -0.55)
|
| ATR_PERCENTILE_WINDOW_DAYS: int = _get("ATR_PERCENTILE_WINDOW_DAYS", int, 30)
|
|
|
| ACCOUNT_BALANCE_ABOVE_25K: bool = _get("ACCOUNT_BALANCE_ABOVE_25K", bool, True)
|
| PDT_MAX_DAY_TRADES: int = _get("PDT_MAX_DAY_TRADES", int, 3)
|
|
|
| HISTORY_DAILY_DAYS: int = max(_get("HISTORY_DAILY_DAYS", int, 500), 250)
|
| HISTORY_HOURLY_DAYS: int = _get("HISTORY_HOURLY_DAYS", int, 180)
|
| HISTORY_5MIN_DAYS: int = _get("HISTORY_5MIN_DAYS", int, 30)
|
| DB_BACKEND: str = _get("DB_BACKEND", str, "sqlite")
|
|
|
|
|
| TRADING_START_ET: str = _get("TRADING_START_ET", str, "09:45")
|
| TRADING_STOP_ET: str = _get("TRADING_STOP_ET", str, "14:00")
|
| FORCE_CLOSE_ET: str = _get("FORCE_CLOSE_ET", str, "14:00")
|
| SYSTEM_START_MINUTES_BEFORE_OPEN: int = _get("SYSTEM_START_MINUTES_BEFORE_OPEN", int, 10)
|
|
|
| def _parse_et_time(hhmm: str) -> tuple[int, int]:
|
| """Parse 'HH:MM' string to (hour, minute)."""
|
| parts = hhmm.strip().split(":")
|
| return int(parts[0]), int(parts[1])
|
|
|
| TRADING_START_HOUR, TRADING_START_MIN = _parse_et_time(TRADING_START_ET)
|
| TRADING_STOP_HOUR, TRADING_STOP_MIN = _parse_et_time(TRADING_STOP_ET)
|
| FORCE_CLOSE_HOUR, FORCE_CLOSE_MIN = _parse_et_time(FORCE_CLOSE_ET)
|
|
|
|
|
|
|
|
|
| def migrate():
|
| """Add any missing keys from .env.example to .env without deleting existing ones."""
|
| example_keys = _parse_example_keys()
|
| loaded = dotenv_values(ENV_FILE) if ENV_FILE.exists() else {}
|
| example_vals = dotenv_values(ENV_EXAMPLE) if ENV_EXAMPLE.exists() else {}
|
| added = []
|
| for key in sorted(example_keys):
|
| if key not in loaded:
|
| default = example_vals.get(key, "")
|
| set_key(str(ENV_FILE), key, default)
|
| added.append(key)
|
|
|
| set_key(str(ENV_FILE), "ENV_SCHEMA_VERSION", str(EXPECTED_SCHEMA_VERSION))
|
| if added:
|
| print(f"Added {len(added)} missing keys: {', '.join(added)}")
|
| else:
|
| print("No missing keys. .env is up to date.")
|
| print(f"ENV_SCHEMA_VERSION set to {EXPECTED_SCHEMA_VERSION}.")
|
|
|
|
|
| if __name__ == "__main__":
|
| if len(sys.argv) > 1 and sys.argv[1] == "migrate":
|
| migrate()
|
| else:
|
| print("Config loaded successfully. All validations passed.")
|
| print(f" TRADING_MODE={TRADING_MODE}")
|
| print(f" DRY_RUN={DRY_RUN}")
|
| print(f" UNIVERSE={UNIVERSE}")
|
| print(f" GPU_AVAILABLE={GPU_AVAILABLE}")
|
| print(f" DB_BACKEND={DB_BACKEND}")
|
| print(f" ACCOUNT_BALANCE_ABOVE_25K={ACCOUNT_BALANCE_ABOVE_25K}")
|
|
|
