History Collection (SQLite)¶

def append_dataframe(
    conn: sqlite3.Connection,
    frame: pd.DataFrame,
    table_name: str,
    if_exists: IfExists,
) -> bool:
    """Append a DataFrame to SQLite when it has a schema.

    Returns:
        True if a table was written, False if the frame had no columns.
    """
    if len(frame.columns) == 0:
        logger.warning("Skipping %s: dataset returned no columns", table_name)
        return False
    frame.to_sql(  # type: ignore[reportUnknownMemberType]
        table_name,
        conn,
        if_exists=if_exists.value,
        index=False,
        chunksize=50_000,
    )
    return True

def augment_written_columns_from_sqlite(
    conn: sqlite3.Connection,
    datasets: set[Dataset],
    written_columns: dict[Dataset, set[str]],
) -> None:
    """Add existing table columns to the written column map."""
    for dataset in datasets:
        columns = get_table_columns(conn, dataset.table_name)
        if not columns:
            continue
        if dataset in written_columns:
            written_columns[dataset].update(columns)
        else:
            written_columns[dataset] = columns

def build_rate_targets(
    symbols: Sequence[str],
    timeframes: Sequence[int | str],
    *,
    allow_missing_symbol: bool = False,
) -> list[RateTarget]:
    """Build rate targets for every symbol and timeframe combination.

    Args:
        symbols: MT5 symbol names. May be empty when ``allow_missing_symbol``.
        timeframes: MT5 timeframes as integers or names (for example ``M1``).
        allow_missing_symbol: When True and ``symbols`` is empty, build targets
            with ``symbol=None`` for each timeframe instead of raising.

    Returns:
        Targets in row-major order: every timeframe for the first symbol, then
        every timeframe for the next symbol, and so on.

    Raises:
        ValueError: If ``timeframes`` is empty, or ``symbols`` is empty and
            ``allow_missing_symbol`` is False.
    """
    if not timeframes:
        msg = "At least one timeframe is required."
        raise ValueError(msg)
    if not symbols:
        if not allow_missing_symbol:
            msg = "At least one symbol is required."
            raise ValueError(msg)
        return [RateTarget(symbol=None, timeframe=tf) for tf in timeframes]
    return [
        RateTarget(symbol=symbol, timeframe=tf)
        for symbol in symbols
        for tf in timeframes
    ]

def build_rate_view_name(
    *,
    symbol: str,
    granularity: str,
    granularity_count: int,
    timeframe: int,
) -> str:
    """Return a collision-free offline optimize view name.

    View names always include the timeframe integer after a ``__`` separator so
    a symbol such as ``EURUSD_M1`` cannot collide with ``EURUSD`` at timeframe
    ``M1``.
    """
    if granularity_count == 1:
        return f"rate_{symbol}__{timeframe}"
    return f"rate_{symbol}__{granularity}_{timeframe}"

def create_cash_events_view(
    conn: sqlite3.Connection,
    deals_columns: set[str],
) -> bool:
    """Create the cash_events SQLite view derived from history_deals.

    Returns:
        True if the view was created, False if required columns are missing.
    """
    if "type" not in deals_columns:
        logger.warning("Skipping cash_events view: history_deals.type is missing")
        return False
    conn.execute("DROP VIEW IF EXISTS cash_events")
    conn.execute(
        "CREATE VIEW cash_events AS"  # noqa: S608
        f" SELECT * FROM history_deals WHERE type NOT IN {_TRADE_DEAL_TYPES_SQL}",
    )
    return True

def create_history_indexes(
    conn: sqlite3.Connection,
    written_columns: dict[Dataset, set[str]],
) -> None:
    """Create useful indexes for collected history tables when present."""
    if {"symbol", "timeframe", "time"}.issubset(
        written_columns.get(Dataset.rates, set()),
    ):
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_rates_symbol_timeframe_time"
            " ON rates(symbol, timeframe, time)",
        )
    if {"symbol", "time"}.issubset(written_columns.get(Dataset.ticks, set())):
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_ticks_symbol_time ON ticks(symbol, time)",
        )
    if {"position_id", "symbol"}.issubset(
        written_columns.get(Dataset.history_deals, set()),
    ):
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_history_deals_position_symbol"
            " ON history_deals(position_id, symbol)",
        )

def create_positions_reconstructed_view(
    conn: sqlite3.Connection,
    deals_columns: set[str],
) -> bool:
    """Create the positions_reconstructed SQLite view derived from history_deals.

    Returns:
        True if the view was created, False if required columns are missing.
    """
    if not _POSITIONS_VIEW_REQUIRED_COLUMNS.issubset(deals_columns):
        missing = ", ".join(sorted(_POSITIONS_VIEW_REQUIRED_COLUMNS - deals_columns))
        logger.warning(
            "Skipping positions_reconstructed view: history_deals missing columns: %s",
            missing,
        )
        return False
    conn.execute("DROP VIEW IF EXISTS positions_reconstructed")
    conn.execute(
        "CREATE VIEW positions_reconstructed AS"  # noqa: S608
        " SELECT"
        " position_id,"
        " symbol,"
        " MIN(CASE WHEN entry = 0 THEN time END) AS open_time,"
        " MAX(CASE WHEN entry IN (1, 2, 3) THEN time END) AS close_time,"
        " MIN(CASE WHEN entry = 0 THEN type END) AS direction,"
        " SUM(CASE WHEN entry = 0 THEN volume ELSE 0 END) AS volume_open,"
        " SUM(CASE WHEN entry IN (1, 2, 3) THEN volume ELSE 0 END) AS volume_close,"
        " SUM(CASE WHEN entry = 2 THEN volume ELSE 0 END) AS volume_reversal,"
        " CASE"
        " WHEN SUM(CASE WHEN entry = 0 THEN volume ELSE 0 END) > 0"
        " THEN SUM(CASE WHEN entry = 0 THEN price * volume ELSE 0 END)"
        " / SUM(CASE WHEN entry = 0 THEN volume ELSE 0 END)"
        " END AS open_price,"
        " CASE"
        " WHEN SUM(CASE WHEN entry IN (1, 2, 3) THEN volume ELSE 0 END) > 0"
        " THEN SUM(CASE WHEN entry IN (1, 2, 3) THEN price * volume ELSE 0 END)"
        " / SUM(CASE WHEN entry IN (1, 2, 3) THEN volume ELSE 0 END)"
        " END AS close_price,"
        " SUM(profit) AS total_profit,"
        " SUM(CASE WHEN entry = 2 THEN 1 ELSE 0 END) AS reversal_count,"
        " COUNT(*) AS deals_count"
        " FROM history_deals"
        f" WHERE type IN {_TRADE_DEAL_TYPES_SQL} AND position_id != 0"
        " GROUP BY position_id, symbol"
        " HAVING SUM(CASE WHEN entry IN (1, 2, 3) THEN 1 ELSE 0 END) > 0",
    )
    return True

def create_rate_compatibility_views(conn: sqlite3.Connection) -> None:
    """Create rate compatibility views from the normalized rates table."""
    columns = get_table_columns(conn, Dataset.rates.table_name)
    if not {"symbol", "timeframe", "time"}.issubset(columns):
        return
    drop_rate_compatibility_views(conn)
    select_columns = sorted(columns - {"symbol", "timeframe"})
    quoted_columns = ", ".join(f'"{column}"' for column in select_columns)
    rows = conn.execute(
        "SELECT DISTINCT symbol, timeframe FROM rates ORDER BY symbol, timeframe",
    ).fetchall()
    timeframes_by_symbol: dict[str, list[int]] = {}
    for symbol, timeframe in rows:
        timeframes_by_symbol.setdefault(str(symbol), []).append(int(timeframe))
    for symbol, timeframes in timeframes_by_symbol.items():
        for timeframe in timeframes:
            granularity = resolve_granularity_name(timeframe)
            view_name = build_rate_view_name(
                symbol=symbol,
                granularity=granularity,
                granularity_count=len(timeframes),
                timeframe=timeframe,
            )
            quoted_view_name = quote_sqlite_identifier(view_name)
            escaped_symbol = symbol.replace("'", "''")
            conn.execute(
                f"CREATE VIEW {quoted_view_name} AS"  # noqa: S608
                f" SELECT {quoted_columns} FROM rates"
                f" WHERE symbol = '{escaped_symbol}'"
                f" AND timeframe = {timeframe}",
            )

def deduplicate_history_tables(
    conn: sqlite3.Connection,
    written_columns: dict[Dataset, set[str]],
    written_tables: set[Dataset],
    dedup_scopes: Mapping[Dataset, Sequence[DedupScope]] | None = None,
) -> None:
    """Deduplicate appended history tables by stable identifiers.

    Scopes whose required columns are not present in the written table are
    skipped. If all scopes for a dataset are skipped, the table receives one
    unscoped deduplication pass instead.
    """
    cursor = conn.cursor()
    for dataset in written_tables:
        columns = written_columns.get(dataset, set())
        table = dataset.table_name
        keys = next(
            (
                candidate
                for candidate in _HISTORY_DEDUP_KEYS[dataset]
                if set(candidate).issubset(columns)
            ),
            None,
        )
        if keys is None:
            logger.warning(
                "Skipping %s deduplication: no supported key columns",
                table,
            )
            continue
        raw_scopes: Sequence[DedupScope] = (
            dedup_scopes.get(dataset, ()) if dedup_scopes else ()
        )
        scopes = [scope for scope in raw_scopes if scope.required_columns <= columns]
        if scopes:
            for scope in scopes:
                drop_duplicates_in_table(
                    cursor,
                    table,
                    list(keys),
                    keep="last",
                    scope_where=scope.where,
                    scope_params=scope.params,
                )
            continue
        drop_duplicates_in_table(cursor, table, list(keys), keep="last")

def drop_duplicates_in_table(
    cursor: sqlite3.Cursor,
    table: str,
    ids: list[str],
    *,
    keep: Literal["first", "last"] = "last",
    scope_where: str | None = None,
    scope_params: tuple[object, ...] = (),
) -> None:
    """Remove duplicate rows, keeping the first or last ROWID per key group.

    Raises:
        ValueError: If the table or column names are invalid.
    """
    if not table.isidentifier():
        msg = f"Invalid table name: {table}"
        raise ValueError(msg)
    if invalid := {column for column in ids if not column.isidentifier()}:
        msg = f"Invalid column names: {', '.join(sorted(invalid))}"
        raise ValueError(msg)
    ids_csv = ", ".join(f'"{column}"' for column in ids)
    rowid_selector = "MIN" if keep == "first" else "MAX"
    if scope_where:
        delete_sql = (
            f"DELETE FROM {table} WHERE {scope_where} AND ROWID NOT IN"  # noqa: S608
            f" (SELECT {rowid_selector}(ROWID) FROM {table} WHERE {scope_where}"
            f" GROUP BY {ids_csv})"
        )
        cursor.execute(delete_sql, scope_params + scope_params)
        return
    cursor.execute(
        f"DELETE FROM {table} WHERE ROWID NOT IN"  # noqa: S608
        f" (SELECT {rowid_selector}(ROWID) FROM {table} GROUP BY {ids_csv})",
    )

def drop_forming_rate_bar(df_rate: pd.DataFrame) -> pd.DataFrame:
    """Return closed bars from chronologically ordered MT5 rate data.

    MetaTrader 5 ``copy_rates_from_pos(start_pos=0)`` includes the still-forming
    current bar as the last row. Slice it off so downstream logic only sees
    completed bars. Empty frames and single-row frames return empty results.

    Args:
        df_rate: Rate data ordered oldest-to-newest with the forming bar last.

    Returns:
        A new DataFrame with all rows except the last. Index and columns are
        preserved. The input frame is not modified.
    """
    return df_rate.iloc[:-1].copy()

def drop_rate_compatibility_views(conn: sqlite3.Connection) -> None:
    """Drop all mt5cli-managed ``rate_*`` compatibility views."""
    rows = conn.execute(
        "SELECT name FROM sqlite_master WHERE type = 'view' AND name GLOB 'rate_*'",
    ).fetchall()
    for (view_name,) in rows:
        quoted_view_name = quote_sqlite_identifier(str(view_name))
        conn.execute(f"DROP VIEW IF EXISTS {quoted_view_name}")

def filter_incremental_history_deals_frame(
    frame: pd.DataFrame,
    symbols: Sequence[str],
    start_by_symbol: dict[str, datetime],
    account_event_start: datetime,
) -> pd.DataFrame:
    """Filter incrementally fetched history_deals by symbol and event start times.

    Returns:
        Rows for selected symbols at or after each symbol start, plus account
        events at or after ``account_event_start``.
    """
    if frame.empty:
        return frame.copy()
    parsed_times = _frame_parsed_times(frame)
    time_valid = parsed_times.notna()
    account_event_mask = _history_deals_account_event_mask(frame)
    account_keep = account_event_mask & (parsed_times >= account_event_start)
    trade_keep = pd.Series(data=False, index=frame.index)
    if "symbol" in frame.columns:
        for symbol in symbols:
            trade_keep |= (
                (frame["symbol"] == symbol)
                & (parsed_times >= start_by_symbol[symbol])
                & ~account_event_mask
            )
    keep = (account_keep | trade_keep) & time_valid
    return frame.loc[keep].copy()

def filter_trade_history_frame(
    frame: pd.DataFrame,
    symbols: Sequence[str],
    *,
    include_account_events: bool,
) -> pd.DataFrame:
    """Filter trade history rows to selected symbols and account events.

    Returns:
        Filtered history rows.
    """
    if "symbol" not in frame.columns:
        return frame
    symbol_mask = frame["symbol"].isin(symbols)
    if not include_account_events:
        return frame.loc[symbol_mask].copy()
    account_event_mask = _history_deals_account_event_mask(frame)
    return frame.loc[symbol_mask | account_event_mask].copy()

def get_history_deals_account_event_start_datetime(
    conn: sqlite3.Connection,
    *,
    fallback_start: datetime,
) -> datetime:
    """Return the next update start for account-level history_deals rows."""
    table = Dataset.history_deals.table_name
    columns = get_table_columns(conn, table)
    if "time" not in columns:
        return fallback_start
    if "type" in columns:
        where_clause = f"type NOT IN {_TRADE_DEAL_TYPES_SQL}"
    elif "symbol" in columns:
        where_clause = "symbol IS NULL OR symbol = ''"
    else:
        return fallback_start
    row = conn.execute(
        f"SELECT MAX(time) FROM {table} WHERE {where_clause}",  # noqa: S608
    ).fetchone()
    parsed = parse_sqlite_timestamp(row[0] if row else None)
    return parsed if parsed is not None else fallback_start

def get_incremental_start_datetime(
    conn: sqlite3.Connection,
    dataset: Dataset,
    *,
    symbol: str,
    timeframe: int | None,
    fallback_start: datetime,
) -> datetime:
    """Return the next update start datetime from existing MAX(time)."""
    timeframes = [timeframe] if timeframe is not None else None
    starts = load_incremental_start_datetimes(
        conn,
        dataset,
        symbols=[symbol],
        timeframes=timeframes,
        fallback_start=fallback_start,
    )
    return starts[symbol, timeframe]

def get_table_columns(conn: sqlite3.Connection, table: str) -> set[str]:
    """Return existing SQLite columns for a table."""
    quoted_table = quote_sqlite_identifier(table)
    rows = conn.execute(f"PRAGMA table_info({quoted_table})").fetchall()
    return {str(row[1]) for row in rows}

def load_incremental_start_datetimes(
    conn: sqlite3.Connection,
    dataset: Dataset,
    *,
    symbols: Sequence[str],
    timeframes: Sequence[int] | None = None,
    fallback_start: datetime,
) -> dict[tuple[str, int | None], datetime]:
    """Return next update start datetimes keyed by symbol and optional timeframe."""
    table = dataset.table_name
    columns = get_table_columns(conn, table)
    if dataset is Dataset.rates and columns:
        _validate_rates_schema(columns)

    if "time" not in columns:
        if dataset is Dataset.rates and timeframes is not None:
            return {
                (symbol, timeframe): fallback_start
                for symbol in symbols
                for timeframe in timeframes
            }
        return {(symbol, None): fallback_start for symbol in symbols}

    parsed_by_key: dict[tuple[str, int | None], datetime] = {}
    if (
        dataset is Dataset.rates
        and timeframes is not None
        and {"symbol", "timeframe"}.issubset(columns)
    ):
        symbol_placeholders = ", ".join("?" for _ in symbols)
        timeframe_placeholders = ", ".join("?" for _ in timeframes)
        grouped_rates_query = (
            "SELECT symbol, timeframe, MAX(time) FROM "  # noqa: S608
            f"{table} WHERE symbol IN ({symbol_placeholders})"
            f" AND timeframe IN ({timeframe_placeholders})"
            " GROUP BY symbol, timeframe"
        )
        rows = conn.execute(
            grouped_rates_query,
            [*symbols, *timeframes],
        ).fetchall()
        for row_symbol, row_timeframe, max_time in rows:
            parsed = parse_sqlite_timestamp(max_time)
            if parsed is not None:
                parsed_by_key[str(row_symbol), int(row_timeframe)] = parsed
        return {
            (symbol, timeframe): parsed_by_key.get(
                (symbol, timeframe),
                fallback_start,
            )
            for symbol in symbols
            for timeframe in timeframes
        }

    if "symbol" in columns:
        symbol_placeholders = ", ".join("?" for _ in symbols)
        rows = conn.execute(
            f"SELECT symbol, MAX(time) FROM {table}"  # noqa: S608
            f" WHERE symbol IN ({symbol_placeholders}) GROUP BY symbol",
            list(symbols),
        ).fetchall()
        for row_symbol, max_time in rows:
            parsed = parse_sqlite_timestamp(max_time)
            if parsed is not None:
                parsed_by_key[str(row_symbol), None] = parsed
        return {
            (symbol, None): parsed_by_key.get((symbol, None), fallback_start)
            for symbol in symbols
        }

    row = conn.execute(f"SELECT MAX(time) FROM {table}").fetchone()  # noqa: S608
    parsed = parse_sqlite_timestamp(row[0] if row else None)
    shared_start = parsed if parsed is not None else fallback_start
    return {(symbol, None): shared_start for symbol in symbols}

def load_rate_data(
    conn_or_path: SqliteConnOrPath,
    table: str,
    count: int | None = None,
) -> pd.DataFrame:
    """Load rate-like data from a SQLite database path or connection.

    Args:
        conn_or_path: SQLite database path or open connection.
        table: Source table or view name.
        count: Optional number of most recent rows to load.

    Returns:
        DataFrame indexed by ascending ``time``.

    """
    conn, should_close = _open_existing_sqlite_database(conn_or_path)
    try:
        return load_rate_data_from_connection(conn, table, count=count)
    finally:
        if should_close:
            conn.close()

def load_rate_data_from_connection(
    connection: sqlite3.Connection,
    table: str,
    count: int | None = None,
) -> pd.DataFrame:
    """Load rate-like data from a SQLite table or view.

    Args:
        connection: Open SQLite connection.
        table: Source table or view name.
        count: Optional number of most recent rows to load.

    Returns:
        DataFrame indexed by ascending ``time``.

    Raises:
        ValueError: If inputs, schema, timestamps are invalid, or the table
            or view contains no rows.
    """
    table_name = _validate_rate_load_request(table, count)
    columns = get_table_columns(connection, table_name)
    _ensure_rate_columns(columns, table_name)
    quoted_table = quote_sqlite_identifier(table_name)
    if count is None:
        frame = cast(
            "pd.DataFrame",
            pd.read_sql_query(  # type: ignore[reportUnknownMemberType]
                f"SELECT * FROM {quoted_table} ORDER BY time ASC",  # noqa: S608
                connection,
            ),
        )
    else:
        frame = cast(
            "pd.DataFrame",
            pd.read_sql_query(  # type: ignore[reportUnknownMemberType]
                f"SELECT * FROM {quoted_table} ORDER BY time DESC LIMIT ?",  # noqa: S608
                connection,
                params=(count,),
            ),
        )
    if frame.empty:
        msg = f"SQLite table or view {table_name!r} contains no rows."
        raise ValueError(msg)
    return _parse_rate_time_index(frame, table_name)

def load_rate_series_by_granularity(
    conn_or_path: SqliteConnOrPath,
    symbols: Sequence[str],
    granularities: Sequence[int | str],
    count: int,
    *,
    explicit_tables: Sequence[str] | None = None,
    allow_missing_symbol: bool = False,
) -> dict[tuple[str | None, str], pd.DataFrame]:
    """Load rate series keyed by symbol and string granularity name.

    Builds targets with :func:`build_rate_targets` and loads them with
    :func:`load_rate_series_from_sqlite`, then rekeys the result by granularity
    name (for example ``M1``) instead of the integer timeframe to reduce
    downstream boilerplate.

    Args:
        conn_or_path: SQLite database path or open connection.
        symbols: MT5 symbol names. May be empty when ``allow_missing_symbol``.
        granularities: MT5 timeframes as integers or names (for example ``M1``).
        count: Number of most recent rows to load per series.
        explicit_tables: Optional explicit table or view names matching the
            built targets in row-major order. Required when symbols are omitted.
        allow_missing_symbol: When True and ``symbols`` is empty, build targets
            with ``symbol=None`` for each granularity instead of raising.

    Returns:
        Mapping keyed by ``(symbol | None, granularity_name)`` to each rate
        DataFrame. Propagates ``ValueError`` (via :func:`build_rate_targets` and
        :func:`load_rate_series_from_sqlite`) when inputs are empty or invalid,
        table resolution fails, or duplicate targets are present.
    """
    targets = build_rate_targets(
        symbols,
        granularities,
        allow_missing_symbol=allow_missing_symbol,
    )
    series = load_rate_series_from_sqlite(
        conn_or_path,
        targets,
        count,
        explicit_tables=explicit_tables,
    )
    return {
        (symbol, resolve_granularity_name(timeframe)): frame
        for (symbol, timeframe), frame in series.items()
    }

def load_rate_series_from_sqlite(
    conn_or_path: SqliteConnOrPath,
    targets: Sequence[RateTarget] | None = None,
    count: int | None = None,
    explicit_tables: Sequence[str] | None = None,
    *,
    table: str | None = None,
) -> dict[tuple[str | None, int], pd.DataFrame] | pd.DataFrame:
    """Load one table/view or multiple rate series from a SQLite database.

    Args:
        conn_or_path: SQLite database path or open connection.
        targets: Rate targets to load. Each ``(symbol, timeframe_int)`` pair must
            be unique. Omit when loading a single explicit ``table``.
        count: Optional number of most recent rows to load per series.
        explicit_tables: Optional explicit table or view names matching targets.
            When omitted, managed ``rate_*`` compatibility views must already
            exist in the database.
        table: Optional single table or view name to load directly.

    Returns:
        A DataFrame when ``table`` is provided, otherwise a mapping keyed by
        ``(symbol, timeframe_int)`` to each rate DataFrame.

    Raises:
        ValueError: If ``count`` is not positive, targets are empty, duplicate
            ``(symbol, timeframe_int)`` pairs are present, or table resolution
            fails.
    """
    if table is not None:
        return load_rate_data(conn_or_path, table, count=count)
    if count is None or count <= 0:
        msg = "count must be positive."
        raise ValueError(msg)
    if targets is None:
        msg = "targets are required when table is not provided."
        raise ValueError(msg)
    target_list = list(targets)
    if not target_list:
        msg = "At least one rate target is required."
        raise ValueError(msg)
    if explicit_tables is None and any(target.symbol is None for target in target_list):
        msg = (
            "Cannot resolve a rate table for a target without a symbol; "
            "provide explicit_tables."
        )
        raise ValueError(msg)
    seen_keys: set[tuple[str | None, int]] = set()
    for target in target_list:
        key = (target.symbol, target.timeframe_int)
        if key in seen_keys:
            symbol_repr = repr(target.symbol)
            msg = f"Duplicate rate target: ({symbol_repr}, {target.timeframe_int})"
            raise ValueError(msg)
        seen_keys.add(key)
    tables = (
        resolve_rate_tables(None, target_list, explicit_tables)
        if explicit_tables is not None
        else None
    )
    conn, should_close = _open_existing_sqlite_database(conn_or_path)
    try:
        resolved_tables = tables or resolve_rate_tables(
            conn,
            target_list,
            require_existing=True,
        )
        return {
            (target.symbol, target.timeframe_int): load_rate_data_from_connection(
                conn,
                table,
                count=count,
            )
            for target, table in zip(target_list, resolved_tables, strict=True)
        }
    finally:
        if should_close:
            conn.close()

def parse_sqlite_timestamp(value: object) -> datetime | None:
    """Parse a SQLite history timestamp value.

    Returns:
        Parsed timezone-aware datetime, or None when parsing fails.
    """
    if value is None:
        return None
    if isinstance(value, datetime):
        return value if value.tzinfo is not None else value.replace(tzinfo=UTC)
    if isinstance(value, int | float):
        return datetime.fromtimestamp(float(value), tz=UTC)
    if isinstance(value, str):
        return _parse_string_sqlite_timestamp(value)
    logger.warning("Ignoring unsupported history timestamp type: %s", type(value))
    return None

def quote_sqlite_identifier(identifier: str) -> str:
    """Return a safely quoted SQLite identifier using double quotes."""
    return '"' + identifier.replace('"', '""') + '"'

def record_written_columns(
    written_columns: dict[Dataset, set[str]],
    dataset: Dataset,
    frame: pd.DataFrame,
) -> None:
    """Remember columns for datasets written during collection."""
    columns = set(frame.columns)
    if dataset in written_columns:
        written_columns[dataset].update(columns)
    else:
        written_columns[dataset] = columns