"""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")
TRADING_INSTRUCTIONS_COLUMNS = ("time_ns", "tstamp", "data")
OHLCV_1MIN_COLUMNS = ("tstamp", "tstamp_ns", "exch_acct", "instrument_id", "close")
INITIAL_THEO_CAPITAL_USD = 10_000.0
SQLITE_EXTENSIONS = {".db", ".sqlite", ".sqlite3"}
PAIR_NAME_DISPLAY_SUFFIX = ":USD"
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 validate_trading_instructions_table(conn: sqlite3.Connection) -> None:
    """Raise an actionable error if trading_instructions lacks required columns."""
    table_info = conn.execute("PRAGMA table_info(trading_instructions)").fetchall()
    if not table_info:
        raise ValueError("SQLite database is missing required table: trading_instructions")

    existing_columns = {row[1] for row in table_info}
    missing_columns = set(TRADING_INSTRUCTIONS_COLUMNS) - existing_columns
    if missing_columns:
        missing = ", ".join(sorted(missing_columns))
        raise ValueError(
            f"trading_instructions is missing required column(s): {missing}"
        )


def load_trading_instructions(conn: sqlite3.Connection) -> pd.DataFrame:
    """Load the full trading_instructions table ordered by timestamp."""
    validate_trading_instructions_table(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 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

    try:
        parsed = json.loads(raw_data)
    except (TypeError, json.JSONDecodeError):
        return None

    return parsed if isinstance(parsed, dict) else None


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)

    if "data" not in trd_inst_df.columns:
        raise ValueError("trading instructions dataframe is missing column: data")

    selected_instructions: list[dict[str, Any]] = []
    for instruction_row in _sort_trading_instructions(trd_inst_df).itertuples(
        index=False
    ):
        raw_data = getattr(instruction_row, "data")
        parsed = _parse_instruction_data(raw_data)
        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),
                "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,
                        "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,
                        "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",
        "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 1-minute close data from trading-day start for selected instruments."""
    validate_selector_pair_instrument_columns(conn)
    validate_ohlcv_1min_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 ohlcv_1min
            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"ohlcv_1min does not contain market 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(
            "ohlcv_1min 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"ohlcv_1min 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 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"},
    )