124 lines
3.8 KiB
Python
124 lines
3.8 KiB
Python
"""FastAPI web demo for the Adaptive Barrier Monitor."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any, Literal
|
|
|
|
import numpy as np
|
|
from fastapi import FastAPI
|
|
from fastapi.responses import HTMLResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import BaseModel, Field
|
|
|
|
# Support both an installed package (`pip install -e .`) and direct launches
|
|
# from a source checkout (`uvicorn webapp.app:app`).
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
SRC_DIR = REPO_ROOT / "src"
|
|
if str(SRC_DIR) not in sys.path:
|
|
sys.path.insert(0, str(SRC_DIR))
|
|
|
|
from adaptive_barrier import __version__
|
|
from adaptive_barrier.engine import (
|
|
DROP_FRACTION,
|
|
LOG_BARRIER,
|
|
RISE_FRACTION,
|
|
TYPICAL_ANNUAL_VOL,
|
|
WINDOW_MINUTES,
|
|
run_monte_carlo_simulation,
|
|
)
|
|
|
|
STATIC_DIR = REPO_ROOT / "webapp" / "static"
|
|
|
|
|
|
class SimulateRequest(BaseModel):
|
|
"""Validated Monte Carlo and detector-comparison parameters."""
|
|
|
|
S0: float = Field(100.0, gt=0.0, le=10000.0)
|
|
sigma_annual: float = Field(TYPICAL_ANNUAL_VOL, gt=0.0, le=2.0)
|
|
mu_annual: float = Field(0.0, ge=-0.5, le=0.5)
|
|
window_minutes: float = Field(1950.0, gt=0.0, le=39000.0)
|
|
n_paths: int = Field(20, ge=1, le=100)
|
|
n_steps: int = Field(500, ge=50, le=2000)
|
|
use_jumps: bool = Field(
|
|
False,
|
|
description="Use Merton jump diffusion. The Brownian miss proxy does not control jumps.",
|
|
)
|
|
jump_intensity: float = Field(25.0, ge=0.0, le=500.0)
|
|
jump_mean: float = Field(-0.02, ge=-0.5, le=0.5)
|
|
jump_sigma: float = Field(0.05, ge=0.0, le=0.5)
|
|
eps: float = Field(1e-3, gt=0.0, lt=1.0)
|
|
dt_cap_minutes: float | None = Field(None, gt=0.0, le=1440.0)
|
|
rng_seed: int | None = Field(None, ge=0, le=2_147_483_647)
|
|
drop_fraction: float = Field(DROP_FRACTION, gt=0.0, lt=1.0)
|
|
rise_fraction: float = Field(RISE_FRACTION, gt=0.0, le=1.0)
|
|
max_detection_lag_steps: int | None = Field(3, ge=0, le=100)
|
|
comparison_mode: Literal["equal_budget", "fixed_cadence"] = "equal_budget"
|
|
fixed_cadence_minutes: float = Field(60.0, gt=0.0, le=39000.0)
|
|
|
|
|
|
def _file_version(path: Path) -> str:
|
|
try:
|
|
stat = path.stat()
|
|
return f"{int(stat.st_mtime)}-{stat.st_size}"
|
|
except FileNotFoundError:
|
|
return "missing"
|
|
|
|
|
|
def _json_safe(obj: Any) -> Any:
|
|
if isinstance(obj, dict):
|
|
return {str(key): _json_safe(value) for key, value in obj.items()}
|
|
if isinstance(obj, (list, tuple)):
|
|
return [_json_safe(value) for value in obj]
|
|
if isinstance(obj, np.integer):
|
|
return int(obj)
|
|
if isinstance(obj, np.floating):
|
|
value = float(obj)
|
|
return value if np.isfinite(value) else None
|
|
if isinstance(obj, np.ndarray):
|
|
return _json_safe(obj.tolist())
|
|
if isinstance(obj, np.bool_):
|
|
return bool(obj)
|
|
return obj
|
|
|
|
|
|
app = FastAPI(title="Adaptive Barrier Monitor", version=__version__)
|
|
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
|
|
|
|
|
|
@app.get("/")
|
|
def index() -> HTMLResponse:
|
|
html = (STATIC_DIR / "index.html").read_text(encoding="utf-8")
|
|
html = re.sub(
|
|
r'app\.css\?v=[^\s"\']+',
|
|
f"app.css?v={_file_version(STATIC_DIR / 'app.css')}",
|
|
html,
|
|
)
|
|
html = re.sub(
|
|
r'app\.js\?v=[^\s"\']+',
|
|
f"app.js?v={_file_version(STATIC_DIR / 'app.js')}",
|
|
html,
|
|
)
|
|
return HTMLResponse(content=html)
|
|
|
|
|
|
@app.get("/api/health")
|
|
def health() -> dict:
|
|
return {
|
|
"ok": True,
|
|
"version": __version__,
|
|
"default_sigma_annual": TYPICAL_ANNUAL_VOL,
|
|
"default_window_minutes": WINDOW_MINUTES,
|
|
"drop_fraction": DROP_FRACTION,
|
|
"rise_fraction": RISE_FRACTION,
|
|
"log_barrier": LOG_BARRIER,
|
|
}
|
|
|
|
|
|
@app.post("/api/simulate")
|
|
def simulate(req: SimulateRequest) -> dict:
|
|
result = run_monte_carlo_simulation(**req.model_dump())
|
|
return _json_safe(result)
|