{
"cells": [
{
"cell_type": "markdown",
"id": "single-day-title",
"metadata": {},
"source": [
"# Single-Day Backtest Result Analysis\n",
"\n",
"This notebook analyzes the result of one single-day backtest stored in a SQLite database. Development is staged; Step 1 only selects the database file that later sections will read.\n",
"\n",
"Input assumptions for Step 1:\n",
"\n",
"- The default data directory is `data/` at the repository root.\n",
"- SQLite result files usually use `.db`, `.sqlite`, or `.sqlite3` extensions.\n",
"- The directory can be changed interactively if the result file lives elsewhere."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "imports-and-paths",
"metadata": {},
"outputs": [],
"source": [
"from html import escape\n",
"from pathlib import Path\n",
"import sqlite3\n",
"import sys\n",
"from urllib.parse import quote\n",
"\n",
"from IPython.display import display\n",
"import ipywidgets as widgets\n",
"\n",
"\n",
"def find_repo_root(start: Path | None = None) -> Path:\n",
" \"\"\"Return the nearest parent containing repository-level files.\"\"\"\n",
" current = (start or Path.cwd()).resolve()\n",
" for candidate in (current, *current.parents):\n",
" if (candidate / \"requirements.txt\").exists() and (candidate / \"notebooks\").is_dir():\n",
" return candidate\n",
" return current\n",
"\n",
"\n",
"REPO_ROOT = find_repo_root()\n",
"if str(REPO_ROOT) not in sys.path:\n",
" sys.path.insert(0, str(REPO_ROOT))\n",
"\n",
"from scripts.spbt_day import (\n",
" calculate_ranked_pairs_theo_ret,\n",
" load_selector_pair_rankings,\n",
" load_trading_instructions,\n",
")\n",
"\n",
"DEFAULT_DATA_DIR = REPO_ROOT / \"data\"\n",
"SQLITE_EXTENSIONS = {\".db\", \".sqlite\", \".sqlite3\"}\n",
"\n",
"selected_db_path: Path | None = None\n",
"\n",
"REPO_ROOT, DEFAULT_DATA_DIR"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "database-file-selector",
"metadata": {},
"outputs": [],
"source": [
"directory_input = widgets.Text(\n",
" value=str(DEFAULT_DATA_DIR),\n",
" description=\"Directory\",\n",
" continuous_update=False,\n",
" layout=widgets.Layout(width=\"100%\"),\n",
" style={\"description_width\": \"90px\"},\n",
")\n",
"\n",
"show_all_files = widgets.Checkbox(\n",
" value=False,\n",
" description=\"Show all files\",\n",
" indent=False,\n",
")\n",
"\n",
"refresh_button = widgets.Button(\n",
" description=\"Refresh\",\n",
" icon=\"refresh\",\n",
" button_style=\"\",\n",
" tooltip=\"Rescan the selected directory\",\n",
")\n",
"\n",
"file_select = widgets.Select(\n",
" options=[],\n",
" rows=12,\n",
" description=\"Files\",\n",
" layout=widgets.Layout(width=\"100%\"),\n",
" style={\"description_width\": \"90px\"},\n",
")\n",
"\n",
"selected_path_display = widgets.HTML(value=\"Selected database: none\")\n",
"status_output = widgets.Output()\n",
"\n",
"\n",
"def normalize_directory(raw_path: str) -> Path:\n",
" path = Path(raw_path).expanduser()\n",
" if not path.is_absolute():\n",
" path = REPO_ROOT / path\n",
" return path.resolve()\n",
"\n",
"\n",
"def list_candidate_files(directory: Path, show_all: bool = False) -> list[Path]:\n",
" def result_file_sort_key(path: Path) -> tuple[int, str]:\n",
" name = path.name.lower()\n",
" if \".spbt_results.\" in name:\n",
" priority = 0\n",
" elif \"selector\" in name and \"results\" in name:\n",
" priority = 1\n",
" elif \"results\" in name:\n",
" priority = 2\n",
" else:\n",
" priority = 3\n",
" return priority, name\n",
"\n",
" if show_all:\n",
" return sorted(\n",
" (path for path in directory.iterdir() if path.is_file()),\n",
" key=result_file_sort_key,\n",
" )\n",
" return sorted(\n",
" (\n",
" path\n",
" for path in directory.iterdir()\n",
" if path.is_file() and path.suffix.lower() in SQLITE_EXTENSIONS\n",
" ),\n",
" key=result_file_sort_key,\n",
" )\n",
"\n",
"\n",
"def set_selected_database(path_value: str | None) -> None:\n",
" global selected_db_path\n",
" selected_db_path = Path(path_value).resolve() if path_value else None\n",
" label = str(selected_db_path) if selected_db_path else \"none\"\n",
" selected_path_display.value = f\"Selected database: {escape(label)}\"\n",
"\n",
"\n",
"def refresh_file_list(*_args) -> None:\n",
" directory = normalize_directory(directory_input.value)\n",
" with status_output:\n",
" status_output.clear_output()\n",
" if not directory.exists():\n",
" file_select.options = []\n",
" set_selected_database(None)\n",
" print(f\"Directory does not exist: {directory}\")\n",
" return\n",
" if not directory.is_dir():\n",
" file_select.options = []\n",
" set_selected_database(None)\n",
" print(f\"Path is not a directory: {directory}\")\n",
" return\n",
"\n",
" candidates = list_candidate_files(directory, show_all=show_all_files.value)\n",
" candidate_values = [str(path) for path in candidates]\n",
" previous_value = file_select.value\n",
" file_select.options = [(path.name, str(path)) for path in candidates]\n",
" if candidates:\n",
" file_select.value = previous_value if previous_value in candidate_values else candidate_values[0]\n",
" set_selected_database(file_select.value)\n",
" else:\n",
" set_selected_database(None)\n",
"\n",
" if candidates:\n",
" print(f\"Found {len(candidates)} file(s) in {directory}\")\n",
" else:\n",
" suffixes = \", \".join(sorted(SQLITE_EXTENSIONS))\n",
" print(f\"No SQLite files ({suffixes}) found in {directory}\")\n",
"\n",
"\n",
"def on_file_selected(change) -> None:\n",
" if change[\"name\"] == \"value\":\n",
" set_selected_database(change[\"new\"])\n",
"\n",
"\n",
"refresh_button.on_click(refresh_file_list)\n",
"show_all_files.observe(refresh_file_list, names=\"value\")\n",
"directory_input.observe(refresh_file_list, names=\"value\")\n",
"file_select.observe(on_file_selected, names=\"value\")\n",
"\n",
"display(\n",
" widgets.VBox(\n",
" [\n",
" widgets.HBox([directory_input, refresh_button]),\n",
" show_all_files,\n",
" file_select,\n",
" selected_path_display,\n",
" status_output,\n",
" ]\n",
" )\n",
")\n",
"\n",
"refresh_file_list()"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "selected-database-helpers",
"metadata": {},
"outputs": [],
"source": [
"def selected_database_path() -> Path:\n",
" \"\"\"Return the interactively selected SQLite result path.\"\"\"\n",
" if selected_db_path is None:\n",
" raise ValueError(\"Choose a SQLite result file before continuing.\")\n",
" if not selected_db_path.exists():\n",
" raise FileNotFoundError(f\"Selected database does not exist: {selected_db_path}\")\n",
" if not selected_db_path.is_file():\n",
" raise ValueError(f\"Selected database path is not a file: {selected_db_path}\")\n",
" return selected_db_path\n",
"\n",
"\n",
"def connect_selected_database() -> sqlite3.Connection:\n",
" \"\"\"Open a read-only SQLite connection to the selected result database.\"\"\"\n",
" db_path = selected_database_path()\n",
" uri = f\"file:{quote(db_path.as_posix(), safe='/:')}?mode=ro\"\n",
" return sqlite3.connect(uri, uri=True)\n",
"\n",
"\n",
"# Later notebook sections can call selected_database_path() or connect_selected_database()."
]
},
{
"cell_type": "markdown",
"id": "selector-pair-rankings-context",
"metadata": {},
"source": [
"## Selector Pair Rankings\n",
"\n",
"Load `selector_pairs.pair_name` and `selector_pairs.mr_score` from the selected SQLite database. The JSON field `mr_score.final` is parsed as a numeric score and ranked descending with dense ranks, so tied scores share the same rank and the next distinct score gets the next rank.\n",
"\n",
"Rows with missing, malformed, non-numeric, or non-finite `mr_score.final` values are preserved, sorted after ranked rows, and marked in `mr_score_parse_status`."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "load-selector-pair-rankings",
"metadata": {},
"outputs": [],
"source": [
"conn = connect_selected_database()\n",
"try:\n",
" selector_pair_rankings = load_selector_pair_rankings(conn)\n",
"finally:\n",
" conn.close()\n",
"\n",
"selector_pair_rankings"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "selector-pair-ranking-summary",
"metadata": {},
"outputs": [],
"source": [
"selector_pair_ranking_summary = (\n",
" selector_pair_rankings[\"mr_score_parse_status\"]\n",
" .value_counts(dropna=False)\n",
" .rename_axis(\"mr_score_parse_status\")\n",
" .reset_index(name=\"row_count\")\n",
")\n",
"\n",
"selector_pair_ranking_summary"
]
},
{
"cell_type": "markdown",
"id": "theoretical-return-context",
"metadata": {},
"source": [
"## Theoretical Return by Pair\n",
"\n",
"Load `trading_instructions` and calculate theoretical return for each ranked pair. Each pair starts from a fixed `$10,000` theoretical USD base. `TARGET` opens or replaces the current theoretical position using each asset's `strength` and `reference_price`; `CLOSE` liquidates the open position at the close row's `reference_price`; `HOLD` is ignored.\n",
"\n",
"`realized_pnl` and `unrealized_pnl` are percentage returns relative to `$10,000`, sorted by ascending MR rank."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "load-trading-instructions",
"metadata": {},
"outputs": [],
"source": [
"conn = connect_selected_database()\n",
"try:\n",
" trading_instructions = load_trading_instructions(conn)\n",
"finally:\n",
" conn.close()\n",
"\n",
"trading_instructions"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "calculate-pair-theoretical-returns",
"metadata": {},
"outputs": [],
"source": [
"pair_theo_ret = calculate_ranked_pairs_theo_ret(\n",
" selector_pair_rankings,\n",
" trading_instructions,\n",
")\n",
"\n",
"pair_theo_ret"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "python3.12-venv (3.12.13.final.0)",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.12.13"
}
},
"nbformat": 4,
"nbformat_minor": 5
}