research/stat_pairs_backtest
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Release v1.0.2

Browse Source
This commit is contained in:
Oleg Sheynin
2026-07-29 00:39:10 +00:00
parent 49c91e5d85
commit a3e5acd765
8 changed files with 764 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+14
View File
@@ -6,6 +6,20 @@ All notable changes to this project are documented in this file.
No unreleased changes yet.
## 2026-07-29 v1.0.2
- Added a Panel application for single-day SPBT result analysis with result-file
selection, minimum TARGET-change input, pair TheoRet table, pair selector,
selected-pair execution table, and market/trade chart.
- Added a launcher script for the Panel application.
- Changed notebook and Panel pair analysis to use per-row Analyze actions from
the Pair TheoRet grid, deferring selected-pair calculations until clicked.
- Adjusted Panel sizing so key controls use compact widths and Pair TheoRet uses
content width with vertical scrolling instead of full-width paginated layout.
- Added a FastListTemplate shell to the Panel application for sidebar controls
and configurable app color accents.
- Made Plotly chart panes use all available horizontal space.
## 2026-07-28 v1.0.1
- Added the `spbt_day` notebook for interactive single-day backtest result
notebooks/spbt_day.ipynb
+73 -76
View File
@@ -30,6 +30,7 @@
"from IPython.display import display\n",
"import ipywidgets as widgets\n",
"import pandas as pd\n",
"import panel as pn\n",
"\n",
"START_DIR = Path.cwd().resolve()\n",
"for candidate in (START_DIR, *START_DIR.parents):\n",
@@ -41,21 +42,27 @@
"import scripts.spbt_day as spbt_day\n",
"\n",
"spbt_day = importlib.reload(spbt_day)\n",
"pn.extension(\"tabulator\", \"plotly\")\n",
"\n",
"ANALYZE_BUTTON_COLUMN = spbt_day.ANALYZE_BUTTON_COLUMN\n",
"SELECTED_PAIR_EXECUTION_DISPLAY_COLUMNS = spbt_day.SELECTED_PAIR_EXECUTION_DISPLAY_COLUMNS\n",
"add_total_pnl = spbt_day.add_total_pnl\n",
"calculate_pair_theo_executions = spbt_day.calculate_pair_theo_executions\n",
"calculate_ranked_pairs_theo_ret = spbt_day.calculate_ranked_pairs_theo_ret\n",
"create_database_file_selector = spbt_day.create_database_file_selector\n",
"create_pair_name_dropdown = spbt_day.create_pair_name_dropdown\n",
"create_pair_theo_ret_analyze_grid = spbt_day.create_pair_theo_ret_analyze_grid\n",
"create_pair_trades_market_plot = spbt_day.create_pair_trades_market_plot\n",
"create_selected_pair_executions_grid = spbt_day.create_selected_pair_executions_grid\n",
"create_total_pnl_histogram = spbt_day.create_total_pnl_histogram\n",
"find_repo_root = spbt_day.find_repo_root\n",
"format_pair_name_for_display = spbt_day.format_pair_name_for_display\n",
"format_pair_names_for_display = spbt_day.format_pair_names_for_display\n",
"format_pair_theo_ret_for_analyze_grid = spbt_day.format_pair_theo_ret_for_analyze_grid\n",
"infer_trading_day_start_ns = spbt_day.infer_trading_day_start_ns\n",
"load_selector_pair_rankings = spbt_day.load_selector_pair_rankings\n",
"load_pair_market_data = spbt_day.load_pair_market_data\n",
"load_trading_instructions = spbt_day.load_trading_instructions\n",
"pair_name_from_analyze_event = spbt_day.pair_name_from_analyze_event\n",
"show_interactive_dataframe = spbt_day.show_interactive_dataframe\n",
"\n",
"REPO_ROOT = find_repo_root()\n",
@@ -190,12 +197,12 @@
" kind=\"mergesort\",\n",
").drop(columns=\"total_pnl\").reset_index(drop=True)\n",
"\n",
"pair_theo_ret_display = format_pair_names_for_display(pair_theo_ret)\n",
"\n",
"show_interactive_dataframe(\n",
"pair_theo_ret_display = format_pair_theo_ret_for_analyze_grid(pair_theo_ret)\n",
"pair_theo_ret_grid = create_pair_theo_ret_analyze_grid(\n",
" pair_theo_ret_display,\n",
" table_id=\"pair-theo-ret-grid\",\n",
")"
" height=520,\n",
")\n",
"display(pair_theo_ret_grid)"
]
},
{
@@ -216,8 +223,13 @@
"outputs": [],
"source": [
"total_pnl_histogram = create_total_pnl_histogram(pair_theo_ret)\n",
"total_pnl_histogram_pane = pn.pane.Plotly(\n",
" total_pnl_histogram,\n",
" height=360,\n",
" sizing_mode=\"stretch_width\",\n",
")\n",
"\n",
"total_pnl_histogram"
"display(total_pnl_histogram_pane)"
]
},
{
@@ -227,92 +239,62 @@
"source": [
"## Individual Pair Analysis\n",
"\n",
"Choose one pair for detailed follow-up analysis. Pair names are sorted alphabetically."
"Click the Analyze button in the Pair TheoRet grid to load detailed follow-up analysis for that row. The selected-pair execution table and market/trade chart are not calculated until an Analyze button is clicked."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "choose-individual-pair",
"id": "individual-pair-analysis",
"metadata": {},
"outputs": [],
"source": [
"pair_name_dropdown = create_pair_name_dropdown(selector_pair_rankings)\n",
"display(pair_name_dropdown)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "selected-individual-pair",
"metadata": {},
"outputs": [],
"source": [
"selected_pair_name = pair_name_dropdown.value\n",
"format_pair_name_for_display(selected_pair_name)"
]
},
{
"cell_type": "markdown",
"id": "selected-pair-theo-executions-context",
"metadata": {},
"source": [
"### Selected Pair Theoretical Executions\n",
"selected_pair_name = None\n",
"selected_pair_theo_executions = pd.DataFrame()\n",
"selected_pair_theo_executions_display = pd.DataFrame(\n",
" columns=SELECTED_PAIR_EXECUTION_DISPLAY_COLUMNS\n",
")\n",
"selected_pair_market_data = pd.DataFrame()\n",
"selected_pair_market_trades_plot = None\n",
"\n",
"selected_pair_message = pn.pane.Markdown(\n",
" \"Click Analyze in the Pair TheoRet grid to load individual-pair details.\"\n",
")\n",
"selected_pair_theo_executions_grid = create_selected_pair_executions_grid(\n",
" selected_pair_theo_executions_display,\n",
" height=360,\n",
")\n",
"selected_pair_market_trades_plot_pane = pn.pane.Plotly(\n",
" None,\n",
" height=520,\n",
" sizing_mode=\"stretch_width\",\n",
")\n",
"\n",
"\n",
"def analyze_pair_click(event):\n",
" global selected_pair_name\n",
" global selected_pair_theo_executions\n",
" global selected_pair_theo_executions_display\n",
" global selected_pair_market_data\n",
" global selected_pair_market_trades_plot\n",
"\n",
" try:\n",
" selected_pair_name = pair_name_from_analyze_event(pair_theo_ret_grid, event)\n",
" selected_pair_message.object = (\n",
" f\"Selected pair: **{format_pair_name_for_display(selected_pair_name)}**\"\n",
" )\n",
"\n",
"Create the theoretical asset-level executions used by the PnL calculation for the selected pair. `TARGET` rows trade the position difference from the current theoretical position to the new target position, where target size is `10000 * strength / reference_price`; `CLOSE` rows flatten the current theoretical position. Positive size is `BUY`; negative size is `SELL`; USD value is signed as the opposite cash movement."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "selected-pair-theo-executions",
"metadata": {},
"outputs": [],
"source": [
" selected_pair_theo_executions = calculate_pair_theo_executions(\n",
" selected_pair_name,\n",
" trading_instructions,\n",
" min_pctg_change=MIN_TARGET_STRENGTH_CHANGE_PCTG,\n",
" )\n",
"\n",
"selected_pair_theo_execution_columns = [\n",
" \"time\",\n",
" \"asset\",\n",
" \"action\",\n",
" \"side\",\n",
" \"strength\",\n",
" \"size\",\n",
" \"price\",\n",
" \"usd_value\",\n",
"]\n",
" selected_pair_theo_executions_display = selected_pair_theo_executions.reindex(\n",
" columns=selected_pair_theo_execution_columns\n",
" columns=SELECTED_PAIR_EXECUTION_DISPLAY_COLUMNS\n",
" )\n",
"show_interactive_dataframe(\n",
" selected_pair_theo_executions_display,\n",
" table_id=\"selected-pair-theo-executions-grid\",\n",
")"
]
},
{
"cell_type": "markdown",
"id": "selected-pair-market-trades-context",
"metadata": {},
"source": [
"### Selected Pair Trades on Market Data\n",
" selected_pair_theo_executions_grid.value = selected_pair_theo_executions_display\n",
"\n",
"Load full available 1-minute market data for the selected pair's instruments from `ohlcv_1min`, starting at midnight UTC of the trading day inferred from `trading_instructions`. Close prices are shown as relative prices from each instrument's close at that midnight. Theoretical executions are overlaid at their execution `reference_price`, normalized by the same midnight close. Execution markers use execution timestamps directly and do not require a matching OHLCV row."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "selected-pair-market-trades-plot",
"metadata": {},
"outputs": [],
"source": [
" trading_day_start_ns = infer_trading_day_start_ns(trading_instructions)\n",
"\n",
" conn = connect_selected_database()\n",
" try:\n",
" selected_pair_market_data = load_pair_market_data(\n",
@@ -328,8 +310,23 @@
" selected_pair_market_data,\n",
" selected_pair_theo_executions,\n",
" )\n",
" selected_pair_market_trades_plot_pane.object = selected_pair_market_trades_plot\n",
" except Exception as exc:\n",
" selected_pair_message.object = f\"**Error:** {exc}\"\n",
" selected_pair_market_trades_plot_pane.object = None\n",
"\n",
"selected_pair_market_trades_plot"
"\n",
"pair_theo_ret_grid.on_click(analyze_pair_click, column=ANALYZE_BUTTON_COLUMN)\n",
"\n",
"display(\n",
" pn.Column(\n",
" selected_pair_message,\n",
" \"### Theoretical Executions\",\n",
" selected_pair_theo_executions_grid,\n",
" \"### Trades on Market Data\",\n",
" selected_pair_market_trades_plot_pane,\n",
" )\n",
")"
]
}
],
panel/spbt_day_panel.py
+295
View File
@@ -0,0 +1,295 @@
"""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.refresh_button = pn.widgets.Button(label="Refresh")
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.refresh_button.on_click(self.refresh_files)
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) -> None:
"""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
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}.")
def calculate(self, *_events: Any) -> None:
"""Load selected data and calculate all-pair TheoRet."""
self.calculate_button.loading = True
try:
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",
pn.Row(self.directory_input, self.refresh_button),
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)
requirements.txt
+1
View File
@@ -5,6 +5,7 @@ itables>=2.2,<3
jupyter>=1.1,<2
nbformat>=5.10,<6
pandas>=2.2,<3
panel>=1.5,<2
plotly>=5.24,<7
# Verification
scripts/run_spbt_day_panel.sh
Executable
+7
View File
@@ -0,0 +1,7 @@
#!/usr/bin/env bash
set -euo pipefail
repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$repo_root"
panel serve panel/spbt_day_panel.py --show "$@"
scripts/spbt_day.py
+85
View File
@@ -20,6 +20,22 @@ OHLCV_1MIN_COLUMNS = ("tstamp", "tstamp_ns", "exch_acct", "instrument_id", "clos
INITIAL_THEO_CAPITAL_USD = 10_000.0
SQLITE_EXTENSIONS = {".db", ".sqlite", ".sqlite3"}
PAIR_NAME_DISPLAY_SUFFIX = ":USD"
PAIR_NAME_VALUE_COLUMN = "_pair_name_value"
ANALYZE_BUTTON_COLUMN = "analyze"
ANALYZE_BUTTON_HTML = (
'<button type="button" title="Analyze pair" aria-label="Analyze pair" '
'style="padding: 1px 6px; line-height: 1.2;">&#9654;</button>'
)
SELECTED_PAIR_EXECUTION_DISPLAY_COLUMNS = [
"time",
"asset",
"action",
"side",
"strength",
"size",
"price",
"usd_value",
]
INTERACTIVE_TABLE_CSS = """
table.dataTable,
table.dataTable th,
@@ -1081,6 +1097,75 @@ def format_pair_names_for_display(
return formatted
def format_pair_theo_ret_for_analyze_grid(pair_theo_ret: pd.DataFrame) -> pd.DataFrame:
"""Format pair TheoRet rows for an Analyze-button grid.
The visible pair name is display-cleaned, while the original pair name is
retained in a hidden column for callbacks that need the calculation key.
"""
if "pair_name" not in pair_theo_ret.columns:
raise ValueError("pair TheoRet dataframe missing column: pair_name")
formatted = format_pair_names_for_display(pair_theo_ret)
formatted[PAIR_NAME_VALUE_COLUMN] = pair_theo_ret["pair_name"].to_numpy()
return formatted
def pair_name_from_analyze_event(table: Any, event: Any) -> str:
"""Resolve the full pair name from a Panel Tabulator click event."""
if PAIR_NAME_VALUE_COLUMN not in table.value.columns:
raise ValueError(f"Analyze table missing column: {PAIR_NAME_VALUE_COLUMN}")
pair_name = table.value[PAIR_NAME_VALUE_COLUMN].iloc[event.row]
if pd.isna(pair_name) or not str(pair_name):
raise ValueError(f"Analyze row {event.row} does not contain a pair name")
return str(pair_name)
def create_pair_theo_ret_analyze_grid(
pair_theo_ret_display: pd.DataFrame,
*,
height: int = 420,
) -> Any:
"""Create a sortable Panel grid with a compact Analyze button column."""
import panel as pn
return pn.widgets.Tabulator(
pair_theo_ret_display,
buttons={ANALYZE_BUTTON_COLUMN: ANALYZE_BUTTON_HTML},
hidden_columns=[PAIR_NAME_VALUE_COLUMN],
show_index=False,
pagination=None,
layout="fit_data_table",
height=height,
sizing_mode="fixed",
selectable=False,
)
def create_selected_pair_executions_grid(
dataframe: pd.DataFrame | None = None,
*,
height: int = 320,
) -> Any:
"""Create a sortable Panel grid for selected-pair theoretical executions."""
import panel as pn
source = (
dataframe
if dataframe is not None
else pd.DataFrame(columns=SELECTED_PAIR_EXECUTION_DISPLAY_COLUMNS)
)
return pn.widgets.Tabulator(
source.reindex(columns=SELECTED_PAIR_EXECUTION_DISPLAY_COLUMNS),
show_index=False,
pagination=None,
layout="fit_data_table",
height=height,
sizing_mode="fixed",
selectable=False,
)
def add_total_pnl(pair_theo_ret: pd.DataFrame) -> pd.DataFrame:
"""Return a copy of pair TheoRet rows with total realized plus unrealized PnL."""
required_columns = {"realized_pnl", "unrealized_pnl"}
tests/test_spbt_day.py
+51
View File
@@ -5,17 +5,22 @@ import pandas as pd
import pytest
from scripts.spbt_day import (
ANALYZE_BUTTON_COLUMN,
PAIR_NAME_VALUE_COLUMN,
add_total_pnl,
calculate_pair_theo_executions,
calculate_pair_theo_ret,
calculate_ranked_pairs_theo_ret,
create_pair_name_dropdown,
create_pair_theo_ret_analyze_grid,
create_pair_trades_market_plot,
create_selected_pair_executions_grid,
connect_sqlite_read_only,
create_total_pnl_histogram,
find_repo_root,
format_pair_name_for_display,
format_pair_names_for_display,
format_pair_theo_ret_for_analyze_grid,
infer_trading_day_start_ns,
list_candidate_files,
load_pair_market_data,
@@ -985,6 +990,52 @@ def test_format_pair_names_for_display_removes_usd_suffix_without_mutating_sourc
]
def test_format_pair_theo_ret_for_analyze_grid_preserves_full_pair_name():
pair_theo_ret = pd.DataFrame(
{
"pair_name": ["AAA:USD-BBB:USD"],
"mr_ranking": [1],
"realized_pnl": [1.0],
"unrealized_pnl": [0.0],
}
)
formatted = format_pair_theo_ret_for_analyze_grid(pair_theo_ret)
assert formatted["pair_name"].tolist() == ["AAA-BBB"]
assert formatted[PAIR_NAME_VALUE_COLUMN].tolist() == ["AAA:USD-BBB:USD"]
assert pair_theo_ret["pair_name"].tolist() == ["AAA:USD-BBB:USD"]
def test_create_panel_grids_use_analyze_button_and_hidden_pair_column():
pair_grid = create_pair_theo_ret_analyze_grid(
pd.DataFrame(
{
"pair_name": ["AAA-BBB"],
PAIR_NAME_VALUE_COLUMN: ["AAA:USD-BBB:USD"],
}
)
)
executions_grid = create_selected_pair_executions_grid()
assert pair_grid.buttons.keys() == {ANALYZE_BUTTON_COLUMN}
assert pair_grid.hidden_columns == [PAIR_NAME_VALUE_COLUMN]
assert pair_grid.disabled is False
assert pair_grid.pagination is None
assert pair_grid.layout == "fit_data_table"
assert pair_grid.sizing_mode == "fixed"
assert executions_grid.value.columns.tolist() == [
"time",
"asset",
"action",
"side",
"strength",
"size",
"price",
"usd_value",
]
def test_show_interactive_dataframe_uses_sortable_grid_defaults(monkeypatch):
calls = []
tests/test_spbt_day_panel.py
+218
View File
@@ -0,0 +1,218 @@
import importlib.util
import sqlite3
from pathlib import Path
import pandas as pd
def load_panel_app_module():
module_path = Path("panel/spbt_day_panel.py").resolve()
spec = importlib.util.spec_from_file_location("spbt_day_panel_app", module_path)
module = importlib.util.module_from_spec(spec)
assert spec.loader is not None
spec.loader.exec_module(module)
return module
def create_panel_fixture_db(db_path: Path) -> None:
trading_day_start_ns = pd.Timestamp("2026-06-17T00:00:00Z").value
conn = sqlite3.connect(db_path)
try:
conn.execute(
"""
CREATE TABLE selector_pairs (
time_ns INTEGER,
tstamp TEXT,
pair_name TEXT,
instrument_a TEXT,
instrument_b TEXT,
mr_score TEXT
)
"""
)
conn.execute(
"""
CREATE TABLE trading_instructions (
time_ns INTEGER,
tstamp TEXT,
book_id TEXT,
strategy_id TEXT,
type TEXT,
data TEXT
)
"""
)
conn.execute(
"""
CREATE TABLE ohlcv_1min (
tstamp TEXT,
tstamp_ns INTEGER,
exch_acct TEXT,
exchange_id TEXT,
instrument_id TEXT,
interval_sec INTEGER,
open REAL,
high REAL,
low REAL,
close REAL,
volume REAL,
vwap REAL,
num_trades INTEGER
)
"""
)
conn.execute(
"INSERT INTO selector_pairs VALUES (?, ?, ?, ?, ?, ?)",
(
10,
"2026-06-17T00:00:00Z",
"AAA:USD-BBB:USD",
"EXCH:PAIR-AAA-USD",
"EXCH:PAIR-BBB-USD",
'{"final":"0.5"}',
),
)
conn.executemany(
"INSERT INTO trading_instructions VALUES (?, ?, ?, ?, ?, ?)",
[
(
trading_day_start_ns,
"2026-06-17T00:00:00Z",
"book",
"strategy-AAA:USD-BBB:USD",
"TARGET_POSITION",
(
'{"action":"TARGET","quote_asset":"USD","assets":'
'{"AAA":{"reference_price":"100","strength":"0.5"},'
'"BBB":{"reference_price":"50","strength":"-0.5"}}}'
),
),
(
trading_day_start_ns + 60_000_000_000,
"2026-06-17T00:01:00Z",
"book",
"strategy-AAA:USD-BBB:USD",
"CLOSE_POSITION",
(
'{"action":"CLOSE","quote_asset":"USD","assets":'
'{"AAA":{"reference_price":"110"},'
'"BBB":{"reference_price":"45"}}}'
),
),
],
)
conn.executemany(
"INSERT INTO ohlcv_1min VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
[
(
"2026-06-17T00:00:00Z",
trading_day_start_ns,
"EXCH",
"EXCH",
"PAIR-AAA-USD",
60,
100.0,
100.0,
100.0,
100.0,
1.0,
100.0,
1,
),
(
"2026-06-17T00:00:00Z",
trading_day_start_ns,
"EXCH",
"EXCH",
"PAIR-BBB-USD",
60,
50.0,
50.0,
50.0,
50.0,
1.0,
50.0,
1,
),
],
)
conn.commit()
finally:
conn.close()
def test_pair_analyze_grid_keeps_clean_labels_and_full_pair_values():
module = load_panel_app_module()
pair_theo_ret = pd.DataFrame(
{
"pair_name": ["BTC:USD-ETH:USD", "ADA:USD-BTC:USD"],
"mr_ranking": [2, 1],
"realized_pnl": [0.0, 0.0],
"unrealized_pnl": [0.0, 0.0],
}
)
formatted = module.spbt_day.format_pair_theo_ret_for_analyze_grid(pair_theo_ret)
assert formatted["pair_name"].tolist() == ["BTC-ETH", "ADA-BTC"]
assert formatted[module.spbt_day.PAIR_NAME_VALUE_COLUMN].tolist() == [
"BTC:USD-ETH:USD",
"ADA:USD-BTC:USD",
]
def test_panel_app_uses_fast_list_template(tmp_path):
module = load_panel_app_module()
app = module.SpbtDayPanelApp(repo_root=tmp_path)
view = app.view
assert isinstance(view, module.pn.template.FastListTemplate)
assert view.title == module.APP_TITLE
assert view.sidebar_width == 430
assert view.accent_base_color == module.APP_ACCENT_COLOR
assert view.header_background == module.APP_HEADER_COLOR
assert len(view.sidebar) == 1
assert len(view.main) == 1
def test_panel_app_calculates_pairs_and_selected_pair_outputs(tmp_path):
module = load_panel_app_module()
data_dir = tmp_path / "data"
data_dir.mkdir()
db_path = data_dir / "20260617.spbt_results.db"
create_panel_fixture_db(db_path)
app = module.SpbtDayPanelApp(repo_root=tmp_path)
app.directory_input.value = str(data_dir)
app.refresh_files()
app.min_pctg_change_input.value = 0.0
app.calculate()
assert app.file_select.value == str(db_path)
assert app.file_select.width == 360
assert app.min_pctg_change_input.width == 220
assert app.calculate_button.width == 110
assert app.total_pnl_histogram.sizing_mode == "stretch_width"
assert app.selected_pair_market_plot.sizing_mode == "stretch_width"
assert app.pair_theo_ret_table.pagination is None
assert app.pair_theo_ret_table.layout == "fit_data_table"
assert app.pair_theo_ret_table.value["pair_name"].tolist() == ["AAA-BBB"]
assert (
app.pair_theo_ret_table.value[module.spbt_day.PAIR_NAME_VALUE_COLUMN].tolist()
== ["AAA:USD-BBB:USD"]
)
assert app.selected_pair_name is None
assert app.selected_pair_executions_table.value.empty
assert app.selected_pair_market_plot.object is None
app.analyze_pair_row(0)
assert app.selected_pair_name == "AAA:USD-BBB:USD"
assert app.selected_pair_executions_table.value["action"].tolist() == [
"TARGET",
"TARGET",
"CLOSE",
"CLOSE",
]
assert app.selected_pair_market_plot.object is not None