diff --git a/README.md b/README.md index 1271ccd..25e4fea 100644 --- a/README.md +++ b/README.md @@ -87,7 +87,46 @@ mt5cli -o history.db collect-history \ --timeframe M1 --flags ALL --if-exists append --with-views ``` -History orders and deals are fetched per symbol and concatenated, so the symbol filter is applied consistently across all datasets. The `cash_events` view is derived from symbol-filtered `history_deals`, so account-level cash events with empty or non-matching symbols may be excluded. The `rates` table records the requested `timeframe` so appended runs at different timeframes remain distinguishable. The `positions_reconstructed` view aggregates trade deals by `position_id`, excludes positions without closing deals, and uses volume-weighted open/close prices; reversal deals (`DEAL_ENTRY_INOUT`) are reported via `volume_reversal` / `reversal_count` columns and do not contribute to the weighted prices. +History orders and deals are fetched per symbol and concatenated, so the symbol filter is applied consistently across all datasets. The `cash_events` view is derived from symbol-filtered `history_deals`, so account-level cash events with empty or non-matching symbols may be excluded. The `rates` table records the requested `timeframe` so appended runs at different timeframes remain distinguishable. The `positions_reconstructed` view aggregates trade deals by `position_id`, excludes positions without closing-side entries, and uses volume-weighted open/close prices; reversal deals (`DEAL_ENTRY_INOUT`) are reported via `volume_reversal` / `reversal_count` columns. + +### Incremental history SDK + +For automated pipelines, use the importable incremental API instead of re-fetching fixed date ranges: + +```python +from pdmt5 import Mt5Config, Mt5DataClient +from mt5cli import Dataset, update_history, update_history_with_config + +# Reuse an already-connected pdmt5 client (does not open/close MT5) +client = Mt5DataClient(config=Mt5Config(login=12345)) +client.initialize_and_login_mt5() +try: + update_history( + client=client, + output="history.db", + symbols=["EURUSD", "GBPUSD"], + datasets={Dataset.rates, Dataset.history_deals}, + timeframes=["M1", "H1"], # default: all fixed MT5 timeframes + lookback_hours=24, + create_rate_views=True, + with_views=True, + include_account_events=True, + ) +finally: + client.shutdown() + +# Standalone wrapper that opens and closes MT5 for you +update_history_with_config( + output="history.db", + symbols=["EURUSD"], + config=Mt5Config(login=12345), +) +``` + +- **`collect-history`**: explicit date-range export into SQLite. +- **`update_history`**: incremental append based on existing SQLite `MAX(time)` per symbol (and timeframe for rates); account-level deals use a separate cursor when `include_account_events=True`. +- **`rates` table**: normalized storage with `symbol` and `timeframe` columns. +- **Rate compatibility views**: mt5cli manages all `rate_*` views. Naming is `rate___` when a symbol has one timeframe, otherwise `rate____` (for example `rate_EURUSD__M1_1`). Stale `rate_*` views are dropped and recreated when rates change for offline tools such as mteor optimize. ## Requirements diff --git a/docs/api/sqlite_history.md b/docs/api/sqlite_history.md new file mode 100644 index 0000000..628c73c --- /dev/null +++ b/docs/api/sqlite_history.md @@ -0,0 +1,131 @@ +# SQLite History Module + +::: mt5cli.sqlite_history + +## `collect-history` schema + +The `collect-history` command (and the matching `collect_history` SDK function) writes +selected MT5 datasets into one SQLite database. Each dataset becomes a table; column +names and types mirror the pdmt5 DataFrame schema for that export, with two additions: + +- `symbol` is prepended on every table. +- `timeframe` is prepended on `rates` so appended runs at different bar sizes stay + distinguishable. + +SQLite does not declare foreign keys. Rows are linked logically by `symbol`, time +windows, and (for deals) `position_id` / `order`. Duplicate rows are removed on +append using dataset-specific keys (for example `ticket` on history tables, or +`(symbol, timeframe, time)` on rates). + +Optional views are created when `--with-views` is set and the `history-deals` dataset +was written. + +### Entity-relationship diagram + +Sample layout for a full collection with `--with-views`: + +```mermaid +erDiagram + rates { + TEXT symbol "dedup key" + INTEGER timeframe "dedup key" + TEXT time "dedup key" + REAL open + REAL high + REAL low + REAL close + INTEGER tick_volume + INTEGER spread + INTEGER real_volume + } + + ticks { + TEXT symbol "dedup key" + TEXT time "dedup key" + INTEGER time_msc "dedup key (preferred)" + REAL bid + REAL ask + REAL last + INTEGER volume + INTEGER flags + REAL volume_real + } + + history_orders { + INTEGER ticket "dedup key" + TEXT symbol + TEXT time + INTEGER type + INTEGER state + REAL volume_initial + REAL price_open + REAL price_current + INTEGER magic + } + + history_deals { + INTEGER ticket "dedup key" + INTEGER order + INTEGER position_id "groups position view" + TEXT symbol + TEXT time + INTEGER type "0/1 trade, else cash event" + INTEGER entry "0 IN, 1 OUT, 2 INOUT, 3 OUT_BY" + REAL volume + REAL price + REAL profit + REAL commission + REAL swap + REAL fee + } + + cash_events { + INTEGER ticket + TEXT symbol + TEXT time + INTEGER type + REAL profit + } + + positions_reconstructed { + INTEGER position_id + TEXT symbol + TEXT open_time + TEXT close_time + INTEGER direction + REAL volume_open + REAL volume_close + REAL volume_reversal + REAL open_price + REAL close_price + REAL total_profit + INTEGER reversal_count + INTEGER deals_count + } + + rates ||--o{ history_deals : "symbol (logical)" + ticks ||--o{ history_deals : "symbol (logical)" + history_orders ||--o{ history_deals : "order ~ ticket (logical)" + history_deals ||--|| cash_events : "VIEW: type NOT IN (0,1)" + history_deals ||--o{ positions_reconstructed : "VIEW: GROUP BY position_id" +``` + +### Tables and views + +| Object | Kind | Source | Notes | +| ------------------------- | ----- | -------------------- | ------------------------------------------------------------------------------------------- | +| `rates` | table | `copy_rates_range` | Indexed on `(symbol, timeframe, time)` when columns exist. | +| `ticks` | table | `copy_ticks_range` | Indexed on `(symbol, time)` when columns exist. | +| `history_orders` | table | `history_orders_get` | Fetched per `--symbol`, then concatenated. | +| `history_deals` | table | `history_deals_get` | Fetched per `--symbol`, then concatenated. Indexed on `(position_id, symbol)` when present. | +| `cash_events` | view | `history_deals` | Non-trade deal types (deposits, balance ops, etc.). Requires `type` column. | +| `positions_reconstructed` | view | `history_deals` | One row per closed `position_id`; volume-weighted prices and reversal stats. | + +Column sets can vary with terminal and pdmt5 version. Views are skipped with a warning +when required columns are missing. + +### Incremental collection + +The `update_history` SDK path uses the same base tables and optional +`cash_events` / `positions_reconstructed` views. It additionally maintains +`rate___` compatibility views when `create_rate_views=True`. diff --git a/docs/index.md b/docs/index.md index 692cecf..955e449 100644 --- a/docs/index.md +++ b/docs/index.md @@ -152,6 +152,8 @@ mt5cli -o history.db collect-history \ History orders and deals are fetched per symbol and concatenated, so the symbol filter is applied consistently across all datasets. The `cash_events` view is derived from symbol-filtered `history_deals`, so account-level cash events with empty or non-matching symbols may be excluded. The `positions_reconstructed` view excludes positions with no closing deal, uses volume-weighted open/close prices, and reports reversal deals (`DEAL_ENTRY_INOUT`) via `volume_reversal` / `reversal_count`. +See the [SQLite History schema diagram](api/sqlite_history.md#entity-relationship-diagram) for a sample ER layout of the resulting database. + ## Global Options | Option | Description | diff --git a/mkdocs.yml b/mkdocs.yml index 53e9d48..b4cfc37 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -24,6 +24,7 @@ theme: features: - content.code.annotate - content.code.copy + - content.code.mermaid - navigation.indexes - navigation.sections - navigation.tabs @@ -57,6 +58,7 @@ nav: - Overview: api/index.md - CLI: api/cli.md - SDK: api/sdk.md + - SQLite History: api/sqlite_history.md - Utils: api/utils.md markdown_extensions: diff --git a/mt5cli/__init__.py b/mt5cli/__init__.py index ee57b80..f01ac06 100644 --- a/mt5cli/__init__.py +++ b/mt5cli/__init__.py @@ -22,15 +22,19 @@ from .sdk import ( symbol_info_tick, symbols, terminal_info, + update_history, + update_history_with_config, ) from .sdk import ( version as mt5_version, ) -from .utils import detect_format, export_dataframe +from .utils import Dataset, IfExists, detect_format, export_dataframe __version__ = version(__package__) if __package__ else None __all__ = [ + "Dataset", + "IfExists", "Mt5CliClient", "account_info", "build_config", @@ -53,4 +57,6 @@ __all__ = [ "symbol_info_tick", "symbols", "terminal_info", + "update_history", + "update_history_with_config", ] diff --git a/mt5cli/sdk.py b/mt5cli/sdk.py index 28ada99..93d3442 100644 --- a/mt5cli/sdk.py +++ b/mt5cli/sdk.py @@ -5,12 +5,23 @@ from __future__ import annotations import logging import sqlite3 from contextlib import contextmanager -from datetime import datetime -from pathlib import Path # noqa: TC003 +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from pathlib import Path from typing import TYPE_CHECKING, Self, TypeVar from pdmt5 import Mt5Config, Mt5DataClient +from .sqlite_history import ( + create_cash_events_view, + create_history_indexes, + create_positions_reconstructed_view, + resolve_history_datasets, + resolve_history_tick_flags, + resolve_history_timeframes, + write_collected_datasets, + write_incremental_datasets, +) from .utils import ( Dataset, IfExists, @@ -20,7 +31,7 @@ from .utils import ( ) if TYPE_CHECKING: - from collections.abc import Callable, Iterator + from collections.abc import Callable, Iterator, Sequence import pandas as pd @@ -48,22 +59,11 @@ __all__ = [ "symbol_info_tick", "symbols", "terminal_info", + "update_history", + "update_history_with_config", "version", ] -_TRADE_DEAL_TYPES: tuple[int, int] = (0, 1) -_TRADE_DEAL_TYPES_SQL = f"({', '.join(str(value) for value in _TRADE_DEAL_TYPES)})" -_POSITIONS_VIEW_REQUIRED_COLUMNS: frozenset[str] = frozenset({ - "position_id", - "symbol", - "time", - "type", - "entry", - "volume", - "price", - "profit", -}) - def _coerce_timeframe(timeframe: int | str) -> int: if isinstance(timeframe, int): @@ -419,325 +419,219 @@ class Mt5CliClient: return self._fetch(lambda c: c.market_book_get_as_df(symbol=symbol)) -def _create_cash_events_view( - conn: sqlite3.Connection, - deals_columns: set[str], -) -> bool: - """Create the cash_events SQLite view derived from history_deals. +def _resolve_incremental_settings( + selected_datasets: set[Dataset], + timeframes: Sequence[int | str] | None, + flags: int | str, +) -> tuple[list[int], int]: + """Resolve dataset-specific incremental update settings. Returns: - True if the view was created, False if required columns are missing. + Tuple of resolved rate timeframes and tick copy flags. + + Raises: + ValueError: If timeframe or tick flag values are invalid. """ - 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 + resolved_timeframes: list[int] = [] + if Dataset.rates in selected_datasets: + try: + resolved_timeframes = resolve_history_timeframes(timeframes) + except ValueError as exc: + msg = str(exc) + raise ValueError(msg) from exc + resolved_tick_flags = 0 + if Dataset.ticks in selected_datasets: + try: + resolved_tick_flags = resolve_history_tick_flags(flags) + except ValueError as exc: + msg = str(exc) + raise ValueError(msg) from exc + return resolved_timeframes, resolved_tick_flags -def _create_positions_reconstructed_view( - conn: sqlite3.Connection, - deals_columns: set[str], -) -> bool: - """Create the positions_reconstructed SQLite view derived from history_deals. +@dataclass(frozen=True) +class _UpdateHistoryRequest: + selected: set[Dataset] + end: datetime + fallback_start: datetime + resolved_timeframes: list[int] + resolved_tick_flags: int + output_path: Path + + +def _resolve_update_history_request( + *, + output: Path | str, + symbols: Sequence[str], + datasets: set[Dataset] | None, + timeframes: Sequence[int | str] | None, + flags: int | str, + lookback_hours: float, + date_to: datetime | str | None, +) -> _UpdateHistoryRequest | None: + """Validate and resolve incremental history update inputs. Returns: - True if the view was created, False if required columns are missing. + Resolved request parameters, or None when no datasets are selected. + + Raises: + ValueError: If symbols are empty, lookback_hours is not positive, or + timeframe/flag values are invalid. """ - 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, 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, 3) THEN volume ELSE 0 END) > 0" - " THEN SUM(CASE WHEN entry IN (1, 3) THEN price * volume ELSE 0 END)" - " / SUM(CASE WHEN entry IN (1, 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, 3) THEN 1 ELSE 0 END) > 0", - ) - return True + if lookback_hours <= 0: + msg = "lookback_hours must be positive." + raise ValueError(msg) + selected = resolve_history_datasets(datasets) + if not selected: + logger.info("Skipping SQLite history update: no datasets selected.") + return None + if not symbols: + msg = "At least one symbol is required." + raise ValueError(msg) - -def _write_frame_to_sqlite( - conn: sqlite3.Connection, - frame: pd.DataFrame, - table_name: str, - if_exists: IfExists, -) -> bool: - """Write a non-empty-schema frame to SQLite. - - 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, - method="multi", - ) - return True - - -def _create_collect_history_indexes( - conn: sqlite3.Connection, - written_columns: dict[Dataset, set[str]], -) -> None: - """Create useful indexes for collected history tables when present.""" - if {"symbol", "time"}.issubset(written_columns.get(Dataset.rates, set())): - conn.execute( - "CREATE INDEX IF NOT EXISTS idx_rates_symbol_time ON rates(symbol, 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 _record_written_columns( - written_columns: dict[Dataset, set[str]], - dataset: Dataset, - frame: pd.DataFrame, -) -> None: - """Remember columns for datasets written during streaming collection.""" - columns = set(frame.columns) - if dataset in written_columns: - written_columns[dataset].update(columns) + if date_to is not None: + resolved_end = _coerce_datetime(date_to) else: - written_columns[dataset] = columns - - -def _write_streamed_frame( - conn: sqlite3.Connection, - frame: pd.DataFrame, - dataset: Dataset, - table_exists: bool, - if_exists: IfExists, - written_columns: dict[Dataset, set[str]], -) -> bool: - """Write one streamed dataset frame and track table state. - - Returns: - True if the dataset table exists after this write attempt. - """ - write_mode = IfExists.APPEND if table_exists else if_exists - if _write_frame_to_sqlite( - conn, - frame, - dataset.table_name, - write_mode, - ): - _record_written_columns(written_columns, dataset, frame) - return True - return table_exists - - -def _write_rates_dataset( - conn: sqlite3.Connection, - client: Mt5DataClient, - symbols: list[str], - timeframe: int, - date_from: datetime, - date_to: datetime, - if_exists: IfExists, - written_columns: dict[Dataset, set[str]], -) -> bool: - """Stream rates frames into SQLite. - - Returns: - True if the rates table was written. - """ - table_exists = False - for sym in symbols: - frame = client.copy_rates_range_as_df( - symbol=sym, - timeframe=timeframe, - date_from=date_from, - date_to=date_to, - ) - frame.insert(0, "symbol", sym) - frame.insert(1, "timeframe", timeframe) - table_exists = _write_streamed_frame( - conn, - frame, - Dataset.rates, - table_exists, - if_exists, - written_columns, - ) - return table_exists - - -def _write_ticks_dataset( - conn: sqlite3.Connection, - client: Mt5DataClient, - symbols: list[str], - flags: int, - date_from: datetime, - date_to: datetime, - if_exists: IfExists, - written_columns: dict[Dataset, set[str]], -) -> bool: - """Stream ticks frames into SQLite. - - Returns: - True if the ticks table was written. - """ - table_exists = False - for sym in symbols: - frame = client.copy_ticks_range_as_df( - symbol=sym, - date_from=date_from, - date_to=date_to, - flags=flags, - ) - frame.insert(0, "symbol", sym) - table_exists = _write_streamed_frame( - conn, - frame, - Dataset.ticks, - table_exists, - if_exists, - written_columns, - ) - return table_exists - - -def _write_history_dataset( - conn: sqlite3.Connection, - fetch: Callable[..., pd.DataFrame], - dataset: Dataset, - symbols: list[str], - date_from: datetime, - date_to: datetime, - if_exists: IfExists, - written_columns: dict[Dataset, set[str]], -) -> bool: - """Stream a history dataset into SQLite with exact symbol filtering. - - Returns: - True if the history table was written. - """ - table_exists = False - for sym in symbols: - frame = fetch(date_from=date_from, date_to=date_to, symbol=sym) - if "symbol" in frame.columns: - frame = frame[frame["symbol"] == sym] - table_exists = _write_streamed_frame( - conn, - frame, - dataset, - table_exists, - if_exists, - written_columns, - ) - return table_exists - - -def _write_collected_datasets( - conn: sqlite3.Connection, - client: Mt5DataClient, - symbols: list[str], - datasets: set[Dataset], - timeframe: int, - flags: int, - date_from: datetime, - date_to: datetime, - if_exists: IfExists, -) -> tuple[set[Dataset], dict[Dataset, set[str]]]: - """Collect selected datasets and stream each symbol frame into SQLite. - - Returns: - Written datasets and their columns. - """ - written_columns: dict[Dataset, set[str]] = {} - written_tables: set[Dataset] = set() - if Dataset.rates in datasets and _write_rates_dataset( - conn, - client, - symbols, - timeframe, - date_from, - date_to, - if_exists, - written_columns, - ): - written_tables.add(Dataset.rates) - if Dataset.ticks in datasets and _write_ticks_dataset( - conn, - client, - symbols, + resolved_end = datetime.now(UTC) + end = resolved_end if resolved_end is not None else datetime.now(UTC) + fallback_start = end - timedelta(hours=lookback_hours) + resolved_timeframes, resolved_tick_flags = _resolve_incremental_settings( + selected, + timeframes, flags, - date_from, - date_to, - if_exists, - written_columns, - ): - written_tables.add(Dataset.ticks) - if Dataset.history_orders in datasets and _write_history_dataset( - conn, - client.history_orders_get_as_df, - Dataset.history_orders, - symbols, - date_from, - date_to, - if_exists, - written_columns, - ): - written_tables.add(Dataset.history_orders) - if Dataset.history_deals in datasets and _write_history_dataset( - conn, - client.history_deals_get_as_df, - Dataset.history_deals, - symbols, - date_from, - date_to, - if_exists, - written_columns, - ): - written_tables.add(Dataset.history_deals) - return written_tables, written_columns + ) + return _UpdateHistoryRequest( + selected=selected, + end=end, + fallback_start=fallback_start, + resolved_timeframes=resolved_timeframes, + resolved_tick_flags=resolved_tick_flags, + output_path=Path(output), + ) + + +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___`` 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, + ) + + +def update_history_with_config( # noqa: PLR0913 + *, + output: Path | str, + symbols: Sequence[str], + config: Mt5Config | None = None, + 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, opening and closing the MT5 connection. + + Convenience wrapper around :func:`update_history` for standalone use. + """ + 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 + mt5_config = config or build_config() + with _connected_client(mt5_config) as client: + update_history( + client=client, + output=output, + symbols=symbols, + datasets=datasets, + timeframes=timeframes, + flags=flags, + lookback_hours=lookback_hours, + date_to=date_to, + deduplicate=deduplicate, + create_rate_views=create_rate_views, + with_views=with_views, + include_account_events=include_account_events, + ) def collect_history( @@ -776,7 +670,7 @@ def collect_history( 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( + written_tables, written_columns = write_collected_datasets( conn, client, symbols, @@ -787,10 +681,10 @@ def collect_history( end, if_exists, ) - _create_collect_history_indexes(conn, written_columns) + 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( + create_cash_events_view(conn, written_columns[Dataset.history_deals]) + create_positions_reconstructed_view( conn, written_columns[Dataset.history_deals], ) diff --git a/mt5cli/sqlite_history.py b/mt5cli/sqlite_history.py new file mode 100644 index 0000000..9d9f2fd --- /dev/null +++ b/mt5cli/sqlite_history.py @@ -0,0 +1,1176 @@ +"""SQLite helpers for incremental MT5 history collection.""" + +from __future__ import annotations + +import logging +import sqlite3 +from datetime import UTC, datetime +from typing import TYPE_CHECKING, Literal + +import pandas as pd + +from .utils import ( + TIMEFRAME_MAP, + Dataset, + IfExists, + parse_datetime, + parse_tick_flags, + parse_timeframe, +) + +if TYPE_CHECKING: + from collections.abc import Callable, Sequence + + from pdmt5 import Mt5DataClient + +logger = logging.getLogger(__name__) + +DEFAULT_HISTORY_TIMEFRAMES: tuple[str, ...] = tuple(TIMEFRAME_MAP) + +_HISTORY_DEDUP_KEYS: dict[Dataset, tuple[tuple[str, ...], ...]] = { + Dataset.rates: (("symbol", "timeframe", "time"), ("symbol", "time")), + Dataset.ticks: (("symbol", "time_msc"), ("symbol", "time")), + Dataset.history_orders: (("ticket",), ("symbol", "time", "type")), + Dataset.history_deals: (("ticket",), ("symbol", "time", "type", "entry")), +} + +_TRADE_DEAL_TYPES: tuple[int, int] = (0, 1) +_TRADE_DEAL_TYPES_SQL = f"({', '.join(str(value) for value in _TRADE_DEAL_TYPES)})" + +_POSITIONS_VIEW_REQUIRED_COLUMNS: frozenset[str] = frozenset({ + "position_id", + "symbol", + "time", + "type", + "entry", + "volume", + "price", + "profit", +}) + + +def quote_sqlite_identifier(identifier: str) -> str: + """Return a safely quoted SQLite identifier using double quotes.""" + return '"' + identifier.replace('"', '""') + '"' + + +def resolve_history_datasets(datasets: set[Dataset] | None) -> set[Dataset]: + """Resolve configured history datasets. + + Returns: + All supported datasets when ``datasets`` is None, otherwise the + configured selection (which may be empty). + """ + if datasets is None: + return set(Dataset) + return set(datasets) + + +def resolve_history_timeframes( + timeframes: Sequence[int | str] | None, +) -> list[int]: + """Resolve rate timeframes, deduplicating aliases for the same integer. + + Returns: + Ordered list of unique timeframe integers. + """ + raw = timeframes if timeframes is not None else DEFAULT_HISTORY_TIMEFRAMES + seen: set[int] = set() + resolved: list[int] = [] + for value in raw: + tf = value if isinstance(value, int) else parse_timeframe(str(value)) + if tf not in seen: + seen.add(tf) + resolved.append(tf) + return resolved + + +def resolve_history_tick_flags(flags: int | str) -> int: + """Resolve tick copy flags from an integer or name. + + Returns: + Integer tick flag value. + """ + if isinstance(flags, int): + return flags + return parse_tick_flags(flags) + + +def resolve_granularity_name(timeframe: int) -> str: + """Return a granularity name for a timeframe integer when known.""" + for name, value in TIMEFRAME_MAP.items(): + if value == timeframe: + return name + return str(timeframe) + + +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 get_table_columns(conn: sqlite3.Connection, table: str) -> set[str]: + """Return existing SQLite columns for a table.""" + rows = conn.execute(f"PRAGMA table_info({table})").fetchall() + return {str(row[1]) for row in rows} + + +def _parse_string_sqlite_timestamp(value: str) -> datetime | None: + try: + return parse_datetime(value) + except ValueError: + parsed_ts = pd.to_datetime(value, utc=True, errors="coerce") + if pd.isna(parsed_ts): + logger.warning("Ignoring unparseable history timestamp: %s", value) + return None + return parsed_ts.to_pydatetime() + + +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 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 + + +_REQUIRED_RATE_COLUMNS = frozenset({"symbol", "timeframe", "time"}) + + +def _validate_rates_schema(columns: set[str]) -> None: + """Validate an existing rates table has normalized incremental columns. + + Raises: + ValueError: If required columns are missing. + """ + missing = _REQUIRED_RATE_COLUMNS - columns + if missing: + msg = ( + "The rates table must include symbol, timeframe, and time columns" + " for incremental updates; " + f"missing: {', '.join(sorted(missing))}." + ) + raise ValueError(msg) + + +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 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 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 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 + + +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 write_streamed_frame( + conn: sqlite3.Connection, + frame: pd.DataFrame, + dataset: Dataset, + table_exists: bool, + if_exists: IfExists, + written_columns: dict[Dataset, set[str]], +) -> bool: + """Write one streamed dataset frame and track table state. + + Returns: + True if the dataset table exists after this write attempt. + """ + write_mode = IfExists.APPEND if table_exists else if_exists + if append_dataframe(conn, frame, dataset.table_name, write_mode): + record_written_columns(written_columns, dataset, frame) + return True + return table_exists + + +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})", + ) + + +DedupScope = tuple[str, tuple[object, ...]] + + +def _record_dedup_scope( + dedup_scopes: dict[Dataset, list[DedupScope]], + dataset: Dataset, + scope_where: str, + scope_params: tuple[object, ...], +) -> None: + dedup_scopes.setdefault(dataset, []).append((scope_where, scope_params)) + + +def deduplicate_history_tables( + conn: sqlite3.Connection, + written_columns: dict[Dataset, set[str]], + written_tables: set[Dataset], + dedup_scopes: dict[Dataset, list[DedupScope]] | None = None, +) -> None: + """Deduplicate appended history tables by stable identifiers.""" + 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 + scopes = dedup_scopes.get(dataset, []) if dedup_scopes else [] + if scopes: + for scope_where, scope_params 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 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 _history_deals_account_event_mask(frame: pd.DataFrame) -> pd.Series: + if "type" in frame.columns: + return ~frame["type"].isin(_TRADE_DEAL_TYPES) + if "symbol" in frame.columns: + return frame["symbol"].isna() | frame["symbol"].eq("") + return pd.Series(data=False, index=frame.index) + + +def _frame_parsed_times(frame: pd.DataFrame) -> pd.Series: + if "time" not in frame.columns: + return pd.Series([pd.NaT] * len(frame), index=frame.index, dtype="object") + return frame["time"].map(parse_sqlite_timestamp) + + +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 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_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 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 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 write_rates_dataset( + conn: sqlite3.Connection, + client: Mt5DataClient, + symbols: Sequence[str], + timeframe: int, + date_from: datetime, + date_to: datetime, + if_exists: IfExists, + written_columns: dict[Dataset, set[str]], +) -> bool: + """Stream rates frames into SQLite. + + Returns: + True if the rates table was written. + """ + table_exists = False + for sym in symbols: + frame = client.copy_rates_range_as_df( + symbol=sym, + timeframe=timeframe, + date_from=date_from, + date_to=date_to, + ).drop(columns=["symbol", "timeframe"], errors="ignore") + if len(frame.columns) != 0: + frame.insert(0, "symbol", sym) + frame.insert(1, "timeframe", timeframe) + table_exists = write_streamed_frame( + conn, + frame, + Dataset.rates, + table_exists, + if_exists, + written_columns, + ) + return table_exists + + +def write_ticks_dataset( + conn: sqlite3.Connection, + client: Mt5DataClient, + symbols: Sequence[str], + flags: int, + date_from: datetime, + date_to: datetime, + if_exists: IfExists, + written_columns: dict[Dataset, set[str]], +) -> bool: + """Stream ticks frames into SQLite. + + Returns: + True if the ticks table was written. + """ + table_exists = False + for sym in symbols: + frame = client.copy_ticks_range_as_df( + symbol=sym, + date_from=date_from, + date_to=date_to, + flags=flags, + ).drop(columns=["symbol"], errors="ignore") + if len(frame.columns) != 0: + frame.insert(0, "symbol", sym) + table_exists = write_streamed_frame( + conn, + frame, + Dataset.ticks, + table_exists, + if_exists, + written_columns, + ) + return table_exists + + +def write_history_dataset( + conn: sqlite3.Connection, + fetch: Callable[..., pd.DataFrame], + dataset: Dataset, + symbols: Sequence[str], + date_from: datetime, + date_to: datetime, + if_exists: IfExists, + written_columns: dict[Dataset, set[str]], + *, + include_account_events: bool = False, +) -> bool: + """Stream a history dataset into SQLite. + + Returns: + True if the target table was written. + """ + table_exists = False + if include_account_events: + frame = filter_trade_history_frame( + fetch(date_from=date_from, date_to=date_to), + symbols, + include_account_events=True, + ) + return write_streamed_frame( + conn, + frame, + dataset, + table_exists, + if_exists, + written_columns, + ) + for sym in symbols: + frame = fetch(date_from=date_from, date_to=date_to, symbol=sym) + frame = filter_trade_history_frame( + frame, + [sym], + include_account_events=False, + ) + table_exists = write_streamed_frame( + conn, + frame, + dataset, + table_exists, + if_exists, + written_columns, + ) + return table_exists + + +def _write_incremental_rates( + conn: sqlite3.Connection, + client: Mt5DataClient, + symbols: Sequence[str], + resolved_timeframes: list[int], + fallback_start: datetime, + end_date: datetime, + written_columns: dict[Dataset, set[str]], + written_tables: set[Dataset], + dedup_scopes: dict[Dataset, list[DedupScope]], +) -> None: + start_by_key = load_incremental_start_datetimes( + conn, + Dataset.rates, + symbols=symbols, + timeframes=resolved_timeframes, + fallback_start=fallback_start, + ) + for symbol in symbols: + for timeframe in resolved_timeframes: + start_date = start_by_key[symbol, timeframe] + if write_rates_dataset( + conn, + client, + [symbol], + timeframe, + start_date, + end_date, + IfExists.APPEND, + written_columns, + ): + written_tables.add(Dataset.rates) + _record_dedup_scope( + dedup_scopes, + Dataset.rates, + "symbol = ? AND timeframe = ? AND time >= ?", + (symbol, timeframe, start_date), + ) + + +def _write_incremental_ticks( + conn: sqlite3.Connection, + client: Mt5DataClient, + symbols: Sequence[str], + resolved_tick_flags: int, + fallback_start: datetime, + end_date: datetime, + written_columns: dict[Dataset, set[str]], + written_tables: set[Dataset], + dedup_scopes: dict[Dataset, list[DedupScope]], +) -> None: + start_by_symbol = load_incremental_start_datetimes( + conn, + Dataset.ticks, + symbols=symbols, + fallback_start=fallback_start, + ) + for symbol in symbols: + start_date = start_by_symbol[symbol, None] + if write_ticks_dataset( + conn, + client, + [symbol], + resolved_tick_flags, + start_date, + end_date, + IfExists.APPEND, + written_columns, + ): + written_tables.add(Dataset.ticks) + _record_dedup_scope( + dedup_scopes, + Dataset.ticks, + "symbol = ? AND time >= ?", + (symbol, start_date), + ) + + +def _write_incremental_history_orders( + conn: sqlite3.Connection, + client: Mt5DataClient, + symbols: Sequence[str], + fallback_start: datetime, + end_date: datetime, + written_columns: dict[Dataset, set[str]], + written_tables: set[Dataset], + dedup_scopes: dict[Dataset, list[DedupScope]], +) -> None: + start_by_symbol = load_incremental_start_datetimes( + conn, + Dataset.history_orders, + symbols=symbols, + fallback_start=fallback_start, + ) + for symbol in symbols: + start_date = start_by_symbol[symbol, None] + if write_history_dataset( + conn, + client.history_orders_get_as_df, + Dataset.history_orders, + [symbol], + start_date, + end_date, + IfExists.APPEND, + written_columns, + include_account_events=False, + ): + written_tables.add(Dataset.history_orders) + _record_dedup_scope( + dedup_scopes, + Dataset.history_orders, + "symbol = ? AND time >= ?", + (symbol, start_date), + ) + + +def _write_incremental_history_deals( + conn: sqlite3.Connection, + client: Mt5DataClient, + symbols: Sequence[str], + fallback_start: datetime, + end_date: datetime, + written_columns: dict[Dataset, set[str]], + written_tables: set[Dataset], + dedup_scopes: dict[Dataset, list[DedupScope]], + *, + include_account_events: bool, +) -> None: + if include_account_events: + start_by_symbol = load_incremental_start_datetimes( + conn, + Dataset.history_deals, + symbols=symbols, + fallback_start=fallback_start, + ) + account_event_start = get_history_deals_account_event_start_datetime( + conn, + fallback_start=fallback_start, + ) + fetch_start = min([*start_by_symbol.values(), account_event_start]) + frame = filter_incremental_history_deals_frame( + client.history_deals_get_as_df( + date_from=fetch_start, + date_to=end_date, + ), + symbols, + {symbol: start_by_symbol[symbol, None] for symbol in symbols}, + account_event_start, + ) + if write_streamed_frame( + conn, + frame, + Dataset.history_deals, + table_exists=False, + if_exists=IfExists.APPEND, + written_columns=written_columns, + ): + written_tables.add(Dataset.history_deals) + columns = get_table_columns(conn, Dataset.history_deals.table_name) + if "symbol" in columns: + for symbol in symbols: + _record_dedup_scope( + dedup_scopes, + Dataset.history_deals, + "symbol = ? AND time >= ?", + (symbol, start_by_symbol[symbol, None]), + ) + if "type" in columns: + _record_dedup_scope( + dedup_scopes, + Dataset.history_deals, + f"type NOT IN {_TRADE_DEAL_TYPES_SQL} AND time >= ?", + (account_event_start,), + ) + if "type" not in columns and "symbol" in columns: + _record_dedup_scope( + dedup_scopes, + Dataset.history_deals, + "(symbol IS NULL OR symbol = '') AND time >= ?", + (account_event_start,), + ) + return + start_by_symbol = load_incremental_start_datetimes( + conn, + Dataset.history_deals, + symbols=symbols, + fallback_start=fallback_start, + ) + for symbol in symbols: + start_date = start_by_symbol[symbol, None] + if write_history_dataset( + conn, + client.history_deals_get_as_df, + Dataset.history_deals, + [symbol], + start_date, + end_date, + IfExists.APPEND, + written_columns, + include_account_events=False, + ): + written_tables.add(Dataset.history_deals) + _record_dedup_scope( + dedup_scopes, + Dataset.history_deals, + "symbol = ? AND time >= ?", + (symbol, start_date), + ) + + +def _finalize_incremental_writes( + conn: sqlite3.Connection, + selected_datasets: set[Dataset], + written_columns: dict[Dataset, set[str]], + written_tables: set[Dataset], + dedup_scopes: dict[Dataset, list[DedupScope]], + *, + deduplicate: bool, + create_rate_views: bool, + with_views: bool, +) -> None: + augment_written_columns_from_sqlite(conn, selected_datasets, written_columns) + create_history_indexes(conn, written_columns) + if deduplicate: + deduplicate_history_tables( + conn, + written_columns, + written_tables, + dedup_scopes, + ) + if create_rate_views and Dataset.rates in written_tables: + create_rate_compatibility_views(conn) + if with_views and Dataset.history_deals in selected_datasets: + if Dataset.history_deals in written_columns: + deal_columns = written_columns[Dataset.history_deals] + create_cash_events_view(conn, deal_columns) + create_positions_reconstructed_view(conn, deal_columns) + if ( + Dataset.history_deals not in written_columns + and Dataset.history_deals not in written_tables + ): + logger.warning( + "with_views ignored: history_deals table was not available", + ) + + +def write_incremental_datasets( # noqa: PLR0913 + conn: sqlite3.Connection, + client: Mt5DataClient, + symbols: Sequence[str], + selected_datasets: set[Dataset], + resolved_timeframes: list[int], + resolved_tick_flags: int, + fallback_start: datetime, + end_date: datetime, + *, + deduplicate: bool, + create_rate_views: bool, + with_views: bool, + include_account_events: bool, +) -> tuple[set[Dataset], dict[Dataset, set[str]]]: + """Append selected datasets incrementally and refresh indexes and views. + + Returns: + Written datasets and their columns. + """ + written_columns: dict[Dataset, set[str]] = {} + written_tables: set[Dataset] = set() + dedup_scopes: dict[Dataset, list[DedupScope]] = {} + if Dataset.rates in selected_datasets: + _write_incremental_rates( + conn, + client, + symbols, + resolved_timeframes, + fallback_start, + end_date, + written_columns, + written_tables, + dedup_scopes, + ) + if Dataset.ticks in selected_datasets: + _write_incremental_ticks( + conn, + client, + symbols, + resolved_tick_flags, + fallback_start, + end_date, + written_columns, + written_tables, + dedup_scopes, + ) + if Dataset.history_orders in selected_datasets: + _write_incremental_history_orders( + conn, + client, + symbols, + fallback_start, + end_date, + written_columns, + written_tables, + dedup_scopes, + ) + if Dataset.history_deals in selected_datasets: + _write_incremental_history_deals( + conn, + client, + symbols, + fallback_start, + end_date, + written_columns, + written_tables, + dedup_scopes, + include_account_events=include_account_events, + ) + _finalize_incremental_writes( + conn, + selected_datasets, + written_columns, + written_tables, + dedup_scopes, + deduplicate=deduplicate, + create_rate_views=create_rate_views, + with_views=with_views, + ) + return written_tables, written_columns + + +def write_collected_datasets( + conn: sqlite3.Connection, + client: Mt5DataClient, + symbols: Sequence[str], + datasets: set[Dataset], + timeframe: int, + flags: int, + date_from: datetime, + date_to: datetime, + if_exists: IfExists, +) -> tuple[set[Dataset], dict[Dataset, set[str]]]: + """Collect selected datasets and stream each symbol frame into SQLite. + + Returns: + Written datasets and their columns. + """ + written_columns: dict[Dataset, set[str]] = {} + written_tables: set[Dataset] = set() + if Dataset.rates in datasets and write_rates_dataset( + conn, + client, + symbols, + timeframe, + date_from, + date_to, + if_exists, + written_columns, + ): + written_tables.add(Dataset.rates) + if Dataset.ticks in datasets and write_ticks_dataset( + conn, + client, + symbols, + flags, + date_from, + date_to, + if_exists, + written_columns, + ): + written_tables.add(Dataset.ticks) + if Dataset.history_orders in datasets and write_history_dataset( + conn, + client.history_orders_get_as_df, + Dataset.history_orders, + symbols, + date_from, + date_to, + if_exists, + written_columns, + include_account_events=False, + ): + written_tables.add(Dataset.history_orders) + if Dataset.history_deals in datasets and write_history_dataset( + conn, + client.history_deals_get_as_df, + Dataset.history_deals, + symbols, + date_from, + date_to, + if_exists, + written_columns, + include_account_events=False, + ): + written_tables.add(Dataset.history_deals) + return written_tables, written_columns diff --git a/pyproject.toml b/pyproject.toml index c663a5a..aded71a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "mt5cli" -version = "0.4.0" +version = "0.4.1" description = "Command-line tool for MetaTrader 5" authors = [{name = "dceoy", email = "dceoy@users.noreply.github.com"}] maintainers = [{name = "dceoy", email = "dceoy@users.noreply.github.com"}] @@ -124,6 +124,7 @@ ignore = [ ] [tool.ruff.lint.per-file-ignores] +"mt5cli/sqlite_history.py" = ["TC003"] "tests/**/*.py" = [ "DOC201", # Missing return documentation "DOC501", # Raised exception missing from docstring diff --git a/tests/test_cli.py b/tests/test_cli.py index 79c21ab..8b91700 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1089,7 +1089,7 @@ class TestCollectHistory: assert all(row[0] not in {0, 1} for row in cash) # Position 100 (BUY 1@1.10 + BUY 3@1.20 then SELL 4@1.50) is closed. # Position 200 (BUY 2@2.00 then SELL 2@2.20) is closed. - # Position 300 (open-only) and 400 (reversal-only) are excluded. + # Position 400 (reversal-only with non-trade deal type) stays excluded. assert set(positions) == {100, 200, 500, 600} pos_100 = positions[100] tol = 1e-9 @@ -1106,10 +1106,10 @@ class TestCollectHistory: assert abs(pos_500[5] - 1.05) < tol pos_600 = positions[600] assert abs(pos_600[1] - 3.0) < tol - assert abs(pos_600[2] - 3.0) < tol + assert abs(pos_600[2] - 4.0) < tol # reversal + close volumes assert abs(pos_600[3] - 1.0) < tol assert abs(pos_600[4] - 1.10) < tol - assert abs(pos_600[5] - 1.40) < tol + assert abs(pos_600[5] - 3.5475) < tol assert pos_600[6] == 1 def test_collect_history_filters_history_symbols_exactly( diff --git a/tests/test_sdk.py b/tests/test_sdk.py index 3432ef1..4d68c78 100644 --- a/tests/test_sdk.py +++ b/tests/test_sdk.py @@ -36,8 +36,11 @@ from mt5cli.sdk import ( symbol_info_tick, symbols, terminal_info, + update_history, + update_history_with_config, version, ) +from mt5cli.sqlite_history import DEFAULT_HISTORY_TIMEFRAMES from mt5cli.utils import Dataset _DEALS_FIXTURE: dict[str, list[object]] = { @@ -464,3 +467,344 @@ class TestCollectHistory: } assert "cash_events" not in views assert "positions_reconstructed" not in views + + +class TestUpdateHistory: + """Tests for update_history SDK functions.""" + + @pytest.fixture + def connected_client(self) -> MagicMock: + """Create a connected mock client without MT5 lifecycle patching.""" + return MagicMock() + + def test_update_history_appends_incrementally( + self, + connected_client: MagicMock, + mocker: MockerFixture, + tmp_path: Path, + ) -> None: + """Test sequential SQLite history updates use existing max timestamps.""" + date_to = datetime(2024, 1, 2, tzinfo=UTC) + first_expected_start = datetime(2024, 1, 1, tzinfo=UTC) + second_expected_start = datetime(2024, 1, 1, 12, tzinfo=UTC) + rate_starts: list[datetime] = [] + deal_starts: list[datetime] = [] + + def make_rates(**kwargs: object) -> pd.DataFrame: + assert kwargs["symbol"] == "EURUSD" + assert kwargs["timeframe"] == 1 + assert kwargs["date_to"] == date_to + rate_starts.append(kwargs["date_from"]) # type: ignore[arg-type] + return pd.DataFrame({ + "time": ["2024-01-01T12:00:00+00:00"], + "open": [1.0 + len(rate_starts) / 10], + }) + + def make_deals(**kwargs: object) -> pd.DataFrame: + assert kwargs["date_to"] == date_to + deal_starts.append(kwargs["date_from"]) # type: ignore[arg-type] + return pd.DataFrame({ + "ticket": [10], + "position_id": [100], + "symbol": ["EURUSD"], + "time": ["2024-01-01T12:00:00+00:00"], + "type": [0], + "entry": [0], + "volume": [1.0], + "price": [1.1], + "profit": [0.0], + }) + + connected_client.copy_rates_range_as_df.side_effect = make_rates + connected_client.history_deals_get_as_df.side_effect = make_deals + mocker.patch("mt5cli.sdk.Mt5DataClient") + output = tmp_path / "incremental-history.db" + + for _ in range(2): + update_history( + client=connected_client, + output=output, + symbols=["EURUSD"], + datasets={Dataset.rates, Dataset.history_deals}, + timeframes=["M1"], + lookback_hours=24, + date_to=date_to, + with_views=True, + ) + + assert rate_starts == [first_expected_start, second_expected_start] + assert deal_starts == [first_expected_start, first_expected_start] + connected_client.initialize_and_login_mt5.assert_not_called() + connected_client.shutdown.assert_not_called() + with sqlite3.connect(output) as conn: + assert conn.execute("SELECT COUNT(*) FROM rates").fetchone() == (1,) + assert conn.execute("SELECT open FROM rates").fetchone() == (1.2,) + assert conn.execute( + "SELECT COUNT(*) FROM history_deals", + ).fetchone() == (1,) + assert conn.execute( + "SELECT name FROM sqlite_master WHERE name = 'cash_events'", + ).fetchone() == ("cash_events",) + + def test_update_history_rejects_invalid_inputs( + self, + connected_client: MagicMock, + tmp_path: Path, + ) -> None: + """Test validation errors for incremental history updates.""" + output = tmp_path / "invalid-update.db" + with pytest.raises(ValueError, match="At least one symbol"): + update_history( + client=connected_client, + output=output, + symbols=[], + ) + with pytest.raises(ValueError, match="lookback_hours must be positive"): + update_history( + client=connected_client, + output=output, + symbols=["EURUSD"], + lookback_hours=0, + ) + with pytest.raises(ValueError, match="Invalid timeframe"): + update_history( + client=connected_client, + output=output, + symbols=["EURUSD"], + datasets={Dataset.rates}, + timeframes=["BAD"], + ) + with pytest.raises(ValueError, match="Invalid tick flags"): + update_history( + client=connected_client, + output=output, + symbols=["EURUSD"], + datasets={Dataset.ticks}, + flags="BAD", + ) + + def test_update_history_noops_for_empty_datasets( + self, + connected_client: MagicMock, + mocker: MockerFixture, + tmp_path: Path, + ) -> None: + """Test empty dataset selection skips MT5 and SQLite writes.""" + writer = mocker.patch("mt5cli.sdk.write_incremental_datasets") + connect = mocker.patch("mt5cli.sdk.sqlite3.connect") + update_history( + client=connected_client, + output=tmp_path / "empty-datasets.db", + symbols=["EURUSD"], + datasets=set(), + ) + writer.assert_not_called() + connect.assert_not_called() + + def test_update_history_uses_all_default_timeframes( + self, + connected_client: MagicMock, + mocker: MockerFixture, + tmp_path: Path, + ) -> None: + """Test that timeframes=None writes rates for all default MT5 timeframes.""" + timeframes_written: list[int] = [] + + def capture( + *args: object, + **_kwargs: object, + ) -> tuple[set[Dataset], dict[Dataset, set[str]]]: + timeframes_written.extend(args[4]) # type: ignore[arg-type] + return set(), {} + + mocker.patch("mt5cli.sdk.write_incremental_datasets", side_effect=capture) + update_history( + client=connected_client, + output=tmp_path / "default-timeframes.db", + symbols=["EURUSD"], + datasets={Dataset.rates}, + timeframes=None, + lookback_hours=1, + date_to=datetime(2024, 1, 1, tzinfo=UTC), + ) + assert len(timeframes_written) == len(DEFAULT_HISTORY_TIMEFRAMES) + + def test_update_history_uses_specified_timeframes( + self, + connected_client: MagicMock, + mocker: MockerFixture, + tmp_path: Path, + ) -> None: + """Test explicit timeframes limit rate updates.""" + timeframes_written: list[int] = [] + + def capture( + *args: object, + **_kwargs: object, + ) -> tuple[set[Dataset], dict[Dataset, set[str]]]: + timeframes_written.extend(args[4]) # type: ignore[arg-type] + return set(), {} + + mocker.patch("mt5cli.sdk.write_incremental_datasets", side_effect=capture) + update_history( + client=connected_client, + output=tmp_path / "specific-timeframes.db", + symbols=["EURUSD"], + datasets={Dataset.rates}, + timeframes=["M1", "H1"], + lookback_hours=1, + date_to=datetime(2024, 1, 1, tzinfo=UTC), + ) + assert timeframes_written == [1, 16385] + + def test_update_history_updates_ticks_and_orders( + self, + connected_client: MagicMock, + tmp_path: Path, + ) -> None: + """Test incremental update writes selected ticks and orders datasets.""" + date_to = datetime(2024, 1, 2, tzinfo=UTC) + expected_start = datetime(2024, 1, 1, tzinfo=UTC) + + def make_ticks(**kwargs: object) -> pd.DataFrame: + assert kwargs["symbol"] == "EURUSD" + assert kwargs["date_from"] == expected_start + assert kwargs["date_to"] == date_to + assert kwargs["flags"] == 1 + return pd.DataFrame({ + "time": ["2024-01-01T12:00:00+00:00"], + "time_msc": [1_704_110_400_000], + "bid": [1.1], + }) + + def make_orders(**kwargs: object) -> pd.DataFrame: + assert kwargs["symbol"] == "EURUSD" + assert kwargs["date_from"] == expected_start + assert kwargs["date_to"] == date_to + return pd.DataFrame({ + "ticket": [1], + "symbol": ["EURUSD"], + "time": ["2024-01-01T12:00:00+00:00"], + "type": [0], + }) + + connected_client.copy_ticks_range_as_df.side_effect = make_ticks + connected_client.history_orders_get_as_df.side_effect = make_orders + output = tmp_path / "ticks-orders.db" + update_history( + client=connected_client, + output=output, + symbols=["EURUSD"], + datasets={Dataset.ticks, Dataset.history_orders}, + lookback_hours=24, + date_to=date_to, + ) + with sqlite3.connect(output) as conn: + assert conn.execute("SELECT COUNT(*) FROM ticks").fetchone() == (1,) + assert conn.execute( + "SELECT COUNT(*) FROM history_orders", + ).fetchone() == (1,) + + def test_update_history_with_config_opens_and_closes_connection( + self, + mocker: MockerFixture, + tmp_path: Path, + ) -> None: + """Test update_history_with_config manages MT5 connection lifecycle.""" + mock_client = MagicMock() + mocker.patch("mt5cli.sdk.Mt5DataClient", return_value=mock_client) + updater = mocker.patch("mt5cli.sdk.update_history") + update_history_with_config( + output=tmp_path / "config-wrapper.db", + symbols=["EURUSD"], + datasets={Dataset.history_deals}, + timeframes=["M1"], + flags="ALL", + lookback_hours=1, + date_to=datetime(2024, 1, 1, tzinfo=UTC), + deduplicate=False, + create_rate_views=False, + with_views=True, + include_account_events=False, + ) + mock_client.initialize_and_login_mt5.assert_called_once() + mock_client.shutdown.assert_called_once() + updater.assert_called_once() + assert updater.call_args.kwargs == { + "client": mock_client, + "output": tmp_path / "config-wrapper.db", + "symbols": ["EURUSD"], + "datasets": {Dataset.history_deals}, + "timeframes": ["M1"], + "flags": "ALL", + "lookback_hours": 1, + "date_to": datetime(2024, 1, 1, tzinfo=UTC), + "deduplicate": False, + "create_rate_views": False, + "with_views": True, + "include_account_events": False, + } + + def test_update_history_with_config_validates_before_connecting( + self, + mocker: MockerFixture, + tmp_path: Path, + ) -> None: + """Test invalid inputs fail before MT5 is initialized.""" + mock_client = MagicMock() + mocker.patch("mt5cli.sdk.Mt5DataClient", return_value=mock_client) + with pytest.raises(ValueError, match="lookback_hours must be positive"): + update_history_with_config( + output=tmp_path / "invalid-config.db", + symbols=["EURUSD"], + lookback_hours=0, + ) + mock_client.initialize_and_login_mt5.assert_not_called() + mock_client.shutdown.assert_not_called() + + def test_update_history_with_config_noops_for_empty_datasets( + self, + mocker: MockerFixture, + tmp_path: Path, + ) -> None: + """Test empty dataset selection skips MT5 initialization.""" + mock_client = MagicMock() + mocker.patch("mt5cli.sdk.Mt5DataClient", return_value=mock_client) + updater = mocker.patch("mt5cli.sdk.update_history") + update_history_with_config( + output=tmp_path / "empty-config.db", + symbols=["EURUSD"], + datasets=set(), + ) + mock_client.initialize_and_login_mt5.assert_not_called() + mock_client.shutdown.assert_not_called() + updater.assert_not_called() + + def test_update_history_defaults_date_to_now( + self, + connected_client: MagicMock, + mocker: MockerFixture, + tmp_path: Path, + ) -> None: + """Test update_history uses current UTC time when date_to is omitted.""" + captured: dict[str, datetime] = {} + + def capture( + *args: object, + **_kwargs: object, + ) -> tuple[set[Dataset], dict[Dataset, set[str]]]: + captured["end"] = args[7] # type: ignore[assignment] + return set(), {} + + mocker.patch("mt5cli.sdk.write_incremental_datasets", side_effect=capture) + before = datetime.now(UTC) + update_history( + client=connected_client, + output=tmp_path / "now-default.db", + symbols=["EURUSD"], + datasets={Dataset.rates}, + timeframes=["M1"], + lookback_hours=12, + ) + after = datetime.now(UTC) + assert before <= captured["end"] <= after diff --git a/tests/test_sqlite_history.py b/tests/test_sqlite_history.py new file mode 100644 index 0000000..7771dd1 --- /dev/null +++ b/tests/test_sqlite_history.py @@ -0,0 +1,1497 @@ +"""Tests for mt5cli.sqlite_history module.""" + +from __future__ import annotations + +import logging +import sqlite3 +from datetime import UTC, datetime +from typing import TYPE_CHECKING +from unittest.mock import MagicMock + +import pandas as pd +import pytest + +if TYPE_CHECKING: + from pathlib import Path + +from mt5cli.sqlite_history import ( + DEFAULT_HISTORY_TIMEFRAMES, + append_dataframe, + augment_written_columns_from_sqlite, + build_rate_view_name, + create_cash_events_view, + create_history_indexes, + create_positions_reconstructed_view, + create_rate_compatibility_views, + deduplicate_history_tables, + drop_duplicates_in_table, + filter_incremental_history_deals_frame, + filter_trade_history_frame, + get_history_deals_account_event_start_datetime, + get_incremental_start_datetime, + get_table_columns, + load_incremental_start_datetimes, + parse_sqlite_timestamp, + quote_sqlite_identifier, + record_written_columns, + resolve_granularity_name, + resolve_history_datasets, + resolve_history_tick_flags, + resolve_history_timeframes, + write_collected_datasets, + write_history_dataset, + write_incremental_datasets, + write_rates_dataset, + write_streamed_frame, +) +from mt5cli.utils import TIMEFRAME_MAP, Dataset, IfExists + + +class TestQuoteSqliteIdentifier: + """Tests for quote_sqlite_identifier.""" + + @pytest.mark.parametrize( + "symbol", + ["EUR/USD", "US500.cash", "#US500"], + ) + def test_quotes_broker_specific_symbols(self, symbol: str) -> None: + """Test broker-specific symbols are safely quoted.""" + quoted = quote_sqlite_identifier(f"rate_{symbol}") + assert quoted.startswith('"') + assert quoted.endswith('"') + + +class TestResolveHistorySettings: + """Tests for history dataset and timeframe resolution.""" + + def test_resolve_history_datasets_defaults_and_empty(self) -> None: + """Test dataset resolution distinguishes None from empty selection.""" + assert resolve_history_datasets(None) == set(Dataset) + assert resolve_history_datasets(set()) == set() + + def test_resolve_history_timeframes_defaults(self) -> None: + """Test default timeframes include all fixed MT5 values.""" + resolved = resolve_history_timeframes(None) + assert len(resolved) == len(DEFAULT_HISTORY_TIMEFRAMES) + assert 1 in resolved + assert TIMEFRAME_MAP["H1"] in resolved + + def test_resolve_history_timeframes_deduplicates_aliases(self) -> None: + """Test duplicate aliases for the same timeframe are deduplicated.""" + assert resolve_history_timeframes(["M1", "1", "H1"]) == [1, TIMEFRAME_MAP["H1"]] + + def test_resolve_history_tick_flags(self) -> None: + """Test tick flag resolution.""" + assert resolve_history_tick_flags("ALL") == 1 + assert resolve_history_tick_flags(2) == 2 + + def test_resolve_granularity_name_falls_back_to_integer(self) -> None: + """Test unknown timeframe constants fall back to integer text.""" + assert resolve_granularity_name(999) == "999" + assert resolve_granularity_name(1) == "M1" + + +class TestParseSqliteTimestamp: + """Tests for parse_sqlite_timestamp.""" + + def test_parses_iso_and_pandas_strings(self) -> None: + """Test ISO, pandas-compatible, numeric, and datetime values.""" + assert parse_sqlite_timestamp(None) is None + assert parse_sqlite_timestamp(1_704_067_200) == datetime(2024, 1, 1, tzinfo=UTC) + assert parse_sqlite_timestamp( + datetime.fromisoformat("2024-01-01T00:00:00"), + ) == datetime(2024, 1, 1, tzinfo=UTC) + assert parse_sqlite_timestamp("Jan 1 2024") == datetime(2024, 1, 1, tzinfo=UTC) + assert parse_sqlite_timestamp("not-a-datetime") is None + assert parse_sqlite_timestamp(object()) is None + + +class TestIncrementalStart: + """Tests for get_incremental_start_datetime.""" + + def test_uses_max_time_scoped_by_symbol_and_timeframe( + self, + tmp_path: Path, + ) -> None: + """Test rates increment is scoped by symbol and timeframe.""" + db_path = tmp_path / "scoped-rates.db" + fallback = datetime(2024, 1, 1, tzinfo=UTC) + with sqlite3.connect(db_path) as conn: + conn.execute( + "CREATE TABLE rates(" + " symbol TEXT, timeframe INTEGER, time TEXT, open REAL)", + ) + conn.executemany( + "INSERT INTO rates(symbol, timeframe, time, open) VALUES (?, ?, ?, ?)", + [ + ("EURUSD", 1, "2024-01-02T00:00:00+00:00", 1.0), + ("EURUSD", 16385, "2024-01-03T00:00:00+00:00", 1.1), + ("GBPUSD", 1, "2024-01-04T00:00:00+00:00", 1.2), + ], + ) + assert get_incremental_start_datetime( + conn, + Dataset.rates, + symbol="EURUSD", + timeframe=1, + fallback_start=fallback, + ) == datetime(2024, 1, 2, tzinfo=UTC) + assert get_incremental_start_datetime( + conn, + Dataset.rates, + symbol="EURUSD", + timeframe=16385, + fallback_start=fallback, + ) == datetime(2024, 1, 3, tzinfo=UTC) + + def test_load_incremental_start_datetimes_batches_rates( + self, tmp_path: Path + ) -> None: + """Test grouped rates resume query returns all symbol/timeframe pairs.""" + fallback = datetime(2024, 1, 1, tzinfo=UTC) + with sqlite3.connect(tmp_path / "batch-rates.db") as conn: + conn.execute( + "CREATE TABLE rates(" + " symbol TEXT, timeframe INTEGER, time TEXT, open REAL)", + ) + conn.executemany( + "INSERT INTO rates(symbol, timeframe, time, open) VALUES (?, ?, ?, ?)", + [ + ("EURUSD", 1, "2024-01-02T00:00:00+00:00", 1.0), + ("GBPUSD", 1, "2024-01-03T00:00:00+00:00", 1.1), + ], + ) + starts = load_incremental_start_datetimes( + conn, + Dataset.rates, + symbols=["EURUSD", "GBPUSD"], + timeframes=[1], + fallback_start=fallback, + ) + assert starts["EURUSD", 1] == datetime(2024, 1, 2, tzinfo=UTC) + assert starts["GBPUSD", 1] == datetime(2024, 1, 3, tzinfo=UTC) + + def test_load_incremental_start_datetimes_requires_timeframe_column( + self, + tmp_path: Path, + ) -> None: + """Test rates tables without timeframe fail fast during incremental resume.""" + fallback = datetime(2024, 1, 1, tzinfo=UTC) + with sqlite3.connect(tmp_path / "legacy-rates.db") as conn: + conn.execute("CREATE TABLE rates(symbol TEXT, time TEXT, open REAL)") + conn.execute( + "INSERT INTO rates(symbol, time, open) VALUES (?, ?, ?)", + ("EURUSD", "2024-01-02T00:00:00+00:00", 1.0), + ) + with pytest.raises(ValueError, match="missing: timeframe") as exc_info: + load_incremental_start_datetimes( + conn, + Dataset.rates, + symbols=["EURUSD"], + timeframes=[1], + fallback_start=fallback, + ) + assert "timeframe" in str(exc_info.value) + + def test_load_incremental_start_datetimes_requires_symbol_column( + self, + tmp_path: Path, + ) -> None: + """Test rates tables without symbol fail fast during incremental resume.""" + fallback = datetime(2024, 1, 1, tzinfo=UTC) + with sqlite3.connect(tmp_path / "rates-no-symbol.db") as conn: + conn.execute( + "CREATE TABLE rates(timeframe INTEGER, time TEXT, open REAL)", + ) + with pytest.raises(ValueError, match="missing: symbol") as exc_info: + load_incremental_start_datetimes( + conn, + Dataset.rates, + symbols=["EURUSD"], + timeframes=[1], + fallback_start=fallback, + ) + assert "symbol" in str(exc_info.value) + + def test_load_incremental_start_datetimes_requires_time_column( + self, + tmp_path: Path, + ) -> None: + """Test rates tables without time fail fast during incremental resume.""" + fallback = datetime(2024, 1, 1, tzinfo=UTC) + with sqlite3.connect(tmp_path / "rates-no-time.db") as conn: + conn.execute( + "CREATE TABLE rates(symbol TEXT, timeframe INTEGER, open REAL)", + ) + with pytest.raises(ValueError, match="missing: time") as exc_info: + load_incremental_start_datetimes( + conn, + Dataset.rates, + symbols=["EURUSD"], + timeframes=[1], + fallback_start=fallback, + ) + assert "time" in str(exc_info.value) + + def test_load_incremental_start_datetimes_rejects_unrelated_rates_columns( + self, + tmp_path: Path, + ) -> None: + """Test rates tables with only unrelated columns fail fast.""" + fallback = datetime(2024, 1, 1, tzinfo=UTC) + with sqlite3.connect(tmp_path / "rates-open-only.db") as conn: + conn.execute("CREATE TABLE rates(open REAL)") + with pytest.raises(ValueError, match="missing:") as exc_info: + load_incremental_start_datetimes( + conn, + Dataset.rates, + symbols=["EURUSD"], + timeframes=[1], + fallback_start=fallback, + ) + message = str(exc_info.value) + assert "symbol" in message + assert "timeframe" in message + assert "time" in message + + def test_load_incremental_start_skips_unparseable_max_time( + self, + tmp_path: Path, + ) -> None: + """Test grouped resume ignores rows whose MAX(time) cannot be parsed.""" + fallback = datetime(2024, 1, 1, tzinfo=UTC) + with sqlite3.connect(tmp_path / "bad-max-time.db") as conn: + conn.execute("CREATE TABLE ticks(symbol TEXT, time TEXT)") + conn.execute( + "INSERT INTO ticks(symbol, time) VALUES (?, ?)", + ("EURUSD", "not-a-datetime"), + ) + starts = load_incremental_start_datetimes( + conn, + Dataset.ticks, + symbols=["EURUSD"], + fallback_start=fallback, + ) + assert starts["EURUSD", None] == fallback + + def test_load_incremental_start_uses_table_max_without_symbol_column( + self, + tmp_path: Path, + ) -> None: + """Test grouped resume uses table-wide MAX(time) without symbol column.""" + fallback = datetime(2024, 1, 1, tzinfo=UTC) + with sqlite3.connect(tmp_path / "batch-no-symbol.db") as conn: + conn.execute("CREATE TABLE ticks(time TEXT)") + conn.execute( + "INSERT INTO ticks(time) VALUES (?)", + ("2024-01-02T00:00:00+00:00",), + ) + starts = load_incremental_start_datetimes( + conn, + Dataset.ticks, + symbols=["EURUSD", "GBPUSD"], + fallback_start=fallback, + ) + expected = datetime(2024, 1, 2, tzinfo=UTC) + assert starts["EURUSD", None] == expected + assert starts["GBPUSD", None] == expected + + def test_load_incremental_start_skips_unparseable_rates_max_time( + self, + tmp_path: Path, + ) -> None: + """Test grouped rates resume ignores unparseable MAX(time) values.""" + fallback = datetime(2024, 1, 1, tzinfo=UTC) + with sqlite3.connect(tmp_path / "bad-rates-max-time.db") as conn: + conn.execute( + "CREATE TABLE rates(" + " symbol TEXT, timeframe INTEGER, time TEXT, open REAL)", + ) + conn.execute( + "INSERT INTO rates(symbol, timeframe, time, open) VALUES (?, ?, ?, ?)", + ("EURUSD", 1, "not-a-datetime", 1.0), + ) + starts = load_incremental_start_datetimes( + conn, + Dataset.rates, + symbols=["EURUSD"], + timeframes=[1], + fallback_start=fallback, + ) + assert starts["EURUSD", 1] == fallback + + +class TestDeduplication: + """Tests for SQLite deduplication helpers.""" + + def test_append_dedup_keeps_latest_rowid(self, tmp_path: Path) -> None: + """Test deduplication keeps the latest ROWID for stable keys.""" + db_path = tmp_path / "dedup.db" + with sqlite3.connect(db_path) as conn: + conn.execute( + "CREATE TABLE rates(" + " symbol TEXT, timeframe INTEGER, time TEXT, open REAL)", + ) + conn.executemany( + "INSERT INTO rates(symbol, timeframe, time, open) VALUES (?, ?, ?, ?)", + [ + ("EURUSD", 1, "2024-01-01T00:00:00+00:00", 1.0), + ("EURUSD", 1, "2024-01-01T00:00:00+00:00", 9.9), + ], + ) + deduplicate_history_tables( + conn, + {Dataset.rates: {"symbol", "timeframe", "time", "open"}}, + {Dataset.rates}, + ) + assert conn.execute("SELECT COUNT(*) FROM rates").fetchone() == (1,) + assert conn.execute("SELECT open FROM rates").fetchone() == (9.9,) + + def test_drop_duplicates_rejects_invalid_identifiers(self) -> None: + """Test invalid table or column names raise ValueError.""" + cursor = sqlite3.connect(":memory:").cursor() + with pytest.raises(ValueError, match="Invalid table name"): + drop_duplicates_in_table(cursor, "bad table", ["id"]) + with pytest.raises(ValueError, match="Invalid column names"): + drop_duplicates_in_table(cursor, "rates", ["bad column"]) + + @pytest.mark.parametrize( + ("dataset", "table_sql", "insert_sql", "rows", "columns"), + [ + ( + Dataset.ticks, + ( + "CREATE TABLE ticks(" + " symbol TEXT, time_msc INTEGER, time TEXT, bid REAL)" + ), + "INSERT INTO ticks(symbol, time_msc, time, bid) VALUES (?, ?, ?, ?)", + [ + ("EURUSD", 1, "2024-01-01T00:00:00+00:00", 1.0), + ("EURUSD", 1, "2024-01-01T00:00:00+00:00", 9.9), + ], + {"symbol", "time_msc", "time", "bid"}, + ), + ( + Dataset.history_orders, + ( + "CREATE TABLE history_orders(" + " ticket INTEGER, symbol TEXT, time TEXT, type INTEGER)" + ), + ( + "INSERT INTO history_orders(ticket, symbol, time, type)" + " VALUES (?, ?, ?, ?)" + ), + [ + (1, "EURUSD", "2024-01-01T00:00:00+00:00", 0), + (1, "EURUSD", "2024-01-01T00:00:00+00:00", 1), + ], + {"ticket", "symbol", "time", "type"}, + ), + ( + Dataset.history_deals, + ( + "CREATE TABLE history_deals(" + " ticket INTEGER, symbol TEXT, time TEXT, type INTEGER," + " entry INTEGER)" + ), + ( + "INSERT INTO history_deals(ticket, symbol, time, type, entry)" + " VALUES (?, ?, ?, ?, ?)" + ), + [ + (1, "EURUSD", "2024-01-01T00:00:00+00:00", 0, 0), + (1, "EURUSD", "2024-01-01T00:00:00+00:00", 0, 1), + ], + {"ticket", "symbol", "time", "type", "entry"}, + ), + ], + ) + def test_deduplicates_non_rate_datasets_by_stable_keys( + self, + tmp_path: Path, + dataset: Dataset, + table_sql: str, + insert_sql: str, + rows: list[tuple[object, ...]], + columns: set[str], + ) -> None: + """Test deduplication keys for ticks, orders, and deals.""" + with sqlite3.connect(tmp_path / f"{dataset.value}-dedup.db") as conn: + conn.execute(table_sql) + conn.executemany(insert_sql, rows) + deduplicate_history_tables(conn, {dataset: columns}, {dataset}) + assert conn.execute( + f"SELECT COUNT(*) FROM {dataset.table_name}", # noqa: S608 + ).fetchone() == (1,) + + def test_scoped_dedup_preserves_older_rows(self, tmp_path: Path) -> None: + """Test scoped deduplication only rewrites the appended boundary.""" + boundary = datetime(2024, 1, 2, tzinfo=UTC) + with sqlite3.connect(tmp_path / "scoped-dedup.db") as conn: + conn.execute( + "CREATE TABLE rates(" + " symbol TEXT, timeframe INTEGER, time TEXT, open REAL)", + ) + conn.executemany( + "INSERT INTO rates(symbol, timeframe, time, open) VALUES (?, ?, ?, ?)", + [ + ("EURUSD", 1, "2024-01-01T00:00:00+00:00", 1.0), + ("EURUSD", 1, "2024-01-02T00:00:00+00:00", 2.0), + ("EURUSD", 1, "2024-01-02T00:00:00+00:00", 9.9), + ], + ) + deduplicate_history_tables( + conn, + {Dataset.rates: {"symbol", "timeframe", "time", "open"}}, + {Dataset.rates}, + { + Dataset.rates: [ + ( + "symbol = ? AND timeframe = ? AND time >= ?", + ("EURUSD", 1, boundary), + ), + ], + }, + ) + rows = conn.execute( + "SELECT time, open FROM rates ORDER BY time, open", + ).fetchall() + assert rows == [ + ("2024-01-01T00:00:00+00:00", 1.0), + ("2024-01-02T00:00:00+00:00", 9.9), + ] + + +class TestRateCompatibilityViews: + """Tests for rate compatibility view creation.""" + + def test_creates_views_for_multiple_timeframes(self, tmp_path: Path) -> None: + """Test rate views are created per symbol and timeframe.""" + db_path = tmp_path / "rate-views.db" + with sqlite3.connect(db_path) as conn: + conn.execute( + "CREATE TABLE rates(" + " symbol TEXT, timeframe INTEGER, time TEXT, close REAL)", + ) + conn.executemany( + "INSERT INTO rates(symbol, timeframe, time, close) VALUES (?, ?, ?, ?)", + [ + ("EURUSD", 1, "2024-01-01T00:00:00+00:00", 1.0), + ("EURUSD", 16385, "2024-01-01T01:00:00+00:00", 1.1), + ], + ) + create_rate_compatibility_views(conn) + views = { + row[0] + for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type='view'" + " AND name LIKE 'rate_EURUSD%'", + ).fetchall() + } + assert views == {"rate_EURUSD__M1_1", "rate_EURUSD__H1_16385"} + assert conn.execute('SELECT close FROM "rate_EURUSD__M1_1"').fetchone() == ( + 1.0, + ) + + def test_drops_stale_single_timeframe_view(self, tmp_path: Path) -> None: + """Test stale rate_ views are removed when timeframes change.""" + db_path = tmp_path / "stale-rate-view.db" + with sqlite3.connect(db_path) as conn: + conn.execute( + "CREATE TABLE rates(" + " symbol TEXT, timeframe INTEGER, time TEXT, close REAL)", + ) + conn.execute( + "INSERT INTO rates(symbol, timeframe, time, close) VALUES (?, ?, ?, ?)", + ("EURUSD", 1, "2024-01-01T00:00:00+00:00", 1.0), + ) + create_rate_compatibility_views(conn) + assert conn.execute( + "SELECT 1 FROM sqlite_master" + " WHERE type='view' AND name = 'rate_EURUSD__1'", + ).fetchone() == (1,) + + conn.execute( + "INSERT INTO rates(symbol, timeframe, time, close) VALUES (?, ?, ?, ?)", + ("EURUSD", TIMEFRAME_MAP["H1"], "2024-01-01T01:00:00+00:00", 1.1), + ) + create_rate_compatibility_views(conn) + views = { + row[0] + for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type='view'" + " AND name LIKE 'rate_EURUSD%'", + ).fetchall() + } + assert views == {"rate_EURUSD__M1_1", "rate_EURUSD__H1_16385"} + assert "rate_EURUSD__1" not in views + + def test_does_not_drop_views_outside_rate_prefix(self, tmp_path: Path) -> None: + """Test only literal rate_* views are dropped during recreation.""" + db_path = tmp_path / "rate-prefix-views.db" + with sqlite3.connect(db_path) as conn: + conn.execute( + "CREATE TABLE rates(" + " symbol TEXT, timeframe INTEGER, time TEXT, close REAL)", + ) + conn.execute( + "INSERT INTO rates(symbol, timeframe, time, close) VALUES (?, ?, ?, ?)", + ("EURUSD", 1, "2024-01-01T00:00:00+00:00", 1.0), + ) + conn.execute("CREATE VIEW rate_EURUSD AS SELECT 1 AS close") + conn.execute("CREATE VIEW rateX_custom AS SELECT 2 AS close") + create_rate_compatibility_views(conn) + views = { + row[0] + for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type='view'", + ).fetchall() + } + assert "rateX_custom" in views + assert "rate_EURUSD__1" in views + assert conn.execute("SELECT close FROM rate_EURUSD__1").fetchone() == (1.0,) + + @pytest.mark.parametrize( + "symbol", + ["EUR/USD", "US500.cash", "#US500"], + ) + def test_supports_broker_specific_symbols( + self, + tmp_path: Path, + symbol: str, + ) -> None: + """Test broker-specific symbols get readable rate views.""" + db_path = tmp_path / "broker-symbol.db" + view_name = build_rate_view_name( + symbol=symbol, + granularity="M1", + granularity_count=1, + timeframe=1, + ) + with sqlite3.connect(db_path) as conn: + conn.execute( + "CREATE TABLE rates(" + " symbol TEXT, timeframe INTEGER, time TEXT, close REAL)", + ) + conn.execute( + "INSERT INTO rates(symbol, timeframe, time, close) VALUES (?, ?, ?, ?)", + (symbol, 1, "2024-01-01T00:00:00+00:00", 1.0), + ) + create_rate_compatibility_views(conn) + assert conn.execute( + "SELECT 1 FROM sqlite_master WHERE type='view' AND name = ?", + (view_name,), + ).fetchone() == (1,) + quoted = quote_sqlite_identifier(view_name) + query = "SELECT close FROM " + quoted # noqa: S608 + assert conn.execute(query).fetchone() == (1.0,) + + +class TestDerivedViews: + """Tests for cash_events and positions_reconstructed views.""" + + def test_creates_views_when_columns_present(self, tmp_path: Path) -> None: + """Test cash_events and positions_reconstructed views are created.""" + db_path = tmp_path / "derived-views.db" + columns = { + "ticket", + "position_id", + "symbol", + "time", + "type", + "entry", + "volume", + "price", + "profit", + } + with sqlite3.connect(db_path) as conn: + conn.execute( + "CREATE TABLE history_deals(" + " ticket INTEGER, position_id INTEGER, symbol TEXT, time INTEGER," + " type INTEGER, entry INTEGER, volume REAL, price REAL, profit REAL)", + ) + conn.executemany( + "INSERT INTO history_deals VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + [ + (1, 100, "EURUSD", 1, 0, 0, 1.0, 1.1, 0.0), + (2, 100, "EURUSD", 2, 0, 1, 1.0, 1.2, 5.0), + (3, 0, "", 3, 2, 0, 0.0, 0.0, 10.0), + ], + ) + assert create_cash_events_view(conn, columns) + assert create_positions_reconstructed_view(conn, columns) + assert conn.execute("SELECT COUNT(*) FROM cash_events").fetchone() == (1,) + assert conn.execute( + "SELECT position_id FROM positions_reconstructed", + ).fetchall() == [(100,)] + + +class TestFilterTradeHistoryFrame: + """Tests for filter_trade_history_frame.""" + + def test_includes_account_events_when_requested(self) -> None: + """Test account events are kept alongside selected symbols.""" + frame = pd.DataFrame({ + "symbol": ["EURUSD", None, "OTHER"], + "type": [0, 2, 0], + }) + filtered = filter_trade_history_frame( + frame, + ["EURUSD"], + include_account_events=True, + ) + assert filtered["symbol"].tolist()[0] == "EURUSD" + assert pd.isna(filtered["symbol"].tolist()[1]) + + +class TestIncrementalHistoryDealsHelpers: + """Tests for incremental history_deals helper functions.""" + + def test_account_event_start_uses_type_column(self, tmp_path: Path) -> None: + """Test account-event start uses non-trade deal types when available.""" + fallback = datetime(2024, 1, 1, tzinfo=UTC) + with sqlite3.connect(tmp_path / "account-start-type.db") as conn: + conn.execute( + "CREATE TABLE history_deals(" + " ticket INTEGER, symbol TEXT, time TEXT, type INTEGER)", + ) + conn.executemany( + "INSERT INTO history_deals(ticket, symbol, time, type)" + " VALUES (?, ?, ?, ?)", + [ + (1, "EURUSD", "2024-01-05T00:00:00+00:00", 0), + (2, "", "2024-01-08T00:00:00+00:00", 2), + ], + ) + assert get_history_deals_account_event_start_datetime( + conn, + fallback_start=fallback, + ) == datetime(2024, 1, 8, tzinfo=UTC) + + def test_account_event_start_falls_back_to_empty_symbol( + self, tmp_path: Path + ) -> None: + """Test account-event start uses empty symbols when type is missing.""" + fallback = datetime(2024, 1, 1, tzinfo=UTC) + with sqlite3.connect(tmp_path / "account-start-symbol.db") as conn: + conn.execute( + "CREATE TABLE history_deals(ticket INTEGER, symbol TEXT, time TEXT)", + ) + conn.executemany( + "INSERT INTO history_deals(ticket, symbol, time) VALUES (?, ?, ?)", + [ + (1, "EURUSD", "2024-01-05T00:00:00+00:00"), + (2, "", "2024-01-07T00:00:00+00:00"), + ], + ) + assert get_history_deals_account_event_start_datetime( + conn, + fallback_start=fallback, + ) == datetime(2024, 1, 7, tzinfo=UTC) + + def test_account_event_start_without_identifying_columns( + self, + tmp_path: Path, + ) -> None: + """Test account-event start falls back when type and symbol are missing.""" + fallback = datetime(2024, 1, 1, tzinfo=UTC) + with sqlite3.connect(tmp_path / "account-start-fallback.db") as conn: + conn.execute("CREATE TABLE history_deals(ticket INTEGER, time TEXT)") + conn.execute( + "INSERT INTO history_deals(ticket, time) VALUES (?, ?)", + (1, "2024-01-05T00:00:00+00:00"), + ) + assert ( + get_history_deals_account_event_start_datetime( + conn, + fallback_start=fallback, + ) + == fallback + ) + + def test_filter_incremental_history_deals_frame(self) -> None: + """Test incremental deal filtering applies per-symbol and account starts.""" + frame = pd.DataFrame({ + "ticket": [1, 2, 3, 4, 5], + "symbol": ["EURUSD", "EURUSD", "GBPUSD", "OTHER", ""], + "time": [ + "2024-01-05T00:00:00+00:00", + "2024-01-11T00:00:00+00:00", + "2024-01-02T00:00:00+00:00", + "2024-01-02T00:00:00+00:00", + "2024-01-03T00:00:00+00:00", + ], + "type": [0, 0, 0, 0, 2], + }) + start_by_symbol = { + "EURUSD": datetime(2024, 1, 10, tzinfo=UTC), + "GBPUSD": datetime(2024, 1, 1, tzinfo=UTC), + } + filtered = filter_incremental_history_deals_frame( + frame, + ["EURUSD", "GBPUSD"], + start_by_symbol, + datetime(2024, 1, 2, tzinfo=UTC), + ) + assert filtered["ticket"].tolist() == [2, 3, 5] + + def test_filter_incremental_excludes_symbolized_account_events_from_trade_cursor( + self, + ) -> None: + """Test symbolized account events follow only account_event_start. + + An account-event row with a matching symbol must not be kept by the + EURUSD trade cursor when its time is after the trade start but before + account_event_start. + """ + trade_cursor = datetime(2024, 1, 1, tzinfo=UTC) + account_event_start = datetime(2024, 1, 10, tzinfo=UTC) + row_time = datetime(2024, 1, 5, tzinfo=UTC) + frame = pd.DataFrame({ + "ticket": [1], + "symbol": ["EURUSD"], + "time": [row_time.isoformat()], + "type": [2], + }) + filtered = filter_incremental_history_deals_frame( + frame, + ["EURUSD"], + {"EURUSD": trade_cursor}, + account_event_start, + ) + assert filtered.empty + + def test_filter_incremental_rejects_rows_without_parseable_time(self) -> None: + """Test incremental filtering drops rows when time cannot be parsed.""" + frame = pd.DataFrame({ + "ticket": [1], + "symbol": ["EURUSD"], + "time": ["not-a-datetime"], + "type": [0], + }) + filtered = filter_incremental_history_deals_frame( + frame, + ["EURUSD"], + {"EURUSD": datetime(2024, 1, 1, tzinfo=UTC)}, + datetime(2024, 1, 1, tzinfo=UTC), + ) + assert filtered.empty + + def test_filter_incremental_keeps_account_events_without_symbol_column( + self, + ) -> None: + """Test account events are kept when history_deals has no symbol column.""" + frame = pd.DataFrame({ + "ticket": [1], + "time": ["2024-01-03T00:00:00+00:00"], + "type": [2], + }) + filtered = filter_incremental_history_deals_frame( + frame, + ["EURUSD"], + {"EURUSD": datetime(2024, 1, 10, tzinfo=UTC)}, + datetime(2024, 1, 2, tzinfo=UTC), + ) + assert filtered["ticket"].tolist() == [1] + + def test_filter_incremental_skips_trade_rows_without_symbol_column(self) -> None: + """Test trade rows are excluded when symbol column is unavailable.""" + frame = pd.DataFrame({ + "ticket": [1], + "time": ["2024-01-03T00:00:00+00:00"], + }) + filtered = filter_incremental_history_deals_frame( + frame, + ["EURUSD"], + {"EURUSD": datetime(2024, 1, 1, tzinfo=UTC)}, + datetime(2024, 1, 1, tzinfo=UTC), + ) + assert filtered.empty + + def test_filter_incremental_rejects_rows_without_time_column(self) -> None: + """Test incremental filtering drops rows when time column is missing.""" + frame = pd.DataFrame({ + "ticket": [1], + "symbol": ["EURUSD"], + "type": [0], + }) + filtered = filter_incremental_history_deals_frame( + frame, + ["EURUSD"], + {"EURUSD": datetime(2024, 1, 1, tzinfo=UTC)}, + datetime(2024, 1, 1, tzinfo=UTC), + ) + assert filtered.empty + + +class TestIncrementalIntegration: + """Integration tests for incremental write helpers.""" + + def test_write_incremental_datasets_end_to_end( + self, + tmp_path: Path, + ) -> None: + """Test incremental dataset writer covers finalize branches.""" + client = MagicMock() + client.copy_rates_range_as_df.return_value = pd.DataFrame({ + "time": ["2024-01-01T00:00:00+00:00"], + "open": [1.0], + }) + client.copy_ticks_range_as_df.return_value = pd.DataFrame() + client.history_orders_get_as_df.return_value = pd.DataFrame({ + "ticket": [1], + "symbol": ["EURUSD"], + "time": ["2024-01-01T00:00:00+00:00"], + "type": [0], + }) + client.history_deals_get_as_df.return_value = pd.DataFrame({ + "ticket": [2], + "position_id": [100], + "symbol": ["EURUSD"], + "time": ["2024-01-01T00:00:00+00:00"], + "type": [0], + "entry": [0], + "volume": [1.0], + "price": [1.1], + "profit": [0.0], + }) + db_path = tmp_path / "incremental-integration.db" + start = datetime(2024, 1, 1, tzinfo=UTC) + end = datetime(2024, 1, 2, tzinfo=UTC) + with sqlite3.connect(db_path) as conn: + write_incremental_datasets( + conn, + client, + ["EURUSD"], + set(Dataset), + [1], + 1, + start, + end, + deduplicate=True, + create_rate_views=True, + with_views=True, + include_account_events=True, + ) + tables = { + row[0] + for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'", + ).fetchall() + } + views = { + row[0] + for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type='view'", + ).fetchall() + } + assert {"rates", "history_orders", "history_deals"} <= tables + assert "rate_EURUSD__1" in views + assert "cash_events" in views + + def test_rate_view_names_do_not_collide_for_symbol_suffix( + self, + tmp_path: Path, + ) -> None: + """Test symbols resembling generated view suffixes stay distinct.""" + db_path = tmp_path / "rate-view-collision.db" + with sqlite3.connect(db_path) as conn: + conn.execute( + "CREATE TABLE rates(" + " symbol TEXT, timeframe INTEGER, time TEXT, close REAL)", + ) + conn.executemany( + "INSERT INTO rates(symbol, timeframe, time, close) VALUES (?, ?, ?, ?)", + [ + ("EURUSD", 1, "2024-01-01T00:00:00+00:00", 1.0), + ("EURUSD", 16385, "2024-01-01T01:00:00+00:00", 1.1), + ("EURUSD_M1", 1, "2024-01-01T02:00:00+00:00", 1.2), + ], + ) + create_rate_compatibility_views(conn) + views = { + row[0] + for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type='view'", + ).fetchall() + } + assert views == { + "rate_EURUSD__M1_1", + "rate_EURUSD__H1_16385", + "rate_EURUSD_M1__1", + } + + def test_write_collected_datasets_and_edge_branches( + self, + tmp_path: Path, + ) -> None: + """Test collected dataset writer and helper edge branches.""" + client = MagicMock() + client.copy_rates_range_as_df.return_value = pd.DataFrame() + client.copy_ticks_range_as_df.return_value = pd.DataFrame({ + "time": ["2024-01-01T00:00:00+00:00"], + "bid": [1.0], + }) + client.history_orders_get_as_df.return_value = pd.DataFrame({ + "ticket": [1], + "symbol": ["EURUSD"], + "time": [1], + "type": [0], + }) + client.history_deals_get_as_df.return_value = pd.DataFrame({ + "ticket": [1], + "symbol": ["EURUSD"], + "time": [1], + "type": [0], + }) + db_path = tmp_path / "collected-integration.db" + start = datetime(2024, 1, 1, tzinfo=UTC) + end = datetime(2024, 1, 2, tzinfo=UTC) + with sqlite3.connect(db_path) as conn: + write_collected_datasets( + conn, + client, + ["EURUSD"], + {Dataset.ticks, Dataset.history_orders, Dataset.history_deals}, + 1, + 1, + start, + end, + IfExists.APPEND, + ) + assert ( + get_incremental_start_datetime( + conn, + Dataset.ticks, + symbol="EURUSD", + timeframe=None, + fallback_start=start, + ) + == start + ) + deduplicate_history_tables( + conn, + {Dataset.ticks: {"time"}}, + {Dataset.ticks}, + ) + filtered = filter_trade_history_frame( + pd.DataFrame({"ticket": [1]}), + ["EURUSD"], + include_account_events=False, + ) + assert len(filtered) == 1 + account_filtered = filter_trade_history_frame( + pd.DataFrame({"symbol": ["", "EURUSD"], "type": [2, 0]}), + ["EURUSD"], + include_account_events=True, + ) + assert len(account_filtered) == 2 + no_type_filtered = filter_trade_history_frame( + pd.DataFrame({"symbol": ["", "EURUSD"]}), + ["EURUSD"], + include_account_events=True, + ) + assert len(no_type_filtered) == 2 + + def test_get_incremental_start_without_symbol_column( + self, + tmp_path: Path, + ) -> None: + """Test incremental start ignores missing symbol column filters.""" + db_path = tmp_path / "no-symbol-column.db" + fallback = datetime(2024, 1, 1, tzinfo=UTC) + with sqlite3.connect(db_path) as conn: + conn.execute("CREATE TABLE ticks(time TEXT)") + conn.execute( + "INSERT INTO ticks(time) VALUES (?)", + ("2024-01-02T00:00:00+00:00",), + ) + assert get_incremental_start_datetime( + conn, + Dataset.ticks, + symbol="EURUSD", + timeframe=None, + fallback_start=fallback, + ) == datetime(2024, 1, 2, tzinfo=UTC) + + def test_deduplicate_skips_unsupported_keys( + self, + tmp_path: Path, + caplog: pytest.LogCaptureFixture, + ) -> None: + """Test deduplication logs when no stable key columns exist.""" + with ( + sqlite3.connect(tmp_path / "no-keys.db") as conn, + caplog.at_level( + logging.WARNING, + logger="mt5cli.sqlite_history", + ), + ): + deduplicate_history_tables(conn, {Dataset.ticks: {"time"}}, {Dataset.ticks}) + assert "Skipping ticks deduplication" in caplog.text + + def test_write_rates_skips_empty_schema( + self, + tmp_path: Path, + ) -> None: + """Test rates writer skips frames with no columns after normalization.""" + client = MagicMock() + client.copy_rates_range_as_df.return_value = pd.DataFrame() + written_columns: dict[Dataset, set[str]] = {} + with sqlite3.connect(tmp_path / "empty-rates.db") as conn: + assert not write_rates_dataset( + conn, + client, + ["EURUSD"], + 1, + datetime(2024, 1, 1, tzinfo=UTC), + datetime(2024, 1, 2, tzinfo=UTC), + IfExists.APPEND, + written_columns, + ) + + def test_finalize_with_views_warning_when_deals_missing( + self, + tmp_path: Path, + caplog: pytest.LogCaptureFixture, + ) -> None: + """Test with_views warning when history_deals was not written.""" + client = MagicMock() + client.copy_rates_range_as_df.return_value = pd.DataFrame() + with ( + caplog.at_level(logging.WARNING, logger="mt5cli.sqlite_history"), + sqlite3.connect(tmp_path / "views-warning.db") as conn, + ): + write_incremental_datasets( + conn, + client, + ["EURUSD"], + {Dataset.rates, Dataset.history_deals}, + [1], + 0, + datetime(2024, 1, 1, tzinfo=UTC), + datetime(2024, 1, 2, tzinfo=UTC), + deduplicate=False, + create_rate_views=False, + with_views=True, + include_account_events=True, + ) + assert "with_views ignored" in caplog.text + + def test_augment_written_columns_creates_new_entry( + self, + tmp_path: Path, + ) -> None: + """Test augment helper initializes dataset column maps.""" + db_path = tmp_path / "augment-new.db" + written_columns: dict[Dataset, set[str]] = {} + with sqlite3.connect(db_path) as conn: + conn.execute("CREATE TABLE rates(time TEXT)") + augment_written_columns_from_sqlite( + conn, + {Dataset.rates}, + written_columns, + ) + assert written_columns == {Dataset.rates: {"time"}} + + def test_create_rate_views_noop_without_required_columns( + self, + tmp_path: Path, + ) -> None: + """Test rate view creation skips tables missing required columns.""" + with sqlite3.connect(tmp_path / "missing-rate-columns.db") as conn: + conn.execute("CREATE TABLE rates(open REAL)") + create_rate_compatibility_views(conn) + views = conn.execute( + "SELECT name FROM sqlite_master WHERE type='view'", + ).fetchall() + assert views == [] + + def test_incremental_history_noops_when_fetch_returns_empty( + self, + tmp_path: Path, + ) -> None: + """Test incremental history skips table tracking for empty writes.""" + client = MagicMock() + client.history_orders_get_as_df.return_value = pd.DataFrame() + client.history_deals_get_as_df.return_value = pd.DataFrame() + start = datetime(2024, 1, 1, tzinfo=UTC) + end = datetime(2024, 1, 2, tzinfo=UTC) + with sqlite3.connect(tmp_path / "empty-history.db") as conn: + written_tables, _ = write_incremental_datasets( + conn, + client, + ["EURUSD"], + {Dataset.history_orders, Dataset.history_deals}, + [], + 0, + start, + end, + deduplicate=False, + create_rate_views=False, + with_views=False, + include_account_events=True, + ) + assert written_tables == set() + + def test_resolve_history_tick_flags_invalid(self) -> None: + """Test invalid tick flags raise ValueError.""" + with pytest.raises(ValueError, match="Invalid tick flags"): + resolve_history_tick_flags("BAD") + + def test_resolve_history_timeframes_invalid(self) -> None: + """Test invalid timeframes raise ValueError.""" + with pytest.raises(ValueError, match="Invalid timeframe"): + resolve_history_timeframes(["BAD"]) + + +class TestIncrementalHistoryDeals: + """Tests for incremental history_deals account-event handling.""" + + def test_fetches_per_symbol_when_account_events_disabled( + self, + tmp_path: Path, + ) -> None: + """Test history_deals are fetched per symbol without account events.""" + client = MagicMock() + client.history_deals_get_as_df.return_value = pd.DataFrame({ + "ticket": [1], + "symbol": ["EURUSD"], + "time": [1], + "type": [0], + "entry": [0], + }) + start = datetime(2024, 1, 1, tzinfo=UTC) + end = datetime(2024, 1, 2, tzinfo=UTC) + with sqlite3.connect(tmp_path / "per-symbol-deals.db") as conn: + write_incremental_datasets( + conn, + client, + ["EURUSD", "GBPUSD"], + {Dataset.history_deals}, + [], + 0, + start, + end, + deduplicate=False, + create_rate_views=False, + with_views=False, + include_account_events=False, + ) + assert client.history_deals_get_as_df.call_count == 2 + + def test_skips_history_deals_when_fetch_returns_empty( + self, + tmp_path: Path, + ) -> None: + """Test incremental history_deals skips tracking when writes are empty.""" + client = MagicMock() + client.history_deals_get_as_df.return_value = pd.DataFrame() + start = datetime(2024, 1, 1, tzinfo=UTC) + end = datetime(2024, 1, 2, tzinfo=UTC) + with sqlite3.connect(tmp_path / "empty-per-symbol-deals.db") as conn: + written_tables, _ = write_incremental_datasets( + conn, + client, + ["EURUSD", "GBPUSD"], + {Dataset.history_deals}, + [], + 0, + start, + end, + deduplicate=False, + create_rate_views=False, + with_views=False, + include_account_events=False, + ) + assert written_tables == set() + + def test_write_history_dataset_fetches_account_events_once( + self, + tmp_path: Path, + ) -> None: + """Test write_history_dataset account-event path fetches once.""" + client = MagicMock() + client.history_deals_get_as_df.return_value = pd.DataFrame({ + "ticket": [1, 2], + "symbol": ["EURUSD", ""], + "time": [1, 2], + "type": [0, 2], + "entry": [0, 0], + }) + start = datetime(2024, 1, 1, tzinfo=UTC) + end = datetime(2024, 1, 2, tzinfo=UTC) + written_columns: dict[Dataset, set[str]] = {} + with sqlite3.connect(tmp_path / "history-dataset-account.db") as conn: + assert write_history_dataset( + conn, + client.history_deals_get_as_df, + Dataset.history_deals, + ["EURUSD"], + start, + end, + IfExists.APPEND, + written_columns, + include_account_events=True, + ) + rows = conn.execute( + "SELECT ticket, symbol, type FROM history_deals ORDER BY ticket", + ).fetchall() + client.history_deals_get_as_df.assert_called_once() + assert rows == [(1, "EURUSD", 0), (2, "", 2)] + + def test_fetches_account_events_once_for_multiple_symbols( + self, + tmp_path: Path, + ) -> None: + """Test account events are fetched once and not duplicated.""" + client = MagicMock() + client.history_deals_get_as_df.return_value = pd.DataFrame({ + "ticket": [1, 2, 3, 4], + "symbol": ["EURUSD", "GBPUSD", "OTHER", ""], + "time": [ + "2024-01-01T00:00:00+00:00", + "2024-01-01T01:00:00+00:00", + "2024-01-01T02:00:00+00:00", + "2024-01-01T03:00:00+00:00", + ], + "type": [0, 0, 0, 2], + "entry": [0, 0, 0, 0], + }) + start = datetime(2024, 1, 1, tzinfo=UTC) + end = datetime(2024, 1, 2, tzinfo=UTC) + with sqlite3.connect(tmp_path / "account-events.db") as conn: + write_incremental_datasets( + conn, + client, + ["EURUSD", "GBPUSD"], + {Dataset.history_deals}, + [], + 0, + start, + end, + deduplicate=False, + create_rate_views=False, + with_views=False, + include_account_events=True, + ) + rows = conn.execute( + "SELECT ticket, symbol, type FROM history_deals ORDER BY ticket", + ).fetchall() + row_count = conn.execute("SELECT COUNT(*) FROM history_deals").fetchone() + client.history_deals_get_as_df.assert_called_once() + assert rows == [(1, "EURUSD", 0), (2, "GBPUSD", 0), (4, "", 2)] + assert row_count == (3,) + + def test_records_account_event_dedup_scope_without_type_column( + self, + tmp_path: Path, + ) -> None: + """Test account-event dedup scope falls back to empty symbols.""" + client = MagicMock() + client.history_deals_get_as_df.return_value = pd.DataFrame({ + "ticket": [1], + "symbol": [""], + "time": ["2024-01-02T00:00:00+00:00"], + }) + start = datetime(2024, 1, 1, tzinfo=UTC) + end = datetime(2024, 1, 3, tzinfo=UTC) + with sqlite3.connect(tmp_path / "legacy-deals.db") as conn: + conn.execute( + "CREATE TABLE history_deals( ticket INTEGER, symbol TEXT, time TEXT)", + ) + write_incremental_datasets( + conn, + client, + ["EURUSD"], + {Dataset.history_deals}, + [], + 0, + start, + end, + deduplicate=True, + create_rate_views=False, + with_views=False, + include_account_events=True, + ) + assert conn.execute("SELECT COUNT(*) FROM history_deals").fetchone() == (1,) + + def test_skips_symbol_dedup_scope_when_symbol_column_missing( + self, + tmp_path: Path, + ) -> None: + """Test account-event writes skip per-symbol dedup without symbol column.""" + client = MagicMock() + client.history_deals_get_as_df.return_value = pd.DataFrame({ + "ticket": [1], + "time": ["2024-01-02T00:00:00+00:00"], + "type": [2], + }) + start = datetime(2024, 1, 1, tzinfo=UTC) + end = datetime(2024, 1, 3, tzinfo=UTC) + with sqlite3.connect(tmp_path / "no-symbol-deals.db") as conn: + conn.execute( + "CREATE TABLE history_deals(ticket INTEGER, time TEXT, type INTEGER)", + ) + write_incremental_datasets( + conn, + client, + ["EURUSD"], + {Dataset.history_deals}, + [], + 0, + start, + end, + deduplicate=True, + create_rate_views=False, + with_views=False, + include_account_events=True, + ) + assert conn.execute("SELECT COUNT(*) FROM history_deals").fetchone() == (1,) + + def test_creates_views_when_history_deals_written_with_with_views( + self, + tmp_path: Path, + ) -> None: + """Test derived views are rebuilt when history_deals is written.""" + client = MagicMock() + client.history_deals_get_as_df.return_value = pd.DataFrame({ + "ticket": [1, 2], + "position_id": [100, 100], + "symbol": ["EURUSD", "EURUSD"], + "time": ["2024-01-01T00:00:00+00:00", "2024-01-02T00:00:00+00:00"], + "type": [0, 1], + "entry": [0, 1], + "volume": [1.0, 1.0], + "price": [1.1, 1.2], + "profit": [0.0, 5.0], + }) + start = datetime(2024, 1, 1, tzinfo=UTC) + end = datetime(2024, 1, 3, tzinfo=UTC) + with sqlite3.connect(tmp_path / "deals-with-views.db") as conn: + write_incremental_datasets( + conn, + client, + ["EURUSD"], + {Dataset.history_deals}, + [], + 0, + start, + end, + deduplicate=False, + create_rate_views=False, + with_views=True, + include_account_events=False, + ) + views = { + row[0] + for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type='view'", + ).fetchall() + } + assert {"cash_events", "positions_reconstructed"} <= views + + def test_applies_per_symbol_start_when_account_events_enabled( + self, + tmp_path: Path, + ) -> None: + """Test account-event fetch keeps per-symbol incremental boundaries.""" + client = MagicMock() + client.history_deals_get_as_df.return_value = pd.DataFrame({ + "ticket": [1, 2, 3, 4, 5], + "symbol": ["EURUSD", "EURUSD", "GBPUSD", "OTHER", ""], + "time": [ + "2024-01-05T00:00:00+00:00", + "2024-01-11T00:00:00+00:00", + "2024-01-02T00:00:00+00:00", + "2024-01-02T00:00:00+00:00", + "2024-01-03T00:00:00+00:00", + ], + "type": [0, 0, 0, 0, 2], + "entry": [0, 0, 0, 0, 0], + }) + fallback = datetime(2024, 1, 1, tzinfo=UTC) + end = datetime(2024, 1, 15, tzinfo=UTC) + with sqlite3.connect(tmp_path / "per-symbol-account-events.db") as conn: + conn.execute( + "CREATE TABLE history_deals(" + " ticket INTEGER, symbol TEXT, time TEXT, type INTEGER, entry INTEGER)", + ) + conn.executemany( + "INSERT INTO history_deals(ticket, symbol, time, type, entry)" + " VALUES (?, ?, ?, ?, ?)", + [ + (100, "EURUSD", "2024-01-10T00:00:00+00:00", 0, 0), + (200, "GBPUSD", "2024-01-01T00:00:00+00:00", 0, 0), + (300, "", "2024-01-02T00:00:00+00:00", 2, 0), + ], + ) + write_incremental_datasets( + conn, + client, + ["EURUSD", "GBPUSD"], + {Dataset.history_deals}, + [], + 0, + fallback, + end, + deduplicate=False, + create_rate_views=False, + with_views=False, + include_account_events=True, + ) + rows = conn.execute( + "SELECT ticket, symbol, type FROM history_deals" + " WHERE ticket IN (1, 2, 3, 4, 5) ORDER BY ticket", + ).fetchall() + client.history_deals_get_as_df.assert_called_once_with( + date_from=datetime(2024, 1, 1, tzinfo=UTC), + date_to=end, + ) + assert rows == [(2, "EURUSD", 0), (3, "GBPUSD", 0), (5, "", 2)] + + +class TestWriteHelpers: + """Tests for SQLite write helper branches.""" + + def test_append_dataframe_handles_wide_frames(self, tmp_path: Path) -> None: + """Test wide DataFrames append without exceeding SQLite variable limits.""" + db_path = tmp_path / "wide-frame.db" + columns = {f"col_{index}": [float(index)] for index in range(80)} + frame = pd.DataFrame(columns) + with sqlite3.connect(db_path) as conn: + assert append_dataframe(conn, frame, "wide_rates", IfExists.APPEND) + assert get_table_columns(conn, "wide_rates") == set(columns) + + def test_write_streamed_frame_and_column_tracking(self, tmp_path: Path) -> None: + """Test append helpers track columns and skip empty frames.""" + db_path = tmp_path / "write-helpers.db" + written_columns: dict[Dataset, set[str]] = {} + with sqlite3.connect(db_path) as conn: + assert not write_streamed_frame( + conn, + pd.DataFrame(), + Dataset.rates, + table_exists=False, + if_exists=IfExists.APPEND, + written_columns=written_columns, + ) + assert append_dataframe( + conn, + pd.DataFrame({"time": [1], "open": [1.0]}), + "rates", + IfExists.APPEND, + ) + record_written_columns( + written_columns, + Dataset.rates, + pd.DataFrame({"close": [1.1]}), + ) + assert "close" in written_columns[Dataset.rates] + augment_written_columns_from_sqlite( + conn, + {Dataset.rates}, + written_columns, + ) + assert get_table_columns(conn, "rates") == {"time", "open"} + create_history_indexes(conn, written_columns) diff --git a/uv.lock b/uv.lock index eae6710..c13da59 100644 --- a/uv.lock +++ b/uv.lock @@ -487,7 +487,7 @@ wheels = [ [[package]] name = "mt5cli" -version = "0.4.0" +version = "0.4.1" source = { editable = "." } dependencies = [ { name = "click" },