"""Tests for mt5cli.history module.""" from __future__ import annotations import logging import sqlite3 from datetime import UTC, datetime from typing import TYPE_CHECKING, cast from unittest.mock import MagicMock import pandas as pd import pytest from pytest_mock import MockerFixture # noqa: TC002 if TYPE_CHECKING: from collections.abc import Callable from pathlib import Path from pdmt5 import TIMEFRAME_MAP from mt5cli import history from mt5cli.history import ( DEFAULT_HISTORY_DATASETS, DEFAULT_HISTORY_TIMEFRAMES, DedupScope, RateTarget, append_dataframe, augment_written_columns_from_sqlite, build_rate_targets, 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, drop_forming_rate_bar, 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, load_rate_data, load_rate_data_from_connection, load_rate_series_by_granularity, load_rate_series_from_sqlite, parse_sqlite_timestamp, quote_sqlite_identifier, record_written_columns, report_rate_gaps, resolve_granularity_name, resolve_history_datasets, resolve_history_tick_flags, resolve_history_timeframes, resolve_rate_table_name, resolve_rate_tables, resolve_rate_view_name, resolve_rate_view_names, write_collected_datasets, write_history_dataset, write_incremental_datasets, write_rates_dataset, write_streamed_frame, ) from mt5cli.utils import Dataset, IfExists def _resolve_rate_view_single( conn_or_path: sqlite3.Connection | Path | str | None, *, granularity: str = "M1", require_existing: bool = False, ) -> str: """Resolve a single EURUSD rate view name via resolve_rate_view_name.""" return resolve_rate_view_name( conn_or_path, "EURUSD", granularity, require_existing=require_existing, ) def _resolve_rate_view_batch( conn_or_path: sqlite3.Connection | Path | str | None, *, granularity: str = "M1", require_existing: bool = False, ) -> list[str]: """Resolve batch EURUSD rate view names via resolve_rate_view_names.""" return resolve_rate_view_names( conn_or_path, ["EURUSD"], [granularity], require_existing=require_existing, ) class TestResolveRateViewName: """Tests for resolve_rate_view_name and resolve_rate_view_names.""" def test_resolve_rate_table_name_returns_normalized_table(self) -> None: """Test canonical normalized rates table name is stable.""" assert resolve_rate_table_name("EURUSD", "M1") == "rates" def test_resolve_rate_table_name_rejects_empty_symbol(self) -> None: """Test canonical rate table resolution validates symbols.""" with pytest.raises(ValueError, match="symbol must not be empty"): resolve_rate_table_name(" ", "M1") def test_missing_database_path_does_not_create_file(self, tmp_path: Path) -> None: """Test resolving against a missing path does not create a database.""" db_path = tmp_path / "missing.db" assert resolve_rate_view_name(db_path, "EURUSD", "M1") == "rate_EURUSD__1" assert not db_path.exists() def test_none_path_returns_default_name(self) -> None: """Test a None connection or path returns the deterministic default.""" assert resolve_rate_view_name(None, "EURUSD", "M1") == "rate_EURUSD__1" assert resolve_rate_view_names(None, ["EURUSD"], ["M1", "H1"]) == [ "rate_EURUSD__1", "rate_EURUSD__16385", ] @pytest.mark.parametrize( "resolve", [_resolve_rate_view_single, _resolve_rate_view_batch], ids=["single", "batch"], ) def test_none_path_with_require_existing_raises( self, resolve: Callable[..., object], ) -> None: """Test a None path under strict mode raises a clear error.""" with pytest.raises(ValueError, match="SQLite database not found"): resolve(None, require_existing=True) def test_no_rates_table_falls_back_to_single_timeframe_name( self, tmp_path: Path, ) -> None: """Test databases without a rates table use single-timeframe naming.""" db_path = tmp_path / "no-rates.db" with sqlite3.connect(db_path) as conn: conn.execute("CREATE TABLE ticks(symbol TEXT, time TEXT)") assert resolve_rate_view_name(db_path, "EURUSD", "M1") == "rate_EURUSD__1" def test_single_timeframe_for_one_symbol(self, tmp_path: Path) -> None: """Test one stored timeframe resolves to the short view name.""" db_path = tmp_path / "single-timeframe.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 resolve_rate_view_name(db_path, "EURUSD", "M1") == "rate_EURUSD__1" def test_multiple_timeframes_for_one_symbol(self, tmp_path: Path) -> None: """Test multiple stored timeframes resolve to disambiguated view names.""" db_path = tmp_path / "multi-timeframe.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", TIMEFRAME_MAP["H1"], "2024-01-01T01:00:00+00:00", 1.1), ], ) create_rate_compatibility_views(conn) assert resolve_rate_view_name(db_path, "EURUSD", "M1") == "rate_EURUSD__M1_1" assert ( resolve_rate_view_name(db_path, "EURUSD", "H1") == "rate_EURUSD__H1_16385" ) def test_prefers_multi_name_when_both_candidate_views_exist( self, tmp_path: Path, ) -> None: """Test multi-timeframe metadata wins over stale single-timeframe views.""" db_path = tmp_path / "stale-and-current-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", TIMEFRAME_MAP["H1"], "2024-01-01T01:00:00+00:00", 1.1), ], ) conn.execute( 'CREATE VIEW "rate_EURUSD__1" AS' " SELECT time, close FROM rates" " WHERE symbol = 'EURUSD' AND timeframe = 1", ) conn.execute( 'CREATE VIEW "rate_EURUSD__M1_1" AS' " SELECT time, close FROM rates" " WHERE symbol = 'EURUSD' AND timeframe = 1", ) assert resolve_rate_view_name(db_path, "EURUSD", "M1") == "rate_EURUSD__M1_1" def test_prefers_existing_view_when_metadata_unavailable( self, tmp_path: Path, ) -> None: """Test an existing managed view is preferred without rates metadata.""" db_path = tmp_path / "view-only.db" with sqlite3.connect(db_path) as conn: conn.execute("CREATE TABLE ticks(symbol TEXT, time TEXT)") conn.execute('CREATE VIEW "rate_EURUSD__M1_1" AS SELECT 1 AS close') assert resolve_rate_view_name(db_path, "EURUSD", "M1") == "rate_EURUSD__M1_1" def test_symbol_absent_from_rates_metadata_uses_candidate_pair( self, tmp_path: Path, ) -> None: """Test symbols missing from rates metadata still resolve known views.""" db_path = tmp_path / "other-symbol-only.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 (?, ?, ?, ?)", ("GBPUSD", 1, "2024-01-01T00:00:00+00:00", 1.0), ) conn.execute( 'CREATE VIEW "rate_EURUSD__1" AS' " SELECT time, close FROM rates" " WHERE symbol = 'EURUSD' AND timeframe = 1", ) assert resolve_rate_view_name(db_path, "EURUSD", "M1") == "rate_EURUSD__1" def test_ignores_non_compatibility_rate_views(self, tmp_path: Path) -> None: """Test unrelated rate_* views without the __ separator are ignored.""" db_path = tmp_path / "summary-view.db" with sqlite3.connect(db_path) as conn: conn.execute("CREATE TABLE ticks(symbol TEXT, time TEXT)") conn.execute('CREATE VIEW "rate_summary" AS SELECT 1 AS close') assert resolve_rate_view_name(db_path, "EURUSD", "M1") == "rate_EURUSD__1" @pytest.mark.parametrize( "resolve", [_resolve_rate_view_single, _resolve_rate_view_batch], ids=["single", "batch"], ) def test_invalid_granularity_propagates_value_error( self, tmp_path: Path, resolve: Callable[..., object], ) -> None: """Test invalid granularities raise ValueError from parse_timeframe.""" with pytest.raises(ValueError, match="Invalid timeframe"): resolve(tmp_path / "unused.db", granularity="BAD") def test_resolve_rate_view_names_for_multiple_pairs(self, tmp_path: Path) -> None: """Test batch resolution returns row-major symbol/granularity pairs.""" db_path = tmp_path / "batch-resolve.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", TIMEFRAME_MAP["H1"], "2024-01-01T01:00:00+00:00", 1.1), ("GBPUSD", 1, "2024-01-01T00:00:00+00:00", 1.2), ], ) create_rate_compatibility_views(conn) assert resolve_rate_view_names( db_path, ["EURUSD", "GBPUSD"], ["M1", "H1"], ) == [ "rate_EURUSD__M1_1", "rate_EURUSD__H1_16385", "rate_GBPUSD__1", "rate_GBPUSD__16385", ] @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 resolve to safely created view names.""" db_path = tmp_path / "broker-symbol-resolve.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 (?, ?, ?, ?)", (symbol, 1, "2024-01-01T00:00:00+00:00", 1.0), ) create_rate_compatibility_views(conn) assert resolve_rate_view_name(db_path, symbol, "M1") == build_rate_view_name( symbol=symbol, granularity="M1", granularity_count=1, timeframe=1, ) def test_accepts_open_sqlite_connection(self, tmp_path: Path) -> None: """Test resolver accepts an already-open SQLite connection.""" db_path = tmp_path / "open-connection.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 resolve_rate_view_name(conn, "EURUSD", "M1") == "rate_EURUSD__1" @pytest.mark.parametrize( "resolve", [_resolve_rate_view_single, _resolve_rate_view_batch], ids=["single", "batch"], ) def test_require_existing_raises_when_database_missing( self, tmp_path: Path, resolve: Callable[..., object], ) -> None: """Test strict mode rejects missing database paths.""" db_path = tmp_path / "missing.db" with pytest.raises(ValueError, match="SQLite database not found"): resolve(db_path, require_existing=True) @pytest.mark.parametrize( "resolve", [_resolve_rate_view_single, _resolve_rate_view_batch], ids=["single", "batch"], ) def test_require_existing_raises_when_view_missing( self, tmp_path: Path, resolve: Callable[..., object], ) -> None: """Test strict mode rejects databases without matching rate views.""" db_path = tmp_path / "no-view.db" with sqlite3.connect(db_path) as conn: conn.execute("CREATE TABLE ticks(symbol TEXT, time TEXT)") with pytest.raises(ValueError, match="No rate compatibility view exists"): resolve(db_path, require_existing=True) def test_require_existing_returns_existing_view(self, tmp_path: Path) -> None: """Test strict mode returns a view when one exists.""" db_path = tmp_path / "existing-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 ( resolve_rate_view_name( db_path, "EURUSD", "M1", require_existing=True, ) == "rate_EURUSD__1" ) assert resolve_rate_view_names( db_path, ["EURUSD"], ["M1"], require_existing=True, ) == ["rate_EURUSD__1"] 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 TestLoadRateData: """Tests for SQLite rate-like table and view loading.""" def test_loads_close_rates_from_path_with_count(self, tmp_path: Path) -> None: """Test loading the latest close-based rates in ascending time order.""" db_path = tmp_path / "rates.db" with sqlite3.connect(db_path) as conn: conn.execute("CREATE TABLE rates(time TEXT, close REAL)") conn.executemany( "INSERT INTO rates(time, close) VALUES (?, ?)", [ ("2024-01-01T00:00:00+00:00", 1.0), ("2024-01-01T00:02:00+00:00", 1.2), ("2024-01-01T00:01:00+00:00", 1.1), ], ) frame = load_rate_data(db_path, "rates", count=2) assert list(frame["close"]) == [1.1, 1.2] assert isinstance(frame.index, pd.DatetimeIndex) assert frame.index.name == "time" assert frame.index.is_monotonic_increasing def test_loads_ask_bid_tick_like_rates_from_connection( self, tmp_path: Path, ) -> None: """Test loading tick-like tables with bid and ask columns.""" db_path = tmp_path / "ticks.db" with sqlite3.connect(db_path) as conn: conn.execute("CREATE TABLE ticks(time TEXT, bid REAL, ask REAL)") conn.execute( "INSERT INTO ticks(time, bid, ask) VALUES (?, ?, ?)", ("2024-01-01T00:00:00+00:00", 1.0, 1.1), ) frame = load_rate_data_from_connection(conn, "ticks") path_frame = load_rate_data(conn, "ticks") assert frame.iloc[0].to_dict() == {"bid": 1.0, "ask": 1.1} assert path_frame.iloc[0].to_dict() == {"bid": 1.0, "ask": 1.1} def test_loads_from_view(self, tmp_path: Path) -> None: """Test loading from a SQLite view.""" db_path = tmp_path / "view.db" with sqlite3.connect(db_path) as conn: conn.execute("CREATE TABLE rates(time TEXT, close REAL)") conn.execute( "INSERT INTO rates(time, close) VALUES (?, ?)", ("2024-01-01T00:00:00+00:00", 1.0), ) conn.execute("CREATE VIEW rate_view AS SELECT time, close FROM rates") frame = load_rate_data_from_connection(conn, "rate_view") assert list(frame["close"]) == [1.0] def test_load_rate_series_from_sqlite_table_style( self, tmp_path: Path, ) -> None: """Test public table-style loader returns one rate DataFrame.""" db_path = tmp_path / "table-style.db" with sqlite3.connect(db_path) as conn: conn.execute("CREATE TABLE rates(time TEXT, close REAL)") conn.executemany( "INSERT INTO rates(time, close) VALUES (?, ?)", [ ("2024-01-01T00:00:00+00:00", 1.0), ("2024-01-01T00:01:00+00:00", 1.1), ], ) frame = load_rate_series_from_sqlite(db_path, table="rates", count=1) assert isinstance(frame, pd.DataFrame) assert list(frame["close"]) == [1.1] def test_load_rate_series_from_sqlite_requires_targets_without_table(self) -> None: """Test multi-series loading requires targets when table is omitted.""" with pytest.raises(ValueError, match="targets are required"): load_rate_series_from_sqlite("unused.db", count=1) def test_loads_quoted_identifier(self, tmp_path: Path) -> None: """Test table names are quoted safely.""" db_path = tmp_path / "quoted.db" table = 'rate "quoted"' quoted = quote_sqlite_identifier(table) with sqlite3.connect(db_path) as conn: conn.execute(f"CREATE TABLE {quoted}(time TEXT, close REAL)") conn.execute( f"INSERT INTO {quoted}(time, close) VALUES (?, ?)", # noqa: S608 ("2024-01-01T00:00:00+00:00", 1.0), ) frame = load_rate_data_from_connection(conn, table) assert list(frame["close"]) == [1.0] @pytest.mark.parametrize( ("db_path", "match"), [ pytest.param( "missing.db", "SQLite database not found", id="missing-database" ), pytest.param(".", "not a file", id="non-file"), ], ) def test_rejects_missing_database_and_non_file( self, tmp_path: Path, db_path: str, match: str, ) -> None: """Test path validation for SQLite database inputs.""" with pytest.raises(ValueError, match=match): load_rate_data(tmp_path / db_path, "rates") @pytest.mark.parametrize( ("table", "count", "match"), [ ("", None, "must not be empty"), ("rates", 0, "count must be positive"), ("rates", -1, "count must be positive"), ], ) def test_rejects_invalid_inputs( self, tmp_path: Path, table: str, count: int | None, match: str, ) -> None: """Test request validation.""" db_path = tmp_path / "invalid-inputs.db" with sqlite3.connect(db_path) as conn: conn.execute("CREATE TABLE rates(time TEXT, close REAL)") with pytest.raises(ValueError, match=match): load_rate_data_from_connection(conn, table, count=count) @pytest.mark.parametrize( ("ddl", "match"), [ ("CREATE TABLE rates(time TEXT, close REAL)", "contains no rows"), ("CREATE TABLE rates(close REAL)", "time column"), ("CREATE TABLE rates(time TEXT, open REAL)", "close, or both ask and bid"), ], ) def test_rejects_invalid_tables( self, tmp_path: Path, ddl: str, match: str, ) -> None: """Test missing table, empty table, and invalid schemas.""" db_path = tmp_path / "invalid-tables.db" with sqlite3.connect(db_path) as conn: conn.execute(ddl) with pytest.raises(ValueError, match=match): load_rate_data_from_connection(conn, "rates") with pytest.raises(ValueError, match="not found"): load_rate_data_from_connection(conn, "missing") def test_rejects_invalid_timestamp(self, tmp_path: Path) -> None: """Test unparsable timestamps fail clearly.""" db_path = tmp_path / "invalid-time.db" with sqlite3.connect(db_path) as conn: conn.execute("CREATE TABLE rates(time TEXT, close REAL)") conn.execute("INSERT INTO rates(time, close) VALUES (?, ?)", ("bad", 1.0)) with pytest.raises(ValueError, match="unparsable time"): load_rate_data_from_connection(conn, "rates") def test_loads_numeric_mt5_epoch_seconds(self, tmp_path: Path) -> None: """Test MT5-native integer timestamps are parsed as epoch seconds.""" db_path = tmp_path / "epoch-rates.db" with sqlite3.connect(db_path) as conn: conn.execute("CREATE TABLE rates(time INTEGER, close REAL)") conn.execute( "INSERT INTO rates(time, close) VALUES (?, ?)", (1_704_067_200, 1.0), ) frame = load_rate_data_from_connection(conn, "rates") assert frame.index[0] == pd.Timestamp("2024-01-01", tz="UTC") assert list(frame["close"]) == [1.0] class TestResolveHistorySettings: """Tests for history dataset and timeframe resolution.""" @pytest.mark.parametrize( ("datasets", "expected"), [ ( None, { Dataset.rates, Dataset.history_orders, Dataset.history_deals, }, ), ( cast("set[Dataset]", set()), cast("set[Dataset]", set()), ), ({Dataset.ticks}, {Dataset.ticks}), (set(Dataset), set(Dataset)), ], ids=["defaults", "empty", "explicit-ticks", "all-datasets"], ) def test_resolve_history_datasets( self, datasets: set[Dataset] | None, expected: set[Dataset], ) -> None: """Test dataset resolution defaults and explicit selections.""" resolved = resolve_history_datasets(datasets) assert resolved == expected if datasets is None: assert resolved == set(DEFAULT_HISTORY_DATASETS) assert Dataset.ticks not in resolved 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 not any( name.startswith("TIMEFRAME_") for name in 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"]] @pytest.mark.parametrize( ("flags", "expected"), [("ALL", -1), (2, 2)], ids=["named-all", "numeric"], ) def test_resolve_history_tick_flags( self, flags: str | int, expected: int, ) -> None: """Test tick flag resolution accepts named and numeric flags.""" assert resolve_history_tick_flags(flags) == expected @pytest.mark.parametrize( ("timeframe", "expected"), [(999, "999"), (1, "M1")], ids=["unknown-integer-fallback", "known-m1"], ) def test_resolve_granularity_name_falls_back_to_integer( self, timeframe: int, expected: str, ) -> None: """Test unknown timeframe constants fall back to integer text.""" assert resolve_granularity_name(timeframe) == expected def test_resolve_granularity_name_strips_official_prefix( self, mocker: MockerFixture, ) -> None: """Test official pdmt5 timeframe names are normalized to short aliases.""" mocker.patch( "mt5cli.history._get_timeframe_name", return_value="TIMEFRAME_H1", ) assert resolve_granularity_name(16385) == "H1" class TestDropFormingRateBar: """Tests for drop_forming_rate_bar.""" def test_drops_still_forming_last_bar(self) -> None: """Test the still-forming last bar is removed.""" df_rate = pd.DataFrame( {"time": [1, 2, 3], "close": [1.1, 1.2, 1.3]}, index=pd.Index(["a", "b", "c"], name="idx"), ) result = drop_forming_rate_bar(df_rate) pd.testing.assert_frame_equal( result, pd.DataFrame( {"time": [1, 2], "close": [1.1, 1.2]}, index=pd.Index(["a", "b"], name="idx"), ), ) assert df_rate.shape == (3, 2) @pytest.mark.parametrize( "df_rate", [ pytest.param( pd.DataFrame(columns=["time", "close"]), id="empty-input", ), pytest.param( pd.DataFrame({"time": [1], "close": [1.1]}), id="single-forming-bar", ), ], ) def test_returns_empty_frame_for_empty_result_cases( self, df_rate: pd.DataFrame, ) -> None: """Test empty and single-bar frames stay empty after dropping.""" result = drop_forming_rate_bar(df_rate) assert result.empty assert list(result.columns) == ["time", "close"] class TestParseSqliteTimestamp: """Tests for parse_sqlite_timestamp.""" @pytest.mark.parametrize( ("value", "expected"), [ pytest.param(None, None, id="none"), pytest.param( 1_704_067_200, datetime(2024, 1, 1, tzinfo=UTC), id="mt5-epoch-seconds", ), pytest.param( datetime.fromisoformat("2024-01-01T00:00:00"), datetime(2024, 1, 1, tzinfo=UTC), id="naive-datetime", ), pytest.param( "Jan 1 2024", datetime(2024, 1, 1, tzinfo=UTC), id="pandas-string", ), pytest.param("not-a-datetime", None, id="invalid-string"), pytest.param(object(), None, id="unsupported-object"), ], ) def test_parses_various_inputs( self, value: object, expected: datetime | None, ) -> None: """Test ISO, pandas-compatible, numeric, and datetime values.""" assert parse_sqlite_timestamp(value) == expected def test_serialize_invalid_timestamp_returns_none(self) -> None: """Test invalid SQLite timestamp serialization returns None.""" assert history._serialize_sqlite_timestamp(object()) is None # type: ignore[reportPrivateUsage] def test_require_serialized_timestamp_raises_for_invalid(self) -> None: """Test invalid SQLite timestamp boundaries fail fast.""" with pytest.raises(ValueError, match="Invalid SQLite timestamp boundary"): history._require_serialized_sqlite_timestamp( # type: ignore[reportPrivateUsage] object() ) 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) @pytest.mark.parametrize( ( "dataset", "db_name", "ddl", "insert_sql", "insert_rows", "symbols", "timeframes", "expected_starts", ), [ pytest.param( Dataset.rates, "duplicate-rate-groups", ( "CREATE TABLE rates(" " symbol TEXT, timeframe INTEGER, time TEXT, open REAL)" ), ( "INSERT INTO rates(symbol, timeframe, time, open)" " VALUES (?, ?, ?, ?)" ), [ ("EURUSD", 1, "2024-01-03T00:00:00+00:00", 1.2), ("EURUSD", 1, "2024-01-02T00:00:00+00:00", 1.1), ("GBPUSD", 1, "2024-01-04T00:00:00+00:00", 1.3), ], ["EURUSD", "GBPUSD"], [1], { ("EURUSD", 1): datetime(2024, 1, 3, tzinfo=UTC), ("GBPUSD", 1): datetime(2024, 1, 4, tzinfo=UTC), }, id="rates-latest-row-per-group", ), pytest.param( Dataset.ticks, "duplicate-symbol-groups", "CREATE TABLE ticks(symbol TEXT, time TEXT)", "INSERT INTO ticks(symbol, time) VALUES (?, ?)", [ ("EURUSD", "2024-01-03T00:00:00+00:00"), ("EURUSD", "2024-01-02T00:00:00+00:00"), ("GBPUSD", "2024-01-04T00:00:00+00:00"), ], ["EURUSD", "GBPUSD"], None, { ("EURUSD", None): datetime(2024, 1, 3, tzinfo=UTC), ("GBPUSD", None): datetime(2024, 1, 4, tzinfo=UTC), }, id="ticks-latest-row-per-symbol", ), ], ) def test_load_incremental_start_prefers_latest_row_per_group( self, tmp_path: Path, dataset: Dataset, db_name: str, ddl: str, insert_sql: str, insert_rows: list[tuple[object, ...]], symbols: list[str], timeframes: list[int] | None, expected_starts: dict[tuple[str, int | None], datetime], ) -> None: """Test incremental resume keeps only the latest row per scoped group.""" fallback = datetime(2024, 1, 1, tzinfo=UTC) with sqlite3.connect(tmp_path / f"{db_name}.db") as conn: conn.execute(ddl) conn.executemany(insert_sql, insert_rows) starts = load_incremental_start_datetimes( conn, dataset, symbols=symbols, timeframes=timeframes, fallback_start=fallback, ) for key, expected in expected_starts.items(): assert starts[key] == expected @pytest.mark.parametrize( ( "dataset", "db_name", "ddl", "insert_sql", "insert_rows", "symbols", "timeframes", "expected_starts", ), [ pytest.param( Dataset.rates, "numeric-rate-cursor", "CREATE TABLE rates( symbol TEXT, timeframe INTEGER, time, open REAL)", ( "INSERT INTO rates(symbol, timeframe, time, open)" " VALUES (?, ?, ?, ?)" ), [ ("EURUSD", 1, 1704153600, 1.2), ("GBPUSD", 1, "2024-01-03T00:00:00+00:00", 1.3), ], ["EURUSD", "GBPUSD"], [1], { ("EURUSD", 1): datetime(2024, 1, 2, tzinfo=UTC), ("GBPUSD", 1): datetime(2024, 1, 3, tzinfo=UTC), }, id="rates-numeric-cursor", ), pytest.param( Dataset.ticks, "numeric-symbol-cursor", "CREATE TABLE ticks(symbol TEXT, time)", "INSERT INTO ticks(symbol, time) VALUES (?, ?)", [ ("EURUSD", 1704153600), ("GBPUSD", "2024-01-03T00:00:00+00:00"), ], ["EURUSD", "GBPUSD"], None, { ("EURUSD", None): datetime(2024, 1, 2, tzinfo=UTC), ("GBPUSD", None): datetime(2024, 1, 3, tzinfo=UTC), }, id="ticks-numeric-cursor", ), ], ) def test_load_incremental_start_accepts_numeric_cursor( self, tmp_path: Path, dataset: Dataset, db_name: str, ddl: str, insert_sql: str, insert_rows: list[tuple[object, ...]], symbols: list[str], timeframes: list[int] | None, expected_starts: dict[tuple[str, int | None], datetime], ) -> None: """Test incremental resume preserves numeric epoch cursors.""" fallback = datetime(2024, 1, 1, tzinfo=UTC) with sqlite3.connect(tmp_path / f"{db_name}.db") as conn: conn.execute(ddl) conn.executemany(insert_sql, insert_rows) starts = load_incremental_start_datetimes( conn, dataset, symbols=symbols, timeframes=timeframes, fallback_start=fallback, ) for key, expected in expected_starts.items(): assert starts[key] == expected @pytest.mark.parametrize( ("ddl", "missing_col"), [ ("CREATE TABLE rates(symbol TEXT, time TEXT, open REAL)", "timeframe"), ("CREATE TABLE rates(timeframe INTEGER, time TEXT, open REAL)", "symbol"), ("CREATE TABLE rates(symbol TEXT, timeframe INTEGER, open REAL)", "time"), ], ) def test_load_incremental_start_datetimes_requires_column( self, tmp_path: Path, ddl: str, missing_col: str, ) -> None: """Test rates tables missing a required column fail fast.""" fallback = datetime(2024, 1, 1, tzinfo=UTC) with sqlite3.connect(tmp_path / f"rates-no-{missing_col}.db") as conn: conn.execute(ddl) with pytest.raises(ValueError, match=f"missing: {missing_col}") as exc_info: load_incremental_start_datetimes( conn, Dataset.rates, symbols=["EURUSD"], timeframes=[1], fallback_start=fallback, ) assert missing_col 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 @pytest.mark.parametrize( ( "dataset", "ddl", "insert_sql", "insert_args", "symbols", "timeframes", "start_key", "db_name", ), [ pytest.param( Dataset.ticks, "CREATE TABLE ticks(symbol TEXT, time TEXT)", "INSERT INTO ticks(symbol, time) VALUES (?, ?)", ("EURUSD", "not-a-datetime"), ["EURUSD"], None, ("EURUSD", None), "bad-max-time", id="ticks-unparseable-max-time", ), pytest.param( Dataset.rates, ( "CREATE TABLE rates(" " symbol TEXT, timeframe INTEGER, time TEXT, open REAL)" ), ( "INSERT INTO rates(symbol, timeframe, time, open)" " VALUES (?, ?, ?, ?)" ), ("EURUSD", 1, "not-a-datetime", 1.0), ["EURUSD"], [1], ("EURUSD", 1), "bad-rates-max-time", id="rates-unparseable-max-time", ), ], ) def test_load_incremental_start_skips_unparseable_max_time( self, tmp_path: Path, dataset: Dataset, ddl: str, insert_sql: str, insert_args: tuple[object, ...], symbols: list[str], timeframes: list[int] | None, start_key: tuple[str, int | None], db_name: str, ) -> None: """Test grouped resume ignores rows whose MAX(time) cannot be parsed.""" fallback = datetime(2024, 1, 1, tzinfo=UTC) with sqlite3.connect(tmp_path / f"{db_name}.db") as conn: conn.execute(ddl) conn.execute(insert_sql, insert_args) starts = load_incremental_start_datetimes( conn, dataset, symbols=symbols, timeframes=timeframes, fallback_start=fallback, ) assert starts[start_key] == 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 @pytest.mark.parametrize( ( "db_name", "ddl", "table_name", "insert_sql", "insert_args", "loader_name", "loader_kwargs", "expected_key", ), [ pytest.param( "grouped-rate-parse-none", ( "CREATE TABLE rates(" " symbol TEXT, timeframe INTEGER, time TEXT, open REAL)" ), "rates", ( "INSERT INTO rates(symbol, timeframe, time, open)" " VALUES (?, ?, ?, ?)" ), ("EURUSD", 1, "2024-01-03T00:00:00+00:00", 1.2), "_load_grouped_rate_start_datetimes", {"symbols": ["EURUSD"], "timeframes": [1]}, ("EURUSD", 1), id="grouped-rate-parse-failure-fallback", ), pytest.param( "symbol-start-parse-none", "CREATE TABLE ticks(symbol TEXT, time TEXT)", "ticks", "INSERT INTO ticks(symbol, time) VALUES (?, ?)", ("EURUSD", "2024-01-03T00:00:00+00:00"), "_load_symbol_start_datetimes", {"symbols": ["EURUSD"]}, ("EURUSD", None), id="symbol-scoped-parse-failure-fallback", ), ], ) def test_incremental_start_skips_rows_when_timestamp_parse_fails( self, tmp_path: Path, mocker: MockerFixture, db_name: str, ddl: str, table_name: str, insert_sql: str, insert_args: tuple[object, ...], loader_name: str, loader_kwargs: dict[str, object], expected_key: tuple[str, int | None], ) -> None: """Test incremental starts fall back when a parsed row returns None.""" fallback = datetime(2024, 1, 1, tzinfo=UTC) with sqlite3.connect(tmp_path / f"{db_name}.db") as conn: conn.execute(ddl) conn.execute(insert_sql, insert_args) mocker.patch.object(history, "parse_sqlite_timestamp", return_value=None) loader = getattr(history, loader_name) starts = loader( conn, table_name, fallback_start=fallback, **loader_kwargs, ) assert starts[expected_key] == fallback @pytest.mark.parametrize( ( "db_name", "ddl", "table_name", "insert_sql", "insert_rows", "loader_name", "loader_kwargs", "expected_starts", ), [ pytest.param( "grouped-rate-mixed-formats", "CREATE TABLE rates( symbol TEXT, timeframe INTEGER, time, open REAL)", "rates", ( "INSERT INTO rates(symbol, timeframe, time, open)" " VALUES (?, ?, ?, ?)" ), [ ("EURUSD", 1, "2024-01-02 00:00:00", 1.0), ("EURUSD", 1, "2024-01-03T00:00:00+00:00", 1.1), ("EURUSD", 1, 1704240000, 1.2), ("GBPUSD", 1, 1704153600, 1.3), ("GBPUSD", 1, "2024-01-04T00:00:00+00:00", 1.4), ], "_load_grouped_rate_start_datetimes", {"symbols": ["EURUSD", "GBPUSD"], "timeframes": [1]}, { ("EURUSD", 1): datetime(2024, 1, 3, tzinfo=UTC), ("GBPUSD", 1): datetime(2024, 1, 4, tzinfo=UTC), }, id="grouped-rate-mixed-legacy-timestamps", ), pytest.param( "symbol-mixed-formats", "CREATE TABLE ticks(symbol TEXT, time)", "ticks", "INSERT INTO ticks(symbol, time) VALUES (?, ?)", [ ("EURUSD", "2024-01-02 00:00:00"), ("EURUSD", "2024-01-03T00:00:00+00:00"), ("EURUSD", 1704240000), ("GBPUSD", 1704153600), ("GBPUSD", "2024-01-04T00:00:00+00:00"), ], "_load_symbol_start_datetimes", {"symbols": ["EURUSD", "GBPUSD"]}, { ("EURUSD", None): datetime(2024, 1, 3, tzinfo=UTC), ("GBPUSD", None): datetime(2024, 1, 4, tzinfo=UTC), }, id="symbol-scoped-mixed-legacy-timestamps", ), ], ) def test_incremental_start_aggregates_mixed_legacy_timestamps( self, tmp_path: Path, db_name: str, ddl: str, table_name: str, insert_sql: str, insert_rows: list[tuple[object, ...]], loader_name: str, loader_kwargs: dict[str, object], expected_starts: dict[tuple[str, int | None], datetime], ) -> None: """Test incremental resume uses SQL MAX over mixed legacy time formats.""" fallback = datetime(2024, 1, 1, tzinfo=UTC) with sqlite3.connect(tmp_path / f"{db_name}.db") as conn: conn.execute(ddl) conn.executemany(insert_sql, insert_rows) loader = getattr(history, loader_name) starts = loader( conn, table_name, fallback_start=fallback, **loader_kwargs, ) for key, expected in expected_starts.items(): assert starts[key] == expected @pytest.mark.parametrize( ( "db_name", "ddl", "table_name", "insert_sql", "insert_rows", "loader_name", "loader_kwargs", "expected", ), [ pytest.param( "grouped-rate-aggregation", ( "CREATE TABLE rates(" " symbol TEXT, timeframe INTEGER, time TEXT, open REAL)" ), "rates", ( "INSERT INTO rates(symbol, timeframe, time, open)" " VALUES (?, ?, ?, ?)" ), [ ("EURUSD", 1, f"2024-01-01T{hour:02d}:00:00+00:00", float(hour)) for hour in range(24) ], "_load_grouped_rate_start_datetimes", {"symbols": ["EURUSD"], "timeframes": [1]}, ( "GROUP BY symbol, timeframe", ("EURUSD", 1), datetime(2024, 1, 1, 23, tzinfo=UTC), ), id="grouped-rate-sqlite-aggregation", ), pytest.param( "symbol-aggregation", "CREATE TABLE ticks(symbol TEXT, time TEXT)", "ticks", "INSERT INTO ticks(symbol, time) VALUES (?, ?)", [ ("EURUSD", f"2024-01-01T{hour:02d}:00:00+00:00") for hour in range(24) ], "_load_symbol_start_datetimes", {"symbols": ["EURUSD"]}, ( "GROUP BY symbol", ("EURUSD", None), datetime(2024, 1, 1, 23, tzinfo=UTC), ), id="symbol-scoped-sqlite-aggregation", ), ], ) def test_incremental_start_query_uses_sqlite_aggregation( self, tmp_path: Path, mocker: MockerFixture, db_name: str, ddl: str, table_name: str, insert_sql: str, insert_rows: list[tuple[object, ...]], loader_name: str, loader_kwargs: dict[str, object], expected: tuple[str, tuple[str, int | None], datetime], ) -> None: """Test incremental resume aggregates in SQLite instead of scanning rows.""" expected_group_by, expected_key, expected_start = expected fallback = datetime(2024, 1, 1, tzinfo=UTC) with sqlite3.connect(tmp_path / f"{db_name}.db") as conn: conn.execute(ddl) conn.executemany(insert_sql, insert_rows) execute_spy = mocker.spy(conn, "execute") loader = getattr(history, loader_name) starts = loader( conn, table_name, fallback_start=fallback, **loader_kwargs, ) query = str(execute_spy.call_args[0][0]) assert "MAX(" in query assert expected_group_by in query assert "ORDER BY" not in query assert starts[expected_key] == expected_start 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,) @pytest.mark.parametrize( ("table", "columns", "match"), [ ("bad table", ["id"], "Invalid table name"), ("rates", ["bad column"], "Invalid column names"), ], ids=["invalid-table", "invalid-columns"], ) def test_drop_duplicates_rejects_invalid_identifiers( self, table: str, columns: list[str], match: str, ) -> None: """Test invalid table or column names raise ValueError.""" cursor = sqlite3.connect(":memory:").cursor() with pytest.raises(ValueError, match=match): drop_duplicates_in_table(cursor, table, columns) @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: [ DedupScope( "symbol = ? AND timeframe = ? AND time >= ?", ("EURUSD", 1, boundary), frozenset({"symbol", "timeframe", "time"}), ), ], }, ) 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), ] def test_scoped_dedup_matches_numeric_and_iso_times(self, tmp_path: Path) -> None: """Test scoped dedup collapses numeric epoch rows with canonical ISO writes.""" boundary = datetime(2024, 1, 2, tzinfo=UTC) dedup_scopes: dict[Dataset, list[DedupScope]] = {} time_expr = history._sqlite_normalized_time_expression( # type: ignore[reportPrivateUsage] "time" ) with sqlite3.connect(tmp_path / "numeric-time-dedup.db") as conn: conn.execute( "CREATE TABLE rates( symbol TEXT, timeframe INTEGER, time, 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, 1704153600, 2.0), ("EURUSD", 1, "2024-01-02T00:00:00+00:00", 9.9), ], ) history._record_dedup_scope( # type: ignore[reportPrivateUsage] dedup_scopes, Dataset.rates, f"symbol = ? AND timeframe = ? AND {time_expr} >= ?", ("EURUSD", 1, boundary.isoformat()), frozenset({"symbol", "timeframe", "time"}), ) deduplicate_history_tables( conn, {Dataset.rates: {"symbol", "timeframe", "time", "open"}}, {Dataset.rates}, dedup_scopes, ) rows = conn.execute( "SELECT time, typeof(time), open FROM rates ORDER BY ROWID", ).fetchall() assert rows == [ ("2024-01-01T00:00:00+00:00", "text", 1.0), ("2024-01-02T00:00:00+00:00", "text", 9.9), ] def test_unusable_scope_falls_back_to_table_dedup(self, tmp_path: Path) -> None: """Test scopes with missing columns do not break stable-key dedup.""" boundary = datetime(2024, 1, 1, tzinfo=UTC) with sqlite3.connect(tmp_path / "orders-without-time.db") as conn: conn.execute( "CREATE TABLE history_orders(" " ticket INTEGER, symbol TEXT, time_setup TEXT, type INTEGER)", ) conn.executemany( "INSERT INTO history_orders(ticket, symbol, time_setup, type)" " VALUES (?, ?, ?, ?)", [ (1, "EURUSD", "2024-01-01T00:00:00+00:00", 0), (1, "EURUSD", "2024-01-01T00:00:01+00:00", 1), ], ) deduplicate_history_tables( conn, {Dataset.history_orders: {"ticket", "symbol", "time_setup", "type"}}, {Dataset.history_orders}, { Dataset.history_orders: [ DedupScope( "symbol = ? AND time >= ?", ("EURUSD", boundary), frozenset({"symbol", "time"}), ), ], }, ) rows = conn.execute( "SELECT ticket, time_setup, type FROM history_orders", ).fetchall() assert rows == [(1, "2024-01-01T00:00:01+00:00", 1)] def test_partially_unusable_scopes_only_run_usable_scopes( self, tmp_path: Path, ) -> None: """Test mixed scope filtering skips only scopes with missing columns.""" boundary = datetime(2024, 1, 2, tzinfo=UTC) with sqlite3.connect(tmp_path / "partial-scope-filter.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", 2.0), ("EURUSD", 1, "2024-01-02T00:00:00+00:00", 9.9), ("USDJPY", 1, "2024-01-02T00:00:00+00:00", 100.0), ("USDJPY", 1, "2024-01-02T00:00:00+00:00", 101.0), ], ) deduplicate_history_tables( conn, {Dataset.rates: {"symbol", "timeframe", "time", "open"}}, {Dataset.rates}, { Dataset.rates: [ DedupScope( "symbol = ? AND timeframe = ? AND time >= ?", ("EURUSD", 1, boundary), frozenset({"symbol", "timeframe", "time"}), ), DedupScope( "symbol = ? AND timeframe = ? AND broker = ?", ("USDJPY", 1, "demo"), frozenset({"symbol", "timeframe", "broker"}), ), ], }, ) rows = conn.execute( "SELECT symbol, open FROM rates ORDER BY symbol, open", ).fetchall() assert rows == [ ("EURUSD", 9.9), ("USDJPY", 100.0), ("USDJPY", 101.0), ] 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.""" @pytest.mark.parametrize( ("ddl", "insert_sql", "rows", "expected"), [ pytest.param( ( "CREATE TABLE history_deals(" " ticket INTEGER, symbol TEXT, time TEXT, type INTEGER)" ), ( "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), ], datetime(2024, 1, 8, tzinfo=UTC), id="uses-type-column", ), pytest.param( ("CREATE TABLE history_deals( ticket INTEGER, symbol TEXT, time TEXT)"), "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"), ], datetime(2024, 1, 7, tzinfo=UTC), id="falls-back-to-empty-symbol", ), pytest.param( "CREATE TABLE history_deals(ticket INTEGER, time TEXT)", "INSERT INTO history_deals(ticket, time) VALUES (?, ?)", [(1, "2024-01-05T00:00:00+00:00")], datetime(2024, 1, 1, tzinfo=UTC), id="without-identifying-columns", ), ], ) def test_get_history_deals_account_event_start_datetime( self, tmp_path: Path, ddl: str, insert_sql: str, rows: list[tuple[object, ...]], expected: datetime, ) -> None: """Test account-event start resolution across identifying-column variants.""" fallback = datetime(2024, 1, 1, tzinfo=UTC) with sqlite3.connect(tmp_path / "account-start.db") as conn: conn.execute(ddl) conn.executemany(insert_sql, rows) assert ( get_history_deals_account_event_start_datetime( conn, fallback_start=fallback, ) == expected ) 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] @pytest.mark.parametrize( ("frame", "start_by_symbol", "account_event_start", "expected_tickets"), [ pytest.param( pd.DataFrame({ "ticket": [1], "symbol": ["EURUSD"], "time": [datetime(2024, 1, 5, tzinfo=UTC).isoformat()], "type": [2], }), {"EURUSD": datetime(2024, 1, 1, tzinfo=UTC)}, datetime(2024, 1, 10, tzinfo=UTC), [], id="excludes-symbolized-account-events-from-trade-cursor", ), pytest.param( pd.DataFrame({ "ticket": [1], "time": ["2024-01-03T00:00:00+00:00"], "type": [2], }), {"EURUSD": datetime(2024, 1, 10, tzinfo=UTC)}, datetime(2024, 1, 2, tzinfo=UTC), [1], id="keeps-account-events-without-symbol-column", ), ], ) def test_filter_incremental_account_event_edge_cases( self, frame: pd.DataFrame, start_by_symbol: dict[str, datetime], account_event_start: datetime, expected_tickets: list[int], ) -> None: """Test account-event rows are scoped by symbol column presence. A symbolized account-event row follows only the EURUSD trade cursor and is excluded when its time falls before account_event_start; when history_deals has no symbol column, account events are kept using account_event_start. """ filtered = filter_incremental_history_deals_frame( frame, ["EURUSD"], start_by_symbol, account_event_start, ) assert filtered["ticket"].tolist() == expected_tickets @pytest.mark.parametrize( "frame", [ pytest.param( pd.DataFrame({ "ticket": [1], "symbol": ["EURUSD"], "time": ["not-a-datetime"], "type": [0], }), id="unparseable-time", ), pytest.param( pd.DataFrame({ "ticket": [1], "time": ["2024-01-03T00:00:00+00:00"], }), id="trade-rows-without-symbol-column", ), pytest.param( pd.DataFrame({ "ticket": [1], "symbol": ["EURUSD"], "type": [0], }), id="rows-without-time-column", ), ], ) def test_filter_incremental_returns_empty_for_invalid_rows( self, frame: pd.DataFrame, ) -> None: """Test incremental filtering drops rows that cannot be evaluated.""" 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_incremental_rates_deduplicate_legacy_naive_boundary_rows( self, tmp_path: Path, ) -> None: """Test repeated boundary refetches deduplicate legacy naive SQLite times.""" def copy_rates_range_as_df( *, symbol: str, timeframe: int, date_from: datetime, date_to: datetime, ) -> pd.DataFrame: del symbol, timeframe, date_to rows_by_start = { datetime(2024, 1, 1, 0, 1, tzinfo=UTC): [ ("2024-01-01 00:01:00", 1.1), ("2024-01-01 00:02:00", 1.2), ], datetime(2024, 1, 1, 0, 2, tzinfo=UTC): [ ("2024-01-01 00:02:00", 1.2), ("2024-01-01 00:03:00", 1.3), ], } rows = rows_by_start[date_from] return pd.DataFrame({ "time": pd.to_datetime([row[0] for row in rows]), "open": [row[1] for row in rows], }) client = MagicMock() client.copy_rates_range_as_df.side_effect = copy_rates_range_as_df fallback = datetime(2024, 1, 1, tzinfo=UTC) end = datetime(2024, 1, 2, tzinfo=UTC) with sqlite3.connect(tmp_path / "legacy-naive-boundary.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-01 00:00:00", 1.0), ("EURUSD", 1, "2024-01-01 00:01:00", 1.1), ], ) write_incremental_datasets( conn, client, ["EURUSD"], {Dataset.rates}, [1], 0, fallback, end, deduplicate=True, create_rate_views=False, with_views=False, include_account_events=False, ) write_incremental_datasets( conn, client, ["EURUSD"], {Dataset.rates}, [1], 0, fallback, end, deduplicate=True, create_rate_views=False, with_views=False, include_account_events=False, ) result = pd.read_sql_query( # type: ignore[reportUnknownMemberType] "SELECT symbol, timeframe, time, open FROM rates ORDER BY ROWID", conn, ) result["time"] = pd.to_datetime(result["time"], utc=True, format="mixed") duplicate_counts = result.groupby(["symbol", "timeframe", "time"]).size() assert duplicate_counts[duplicate_counts > 1].empty pd.testing.assert_frame_equal( result.reset_index(drop=True), pd.DataFrame({ "symbol": ["EURUSD", "EURUSD", "EURUSD", "EURUSD"], "timeframe": [1, 1, 1, 1], "time": pd.to_datetime( [ "2024-01-01T00:00:00+00:00", "2024-01-01T00:01:00+00:00", "2024-01-01T00:02:00+00:00", "2024-01-01T00:03:00+00:00", ], utc=True, ), "open": [1.0, 1.1, 1.2, 1.3], }), ) 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_incremental_orders_without_time_deduplicate_by_ticket( self, tmp_path: Path, caplog: pytest.LogCaptureFixture, ) -> None: """Test incremental history_orders without time deduplicate safely.""" def history_orders_get_as_df(**kwargs: object) -> pd.DataFrame: if kwargs["symbol"] == "GBPUSD": return pd.DataFrame() return pd.DataFrame({ "ticket": [1, 1], "symbol": ["EURUSD", "EURUSD"], "time_setup": [ "2024-01-01T00:00:00+00:00", "2024-01-01T00:00:01+00:00", ], "type": [0, 1], }) client = MagicMock() client.history_orders_get_as_df.side_effect = history_orders_get_as_df start = datetime(2024, 1, 1, tzinfo=UTC) end = datetime(2024, 1, 2, tzinfo=UTC) with ( sqlite3.connect(tmp_path / "incremental-orders-without-time.db") as conn, caplog.at_level(logging.WARNING, logger="mt5cli.history"), ): write_incremental_datasets( conn, client, ["EURUSD", "GBPUSD"], {Dataset.history_orders}, [], 0, start, end, deduplicate=True, create_rate_views=False, with_views=False, include_account_events=False, ) rows = conn.execute( "SELECT ticket, time_setup, type FROM history_orders", ).fetchall() assert rows == [(1, "2024-01-01T00:00:01+00:00", 1)] assert "Skipping history_orders: dataset returned no columns" in caplog.text 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.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.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() @pytest.mark.parametrize("flags", ["BAD", 7]) def test_resolve_history_tick_flags_invalid(self, flags: str | int) -> None: """Test invalid tick flags raise ValueError.""" with pytest.raises(ValueError, match="Invalid tick flags"): resolve_history_tick_flags(flags) 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,) @pytest.mark.parametrize( ("frame", "ddl", "db_name"), [ pytest.param( pd.DataFrame({ "ticket": [1], "symbol": [""], "time": ["2024-01-02T00:00:00+00:00"], }), "CREATE TABLE history_deals( ticket INTEGER, symbol TEXT, time TEXT)", "deals-without-type.db", id="dedup-scope-without-type-column", ), pytest.param( pd.DataFrame({ "ticket": [1], "time": ["2024-01-02T00:00:00+00:00"], "type": [2], }), "CREATE TABLE history_deals(ticket INTEGER, time TEXT, type INTEGER)", "no-symbol-deals.db", id="dedup-scope-without-symbol-column", ), ], ) def test_account_event_dedup_scope_with_missing_column( self, tmp_path: Path, frame: pd.DataFrame, ddl: str, db_name: str, ) -> None: """Test account-event dedup falls back safely when a scope column is missing. Without a type column, dedup scope falls back to empty symbols; without a symbol column, per-symbol dedup scope is skipped. Both still write the row. """ client = MagicMock() client.history_deals_get_as_df.return_value = frame start = datetime(2024, 1, 1, tzinfo=UTC) end = datetime(2024, 1, 3, tzinfo=UTC) with sqlite3.connect(tmp_path / db_name) as conn: conn.execute(ddl) 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) class TestRateSourceHelpers: """Tests for generic rate-source SDK helpers.""" def test_rate_target_timeframe_int(self) -> None: """Test RateTarget resolves named and integer timeframes.""" target = RateTarget(symbol="EURUSD", timeframe="M1") assert target.timeframe == 1 assert target.timeframe_int == 1 assert RateTarget(symbol="EURUSD", timeframe=16385).timeframe_int == 16385 def test_build_rate_targets_row_major(self) -> None: """Test targets are built in row-major symbol/timeframe order.""" targets = build_rate_targets(["EURUSD", "GBPUSD"], ["M1", "H1"]) assert [(t.symbol, t.timeframe) for t in targets] == [ ("EURUSD", 1), ("EURUSD", 16385), ("GBPUSD", 1), ("GBPUSD", 16385), ] def test_build_rate_targets_allows_missing_symbol(self) -> None: """Test missing symbols produce None-symbol targets when allowed.""" targets = build_rate_targets([], ["M1", "H1"], allow_missing_symbol=True) assert [(t.symbol, t.timeframe) for t in targets] == [ (None, 1), (None, 16385), ] @pytest.mark.parametrize( ("symbols", "timeframes", "match"), [ (["EURUSD"], [], "At least one timeframe"), ([], ["M1"], "At least one symbol"), ], ) def test_build_rate_targets_rejects_empty( self, symbols: list[str], timeframes: list[str], match: str, ) -> None: """Test target building input validation.""" with pytest.raises(ValueError, match=match): build_rate_targets(symbols, timeframes) def test_resolve_rate_tables_uses_explicit_tables(self) -> None: """Test explicit tables bypass view resolution when counts match.""" targets = build_rate_targets([], ["M1", "H1"], allow_missing_symbol=True) assert resolve_rate_tables(None, targets, ["t1", "t2"]) == ["t1", "t2"] @pytest.mark.parametrize( ("targets", "explicit_tables", "path_kind", "require_existing", "match"), [ pytest.param( build_rate_targets(["EURUSD"], ["M1"]), ["t1", "t2"], "none", False, "Expected 1 explicit table", id="mismatched-explicit-count", ), pytest.param( [], None, "none", False, "At least one rate target", id="empty-targets", ), pytest.param( build_rate_targets([], ["M1"], allow_missing_symbol=True), None, "none", False, "without a symbol", id="none-symbol-without-explicit", ), pytest.param( build_rate_targets(["EURUSD"], ["M1"]), None, "none", True, "SQLite database not found", id="none-path-require-existing", ), pytest.param( build_rate_targets(["EURUSD"], ["M1"]), None, "missing", True, "SQLite database not found", id="missing-db-require-existing", ), ], ) def test_resolve_rate_tables_rejects_invalid_inputs( self, tmp_path: Path, targets: list[RateTarget], explicit_tables: list[str] | None, path_kind: str, require_existing: bool, match: str, ) -> None: """Test resolve_rate_tables input and strict-mode validation.""" db_path = None if path_kind == "none" else tmp_path / "missing.db" with pytest.raises(ValueError, match=match): resolve_rate_tables( db_path, targets, explicit_tables=explicit_tables, require_existing=require_existing, ) def test_resolve_rate_tables_resolves_view_names(self) -> None: """Test symbol targets resolve to default view names without a database.""" targets = build_rate_targets(["EURUSD"], ["M1", "H1"]) assert resolve_rate_tables(None, targets) == [ "rate_EURUSD__1", "rate_EURUSD__16385", ] def test_resolve_rate_tables_missing_view_with_require_existing_raises( self, tmp_path: Path, ) -> None: """Test strict mode rejects databases without managed rate views.""" db_path = tmp_path / "no-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), ) targets = build_rate_targets(["EURUSD"], ["M1"]) with pytest.raises(ValueError, match="No rate compatibility view exists"): resolve_rate_tables(db_path, targets, require_existing=True) def test_resolve_rate_tables_with_require_existing_resolves_views( self, tmp_path: Path, ) -> None: """Test strict mode resolves existing managed rate views.""" db_path = tmp_path / "strict-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), ) create_rate_compatibility_views(conn) targets = build_rate_targets(["EURUSD"], ["M1"]) assert resolve_rate_tables(db_path, targets, require_existing=True) == [ "rate_EURUSD__1", ] def test_resolve_rate_tables_batches_sqlite_metadata( self, tmp_path: Path, mocker: MockerFixture, ) -> None: """Test resolving multiple targets loads SQLite metadata once.""" db_path = tmp_path / "batch-rate-tables.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), ("GBPUSD", 1, "2024-01-01T00:00:00+00:00", 1.2), ], ) create_rate_compatibility_views(conn) counts_spy = mocker.spy(history, "_load_rates_timeframe_counts") views_spy = mocker.spy(history, "_load_existing_rate_views") targets = build_rate_targets(["EURUSD", "GBPUSD"], ["M1", "H1"]) assert resolve_rate_tables(db_path, targets) == [ "rate_EURUSD__M1_1", "rate_EURUSD__H1_16385", "rate_GBPUSD__1", "rate_GBPUSD__16385", ] assert counts_spy.call_count == 1 assert views_spy.call_count == 1 def test_load_rate_series_from_sqlite(self, tmp_path: Path) -> None: """Test loading multiple rate series keyed by symbol and timeframe.""" db_path = tmp_path / "series.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", 1, "2024-01-01T00:01:00+00:00", 1.1), ], ) create_rate_compatibility_views(conn) targets = build_rate_targets(["EURUSD"], ["M1"]) result = load_rate_series_from_sqlite(db_path, targets, count=2) assert set(result) == {("EURUSD", 1)} assert len(result["EURUSD", 1]) == 2 def test_load_rate_series_by_granularity(self, tmp_path: Path) -> None: """Test loading rate series keyed by symbol and granularity name.""" db_path = tmp_path / "granularity.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-01T00:00:00+00:00", 1.1), ], ) create_rate_compatibility_views(conn) result = load_rate_series_by_granularity( db_path, ["EURUSD"], ["M1", "H1"], count=1, ) assert set(result) == {("EURUSD", "M1"), ("EURUSD", "H1")} def test_report_rate_gaps_reports_one_row_per_gap(self, tmp_path: Path) -> None: """Gap reports emit one row for each detected missing interval run.""" db_path = tmp_path / "gaps.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", 1, "2024-01-01T00:01:00+00:00", 1.1), ("EURUSD", 1, "2024-01-01T00:03:00+00:00", 1.2), ], ) conn.execute( 'CREATE VIEW "rate_EURUSD__M1_1" AS ' "SELECT time, close FROM rates " "WHERE symbol = 'EURUSD' AND timeframe = 1", ) result = report_rate_gaps( conn, "rate_EURUSD__M1_1", granularity_seconds=60, ) records = cast("list[dict[str, object]]", result.to_dict("records")) assert records == [ { "table": "rate_EURUSD__M1_1", "symbol": "EURUSD", "timeframe": 1, "granularity": "M1", "granularity_seconds": 60, "gap_start": datetime(2024, 1, 1, 0, 2, tzinfo=UTC), "gap_end": datetime(2024, 1, 1, 0, 2, tzinfo=UTC), "missing_intervals": 1, }, ] def test_report_rate_gaps_parses_numeric_sqlite_times_as_epoch_seconds( self, tmp_path: Path, ) -> None: """Numeric SQLite timestamps must be interpreted as Unix seconds.""" db_path = tmp_path / "numeric-gaps.db" with sqlite3.connect(db_path) as conn: conn.execute("CREATE TABLE custom_rates(time INTEGER, close REAL)") conn.executemany( "INSERT INTO custom_rates(time, close) VALUES (?, ?)", [ (1704067200, 1.0), (1704067320, 1.1), ], ) result = report_rate_gaps( conn, "custom_rates", granularity_seconds=60, ) assert len(result) == 1 assert result.iloc[0]["missing_intervals"] == 1 def test_report_rate_gaps_empty_schema_for_zero_gap_tables( self, tmp_path: Path, ) -> None: """Tables without gaps return the stable empty result schema.""" db_path = tmp_path / "no-gaps.db" with sqlite3.connect(db_path) as conn: conn.execute("CREATE TABLE custom_rates(time TEXT, close REAL)") conn.executemany( "INSERT INTO custom_rates(time, close) VALUES (?, ?)", [ ("2024-01-01T00:00:00+00:00", 1.0), ("2024-01-01T00:01:00+00:00", 1.1), ], ) result = report_rate_gaps( conn, "custom_rates", granularity_seconds=60, ) assert list(result.columns) == [ "table", "symbol", "timeframe", "granularity", "granularity_seconds", "gap_start", "gap_end", "missing_intervals", ] assert result.empty def test_report_rate_gaps_filters_by_min_gap_intervals( self, tmp_path: Path, ) -> None: """Small gaps are filtered out when min_gap_intervals is raised.""" db_path = tmp_path / "filtered-gaps.db" with sqlite3.connect(db_path) as conn: conn.execute("CREATE TABLE custom_rates(time TEXT, close REAL)") conn.executemany( "INSERT INTO custom_rates(time, close) VALUES (?, ?)", [ ("2024-01-01T00:00:00+00:00", 1.0), ("2024-01-01T00:02:00+00:00", 1.1), ], ) result = report_rate_gaps( conn, "custom_rates", granularity_seconds=60, min_gap_intervals=2, ) assert result.empty def test_report_rate_gaps_computes_gaps_per_series_key( self, tmp_path: Path, ) -> None: """Managed rates tables must detect gaps within each symbol/timeframe series.""" db_path = tmp_path / "series-gaps.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), ("GBPUSD", 1, "2024-01-01T00:01:00+00:00", 1.1), ("EURUSD", 1, "2024-01-01T00:02:00+00:00", 1.2), ], ) result = report_rate_gaps( conn, "rates", granularity_seconds=60, ) records = cast("list[dict[str, object]]", result.to_dict("records")) assert records == [ { "table": "rates", "symbol": "EURUSD", "timeframe": 1, "granularity": "M1", "granularity_seconds": 60, "gap_start": datetime(2024, 1, 1, 0, 1, tzinfo=UTC), "gap_end": datetime(2024, 1, 1, 0, 1, tzinfo=UTC), "missing_intervals": 1, }, ] def test_rate_gap_private_helpers_and_validation(self) -> None: """Private helpers should preserve schema and reject invalid inputs.""" empty = history._empty_rate_gap_report() # type: ignore[attr-defined] assert list(empty.columns) == [ "table", "symbol", "timeframe", "granularity", "granularity_seconds", "gap_start", "gap_end", "missing_intervals", ] class _BadInt: def __int__(self) -> int: msg = "bad-int" raise ValueError(msg) assert history._coerce_optional_int(None) is None # type: ignore[attr-defined] false_value: object = False assert history._coerce_optional_int(false_value) is None # type: ignore[attr-defined] assert history._coerce_optional_int(" +7 ") == 7 # type: ignore[attr-defined] assert history._coerce_optional_int("bad") is None # type: ignore[attr-defined] assert history._coerce_optional_int(object()) is None # type: ignore[attr-defined] assert history._coerce_optional_int(_BadInt()) is None # type: ignore[attr-defined] metadata = history._rate_gap_metadata( # type: ignore[attr-defined] "custom_rates", pd.DataFrame({ "symbol": ["EURUSD", "GBPUSD"], "timeframe": [1, 1], }), granularity_seconds=60, ) assert metadata["symbol"] is None assert metadata["timeframe"] == 1 assert metadata["granularity"] == "M1" fallback_metadata = history._rate_gap_metadata( # type: ignore[attr-defined] "rate_USDJPY__M1_1", pd.DataFrame({"time": []}), granularity_seconds=60, ) assert fallback_metadata["symbol"] == "USDJPY" assert fallback_metadata["timeframe"] == 1 assert fallback_metadata["granularity"] == "M1" unique_symbol_metadata = history._rate_gap_metadata( # type: ignore[attr-defined] "custom_rates", pd.DataFrame({"symbol": ["EURUSD"], "timeframe": [1]}), granularity_seconds=60, ) assert unique_symbol_metadata["symbol"] == "EURUSD" multi_timeframe_metadata = history._rate_gap_metadata( # type: ignore[attr-defined] "custom_rates", pd.DataFrame({"timeframe": [1, 5]}), granularity_seconds=60, ) assert multi_timeframe_metadata["timeframe"] is None assert multi_timeframe_metadata["granularity"] is None @pytest.mark.parametrize( ("granularity_seconds", "min_gap_intervals", "match"), [ pytest.param( 0, 1, "granularity_seconds must be positive", id="bad-seconds" ), pytest.param(60, 0, "min_gap_intervals must be positive", id="bad-min-gap"), ], ) def test_report_rate_gaps_rejects_invalid_parameters( self, tmp_path: Path, granularity_seconds: int, min_gap_intervals: int, match: str, ) -> None: """Gap reports reject non-positive granularity and min-gap values.""" db_path = tmp_path / "invalid-gaps.db" with sqlite3.connect(db_path) as conn: conn.execute("CREATE TABLE custom_rates(time TEXT, close REAL)") conn.execute( "INSERT INTO custom_rates(time, close) VALUES (?, ?)", ("2024-01-01T00:00:00+00:00", 1.0), ) with pytest.raises(ValueError, match=match): report_rate_gaps( conn, "custom_rates", granularity_seconds=granularity_seconds, min_gap_intervals=min_gap_intervals, ) def test_report_rate_gaps_rejects_unparseable_timestamps( self, tmp_path: Path, ) -> None: """Unparseable table times should fail clearly.""" db_path = tmp_path / "bad-times.db" with sqlite3.connect(db_path) as conn: conn.execute("CREATE TABLE custom_rates(time TEXT, close REAL)") conn.execute( "INSERT INTO custom_rates(time, close) VALUES (?, ?)", ("bad-time", 1.0), ) with pytest.raises(ValueError, match="contains unparsable time values"): report_rate_gaps( conn, "custom_rates", granularity_seconds=60, ) def test_report_rate_gaps_empty_and_single_row_sources_return_empty( self, tmp_path: Path, ) -> None: """Empty or single-row sources cannot produce gap rows.""" db_path = tmp_path / "too-short.db" with sqlite3.connect(db_path) as conn: conn.execute("CREATE TABLE custom_rates(time TEXT, close REAL)") assert report_rate_gaps(conn, "custom_rates", granularity_seconds=60).empty conn.execute( "INSERT INTO custom_rates(time, close) VALUES (?, ?)", ("2024-01-01T00:00:00+00:00", 1.0), ) assert report_rate_gaps(conn, "custom_rates", granularity_seconds=60).empty def test_load_rate_series_by_granularity_explicit_tables( self, tmp_path: Path, ) -> None: """Test explicit tables with None-symbol targets key by granularity.""" db_path = tmp_path / "granularity-explicit.db" with sqlite3.connect(db_path) as conn: conn.execute("CREATE TABLE custom_view(time TEXT, close REAL)") conn.execute( "INSERT INTO custom_view(time, close) VALUES (?, ?)", ("2024-01-01T00:00:00+00:00", 1.0), ) result = load_rate_series_by_granularity( db_path, [], ["M1"], count=1, explicit_tables=["custom_view"], allow_missing_symbol=True, ) assert set(result) == {(None, "M1")} def test_load_rate_series_reuses_path_connection( self, tmp_path: Path, mocker: MockerFixture, ) -> None: """Test loading from a path opens SQLite once for resolve and reads.""" db_path = tmp_path / "single-open-series.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) connect_spy = mocker.spy(history.sqlite3, "connect") result = load_rate_series_from_sqlite( db_path, build_rate_targets(["EURUSD"], ["M1"]), count=1, ) assert set(result) == {("EURUSD", 1)} assert connect_spy.call_count == 1 def test_load_rate_series_with_explicit_tables(self, tmp_path: Path) -> None: """Test explicit tables and None-symbol targets load series.""" db_path = tmp_path / "explicit.db" with sqlite3.connect(db_path) as conn: conn.execute("CREATE TABLE custom_view(time TEXT, close REAL)") conn.execute( "INSERT INTO custom_view(time, close) VALUES (?, ?)", ("2024-01-01T00:00:00+00:00", 1.0), ) targets = build_rate_targets([], ["M1"], allow_missing_symbol=True) result = load_rate_series_from_sqlite( db_path, targets, count=1, explicit_tables=["custom_view"], ) assert set(result) == {(None, 1)} @pytest.mark.parametrize( ("targets", "count", "explicit_tables", "match"), [ pytest.param( build_rate_targets(["EURUSD"], ["M1"]), 0, None, "count must be positive", id="non-positive-count", ), pytest.param( [], 1, None, "At least one rate target", id="empty-targets", ), pytest.param( build_rate_targets([], ["M1"], allow_missing_symbol=True), 1, None, "without a symbol", id="none-symbol-without-explicit", ), pytest.param( [RateTarget("EURUSD", 1), RateTarget("EURUSD", "M1")], 1, None, r"Duplicate rate target: \('EURUSD', 1\)", id="duplicate-targets", ), pytest.param( [RateTarget("EURUSD", 1), RateTarget("EURUSD", 1)], 1, ["custom_view", "custom_view"], r"Duplicate rate target: \('EURUSD', 1\)", id="duplicate-targets-with-explicit", ), ], ) def test_load_rate_series_rejects_invalid_inputs( self, targets: list[RateTarget], count: int, explicit_tables: list[str] | None, match: str, ) -> None: """Test load_rate_series_from_sqlite validation runs before opening SQLite.""" with pytest.raises(ValueError, match=match): load_rate_series_from_sqlite( "unused.db", targets, count=count, explicit_tables=explicit_tables, ) def test_load_rate_series_requires_existing_managed_views( self, tmp_path: Path, ) -> None: """Test loading without explicit tables requires managed rate views.""" db_path = tmp_path / "no-managed-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), ) targets = build_rate_targets(["EURUSD"], ["M1"]) with pytest.raises(ValueError, match="No rate compatibility view exists"): load_rate_series_from_sqlite(db_path, targets, count=1)