from pathlib import Path from pydantic_settings import BaseSettings, SettingsConfigDict from app.bars.models import Timeframe TIMEFRAME_WEIGHT = { Timeframe.M1: 1, Timeframe.M2: 1, Timeframe.M5: 1, Timeframe.M15: 2, Timeframe.M30: 3, Timeframe.H1: 4, Timeframe.D1: 16, } MA_WEIGHT_FACTOR = 0.75 class Settings(BaseSettings): model_config = SettingsConfigDict(env_file=".env", extra="ignore") live_source: str = "yahoo" seed_source: str = "yahoo" yahoo_symbol: str = "ES=F" yahoo_poll_seconds: float = 20 seed_1h_range: str = "730d" seed_1m_range: str = "8d" timeframes: str = "1m,5m,15m,30m,1h,1d" base_timeframes: str = "1m,30m,1d" max_bars_per_tf: int = 5000 ma_sets__1d: str = "sma10,sma20,sma50,sma100,sma200" daily_anchor_et: str = "18:00" manual_lines_path: Path = Path("./data/manual_lines.json") # Schwab. Empty until the app's keys are issued; nothing reads them while # live_source is yahoo. The token lives under data/ so it lands on the # Coolify persistent volume — a rebuild would otherwise log you out, and # re-authenticating is an interactive browser flow. schwab_api_key: str = "" schwab_app_secret: str = "" schwab_callback_url: str = "https://chart.amow.com/api/qt" schwab_token_path: Path = Path("./data/.schwab_token.json") schwab_symbol: str = "/ES" confluence_min_score: float = 28 # Four hours, chosen from the sweep in scripts/calibrate_alerts.py. Suppression is # per price zone, so an unrelated zone still alerts immediately; this only # governs how often the *same* area repeats itself. alert_cooldown_seconds: int = 14400 ntfy_topic: str = "" ntfy_server: str = "https://ntfy.sh" chart_auth_token: str = "" replay_file: Path | None = None @property def live_symbol(self) -> str: """What the live source calls the instrument. Yahoo wants ES=F, Schwab wants /ES. Seeding always uses the Yahoo symbol, because Yahoo is always the source of history. """ return self.schwab_symbol if self.live_source == "schwab" else self.yahoo_symbol @property def enabled_timeframes(self) -> list[Timeframe]: return [Timeframe(value.strip()) for value in self.timeframes.split(",") if value.strip()] @property def ma_sets(self) -> dict[Timeframe, list[tuple[str, int]]]: configured = {Timeframe.D1: self.ma_sets__1d} result: dict[Timeframe, list[tuple[str, int]]] = {} for tf, value in configured.items(): definitions = [] for item in filter(None, (part.strip().lower() for part in value.split(","))): kind = "sma" if item.startswith("sma") else "ema" if item.startswith("ema") else "" if not kind or not item[len(kind) :].isdigit(): raise ValueError(f"Invalid MA definition: {item}") definitions.append((kind, int(item[len(kind) :]))) result[tf] = definitions return result