"""Panel application for single-day SPBT result analysis.""" from __future__ import annotations from pathlib import Path import sys from typing import Any import pandas as pd import panel as pn APP_DIR = Path(__file__).resolve().parent REPO_ROOT = APP_DIR.parent if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) from scripts import spbt_day pn.extension("tabulator", "plotly") PAIR_THEO_RET_SORT_COLUMNS = ["total_pnl", "pair_name"] PAIR_THEO_RET_DISPLAY_DROP_COLUMNS = ["total_pnl"] APP_TITLE = "SPBT Day Analysis" APP_ACCENT_COLOR = "#226c67" APP_HEADER_COLOR = "#184c47" class SpbtDayPanelApp: """Stateful Panel UI for single-day SPBT analysis.""" def __init__(self, repo_root: Path | None = None) -> None: self.repo_root = (repo_root or spbt_day.find_repo_root(REPO_ROOT)).resolve() self.selector_pair_rankings = pd.DataFrame() self.trading_instructions = pd.DataFrame() self.pair_theo_ret = pd.DataFrame() self.selected_pair_theo_executions = pd.DataFrame() self.selected_pair_name: str | None = None self.min_pctg_change = 0.0 self.directory_input = pn.widgets.TextInput( label="Directory", value=str(self.repo_root / "data"), ) self.show_all_files = pn.widgets.Checkbox(label="Show all files", value=False) self.file_select = pn.widgets.Select( label="SQLite result file", options={}, width=360, ) self.min_pctg_change_input = pn.widgets.FloatInput( label="Mininal TARGET change (%)", value=0.0, step=1.0, width=220, ) self.calculate_button = pn.widgets.Button( label="Calculate", color="primary", width=110, ) self.status = pn.pane.Markdown("") self.pair_theo_ret_table = spbt_day.create_pair_theo_ret_analyze_grid( pd.DataFrame(), height=420, ) self.total_pnl_histogram = pn.pane.Plotly( None, height=360, sizing_mode="stretch_width", ) self.selected_pair_message = pn.pane.Markdown( "Click Analyze in the Pair TheoRet grid to load individual-pair details." ) self.selected_pair_executions_table = spbt_day.create_selected_pair_executions_grid( height=320, ) self.selected_pair_market_plot = pn.pane.Plotly( None, height=520, sizing_mode="stretch_width", ) self.calculate_button.on_click(self.calculate) self.directory_input.param.watch(self.refresh_files, "value") self.show_all_files.param.watch(self.refresh_files, "value") self.pair_theo_ret_table.on_click( self.analyze_pair_click, column=spbt_day.ANALYZE_BUTTON_COLUMN, ) self.refresh_files() def set_status(self, message: str, *, error: bool = False) -> None: """Update visible status text.""" prefix = "**Error:** " if error else "" self.status.object = f"{prefix}{message}" if message else "" def selected_database_path(self) -> Path: """Return the selected result database path.""" if not self.file_select.value: raise ValueError("Select a SQLite result file before calculating.") db_path = Path(str(self.file_select.value)).resolve() if not db_path.exists(): raise FileNotFoundError(f"Selected database does not exist: {db_path}") if not db_path.is_file(): raise ValueError(f"Selected database path is not a file: {db_path}") return db_path def refresh_files(self, *_events: Any) -> bool: """Refresh selectable SQLite files from the configured directory.""" try: directory = spbt_day.normalize_directory( self.directory_input.value, self.repo_root, ) candidates = spbt_day.list_candidate_files( directory, show_all=self.show_all_files.value, ) except Exception as exc: self.file_select.options = {} self.file_select.value = None self.set_status(str(exc), error=True) return False options = {path.name: str(path) for path in candidates} previous_value = self.file_select.value self.file_select.options = options if previous_value in options.values(): self.file_select.value = previous_value elif options: self.file_select.value = next(iter(options.values())) else: self.file_select.value = None if options: self.set_status(f"Found {len(options):,} file(s) in {directory}.") else: self.set_status(f"No selectable files found in {directory}.") return True def calculate(self, *_events: Any) -> None: """Load selected data and calculate all-pair TheoRet.""" self.calculate_button.loading = True try: if not self.refresh_files(): return db_path = self.selected_database_path() self.min_pctg_change = float(self.min_pctg_change_input.value) conn = spbt_day.connect_sqlite_read_only(db_path) try: self.selector_pair_rankings = spbt_day.load_selector_pair_rankings(conn) self.trading_instructions = spbt_day.load_trading_instructions(conn) finally: conn.close() self.pair_theo_ret = ( spbt_day.add_total_pnl( spbt_day.calculate_ranked_pairs_theo_ret( self.selector_pair_rankings, self.trading_instructions, min_pctg_change=self.min_pctg_change, ) ) .sort_values( PAIR_THEO_RET_SORT_COLUMNS, ascending=[True, True], kind="mergesort", ) .drop(columns=PAIR_THEO_RET_DISPLAY_DROP_COLUMNS) .reset_index(drop=True) ) self.pair_theo_ret_table.value = spbt_day.format_pair_theo_ret_for_analyze_grid( self.pair_theo_ret ) self.total_pnl_histogram.object = spbt_day.create_total_pnl_histogram( self.pair_theo_ret ) self.clear_selected_pair_analysis() self.set_status( f"Calculated {len(self.pair_theo_ret):,} pair row(s) from {db_path.name}." ) except Exception as exc: self.set_status(str(exc), error=True) finally: self.calculate_button.loading = False def clear_selected_pair_analysis(self) -> None: """Clear individual-pair outputs until a row Analyze button is clicked.""" self.selected_pair_name = None self.selected_pair_theo_executions = pd.DataFrame() self.selected_pair_message.object = ( "Click Analyze in the Pair TheoRet grid to load individual-pair details." ) self.selected_pair_executions_table.value = pd.DataFrame( columns=spbt_day.SELECTED_PAIR_EXECUTION_DISPLAY_COLUMNS ) self.selected_pair_market_plot.object = None def analyze_pair_click(self, event: Any) -> None: """Run selected-pair analysis from a Pair TheoRet Analyze button click.""" self.update_selected_pair( spbt_day.pair_name_from_analyze_event(self.pair_theo_ret_table, event) ) def analyze_pair_row(self, row: int) -> None: """Run selected-pair analysis for a Pair TheoRet table row.""" event = type("AnalyzeEvent", (), {"row": row})() self.analyze_pair_click(event) def update_selected_pair(self, pair_name: str) -> None: """Calculate selected-pair executions and market plot.""" if self.trading_instructions.empty: self.clear_selected_pair_analysis() return self.selected_pair_name = pair_name self.selected_pair_message.object = ( f"Selected pair: **{spbt_day.format_pair_name_for_display(pair_name)}**" ) self.selected_pair_theo_executions = spbt_day.calculate_pair_theo_executions( pair_name, self.trading_instructions, min_pctg_change=self.min_pctg_change, ) self.selected_pair_executions_table.value = ( self.selected_pair_theo_executions.reindex( columns=spbt_day.SELECTED_PAIR_EXECUTION_DISPLAY_COLUMNS ) ) try: trading_day_start_ns = spbt_day.infer_trading_day_start_ns( self.trading_instructions ) conn = spbt_day.connect_sqlite_read_only(self.selected_database_path()) try: selected_pair_market_data = spbt_day.load_pair_market_data( conn, pair_name, trading_day_start_ns=trading_day_start_ns, ) finally: conn.close() self.selected_pair_market_plot.object = spbt_day.create_pair_trades_market_plot( pair_name, selected_pair_market_data, self.selected_pair_theo_executions, ) except Exception as exc: self.selected_pair_market_plot.object = None self.set_status(str(exc), error=True) @property def view(self) -> pn.template.FastListTemplate: """Return the app layout.""" controls = pn.Column( "## Inputs", self.directory_input, self.show_all_files, self.file_select, self.min_pctg_change_input, self.calculate_button, self.status, width=400, ) main = pn.Column( "## Pair TheoRet", self.pair_theo_ret_table, self.total_pnl_histogram, "## Individual Pair", self.selected_pair_message, "### Theoretical Executions", self.selected_pair_executions_table, "### Trades on Market Data", self.selected_pair_market_plot, ) return pn.template.FastListTemplate( title=APP_TITLE, sidebar=[controls], main=[main], sidebar_width=430, accent_base_color=APP_ACCENT_COLOR, header_background=APP_HEADER_COLOR, main_layout=None, ) app_controller = SpbtDayPanelApp() app = app_controller.view app.servable(title=APP_TITLE)