Spaces:
Paused
Paused
File size: 3,738 Bytes
37173e0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 |
from __future__ import annotations
import html
import json
import os
from pathlib import Path
from typing import Any, Dict, List
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse
from fastapi.staticfiles import StaticFiles
from pygments import highlight
from pygments.formatters import HtmlFormatter
from pygments.lexers import PythonLexer
from runtime import MarketRuntime, ensure_repo_engine_path
ROOT = Path(__file__).resolve().parent
FRONTEND = ROOT / "frontend"
SOURCE_PATH = ensure_repo_engine_path()
app = FastAPI(title="MAYTHOS Live Space", version="1.0.0")
app.mount("/static", StaticFiles(directory=str(FRONTEND)), name="static")
runtime = MarketRuntime(
default_symbol=os.getenv("SYMBOL_DEFAULT", "frxEURUSD"),
base_timeframe=int(os.getenv("BASE_TIMEFRAME", "30")),
debug_mode=True,
)
_SOURCE_RAW = SOURCE_PATH.read_text(encoding="utf-8")
_SOURCE_LINE_COUNT = len(_SOURCE_RAW.splitlines())
_SOURCE_HTML = None
def _render_source_html(source: str) -> str:
lexer = PythonLexer()
formatter = HtmlFormatter(nowrap=True)
lines = []
for idx, line in enumerate(source.splitlines(), start=1):
display = line if line.strip() else " "
hl = highlight(display, lexer, formatter).rstrip("\n")
data_text = html.escape(line)
lines.append(
f'<div class="code-line" data-line="{idx}" data-text="{data_text}">'
f'<span class="ln">{idx}</span>'
f'<span class="code">{hl or " "}</span>'
f"</div>"
)
return "\n".join(lines)
_SOURCE_HTML = _render_source_html(_SOURCE_RAW)
@app.on_event("startup")
async def _startup() -> None:
await runtime.start()
@app.on_event("shutdown")
async def _shutdown() -> None:
await runtime.stop()
@app.get("/", response_class=HTMLResponse)
async def index() -> HTMLResponse:
html_path = FRONTEND / "index.html"
if html_path.exists():
return HTMLResponse(html_path.read_text(encoding="utf-8"))
raise HTTPException(404, "frontend/index.html missing")
@app.get("/api/state")
async def api_state() -> JSONResponse:
return JSONResponse(await runtime.snapshot())
@app.get("/api/health")
async def api_health() -> JSONResponse:
return JSONResponse(await runtime.health())
@app.get("/health")
async def health_plain() -> PlainTextResponse:
data = await runtime.health()
if data["ok"]:
return PlainTextResponse("OK")
return PlainTextResponse("ERROR", status_code=503)
@app.get("/api/source")
async def api_source() -> JSONResponse:
return JSONResponse({"source": _SOURCE_RAW, "line_count": _SOURCE_LINE_COUNT})
@app.get("/api/source-html")
async def api_source_html() -> HTMLResponse:
return HTMLResponse(_SOURCE_HTML)
@app.get("/api/symbols")
async def api_symbols() -> JSONResponse:
return JSONResponse(
{
"symbols": runtime.symbol_options(),
"selected": runtime.selected_symbol,
"market_type": runtime.market_type,
}
)
@app.post("/api/config/symbol/{symbol}")
async def api_set_symbol(symbol: str) -> JSONResponse:
await runtime.set_symbol(symbol)
return JSONResponse({"ok": True, "symbol": runtime.selected_symbol, "market_type": runtime.market_type})
@app.post("/api/config/timeframe/{tf_seconds}")
async def api_set_tf(tf_seconds: int) -> JSONResponse:
await runtime.set_display_timeframe(tf_seconds)
return JSONResponse({"ok": True, "display_timeframe": tf_seconds})
@app.get("/api/source-meta")
async def api_source_meta() -> JSONResponse:
return JSONResponse({
"path": str(SOURCE_PATH),
"line_count": _SOURCE_LINE_COUNT,
})
|