#!/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 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") _opt_thread = None class ConfigUpdate(BaseModel): config: dict @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) DASHBOARD_HTML = """
| # | Cost Imp% | Signals | Frequency | R2 | Bottoms | Tops | Model |
|---|