""" webapp/server.py FastAPI backend for the dashboard — reachable two ways: 1. As a Telegram Mini App, opened via the /dashboard button (auth via Telegram's signed initData). 2. Directly in a plain browser — e.g. a Hugging Face Space's landing page after container startup, where there's no Telegram context at all (auth via a plain access token — see webapp/auth.py). This is intentionally created as a factory (create_app) that takes the *already-running* connector/executor/telegram objects from main.py and runs inside the SAME process (via a background thread started in main.py). That matters: it means the dashboard reads and mutates the exact same in-memory state as the bot's trading loop and command bot — no second MT5 connection, no state drifting out of sync, no IPC needed. Every route (except the static frontend and the health check) requires valid credentials via one of the two methods above — see webapp.auth.authorize(). """ import logging import os from typing import Optional from fastapi import FastAPI, HTTPException, Request from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel import performance_tracker as perf import goal_tracker import position_manager import account_manager import auto_tuner import strategies import market_sessions from config import Config, TUNABLE_REGISTRY, get_setting_default import runtime_config import presets from webapp.auth import authorize log = logging.getLogger("wickbot.webapp") STATIC_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "static") def _perf_by_key(key_column: str) -> dict: """Builds per-key performance stats (win rate, total trades, profit factor, avg R, net R) from the trade log, keyed by `key_column` ('strategy' or 'symbol'). Reads only closed demo trades so the numbers match the graduation gate in performance_tracker.py.""" conn = perf._connect() try: rows = conn.execute( f"SELECT {key_column}, result_r, status FROM trades " "WHERE account_type='demo' AND status LIKE 'closed_%'" ).fetchall() finally: conn.close() buckets = {} for key, result_r, status in rows: buckets.setdefault(key, []).append((result_r or 0.0, status)) out = {} for key, vals in buckets.items(): total = len(vals) wins = [v for v, s in vals if v > 0] losses = [v for v, s in vals if v <= 0] gross_profit = sum(wins) gross_loss = abs(sum(losses)) if gross_loss > 0: profit_factor = round(gross_profit / gross_loss, 2) elif gross_profit == 0: profit_factor = None else: profit_factor = 999.0 out[key] = { "total_trades": total, "wins": len(wins), "losses": len(losses), "win_rate": round(len(wins) / total, 2), "avg_r": round(sum(v for v, _ in vals) / total, 2), "net_r": round(sum(v for v, _ in vals), 2), "profit_factor": profit_factor, } return out def _creds_from_request(request: Request) -> tuple: """Pulls the two supported credential forms out of a GET request: Telegram `init_data` (query string) or a browser `access_token` (query string or Authorization: Bearer header). Returns (init_data, access_token) for passing to authorize().""" init_data = request.query_params.get("init_data") access_token = request.query_params.get("access_token") if not access_token: auth_header = request.headers.get("Authorization", "") if auth_header.lower().startswith("bearer "): access_token = auth_header[7:].strip() return init_data, access_token class AuthPayload(BaseModel): init_data: Optional[str] = None access_token: Optional[str] = None class ArmRealPayload(AuthPayload): confirm_phrase: str class GoalSetPayload(AuthPayload): text: str class PositionModifyPayload(AuthPayload): ticket: int new_sl: Optional[float] = None new_tp: Optional[float] = None class PositionClosePayload(AuthPayload): ticket: int class PendingCancelPayload(AuthPayload): ticket: int class SettingUpdatePayload(AuthPayload): key: str value: str class SettingResetPayload(AuthPayload): key: str class AccountAddPayload(AuthPayload): label: str login: int password: str server: str account_type: str = "demo" broker: str = "" mt5_path: Optional[str] = None notes: str = "" class AccountSwitchPayload(AuthPayload): account_id: int class AccountRemovePayload(AuthPayload): account_id: int class TuningActionPayload(AuthPayload): suggestion_id: int def create_app(connector, executor, tg) -> FastAPI: """ connector: MT5Connector instance (already connected) executor: TradeExecutor instance tg: WickBotTelegram instance (provides .paused and notify()) """ # Imported here (not at module top) to avoid a circular import, since # telegram_bot.py is what constructs `tg` in the first place. from telegram_bot import CONFIRM_PHRASE app = FastAPI(title="WickBot Dashboard") @app.get("/") def index(): # Serves the same dashboard whether opened via Telegram or a # plain browser — index.html itself detects which context it's # in and asks for a token only when there's no Telegram initData. return FileResponse(f"{STATIC_DIR}/index.html") @app.get("/healthz") def healthz(): # Unauthenticated on purpose — this is just a container/platform # liveness check (e.g. for Hugging Face Spaces), not a data route. return {"status": "ok"} app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") @app.post("/api/whoami") def api_whoami(payload: AuthPayload): """Lets the frontend validate a freshly-entered browser token (or Telegram initData) before rendering the rest of the dashboard, without needing a dedicated 'login' endpoint.""" identity = authorize(payload.init_data, payload.access_token) return {"ok": True, "auth_via": identity.get("auth_via")} @app.post("/api/status") def api_status(payload: AuthPayload): authorize(payload.init_data, payload.access_token) status = connector.get_account_status() return { "login": status.login, "server": status.server, "is_demo": status.is_demo, "is_real": status.is_real, "balance": status.balance, "equity": status.equity, "currency": status.currency, "open_positions": connector.positions_count(), "paused": tg.paused, "real_trading_armed": executor.real_trading_armed, "allow_real_trading_config": Config.ALLOW_REAL_TRADING, "symbols": Config.SYMBOLS, "timeframe": Config.TIMEFRAME, } @app.post("/api/stats") def api_stats(payload: AuthPayload): authorize(payload.init_data, payload.access_token) grad = perf.graduation_status() s = grad["stats"] return { "total_closed": s.total_closed, "win_rate": s.win_rate, "expectancy_r": s.expectancy_r, "by_strategy": s.by_strategy, "by_symbol": s.by_symbol, "eligible": grad["eligible"], "requirements": grad["requirements"], } # ------------------------------------------------------------------ # Read-only GET endpoints for the redesigned dashboard frontend. # These still require the same credentials as every other route via # authorize(), but use GET so the frontend can poll them on a timer # with a plain fetch() (credentials passed as query params / Bearer # header) instead of building a JSON body each time. # ------------------------------------------------------------------ # Module-level cache for /api/symbols so we don't call symbols_get() # on every poll (the frontend polls this on a timer). 30s TTL. _symbols_cache = {"ts": 0.0, "connected": False, "symbols": []} _SYMBOLS_CACHE_TTL = 30.0 _SYMBOLS_MAX = 200 def _classify_symbol(name: str) -> str: """Classify a symbol into a group: forex, crypto, metal, commodity, index, or stock.""" upper = name.upper() # Remove common suffixes like m, .a, .n etc. for pattern matching import re base = re.sub(r'[\._][a-zA-Z0-9]+$', '', upper) # Indices if any(base.startswith(p) for p in ("US30", "US100", "US500", "UK100", "JP225", "GER40", "FRA40", "AUS200", "SPX", "NAS", "DJI", "IN50", "NIFTY", "HSI", "CHINA50", "VIX", "EAm", "AMCm")): return "index" # Metals if base.startswith(("XAU", "XAG", "XPD", "XPT", "GOLD", "SILVER", "PALLADIUM", "PLATINUM")): return "metal" # Commodities if any(base.startswith(p) for p in ("XTI", "XBR", "XNG", "UKOIL", "USOIL", "NATGAS", "COPPER", "WHEAT", "CORN", "SOYBEAN", "SUGAR", "COFFEE", "COCOA", "COTTON", "BRENT", "WTI", "NG", "CL", "HO", "RB")): return "commodity" # Crypto crypto_pairs = ("BTC", "ETH", "XRP", "LTC", "BCH", "ADA", "DOT", "SOL", "DOGE", "LINK", "UNI", "MATIC", "AAVE", "BNB", "XLM", "TRX", "EOS", "FIL", "ATOM", "ALGO", "AVAX", "NEAR", "SAND", "MANA", "AXS", "CAKE", "COMP", "MKR", "YFI", "SNX", "SUSHI", "CRV", "1INCH", "ENJ", "XTZ", "IOST", "THETA", "HBAR", "HT", "BAT", "MBT", "TET", "IQ", "RLX") if any(base.startswith(p) for p in crypto_pairs): return "crypto" # Also check for common crypto suffixes like USDm after a crypto name if base.endswith("USDm") and len(base) > 5: stem = base[:-4] if any(c.isdigit() for c in stem) or len(stem) >= 3: return "crypto" # Stocks stock_keywords = ("BRQS", "BYND", "CAN", "BBB", "BB", "TIGR") if any(name.upper().startswith(p) for p in stock_keywords): return "stock" # Forex — 6-char currency pairs, optionally with a suffix if len(base) >= 6: currencies = {"EUR","GBP","USD","JPY","CHF","AUD","NZD","CAD","SGD","HKD", "SEK","NOK","DKK","ZAR","MXN","TRY","PLN","HUF","CZK","RON", "BGN","CNY","INR","KRW","TWD","MYR","THB","IDR","PHP","VND", "BRL","CLP","COP","EGP","ILS","KWD","SAR","QAR","AED","JOD", "OMR","BHD","KZT","UAH","NGN","KES","GHS","TZS","UGX","MAD", "XOF","LKR","PKR","BDT","NPR","ISK","GEL","AMD","AZN","BYN", "MNT","MUR","MVR","SCR","LBP","IQD","YER","SYP","TND","DZD", "ARS","BOB","PYG","UYU","VES","CRC","DOP","GTQ","HNL","NIO", "PAB","PEN","TTD"} # Check if first 3 chars and last 3 chars (or 3 chars after suffix strip) are known currencies if len(base) >= 6: first3 = base[:3] # Try to find a second currency after the first one for i in range(3, min(7, len(base))): if base[3:i] in currencies and (base[i:] in currencies or base[i:] == ""): break if first3 in currencies and base[3:6] in currencies: return "forex" # Also check with exotic patterns: USDCZKm -> USDCZK if len(base) >= 6 and base[:3] in currencies and base[3:6] in currencies: return "forex" # If it has a currency-like pattern with a suffix (e.g. EURUSDm) if len(upper) >= 6: curr_prefixes = ("EUR","GBP","USD","JPY","CHF","AUD","NZD","CAD") for c in curr_prefixes: if upper.startswith(c): rest = upper[len(c):] if any(rest.startswith(cc) for cc in curr_prefixes): return "forex" return "other" @app.get("/api/symbols") def api_symbols(request: Request): authorize(*_creds_from_request(request)) mt5 = connector.mt5 connected = getattr(connector, "_connected", False) # Serve from cache if still fresh. import time as _time if _time.time() - _symbols_cache["ts"] < _SYMBOLS_CACHE_TTL and _symbols_cache["symbols"]: return {"connected": _symbols_cache["connected"], "symbols": _symbols_cache["symbols"]} symbols = [] if connected and mt5 is not None: try: # The standard MetaTrader5 package has no "market watch only" # function — symbols_get() returns ALL available symbols. We # approximate Market Watch by preferring those flagged visible # (sym.visible), and fall back to returning all (capped) when # visibility isn't reliable. all_syms = mt5.symbols_get() if all_syms: visible = [s for s in all_syms if getattr(s, "visible", False)] chosen = visible if visible else all_syms # Cap the payload to avoid huge responses. chosen = chosen[:_SYMBOLS_MAX] for info in chosen: if info is None: continue spread = None if info.bid and info.ask and info.point: spread = round((info.ask - info.bid) / info.point, 2) symbols.append({ "symbol": info.name, "description": getattr(info, "description", ""), "bid": info.bid, "ask": info.ask, "spread": spread, "digits": info.digits, "visible": bool(getattr(info, "visible", False)), "trade_mode": info.trade_mode, "last": getattr(info, "last", None), "volume_min": getattr(info, "volume_min", None), "volume_max": getattr(info, "volume_max", None), "volume_step": getattr(info, "volume_step", None), "point": info.point, "trade_stops_level": getattr(info, "trade_stops_level", None), }) except Exception as e: log.warning("Failed to enumerate Market Watch symbols: %s", e) connected = False # Add category to each symbol for sym in symbols: sym["category"] = _classify_symbol(sym["symbol"]) _symbols_cache["ts"] = _time.time() _symbols_cache["connected"] = connected _symbols_cache["symbols"] = symbols return {"connected": connected, "symbols": symbols} @app.get("/api/strategies") def api_strategies(request: Request): authorize(*_creds_from_request(request)) names = [s.name for s in strategies.ALL_STRATEGIES] return {"strategies": names} @app.get("/api/performance/strategy") def api_perf_strategy(request: Request): authorize(*_creds_from_request(request)) return {"performance": _perf_by_key("strategy")} @app.get("/api/performance/symbol") def api_perf_symbol(request: Request): authorize(*_creds_from_request(request)) return {"performance": _perf_by_key("symbol")} @app.get("/api/market-sessions") def api_market_sessions(request: Request): authorize(*_creds_from_request(request)) return market_sessions.get_session_status() @app.post("/api/pause") def api_pause(payload: AuthPayload): authorize(payload.init_data, payload.access_token) tg.paused = True log.info("Trading paused via dashboard") return {"paused": True} @app.post("/api/resume") def api_resume(payload: AuthPayload): authorize(payload.init_data, payload.access_token) tg.paused = False log.info("Trading resumed via dashboard") return {"paused": False} @app.post("/api/arm_real") def api_arm_real(payload: ArmRealPayload): authorize(payload.init_data, payload.access_token) if payload.confirm_phrase.strip() != CONFIRM_PHRASE: raise HTTPException(status_code=400, detail="Confirmation phrase did not match") status = connector.get_account_status() if not status.is_real: raise HTTPException(status_code=400, detail="Current MT5 account is not REAL — nothing to arm") if not Config.ALLOW_REAL_TRADING: raise HTTPException(status_code=400, detail="ALLOW_REAL_TRADING is false in .env") grad = perf.graduation_status() if not grad["eligible"]: raise HTTPException(status_code=400, detail="Demo graduation criteria not yet met") executor.arm_real_trading(True) log.warning("Real trading ARMED via dashboard by verified owner") tg.notify("🔴 Real trading was just ARMED via the dashboard.") return {"armed": True} @app.post("/api/disarm_real") def api_disarm_real(payload: AuthPayload): authorize(payload.init_data, payload.access_token) executor.real_trading_armed = False log.info("Real trading disarmed via dashboard") tg.notify("🔒 Real trading was disarmed via the dashboard.") return {"armed": False} @app.post("/api/goal") def api_goal(payload: AuthPayload): authorize(payload.init_data, payload.access_token) status = connector.get_account_status() prog = goal_tracker.progress(status.equity) return {"active": prog is not None, "progress": prog} @app.post("/api/goal/set") def api_goal_set(payload: GoalSetPayload): authorize(payload.init_data, payload.access_token) parsed = goal_tracker.parse_goal_text(payload.text) if "error" in parsed: raise HTTPException(status_code=400, detail=parsed["error"]) status = connector.get_account_status() goal_tracker.set_goal( target_usd=parsed["target_usd"], duration_hours=parsed["duration_hours"], starting_balance=status.equity, ) tg.notify(f"🎯 Goal set via dashboard: ${parsed['target_usd']:.2f} in {parsed['duration_hours']:.1f}h.") return {"ok": True, "parsed": parsed} @app.post("/api/goal/cancel") def api_goal_cancel(payload: AuthPayload): authorize(payload.init_data, payload.access_token) goal_tracker.cancel_active_goal() return {"ok": True} @app.post("/api/positions") def api_positions(payload: AuthPayload): authorize(payload.init_data, payload.access_token) return {"positions": position_manager.get_status(connector)} @app.post("/api/positions/modify") def api_positions_modify(payload: PositionModifyPayload): authorize(payload.init_data, payload.access_token) if payload.new_sl is None and payload.new_tp is None: raise HTTPException(status_code=400, detail="Provide new_sl and/or new_tp") try: ok = position_manager.manual_modify(connector, payload.ticket, payload.new_sl, payload.new_tp) except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) if not ok: raise HTTPException(status_code=400, detail="Broker rejected the modification") tg.notify(f"✏️ Position {payload.ticket} manually modified via dashboard.") return {"ok": True} @app.post("/api/positions/close") def api_positions_close(payload: PositionClosePayload): authorize(payload.init_data, payload.access_token) try: ok = position_manager.manual_close(connector, payload.ticket) except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) if not ok: raise HTTPException(status_code=400, detail="Broker rejected the close") tg.notify(f"❌ Position {payload.ticket} manually closed via dashboard.") return {"ok": True} @app.post("/api/pending_orders") def api_pending_orders(payload: AuthPayload): authorize(payload.init_data, payload.access_token) return {"orders": position_manager.get_pending_status(connector)} @app.post("/api/pending_orders/cancel") def api_pending_cancel(payload: PendingCancelPayload): authorize(payload.init_data, payload.access_token) result = connector.cancel_pending_order(payload.ticket) ok = getattr(result, "retcode", None) == 10009 if not ok: raise HTTPException(status_code=400, detail="Broker rejected the cancellation") tg.notify(f"🗑️ Pending order {payload.ticket} cancelled via dashboard.") return {"ok": True} # ------------------------------------------------------------------ # Settings — dynamic/tunable config only. TUNABLE_REGISTRY is the # allowlist; a key not in it is structurally impossible to set here, # which is what keeps ALLOW_REAL_TRADING/MIN_DEMO_*/MAGIC_NUMBER # un-editable from this surface no matter what. # ------------------------------------------------------------------ @app.post("/api/settings") def api_settings(payload: AuthPayload): authorize(payload.init_data, payload.access_token) out = [] for spec in TUNABLE_REGISTRY: key = spec["key"] current = getattr(Config, key) default = get_setting_default(key) out.append({ "key": key, "type": spec["type"], "group": spec["group"], "label": spec["label"], "help": spec["help"], "value": current, "default": default, "is_overridden": runtime_config.get_raw(key) is not None, }) return {"settings": out} @app.post("/api/settings/update") def api_settings_update(payload: SettingUpdatePayload): authorize(payload.init_data, payload.access_token) valid_keys = {s["key"] for s in TUNABLE_REGISTRY} if payload.key not in valid_keys: raise HTTPException(status_code=400, detail=f"'{payload.key}' is not an editable setting") value = payload.value # SYMBOLS must be stored as a proper list, never a stringified # string (which would double-encode on the next read). Parse the # incoming value into a real list and store that. if payload.key == "SYMBOLS": from runtime_config import parse_symbols_list value = parse_symbols_list(payload.value) runtime_config.set_raw(payload.key, value) tg.notify(f"⚙️ Setting changed via dashboard: `{payload.key}` = `{payload.value}`") return {"ok": True, "key": payload.key, "value": getattr(Config, payload.key)} @app.post("/api/settings/reset") def api_settings_reset(payload: SettingResetPayload): authorize(payload.init_data, payload.access_token) valid_keys = {s["key"] for s in TUNABLE_REGISTRY} if payload.key not in valid_keys: raise HTTPException(status_code=400, detail=f"'{payload.key}' is not an editable setting") runtime_config.delete(payload.key) return {"ok": True, "key": payload.key, "value": getattr(Config, payload.key)} # ------------------------------------------------------------------ # Accounts — never returns a decrypted password over the API. # Switching only works within a broker/terminal the currently # running MT5 instance can actually log into — see account_manager.py # and docs/MULTI_ACCOUNT.md for the real constraints here. # ------------------------------------------------------------------ @app.post("/api/accounts") def api_accounts(payload: AuthPayload): authorize(payload.init_data, payload.access_token) return {"accounts": account_manager.list_accounts()} @app.post("/api/accounts/add") def api_accounts_add(payload: AccountAddPayload): authorize(payload.init_data, payload.access_token) account_id = account_manager.add_account( label=payload.label, login=payload.login, password=payload.password, server=payload.server, account_type=payload.account_type, broker=payload.broker, mt5_path=payload.mt5_path, notes=payload.notes, ) tg.notify(f"➕ Account added via dashboard: {payload.label} ({payload.server})") return {"ok": True, "account_id": account_id} @app.post("/api/accounts/switch") def api_accounts_switch(payload: AccountSwitchPayload): authorize(payload.init_data, payload.access_token) try: status = connector.switch_account(payload.account_id) except Exception as e: raise HTTPException(status_code=400, detail=str(e)) tg.notify( f"🔀 Switched active account via dashboard: {status.login} @ {status.server} " f"({'DEMO' if status.is_demo else 'REAL'})" ) return {"ok": True, "login": status.login, "server": status.server, "is_demo": status.is_demo} @app.post("/api/accounts/remove") def api_accounts_remove(payload: AccountRemovePayload): authorize(payload.init_data, payload.access_token) account_manager.delete_account(payload.account_id) return {"ok": True} # ------------------------------------------------------------------ # AI tuning suggestions — suggestion-only, never auto-applied. See # auto_tuner.py for why the actual proposed values always come from # an auditable statistical rule, never invented by the LLM. # ------------------------------------------------------------------ @app.post("/api/tuning/suggestions") def api_tuning_suggestions(payload: AuthPayload): authorize(payload.init_data, payload.access_token) return {"suggestions": auto_tuner.get_pending_suggestions()} @app.get("/api/tuning/status") def api_tuning_status(request: Request): authorize(*_creds_from_request(request)) return auto_tuner.get_status() @app.post("/api/tuning/generate") def api_tuning_generate(payload: AuthPayload): authorize(payload.init_data, payload.access_token) new_suggestions = auto_tuner.generate_suggestions() return {"generated": len(new_suggestions), "suggestions": new_suggestions} @app.post("/api/tuning/apply") def api_tuning_apply(payload: TuningActionPayload): authorize(payload.init_data, payload.access_token) # If a specific suggestion_id is supplied, apply just that one # (back-compat with the per-suggestion Apply button). Otherwise # apply every currently-pending suggestion at once. if payload.suggestion_id: try: result = auto_tuner.apply_suggestion(payload.suggestion_id) except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) tg.notify(f"🤖 Applied tuning suggestion via dashboard: `{result['key']}` -> `{result['value']}`") return {"ok": True, **result} applied = auto_tuner.apply_suggestions() if applied: summary = ", ".join(f"`{a['key']}`→`{a['value']}`" for a in applied) tg.notify(f"🤖 Applied {len(applied)} tuning suggestion(s) via dashboard: {summary}") return {"ok": True, "applied": applied} @app.post("/api/tuning/dismiss") def api_tuning_dismiss(payload: TuningActionPayload): authorize(payload.init_data, payload.access_token) auto_tuner.dismiss_suggestion(payload.suggestion_id) return {"ok": True} # ------------------------------------------------------------------ # Config presets — named bundles of tunable overrides. See presets.py. # ------------------------------------------------------------------ @app.get("/api/presets") def api_presets(request: Request, strategy: str = None, mode: str = None, timeframe: str = None): authorize(*_creds_from_request(request)) # 'all' or None act as wildcards in presets.list_presets. strat = strategy if (strategy and strategy != "all") else None md = mode if (mode and mode != "all") else None tf = timeframe if (timeframe and timeframe != "all") else None return {"presets": presets.list_presets(strat, md, tf)} @app.get("/api/presets/{preset_id}") def api_preset_detail(request: Request, preset_id: str): authorize(*_creds_from_request(request)) # Accept either an integer id or a name. try: ident = int(preset_id) except ValueError: ident = preset_id preset = presets.get_preset(ident) if preset is None: raise HTTPException(status_code=404, detail="preset not found") return preset class PresetApplyPayload(AuthPayload): preset: str @app.post("/api/presets/apply") def api_preset_apply(payload: PresetApplyPayload): authorize(payload.init_data, payload.access_token) if not payload.preset: raise HTTPException(status_code=400, detail="preset id or name is required") try: applied = presets.apply_preset(payload.preset) except Exception as e: raise HTTPException(status_code=400, detail=str(e)) return {"ok": True, "applied": applied} class PresetGeneratePayload(AuthPayload): text: str @app.post("/api/presets/generate") def api_preset_generate(payload: PresetGeneratePayload): authorize(payload.init_data, payload.access_token) if not payload.text or not payload.text.strip(): raise HTTPException(status_code=400, detail="text is required") try: generated = presets.generate_preset_from_text(payload.text) pid = presets.save_preset( name=generated["name"], config_dict=generated["config"], description=generated.get("description", ""), strategy=generated.get("strategy", "all"), mode=generated.get("mode", "all"), timeframe=generated.get("timeframe", "all"), created_by="ai", ) preset = presets.get_preset(pid) except Exception as e: raise HTTPException(status_code=400, detail=str(e)) return {"ok": True, "preset": preset} @app.delete("/api/presets/{preset_id}") def api_preset_delete(request: Request, preset_id: str): authorize(*_creds_from_request(request)) try: ident = int(preset_id) except ValueError: ident = preset_id preset = presets.get_preset(ident) if preset is None: raise HTTPException(status_code=404, detail="preset not found") if preset.get("is_builtin"): raise HTTPException(status_code=403, detail="cannot delete a built-in preset") ok = presets.delete_preset(ident) if not ok: raise HTTPException(status_code=404, detail="preset not found") return {"ok": True} # Seed built-in presets once at startup so they always exist. Wrapped # so a DB failure can never break the dashboard server. try: presets.seed_builtin_presets() except Exception as e: log.error("Failed to seed built-in presets at startup: %s", e) return app