from __future__ import annotations from typing import Any from web_core.constants import INTERVAL_OPTIONS, PERIOD_OPTIONS from web_core.market.presets import MARKET_PRESET_OPTIONS from web_core.time_display import DEFAULT_DISPLAY_TIMEZONE, normalize_display_timezone def _clamp_int(value: Any, fallback: int, minimum: int, maximum: int) -> int: try: parsed = int(value) except (TypeError, ValueError): return fallback return min(maximum, max(minimum, parsed)) def _clamp_float(value: Any, fallback: float, minimum: float, maximum: float) -> float: try: parsed = float(value) except (TypeError, ValueError): return fallback return min(maximum, max(minimum, parsed)) def _to_bool(value: Any, fallback: bool) -> bool: if isinstance(value, bool): return value if isinstance(value, (int, float)): return value != 0 if value is None: return fallback normalized = str(value).strip().lower() if normalized in {"1", "true", "yes", "y", "on"}: return True if normalized in {"0", "false", "no", "n", "off"}: return False return fallback def _clamp_max_bars(value: Any, fallback: int = 500) -> int: return _clamp_int(value=value, fallback=fallback, minimum=20, maximum=5000) def _normalize_watchlist(raw: Any) -> list[str]: tokens: list[str] = [] if isinstance(raw, list): tokens = [str(item) for item in raw] elif isinstance(raw, str): tokens = [raw] else: tokens = [str(raw or "")] seen: set[str] = set() symbols: list[str] = [] for token in tokens: normalized_token = token.replace("\\n", "\n").replace("\\N", "\n").replace(",", "\n") for split_token in normalized_token.splitlines(): symbol = split_token.strip().upper() if not symbol or symbol in seen: continue seen.add(symbol) symbols.append(symbol) return symbols[:40] def _normalize_symbol_list(raw: Any, limit: int = 20) -> list[str]: tokens: list[str] = [] if isinstance(raw, list): tokens = [str(item) for item in raw] else: tokens = [str(raw or "")] seen: set[str] = set() out: list[str] = [] for token in tokens: normalized_token = token.replace("\\n", "\n").replace("\\N", "\n").replace(",", "\n") for split_token in normalized_token.splitlines(): symbol = split_token.strip().upper() if not symbol or symbol in seen: continue seen.add(symbol) out.append(symbol) return out[:limit] def normalize_web_settings(raw: dict[str, Any] | None) -> dict[str, Any]: raw = raw or {} defaults: dict[str, Any] = { "symbol": "AAPL", "interval": "1d", "period": "6mo", "max_bars": 500, "drop_live": True, "use_body_range": False, "volume_filter_enabled": False, "volume_sma_window": 20, "volume_multiplier": 1.0, "gray_fake": True, "hide_market_closed_gaps": True, "show_live_guide": False, "show_past_behavior": False, "show_trade_markers": False, "focus_chart_on_selected_example": False, "max_training_examples": 20, "enable_auto_refresh": False, "refresh_sec": 60, "watchlist": [], "market_preset": "Custom", "compare_symbols": [], "enable_compare_symbols": False, "enable_multi_tf_confirmation": False, "enable_regime_filter": False, "advanced_auto_run": False, "replay_mode_enabled": False, "replay_bars": 120, "enable_alert_rules": False, "alert_on_bull": True, "alert_on_bear": True, "alert_webhook_url": "", "backtest_slippage_bps": 0.0, "backtest_fee_bps": 0.0, "backtest_stop_loss_pct": 0.0, "backtest_take_profit_pct": 0.0, "backtest_min_hold_bars": 1, "backtest_max_hold_bars": 1, "display_timezone": DEFAULT_DISPLAY_TIMEZONE, "use_24h_time": False, } symbol = str(raw.get("symbol", defaults["symbol"])).strip().upper() if not symbol: symbol = str(defaults["symbol"]) interval = str(raw.get("interval", defaults["interval"])) if interval not in INTERVAL_OPTIONS: interval = str(defaults["interval"]) period = str(raw.get("period", defaults["period"])) if period not in PERIOD_OPTIONS: period = str(defaults["period"]) watchlist = _normalize_watchlist(raw.get("watchlist", defaults["watchlist"])) market_preset = str(raw.get("market_preset", defaults["market_preset"])) if market_preset not in MARKET_PRESET_OPTIONS: market_preset = str(defaults["market_preset"]) compare_symbols = _normalize_symbol_list(raw.get("compare_symbols", defaults["compare_symbols"]), limit=12) replay_bars = _clamp_int(raw.get("replay_bars"), fallback=int(defaults["replay_bars"]), minimum=20, maximum=1000) backtest_min_hold_bars = _clamp_int( raw.get("backtest_min_hold_bars"), fallback=int(defaults["backtest_min_hold_bars"]), minimum=1, maximum=20, ) backtest_max_hold_bars = _clamp_int( raw.get("backtest_max_hold_bars"), fallback=int(defaults["backtest_max_hold_bars"]), minimum=1, maximum=40, ) backtest_max_hold_bars = max(backtest_max_hold_bars, backtest_min_hold_bars) return { "symbol": symbol, "interval": interval, "period": period, "max_bars": _clamp_max_bars(raw.get("max_bars"), fallback=int(defaults["max_bars"])), "drop_live": _to_bool(raw.get("drop_live"), fallback=bool(defaults["drop_live"])), "use_body_range": _to_bool(raw.get("use_body_range"), fallback=bool(defaults["use_body_range"])), "volume_filter_enabled": _to_bool( raw.get("volume_filter_enabled"), fallback=bool(defaults["volume_filter_enabled"]) ), "volume_sma_window": _clamp_int( raw.get("volume_sma_window"), fallback=int(defaults["volume_sma_window"]), minimum=2, maximum=100, ), "volume_multiplier": round( _clamp_float( raw.get("volume_multiplier"), fallback=float(defaults["volume_multiplier"]), minimum=0.1, maximum=3.0, ), 1, ), "gray_fake": _to_bool(raw.get("gray_fake"), fallback=bool(defaults["gray_fake"])), "hide_market_closed_gaps": _to_bool( raw.get("hide_market_closed_gaps"), fallback=bool(defaults["hide_market_closed_gaps"]), ), "show_live_guide": _to_bool(raw.get("show_live_guide"), fallback=bool(defaults["show_live_guide"])), "show_past_behavior": _to_bool(raw.get("show_past_behavior"), fallback=bool(defaults["show_past_behavior"])), "show_trade_markers": _to_bool(raw.get("show_trade_markers"), fallback=bool(defaults["show_trade_markers"])), "focus_chart_on_selected_example": _to_bool( raw.get("focus_chart_on_selected_example"), fallback=bool(defaults["focus_chart_on_selected_example"]), ), "max_training_examples": _clamp_int( raw.get("max_training_examples"), fallback=int(defaults["max_training_examples"]), minimum=5, maximum=100, ), "enable_auto_refresh": _to_bool( raw.get("enable_auto_refresh"), fallback=bool(defaults["enable_auto_refresh"]) ), "refresh_sec": _clamp_int( raw.get("refresh_sec"), fallback=int(defaults["refresh_sec"]), minimum=10, maximum=600, ), "watchlist": watchlist, "market_preset": market_preset, "compare_symbols": compare_symbols, "enable_compare_symbols": _to_bool( raw.get("enable_compare_symbols"), fallback=bool(defaults["enable_compare_symbols"]) ), "enable_multi_tf_confirmation": _to_bool( raw.get("enable_multi_tf_confirmation"), fallback=bool(defaults["enable_multi_tf_confirmation"]), ), "advanced_auto_run": _to_bool(raw.get("advanced_auto_run"), fallback=bool(defaults["advanced_auto_run"])), "enable_regime_filter": _to_bool( raw.get("enable_regime_filter"), fallback=bool(defaults["enable_regime_filter"]) ), "replay_mode_enabled": _to_bool( raw.get("replay_mode_enabled"), fallback=bool(defaults["replay_mode_enabled"]) ), "replay_bars": replay_bars, "enable_alert_rules": _to_bool(raw.get("enable_alert_rules"), fallback=bool(defaults["enable_alert_rules"])), "alert_on_bull": _to_bool(raw.get("alert_on_bull"), fallback=bool(defaults["alert_on_bull"])), "alert_on_bear": _to_bool(raw.get("alert_on_bear"), fallback=bool(defaults["alert_on_bear"])), "alert_webhook_url": str(raw.get("alert_webhook_url", defaults["alert_webhook_url"])).strip(), "backtest_slippage_bps": round( _clamp_float(raw.get("backtest_slippage_bps"), fallback=0.0, minimum=0.0, maximum=100.0), 1, ), "backtest_fee_bps": round( _clamp_float(raw.get("backtest_fee_bps"), fallback=0.0, minimum=0.0, maximum=100.0), 1, ), "backtest_stop_loss_pct": round( _clamp_float(raw.get("backtest_stop_loss_pct"), fallback=0.0, minimum=0.0, maximum=25.0), 2, ), "backtest_take_profit_pct": round( _clamp_float(raw.get("backtest_take_profit_pct"), fallback=0.0, minimum=0.0, maximum=25.0), 2, ), "backtest_min_hold_bars": backtest_min_hold_bars, "backtest_max_hold_bars": backtest_max_hold_bars, "display_timezone": normalize_display_timezone( raw.get("display_timezone"), fallback=str(defaults["display_timezone"]), ), "use_24h_time": _to_bool(raw.get("use_24h_time"), fallback=bool(defaults["use_24h_time"])), }