diff --git a/README.md b/README.md index 5ac5317..353bb8d 100644 --- a/README.md +++ b/README.md @@ -302,7 +302,11 @@ WHERE run_id = (SELECT MAX(run_id) FROM snapshot_runs WHERE status = 'ok'); SELECT symbol, total_profit FROM grafana_trade_stats ORDER BY total_profit DESC; ``` -> **Note**: OpenTelemetry integration is intentionally not part of this release and is tracked separately. +#### Grafana and telemetry API docs + +The shipped Grafana helpers are documented in [`docs/api/grafana.md`](docs/api/grafana.md), including `publish_grafana_copy()` for creating a WAL-safe published SQLite copy for Grafana. + +OpenTelemetry metrics are documented in [`docs/api/telemetry.md`](docs/api/telemetry.md), including `enable_otel_metrics()`, `configure_metrics()`, the `mt5cli[otel]` extra, and `OTEL_EXPORTER_OTLP_ENDPOINT`. ### Incremental history SDK diff --git a/docs/api/grafana.md b/docs/api/grafana.md new file mode 100644 index 0000000..d13d4b3 --- /dev/null +++ b/docs/api/grafana.md @@ -0,0 +1,25 @@ +# Grafana + +::: mt5cli.grafana + +## Grafana-ready SQLite workflow + +Use `ensure_grafana_schema(conn)` or the `grafana-schema` CLI command to create +the snapshot tables, `grafana_*` views, and supporting indexes in one step. + +`publish_grafana_copy(source, target)` creates a consistent SQLite copy via the +SQLite backup API, which is useful when the primary database is running in WAL +mode and Grafana should read from a separate published file. + +## Main APIs + +- `ensure_grafana_schema()`: idempotently creates snapshot tables, Grafana + views, and indexes. +- `create_grafana_views()`: rebuilds the shipped `grafana_*` views. +- `create_grafana_indexes()`: creates read-oriented indexes for Grafana + queries. +- `publish_grafana_copy()`: writes an atomic, WAL-safe published copy for a + Grafana datasource. + +For end-to-end snapshot collection examples, see the Grafana and observability +section in the project `README.md`. diff --git a/docs/api/index.md b/docs/api/index.md index 5ab2183..38c5a64 100644 --- a/docs/api/index.md +++ b/docs/api/index.md @@ -18,6 +18,8 @@ responsibilities. | [SDK](sdk.md) | Module-level fetch helpers, multi-account collectors, incremental history | | [Trading](trading.md) | Trading-capable sessions and operational helpers | | [History Collection (SQLite)](history.md) | SQLite schema, incremental writes, dedup, and rate views | +| [Telemetry](telemetry.md) | OpenTelemetry metrics setup, meters, and emitted metric names | +| [Grafana](grafana.md) | Grafana-ready SQLite schema, views, snapshots, and published copies | | [CLI](cli.md) | Typer commands that delegate to the Python API | | [Utils](utils.md) | Parsing helpers and Click parameter types | diff --git a/docs/api/telemetry.md b/docs/api/telemetry.md new file mode 100644 index 0000000..4677583 --- /dev/null +++ b/docs/api/telemetry.md @@ -0,0 +1,50 @@ +# Telemetry + +::: mt5cli.telemetry + +## Enabling OpenTelemetry metrics + +Install the optional exporter dependencies with: + +```bash +uv add 'mt5cli[otel]' +``` + +Then enable the default OTLP HTTP pipeline: + +```python +from mt5cli.telemetry import enable_otel_metrics + +enable_otel_metrics(service_name="mt5cli") +``` + +When `readers=None`, `enable_otel_metrics()` builds a +`PeriodicExportingMetricReader` backed by the OTLP HTTP exporter and reads the +endpoint from `OTEL_EXPORTER_OTLP_ENDPOINT`. + +If your application already owns an OpenTelemetry `Meter`, wire mt5cli into it +directly with `configure_metrics(meter)`. + +## Emitted metric names + +`enable_otel_metrics()` / `configure_metrics()` register these instruments: + +- `mt5_history_update_duration_seconds` +- `mt5_history_update_rows_total` +- `mt5_history_update_failures_total` +- `mt5_snapshot_update_duration_seconds` +- `mt5_snapshot_update_failures_total` +- `mt5_account_balance` +- `mt5_account_equity` +- `mt5_account_margin` +- `mt5_account_margin_free` +- `mt5_account_margin_level` +- `mt5_position_profit` +- `mt5_position_volume` +- `mt5_terminal_connected` +- `mt5_terminal_trade_allowed` +- `mt5_terminal_trade_expert` +- `mt5_last_successful_update_timestamp` + +The history metrics use a `dataset` attribute. Account and position gauges add +labels such as `login`, `server`, and `symbol` where applicable. diff --git a/mkdocs.yml b/mkdocs.yml index a9ce160..9ba7d6d 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -65,6 +65,8 @@ nav: - SDK: api/sdk.md - Trading: api/trading.md - History Collection (SQLite): api/history.md + - Telemetry: api/telemetry.md + - Grafana: api/grafana.md - Utils: api/utils.md markdown_extensions: diff --git a/mt5cli/history.py b/mt5cli/history.py index 566815f..9e61e8a 100644 --- a/mt5cli/history.py +++ b/mt5cli/history.py @@ -12,7 +12,7 @@ from typing import TYPE_CHECKING, Literal, cast, overload import pandas as pd from pdmt5 import get_timeframe_name as _get_timeframe_name -from .schemas import DEDUP_KEYS, DataKind +from .schemas import DEDUP_KEYS, DataKind, ensure_utc_columns from .utils import ( TIMEFRAME_NAMES, Dataset, @@ -29,6 +29,13 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +_SQLITE_TEXT_TIME_COLUMNS: frozenset[str] = frozenset({ + "time", + "time_setup", + "time_done", +}) +_SQLITE_CANONICAL_TIME_FORMAT = "%Y-%m-%dT%H:%M:%f+00:00" + DEFAULT_HISTORY_TIMEFRAMES: tuple[str, ...] = TIMEFRAME_NAMES DEFAULT_HISTORY_DATASETS: frozenset[Dataset] = frozenset({ Dataset.rates, @@ -877,6 +884,72 @@ def parse_sqlite_timestamp(value: object) -> datetime | None: return None +def _serialize_sqlite_timestamp(value: object) -> str | None: + parsed = parse_sqlite_timestamp(value) + if parsed is None: + return None + utc_value = parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC) + utc_value = utc_value.astimezone(UTC) + timespec = "microseconds" if utc_value.microsecond else "seconds" + return utc_value.isoformat(timespec=timespec) + + +def _require_serialized_sqlite_timestamp(value: object) -> str: + serialized = _serialize_sqlite_timestamp(value) + if serialized is None: + msg = f"Invalid SQLite timestamp boundary: {value!r}" + raise ValueError(msg) + return serialized + + +def _canonicalize_sqlite_time_columns(frame: pd.DataFrame) -> pd.DataFrame: + columns = [ + column for column in _SQLITE_TEXT_TIME_COLUMNS if column in frame.columns + ] + if not columns: + return frame + normalized = ensure_utc_columns(frame, columns) + for column in columns: + normalized[column] = normalized[column].map(_serialize_sqlite_timestamp) + return normalized + + +def _sqlite_dedup_key_expression(column: str) -> str: + quoted = quote_sqlite_identifier(column) + if column != "time": + return quoted + normalized = _sqlite_normalized_time_expression(column) + return f"COALESCE({normalized}, CAST({quoted} AS TEXT))" + + +def _sqlite_normalized_time_expression(column: str) -> str: + """Return a canonical UTC timestamp expression for mixed SQLite time values.""" + quoted = quote_sqlite_identifier(column) + return ( + "COALESCE(" + f"strftime('{_SQLITE_CANONICAL_TIME_FORMAT}', {quoted}), " + f"strftime('{_SQLITE_CANONICAL_TIME_FORMAT}', {quoted}, 'unixepoch')" + ")" + ) + + +def _load_latest_parseable_time( + conn: sqlite3.Connection, + table: str, + *, + where_clause: str | None = None, + params: Sequence[object] = (), +) -> datetime | None: + quoted_table = quote_sqlite_identifier(table) + time_expr = _sqlite_normalized_time_expression("time") + query = f"SELECT time FROM {quoted_table} WHERE {time_expr} IS NOT NULL" # noqa: S608 + if where_clause: + query += f" AND {where_clause}" + query += f" ORDER BY {time_expr} DESC, ROWID DESC LIMIT 1" + row = conn.execute(query, tuple(params)).fetchone() + return parse_sqlite_timestamp(row[0] if row else None) + + def get_history_deals_account_event_start_datetime( conn: sqlite3.Connection, *, @@ -893,10 +966,11 @@ def get_history_deals_account_event_start_datetime( where_clause = "symbol IS NULL OR symbol = ''" else: return fallback_start - row = conn.execute( - f"SELECT MAX(time) FROM {table} WHERE {where_clause}", # noqa: S608 - ).fetchone() - parsed = parse_sqlite_timestamp(row[0] if row else None) + parsed = _load_latest_parseable_time( + conn, + table, + where_clause=where_clause, + ) return parsed if parsed is not None else fallback_start @@ -919,6 +993,68 @@ def _validate_rates_schema(columns: set[str]) -> None: raise ValueError(msg) +def _load_grouped_rate_start_datetimes( + conn: sqlite3.Connection, + table: str, + *, + symbols: Sequence[str], + timeframes: Sequence[int], + fallback_start: datetime, +) -> dict[tuple[str, int | None], datetime]: + symbol_placeholders = ", ".join("?" for _ in symbols) + timeframe_placeholders = ", ".join("?" for _ in timeframes) + time_expr = _sqlite_normalized_time_expression("time") + rows = conn.execute( + "SELECT symbol, timeframe, MAX(" # noqa: S608 + f"{time_expr}) FROM " + f"{quote_sqlite_identifier(table)}" + f" WHERE symbol IN ({symbol_placeholders})" + f" AND timeframe IN ({timeframe_placeholders})" + f" AND {time_expr} IS NOT NULL" + f" GROUP BY symbol, timeframe", + [*symbols, *timeframes], + ).fetchall() + parsed_by_key: dict[tuple[str, int | None], datetime] = {} + for row_symbol, row_timeframe, max_time in rows: + parsed = parse_sqlite_timestamp(max_time) + if parsed is not None: + parsed_by_key[str(row_symbol), int(row_timeframe)] = parsed + return { + (symbol, timeframe): parsed_by_key.get((symbol, timeframe), fallback_start) + for symbol in symbols + for timeframe in timeframes + } + + +def _load_symbol_start_datetimes( + conn: sqlite3.Connection, + table: str, + *, + symbols: Sequence[str], + fallback_start: datetime, +) -> dict[tuple[str, int | None], datetime]: + symbol_placeholders = ", ".join("?" for _ in symbols) + time_expr = _sqlite_normalized_time_expression("time") + rows = conn.execute( + "SELECT symbol, MAX(" # noqa: S608 + f"{time_expr}) FROM " + f"{quote_sqlite_identifier(table)}" + f" WHERE symbol IN ({symbol_placeholders})" + f" AND {time_expr} IS NOT NULL" + f" GROUP BY symbol", + list(symbols), + ).fetchall() + parsed_by_key: dict[tuple[str, int | None], datetime] = {} + for row_symbol, max_time in rows: + parsed = parse_sqlite_timestamp(max_time) + if parsed is not None: + parsed_by_key[str(row_symbol), None] = parsed + return { + (symbol, None): parsed_by_key.get((symbol, None), fallback_start) + for symbol in symbols + } + + def load_incremental_start_datetimes( conn: sqlite3.Connection, dataset: Dataset, @@ -942,55 +1078,28 @@ def load_incremental_start_datetimes( } return {(symbol, None): fallback_start for symbol in symbols} - parsed_by_key: dict[tuple[str, int | None], datetime] = {} if ( dataset is Dataset.rates and timeframes is not None and {"symbol", "timeframe"}.issubset(columns) ): - symbol_placeholders = ", ".join("?" for _ in symbols) - timeframe_placeholders = ", ".join("?" for _ in timeframes) - grouped_rates_query = ( - "SELECT symbol, timeframe, MAX(time) FROM " # noqa: S608 - f"{table} WHERE symbol IN ({symbol_placeholders})" - f" AND timeframe IN ({timeframe_placeholders})" - " GROUP BY symbol, timeframe" + return _load_grouped_rate_start_datetimes( + conn, + table, + symbols=symbols, + timeframes=timeframes, + fallback_start=fallback_start, ) - rows = conn.execute( - grouped_rates_query, - [*symbols, *timeframes], - ).fetchall() - for row_symbol, row_timeframe, max_time in rows: - parsed = parse_sqlite_timestamp(max_time) - if parsed is not None: - parsed_by_key[str(row_symbol), int(row_timeframe)] = parsed - return { - (symbol, timeframe): parsed_by_key.get( - (symbol, timeframe), - fallback_start, - ) - for symbol in symbols - for timeframe in timeframes - } if "symbol" in columns: - symbol_placeholders = ", ".join("?" for _ in symbols) - rows = conn.execute( - f"SELECT symbol, MAX(time) FROM {table}" # noqa: S608 - f" WHERE symbol IN ({symbol_placeholders}) GROUP BY symbol", - list(symbols), - ).fetchall() - for row_symbol, max_time in rows: - parsed = parse_sqlite_timestamp(max_time) - if parsed is not None: - parsed_by_key[str(row_symbol), None] = parsed - return { - (symbol, None): parsed_by_key.get((symbol, None), fallback_start) - for symbol in symbols - } + return _load_symbol_start_datetimes( + conn, + table, + symbols=symbols, + fallback_start=fallback_start, + ) - row = conn.execute(f"SELECT MAX(time) FROM {table}").fetchone() # noqa: S608 - parsed = parse_sqlite_timestamp(row[0] if row else None) + parsed = _load_latest_parseable_time(conn, table) shared_start = parsed if parsed is not None else fallback_start return {(symbol, None): shared_start for symbol in symbols} @@ -1029,7 +1138,8 @@ def append_dataframe( if len(frame.columns) == 0: logger.warning("Skipping %s: dataset returned no columns", table_name) return False - frame.to_sql( # type: ignore[reportUnknownMemberType] + writable = _canonicalize_sqlite_time_columns(frame) + writable.to_sql( # type: ignore[reportUnknownMemberType] table_name, conn, if_exists=if_exists.value, @@ -1108,15 +1218,21 @@ def drop_duplicates_in_table( if invalid := {column for column in ids if not column.isidentifier()}: msg = f"Invalid column names: {', '.join(sorted(invalid))}" raise ValueError(msg) - ids_csv = ", ".join(f'"{column}"' for column in ids) + ids_csv = ", ".join(_sqlite_dedup_key_expression(column) for column in ids) rowid_selector = "MIN" if keep == "first" else "MAX" + prepared_scope_params = tuple( + _require_serialized_sqlite_timestamp(value) + if isinstance(value, datetime) + else value + for value in scope_params + ) if scope_where: delete_sql = ( f"DELETE FROM {table} WHERE {scope_where} AND ROWID NOT IN" # noqa: S608 f" (SELECT {rowid_selector}(ROWID) FROM {table} WHERE {scope_where}" f" GROUP BY {ids_csv})" ) - cursor.execute(delete_sql, scope_params + scope_params) + cursor.execute(delete_sql, prepared_scope_params + prepared_scope_params) return cursor.execute( f"DELETE FROM {table} WHERE ROWID NOT IN" # noqa: S608 @@ -1436,11 +1552,12 @@ def _record_symbol_time_dedup( ) -> None: """Record a symbol-scoped deduplication window after an incremental write.""" written_tables.add(dataset) + time_expr = _sqlite_normalized_time_expression("time") _record_dedup_scope( dedup_scopes, dataset, - "symbol = ? AND time >= ?", - (symbol, start_date), + f"symbol = ? AND {time_expr} >= ?", + (symbol, _require_serialized_sqlite_timestamp(start_date)), frozenset({"symbol", "time"}), ) @@ -1602,11 +1719,16 @@ def _write_incremental_rates( written_columns, ): written_tables.add(Dataset.rates) + time_expr = _sqlite_normalized_time_expression("time") _record_dedup_scope( dedup_scopes, Dataset.rates, - "symbol = ? AND timeframe = ? AND time >= ?", - (symbol, timeframe, start_date), + f"symbol = ? AND timeframe = ? AND {time_expr} >= ?", + ( + symbol, + timeframe, + _require_serialized_sqlite_timestamp(start_date), + ), frozenset({"symbol", "timeframe", "time"}), ) @@ -1730,29 +1852,35 @@ def _write_incremental_history_deals( ): written_tables.add(Dataset.history_deals) columns = get_table_columns(conn, Dataset.history_deals.table_name) + time_expr = _sqlite_normalized_time_expression("time") if "symbol" in columns: for symbol in symbols: _record_dedup_scope( dedup_scopes, Dataset.history_deals, - "symbol = ? AND time >= ?", - (symbol, start_by_symbol[symbol, None]), + f"symbol = ? AND {time_expr} >= ?", + ( + symbol, + _require_serialized_sqlite_timestamp( + start_by_symbol[symbol, None] + ), + ), frozenset({"symbol", "time"}), ) if "type" in columns: _record_dedup_scope( dedup_scopes, Dataset.history_deals, - f"type NOT IN {_TRADE_DEAL_TYPES_SQL} AND time >= ?", - (account_event_start,), + f"type NOT IN {_TRADE_DEAL_TYPES_SQL} AND {time_expr} >= ?", + (_require_serialized_sqlite_timestamp(account_event_start),), frozenset({"type", "time"}), ) if "type" not in columns and "symbol" in columns: _record_dedup_scope( dedup_scopes, Dataset.history_deals, - "(symbol IS NULL OR symbol = '') AND time >= ?", - (account_event_start,), + f"(symbol IS NULL OR symbol = '') AND {time_expr} >= ?", + (_require_serialized_sqlite_timestamp(account_event_start),), frozenset({"symbol", "time"}), ) return diff --git a/pyproject.toml b/pyproject.toml index 37811af..7068cf1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "mt5cli" -version = "1.1.1" +version = "1.1.2" description = "Generic MT5 data and execution infrastructure for Python applications" authors = [{name = "dceoy", email = "dceoy@users.noreply.github.com"}] maintainers = [{name = "dceoy", email = "dceoy@users.noreply.github.com"}] @@ -10,6 +10,7 @@ readme = "README.md" requires-python = ">= 3.11, < 3.14" dependencies = [ "pdmt5>=1.0.0", + "pandas >= 2.2.2", "click >= 8.1.0", "typer >= 0.15.0", ] diff --git a/tests/test_history.py b/tests/test_history.py index 244668d..a0361b6 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -757,6 +757,17 @@ class TestParseSqliteTimestamp: """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.""" @@ -823,6 +834,167 @@ class TestIncrementalStart: 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"), [ @@ -962,6 +1134,249 @@ class TestIncrementalStart: 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.""" @@ -1115,6 +1530,46 @@ class TestDeduplication: ("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) @@ -1565,6 +2020,105 @@ class TestIncrementalHistoryDealsHelpers: 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, diff --git a/uv.lock b/uv.lock index 68ee68f..2262a7a 100644 --- a/uv.lock +++ b/uv.lock @@ -499,10 +499,11 @@ wheels = [ [[package]] name = "mt5cli" -version = "1.1.1" +version = "1.1.2" source = { editable = "." } dependencies = [ { name = "click" }, + { name = "pandas" }, { name = "pdmt5" }, { name = "typer" }, ] @@ -540,6 +541,7 @@ requires-dist = [ { name = "opentelemetry-api", marker = "extra == 'otel'" }, { name = "opentelemetry-exporter-otlp-proto-http", marker = "extra == 'otel'" }, { name = "opentelemetry-sdk", marker = "extra == 'otel'" }, + { name = "pandas", specifier = ">=2.2.2" }, { name = "pdmt5", specifier = ">=1.0.0" }, { name = "pyarrow", marker = "extra == 'parquet'", specifier = ">=19.0.0" }, { name = "typer", specifier = ">=0.15.0" },