Oleg Sheynin
304 lines
11 KiB
Python
304 lines
11 KiB
Python
"""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"
|
|
APP_SIDEBAR_WIDTH = 215
|
|
APP_SIDEBAR_CONTROL_WIDTH = 200
|
|
|
|
|
|
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"),
|
|
sizing_mode="stretch_width",
|
|
width=None,
|
|
)
|
|
self.show_all_files = pn.widgets.Checkbox(label="Show all files", value=False)
|
|
self.file_select = pn.widgets.Select(
|
|
label="SQLite result file",
|
|
options={},
|
|
sizing_mode="stretch_width",
|
|
width=None,
|
|
)
|
|
self.min_pctg_change_input = pn.widgets.FloatInput(
|
|
label="Mininal TARGET change (%)",
|
|
value=0.0,
|
|
step=1.0,
|
|
sizing_mode="stretch_width",
|
|
width=None,
|
|
)
|
|
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=APP_SIDEBAR_CONTROL_WIDTH,
|
|
)
|
|
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=APP_SIDEBAR_WIDTH,
|
|
accent_base_color=APP_ACCENT_COLOR,
|
|
header_background=APP_HEADER_COLOR,
|
|
main_layout=None,
|
|
theme=pn.template.DarkTheme,
|
|
)
|
|
|
|
|
|
app_controller = SpbtDayPanelApp()
|
|
app = app_controller.view
|
|
app.servable(title=APP_TITLE)
|