Storage¶

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 collect_history(
    output: Path,
    symbols: list[str],
    date_from: datetime | str,
    date_to: datetime | str,
    *,
    datasets: set[Dataset] | None = None,
    timeframe: int | str = 1,
    flags: int | str = "ALL",
    if_exists: IfExists = IfExists.FAIL,
    with_views: bool = False,
    config: Mt5Config | None = None,
) -> None:
    """Collect historical datasets into a single SQLite database.

    Args:
        output: SQLite database path.
        symbols: Symbols to collect.
        date_from: Start date.
        date_to: End date.
        datasets: Datasets to include (defaults to all).
        timeframe: Rates timeframe as integer or name (e.g. ``M1``).
        flags: Tick copy flags as integer or name (e.g. ``ALL``).
        if_exists: Behavior when a target table already exists.
        with_views: Create ``cash_events`` and ``positions_reconstructed`` views.
        config: MT5 connection configuration.
    """
    start = _require_datetime(date_from)
    end = _require_datetime(date_to)
    selected = datasets if datasets is not None else set(Dataset)
    tf = _coerce_timeframe(timeframe)
    tick_flags = _coerce_tick_flags(flags)
    mt5_config = config or build_config()
    with connected_client(mt5_config) as client, sqlite3.connect(output) as conn:
        conn.execute("PRAGMA journal_mode=WAL")
        conn.execute("PRAGMA synchronous=NORMAL")
        written_tables, written_columns = write_collected_datasets(
            conn,
            client,
            symbols,
            selected,
            tf,
            tick_flags,
            start,
            end,
            if_exists,
        )
        create_history_indexes(conn, written_columns)
        if with_views and Dataset.history_deals in written_tables:
            create_cash_events_view(conn, written_columns[Dataset.history_deals])
            create_positions_reconstructed_view(
                conn,
                written_columns[Dataset.history_deals],
            )
        elif with_views:
            logger.warning(
                "--with-views ignored: history_deals table was not written",
            )
    logger.info(
        "Collected %s for %d symbol(s) into %s",
        ", ".join(sorted(ds.value for ds in selected)),
        len(symbols),
        output,
    )

def detect_format(
    output_path: Path,
    explicit_format: str | None = None,
) -> str:
    """Detect the output format from a file extension or explicit format string.

    Args:
        output_path: Path to the output file.
        explicit_format: Explicitly specified format, if any.

    Returns:
        The detected format string.

    Raises:
        ValueError: If the format cannot be determined.
    """
    if explicit_format is not None:
        return explicit_format
    suffix = output_path.suffix.lower()
    if suffix in _FORMAT_EXTENSIONS:
        return _FORMAT_EXTENSIONS[suffix]
    msg = (
        f"Cannot detect format from extension '{suffix}'."
        " Use --format to specify the output format."
    )
    raise ValueError(msg)

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 export_dataframe(
    df: pd.DataFrame,
    output_path: Path,
    output_format: str,
    table_name: str = "data",
) -> None:
    """Export a pandas DataFrame to the specified file format.

    Args:
        df: DataFrame to export.
        output_path: Path to the output file.
        output_format: Output format (csv, json, parquet, or sqlite3).
        table_name: Table name for SQLite3 output.

    Raises:
        ValueError: If the output format is not supported.
    """
    if output_format == "csv":
        df.to_csv(output_path, index=False)
    elif output_format == "json":
        df.to_json(
            output_path,
            orient="records",
            date_format="iso",
            indent=2,
        )
    elif output_format == "parquet":
        df.to_parquet(output_path, index=False)
    elif output_format == "sqlite3":
        export_dataframe_to_sqlite(
            df,
            output_path,
            table_name,
            if_exists=IfExists.REPLACE,
            index=False,
        )
    else:
        msg = f"Unsupported output format: {output_format}"
        raise ValueError(msg)

def export_dataframe_to_sqlite(
    df: pd.DataFrame,
    output_path: Path,
    table_name: str = "data",
    *,
    if_exists: IfExists = IfExists.APPEND,
    index: bool = False,
    index_label: str | None = None,
    deduplicate_on: Sequence[str] | None = None,
) -> None:
    """Write a DataFrame to SQLite with configurable append and deduplication.

    Args:
        df: DataFrame to export.
        output_path: SQLite database path.
        table_name: Target table name.
        if_exists: Conflict behavior when the table already exists.
        index: Whether to write the DataFrame index as a column.
        index_label: Column name for the index when ``index=True``.
        deduplicate_on: Optional key columns to deduplicate after writing,
            keeping the latest ``ROWID`` per key group. Deduplication scans the
            full table, so repeated appends cost O(table size); index the key
            columns when appending frequently.
    """
    with sqlite3.connect(output_path) as conn:
        df.to_sql(  # type: ignore[reportUnknownMemberType]
            table_name,
            conn,
            if_exists=if_exists.value,
            index=index,
            index_label=index_label,
        )
        if deduplicate_on:
            from .history import drop_duplicates_in_table  # noqa: PLC0415

            drop_duplicates_in_table(
                conn.cursor(),
                table_name,
                list(deduplicate_on),
                keep="last",
            )
            conn.commit()

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 resolve_rate_tables(
    conn_or_path: SqliteConnOrPath | None,
    targets: Sequence[RateTarget],
    explicit_tables: Sequence[str] | None = None,
    *,
    require_existing: bool = False,
) -> list[str]:
    """Resolve SQLite table or view names for rate targets.

    Args:
        conn_or_path: SQLite database path or open connection. May be None when
            ``explicit_tables`` is provided, or when ``require_existing`` is
            False and deterministic default view names are sufficient.
        targets: Rate targets to resolve.
        explicit_tables: Optional explicit table or view names. When provided,
            they are used as-is and must match the number of targets.
        require_existing: When True, require the database and managed views to
            exist for each symbol target. Ignored when ``explicit_tables`` is
            provided.

    Returns:
        Table or view names aligned with ``targets``.

    Raises:
        ValueError: If ``targets`` is empty, ``explicit_tables`` length does not
            match the target count, a target without a symbol is resolved
            without an explicit table, or ``require_existing`` is True and the
            database or a managed view is missing.
    """
    target_list = list(targets)
    if not target_list:
        msg = "At least one rate target is required."
        raise ValueError(msg)
    if explicit_tables is not None:
        tables = list(explicit_tables)
        if len(tables) != len(target_list):
            msg = (
                f"Expected {len(target_list)} explicit table(s) "
                f"to match the targets, got {len(tables)}."
            )
            raise ValueError(msg)
        return tables
    if 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)
    conn, should_close = _open_history_connection(conn_or_path)
    try:
        if conn is None:
            if require_existing:
                path = (
                    conn_or_path
                    if isinstance(conn_or_path, (Path, str))
                    else "database"
                )
                msg = f"SQLite database not found: {path}"
                raise ValueError(msg)
            timeframe_counts = None
            existing_views: set[str] = set()
        else:
            timeframe_counts = _load_rates_timeframe_counts(conn)
            existing_views = _load_existing_rate_views(conn)
        resolved: list[str] = []
        for target in target_list:
            symbol = cast("str", target.symbol)
            timeframe = target.timeframe_int
            resolved.append(
                _resolve_rate_view_name_from_context(
                    symbol=symbol,
                    timeframe=timeframe,
                    granularity_name=resolve_granularity_name(timeframe),
                    timeframe_counts=timeframe_counts,
                    existing_views=existing_views,
                    require_existing=require_existing,
                ),
            )
        return resolved
    finally:
        if should_close and conn is not None:
            conn.close()

def resolve_rate_view_name(
    conn_or_path: SqliteConnOrPath | None,
    symbol: str,
    granularity: str,
    *,
    require_existing: bool = False,
) -> str:
    """Resolve the mt5cli-managed rate compatibility view name.

    Args:
        conn_or_path: SQLite database path or open connection. When None or a
            non-existing path and ``require_existing`` is False, the deterministic
            default view name is returned without creating a database file.
        symbol: Symbol stored in the normalized ``rates`` table.
        granularity: Timeframe name (for example ``M1``) or integer string.
        require_existing: When True, require the database and a managed view to exist.

    Returns:
        View name such as ``rate_EURUSD__1`` or ``rate_EURUSD__M1_1``.

    Raises:
        ValueError: If ``require_existing`` is True and the database or view is missing.
    """
    timeframe = parse_timeframe(granularity)
    granularity_name = resolve_granularity_name(timeframe)
    conn, should_close = _open_history_connection(conn_or_path)
    try:
        if conn is None:
            if require_existing:
                path = (
                    conn_or_path
                    if isinstance(conn_or_path, (Path, str))
                    else "database"
                )
                msg = f"SQLite database not found: {path}"
                raise ValueError(msg)
            return build_rate_view_name(
                symbol=symbol,
                granularity=granularity_name,
                granularity_count=1,
                timeframe=timeframe,
            )
        return _resolve_rate_view_name_from_context(
            symbol=symbol,
            timeframe=timeframe,
            granularity_name=granularity_name,
            timeframe_counts=_load_rates_timeframe_counts(conn),
            existing_views=_load_existing_rate_views(conn),
            require_existing=require_existing,
        )
    finally:
        if should_close and conn is not None:
            conn.close()

def resolve_rate_view_names(
    conn_or_path: SqliteConnOrPath | None,
    symbols: Sequence[str],
    granularities: Sequence[str],
    *,
    require_existing: bool = False,
) -> list[str]:
    """Resolve rate compatibility view names for symbol and granularity pairs.

    Args:
        conn_or_path: SQLite database path or open connection. When None or a
            non-existing path and ``require_existing`` is False, deterministic
            default view names are returned without creating a database file.
        symbols: Symbols stored in the normalized ``rates`` table.
        granularities: Timeframe names (for example ``M1``) or integer strings.
        require_existing: When True, require the database and managed views to exist.

    Returns:
        View names in row-major order: every ``granularity`` for the first
        symbol, then every granularity for the next symbol, and so on.
    """
    conn, should_close = _open_history_connection(conn_or_path)
    try:
        if conn is None:
            return [
                resolve_rate_view_name(
                    conn_or_path,
                    symbol,
                    granularity,
                    require_existing=require_existing,
                )
                for symbol in symbols
                for granularity in granularities
            ]
        timeframe_counts = _load_rates_timeframe_counts(conn)
        existing_views = _load_existing_rate_views(conn)
        resolved: list[str] = []
        for symbol in symbols:
            for granularity in granularities:
                timeframe = parse_timeframe(granularity)
                resolved.append(
                    _resolve_rate_view_name_from_context(
                        symbol=symbol,
                        timeframe=timeframe,
                        granularity_name=resolve_granularity_name(timeframe),
                        timeframe_counts=timeframe_counts,
                        existing_views=existing_views,
                        require_existing=require_existing,
                    ),
                )
        return resolved
    finally:
        if should_close and conn is not None:
            conn.close()

def update_history(  # noqa: PLR0913
    *,
    client: Mt5DataClient,
    output: Path | str,
    symbols: Sequence[str],
    datasets: set[Dataset] | None = None,
    timeframes: Sequence[int | str] | None = None,
    flags: int | str = "ALL",
    lookback_hours: float = 24.0,
    date_to: datetime | str | None = None,
    deduplicate: bool = True,
    create_rate_views: bool = True,
    with_views: bool = False,
    include_account_events: bool = True,
) -> None:
    """Incrementally append MT5 history into a SQLite database.

    Uses an already-connected ``Mt5DataClient`` and does not create or close
    the MT5 connection. For first-time tables, data is fetched from
    ``date_to - lookback_hours``. Subsequent runs resume from existing
    ``MAX(time)`` per symbol (and timeframe for rates); when
    ``include_account_events=True``, account-level deals use a separate cursor
    over ``type NOT IN (0, 1)`` / empty-symbol rows.

    Args:
        client: Connected MT5 data client.
        output: SQLite database path.
        symbols: Symbols to update.
        datasets: Datasets to include (defaults to all).
        timeframes: Rate timeframes to update (defaults to all fixed MT5
            timeframes when None).
        flags: Tick copy flags as integer or name (e.g. ``ALL``).
        lookback_hours: First-run lookback when a table has no prior rows.
        date_to: Optional update end datetime. Defaults to now (UTC).
        deduplicate: Remove duplicate rows after append, keeping latest ROWID.
        create_rate_views: Create ``rate_<symbol>__<timeframe>`` views.
        with_views: Create ``cash_events`` and ``positions_reconstructed`` views.
        include_account_events: Include account-level cash events in
            ``history_deals`` when True.
    """
    request = _resolve_update_history_request(
        output=output,
        symbols=symbols,
        datasets=datasets,
        timeframes=timeframes,
        flags=flags,
        lookback_hours=lookback_hours,
        date_to=date_to,
    )
    if request is None:
        return
    logger.info(
        "Updating history in SQLite: symbols=%s, datasets=%s, path=%s",
        list(symbols),
        sorted(dataset.value for dataset in request.selected),
        request.output_path,
    )
    with sqlite3.connect(request.output_path) as conn:
        conn.execute("PRAGMA journal_mode=WAL")
        conn.execute("PRAGMA synchronous=NORMAL")
        write_incremental_datasets(
            conn,
            client,
            symbols,
            request.selected,
            request.resolved_timeframes,
            request.resolved_tick_flags,
            request.fallback_start,
            request.end,
            deduplicate=deduplicate,
            create_rate_views=create_rate_views,
            with_views=with_views,
            include_account_events=include_account_events,
        )