"""NiceGUI dashboard — port progressivo da Streamlit.
Avvio: ``uv run python -m multi_swarm.dashboard.nicegui_app``
Default port 8080. Streamlit resta su 8501 durante la migrazione.
Riusa ``dashboard.data`` (Repository helpers) senza modifiche al backend.
Palette "Neon Trading Dashboard" (ispirata screenshot 2026-05-11):
- BG: #0A0A0F (near-black con tinge blu)
- Surface: #13131A (card base)
- Surface elevata: #1C1C26 (hover/active)
- Primary pink: #FF2D87 (highlight key metrics, max fitness)
- Secondary cyan: #00D9FF (median, secondary curves)
- Accent amber: #FFB800 (warnings, p90)
- Success neon green: #00E676, Danger neon red: #FF3D60
- Text: #FFFFFF (primary), #7A7A8C (muted)
"""
from __future__ import annotations
import html
import json
import os
from pathlib import Path
from typing import Any
import pandas as pd # type: ignore[import-untyped]
import plotly.graph_objects as go # type: ignore[import-untyped]
from nicegui import app, ui
from multi_swarm.dashboard.data import (
evaluations_df,
generations_df,
genomes_df,
get_repo,
get_run_overview,
list_runs_df,
paper_equity_df,
paper_positions_df,
paper_run_summary,
paper_runs_df,
paper_ticks_df,
paper_trades_df,
)
DB_PATH = os.environ.get("DB_PATH", "./runs.db")
REFRESH_INTERVAL_S = 3.0
# --- Neon Trading Dashboard palette ---
COLOR_BG = "#0A0A0F"
COLOR_SURFACE = "#13131A"
COLOR_SURFACE_2 = "#1C1C26"
COLOR_BORDER = "rgba(255, 45, 135, 0.12)"
COLOR_BORDER_HOVER = "rgba(255, 45, 135, 0.45)"
COLOR_PRIMARY = "#FF2D87"
COLOR_SECONDARY = "#00D9FF"
COLOR_ACCENT = "#FFB800"
COLOR_SUCCESS = "#00E676"
COLOR_DANGER = "#FF3D60"
COLOR_TEXT = "#FFFFFF"
COLOR_TEXT_MUTED = "#7A7A8C"
_STATUS_BADGE: dict[str, tuple[str, str]] = {
"running": ("● running", "positive"),
"completed": ("✓ completed", "positive"),
"failed": ("✕ failed", "negative"),
}
_CUSTOM_CSS = f"""
"""
def _json_to_html(obj: Any, indent: int = 0) -> str:
"""Render JSON con span colorati espliciti. Garantisce leggibilità ovunque."""
pad = " " * indent
inner_pad = " " * (indent + 1)
if isinstance(obj, dict):
if not obj:
return '{}'
items = []
for k, v in obj.items():
key = f'"{html.escape(str(k))}"'
val = _json_to_html(v, indent + 1)
items.append(f"{inner_pad}{key}: {val}")
return ('{\n'
+ ',\n'.join(items)
+ f'\n{pad}}}')
if isinstance(obj, list):
if not obj:
return '[]'
items = [_json_to_html(x, indent + 1) for x in obj]
return ('[\n'
+ ',\n'.join(inner_pad + i for i in items)
+ f'\n{pad}]')
if isinstance(obj, bool):
return f'{str(obj).lower()}'
if obj is None:
return 'null'
if isinstance(obj, (int, float)):
return f'{obj}'
return f'"{html.escape(str(obj))}"'
def _apply_theme() -> None:
ui.add_head_html(_CUSTOM_CSS)
ui.dark_mode().enable()
ui.colors(
primary=COLOR_PRIMARY,
secondary=COLOR_SECONDARY,
accent=COLOR_ACCENT,
dark=COLOR_BG,
dark_page=COLOR_BG,
positive=COLOR_SUCCESS,
negative=COLOR_DANGER,
info=COLOR_PRIMARY,
warning=COLOR_ACCENT,
)
def _build_header(active: str) -> None:
with ui.header().classes("items-center justify-between q-px-lg q-py-md"):
with ui.row().classes("items-center gap-8"):
with ui.row().classes("items-center gap-2").classes("brand"):
ui.html('')
ui.html('Multi-Swarm / Coevolutivo')
with ui.row().classes("items-center gap-1"):
for path, label in (
("/", "Overview"),
("/convergence", "Convergence"),
("/genomes", "Genomes"),
("/paper", "Paper"),
):
cls = "nav-link active" if active == path else "nav-link"
ui.link(label, path).classes(cls)
with ui.row().classes("items-center gap-3"):
ui.html(f''
f'⛁ {Path(DB_PATH).resolve().name}')
def _runs_options() -> dict[str, str]:
repo = get_repo(DB_PATH)
runs = list_runs_df(repo)
if runs.empty:
return {}
return {
row["id"]: f"{row['name']} — {row['status']} ({row['started_at'][:16]})"
for _, row in runs.iterrows()
}
def _snapshot(run_id: str) -> dict[str, Any]:
repo = get_repo(DB_PATH)
ov = get_run_overview(repo, run_id)
evals = evaluations_df(repo, run_id)
gens = generations_df(repo, run_id)
cfg = ov["config"]
pop_size = int(cfg.get("population_size", 0))
n_gens = int(cfg.get("n_generations", 0))
evals_total = max(pop_size * n_gens, 1)
evals_done = len(evals)
gens_done = int(gens["completed_at"].notna().sum()) if not gens.empty else 0
# runs.total_cost_usd è 0 finché complete_run non viene chiamato.
# Per le run in corso leggiamo la somma live da cost_records.
live_cost = float(repo.total_cost(run_id)) if ov["status"] == "running" else float(
ov["total_cost_usd"]
)
top_fit = float(evals["fitness"].max()) if evals_done else float("nan")
median_fit = float(evals["fitness"].median()) if evals_done else float("nan")
parse_success = (
100.0 * float(evals["parse_error"].isna().sum()) / evals_done if evals_done else 0.0
)
return {
"status": ov["status"],
"name": cfg.get("run_name", "—"),
"started_at": ov["started_at"],
"completed_at": ov["completed_at"] or "—",
"cost_usd": live_cost,
"pop_size": pop_size,
"n_gens": n_gens,
"evals_done": evals_done,
"evals_total": evals_total,
"gens_done": gens_done,
"top_fit": top_fit,
"median_fit": median_fit,
"parse_success": parse_success,
"config": cfg,
"gens_df": gens,
}
def _convergence_figure(gens_df: Any) -> go.Figure:
fig = go.Figure()
if gens_df.empty:
fig.add_annotation(
text="Nessuna generazione registrata", x=0.5, y=0.5, showarrow=False,
font={"color": COLOR_TEXT_MUTED, "size": 14},
)
else:
fig.add_trace(
go.Scatter(
x=gens_df["generation_idx"], y=gens_df["fitness_max"],
name="max", mode="lines+markers",
line={"color": COLOR_PRIMARY, "width": 3, "shape": "spline", "smoothing": 0.6},
marker={"size": 9, "color": COLOR_PRIMARY,
"line": {"color": "#fff", "width": 1}},
fill="tozeroy",
fillcolor="rgba(255, 45, 135, 0.12)",
)
)
fig.add_trace(
go.Scatter(
x=gens_df["generation_idx"], y=gens_df["fitness_p90"],
name="p90", mode="lines+markers",
line={"color": COLOR_ACCENT, "width": 2, "dash": "dot", "shape": "spline"},
marker={"size": 7, "color": COLOR_ACCENT},
)
)
fig.add_trace(
go.Scatter(
x=gens_df["generation_idx"], y=gens_df["fitness_median"],
name="median", mode="lines+markers",
line={"color": COLOR_SECONDARY, "width": 2, "shape": "spline"},
marker={"size": 7, "color": COLOR_SECONDARY},
)
)
fig.update_layout(
template="plotly_dark",
paper_bgcolor=COLOR_SURFACE,
plot_bgcolor=COLOR_SURFACE,
font={"color": COLOR_TEXT},
xaxis={"title": "generation", "gridcolor": "rgba(148, 163, 184, 0.08)", "dtick": 1},
yaxis={"title": "fitness", "gridcolor": "rgba(148, 163, 184, 0.08)"},
title={"text": "Fitness convergence", "font": {"color": COLOR_TEXT, "size": 18}},
legend={"bgcolor": "rgba(19, 19, 26, 0.95)", "bordercolor": COLOR_PRIMARY, "borderwidth": 1},
margin={"l": 50, "r": 30, "t": 50, "b": 50},
)
return fig
def _entropy_figure(gens_df: Any) -> go.Figure:
fig = go.Figure()
if not gens_df.empty:
fig.add_trace(
go.Scatter(
x=gens_df["generation_idx"], y=gens_df["entropy"],
mode="lines+markers",
line={"color": COLOR_SECONDARY, "width": 3, "shape": "spline", "smoothing": 0.6},
marker={"size": 9, "color": COLOR_SECONDARY,
"line": {"color": "#fff", "width": 1}},
fill="tozeroy",
fillcolor="rgba(0, 217, 255, 0.12)",
name="entropy",
)
)
fig.add_hline(
y=0.5, line_dash="dash", line_color=COLOR_ACCENT,
annotation_text="gate threshold (0.5)",
annotation_font_color=COLOR_ACCENT,
)
fig.update_layout(
template="plotly_dark",
paper_bgcolor=COLOR_SURFACE,
plot_bgcolor=COLOR_SURFACE,
font={"color": COLOR_TEXT},
xaxis={"title": "generation", "gridcolor": "rgba(148, 163, 184, 0.08)", "dtick": 1},
yaxis={"title": "entropy", "gridcolor": "rgba(148, 163, 184, 0.08)"},
title={"text": "Diversity (fitness entropy)", "font": {"color": COLOR_TEXT, "size": 18}},
margin={"l": 50, "r": 30, "t": 50, "b": 50},
)
return fig
@ui.page("/")
def index() -> None:
_apply_theme()
_build_header(active="/")
options = _runs_options()
if not options:
ui.label("Nessuna run nel database.").classes("text-h5")
return
state: dict[str, Any] = {"run_id": next(iter(options))}
with ui.row().classes("w-full items-center gap-4 q-mb-md"):
select = ui.select(options=options, value=state["run_id"], label="Run").classes(
"flex-grow"
)
status_badge = ui.badge("…", color="primary").classes("text-body1 q-pa-sm")
ui.button("🔄 Refresh", on_click=lambda: refresh()).props("outline color=primary")
with ui.card().classes("w-full"):
ui.label("Progresso run").classes("text-subtitle1")
gen_label = ui.label("Generations: 0/0")
gen_bar = ui.linear_progress(0.0, show_value=False).props("size=20px color=primary")
eval_label = ui.label("Evaluations: 0/0 (0.0%)")
eval_bar = ui.linear_progress(0.0, show_value=False).props("size=20px color=accent")
with ui.row().classes("w-full gap-4"):
with ui.card().classes("flex-grow metric-card accent-cyan"):
ui.label("Top fitness").classes("text-caption")
top_lbl = ui.label("—").classes("text-h4")
with ui.card().classes("flex-grow metric-card accent-purple"):
ui.label("Median fitness").classes("text-caption")
median_lbl = ui.label("—").classes("text-h4")
with ui.card().classes("flex-grow metric-card accent-amber"):
ui.label("Parse success").classes("text-caption")
parse_lbl = ui.label("—").classes("text-h4")
with ui.card().classes("flex-grow metric-card accent-green"):
ui.label("Cost (USD)").classes("text-caption")
cost_lbl = ui.label("—").classes("text-h4")
with ui.row().classes("w-full gap-4 q-mt-md"):
started_lbl = ui.label("Started: —")
completed_lbl = ui.label("Completed: —")
ui.separator()
ui.label("Config").classes("text-subtitle1")
cfg_code = ui.html('').classes("w-full")
def refresh() -> None:
run_id = select.value
if not run_id:
return
try:
s = _snapshot(run_id)
except Exception as e: # noqa: BLE001
ui.notify(f"Errore: {e}", type="negative")
return
text, color = _STATUS_BADGE.get(s["status"], (s["status"], "primary"))
status_badge.text = text
status_badge.props(f"color={color}")
gen_frac = min(s["gens_done"] / max(s["n_gens"], 1), 1.0)
eval_frac = min(s["evals_done"] / s["evals_total"], 1.0)
gen_bar.value = gen_frac
eval_bar.value = eval_frac
gen_label.text = f"Generations: {s['gens_done']}/{s['n_gens']}"
eval_label.text = (
f"Evaluations: {s['evals_done']}/{s['evals_total']} ({100 * eval_frac:.1f}%)"
)
top_lbl.text = f"{s['top_fit']:.4f}" if s["evals_done"] else "—"
median_lbl.text = f"{s['median_fit']:.4f}" if s["evals_done"] else "—"
parse_lbl.text = f"{s['parse_success']:.1f}%" if s["evals_done"] else "—"
cost_lbl.text = f"${s['cost_usd']:.4f}"
started_lbl.text = f"Started: {s['started_at']}"
completed_lbl.text = f"Completed: {s['completed_at']}"
cfg_code.content = f'{_json_to_html(s["config"])}'
def on_select_change() -> None:
state["run_id"] = select.value
refresh()
select.on_value_change(on_select_change)
ui.timer(REFRESH_INTERVAL_S, refresh)
refresh()
@ui.page("/convergence")
def convergence() -> None:
_apply_theme()
_build_header(active="/convergence")
options = _runs_options()
if not options:
ui.label("Nessuna run nel database.").classes("text-h5")
return
state: dict[str, Any] = {"run_id": next(iter(options))}
with ui.row().classes("w-full items-center gap-4 q-mb-md"):
select = ui.select(options=options, value=state["run_id"], label="Run").classes(
"flex-grow"
)
gen_count_lbl = ui.label("Gens: 0/0").classes("text-body1").style(
f"color: {COLOR_PRIMARY}; font-weight: 600;"
)
ui.button("🔄 Refresh", on_click=lambda: refresh()).props("outline color=primary")
fitness_plot = ui.plotly(_convergence_figure(generations_df(get_repo(DB_PATH), state["run_id"]))).classes("w-full")
entropy_plot = ui.plotly(_entropy_figure(generations_df(get_repo(DB_PATH), state["run_id"]))).classes("w-full q-mt-md")
ui.separator()
ui.label("Tabella generazioni").classes("text-subtitle1 q-mt-md")
gens_table = ui.table(
columns=[
{"name": "generation_idx", "label": "gen", "field": "generation_idx", "sortable": True},
{"name": "n_genomes", "label": "n", "field": "n_genomes"},
{"name": "fitness_max", "label": "max", "field": "fitness_max"},
{"name": "fitness_p90", "label": "p90", "field": "fitness_p90"},
{"name": "fitness_median", "label": "median", "field": "fitness_median"},
{"name": "entropy", "label": "entropy", "field": "entropy"},
{"name": "completed_at", "label": "completed", "field": "completed_at"},
],
rows=[],
row_key="generation_idx",
).classes("w-full")
def refresh() -> None:
run_id = select.value
if not run_id:
return
try:
gens = generations_df(get_repo(DB_PATH), run_id)
ov = get_run_overview(get_repo(DB_PATH), run_id)
except Exception as e: # noqa: BLE001
ui.notify(f"Errore: {e}", type="negative")
return
n_gens = int(ov["config"].get("n_generations", 0))
gens_done = int(gens["completed_at"].notna().sum()) if not gens.empty else 0
gen_count_lbl.text = f"Gens: {gens_done}/{n_gens}"
fitness_plot.update_figure(_convergence_figure(gens))
entropy_plot.update_figure(_entropy_figure(gens))
if gens.empty:
gens_table.rows = []
else:
display_cols = [
"generation_idx", "n_genomes",
"fitness_max", "fitness_p90", "fitness_median",
"entropy", "completed_at",
]
gens_table.rows = [
{
col: (round(v, 6) if isinstance(v, float) else v)
for col, v in row.items()
if col in display_cols
}
for _, row in gens.iterrows()
]
gens_table.update()
def on_select_change() -> None:
state["run_id"] = select.value
refresh()
select.on_value_change(on_select_change)
ui.timer(REFRESH_INTERVAL_S, refresh)
refresh()
@ui.page("/genomes")
def genomes() -> None:
_apply_theme()
_build_header(active="/genomes")
options = _runs_options()
if not options:
ui.label("Nessuna run nel database.").classes("text-h5")
return
state: dict[str, Any] = {
"run_id": next(iter(options)),
"selected_gid": None,
"merged": None,
}
with ui.row().classes("w-full items-center gap-4 q-mb-md"):
select = ui.select(options=options, value=state["run_id"], label="Run").classes(
"flex-grow"
)
top_k_select = ui.select(
options={10: "Top 10", 25: "Top 25", 50: "Top 50"},
value=10,
label="Top K",
)
ui.button("🔄 Refresh", on_click=lambda: refresh()).props("outline color=primary")
ui.label("Top genomi per fitness").classes("text-subtitle1 q-mt-sm")
top_table = ui.table(
columns=[
{"name": "genome_id", "label": "id", "field": "genome_id", "align": "left"},
{"name": "fitness", "label": "fitness", "field": "fitness", "sortable": True},
{"name": "dsr", "label": "DSR", "field": "dsr"},
{"name": "sharpe", "label": "Sharpe", "field": "sharpe"},
{"name": "max_dd", "label": "max DD", "field": "max_dd"},
{"name": "n_trades", "label": "trades", "field": "n_trades"},
{"name": "cognitive_style", "label": "style", "field": "cognitive_style"},
{"name": "temperature", "label": "T", "field": "temperature"},
{"name": "lookback_window", "label": "lookback", "field": "lookback_window"},
],
rows=[],
row_key="genome_id",
selection="single",
).classes("w-full")
ui.separator().classes("q-my-md")
with ui.card().classes("w-full"):
ui.label("Ispezione genoma").classes("text-subtitle1")
detail_hint = ui.label("Seleziona un genoma dalla tabella sopra.").classes(
"text-caption"
).style(f"color: {COLOR_TEXT_MUTED};")
with ui.row().classes("w-full gap-4 q-mt-sm"):
with ui.card().classes("flex-grow metric-card accent-cyan"):
ui.label("fitness").classes("text-caption")
fit_lbl = ui.label("—").classes("text-h4")
with ui.card().classes("flex-grow metric-card accent-purple"):
ui.label("DSR").classes("text-caption")
dsr_lbl = ui.label("—").classes("text-h4")
with ui.card().classes("flex-grow metric-card accent-amber"):
ui.label("Sharpe").classes("text-caption")
sharpe_lbl = ui.label("—").classes("text-h4")
with ui.card().classes("flex-grow metric-card"):
ui.label("max DD").classes("text-caption")
dd_lbl = ui.label("—").classes("text-h4")
with ui.card().classes("flex-grow metric-card accent-green"):
ui.label("trades").classes("text-caption")
trades_lbl = ui.label("—").classes("text-h4")
with ui.card().classes("flex-grow metric-card"):
ui.label("style").classes("text-caption")
style_lbl = ui.label("—").classes("text-h4")
ui.label("System prompt").classes("text-subtitle1 q-mt-md")
prompt_code = ui.html('—
').classes("w-full")
ui.label("Raw LLM output").classes("text-subtitle1 q-mt-md")
raw_code = ui.html('—
').classes("w-full")
parse_error_lbl = ui.label("").classes("q-mt-sm").style(
"color: #FF6B6B; font-weight: 600;"
)
def _render_detail(row: dict[str, Any]) -> None:
detail_hint.text = f"Genoma: {row.get('genome_id', '—')}"
fit_lbl.text = f"{float(row.get('fitness', 0)):.4f}"
dsr_lbl.text = f"{float(row.get('dsr', 0)):.4f}"
sharpe_lbl.text = f"{float(row.get('sharpe', 0)):.3f}"
dd_lbl.text = f"{float(row.get('max_dd', 0)):.3f}"
trades_lbl.text = str(int(row.get("n_trades", 0)))
style_lbl.text = str(row.get("cognitive_style", "—"))
prompt_code.content = (
f'{html.escape(str(row.get("system_prompt", "—")))}'
)
raw_code.content = (
f'{html.escape(str(row.get("raw_text", "—") or "—"))}'
)
pe = row.get("parse_error")
parse_error_lbl.text = f"❌ Parse error: {pe}" if pe else ""
def refresh() -> None:
run_id = select.value
if not run_id:
return
try:
repo = get_repo(DB_PATH)
evals = evaluations_df(repo, run_id)
gens = genomes_df(repo, run_id)
except Exception as e: # noqa: BLE001
ui.notify(f"Errore: {e}", type="negative")
return
if evals.empty:
top_table.rows = []
top_table.update()
return
merged = evals.merge(
gens, left_on="genome_id", right_on="id", how="left", suffixes=("", "_g")
)
state["merged"] = merged
k = int(top_k_select.value)
top = merged.sort_values("fitness", ascending=False).head(k)
rows = []
for _, r in top.iterrows():
rows.append(
{
"genome_id": str(r.get("genome_id", "—"))[:12] + "…",
"fitness": round(float(r.get("fitness", 0)), 4),
"dsr": round(float(r.get("dsr", 0)), 4),
"sharpe": round(float(r.get("sharpe", 0)), 3),
"max_dd": round(float(r.get("max_dd", 0)), 3),
"n_trades": int(r.get("n_trades", 0)),
"cognitive_style": str(r.get("cognitive_style", "—")),
"temperature": round(float(r.get("temperature", 0)), 2),
"lookback_window": int(r.get("lookback_window", 0)),
"_full_id": str(r.get("genome_id", "")),
}
)
top_table.rows = rows
top_table.update()
sel = state.get("selected_gid")
if sel:
match = merged[merged["genome_id"] == sel]
if not match.empty:
_render_detail(match.iloc[0].to_dict())
def on_row_selected(e: Any) -> None:
if not e.selection:
return
full_id = e.selection[0].get("_full_id")
if not full_id:
return
state["selected_gid"] = full_id
merged = state.get("merged")
if merged is None:
return
match = merged[merged["genome_id"] == full_id]
if not match.empty:
_render_detail(match.iloc[0].to_dict())
def on_select_change() -> None:
state["run_id"] = select.value
state["selected_gid"] = None
refresh()
select.on_value_change(on_select_change)
top_k_select.on_value_change(lambda _: refresh())
top_table.on("selection", on_row_selected)
ui.timer(REFRESH_INTERVAL_S, refresh)
refresh()
def _paper_runs_options(only_running: bool = False) -> dict[str, str]:
runs = paper_runs_df(DB_PATH)
if runs.empty:
return {}
if only_running:
runs = runs[runs["status"] == "running"]
if runs.empty:
return {}
return {
row["id"]: f"{row['name']} — {row['status']} ({row['started_at'][:16]})"
for _, row in runs.iterrows()
}
def _paper_equity_figure(eq_df: Any, initial_capital: float) -> go.Figure:
fig = go.Figure()
if eq_df is not None and not eq_df.empty:
ts = pd.to_datetime(eq_df["ts"])
fig.add_trace(
go.Scatter(
x=ts,
y=eq_df["equity"],
mode="lines",
line={"color": COLOR_PRIMARY, "width": 2},
name="equity",
)
)
fig.add_hline(
y=initial_capital,
line={"color": COLOR_TEXT_MUTED, "width": 1, "dash": "dash"},
annotation_text=f"initial ${initial_capital:.0f}",
annotation_position="bottom right",
annotation_font_color=COLOR_TEXT_MUTED,
)
fig.update_layout(
title=None,
paper_bgcolor=COLOR_SURFACE,
plot_bgcolor=COLOR_SURFACE,
font={"color": COLOR_TEXT, "family": "Inter"},
xaxis={"gridcolor": COLOR_SURFACE_2, "title": None},
yaxis={"gridcolor": COLOR_SURFACE_2, "title": "Equity ($)"},
margin={"l": 60, "r": 20, "t": 10, "b": 40},
height=320,
showlegend=False,
)
return fig
@ui.page("/paper")
def paper() -> None:
_apply_theme()
_build_header(active="/paper")
options = _paper_runs_options()
if not options:
ui.label("Nessuna paper-trading run nel database.").classes("text-h5")
return
state: dict[str, Any] = {"run_id": next(iter(options))}
with ui.row().classes("w-full items-center gap-4 q-mb-md"):
select = ui.select(options=options, value=state["run_id"], label="Paper run").classes(
"flex-grow"
)
status_badge = ui.badge("…", color="primary").classes("text-body1 q-pa-sm")
ui.button("🔄 Refresh", on_click=lambda: refresh()).props("outline color=primary")
with ui.row().classes("w-full gap-4"):
with ui.card().classes("flex-grow metric-card accent-cyan"):
ui.label("Equity").classes("text-caption")
equity_lbl = ui.label("—").classes("text-h4")
with ui.card().classes("flex-grow metric-card accent-purple"):
ui.label("P/L cumulato").classes("text-caption")
pnl_lbl = ui.label("—").classes("text-h4")
with ui.card().classes("flex-grow metric-card accent-amber"):
ui.label("Trades chiusi").classes("text-caption")
trades_lbl = ui.label("—").classes("text-h4")
with ui.card().classes("flex-grow metric-card accent-green"):
ui.label("Open / Tick").classes("text-caption")
ticks_lbl = ui.label("—").classes("text-h4")
with ui.row().classes("w-full gap-4 q-mt-md"):
started_lbl = ui.label("Started: —")
last_tick_lbl = ui.label("Last tick: —")
cash_lbl = ui.label("Cash: —")
ui.separator()
ui.label("Equity curve").classes("text-subtitle1 q-mt-md")
equity_plot = ui.plotly(_paper_equity_figure(None, 0.0)).classes("w-full")
ui.separator()
ui.label("Open positions").classes("text-subtitle1 q-mt-md")
positions_table = ui.table(
columns=[
{"name": "symbol", "label": "symbol", "field": "symbol"},
{"name": "side", "label": "side", "field": "side"},
{"name": "qty", "label": "qty", "field": "qty"},
{"name": "entry_price", "label": "entry", "field": "entry_price"},
{"name": "entry_ts", "label": "entry ts", "field": "entry_ts"},
],
rows=[],
row_key="symbol",
).classes("w-full")
ui.separator()
ui.label("Ultimi 30 tick").classes("text-subtitle1 q-mt-md")
ticks_table = ui.table(
columns=[
{"name": "ts", "label": "ts", "field": "ts"},
{"name": "symbol", "label": "symbol", "field": "symbol"},
{"name": "bar_ts", "label": "bar", "field": "bar_ts"},
{"name": "close_price", "label": "close", "field": "close_price"},
{"name": "signal", "label": "signal", "field": "signal"},
{"name": "action_taken", "label": "action", "field": "action_taken"},
],
rows=[],
row_key="ts",
).classes("w-full")
ui.separator()
ui.label("Trades chiusi (ultimi 50)").classes("text-subtitle1 q-mt-md")
trades_table = ui.table(
columns=[
{"name": "exit_ts", "label": "exit ts", "field": "exit_ts"},
{"name": "symbol", "label": "symbol", "field": "symbol"},
{"name": "side", "label": "side", "field": "side"},
{"name": "qty", "label": "qty", "field": "qty"},
{"name": "entry_price", "label": "entry", "field": "entry_price"},
{"name": "exit_price", "label": "exit", "field": "exit_price"},
{"name": "pnl", "label": "pnl", "field": "pnl"},
{"name": "fees", "label": "fees", "field": "fees"},
],
rows=[],
row_key="exit_ts",
).classes("w-full")
def refresh() -> None:
run_id = select.value
if not run_id:
return
try:
summary = paper_run_summary(DB_PATH, run_id)
eq = paper_equity_df(DB_PATH, run_id)
positions = paper_positions_df(DB_PATH, run_id)
ticks = paper_ticks_df(DB_PATH, run_id, limit=30)
trades = paper_trades_df(DB_PATH, run_id, limit=50)
except Exception as e: # noqa: BLE001
ui.notify(f"Errore: {e}", type="negative")
return
text, color = _STATUS_BADGE.get(summary["status"], (summary["status"], "primary"))
status_badge.text = text
status_badge.props(f"color={color}")
equity_lbl.text = f"${summary['current_equity']:.2f}"
pnl_lbl.text = f"{summary['pnl_pct']:+.2f}%"
trades_lbl.text = str(summary["n_trades"])
ticks_lbl.text = f"{summary['n_open_positions']} / {summary['n_ticks']}"
started_lbl.text = f"Started: {summary['started_at']}"
last_tick_lbl.text = f"Last tick: {summary['last_tick_ts'] or '—'}"
cash_lbl.text = (
f"Cash: ${summary['current_cash']:.2f} | "
f"Pos value: ${summary['current_positions_value']:.2f}"
)
equity_plot.update_figure(_paper_equity_figure(eq, summary["initial_capital"]))
positions_table.rows = (
[
{col: (round(v, 6) if isinstance(v, float) else v) for col, v in row.items()}
for _, row in positions.iterrows()
]
if not positions.empty
else []
)
positions_table.update()
ticks_table.rows = (
[
{col: (round(v, 6) if isinstance(v, float) else v) for col, v in row.items()}
for _, row in ticks.iterrows()
]
if not ticks.empty
else []
)
ticks_table.update()
trades_table.rows = (
[
{col: (round(v, 6) if isinstance(v, float) else v) for col, v in row.items()}
for _, row in trades.iterrows()
]
if not trades.empty
else []
)
trades_table.update()
def on_select_change() -> None:
state["run_id"] = select.value
refresh()
select.on_value_change(on_select_change)
ui.timer(REFRESH_INTERVAL_S, refresh)
refresh()
def main() -> None:
app.on_startup(lambda: print(f"DB: {Path(DB_PATH).resolve()}"))
ui.run(
host="0.0.0.0",
port=int(os.environ.get("SWARM_DASHBOARD_PORT", "8080")),
title="Multi-Swarm Dashboard",
reload=False,
show=False,
dark=True,
)
if __name__ in {"__main__", "__mp_main__"}:
main()