Consolidate duplicated export and history streaming helpers (#29)

* Consolidate duplicated export and history streaming helpers.

Reduce repeated CLI export plumbing, shared per-symbol SQLite writes, and test mock setup without changing public behavior.

Co-authored-by: Cursor <cursoragent@cursor.com>

* Bump version from 0.7.0 to 0.7.1.

Co-authored-by: Cursor <cursoragent@cursor.com>

---------

Co-authored-by: Cursor <cursoragent@cursor.com>

mt5cli/cli.py

@@ -96,6 +96,15 @@ def _sdk_client(ctx: typer.Context) -> sdk.Mt5CliClient:
     return sdk.Mt5CliClient(config=export_ctx.config)
 
 
+def _export_command(
+    ctx: typer.Context,
+    fetch_fn: Callable[[sdk.Mt5CliClient], pd.DataFrame],
+) -> None:
+    """Create an SDK client, fetch a DataFrame, and export it."""
+    client = _sdk_client(ctx)
+    _execute_export(ctx, lambda: fetch_fn(client))
+
+
 @app.callback()
 def _callback(  # pyright: ignore[reportUnusedFunction]
     ctx: typer.Context,
@@ -193,10 +202,9 @@ def rates_from(
     count: Annotated[int, typer.Option(help="Number of records.")],
 ) -> None:
     """Export rates from a start date."""
-    client = _sdk_client(ctx)
+    _export_command(
-    _execute_export(
         ctx,
-        lambda: client.copy_rates_from(symbol, timeframe, date_from, count),
+        lambda client: client.copy_rates_from(symbol, timeframe, date_from, count),
     )
 
 
@@ -215,10 +223,14 @@ def rates_from_pos(
     count: Annotated[int, typer.Option(help="Number of records.")],
 ) -> None:
     """Export rates from a start position."""
-    client = _sdk_client(ctx)
+    _export_command(
-    _execute_export(
         ctx,
-        lambda: client.copy_rates_from_pos(symbol, timeframe, start_pos, count),
+        lambda client: client.copy_rates_from_pos(
+            symbol,
+            timeframe,
+            start_pos,
+            count,
+        ),
     )
 
 
@@ -240,10 +252,14 @@ def latest_rates(
     ] = 0,
 ) -> None:
     """Export latest rates from a start position."""
-    client = _sdk_client(ctx)
+    _export_command(
-    _execute_export(
         ctx,
-        lambda: client.latest_rates(symbol, timeframe, count, start_pos=start_pos),
+        lambda client: client.latest_rates(
+            symbol,
+            timeframe,
+            count,
+            start_pos=start_pos,
+        ),
     )
 
 
@@ -268,10 +284,9 @@ def rates_range(
     ],
 ) -> None:
     """Export rates for a date range."""
-    client = _sdk_client(ctx)
+    _export_command(
-    _execute_export(
         ctx,
-        lambda: client.copy_rates_range(symbol, timeframe, date_from, date_to),
+        lambda client: client.copy_rates_range(symbol, timeframe, date_from, date_to),
     )
 
 
@@ -293,10 +308,9 @@ def ticks_from(
     ],
 ) -> None:
     """Export ticks from a start date."""
-    client = _sdk_client(ctx)
+    _export_command(
-    _execute_export(
         ctx,
-        lambda: client.copy_ticks_from(symbol, date_from, count, flags),
+        lambda client: client.copy_ticks_from(symbol, date_from, count, flags),
     )
 
 
@@ -318,10 +332,9 @@ def ticks_range(
     ],
 ) -> None:
     """Export ticks for a date range."""
-    client = _sdk_client(ctx)
+    _export_command(
-    _execute_export(
         ctx,
-        lambda: client.copy_ticks_range(symbol, date_from, date_to, flags),
+        lambda client: client.copy_ticks_range(symbol, date_from, date_to, flags),
     )
 
 
@@ -350,10 +363,9 @@ def ticks_recent(
     ] = "ALL",  # pyright: ignore[reportArgumentType]
 ) -> None:
     """Export ticks from a recent time window."""
-    client = _sdk_client(ctx)
+    _export_command(
-    _execute_export(
         ctx,
-        lambda: client.recent_ticks(
+        lambda client: client.recent_ticks(
             symbol,
             seconds,
             date_to=date_to,
@@ -366,13 +378,13 @@ def ticks_recent(
 @app.command()
 def account_info(ctx: typer.Context) -> None:
     """Export account information."""
-    _execute_export(ctx, _sdk_client(ctx).account_info)
+    _export_command(ctx, lambda client: client.account_info())
 
 
 @app.command()
 def terminal_info(ctx: typer.Context) -> None:
     """Export terminal information."""
-    _execute_export(ctx, _sdk_client(ctx).terminal_info)
+    _export_command(ctx, lambda client: client.terminal_info())
 
 
 @app.command()
@@ -384,8 +396,7 @@ def symbols(
     ] = None,
 ) -> None:
     """Export symbol list."""
-    client = _sdk_client(ctx)
+    _export_command(ctx, lambda client: client.symbols(group=group))
-    _execute_export(ctx, lambda: client.symbols(group=group))
 
 
 @app.command()
@@ -394,8 +405,7 @@ def symbol_info(
     symbol: Annotated[str, typer.Option(help="Symbol name.")],
 ) -> None:
     """Export symbol details."""
-    client = _sdk_client(ctx)
+    _export_command(ctx, lambda client: client.symbol_info(symbol))
-    _execute_export(ctx, lambda: client.symbol_info(symbol))
 
 
 @app.command()
@@ -404,8 +414,7 @@ def minimum_margins(
     symbol: Annotated[str, typer.Option(help="Symbol name.")],
 ) -> None:
     """Export minimum-volume buy and sell margin requirements."""
-    client = _sdk_client(ctx)
+    _export_command(ctx, lambda client: client.minimum_margins(symbol))
-    _execute_export(ctx, lambda: client.minimum_margins(symbol))
 
 
 @app.command()
@@ -416,10 +425,9 @@ def orders(
     ticket: Annotated[int | None, typer.Option(help="Ticket filter.")] = None,
 ) -> None:
     """Export active orders."""
-    client = _sdk_client(ctx)
+    _export_command(
-    _execute_export(
         ctx,
-        lambda: client.orders(symbol=symbol, group=group, ticket=ticket),
+        lambda client: client.orders(symbol=symbol, group=group, ticket=ticket),
     )
 
 
@@ -431,10 +439,9 @@ def positions(
     ticket: Annotated[int | None, typer.Option(help="Ticket filter.")] = None,
 ) -> None:
     """Export open positions."""
-    client = _sdk_client(ctx)
+    _export_command(
-    _execute_export(
         ctx,
-        lambda: client.positions(symbol=symbol, group=group, ticket=ticket),
+        lambda client: client.positions(symbol=symbol, group=group, ticket=ticket),
     )
 
 
@@ -455,10 +462,9 @@ def history_orders(
     position: Annotated[int | None, typer.Option(help="Position ticket.")] = None,
 ) -> None:
     """Export historical orders."""
-    client = _sdk_client(ctx)
+    _export_command(
-    _execute_export(
         ctx,
-        lambda: client.history_orders(
+        lambda client: client.history_orders(
             date_from=date_from,
             date_to=date_to,
             group=group,
@@ -486,10 +492,9 @@ def history_deals(
     position: Annotated[int | None, typer.Option(help="Position ticket.")] = None,
 ) -> None:
     """Export historical deals."""
-    client = _sdk_client(ctx)
+    _export_command(
-    _execute_export(
         ctx,
-        lambda: client.history_deals(
+        lambda client: client.history_deals(
             date_from=date_from,
             date_to=date_to,
             group=group,
@@ -512,10 +517,9 @@ def recent_history_deals(
     symbol: Annotated[str | None, typer.Option(help="Symbol filter.")] = None,
 ) -> None:
     """Export historical deals from a recent trailing window."""
-    client = _sdk_client(ctx)
+    _export_command(
-    _execute_export(
         ctx,
-        lambda: client.recent_history_deals(
+        lambda client: client.recent_history_deals(
             hours,
             date_to=date_to,
             group=group,
@@ -527,20 +531,19 @@ def recent_history_deals(
 @app.command()
 def mt5_summary(ctx: typer.Context) -> None:
     """Export a compact terminal/account status summary."""
-    client = _sdk_client(ctx)
+    _export_command(ctx, lambda client: client.mt5_summary_as_df())
-    _execute_export(ctx, client.mt5_summary_as_df)
 
 
 @app.command()
 def version(ctx: typer.Context) -> None:
     """Export MetaTrader5 version information."""
-    _execute_export(ctx, _sdk_client(ctx).version)
+    _export_command(ctx, lambda client: client.version())
 
 
 @app.command()
 def last_error(ctx: typer.Context) -> None:
     """Export the last error information."""
-    _execute_export(ctx, _sdk_client(ctx).last_error)
+    _export_command(ctx, lambda client: client.last_error())
 
 
 @app.command()
@@ -549,8 +552,7 @@ def symbol_info_tick(
     symbol: Annotated[str, typer.Option(help="Symbol name.")],
 ) -> None:
     """Export the last tick for a symbol."""
-    client = _sdk_client(ctx)
+    _export_command(ctx, lambda client: client.symbol_info_tick(symbol))
-    _execute_export(ctx, lambda: client.symbol_info_tick(symbol))
 
 
 @app.command()
@@ -559,8 +561,7 @@ def market_book(
     symbol: Annotated[str, typer.Option(help="Symbol name.")],
 ) -> None:
     """Export market depth (order book) for a symbol."""
-    client = _sdk_client(ctx)
+    _export_command(ctx, lambda client: client.market_book(symbol))
-    _execute_export(ctx, lambda: client.market_book(symbol))
 
 
 @app.command()

mt5cli/history.py

@@ -1332,6 +1332,50 @@ def create_rate_compatibility_views(conn: sqlite3.Connection) -> None:
             )
 
 
+def _stream_symbol_frames(
+    conn: sqlite3.Connection,
+    symbols: Sequence[str],
+    dataset: Dataset,
+    if_exists: IfExists,
+    written_columns: dict[Dataset, set[str]],
+    fetch_frame: Callable[[str], pd.DataFrame],
+) -> bool:
+    """Stream per-symbol frames into SQLite.
+
+    Returns:
+        True if the dataset table was written.
+    """
+    table_exists = False
+    for sym in symbols:
+        table_exists = write_streamed_frame(
+            conn,
+            fetch_frame(sym),
+            dataset,
+            table_exists,
+            if_exists,
+            written_columns,
+        )
+    return table_exists
+
+
+def _record_symbol_time_dedup(
+    dedup_scopes: dict[Dataset, list[DedupScope]],
+    written_tables: set[Dataset],
+    dataset: Dataset,
+    symbol: str,
+    start_date: datetime,
+) -> None:
+    """Record a symbol-scoped deduplication window after an incremental write."""
+    written_tables.add(dataset)
+    _record_dedup_scope(
+        dedup_scopes,
+        dataset,
+        "symbol = ? AND time >= ?",
+        (symbol, start_date),
+        frozenset({"symbol", "time"}),
+    )
+
+
 def write_rates_dataset(
     conn: sqlite3.Connection,
     client: Mt5DataClient,
@@ -1347,8 +1391,8 @@ def write_rates_dataset(
     Returns:
         True if the rates table was written.
     """
-    table_exists = False
+
-    for sym in symbols:
+    def _fetch_rates_frame(sym: str) -> pd.DataFrame:
         frame = client.copy_rates_range_as_df(
             symbol=sym,
             timeframe=timeframe,
@@ -1358,15 +1402,16 @@ def write_rates_dataset(
         if len(frame.columns) != 0:
             frame.insert(0, "symbol", sym)
             frame.insert(1, "timeframe", timeframe)
-        table_exists = write_streamed_frame(
+        return frame
-            conn,
+
-            frame,
+    return _stream_symbol_frames(
-            Dataset.rates,
+        conn,
-            table_exists,
+        symbols,
-            if_exists,
+        Dataset.rates,
-            written_columns,
+        if_exists,
-        )
+        written_columns,
-    return table_exists
+        _fetch_rates_frame,
+    )
 
 
 def write_ticks_dataset(
@@ -1384,8 +1429,8 @@ def write_ticks_dataset(
     Returns:
         True if the ticks table was written.
     """
-    table_exists = False
+
-    for sym in symbols:
+    def _fetch_ticks_frame(sym: str) -> pd.DataFrame:
         frame = client.copy_ticks_range_as_df(
             symbol=sym,
             date_from=date_from,
@@ -1394,15 +1439,16 @@ def write_ticks_dataset(
         ).drop(columns=["symbol"], errors="ignore")
         if len(frame.columns) != 0:
             frame.insert(0, "symbol", sym)
-        table_exists = write_streamed_frame(
+        return frame
-            conn,
+
-            frame,
+    return _stream_symbol_frames(
-            Dataset.ticks,
+        conn,
-            table_exists,
+        symbols,
-            if_exists,
+        Dataset.ticks,
-            written_columns,
+        if_exists,
-        )
+        written_columns,
-    return table_exists
+        _fetch_ticks_frame,
+    )
 
 
 def write_history_dataset(
@@ -1437,22 +1483,22 @@ def write_history_dataset(
             if_exists,
             written_columns,
         )
-    for sym in symbols:
+
-        frame = fetch(date_from=date_from, date_to=date_to, symbol=sym)
+    def _fetch_history_frame(sym: str) -> pd.DataFrame:
-        frame = filter_trade_history_frame(
+        return filter_trade_history_frame(
-            frame,
+            fetch(date_from=date_from, date_to=date_to, symbol=sym),
             [sym],
             include_account_events=False,
         )
-        table_exists = write_streamed_frame(
+
-            conn,
+    return _stream_symbol_frames(
-            frame,
+        conn,
-            dataset,
+        symbols,
-            table_exists,
+        dataset,
-            if_exists,
+        if_exists,
-            written_columns,
+        written_columns,
-        )
+        _fetch_history_frame,
-    return table_exists
+    )
 
 
 def _write_incremental_rates(
@@ -1525,13 +1571,12 @@ def _write_incremental_ticks(
             IfExists.APPEND,
             written_columns,
         ):
-            written_tables.add(Dataset.ticks)
+            _record_symbol_time_dedup(
-            _record_dedup_scope(
                 dedup_scopes,
+                written_tables,
                 Dataset.ticks,
-                "symbol = ? AND time >= ?",
+                symbol,
-                (symbol, start_date),
+                start_date,
-                frozenset({"symbol", "time"}),
             )
 
 
@@ -1564,13 +1609,12 @@ def _write_incremental_history_orders(
             written_columns,
             include_account_events=False,
         ):
-            written_tables.add(Dataset.history_orders)
+            _record_symbol_time_dedup(
-            _record_dedup_scope(
                 dedup_scopes,
+                written_tables,
                 Dataset.history_orders,
-                "symbol = ? AND time >= ?",
+                symbol,
-                (symbol, start_date),
+                start_date,
-                frozenset({"symbol", "time"}),
             )
 
 
@@ -1662,13 +1706,12 @@ def _write_incremental_history_deals(
             written_columns,
             include_account_events=False,
         ):
-            written_tables.add(Dataset.history_deals)
+            _record_symbol_time_dedup(
-            _record_dedup_scope(
                 dedup_scopes,
+                written_tables,
                 Dataset.history_deals,
-                "symbol = ? AND time >= ?",
+                symbol,
-                (symbol, start_date),
+                start_date,
-                frozenset({"symbol", "time"}),
             )
 
 

mt5cli/sdk.py

@@ -64,6 +64,9 @@ _MT5_HISTORY_CLIENT_CALL_FUNCTIONS: frozenset[str] = frozenset({
     "write_ticks_dataset",
     "write_history_dataset",
     "_write_incremental_history_deals",
+    "_fetch_rates_frame",
+    "_fetch_ticks_frame",
+    "_fetch_history_frame",
 })
 _NON_CALLABLE_TYPE_ERROR = re.compile(r"^'[^']+' object is not callable$")
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "mt5cli"
-version = "0.7.0"
+version = "0.7.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"}]

tests/conftest.py

@@ -0,0 +1,52 @@
+"""Shared pytest fixtures for mt5cli tests."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pandas as pd
+import pytest
+from pytest_mock import MockerFixture  # noqa: TC002
+
+_DATAFRAME_METHODS = (
+    "copy_rates_from_as_df",
+    "copy_rates_from_pos_as_df",
+    "copy_rates_range_as_df",
+    "copy_ticks_from_as_df",
+    "copy_ticks_range_as_df",
+    "account_info_as_df",
+    "terminal_info_as_df",
+    "symbols_get_as_df",
+    "symbol_info_as_df",
+    "orders_get_as_df",
+    "positions_get_as_df",
+    "history_orders_get_as_df",
+    "history_deals_get_as_df",
+    "version_as_df",
+    "last_error_as_df",
+    "symbol_info_tick_as_df",
+    "market_book_get_as_df",
+    "order_check_as_df",
+    "order_send_as_df",
+)
+
+
+def build_mock_mt5_data_client() -> MagicMock:
+    """Return a MagicMock Mt5DataClient with common DataFrame stubs."""
+    client = MagicMock()
+    sample_df = pd.DataFrame({"col": [1]})
+    for method_name in _DATAFRAME_METHODS:
+        getattr(client, method_name).return_value = sample_df
+    client.version.return_value = (5, 0, 1)
+    client.terminal_info.return_value = {"connected": True, "paths": ["terminal.exe"]}
+    client.account_info.return_value = {"login": 123, "limits": {"modes": ["demo"]}}
+    client.symbols_total.return_value = 42
+    return client
+
+
+@pytest.fixture
+def mock_client(mocker: MockerFixture) -> MagicMock:
+    """Create and patch a mock Mt5DataClient for CLI and SDK tests."""
+    client = build_mock_mt5_data_client()
+    mocker.patch("mt5cli.sdk.Mt5DataClient", return_value=client)
+    return client

tests/test_cli.py

@@ -69,38 +69,6 @@ class TestExecuteExport:
 # ---------------------------------------------------------------------------
 
 
-@pytest.fixture
-def mock_client(mocker: MockerFixture) -> MagicMock:
-    """Create and patch a mock Mt5DataClient for CLI tests."""
-    client = MagicMock()
-    sample_df = pd.DataFrame({"col": [1]})
-    client.copy_rates_from_as_df.return_value = sample_df
-    client.copy_rates_from_pos_as_df.return_value = sample_df
-    client.copy_rates_range_as_df.return_value = sample_df
-    client.copy_ticks_from_as_df.return_value = sample_df
-    client.copy_ticks_range_as_df.return_value = sample_df
-    client.account_info_as_df.return_value = sample_df
-    client.terminal_info_as_df.return_value = sample_df
-    client.symbols_get_as_df.return_value = sample_df
-    client.symbol_info_as_df.return_value = sample_df
-    client.orders_get_as_df.return_value = sample_df
-    client.positions_get_as_df.return_value = sample_df
-    client.history_orders_get_as_df.return_value = sample_df
-    client.history_deals_get_as_df.return_value = sample_df
-    client.version_as_df.return_value = sample_df
-    client.last_error_as_df.return_value = sample_df
-    client.symbol_info_tick_as_df.return_value = sample_df
-    client.market_book_get_as_df.return_value = sample_df
-    client.order_check_as_df.return_value = sample_df
-    client.order_send_as_df.return_value = sample_df
-    client.version.return_value = (5, 0, 1)
-    client.terminal_info.return_value = {"connected": True, "paths": ["terminal.exe"]}
-    client.account_info.return_value = {"login": 123, "limits": {"modes": ["demo"]}}
-    client.symbols_total.return_value = 42
-    mocker.patch("mt5cli.sdk.Mt5DataClient", return_value=client)
-    return client
-
-
 class TestCommands:
     """Tests for all CLI subcommands via CliRunner."""
 

tests/test_sdk.py

@@ -132,32 +132,6 @@ _DEALS_FIXTURE: dict[str, list[object]] = {
 }
 
 
-@pytest.fixture
-def mock_client(mocker: MockerFixture) -> MagicMock:
-    """Create and patch a mock Mt5DataClient for SDK tests."""
-    client = MagicMock()
-    sample_df = pd.DataFrame({"col": [1]})
-    client.copy_rates_from_as_df.return_value = sample_df
-    client.copy_rates_from_pos_as_df.return_value = sample_df
-    client.copy_rates_range_as_df.return_value = sample_df
-    client.copy_ticks_from_as_df.return_value = sample_df
-    client.copy_ticks_range_as_df.return_value = sample_df
-    client.account_info_as_df.return_value = sample_df
-    client.terminal_info_as_df.return_value = sample_df
-    client.symbols_get_as_df.return_value = sample_df
-    client.symbol_info_as_df.return_value = sample_df
-    client.orders_get_as_df.return_value = sample_df
-    client.positions_get_as_df.return_value = sample_df
-    client.history_orders_get_as_df.return_value = sample_df
-    client.history_deals_get_as_df.return_value = sample_df
-    client.version_as_df.return_value = sample_df
-    client.last_error_as_df.return_value = sample_df
-    client.symbol_info_tick_as_df.return_value = sample_df
-    client.market_book_get_as_df.return_value = sample_df
-    mocker.patch("mt5cli.sdk.Mt5DataClient", return_value=client)
-    return client
-
-
 def _build_history_client(mocker: MockerFixture) -> MagicMock:
     """Build a mocked Mt5DataClient with per-symbol history results."""
     client = MagicMock()

uv.lock

@@ -487,7 +487,7 @@ wheels = [
 
 [[package]]
 name = "mt5cli"
-version = "0.7.0"
+version = "0.7.1"
 source = { editable = "." }
 dependencies = [
     { name = "click" },