Grafana¶

def create_grafana_indexes(conn: sqlite3.Connection) -> None:
    """Create Grafana query performance indexes idempotently."""
    rates_cols = get_table_columns(conn, "rates")
    if {"time", "symbol", "timeframe"}.issubset(rates_cols):
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_rates_time_symbol_timeframe"
            ' ON "rates"("time", "symbol", "timeframe")',
        )

    ticks_cols = get_table_columns(conn, "ticks")
    if {"time", "symbol"}.issubset(ticks_cols):
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_ticks_time_symbol"
            ' ON "ticks"("time", "symbol")',
        )

    deals_cols = get_table_columns(conn, "history_deals")
    if {"time", "symbol"}.issubset(deals_cols):
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_history_deals_time_symbol"
            ' ON "history_deals"("time", "symbol")',
        )
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_history_deals_symbol_time"
            ' ON "history_deals"("symbol", "time")',
        )

    orders_cols = get_table_columns(conn, "history_orders")
    if {"time_setup", "symbol"}.issubset(orders_cols):
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_history_orders_time_setup_symbol"
            ' ON "history_orders"("time_setup", "symbol")',
        )

    if {"run_id", "login"}.issubset(get_table_columns(conn, "account_snapshots")):
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_account_snapshots_time_login"
            ' ON "account_snapshots"("run_id", "login")',
        )
    if {"run_id", "symbol"}.issubset(get_table_columns(conn, "position_snapshots")):
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_position_snapshots_time_symbol"
            ' ON "position_snapshots"("run_id", "symbol")',
        )
    if {"run_id", "symbol"}.issubset(get_table_columns(conn, "order_snapshots")):
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_order_snapshots_time_symbol"
            ' ON "order_snapshots"("run_id", "symbol")',
        )
    if {"observed_at", "status"}.issubset(get_table_columns(conn, "snapshot_runs")):
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_snapshot_runs_time_status"
            ' ON "snapshot_runs"("observed_at", "status")',
        )

def create_grafana_views(conn: sqlite3.Connection) -> None:
    """Create all Grafana-facing views idempotently.

    Missing source tables cause the affected view to be skipped with a warning;
    other views are unaffected. Stale views whose source table or required
    columns have disappeared are dropped before rebuild.
    """
    for name in _GRAFANA_VIEW_NAMES:
        conn.execute(f'DROP VIEW IF EXISTS "{name}"')
    _build_grafana_rates(conn)
    _build_grafana_ticks(conn)
    _build_grafana_history_deals(conn)
    _build_grafana_history_orders(conn)
    _build_grafana_trade_deals(conn)
    _build_grafana_cash_events(conn)
    _build_grafana_realized_pnl(conn)
    _build_grafana_symbol_pnl(conn)
    _build_grafana_trade_stats(conn)
    _build_grafana_account_snapshots(conn)
    _build_grafana_position_snapshots(conn)
    _build_grafana_order_snapshots(conn)
    _build_grafana_terminal_snapshots(conn)

def create_snapshot_tables(conn: sqlite3.Connection) -> None:
    """Create snapshot tables idempotently."""
    for ddl in _SNAPSHOT_TABLE_DDLS:
        conn.execute(ddl)

def ensure_grafana_schema(conn: sqlite3.Connection) -> None:
    """Create snapshot tables, Grafana views, and indexes idempotently."""
    create_snapshot_tables(conn)
    create_grafana_views(conn)
    create_grafana_indexes(conn)

def insert_account_snapshot(
    conn: sqlite3.Connection,
    run_id: int,
    row: dict[str, object],
) -> None:
    """Append one account state row to account_snapshots."""
    conn.execute(
        "INSERT INTO account_snapshots"
        " (run_id, login, currency, balance, equity,"
        "  margin, margin_free, margin_level, profit, leverage)"
        " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
        (
            run_id,
            row.get("login"),
            row.get("currency"),
            row.get("balance"),
            row.get("equity"),
            row.get("margin"),
            row.get("margin_free"),
            row.get("margin_level"),
            row.get("profit"),
            row.get("leverage"),
        ),
    )

def insert_order_snapshots(
    conn: sqlite3.Connection,
    run_id: int,
    login: int | None,
    rows: list[dict[str, object]],
) -> None:
    """Append order rows to order_snapshots; no-op when rows is empty."""
    if not rows:
        return
    conn.executemany(
        "INSERT INTO order_snapshots"
        " (run_id, login, ticket, symbol, type, volume_current,"
        "  price_open, price_current, state, comment, magic, time_setup)"
        " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
        [
            (
                run_id,
                login,
                r.get("ticket"),
                r.get("symbol"),
                r.get("type"),
                r.get("volume_current"),
                r.get("price_open"),
                r.get("price_current"),
                r.get("state"),
                r.get("comment"),
                r.get("magic"),
                _to_epoch_int(r.get("time_setup")),
            )
            for r in rows
        ],
    )

def insert_position_snapshots(
    conn: sqlite3.Connection,
    run_id: int,
    login: int | None,
    rows: list[dict[str, object]],
) -> None:
    """Append position rows to position_snapshots; no-op when rows is empty."""
    if not rows:
        return
    conn.executemany(
        "INSERT INTO position_snapshots"
        " (run_id, login, ticket, position_id, symbol, type, volume,"
        "  price_open, price_current, profit, swap, comment, magic)"
        " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
        [
            (
                run_id,
                login,
                r.get("ticket"),
                r.get("position_id"),
                r.get("symbol"),
                r.get("type"),
                r.get("volume"),
                r.get("price_open"),
                r.get("price_current"),
                r.get("profit"),
                r.get("swap"),
                r.get("comment"),
                r.get("magic"),
            )
            for r in rows
        ],
    )

def insert_terminal_snapshot(
    conn: sqlite3.Connection,
    run_id: int,
    row: dict[str, object],
) -> None:
    """Append one terminal state row to terminal_snapshots."""
    conn.execute(
        "INSERT INTO terminal_snapshots"
        " (run_id, name, connected, community_account,"
        "  trade_allowed, trade_expert, path, company, language)"
        " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
        (
            run_id,
            row.get("name"),
            row.get("connected"),
            row.get("community_account"),
            row.get("trade_allowed"),
            row.get("trade_expert"),
            row.get("path"),
            row.get("company"),
            row.get("language"),
        ),
    )

def publish_grafana_copy(
    source: str | Path,
    target: str | Path,
) -> Path:
    """Publish a consistent SQLite copy for Grafana using the backup API.

    Uses the SQLite online backup API for a WAL-safe, consistent snapshot of
    the source database. Writes to a temporary file beside the target, then
    atomically replaces it so that a previous published copy is preserved if
    publishing fails.

    Args:
        source: Path to the source SQLite database.
        target: Destination path for the published copy.

    Returns:
        The resolved absolute target path.

    Raises:
        FileNotFoundError: If the source database does not exist.
        ValueError: If source and target resolve to the same path.
    """
    source_path = Path(source)
    target_path = Path(target)
    if source_path.resolve() == target_path.resolve():
        msg = "--publish-copy target must differ from the source database: " + str(
            source_path
        )
        raise ValueError(msg)
    if not source_path.exists():
        raise FileNotFoundError(source_path)

    target_path.parent.mkdir(parents=True, exist_ok=True)

    tmp_fd, tmp_str = tempfile.mkstemp(
        dir=target_path.parent,
        suffix=".tmp",
        prefix=target_path.name + ".",
    )
    tmp_path = Path(tmp_str)
    try:
        os.close(tmp_fd)
        with (
            contextlib.closing(sqlite3.connect(source_path)) as src,
            contextlib.closing(sqlite3.connect(tmp_path)) as dst,
        ):
            src.backup(dst)
        try:
            target_mode = target_path.stat().st_mode & 0o777
        except FileNotFoundError:
            target_mode = 0o644
        Path(tmp_path).chmod(target_mode)
        tmp_path.replace(target_path)
    except Exception:
        with contextlib.suppress(OSError):
            tmp_path.unlink()
        raise

    logger.info("Published Grafana copy: %s -> %s", source_path, target_path)
    return target_path.resolve()

def record_snapshot_run(
    conn: sqlite3.Connection,
    run_id: int,
    status: str,
    detail: str | None = None,
) -> None:
    """Finalize a snapshot run by setting its status."""
    conn.execute(
        "UPDATE snapshot_runs SET status = ?, detail = ? WHERE run_id = ?",
        (status, detail, run_id),
    )

def start_snapshot_run(conn: sqlite3.Connection, observed_at: int) -> int:
    """Insert a snapshot_runs row with status 'running' and return its run_id.

    Returns:
        The auto-assigned run_id for the new row.
    """
    cursor = conn.execute(
        "INSERT INTO snapshot_runs (observed_at, status) VALUES (?, 'running')",
        (observed_at,),
    )
    return cast("int", cursor.lastrowid)