gavindiaz/mt5cli
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: stabilize history timestamps and telemetry docs (#100)

Browse Source 
* fix: stabilize history timestamps and telemetry docs

* fix: preserve numeric epoch cursors in history SQLite queries

Normalize mixed ISO and unixepoch time values for incremental resume and scoped dedup so legacy numeric rows are not dropped by julianday filters.

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

* Bump version to v1.1.2

* fix: aggregate incremental start timestamps in SQLite

Use MAX on the normalized time expression with GROUP BY so incremental
resume loaders stay O(groups) instead of materializing every history row.

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

* test: parametrize duplicated incremental-start cases in TestIncrementalStart

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

---------

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Daichi Narushima
2026-07-04 01:03:40 +09:00
committed by GitHub
parent a5c2aa8e8f
commit efc0de230a
9 changed files with 829 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+5 -1
View File
@@ -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
docs/api/grafana.md
+25
View File
@@ -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`.
docs/api/index.md
+2
View File
@@ -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 |
docs/api/telemetry.md
+50
View File
@@ -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.
mkdocs.yml
+2
View File
@@ -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:
mt5cli/history.py
+186 -58
View File
@@ -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
pyproject.toml
+2 -1
View File
@@ -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",
]
tests/test_history.py
+554
View File
@@ -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,
uv.lock
Generated
+3 -1
View File
@@ -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" },