#!/usr/bin/env python3 """ BTC Accumulation Signal Optimizer -- Web Dashboard FastAPI server with inline HTML/CSS/JS dashboard. """ import json import os import sys import threading import requests from fastapi import FastAPI from fastapi.responses import FileResponse, HTMLResponse, JSONResponse from pydantic import BaseModel BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, BASE_DIR) import orchestrator app = FastAPI(title="BTC Accumulation Signal Optimizer") CONFIG_DIR = os.path.join(BASE_DIR, "config") RESULTS_DIR = os.path.join(BASE_DIR, "results") ITERATIONS_LOG = os.path.join(RESULTS_DIR, "iterations.jsonl") LLM_SETTINGS_PATH = os.path.join(CONFIG_DIR, "llm_settings.json") _opt_thread = None class ConfigUpdate(BaseModel): config: dict class LLMSettingsUpdate(BaseModel): provider: str model: str providers: dict class TestConnectionRequest(BaseModel): provider: str providers: dict class FetchModelsRequest(BaseModel): provider: str providers: dict def _load_llm_settings(): if os.path.exists(LLM_SETTINGS_PATH): with open(LLM_SETTINGS_PATH) as f: return json.load(f) return { "provider": "ollama", "model": "qwen3.5:27b", "providers": { "ollama": {"base_url": "http://100.100.242.21:11434"}, "lmstudio": {"base_url": "http://100.100.242.21:1234"}, "openai": {"api_key": ""}, "anthropic": {"api_key": ""}, "openrouter": {"api_key": ""}, }, } def _mask_api_key(key): if not key or len(key) < 8: return "" return "••••••••" + key[-4:] def _safe_settings(settings): """Return settings with API keys masked.""" out = json.loads(json.dumps(settings)) for name, cfg in out.get("providers", {}).items(): if "api_key" in cfg: cfg["api_key"] = _mask_api_key(cfg["api_key"]) return out def _merge_api_keys(new_providers, existing_providers): """Preserve existing API keys when the incoming value is masked.""" for name, cfg in new_providers.items(): if "api_key" in cfg: masked = cfg["api_key"] if masked.startswith("••••") or masked == "": existing_key = existing_providers.get(name, {}).get("api_key", "") cfg["api_key"] = existing_key @app.get("/api/status") def api_status(): return orchestrator.get_status() @app.get("/api/iterations") def api_iterations(): iterations = orchestrator.load_iteration_history() slim = [] for it in iterations: entry = {k: v for k, v in it.items() if k not in ("config", "results")} slim.append(entry) return slim @app.get("/api/config") def api_config(): best = os.path.join(CONFIG_DIR, "best_config.json") initial = os.path.join(CONFIG_DIR, "initial_config.json") path = best if os.path.exists(best) else initial with open(path) as f: return json.load(f) @app.put("/api/config") def api_update_config(body: ConfigUpdate): path = os.path.join(CONFIG_DIR, "current_config.json") with open(path, "w") as f: json.dump(body.config, f, indent=2) return {"ok": True} @app.post("/api/start") def api_start(): global _opt_thread status = orchestrator.get_status() if status["state"] == "running": return JSONResponse({"error": "Already running"}, status_code=409) _opt_thread = threading.Thread( target=orchestrator.run_optimization_loop, daemon=True ) _opt_thread.start() return {"ok": True, "message": "Optimization started"} @app.post("/api/stop") def api_stop(): orchestrator.request_stop() return {"ok": True, "message": "Stop requested"} @app.get("/api/best") def api_best(): best_path = os.path.join(CONFIG_DIR, "best_config.json") if not os.path.exists(best_path): return {"config": None, "best_score": 0} with open(best_path) as f: config = json.load(f) iterations = orchestrator.load_iteration_history() best_iter = ( max(iterations, key=lambda x: x.get("cost_improvement", 0)) if iterations else {} ) return {"config": config, "best_iteration": best_iter} @app.get("/api/download/iterations") def api_download_iterations(): if os.path.exists(ITERATIONS_LOG): return FileResponse(ITERATIONS_LOG, filename="iterations.jsonl") return JSONResponse({"error": "No iterations yet"}, status_code=404) @app.get("/api/download/best-config") def api_download_best_config(): path = os.path.join(CONFIG_DIR, "best_config.json") if os.path.exists(path): return FileResponse(path, filename="best_config.json") return JSONResponse({"error": "No best config yet"}, status_code=404) # ── Settings API ────────────────────────────────────────────────────────── @app.get("/api/settings") def api_get_settings(): settings = _load_llm_settings() return _safe_settings(settings) @app.post("/api/settings") def api_save_settings(body: LLMSettingsUpdate): existing = _load_llm_settings() new_settings = { "provider": body.provider, "model": body.model, "providers": body.providers, } _merge_api_keys(new_settings["providers"], existing.get("providers", {})) with open(LLM_SETTINGS_PATH, "w") as f: json.dump(new_settings, f, indent=2) return {"ok": True, "message": "Settings saved"} @app.post("/api/settings/test") def api_test_connection(body: TestConnectionRequest): """Test connection to a provider and return available models.""" existing = _load_llm_settings() providers = json.loads(json.dumps(body.providers)) _merge_api_keys(providers, existing.get("providers", {})) provider = body.provider try: models = _fetch_models(provider, providers) return {"ok": True, "models": models, "message": f"Connected — {len(models)} model(s) found"} except requests.exceptions.ConnectionError: return JSONResponse( {"ok": False, "error": "Connection refused — is the server running?"}, status_code=502, ) except requests.exceptions.Timeout: return JSONResponse( {"ok": False, "error": "Connection timed out"}, status_code=504, ) except Exception as e: return JSONResponse( {"ok": False, "error": str(e)}, status_code=500, ) @app.post("/api/settings/models") def api_fetch_models(body: FetchModelsRequest): """Fetch available models for a provider (proxied to avoid CORS).""" existing = _load_llm_settings() providers = json.loads(json.dumps(body.providers)) _merge_api_keys(providers, existing.get("providers", {})) try: models = _fetch_models(body.provider, providers) return {"ok": True, "models": models} except Exception as e: return JSONResponse({"ok": False, "error": str(e)}, status_code=500) def _fetch_models(provider, providers): """Fetch model list from a provider. Returns list of {id, name}.""" cfg = providers.get(provider, {}) if provider == "ollama": base_url = cfg.get("base_url", "http://100.100.242.21:11434") resp = requests.get(f"{base_url}/api/tags", timeout=10) resp.raise_for_status() data = resp.json() return [{"id": m["name"], "name": m["name"]} for m in data.get("models", [])] elif provider == "lmstudio": base_url = cfg.get("base_url", "http://100.100.242.21:1234") resp = requests.get(f"{base_url}/v1/models", timeout=10) resp.raise_for_status() data = resp.json() return [{"id": m["id"], "name": m["id"]} for m in data.get("data", [])] elif provider == "openai": api_key = cfg.get("api_key", "") if not api_key: raise ValueError("OpenAI API key is required") resp = requests.get( "https://api.openai.com/v1/models", headers={"Authorization": f"Bearer {api_key}"}, timeout=15, ) resp.raise_for_status() data = resp.json() models = [m for m in data.get("data", []) if m["id"].startswith("gpt-") or "chat" in m.get("id", "")] models.sort(key=lambda m: m["id"]) return [{"id": m["id"], "name": m["id"]} for m in models] elif provider == "anthropic": api_key = cfg.get("api_key", "") if not api_key: raise ValueError("Anthropic API key is required") resp = requests.get( "https://api.anthropic.com/v1/models", headers={ "x-api-key": api_key, "anthropic-version": "2023-06-01", }, timeout=15, ) resp.raise_for_status() data = resp.json() return [{"id": m["id"], "name": m.get("display_name", m["id"])} for m in data.get("data", [])] elif provider == "openrouter": resp = requests.get("https://openrouter.ai/api/v1/models", timeout=15) resp.raise_for_status() data = resp.json() models = data.get("data", []) models.sort(key=lambda m: m.get("id", "")) return [{"id": m["id"], "name": m.get("name", m["id"])} for m in models[:200]] else: raise ValueError(f"Unknown provider: {provider}") # ── HTML Pages ──────────────────────────────────────────────────────────── # Shared CSS used by both pages SHARED_CSS = """ *,*::before,*::after{box-sizing:border-box;margin:0;padding:0} :root{--bg:#0f172a;--card:#1e293b;--card-hover:#253349;--text:#e2e8f0;--text-dim:#94a3b8;--accent:#f7931a;--green:#22c55e;--red:#ef4444;--yellow:#eab308;--border:#334155;--mono:'JetBrains Mono','Fira Code','Courier New',monospace;--cyan:#22d3ee} body{font-family:'Inter',sans-serif;background:var(--bg);color:var(--text);min-height:100vh} .container{max-width:1400px;margin:0 auto;padding:16px} h1{font-size:1.5rem;font-weight:700;display:flex;align-items:center;gap:10px} h1 .btc{color:var(--accent);font-size:1.8rem} h2{font-size:1rem;font-weight:600;color:var(--text-dim);margin-bottom:12px;text-transform:uppercase;letter-spacing:.05em;font-size:.8rem} .header{display:flex;justify-content:space-between;align-items:center;padding:16px 0;border-bottom:1px solid var(--border);margin-bottom:16px;flex-wrap:wrap;gap:12px} .nav{display:flex;gap:4px;align-items:center} .nav a{color:var(--text-dim);text-decoration:none;font-size:.85rem;font-weight:600;padding:6px 14px;border-radius:6px;transition:all .15s} .nav a:hover{color:var(--text);background:var(--card)} .nav a.active{color:var(--cyan);background:var(--card);border:1px solid var(--border)} .controls{display:flex;gap:8px;align-items:center} .btn{padding:8px 18px;border:none;border-radius:6px;font-family:inherit;font-weight:600;font-size:.85rem;cursor:pointer;transition:all .15s} .btn-start{background:var(--green);color:#000}.btn-start:hover{background:#16a34a} .btn-stop{background:var(--red);color:#fff}.btn-stop:hover{background:#dc2626} .btn-secondary{background:var(--border);color:var(--text)}.btn-secondary:hover{background:var(--card-hover)} .btn-accent{background:var(--accent);color:#000}.btn-accent:hover{background:#e8850f} .btn-cyan{background:var(--cyan);color:#000}.btn-cyan:hover{background:#06b6d4} .btn:disabled{opacity:.4;cursor:not-allowed} .status-badge{display:inline-flex;align-items:center;gap:6px;padding:4px 12px;border-radius:20px;font-size:.8rem;font-weight:600} .status-idle{background:#1e3a5f;color:#60a5fa} .status-running{background:#1a3a2a;color:var(--green)} .status-completed{background:#3a2a1a;color:var(--accent)} .status-error{background:#3a1a1a;color:var(--red)} .pulse{width:8px;height:8px;border-radius:50%;background:currentColor;animation:pulse 1.5s infinite} @keyframes pulse{0%,100%{opacity:1}50%{opacity:.3}} .best-score{text-align:right} .best-score .label{font-size:.7rem;text-transform:uppercase;letter-spacing:.1em;color:var(--text-dim)} .best-score .value{font-size:2.2rem;font-weight:700;color:var(--accent);font-family:var(--mono)} .best-score .unit{font-size:1rem;color:var(--text-dim)} .card{background:var(--card);border-radius:10px;padding:16px;border:1px solid var(--border)} .footer{text-align:center;color:var(--text-dim);font-size:.75rem;padding:20px 0;margin-top:16px;border-top:1px solid var(--border)} .toast{position:fixed;top:20px;right:20px;padding:12px 20px;border-radius:8px;font-size:.85rem;font-weight:600;z-index:9999;opacity:0;transform:translateY(-10px);transition:all .3s;pointer-events:none} .toast.show{opacity:1;transform:translateY(0)} .toast-success{background:var(--green);color:#000} .toast-error{background:var(--red);color:#fff} """ # Shared HTML head SHARED_HEAD = """ """ # Navigation bar HTML (parameterized via JS to highlight active page) NAV_HTML = """
Dashboard ⚙ Settings
""" # Toast JS helper (shared) TOAST_JS = """ function showToast(msg, type) { let t = document.getElementById('toast'); if (!t) { t = document.createElement('div'); t.id = 'toast'; t.className = 'toast'; document.body.appendChild(t); } t.textContent = msg; t.className = 'toast toast-' + type + ' show'; setTimeout(() => { t.classList.remove('show'); }, 3500); } """ DASHBOARD_HTML = """ """ + SHARED_HEAD + """ BTC Accumulation Signal Optimizer

₿ Accumulation Signal Optimizer

Idle """ + NAV_HTML + """
Best Cost Improvement
0.0%

Iterations

#Cost Imp%SignalsFrequencyR2BottomsTopsModel

Cost Improvement Over Iterations

LLM Analysis

No suggestions yet.

Downloads

iterations.jsonl best_config.json
▶

Configuration Editor

BTC Accumulation Signal Optimizer — VPS → Windows GPU → Mac Mini LLM
""" SETTINGS_HTML = """ """ + SHARED_HEAD + """ Settings — BTC Accumulation Signal Optimizer

₿ Accumulation Signal Optimizer

""" + NAV_HTML + """

⚙ LLM Provider Settings

Provider

Connection

BTC Accumulation Signal Optimizer — VPS → Windows GPU → Mac Mini LLM
""" @app.get("/", response_class=HTMLResponse) def dashboard(): return DASHBOARD_HTML @app.get("/settings", response_class=HTMLResponse) def settings_page(): return SETTINGS_HTML if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=3088) @app.post("/api/clear") def api_clear(): """Clear iteration history for a fresh run.""" import glob results_dir = os.path.join(BASE_DIR, "results") # Clear iterations log log_path = os.path.join(results_dir, "iterations.jsonl") if os.path.exists(log_path): os.remove(log_path) # Clear individual result files for f in glob.glob(os.path.join(results_dir, "results_iter_*.json")): os.remove(f) # Reset best config to initial initial = os.path.join(BASE_DIR, "config", "initial_config.json") best = os.path.join(BASE_DIR, "config", "best_config.json") current = os.path.join(BASE_DIR, "config", "current_config.json") if os.path.exists(initial): import shutil shutil.copy(initial, best) shutil.copy(initial, current) # Clear in-memory status orchestrator._status["llm_suggestions"] = [] orchestrator._status["best_score"] = 0.0 orchestrator._status["iteration"] = 0 return {"ok": True, "message": "History cleared. Ready for fresh run."}