Oleg Sheynin
1370 lines
47 KiB
Python
1370 lines
47 KiB
Python
"""Helpers for single-day SPBT result analysis notebooks."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from html import escape
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
import sqlite3
|
|
from typing import Any
|
|
from urllib.parse import quote
|
|
|
|
import pandas as pd
|
|
|
|
|
|
SELECTOR_PAIRS_COLUMNS = ("pair_name", "mr_score")
|
|
SELECTOR_PAIR_INSTRUMENT_COLUMNS = ("pair_name", "instrument_a", "instrument_b")
|
|
LEGACY_TRADING_INSTRUCTIONS_COLUMNS = ("time_ns", "tstamp", "data")
|
|
SP_QUANT_TRADING_INSTRUCTIONS_COLUMNS = (
|
|
"tstamp_ns",
|
|
"tstamp",
|
|
"action",
|
|
"quote_asset",
|
|
"assets",
|
|
"scaled_disequilibrium",
|
|
"beta",
|
|
)
|
|
MARKET_COLUMNS = ("tstamp", "tstamp_ns", "exch_acct", "instrument_id", "close")
|
|
OHLCV_1MIN_COLUMNS = MARKET_COLUMNS
|
|
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;">▶</button>'
|
|
)
|
|
SELECTED_PAIR_EXECUTION_DISPLAY_COLUMNS = [
|
|
"time",
|
|
"asset",
|
|
"action",
|
|
"side",
|
|
"strength",
|
|
"scaled_disequilibrium",
|
|
"beta",
|
|
"size",
|
|
"price",
|
|
"usd_value",
|
|
]
|
|
INTERACTIVE_TABLE_CSS = """
|
|
table.dataTable,
|
|
table.dataTable th,
|
|
table.dataTable td {
|
|
background-color: #ffffff !important;
|
|
color: #000000 !important;
|
|
}
|
|
|
|
table.dataTable.display tbody tr.odd,
|
|
table.dataTable.display tbody tr.even,
|
|
table.dataTable.display tbody tr:hover,
|
|
table.dataTable.hover tbody tr:hover {
|
|
background-color: #ffffff !important;
|
|
color: #000000 !important;
|
|
}
|
|
|
|
div.dt-container,
|
|
div.dt-container label,
|
|
div.dt-container input,
|
|
div.dt-container select,
|
|
div.dt-container .dt-info,
|
|
div.dt-container .dt-paging,
|
|
div.dt-container .dt-paging .dt-paging-button {
|
|
background-color: #ffffff !important;
|
|
color: #000000 !important;
|
|
}
|
|
"""
|
|
|
|
|
|
def find_repo_root(start: Path | None = None) -> Path:
|
|
"""Return the nearest parent containing repository-level files."""
|
|
current = (start or Path.cwd()).resolve()
|
|
for candidate in (current, *current.parents):
|
|
has_requirements = (candidate / "requirements.txt").exists()
|
|
has_notebooks = (candidate / "notebooks").is_dir()
|
|
if has_requirements and has_notebooks:
|
|
return candidate
|
|
return current
|
|
|
|
|
|
def normalize_directory(raw_path: str, base_dir: Path) -> Path:
|
|
"""Resolve a user-provided directory path relative to a base directory."""
|
|
path = Path(raw_path).expanduser()
|
|
if not path.is_absolute():
|
|
path = base_dir / path
|
|
return path.resolve()
|
|
|
|
|
|
def result_file_sort_key(path: Path) -> tuple[int, str]:
|
|
"""Sort result-like SQLite files before other local database files."""
|
|
name = path.name.lower()
|
|
if ".spbt_results." in name:
|
|
priority = 0
|
|
elif "selector" in name and "results" in name:
|
|
priority = 1
|
|
elif "results" in name:
|
|
priority = 2
|
|
else:
|
|
priority = 3
|
|
return priority, name
|
|
|
|
|
|
def list_candidate_files(directory: Path, show_all: bool = False) -> list[Path]:
|
|
"""List selectable files, preferring SQLite result databases by default."""
|
|
if show_all:
|
|
candidates = (path for path in directory.iterdir() if path.is_file())
|
|
else:
|
|
candidates = (
|
|
path
|
|
for path in directory.iterdir()
|
|
if path.is_file() and path.suffix.lower() in SQLITE_EXTENSIONS
|
|
)
|
|
return sorted(candidates, key=result_file_sort_key)
|
|
|
|
|
|
def read_only_sqlite_uri(db_path: Path) -> str:
|
|
"""Build a read-only SQLite URI for a local database path."""
|
|
return f"file:{quote(db_path.resolve().as_posix(), safe='/:')}?mode=ro"
|
|
|
|
|
|
def connect_sqlite_read_only(db_path: Path) -> sqlite3.Connection:
|
|
"""Open a read-only SQLite connection for a local database path."""
|
|
return sqlite3.connect(read_only_sqlite_uri(db_path), uri=True)
|
|
|
|
|
|
class DatabaseFileSelector:
|
|
"""Small ipywidgets controller for choosing a local SQLite result database."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
repo_root: Path,
|
|
directory_input: Any,
|
|
show_all_files: Any,
|
|
refresh_button: Any,
|
|
file_select: Any,
|
|
selected_path_display: Any,
|
|
status_output: Any,
|
|
widget: Any,
|
|
) -> None:
|
|
self.repo_root = repo_root
|
|
self.directory_input = directory_input
|
|
self.show_all_files = show_all_files
|
|
self.refresh_button = refresh_button
|
|
self.file_select = file_select
|
|
self.selected_path_display = selected_path_display
|
|
self.status_output = status_output
|
|
self.widget = widget
|
|
self.selected_db_path: Path | None = None
|
|
|
|
def set_selected_database(self, path_value: str | None) -> None:
|
|
"""Set the selected database and update its display label."""
|
|
self.selected_db_path = Path(path_value).resolve() if path_value else None
|
|
label = str(self.selected_db_path) if self.selected_db_path else "none"
|
|
self.selected_path_display.value = f"<b>Selected database:</b> {escape(label)}"
|
|
|
|
def refresh_file_list(self, *_args: Any) -> None:
|
|
"""Rescan the configured directory and refresh selectable files."""
|
|
directory = normalize_directory(self.directory_input.value, self.repo_root)
|
|
with self.status_output:
|
|
self.status_output.clear_output()
|
|
if not directory.exists():
|
|
self.file_select.options = []
|
|
self.set_selected_database(None)
|
|
print(f"Directory does not exist: {directory}")
|
|
return
|
|
if not directory.is_dir():
|
|
self.file_select.options = []
|
|
self.set_selected_database(None)
|
|
print(f"Path is not a directory: {directory}")
|
|
return
|
|
|
|
candidates = list_candidate_files(
|
|
directory,
|
|
show_all=self.show_all_files.value,
|
|
)
|
|
candidate_values = [str(path) for path in candidates]
|
|
previous_value = self.file_select.value
|
|
self.file_select.options = [(path.name, str(path)) for path in candidates]
|
|
if candidates:
|
|
self.file_select.value = (
|
|
previous_value
|
|
if previous_value in candidate_values
|
|
else candidate_values[0]
|
|
)
|
|
self.set_selected_database(self.file_select.value)
|
|
else:
|
|
self.set_selected_database(None)
|
|
|
|
if candidates:
|
|
print(f"Found {len(candidates)} file(s) in {directory}")
|
|
else:
|
|
suffixes = ", ".join(sorted(SQLITE_EXTENSIONS))
|
|
print(f"No SQLite files ({suffixes}) found in {directory}")
|
|
|
|
def on_file_selected(self, change: dict[str, Any]) -> None:
|
|
"""Update selected path when the widget selection changes."""
|
|
if change["name"] == "value":
|
|
self.set_selected_database(change["new"])
|
|
|
|
def selected_database_path(self) -> Path:
|
|
"""Return the interactively selected SQLite result path."""
|
|
if self.selected_db_path is None:
|
|
raise ValueError("Choose a SQLite result file before continuing.")
|
|
if not self.selected_db_path.exists():
|
|
raise FileNotFoundError(
|
|
f"Selected database does not exist: {self.selected_db_path}"
|
|
)
|
|
if not self.selected_db_path.is_file():
|
|
raise ValueError(
|
|
f"Selected database path is not a file: {self.selected_db_path}"
|
|
)
|
|
return self.selected_db_path
|
|
|
|
def connect_selected_database(self) -> sqlite3.Connection:
|
|
"""Open a read-only SQLite connection to the selected result database."""
|
|
return connect_sqlite_read_only(self.selected_database_path())
|
|
|
|
|
|
def create_database_file_selector(
|
|
default_data_dir: Path | None = None,
|
|
repo_root: Path | None = None,
|
|
) -> DatabaseFileSelector:
|
|
"""Create an interactive database file selector for notebook use."""
|
|
import ipywidgets as widgets
|
|
|
|
resolved_repo_root = (repo_root or find_repo_root()).resolve()
|
|
resolved_data_dir = (default_data_dir or resolved_repo_root / "data").resolve()
|
|
|
|
directory_input = widgets.Text(
|
|
value=str(resolved_data_dir),
|
|
description="Directory",
|
|
continuous_update=False,
|
|
layout=widgets.Layout(width="100%"),
|
|
style={"description_width": "90px"},
|
|
)
|
|
show_all_files = widgets.Checkbox(
|
|
value=False,
|
|
description="Show all files",
|
|
indent=False,
|
|
)
|
|
refresh_button = widgets.Button(
|
|
description="Refresh",
|
|
icon="refresh",
|
|
button_style="",
|
|
tooltip="Rescan the selected directory",
|
|
)
|
|
file_select = widgets.Select(
|
|
options=[],
|
|
rows=12,
|
|
description="Files",
|
|
layout=widgets.Layout(width="100%"),
|
|
style={"description_width": "90px"},
|
|
)
|
|
selected_path_display = widgets.HTML(value="<b>Selected database:</b> none")
|
|
status_output = widgets.Output()
|
|
widget = widgets.VBox(
|
|
[
|
|
widgets.HBox([directory_input, refresh_button]),
|
|
show_all_files,
|
|
file_select,
|
|
selected_path_display,
|
|
status_output,
|
|
]
|
|
)
|
|
|
|
selector = DatabaseFileSelector(
|
|
repo_root=resolved_repo_root,
|
|
directory_input=directory_input,
|
|
show_all_files=show_all_files,
|
|
refresh_button=refresh_button,
|
|
file_select=file_select,
|
|
selected_path_display=selected_path_display,
|
|
status_output=status_output,
|
|
widget=widget,
|
|
)
|
|
refresh_button.on_click(selector.refresh_file_list)
|
|
show_all_files.observe(selector.refresh_file_list, names="value")
|
|
directory_input.observe(selector.refresh_file_list, names="value")
|
|
file_select.observe(selector.on_file_selected, names="value")
|
|
selector.refresh_file_list()
|
|
return selector
|
|
|
|
|
|
def parse_mr_score_final(raw_score: Any) -> tuple[float | None, str]:
|
|
"""Parse the JSON mr_score.final value, preserving parse status."""
|
|
if raw_score is None:
|
|
return None, "missing_mr_score"
|
|
|
|
try:
|
|
parsed = json.loads(raw_score)
|
|
except (TypeError, json.JSONDecodeError):
|
|
return None, "malformed_json"
|
|
|
|
if not isinstance(parsed, dict):
|
|
return None, "unexpected_json_type"
|
|
|
|
if "final" not in parsed or parsed["final"] in (None, ""):
|
|
return None, "missing_final"
|
|
|
|
final_value = parsed["final"]
|
|
if isinstance(final_value, bool):
|
|
return None, "non_numeric_final"
|
|
|
|
try:
|
|
numeric_final = float(final_value)
|
|
except (TypeError, ValueError):
|
|
return None, "non_numeric_final"
|
|
|
|
if not math.isfinite(numeric_final):
|
|
return None, "non_finite_final"
|
|
|
|
return numeric_final, "ok"
|
|
|
|
|
|
def validate_selector_pairs_table(conn: sqlite3.Connection) -> None:
|
|
"""Raise an actionable error if selector_pairs lacks required columns."""
|
|
table_info = conn.execute("PRAGMA table_info(selector_pairs)").fetchall()
|
|
if not table_info:
|
|
raise ValueError("SQLite database is missing required table: selector_pairs")
|
|
|
|
existing_columns = {row[1] for row in table_info}
|
|
missing_columns = set(SELECTOR_PAIRS_COLUMNS) - existing_columns
|
|
if missing_columns:
|
|
missing = ", ".join(sorted(missing_columns))
|
|
raise ValueError(f"selector_pairs is missing required column(s): {missing}")
|
|
|
|
|
|
def validate_selector_pair_instrument_columns(conn: sqlite3.Connection) -> None:
|
|
"""Raise if selector_pairs cannot map pair names to market instruments."""
|
|
table_info = conn.execute("PRAGMA table_info(selector_pairs)").fetchall()
|
|
if not table_info:
|
|
raise ValueError("SQLite database is missing required table: selector_pairs")
|
|
|
|
existing_columns = {row[1] for row in table_info}
|
|
missing_columns = set(SELECTOR_PAIR_INSTRUMENT_COLUMNS) - existing_columns
|
|
if missing_columns:
|
|
missing = ", ".join(sorted(missing_columns))
|
|
raise ValueError(f"selector_pairs is missing required column(s): {missing}")
|
|
|
|
|
|
def rank_selector_pairs(selector_pairs: pd.DataFrame) -> pd.DataFrame:
|
|
"""Rank selector pairs by dense descending mr_score.final."""
|
|
missing_columns = set(SELECTOR_PAIRS_COLUMNS) - set(selector_pairs.columns)
|
|
if missing_columns:
|
|
missing = ", ".join(sorted(missing_columns))
|
|
raise ValueError(f"selector_pairs dataframe is missing column(s): {missing}")
|
|
|
|
ranked = selector_pairs.loc[:, list(SELECTOR_PAIRS_COLUMNS)].copy()
|
|
parsed_scores = ranked["mr_score"].map(parse_mr_score_final)
|
|
ranked["mr_score_final"] = [score for score, _status in parsed_scores]
|
|
ranked["mr_score_parse_status"] = [status for _score, status in parsed_scores]
|
|
|
|
ranked["pair_rank"] = (
|
|
ranked["mr_score_final"].rank(method="dense", ascending=False).astype("Int64")
|
|
)
|
|
|
|
return (
|
|
ranked.loc[
|
|
:,
|
|
[
|
|
"pair_rank",
|
|
"pair_name",
|
|
"mr_score_final",
|
|
"mr_score_parse_status",
|
|
"mr_score",
|
|
],
|
|
]
|
|
.sort_values(["pair_rank", "pair_name"], na_position="last", kind="mergesort")
|
|
.reset_index(drop=True)
|
|
)
|
|
|
|
|
|
def load_selector_pair_rankings(conn: sqlite3.Connection) -> pd.DataFrame:
|
|
"""Load selector_pairs from SQLite and return dense-ranked pair rows."""
|
|
validate_selector_pairs_table(conn)
|
|
selector_pairs = pd.read_sql_query(
|
|
"SELECT pair_name, mr_score FROM selector_pairs",
|
|
conn,
|
|
)
|
|
return rank_selector_pairs(selector_pairs)
|
|
|
|
|
|
def table_column_names(conn: sqlite3.Connection, table_name: str) -> set[str]:
|
|
"""Return SQLite column names for an existing table, or an empty set."""
|
|
return {
|
|
row[0]
|
|
for row in conn.execute(
|
|
"SELECT name FROM pragma_table_info(?)",
|
|
(table_name,),
|
|
)
|
|
}
|
|
|
|
|
|
def validate_trading_instructions_table(conn: sqlite3.Connection) -> None:
|
|
"""Raise an actionable error if trading_instructions lacks required columns."""
|
|
existing_columns = table_column_names(conn, "trading_instructions")
|
|
if not existing_columns:
|
|
raise ValueError("SQLite database is missing required table: trading_instructions")
|
|
|
|
required_schemas = (
|
|
set(SP_QUANT_TRADING_INSTRUCTIONS_COLUMNS),
|
|
set(LEGACY_TRADING_INSTRUCTIONS_COLUMNS),
|
|
)
|
|
if any(required_schema <= existing_columns for required_schema in required_schemas):
|
|
return
|
|
|
|
missing_by_schema = [
|
|
", ".join(sorted(required_schema - existing_columns))
|
|
for required_schema in required_schemas
|
|
]
|
|
raise ValueError(
|
|
"trading_instructions does not match a supported schema; missing either "
|
|
f"SP Quant column(s) [{missing_by_schema[0]}] or legacy column(s) "
|
|
f"[{missing_by_schema[1]}]"
|
|
)
|
|
|
|
|
|
def load_trading_instructions(conn: sqlite3.Connection) -> pd.DataFrame:
|
|
"""Load trading_instructions ordered by timestamp.
|
|
|
|
SP Quant result databases store instruction fields as explicit columns. The
|
|
returned dataframe keeps those columns and adds a time_ns alias so existing
|
|
calculations and notebooks can use one timestamp name.
|
|
"""
|
|
validate_trading_instructions_table(conn)
|
|
columns = table_column_names(conn, "trading_instructions")
|
|
if "tstamp_ns" in columns:
|
|
select_expression = "*"
|
|
if "time_ns" not in columns:
|
|
select_expression = "*, tstamp_ns AS time_ns"
|
|
return pd.read_sql_query(
|
|
f"""
|
|
SELECT {select_expression}
|
|
FROM trading_instructions
|
|
ORDER BY tstamp_ns, rowid
|
|
""",
|
|
conn,
|
|
)
|
|
|
|
return pd.read_sql_query(
|
|
"SELECT * FROM trading_instructions ORDER BY time_ns, rowid",
|
|
conn,
|
|
)
|
|
|
|
|
|
def infer_trading_day_start_ns(trd_inst_df: pd.DataFrame) -> int:
|
|
"""Infer the UTC midnight timestamp for the trading-instruction day."""
|
|
if "time_ns" not in trd_inst_df.columns:
|
|
raise ValueError("trading instructions dataframe is missing column: time_ns")
|
|
|
|
time_ns = pd.to_numeric(trd_inst_df["time_ns"], errors="coerce").dropna()
|
|
if time_ns.empty:
|
|
raise ValueError("trading instructions dataframe does not contain timestamps")
|
|
|
|
first_timestamp = pd.to_datetime(int(time_ns.min()), unit="ns", utc=True)
|
|
return int(first_timestamp.floor("D").value)
|
|
|
|
|
|
def validate_ohlcv_1min_table(conn: sqlite3.Connection) -> None:
|
|
"""Raise an actionable error if ohlcv_1min lacks required columns."""
|
|
table_info = conn.execute("PRAGMA table_info(ohlcv_1min)").fetchall()
|
|
if not table_info:
|
|
raise ValueError("SQLite database is missing required table: ohlcv_1min")
|
|
|
|
existing_columns = {row[1] for row in table_info}
|
|
missing_columns = set(OHLCV_1MIN_COLUMNS) - existing_columns
|
|
if missing_columns:
|
|
missing = ", ".join(sorted(missing_columns))
|
|
raise ValueError(f"ohlcv_1min is missing required column(s): {missing}")
|
|
|
|
|
|
def validate_market_table(conn: sqlite3.Connection) -> None:
|
|
"""Raise an actionable error if market lacks required columns."""
|
|
table_info = conn.execute("PRAGMA table_info(market)").fetchall()
|
|
if not table_info:
|
|
raise ValueError("SQLite database is missing required table: market")
|
|
|
|
existing_columns = {row[1] for row in table_info}
|
|
missing_columns = set(MARKET_COLUMNS) - existing_columns
|
|
if missing_columns:
|
|
missing = ", ".join(sorted(missing_columns))
|
|
raise ValueError(f"market is missing required column(s): {missing}")
|
|
|
|
|
|
def pair_assets_and_quote(pair_name: str) -> tuple[tuple[str, ...], str]:
|
|
"""Parse a pair name like ADA:USD-BTC:USD into assets and quote asset."""
|
|
pair_legs = pair_name.split("-")
|
|
if len(pair_legs) != 2:
|
|
raise ValueError(f"Pair name must contain exactly two legs: {pair_name}")
|
|
|
|
assets: list[str] = []
|
|
quote_assets: list[str] = []
|
|
for leg in pair_legs:
|
|
parts = leg.split(":")
|
|
if len(parts) != 2 or not all(parts):
|
|
raise ValueError(f"Pair leg must use ASSET:QUOTE form: {leg}")
|
|
assets.append(parts[0])
|
|
quote_assets.append(parts[1])
|
|
|
|
distinct_quote_assets = set(quote_assets)
|
|
if len(distinct_quote_assets) != 1:
|
|
raise ValueError(f"Pair legs must use the same quote asset: {pair_name}")
|
|
|
|
return tuple(assets), quote_assets[0]
|
|
|
|
|
|
def _parse_instruction_data(raw_data: Any) -> dict[str, Any] | None:
|
|
if raw_data is None:
|
|
return None
|
|
if isinstance(raw_data, dict):
|
|
return raw_data
|
|
|
|
try:
|
|
parsed = json.loads(raw_data)
|
|
except (TypeError, json.JSONDecodeError):
|
|
return None
|
|
|
|
return parsed if isinstance(parsed, dict) else None
|
|
|
|
|
|
def _instruction_value(row: Any, column: str) -> Any:
|
|
return getattr(row, column, None)
|
|
|
|
|
|
def _instruction_payload(row: Any) -> dict[str, Any] | None:
|
|
legacy_payload = _parse_instruction_data(_instruction_value(row, "data"))
|
|
if legacy_payload is not None:
|
|
return legacy_payload
|
|
|
|
assets = _parse_instruction_data(_instruction_value(row, "assets"))
|
|
if assets is None:
|
|
return None
|
|
|
|
return {
|
|
"action": _instruction_value(row, "action"),
|
|
"quote_asset": _instruction_value(row, "quote_asset"),
|
|
"assets": assets,
|
|
}
|
|
|
|
|
|
def _finite_float(value: Any, field_name: str, pair_name: str) -> float:
|
|
if isinstance(value, bool):
|
|
raise ValueError(f"{field_name} for {pair_name} must be numeric, got bool")
|
|
|
|
try:
|
|
numeric_value = float(value)
|
|
except (TypeError, ValueError) as exc:
|
|
raise ValueError(
|
|
f"{field_name} for {pair_name} must be numeric, got {value!r}"
|
|
) from exc
|
|
|
|
if not math.isfinite(numeric_value):
|
|
raise ValueError(f"{field_name} for {pair_name} must be finite")
|
|
|
|
return numeric_value
|
|
|
|
|
|
def _reference_price(asset_data: Any, asset: str, pair_name: str) -> float:
|
|
if not isinstance(asset_data, dict):
|
|
raise ValueError(f"Asset data for {asset} in {pair_name} must be a JSON object")
|
|
|
|
reference_price = _finite_float(
|
|
asset_data.get("reference_price"),
|
|
f"reference_price[{asset}]",
|
|
pair_name,
|
|
)
|
|
if reference_price <= 0:
|
|
raise ValueError(f"reference_price[{asset}] for {pair_name} must be positive")
|
|
return reference_price
|
|
|
|
|
|
def _strength(asset_data: Any, asset: str, pair_name: str) -> float:
|
|
if not isinstance(asset_data, dict):
|
|
raise ValueError(f"Asset data for {asset} in {pair_name} must be a JSON object")
|
|
return _finite_float(asset_data.get("strength"), f"strength[{asset}]", pair_name)
|
|
|
|
|
|
def _sort_trading_instructions(trd_inst_df: pd.DataFrame) -> pd.DataFrame:
|
|
sort_columns = [column for column in ("time_ns", "tstamp") if column in trd_inst_df]
|
|
ordered = trd_inst_df.copy()
|
|
ordered["_input_order"] = range(len(ordered))
|
|
return ordered.sort_values(
|
|
[*sort_columns, "_input_order"],
|
|
kind="mergesort",
|
|
).drop(columns="_input_order")
|
|
|
|
|
|
def _matching_pair_instructions(
|
|
pair_name: str,
|
|
trd_inst_df: pd.DataFrame,
|
|
) -> list[dict[str, Any]]:
|
|
return [
|
|
instruction_row["data"]
|
|
for instruction_row in _matching_pair_instruction_rows(pair_name, trd_inst_df)
|
|
]
|
|
|
|
|
|
def _matching_pair_instruction_rows(
|
|
pair_name: str,
|
|
trd_inst_df: pd.DataFrame,
|
|
) -> list[dict[str, Any]]:
|
|
pair_assets, quote_asset = pair_assets_and_quote(pair_name)
|
|
pair_asset_set = set(pair_assets)
|
|
|
|
legacy_columns = {"data"}
|
|
sp_quant_columns = set(SP_QUANT_TRADING_INSTRUCTIONS_COLUMNS)
|
|
dataframe_columns = set(trd_inst_df.columns)
|
|
if not (
|
|
legacy_columns <= dataframe_columns
|
|
or sp_quant_columns <= dataframe_columns
|
|
):
|
|
raise ValueError(
|
|
"trading instructions dataframe does not match a supported schema"
|
|
)
|
|
|
|
selected_instructions: list[dict[str, Any]] = []
|
|
for instruction_row in _sort_trading_instructions(trd_inst_df).itertuples(
|
|
index=False
|
|
):
|
|
parsed = _instruction_payload(instruction_row)
|
|
if parsed is None or parsed.get("quote_asset") != quote_asset:
|
|
continue
|
|
|
|
assets = parsed.get("assets")
|
|
if not isinstance(assets, dict) or set(assets) != pair_asset_set:
|
|
continue
|
|
|
|
selected_instructions.append(
|
|
{
|
|
"time_ns": getattr(instruction_row, "time_ns", None),
|
|
"tstamp": getattr(instruction_row, "tstamp", None),
|
|
"scaled_disequilibrium": getattr(
|
|
instruction_row,
|
|
"scaled_disequilibrium",
|
|
None,
|
|
),
|
|
"beta": getattr(instruction_row, "beta", None),
|
|
"data": parsed,
|
|
}
|
|
)
|
|
|
|
return selected_instructions
|
|
|
|
|
|
def _execution_side(size: float) -> str:
|
|
return "BUY" if size > 0 else "SELL"
|
|
|
|
|
|
def _validate_min_pctg_change(min_pctg_change: float) -> float:
|
|
try:
|
|
numeric_min_pctg_change = float(min_pctg_change)
|
|
except (TypeError, ValueError) as exc:
|
|
raise ValueError("min_pctg_change must be numeric") from exc
|
|
|
|
if not math.isfinite(numeric_min_pctg_change):
|
|
raise ValueError("min_pctg_change must be finite")
|
|
if numeric_min_pctg_change < 0:
|
|
raise ValueError("min_pctg_change must be non-negative")
|
|
return numeric_min_pctg_change
|
|
|
|
|
|
def _target_strength_change_pctg(
|
|
current_strength: float | None,
|
|
target_strength: float,
|
|
) -> float | None:
|
|
if current_strength is None:
|
|
return None
|
|
if current_strength == 0:
|
|
return math.inf if target_strength != 0 else 0.0
|
|
return abs((target_strength - current_strength) / current_strength) * 100.0
|
|
|
|
|
|
def calculate_pair_theo_executions(
|
|
pair_name: str,
|
|
trd_inst_df: pd.DataFrame,
|
|
min_pctg_change: float = 0,
|
|
) -> pd.DataFrame:
|
|
"""Create asset-level theoretical executions for one selected pair.
|
|
|
|
TARGET rows trade from the current theoretical position to the new target
|
|
position when the absolute percentage strength change since the last
|
|
executed TARGET reaches min_pctg_change. CLOSE rows flatten the current
|
|
theoretical position. Positive size is a BUY; negative size is a SELL. USD
|
|
value is the opposite signed cash movement, so buys are negative and sells
|
|
are positive.
|
|
"""
|
|
min_pctg_change = _validate_min_pctg_change(min_pctg_change)
|
|
pair_assets, _quote_asset = pair_assets_and_quote(pair_name)
|
|
current_sizes = {asset: 0.0 for asset in pair_assets}
|
|
current_strengths: dict[str, float | None] = {asset: None for asset in pair_assets}
|
|
records: list[dict[str, Any]] = []
|
|
execution_order = 0
|
|
|
|
for instruction in _matching_pair_instruction_rows(pair_name, trd_inst_df):
|
|
action = instruction["data"].get("action")
|
|
assets_data = instruction["data"]["assets"]
|
|
|
|
if action == "TARGET":
|
|
for asset in pair_assets:
|
|
asset_data = assets_data[asset]
|
|
target_strength = _strength(
|
|
asset_data,
|
|
asset,
|
|
pair_name,
|
|
)
|
|
price = _reference_price(asset_data, asset, pair_name)
|
|
target_size = INITIAL_THEO_CAPITAL_USD * target_strength / price
|
|
strength_change_pctg = _target_strength_change_pctg(
|
|
current_strengths[asset],
|
|
target_strength,
|
|
)
|
|
if (
|
|
strength_change_pctg is not None
|
|
and strength_change_pctg < min_pctg_change
|
|
and not math.isclose(strength_change_pctg, min_pctg_change)
|
|
):
|
|
continue
|
|
|
|
trade_size = target_size - current_sizes[asset]
|
|
if trade_size == 0:
|
|
continue
|
|
|
|
records.append(
|
|
{
|
|
"time": instruction["tstamp"] or instruction["time_ns"],
|
|
"time_ns": instruction["time_ns"],
|
|
"pair_name": pair_name,
|
|
"asset": asset,
|
|
"action": action,
|
|
"side": _execution_side(trade_size),
|
|
"strength": target_strength,
|
|
"scaled_disequilibrium": instruction[
|
|
"scaled_disequilibrium"
|
|
],
|
|
"beta": instruction["beta"],
|
|
"size": trade_size,
|
|
"price": price,
|
|
"usd_value": -trade_size * price,
|
|
"_execution_order": execution_order,
|
|
}
|
|
)
|
|
execution_order += 1
|
|
current_sizes[asset] = target_size
|
|
current_strengths[asset] = target_strength
|
|
elif action == "CLOSE":
|
|
if not any(current_sizes.values()):
|
|
continue
|
|
|
|
for asset in pair_assets:
|
|
current_size = current_sizes[asset]
|
|
if current_size == 0:
|
|
continue
|
|
|
|
asset_data = assets_data[asset]
|
|
trade_size = -current_size
|
|
price = _reference_price(asset_data, asset, pair_name)
|
|
records.append(
|
|
{
|
|
"time": instruction["tstamp"] or instruction["time_ns"],
|
|
"time_ns": instruction["time_ns"],
|
|
"pair_name": pair_name,
|
|
"asset": asset,
|
|
"action": action,
|
|
"side": _execution_side(trade_size),
|
|
"strength": None,
|
|
"scaled_disequilibrium": instruction[
|
|
"scaled_disequilibrium"
|
|
],
|
|
"beta": instruction["beta"],
|
|
"size": trade_size,
|
|
"price": price,
|
|
"usd_value": -trade_size * price,
|
|
"_execution_order": execution_order,
|
|
}
|
|
)
|
|
execution_order += 1
|
|
current_sizes[asset] = 0.0
|
|
current_strengths[asset] = None
|
|
|
|
columns = [
|
|
"time",
|
|
"time_ns",
|
|
"pair_name",
|
|
"asset",
|
|
"action",
|
|
"side",
|
|
"strength",
|
|
"scaled_disequilibrium",
|
|
"beta",
|
|
"size",
|
|
"price",
|
|
"usd_value",
|
|
"_execution_order",
|
|
]
|
|
return (
|
|
pd.DataFrame.from_records(records, columns=columns)
|
|
.sort_values(
|
|
["time_ns", "_execution_order"],
|
|
kind="mergesort",
|
|
na_position="last",
|
|
)
|
|
.drop(columns="_execution_order")
|
|
.reset_index(drop=True)
|
|
)
|
|
|
|
|
|
def calculate_pair_theo_ret_from_executions(
|
|
pair_name: str,
|
|
trd_inst_df: pd.DataFrame,
|
|
min_pctg_change: float = 0,
|
|
) -> dict[str, float | int | str]:
|
|
"""Calculate pair TheoRet from the generated theoretical executions."""
|
|
executions = calculate_pair_theo_executions(
|
|
pair_name,
|
|
trd_inst_df,
|
|
min_pctg_change=min_pctg_change,
|
|
)
|
|
if executions.empty:
|
|
return {
|
|
"pair_name": pair_name,
|
|
"num_trades": 0,
|
|
"realized_pnl": 0.0,
|
|
"unrealized_pnl": 0.0,
|
|
}
|
|
|
|
pair_assets, _quote_asset = pair_assets_and_quote(pair_name)
|
|
current_sizes = {asset: 0.0 for asset in pair_assets}
|
|
latest_prices = {asset: 0.0 for asset in pair_assets}
|
|
open_cash_flow_usd = 0.0
|
|
realized_pnl_usd = 0.0
|
|
|
|
for execution in executions.itertuples(index=False):
|
|
current_sizes[execution.asset] += execution.size
|
|
latest_prices[execution.asset] = execution.price
|
|
open_cash_flow_usd += execution.usd_value
|
|
if execution.action == "CLOSE" and not any(current_sizes.values()):
|
|
realized_pnl_usd += open_cash_flow_usd
|
|
open_cash_flow_usd = 0.0
|
|
|
|
unrealized_pnl_usd = open_cash_flow_usd + sum(
|
|
current_sizes[asset] * latest_prices[asset] for asset in pair_assets
|
|
)
|
|
|
|
return {
|
|
"pair_name": pair_name,
|
|
"num_trades": len(executions),
|
|
"realized_pnl": realized_pnl_usd / INITIAL_THEO_CAPITAL_USD * 100.0,
|
|
"unrealized_pnl": unrealized_pnl_usd / INITIAL_THEO_CAPITAL_USD * 100.0,
|
|
}
|
|
|
|
|
|
def calculate_pair_theo_ret(
|
|
pair_name: str,
|
|
trd_inst_df: pd.DataFrame,
|
|
min_pctg_change: float = 0,
|
|
) -> dict[str, float | int | str]:
|
|
"""Calculate realized and unrealized TheoRet percentages for one pair.
|
|
|
|
TARGET actions trade the delta between current and target theoretical
|
|
positions when the target strength change reaches min_pctg_change. CLOSE
|
|
actions liquidate the currently open position. HOLD and unknown actions are
|
|
ignored. Returned PnL values are percentages of the fixed 10,000 USD
|
|
theoretical capital base.
|
|
"""
|
|
return calculate_pair_theo_ret_from_executions(
|
|
pair_name,
|
|
trd_inst_df,
|
|
min_pctg_change=min_pctg_change,
|
|
)
|
|
|
|
|
|
def parse_selector_instrument(raw_instrument: Any) -> tuple[str, str]:
|
|
"""Parse selector_pairs instrument value into exch_acct and instrument_id."""
|
|
if not isinstance(raw_instrument, str) or ":" not in raw_instrument:
|
|
raise ValueError(f"selector instrument must use EXCH_ACCT:INSTRUMENT form")
|
|
|
|
exch_acct, instrument_id = raw_instrument.split(":", 1)
|
|
if not exch_acct or not instrument_id:
|
|
raise ValueError(f"selector instrument must use EXCH_ACCT:INSTRUMENT form")
|
|
return exch_acct, instrument_id
|
|
|
|
|
|
def load_pair_market_data(
|
|
conn: sqlite3.Connection,
|
|
pair_name: str,
|
|
*,
|
|
trading_day_start_ns: int,
|
|
) -> pd.DataFrame:
|
|
"""Load market close data from trading-day start for selected instruments."""
|
|
validate_selector_pair_instrument_columns(conn)
|
|
validate_market_table(conn)
|
|
pair_assets, _quote_asset = pair_assets_and_quote(pair_name)
|
|
|
|
selector_pair = pd.read_sql_query(
|
|
"""
|
|
SELECT instrument_a, instrument_b
|
|
FROM selector_pairs
|
|
WHERE pair_name = ?
|
|
ORDER BY rowid
|
|
LIMIT 1
|
|
""",
|
|
conn,
|
|
params=(pair_name,),
|
|
)
|
|
if selector_pair.empty:
|
|
raise ValueError(f"selector_pairs does not contain pair_name: {pair_name}")
|
|
|
|
instrument_values = [
|
|
selector_pair["instrument_a"].iloc[0],
|
|
selector_pair["instrument_b"].iloc[0],
|
|
]
|
|
|
|
market_frames = []
|
|
missing_market_assets = []
|
|
for asset, raw_instrument in zip(pair_assets, instrument_values, strict=True):
|
|
exch_acct, instrument_id = parse_selector_instrument(raw_instrument)
|
|
instrument_market_data = pd.read_sql_query(
|
|
"""
|
|
SELECT
|
|
tstamp AS time,
|
|
tstamp_ns AS time_ns,
|
|
exch_acct,
|
|
instrument_id,
|
|
close
|
|
FROM market
|
|
WHERE exch_acct = ? AND instrument_id = ? AND tstamp_ns >= ?
|
|
ORDER BY tstamp_ns, rowid
|
|
""",
|
|
conn,
|
|
params=(exch_acct, instrument_id, trading_day_start_ns),
|
|
)
|
|
if instrument_market_data.empty:
|
|
missing_market_assets.append(asset)
|
|
continue
|
|
|
|
instrument_market_data["pair_name"] = pair_name
|
|
instrument_market_data["asset"] = asset
|
|
market_frames.append(instrument_market_data)
|
|
|
|
if missing_market_assets:
|
|
missing_assets = ", ".join(sorted(missing_market_assets))
|
|
raise ValueError(
|
|
f"market does not contain data for asset(s): {missing_assets}"
|
|
)
|
|
|
|
market_data = pd.concat(market_frames, ignore_index=True)
|
|
market_data["close"] = pd.to_numeric(market_data["close"], errors="coerce")
|
|
missing_start_price_assets = sorted(
|
|
set(pair_assets)
|
|
- set(market_data.loc[market_data["time_ns"] == trading_day_start_ns, "asset"])
|
|
)
|
|
if missing_start_price_assets:
|
|
missing_assets = ", ".join(missing_start_price_assets)
|
|
raise ValueError(
|
|
"market does not contain trading-day start close for "
|
|
f"asset(s): {missing_assets}"
|
|
)
|
|
|
|
initial_close_by_asset = market_data.drop_duplicates("asset").set_index("asset")[
|
|
"close"
|
|
]
|
|
market_data["initial_close"] = market_data["asset"].map(
|
|
initial_close_by_asset
|
|
)
|
|
invalid_initial_close = (
|
|
market_data["initial_close"].isna() | (market_data["initial_close"] <= 0)
|
|
)
|
|
if invalid_initial_close.any():
|
|
missing_assets = ", ".join(
|
|
sorted(market_data.loc[invalid_initial_close, "asset"].unique())
|
|
)
|
|
raise ValueError(
|
|
f"market initial close must be positive for asset(s): {missing_assets}"
|
|
)
|
|
|
|
market_data["relative_close"] = (
|
|
market_data["close"] - market_data["initial_close"]
|
|
) / market_data["initial_close"]
|
|
return market_data.loc[
|
|
:,
|
|
[
|
|
"pair_name",
|
|
"asset",
|
|
"time",
|
|
"time_ns",
|
|
"exch_acct",
|
|
"instrument_id",
|
|
"close",
|
|
"initial_close",
|
|
"relative_close",
|
|
],
|
|
]
|
|
|
|
|
|
def _normalize_trade_prices_to_initial_close(
|
|
theo_executions: pd.DataFrame,
|
|
market_data: pd.DataFrame,
|
|
) -> pd.DataFrame:
|
|
required_execution_columns = {"asset", "side", "price", "time", "time_ns"}
|
|
missing_execution_columns = required_execution_columns - set(
|
|
theo_executions.columns
|
|
)
|
|
if missing_execution_columns:
|
|
missing = ", ".join(sorted(missing_execution_columns))
|
|
raise ValueError(f"theo executions dataframe missing column(s): {missing}")
|
|
|
|
required_market_columns = {"asset", "initial_close"}
|
|
missing_market_columns = required_market_columns - set(market_data.columns)
|
|
if missing_market_columns:
|
|
missing = ", ".join(sorted(missing_market_columns))
|
|
raise ValueError(f"market data dataframe missing column(s): {missing}")
|
|
|
|
if theo_executions.empty:
|
|
return theo_executions.assign(relative_price=pd.Series(dtype="float64"))
|
|
|
|
initial_close_by_asset = (
|
|
market_data.dropna(subset=["initial_close"])
|
|
.drop_duplicates("asset")
|
|
.set_index("asset")["initial_close"]
|
|
)
|
|
trades = theo_executions.copy()
|
|
trades["initial_close"] = trades["asset"].map(initial_close_by_asset)
|
|
missing_initial_close = trades["initial_close"].isna() | (
|
|
trades["initial_close"] <= 0
|
|
)
|
|
if missing_initial_close.any():
|
|
missing_assets = ", ".join(
|
|
sorted(trades.loc[missing_initial_close, "asset"].dropna().unique())
|
|
)
|
|
raise ValueError(
|
|
f"market data initial close is required for trade asset(s): {missing_assets}"
|
|
)
|
|
|
|
trades["price"] = pd.to_numeric(trades["price"], errors="coerce")
|
|
trades["relative_price"] = (
|
|
trades["price"] - trades["initial_close"]
|
|
) / trades["initial_close"]
|
|
return trades
|
|
|
|
|
|
def create_pair_trades_market_plot(
|
|
pair_name: str,
|
|
market_data: pd.DataFrame,
|
|
theo_executions: pd.DataFrame,
|
|
) -> Any:
|
|
"""Plot relative market closes and selected-pair theoretical trades."""
|
|
import plotly.graph_objects as go
|
|
|
|
required_market_columns = {"asset", "time", "close", "relative_close"}
|
|
missing_market_columns = required_market_columns - set(market_data.columns)
|
|
if missing_market_columns:
|
|
missing = ", ".join(sorted(missing_market_columns))
|
|
raise ValueError(f"market data dataframe missing column(s): {missing}")
|
|
|
|
pair_assets, _quote_asset = pair_assets_and_quote(pair_name)
|
|
trades = _normalize_trade_prices_to_initial_close(theo_executions, market_data)
|
|
figure = go.Figure()
|
|
|
|
for asset in pair_assets:
|
|
asset_market_data = market_data.loc[market_data["asset"] == asset].sort_values(
|
|
["time_ns", "time"],
|
|
kind="mergesort",
|
|
na_position="last",
|
|
)
|
|
figure.add_trace(
|
|
go.Scatter(
|
|
x=asset_market_data["time"],
|
|
y=asset_market_data["relative_close"],
|
|
mode="lines",
|
|
name=f"{asset} close",
|
|
hovertemplate=(
|
|
"Asset=%{customdata[0]}<br>"
|
|
"Time=%{x}<br>"
|
|
"Close=%{customdata[1]:.8g}<br>"
|
|
"Relative=%{y:.4%}<extra></extra>"
|
|
),
|
|
customdata=asset_market_data[["asset", "close"]],
|
|
)
|
|
)
|
|
|
|
for side, color, symbol in (
|
|
("BUY", "darkgreen", "triangle-up"),
|
|
("SELL", "darkred", "triangle-down"),
|
|
):
|
|
asset_side_trades = trades.loc[
|
|
(trades["asset"] == asset) & (trades["side"] == side)
|
|
].sort_values(["time_ns", "time"], kind="mergesort", na_position="last")
|
|
if asset_side_trades.empty:
|
|
continue
|
|
|
|
figure.add_trace(
|
|
go.Scatter(
|
|
x=asset_side_trades["time"],
|
|
y=asset_side_trades["relative_price"],
|
|
mode="markers",
|
|
name=f"{asset} {side}",
|
|
marker={
|
|
"symbol": symbol,
|
|
"color": color,
|
|
"size": 11,
|
|
"line": {"color": "white", "width": 1},
|
|
},
|
|
hovertemplate=(
|
|
"Asset=%{customdata[0]}<br>"
|
|
"Side=%{customdata[1]}<br>"
|
|
"Action=%{customdata[2]}<br>"
|
|
"Time=%{x}<br>"
|
|
"Price=%{customdata[3]:.8g}<br>"
|
|
"Relative=%{y:.4%}<br>"
|
|
"Size=%{customdata[4]:.8g}<extra></extra>"
|
|
),
|
|
customdata=asset_side_trades[
|
|
["asset", "side", "action", "price", "size"]
|
|
],
|
|
)
|
|
)
|
|
|
|
figure.update_layout(
|
|
title=f"{format_pair_name_for_display(pair_name)} Trades on Market Data",
|
|
xaxis_title="Time",
|
|
yaxis_title="Relative price",
|
|
hovermode="x unified",
|
|
legend_title="Series",
|
|
)
|
|
if not market_data.empty:
|
|
figure.update_xaxes(range=[market_data["time"].min(), market_data["time"].max()])
|
|
figure.update_yaxes(tickformat=".2%")
|
|
return figure
|
|
|
|
|
|
def calculate_ranked_pairs_theo_ret(
|
|
selector_pair_rankings: pd.DataFrame,
|
|
trd_inst_df: pd.DataFrame,
|
|
min_pctg_change: float = 0,
|
|
) -> pd.DataFrame:
|
|
"""Calculate TheoRet percentages for every ranked selector pair."""
|
|
min_pctg_change = _validate_min_pctg_change(min_pctg_change)
|
|
required_columns = {"pair_name", "pair_rank"}
|
|
missing_columns = required_columns - set(selector_pair_rankings.columns)
|
|
if missing_columns:
|
|
missing = ", ".join(sorted(missing_columns))
|
|
raise ValueError(f"selector pair rankings missing column(s): {missing}")
|
|
|
|
records = []
|
|
for row in selector_pair_rankings.itertuples(index=False):
|
|
pair_result = calculate_pair_theo_ret(
|
|
row.pair_name,
|
|
trd_inst_df,
|
|
min_pctg_change=min_pctg_change,
|
|
)
|
|
records.append(
|
|
{
|
|
"pair_name": pair_result["pair_name"],
|
|
"mr_ranking": row.pair_rank,
|
|
"num_trades": pair_result["num_trades"],
|
|
"realized_pnl": pair_result["realized_pnl"],
|
|
"unrealized_pnl": pair_result["unrealized_pnl"],
|
|
}
|
|
)
|
|
|
|
return (
|
|
pd.DataFrame.from_records(
|
|
records,
|
|
columns=[
|
|
"pair_name",
|
|
"mr_ranking",
|
|
"num_trades",
|
|
"realized_pnl",
|
|
"unrealized_pnl",
|
|
],
|
|
)
|
|
.sort_values(["mr_ranking", "pair_name"], na_position="last", kind="mergesort")
|
|
.reset_index(drop=True)
|
|
)
|
|
|
|
|
|
def format_pair_name_for_display(pair_name: Any) -> Any:
|
|
"""Return a human-facing pair label without the USD quote suffix."""
|
|
if pd.isna(pair_name):
|
|
return pair_name
|
|
return "-".join(
|
|
leg.removesuffix(PAIR_NAME_DISPLAY_SUFFIX) for leg in str(pair_name).split("-")
|
|
)
|
|
|
|
|
|
def format_pair_names_for_display(
|
|
dataframe: pd.DataFrame,
|
|
column: str = "pair_name",
|
|
) -> pd.DataFrame:
|
|
"""Return a copy with pair-name labels formatted for display."""
|
|
if column not in dataframe.columns:
|
|
raise ValueError(f"dataframe missing column: {column}")
|
|
formatted = dataframe.copy()
|
|
formatted[column] = formatted[column].map(format_pair_name_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,
|
|
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,
|
|
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"}
|
|
missing_columns = required_columns - set(pair_theo_ret.columns)
|
|
if missing_columns:
|
|
missing = ", ".join(sorted(missing_columns))
|
|
raise ValueError(f"pair TheoRet dataframe missing column(s): {missing}")
|
|
return pair_theo_ret.assign(
|
|
total_pnl=pair_theo_ret["realized_pnl"] + pair_theo_ret["unrealized_pnl"]
|
|
)
|
|
|
|
|
|
def create_total_pnl_histogram(pair_theo_ret: pd.DataFrame) -> Any:
|
|
"""Create a Plotly histogram of total theoretical return with automatic bins."""
|
|
import plotly.express as px
|
|
|
|
pair_theo_ret_for_plot = format_pair_names_for_display(add_total_pnl(pair_theo_ret))
|
|
hover_columns = [
|
|
column
|
|
for column in ("pair_name", "mr_ranking", "realized_pnl", "unrealized_pnl")
|
|
if column in pair_theo_ret_for_plot.columns
|
|
]
|
|
total_pnl_histogram = px.histogram(
|
|
pair_theo_ret_for_plot,
|
|
x="total_pnl",
|
|
labels={
|
|
"total_pnl": "Total TheoRet (%)",
|
|
"count": "Pair count",
|
|
},
|
|
title="Total TheoRet Distribution by Pair",
|
|
hover_data=hover_columns,
|
|
)
|
|
total_pnl_histogram.update_layout(
|
|
bargap=0.05,
|
|
yaxis_title="Pair count",
|
|
)
|
|
return total_pnl_histogram
|
|
|
|
|
|
def show_interactive_dataframe(
|
|
dataframe: pd.DataFrame,
|
|
*,
|
|
table_id: str | None = None,
|
|
**kwargs: Any,
|
|
) -> None:
|
|
"""Render a dataframe as an interactive sortable notebook grid."""
|
|
from itables import show
|
|
|
|
options = {
|
|
"paging": True,
|
|
"pageLength": 25,
|
|
"scrollX": True,
|
|
"ordering": True,
|
|
"showIndex": False,
|
|
"maxBytes": "8MB",
|
|
"classes": "display compact stripe hover",
|
|
"css": INTERACTIVE_TABLE_CSS,
|
|
}
|
|
if table_id is not None:
|
|
options["table_id"] = table_id
|
|
options.update(kwargs)
|
|
show(dataframe, **options)
|
|
|
|
|
|
def sorted_pair_names(selector_pair_rankings: pd.DataFrame) -> list[str]:
|
|
"""Return unique pair names sorted alphabetically for pair-level analysis."""
|
|
if "pair_name" not in selector_pair_rankings.columns:
|
|
raise ValueError("selector pair rankings missing column: pair_name")
|
|
return sorted(
|
|
{
|
|
str(pair_name)
|
|
for pair_name in selector_pair_rankings["pair_name"].dropna()
|
|
if str(pair_name)
|
|
}
|
|
)
|
|
|
|
|
|
def create_pair_name_dropdown(selector_pair_rankings: pd.DataFrame) -> Any:
|
|
"""Create a dropdown for choosing one pair name from ranked pairs."""
|
|
import ipywidgets as widgets
|
|
|
|
pair_names = sorted_pair_names(selector_pair_rankings)
|
|
if not pair_names:
|
|
raise ValueError("selector pair rankings do not contain any pair names")
|
|
options = [
|
|
(format_pair_name_for_display(pair_name), pair_name) for pair_name in pair_names
|
|
]
|
|
return widgets.Dropdown(
|
|
options=options,
|
|
value=pair_names[0],
|
|
description="Pair",
|
|
layout=widgets.Layout(width="100%"),
|
|
style={"description_width": "90px"},
|
|
)
|