diff --git a/src/strategy_pythagoras/strategy_pythagoras/frontend/nicegui_app.py b/src/strategy_pythagoras/strategy_pythagoras/frontend/nicegui_app.py new file mode 100644 index 0000000..a4a2638 --- /dev/null +++ b/src/strategy_pythagoras/strategy_pythagoras/frontend/nicegui_app.py @@ -0,0 +1,421 @@ +"""Strategy Pythagoras Dashboard — paper-trading + GA winners page: /. + +Avvio: ``uv run python -m strategy_pythagoras.frontend.nicegui_app`` +Default port 8080. Legge il paper DB (``strategy_pythagoras_paper.db``) per il +tab ``Paper`` e il GA DB (``strategy_pythagoras.db``) per i tab pythagoras-specifici +(Genomes / Patterns / Ratios / Invariance). + +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 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 strategy_pythagoras.frontend.data import ( + load_candle_pattern_usage, + load_invariance_metrics, + paper_equity_df, + paper_positions_df, + paper_run_summary, + paper_runs_df, + paper_ticks_df, + paper_trades_df, +) +from multi_swarm_core.dashboard.theme import ( + COLOR_PRIMARY, + COLOR_SURFACE, + COLOR_SURFACE_2, + COLOR_TEXT, + COLOR_TEXT_MUTED, + _STATUS_BADGE, + _apply_theme, + _build_header, +) + +PAPER_DB_PATH = os.environ.get( + "STRATEGY_PYTHAGORAS_PAPER_DB_PATH", "./state/strategy_pythagoras_paper.db" +) +GA_DB_PATH = os.environ.get( + "STRATEGY_PYTHAGORAS_DB_PATH", "./state/strategy_pythagoras.db" +) +DASHBOARD_ROOT_PATH = os.environ.get("DASHBOARD_ROOT_PATH", "/strategy_pythagoras_gui") +REFRESH_INTERVAL_S = 3.0 + + +def _paper_runs_options(only_running: bool = False) -> dict[str, str]: + try: + runs = paper_runs_df(PAPER_DB_PATH) + except Exception: + return {} + 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 + + +def _render_paper_panel() -> None: + """Rende il pannello paper-trading (equivalente alla pagina root di strategy_crypto).""" + options = _paper_runs_options() + if not options: + ui.label("Nessuna paper-trading run nel database.").classes("text-h6 text-warning") + ui.label( + "Avvia un paper run per popolare strategy_pythagoras_paper.db." + ).classes("text-caption") + 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(PAPER_DB_PATH, run_id) + eq = paper_equity_df(PAPER_DB_PATH, run_id) + positions = paper_positions_df(PAPER_DB_PATH, run_id) + ticks = paper_ticks_df(PAPER_DB_PATH, run_id, limit=30) + trades = paper_trades_df(PAPER_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() + + +@ui.page("/") +def index() -> None: + _apply_theme() + _build_header( + active="/", + brand_subtitle="Strategy Pythagoras", + nav_items=[("/", "Dashboard")], + db_label=f"⛁ {Path(GA_DB_PATH).resolve().name}", + ) + + with ui.tabs() as tabs: + t_genomes = ui.tab("Genomes") + t_patterns = ui.tab("Patterns") + t_ratios = ui.tab("Ratios") + t_invariance = ui.tab("Invariance") + t_paper = ui.tab("Paper") + + with ui.tab_panels(tabs, value=t_genomes).classes("w-full"): + with ui.tab_panel(t_genomes): + ui.label("GA Winners (post pythagoras-smoke-001)").classes("text-h6") + try: + df = load_invariance_metrics(GA_DB_PATH) + if df.empty: + ui.label( + "No winners yet. Run scripts/run_pythagoras_smoke.py first." + ).classes("text-warning") + else: + ui.table( + rows=df.to_dict("records"), + columns=[ + {"name": "genome_id", "label": "Genome ID", "field": "genome_id"}, + { + "name": "cognitive_style", + "label": "Style", + "field": "cognitive_style", + }, + {"name": "fitness", "label": "Fitness", "field": "fitness"}, + {"name": "sharpe_btc", "label": "Sharpe BTC", "field": "sharpe_btc"}, + {"name": "sharpe_eth", "label": "Sharpe ETH", "field": "sharpe_eth"}, + { + "name": "invariance_score", + "label": "Invariance", + "field": "invariance_score", + }, + ], + pagination=10, + ).classes("w-full") + except Exception as e: # noqa: BLE001 + ui.label(f"DB not ready: {e}").classes("text-warning") + + with ui.tab_panel(t_patterns): + ui.label( + "Candle pattern sequences emerged (per cognitive style)" + ).classes("text-h6") + try: + df_pat = load_candle_pattern_usage(GA_DB_PATH) + if df_pat.empty: + ui.label( + "No patterns yet. Run scripts/run_pythagoras_smoke.py first." + ).classes("text-warning") + else: + grouped = ( + df_pat.groupby(["cognitive_style", "sequence"]) + .size() + .reset_index(name="count") + ) + grouped = grouped.sort_values("count", ascending=False).head(50) + ui.table( + rows=grouped.to_dict("records"), + columns=[ + { + "name": "cognitive_style", + "label": "Style", + "field": "cognitive_style", + }, + { + "name": "sequence", + "label": "Sequence (U/D/0)", + "field": "sequence", + }, + {"name": "count", "label": "Count", "field": "count"}, + ], + pagination=15, + ).classes("w-full") + except Exception as e: # noqa: BLE001 + ui.label(f"DB not ready: {e}").classes("text-warning") + + with ui.tab_panel(t_ratios): + ui.label( + "Pythagorean ratio literals — distance from universal constants" + ).classes("text-h6") + try: + df = load_invariance_metrics(GA_DB_PATH) + if df.empty: + ui.label("No winners yet.").classes("text-warning") + else: + ui.label(f"Total winners: {len(df)}").classes("text-body2") + ui.label( + "(Ratio literal histogram available after GA run produces " + "pythagorean_ratio entries.)" + ).classes("text-caption") + except Exception as e: # noqa: BLE001 + ui.label(f"DB not ready: {e}").classes("text-warning") + + with ui.tab_panel(t_invariance): + ui.label( + "Cross-asset invariance: Sharpe BTC vs Sharpe ETH" + ).classes("text-h6") + try: + df = load_invariance_metrics(GA_DB_PATH) + if df.empty: + ui.label("No winners yet.").classes("text-warning") + else: + import plotly.express as px # type: ignore[import-untyped] + + fig = px.scatter( + df, + x="sharpe_btc", + y="sharpe_eth", + color="invariance_score", + hover_data=["genome_id", "cognitive_style", "fitness"], + title="Sharpe BTC vs Sharpe ETH (color = invariance_score)", + ) + ui.plotly(fig).classes("w-full") + except Exception as e: # noqa: BLE001 + ui.label(f"DB not ready: {e}").classes("text-warning") + + with ui.tab_panel(t_paper): + _render_paper_panel() + + +def main() -> None: + app.on_startup( + lambda: print( + f"GA DB: {Path(GA_DB_PATH).resolve()} | " + f"Paper DB: {Path(PAPER_DB_PATH).resolve()} | " + f"root_path: {DASHBOARD_ROOT_PATH or '/'}" + ) + ) + ui.run( + host="0.0.0.0", + port=int(os.environ.get("SWARM_DASHBOARD_PORT", "8080")), + title="Strategy Pythagoras Dashboard", + reload=False, + show=False, + dark=True, + root_path=DASHBOARD_ROOT_PATH, + ) + + +if __name__ in {"__main__", "__mp_main__"}: + main()