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

feat: add Grafana-ready SQLite observability (#86)

Browse Source 
* feat: add Grafana-ready SQLite observability (#79, #80, #81)

New `mt5cli/grafana.py` module with idempotent DDL helpers:
- `create_snapshot_tables` — five SQLite tables for time-series account,
  position, order, terminal, and run-status snapshots
- `create_grafana_views` — 13 `grafana_*` views with integer epoch-second
  `time` columns; missing source tables emit warnings and are skipped
- `create_grafana_indexes` — 9 performance indexes guarded by column checks
- `ensure_grafana_schema` — convenience wrapper calling all three above
- Insert helpers: `insert_account_snapshot`, `insert_position_snapshots`,
  `insert_order_snapshots`, `insert_terminal_snapshot`, `record_snapshot_run`

New stable SDK exports in `mt5cli.__init__` and `mt5cli.contract`:
- `update_observability` — appends a timestamped snapshot to a SQLite db
  using an already-connected `Mt5DataClient`; never places orders
- `update_observability_with_config` — standalone wrapper that opens and
  closes the MT5 connection automatically

New CLI commands (Collection panel):
- `grafana-schema` — idempotent schema setup, no MT5 connection required
- `snapshot` — append account/position/order/terminal rows; supports
  `--symbol`, `--with-account/--no-account`, and equivalent flags

All public modules maintain 100 % branch coverage; 968 tests pass.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* chore: apply mdformat to docs after grafana observability additions

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* chore: bump version to 1.1.0

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: address PR #86 review feedback

- Fix unaggregated time in grafana_realized_pnl GROUP BY query (MAX)
- Replace O(N) per-symbol API calls with single call + client-side filter
- Eliminate double create_snapshot_tables when with_grafana_schema=True
- Move grafana imports to module level in sdk.py (remove PLC0415 noqa)
- Default with_grafana_schema to False (run grafana-schema once for setup)
- Fix README position example to use snapshot_runs for latest snapshot
- Update tests to reflect new behavior and correct patch targets

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* chore: apply ruff formatting and sync lock file

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: address owner review feedback on PR #86

- Filter grafana_*_snapshots views to only expose rows from successful
  runs (JOIN snapshot_runs WHERE status='ok'), closing the partial-snapshot
  visibility gap raised in PRRT_kwDORzI_286MvDJ6
- Add issubset column guards for snapshot table index creation, consistent
  with the rest of create_grafana_indexes (PRRT_kwDORzI_286MvUx1)
- Fix README example queries: views expose 'time' not 'observed_at'
  (PRRT_kwDORzI_286MvUxw)
- Correct public-contract.md default for with_grafana_schema (False, not
  True) and point to grafana-schema command (PRRT_kwDORzI_286MvUxy)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: add run_id to snapshot schema and fix stale-positions README query

Replace second-level observed_at as the join key between snapshot_runs
and snapshot tables with a stable run_id INTEGER PRIMARY KEY. Two runs
in the same second now get distinct run_ids, preventing view duplication
and cross-contamination from a failed run. Update README example to use
snapshot_runs for latest-snapshot lookup so zero-position runs return an
empty result instead of stale rows.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* chore: format markdown tables

Align table column widths in README and public-contract documentation.

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>

* fix: close sqlite connections deterministically

* fix: require entry filter in grafana_realized_pnl and add time to grafana_trade_stats

grafana_realized_pnl now requires the entry column and filters to
close-side deals (entry IN (1, 2, 3)), consistent with grafana_symbol_pnl
and grafana_trade_stats. grafana_trade_stats now requires the time column
and emits MAX(time_expr) AS "time" so it satisfies the documented Grafana
view contract (integer epoch-second time column throughout).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: normalize pd.Timestamp time_setup to epoch int in insert_order_snapshots

orders_get_as_df() returns datetime-converted columns by default, so
time_setup is a pd.Timestamp in normal use. Passing it directly to
sqlite3.executemany raises ProgrammingError. Added _to_epoch_int helper
that converts datetime.datetime subclasses (including pd.Timestamp) and
raw int/float values to integer epoch seconds, returning None for other
types. Regression tests cover all four input paths.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: pre-drop all grafana_* views at start of create_grafana_views

Previously, a builder that skipped due to a missing source table or column
would not drop the view it owned, leaving stale views referencing gone
tables. Now create_grafana_views drops all 13 known grafana_* views before
calling any builder, so a schema refresh always removes views whose source
has disappeared. Regression test covers the create → drop-source → refresh
cycle.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: expose run_id in snapshot views and drop time from summary views

Grafana snapshot views now expose run_id so latest-state queries can use
MAX(run_id) instead of the ambiguous second-level MAX(observed_at).
grafana_realized_pnl and grafana_trade_stats lose their MAX(time) column
and are reclassified as static summary views; their all-time aggregates
are not filterable by Grafana time-range selectors.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: guard snapshot views against missing run_id and fix README view docs

_build_snapshot_view now skips with a warning when the underlying
snapshot table exists but lacks a run_id column, preventing a broken
view that fails at query time. Adds a regression test for that path.

README Grafana section now qualifies that grafana_realized_pnl and
grafana_trade_stats are static summary views (no time column) and splits
the view table to match docs/api/public-contract.md.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

---------

Co-authored-by: agent <agent@localhost>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Daichi Narushima
2026-06-28 07:55:26 +09:00
committed by GitHub
parent 4bc36d09d2
commit d27da02f3f
14 changed files with 2494 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/conftest.py
+38
View File
@@ -2,12 +2,17 @@
from __future__ import annotations
import sqlite3
from typing import TYPE_CHECKING, Any, Literal
from unittest.mock import MagicMock
import pandas as pd
import pytest
from pytest_mock import MockerFixture # noqa: TC002
if TYPE_CHECKING:
from types import TracebackType
_DATAFRAME_METHODS = (
"copy_rates_from_as_df",
"copy_rates_from_pos_as_df",
@@ -30,6 +35,25 @@ _DATAFRAME_METHODS = (
"order_send_as_df",
)
_ORIGINAL_SQLITE_CONNECT = sqlite3.connect
class ClosingSqliteConnection(sqlite3.Connection):
"""SQLite connection that closes after context-manager exit in tests."""
def __exit__(
self,
exc_type: type[BaseException] | None,
exc_value: BaseException | None,
traceback: TracebackType | None,
) -> Literal[False]:
"""Commit or roll back the transaction, then close the connection."""
try:
super().__exit__(exc_type, exc_value, traceback)
finally:
self.close()
return False
def build_mock_mt5_data_client() -> MagicMock:
"""Return a MagicMock Mt5DataClient with common DataFrame stubs."""
@@ -50,3 +74,17 @@ def mock_client(mocker: MockerFixture) -> MagicMock:
client = build_mock_mt5_data_client()
mocker.patch("mt5cli.sdk.Mt5DataClient", return_value=client)
return client
@pytest.fixture(autouse=True)
def close_sqlite_context_connections(monkeypatch: pytest.MonkeyPatch) -> None:
"""Make test SQLite context managers close their connection handles."""
def connect(
*args: Any, # noqa: ANN401
**kwargs: Any, # noqa: ANN401
) -> sqlite3.Connection:
kwargs.setdefault("factory", ClosingSqliteConnection)
return _ORIGINAL_SQLITE_CONNECT(*args, **kwargs)
monkeypatch.setattr(sqlite3, "connect", connect)
tests/test_cli.py
+168
View File
@@ -1855,6 +1855,174 @@ class TestCollectHistory:
)
class TestGrafanaSchemaCommand:
"""Tests for the grafana-schema CLI command."""
def test_grafana_schema_creates_snapshot_tables_in_sqlite(
self,
tmp_path: Path,
) -> None:
"""grafana-schema applies Grafana schema to a SQLite database."""
output = tmp_path / "out.db"
result = runner.invoke(app, ["-o", str(output), "grafana-schema"])
assert result.exit_code == 0, result.output
with sqlite3.connect(output) as conn:
tables = {
row[0]
for row in conn.execute(
"SELECT name FROM sqlite_master WHERE type='table'"
).fetchall()
}
assert "snapshot_runs" in tables
assert "account_snapshots" in tables
def test_grafana_schema_is_idempotent(self, tmp_path: Path) -> None:
"""grafana-schema can be invoked multiple times without error."""
output = tmp_path / "out.db"
result1 = runner.invoke(app, ["-o", str(output), "grafana-schema"])
result2 = runner.invoke(app, ["-o", str(output), "grafana-schema"])
assert result1.exit_code == 0, result1.output
assert result2.exit_code == 0, result2.output
def test_grafana_schema_rejects_non_sqlite_output(
self,
tmp_path: Path,
) -> None:
"""grafana-schema fails when output is not a SQLite3 format."""
result = runner.invoke(
app,
["-o", str(tmp_path / "out.csv"), "grafana-schema"],
)
assert result.exit_code != 0
assert "grafana-schema requires SQLite3 output" in result.output
class TestSnapshotCommand:
"""Tests for the snapshot CLI command."""
def test_snapshot_rejects_non_sqlite_output(self, tmp_path: Path) -> None:
"""Snapshot fails when output is not a SQLite3 format."""
result = runner.invoke(
app,
["-o", str(tmp_path / "out.csv"), "snapshot"],
)
assert result.exit_code != 0
assert "snapshot requires SQLite3 output" in result.output
def test_snapshot_delegates_to_update_observability_with_config(
self,
tmp_path: Path,
mocker: MockerFixture,
) -> None:
"""Snapshot calls sdk.update_observability_with_config."""
updater = mocker.patch("mt5cli.cli.sdk.update_observability_with_config")
output = tmp_path / "out.db"
result = runner.invoke(app, ["-o", str(output), "snapshot"])
assert result.exit_code == 0, result.output
updater.assert_called_once()
kwargs = updater.call_args.kwargs
assert kwargs["output"] == output
assert kwargs["symbols"] is None
assert kwargs["include_account"] is True
assert kwargs["include_positions"] is True
assert kwargs["include_orders"] is True
assert kwargs["include_terminal"] is True
assert kwargs["with_grafana_schema"] is False
def test_snapshot_with_symbol_filter(
self,
tmp_path: Path,
mocker: MockerFixture,
) -> None:
"""Snapshot passes symbol list to update_observability_with_config."""
updater = mocker.patch("mt5cli.cli.sdk.update_observability_with_config")
result = runner.invoke(
app,
[
"-o",
str(tmp_path / "out.db"),
"snapshot",
"--symbol",
"EURUSD",
"--symbol",
"GBPUSD",
],
)
assert result.exit_code == 0, result.output
kwargs = updater.call_args.kwargs
assert kwargs["symbols"] == ["EURUSD", "GBPUSD"]
def test_snapshot_with_no_account_flag(
self,
tmp_path: Path,
mocker: MockerFixture,
) -> None:
"""--no-account disables account snapshotting."""
updater = mocker.patch("mt5cli.cli.sdk.update_observability_with_config")
result = runner.invoke(
app,
["-o", str(tmp_path / "out.db"), "snapshot", "--no-account"],
)
assert result.exit_code == 0, result.output
assert updater.call_args.kwargs["include_account"] is False
def test_snapshot_with_no_positions_flag(
self,
tmp_path: Path,
mocker: MockerFixture,
) -> None:
"""--no-positions disables position snapshotting."""
updater = mocker.patch("mt5cli.cli.sdk.update_observability_with_config")
result = runner.invoke(
app,
["-o", str(tmp_path / "out.db"), "snapshot", "--no-positions"],
)
assert result.exit_code == 0, result.output
assert updater.call_args.kwargs["include_positions"] is False
def test_snapshot_with_no_orders_flag(
self,
tmp_path: Path,
mocker: MockerFixture,
) -> None:
"""--no-orders disables order snapshotting."""
updater = mocker.patch("mt5cli.cli.sdk.update_observability_with_config")
result = runner.invoke(
app,
["-o", str(tmp_path / "out.db"), "snapshot", "--no-orders"],
)
assert result.exit_code == 0, result.output
assert updater.call_args.kwargs["include_orders"] is False
def test_snapshot_with_no_terminal_flag(
self,
tmp_path: Path,
mocker: MockerFixture,
) -> None:
"""--no-terminal disables terminal snapshotting."""
updater = mocker.patch("mt5cli.cli.sdk.update_observability_with_config")
result = runner.invoke(
app,
["-o", str(tmp_path / "out.db"), "snapshot", "--no-terminal"],
)
assert result.exit_code == 0, result.output
assert updater.call_args.kwargs["include_terminal"] is False
def test_snapshot_with_no_grafana_schema_flag(
self,
tmp_path: Path,
mocker: MockerFixture,
) -> None:
"""--no-grafana-schema disables Grafana schema creation."""
updater = mocker.patch("mt5cli.cli.sdk.update_observability_with_config")
result = runner.invoke(
app,
["-o", str(tmp_path / "out.db"), "snapshot", "--no-grafana-schema"],
)
assert result.exit_code == 0, result.output
assert updater.call_args.kwargs["with_grafana_schema"] is False
class TestMain:
"""Tests for the main entry point."""
tests/test_grafana.py
+838
View File
@@ -0,0 +1,838 @@
"""Tests for mt5cli.grafana module."""
from __future__ import annotations
import logging
import sqlite3
from typing import TYPE_CHECKING
from unittest.mock import MagicMock
import pandas as pd
import pytest
if TYPE_CHECKING:
from collections.abc import Iterator
from mt5cli.grafana import (
_build_snapshot_view, # type: ignore[reportPrivateUsage]
_create_view_safe, # type: ignore[reportPrivateUsage]
create_grafana_indexes,
create_grafana_views,
create_snapshot_tables,
ensure_grafana_schema,
insert_account_snapshot,
insert_order_snapshots,
insert_position_snapshots,
insert_terminal_snapshot,
record_snapshot_run,
start_snapshot_run,
)
@pytest.fixture
def conn() -> Iterator[sqlite3.Connection]:
"""Yield an in-memory SQLite connection for each test."""
with sqlite3.connect(":memory:") as c:
yield c
def _get_names(conn: sqlite3.Connection, type_: str) -> set[str]:
return {
row[0]
for row in conn.execute(
"SELECT name FROM sqlite_master WHERE type=?",
(type_,),
).fetchall()
}
def _make_rates_table(conn: sqlite3.Connection) -> None:
conn.execute(
"CREATE TABLE rates"
" (time TEXT, symbol TEXT, timeframe INTEGER,"
" open REAL, high REAL, low REAL, close REAL)"
)
def _make_ticks_table(conn: sqlite3.Connection) -> None:
conn.execute("CREATE TABLE ticks (time TEXT, symbol TEXT, bid REAL, ask REAL)")
def _make_history_deals_full(conn: sqlite3.Connection) -> None:
conn.execute(
"CREATE TABLE history_deals"
" (time TEXT, symbol TEXT, profit REAL, type INTEGER,"
" entry INTEGER, volume REAL, price REAL, ticket INTEGER, position_id INTEGER)"
)
def _make_history_deals_minimal(conn: sqlite3.Connection) -> None:
"""history_deals with only time, type, symbol, profit — no entry/volume/price."""
conn.execute(
"CREATE TABLE history_deals (time TEXT, symbol TEXT, profit REAL, type INTEGER)"
)
def _make_history_orders_table(conn: sqlite3.Connection) -> None:
conn.execute(
"CREATE TABLE history_orders"
" (time_setup TEXT, symbol TEXT, ticket INTEGER, type INTEGER)"
)
# ---------------------------------------------------------------------------
# TestSnapshotTables
# ---------------------------------------------------------------------------
class TestSnapshotTables:
"""Tests for create_snapshot_tables."""
def test_creates_all_five_tables(self, conn: sqlite3.Connection) -> None:
"""All five snapshot tables are created."""
create_snapshot_tables(conn)
tables = _get_names(conn, "table")
assert "snapshot_runs" in tables
assert "account_snapshots" in tables
assert "position_snapshots" in tables
assert "order_snapshots" in tables
assert "terminal_snapshots" in tables
def test_is_idempotent(self, conn: sqlite3.Connection) -> None:
"""Calling create_snapshot_tables twice does not raise."""
create_snapshot_tables(conn)
create_snapshot_tables(conn)
tables = _get_names(conn, "table")
assert "snapshot_runs" in tables
# ---------------------------------------------------------------------------
# TestCreateViewSafe
# ---------------------------------------------------------------------------
class TestCreateViewSafe:
"""Tests for _create_view_safe."""
def test_creates_view_successfully(self, conn: sqlite3.Connection) -> None:
"""A valid select SQL creates the named view."""
_create_view_safe(conn, "test_view", "SELECT 1 AS val")
views = _get_names(conn, "view")
assert "test_view" in views
def test_replaces_existing_view(self, conn: sqlite3.Connection) -> None:
"""Calling again with a new SQL replaces the existing view."""
_create_view_safe(conn, "test_view", "SELECT 1 AS val")
_create_view_safe(conn, "test_view", "SELECT 2 AS val")
result = conn.execute("SELECT val FROM test_view").fetchone()
assert result == (2,)
def test_logs_warning_on_sqlite_error(
self,
caplog: pytest.LogCaptureFixture,
) -> None:
"""sqlite3.Error during CREATE VIEW logs a warning instead of raising."""
mock_conn = MagicMock()
mock_conn.execute.side_effect = [
None,
sqlite3.OperationalError("parse error"),
]
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
_create_view_safe(mock_conn, "bad_view", "SELECT 1")
assert "Skipping view bad_view" in caplog.text
assert "parse error" in caplog.text
# ---------------------------------------------------------------------------
# TestGrafanaViews
# ---------------------------------------------------------------------------
class TestGrafanaViews:
"""Tests for create_grafana_views and individual view builders."""
def test_all_views_created_with_full_schema(
self,
conn: sqlite3.Connection,
) -> None:
"""All 13 Grafana views are created when all source tables are present."""
_make_rates_table(conn)
_make_ticks_table(conn)
_make_history_deals_full(conn)
_make_history_orders_table(conn)
create_snapshot_tables(conn)
create_grafana_views(conn)
views = _get_names(conn, "view")
expected = {
"grafana_rates",
"grafana_ticks",
"grafana_history_deals",
"grafana_history_orders",
"grafana_trade_deals",
"grafana_cash_events",
"grafana_realized_pnl",
"grafana_symbol_pnl",
"grafana_trade_stats",
"grafana_account_snapshots",
"grafana_position_snapshots",
"grafana_order_snapshots",
"grafana_terminal_snapshots",
}
assert expected.issubset(views)
def test_stale_view_dropped_when_source_table_disappears(
self,
conn: sqlite3.Connection,
) -> None:
"""create_grafana_views drops a previously created view whose source is gone."""
_make_ticks_table(conn)
create_grafana_views(conn)
assert "grafana_ticks" in _get_names(conn, "view")
conn.execute("DROP TABLE ticks")
create_grafana_views(conn)
assert "grafana_ticks" not in _get_names(conn, "view")
def test_grafana_rates_skipped_when_table_absent(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
) -> None:
"""grafana_rates is skipped when rates table is missing."""
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
assert "grafana_rates" not in _get_names(conn, "view")
def test_grafana_rates_skipped_when_required_cols_missing(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
) -> None:
"""grafana_rates is skipped when rates table lacks required columns."""
conn.execute("CREATE TABLE rates (open REAL)")
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
assert "grafana_rates" not in _get_names(conn, "view")
assert "Skipping grafana_rates" in caplog.text
def test_grafana_ticks_skipped_when_cols_missing(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
) -> None:
"""grafana_ticks is skipped when ticks table lacks required columns."""
conn.execute("CREATE TABLE ticks (bid REAL)")
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
assert "grafana_ticks" not in _get_names(conn, "view")
assert "Skipping grafana_ticks" in caplog.text
def test_grafana_history_deals_skipped_when_time_missing(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
) -> None:
"""grafana_history_deals is skipped when history_deals.time is missing."""
conn.execute("CREATE TABLE history_deals (symbol TEXT)")
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
assert "grafana_history_deals" not in _get_names(conn, "view")
assert "Skipping grafana_history_deals" in caplog.text
def test_grafana_history_orders_skipped_when_time_setup_missing(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
) -> None:
"""grafana_history_orders is skipped when time_setup is absent."""
conn.execute("CREATE TABLE history_orders (symbol TEXT)")
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
assert "grafana_history_orders" not in _get_names(conn, "view")
assert "Skipping grafana_history_orders" in caplog.text
def test_grafana_trade_deals_skipped_when_cols_missing(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
) -> None:
"""grafana_trade_deals is skipped when history_deals missing time/type."""
conn.execute("CREATE TABLE history_deals (symbol TEXT)")
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
assert "grafana_trade_deals" not in _get_names(conn, "view")
def test_grafana_cash_events_skipped_when_cols_missing(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
) -> None:
"""grafana_cash_events is skipped when history_deals missing time/type."""
conn.execute("CREATE TABLE history_deals (symbol TEXT)")
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
assert "grafana_cash_events" not in _get_names(conn, "view")
def test_grafana_realized_pnl_skipped_when_cols_missing(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
) -> None:
"""grafana_realized_pnl is skipped when history_deals missing required cols."""
conn.execute("CREATE TABLE history_deals (time TEXT, type INTEGER)")
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
assert "grafana_realized_pnl" not in _get_names(conn, "view")
assert "Skipping grafana_realized_pnl" in caplog.text
def test_grafana_realized_pnl_skipped_when_entry_missing(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
) -> None:
"""grafana_realized_pnl is skipped when entry column is absent."""
_make_history_deals_minimal(conn)
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
assert "grafana_realized_pnl" not in _get_names(conn, "view")
assert "Skipping grafana_realized_pnl" in caplog.text
def test_grafana_symbol_pnl_skipped_when_required_cols_missing(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
) -> None:
"""grafana_symbol_pnl is skipped when required columns are absent."""
conn.execute("CREATE TABLE history_deals (time TEXT, type INTEGER)")
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
assert "grafana_symbol_pnl" not in _get_names(conn, "view")
assert "Skipping grafana_symbol_pnl" in caplog.text
def test_grafana_symbol_pnl_without_volume_and_price(
self,
conn: sqlite3.Connection,
) -> None:
"""grafana_symbol_pnl is created with only required columns."""
conn.execute(
"CREATE TABLE history_deals"
" (time TEXT, symbol TEXT, profit REAL, type INTEGER, entry INTEGER)"
)
create_grafana_views(conn)
assert "grafana_symbol_pnl" in _get_names(conn, "view")
def test_grafana_symbol_pnl_with_volume_and_price(
self,
conn: sqlite3.Connection,
) -> None:
"""grafana_symbol_pnl includes volume and price columns when present."""
_make_history_deals_full(conn)
create_grafana_views(conn)
assert "grafana_symbol_pnl" in _get_names(conn, "view")
# View columns include volume and price
cols = {row[1] for row in conn.execute("PRAGMA table_info(grafana_symbol_pnl)")}
assert "volume" in cols
assert "price" in cols
def test_grafana_trade_stats_skipped_when_cols_missing(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
) -> None:
"""grafana_trade_stats is skipped when history_deals missing required cols."""
conn.execute("CREATE TABLE history_deals (time TEXT)")
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
assert "grafana_trade_stats" not in _get_names(conn, "view")
assert "Skipping grafana_trade_stats" in caplog.text
def test_grafana_trade_stats_without_entry_col(
self,
conn: sqlite3.Connection,
) -> None:
"""grafana_trade_stats is a static summary view with no time column."""
_make_history_deals_minimal(conn)
create_grafana_views(conn)
assert "grafana_trade_stats" in _get_names(conn, "view")
cols = {
row[1] for row in conn.execute("PRAGMA table_info(grafana_trade_stats)")
}
assert "time" not in cols
assert "symbol" in cols
def test_grafana_trade_stats_with_entry_col(
self,
conn: sqlite3.Connection,
) -> None:
"""grafana_trade_stats is a static summary view with no time column."""
_make_history_deals_full(conn)
create_grafana_views(conn)
assert "grafana_trade_stats" in _get_names(conn, "view")
cols = {
row[1] for row in conn.execute("PRAGMA table_info(grafana_trade_stats)")
}
assert "time" not in cols
assert "symbol" in cols
def test_snapshot_views_skipped_when_snapshot_tables_absent(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Snapshot views are skipped when snapshot tables are not created."""
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
views = _get_names(conn, "view")
assert "grafana_account_snapshots" not in views
assert "grafana_position_snapshots" not in views
assert "grafana_order_snapshots" not in views
assert "grafana_terminal_snapshots" not in views
def test_build_snapshot_view_with_only_run_id_col(
self,
conn: sqlite3.Connection,
) -> None:
"""_build_snapshot_view exposes time and run_id when table has only run_id."""
create_snapshot_tables(conn)
conn.execute("CREATE TABLE only_run (run_id INTEGER NOT NULL)")
run_id = start_snapshot_run(conn, 1000)
record_snapshot_run(conn, run_id, "ok")
conn.execute("INSERT INTO only_run (run_id) VALUES (?)", (run_id,))
_build_snapshot_view(conn, "test_view", "only_run")
assert "test_view" in _get_names(conn, "view")
cols = {row[1] for row in conn.execute("PRAGMA table_info(test_view)")}
assert "time" in cols
assert "run_id" in cols
def test_build_snapshot_view_skips_when_snapshot_runs_missing(
self,
conn: sqlite3.Connection,
) -> None:
"""_build_snapshot_view skips view when snapshot_runs has wrong columns."""
conn.execute("CREATE TABLE only_run (run_id INTEGER NOT NULL)")
conn.execute("CREATE TABLE snapshot_runs (foo TEXT)")
_build_snapshot_view(conn, "test_view", "only_run")
views = _get_names(conn, "view")
assert "test_view" not in views
def test_build_snapshot_view_skips_when_run_id_col_missing(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
) -> None:
"""_build_snapshot_view skips view when the table lacks run_id."""
create_snapshot_tables(conn)
conn.execute("CREATE TABLE no_run_id (symbol TEXT)")
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
_build_snapshot_view(conn, "test_view", "no_run_id")
assert "test_view" not in _get_names(conn, "view")
assert "missing run_id column" in caplog.text
def test_snapshot_view_excludes_failed_run_rows(
self,
conn: sqlite3.Connection,
) -> None:
"""Snapshot views hide rows from failed runs."""
create_snapshot_tables(conn)
run_id = start_snapshot_run(conn, 1000)
conn.execute(
"INSERT INTO account_snapshots"
" (run_id, login, balance, equity, margin, margin_free, profit)"
" VALUES (?, 12345, 10000.0, 9800.0, 200.0, 9600.0, -200.0)",
(run_id,),
)
record_snapshot_run(conn, run_id, "error", "terminal offline")
create_grafana_views(conn)
rows = conn.execute("SELECT * FROM grafana_account_snapshots").fetchall()
assert rows == []
def test_snapshot_view_includes_ok_run_rows(
self,
conn: sqlite3.Connection,
) -> None:
"""Snapshot views show rows from successful runs and expose run_id."""
create_snapshot_tables(conn)
run_id = start_snapshot_run(conn, 2000)
conn.execute(
"INSERT INTO account_snapshots"
" (run_id, login, balance, equity, margin, margin_free, profit)"
" VALUES (?, 12345, 10000.0, 9800.0, 200.0, 9600.0, -200.0)",
(run_id,),
)
record_snapshot_run(conn, run_id, "ok")
create_grafana_views(conn)
rows = conn.execute(
"SELECT time, run_id, login FROM grafana_account_snapshots"
).fetchall()
assert rows == [(2000, run_id, 12345)]
cols = {
row[1]
for row in conn.execute("PRAGMA table_info(grafana_account_snapshots)")
}
assert "run_id" in cols
def test_snapshot_view_same_second_ok_and_error_no_cross_contamination(
self,
conn: sqlite3.Connection,
) -> None:
"""An ok and error run sharing observed_at expose only the ok run's rows."""
create_snapshot_tables(conn)
run_err = start_snapshot_run(conn, 3000)
conn.execute(
"INSERT INTO account_snapshots (run_id, login) VALUES (?, 99)",
(run_err,),
)
record_snapshot_run(conn, run_err, "error")
run_ok = start_snapshot_run(conn, 3000)
conn.execute(
"INSERT INTO account_snapshots (run_id, login) VALUES (?, 12345)",
(run_ok,),
)
record_snapshot_run(conn, run_ok, "ok")
create_grafana_views(conn)
rows = conn.execute("SELECT login FROM grafana_account_snapshots").fetchall()
assert rows == [(12345,)]
def test_snapshot_view_two_ok_runs_same_second_no_duplication(
self,
conn: sqlite3.Connection,
) -> None:
"""Two ok runs sharing observed_at each produce exactly one row in the view."""
create_snapshot_tables(conn)
run1 = start_snapshot_run(conn, 4000)
conn.execute(
"INSERT INTO account_snapshots (run_id, login) VALUES (?, 1)",
(run1,),
)
record_snapshot_run(conn, run1, "ok")
run2 = start_snapshot_run(conn, 4000)
conn.execute(
"INSERT INTO account_snapshots (run_id, login) VALUES (?, 2)",
(run2,),
)
record_snapshot_run(conn, run2, "ok")
create_grafana_views(conn)
rows = conn.execute("SELECT login FROM grafana_account_snapshots").fetchall()
assert len(rows) == 2
# ---------------------------------------------------------------------------
# TestGrafanaIndexes
# ---------------------------------------------------------------------------
class TestGrafanaIndexes:
"""Tests for create_grafana_indexes."""
def test_all_indexes_created_with_full_schema(
self,
conn: sqlite3.Connection,
) -> None:
"""All 9 indexes are created when all source tables are present."""
_make_rates_table(conn)
_make_ticks_table(conn)
_make_history_deals_full(conn)
_make_history_orders_table(conn)
create_snapshot_tables(conn)
create_grafana_indexes(conn)
indexes = _get_names(conn, "index")
assert "idx_rates_time_symbol_timeframe" in indexes
assert "idx_ticks_time_symbol" in indexes
assert "idx_history_deals_time_symbol" in indexes
assert "idx_history_deals_symbol_time" in indexes
assert "idx_history_orders_time_setup_symbol" in indexes
assert "idx_account_snapshots_time_login" in indexes
assert "idx_position_snapshots_time_symbol" in indexes
assert "idx_order_snapshots_time_symbol" in indexes
assert "idx_snapshot_runs_time_status" in indexes
def test_no_indexes_created_when_tables_absent(
self,
conn: sqlite3.Connection,
) -> None:
"""No indexes are created when tables are absent."""
create_grafana_indexes(conn)
indexes = _get_names(conn, "index")
assert not any(name.startswith("idx_") for name in indexes)
def test_indexes_for_snapshot_tables_skipped_when_absent(
self,
conn: sqlite3.Connection,
) -> None:
"""Snapshot table indexes are skipped when snapshot tables don't exist."""
_make_history_deals_full(conn)
create_grafana_indexes(conn)
indexes = _get_names(conn, "index")
assert "idx_account_snapshots_time_login" not in indexes
assert "idx_position_snapshots_time_symbol" not in indexes
assert "idx_order_snapshots_time_symbol" not in indexes
assert "idx_snapshot_runs_time_status" not in indexes
def test_rates_index_skipped_when_cols_missing(
self,
conn: sqlite3.Connection,
) -> None:
"""Rates index is skipped when required columns are absent."""
conn.execute("CREATE TABLE rates (open REAL)")
create_grafana_indexes(conn)
indexes = _get_names(conn, "index")
assert "idx_rates_time_symbol_timeframe" not in indexes
def test_ticks_index_skipped_when_cols_missing(
self,
conn: sqlite3.Connection,
) -> None:
"""Ticks index is skipped when required columns are absent."""
conn.execute("CREATE TABLE ticks (bid REAL)")
create_grafana_indexes(conn)
indexes = _get_names(conn, "index")
assert "idx_ticks_time_symbol" not in indexes
def test_deals_indexes_skipped_when_cols_missing(
self,
conn: sqlite3.Connection,
) -> None:
"""history_deals indexes are skipped when required columns are absent."""
conn.execute("CREATE TABLE history_deals (ticket INTEGER)")
create_grafana_indexes(conn)
indexes = _get_names(conn, "index")
assert "idx_history_deals_time_symbol" not in indexes
def test_orders_index_skipped_when_cols_missing(
self,
conn: sqlite3.Connection,
) -> None:
"""history_orders index is skipped when required columns are absent."""
conn.execute("CREATE TABLE history_orders (ticket INTEGER)")
create_grafana_indexes(conn)
indexes = _get_names(conn, "index")
assert "idx_history_orders_time_setup_symbol" not in indexes
def test_snapshot_indexes_skipped_when_cols_missing(
self,
conn: sqlite3.Connection,
) -> None:
"""Snapshot table indexes are skipped when required columns are absent."""
conn.execute("CREATE TABLE account_snapshots (foo TEXT)")
conn.execute("CREATE TABLE position_snapshots (foo TEXT)")
conn.execute("CREATE TABLE order_snapshots (foo TEXT)")
conn.execute("CREATE TABLE snapshot_runs (foo TEXT)")
create_grafana_indexes(conn)
indexes = _get_names(conn, "index")
assert "idx_account_snapshots_time_login" not in indexes
assert "idx_position_snapshots_time_symbol" not in indexes
assert "idx_order_snapshots_time_symbol" not in indexes
assert "idx_snapshot_runs_time_status" not in indexes
def test_indexes_are_idempotent(self, conn: sqlite3.Connection) -> None:
"""Creating indexes twice does not raise (IF NOT EXISTS)."""
_make_rates_table(conn)
create_grafana_indexes(conn)
create_grafana_indexes(conn)
indexes = _get_names(conn, "index")
assert "idx_rates_time_symbol_timeframe" in indexes
# ---------------------------------------------------------------------------
# TestEnsureGrafanaSchema
# ---------------------------------------------------------------------------
class TestEnsureGrafanaSchema:
"""Tests for ensure_grafana_schema."""
def test_creates_all_tables_views_and_indexes(
self,
conn: sqlite3.Connection,
) -> None:
"""ensure_grafana_schema creates snapshot tables, views, and indexes."""
_make_rates_table(conn)
_make_history_deals_full(conn)
ensure_grafana_schema(conn)
tables = _get_names(conn, "table")
assert "snapshot_runs" in tables
assert "account_snapshots" in tables
views = _get_names(conn, "view")
assert "grafana_rates" in views
assert "grafana_account_snapshots" in views
indexes = _get_names(conn, "index")
assert "idx_rates_time_symbol_timeframe" in indexes
def test_is_idempotent(self, conn: sqlite3.Connection) -> None:
"""Calling ensure_grafana_schema twice does not raise."""
ensure_grafana_schema(conn)
ensure_grafana_schema(conn)
# ---------------------------------------------------------------------------
# TestSnapshotInserts
# ---------------------------------------------------------------------------
class TestSnapshotInserts:
"""Tests for snapshot insert helpers."""
@pytest.fixture(autouse=True)
def setup_tables(self, conn: sqlite3.Connection) -> None:
"""Create snapshot tables before each insert test."""
create_snapshot_tables(conn)
def test_insert_account_snapshot(self, conn: sqlite3.Connection) -> None:
"""insert_account_snapshot appends a row with correct values."""
run_id = start_snapshot_run(conn, 1700000000)
row: dict[str, object] = {
"login": 12345,
"currency": "USD",
"balance": 10000.0,
"equity": 9800.0,
"margin": 200.0,
"margin_free": 9800.0,
"margin_level": 4900.0,
"profit": -200.0,
"leverage": 100,
}
insert_account_snapshot(conn, run_id, row)
result = conn.execute(
"SELECT login, currency, balance FROM account_snapshots"
).fetchone()
assert result == (12345, "USD", 10000.0)
def test_insert_account_snapshot_partial_row(
self,
conn: sqlite3.Connection,
) -> None:
"""insert_account_snapshot works when some fields are missing (uses None)."""
run_id = start_snapshot_run(conn, 1700000000)
insert_account_snapshot(conn, run_id, {"login": 1})
result = conn.execute(
"SELECT login, currency FROM account_snapshots"
).fetchone()
assert result == (1, None)
def test_insert_position_snapshots_with_rows(
self,
conn: sqlite3.Connection,
) -> None:
"""insert_position_snapshots appends each position row."""
run_id = start_snapshot_run(conn, 1700000000)
rows: list[dict[str, object]] = [
{"ticket": 1, "symbol": "EURUSD", "volume": 0.1, "profit": 10.0},
{"ticket": 2, "symbol": "GBPUSD", "volume": 0.2, "profit": -5.0},
]
insert_position_snapshots(conn, run_id, 12345, rows)
count = conn.execute("SELECT COUNT(*) FROM position_snapshots").fetchone()[0]
assert count == 2
def test_insert_position_snapshots_noop_when_empty(
self,
conn: sqlite3.Connection,
) -> None:
"""insert_position_snapshots is a no-op when rows is empty."""
run_id = start_snapshot_run(conn, 1700000000)
insert_position_snapshots(conn, run_id, 12345, [])
count = conn.execute("SELECT COUNT(*) FROM position_snapshots").fetchone()[0]
assert count == 0
def test_insert_order_snapshots_with_rows(
self,
conn: sqlite3.Connection,
) -> None:
"""insert_order_snapshots appends each order row."""
run_id = start_snapshot_run(conn, 1700000000)
rows: list[dict[str, object]] = [
{"ticket": 10, "symbol": "EURUSD", "type": 2, "volume_current": 0.1},
]
insert_order_snapshots(conn, run_id, 12345, rows)
count = conn.execute("SELECT COUNT(*) FROM order_snapshots").fetchone()[0]
assert count == 1
def test_insert_order_snapshots_noop_when_empty(
self,
conn: sqlite3.Connection,
) -> None:
"""insert_order_snapshots is a no-op when rows is empty."""
run_id = start_snapshot_run(conn, 1700000000)
insert_order_snapshots(conn, run_id, 12345, [])
count = conn.execute("SELECT COUNT(*) FROM order_snapshots").fetchone()[0]
assert count == 0
def test_insert_order_snapshots_normalizes_timestamp_time_setup(
self,
conn: sqlite3.Connection,
) -> None:
"""insert_order_snapshots converts pd.Timestamp time_setup to epoch int."""
run_id = start_snapshot_run(conn, 1700000000)
ts = pd.Timestamp("2024-01-15 10:30:00", tz="UTC")
rows: list[dict[str, object]] = [{"ticket": 10, "time_setup": ts}]
insert_order_snapshots(conn, run_id, 12345, rows)
stored = conn.execute("SELECT time_setup FROM order_snapshots").fetchone()[0]
assert stored == int(ts.timestamp())
def test_insert_order_snapshots_stores_int_time_setup(
self,
conn: sqlite3.Connection,
) -> None:
"""insert_order_snapshots stores an integer time_setup as-is."""
run_id = start_snapshot_run(conn, 1700000000)
rows: list[dict[str, object]] = [{"ticket": 10, "time_setup": 1705314600}]
insert_order_snapshots(conn, run_id, 12345, rows)
stored = conn.execute("SELECT time_setup FROM order_snapshots").fetchone()[0]
assert stored == 1705314600
def test_insert_order_snapshots_stores_null_for_unknown_time_setup_type(
self,
conn: sqlite3.Connection,
) -> None:
"""insert_order_snapshots stores NULL for an unrecognized time_setup type."""
run_id = start_snapshot_run(conn, 1700000000)
rows: list[dict[str, object]] = [{"ticket": 10, "time_setup": "not_a_time"}]
insert_order_snapshots(conn, run_id, 12345, rows)
stored = conn.execute("SELECT time_setup FROM order_snapshots").fetchone()[0]
assert stored is None
def test_insert_terminal_snapshot(self, conn: sqlite3.Connection) -> None:
"""insert_terminal_snapshot appends a terminal info row."""
run_id = start_snapshot_run(conn, 1700000000)
row: dict[str, object] = {
"name": "MetaTrader 5",
"connected": 1,
"community_account": 0,
"trade_allowed": 1,
"trade_expert": 1,
"path": "/mt5",
"company": "Broker",
"language": "en",
}
insert_terminal_snapshot(conn, run_id, row)
result = conn.execute(
"SELECT name, connected FROM terminal_snapshots"
).fetchone()
assert result == ("MetaTrader 5", 1)
def test_start_snapshot_run_returns_incrementing_ids(
self,
conn: sqlite3.Connection,
) -> None:
"""start_snapshot_run returns a unique run_id for each call."""
run1 = start_snapshot_run(conn, 1700000000)
run2 = start_snapshot_run(conn, 1700000000)
assert run1 != run2
def test_record_snapshot_run_with_detail(
self,
conn: sqlite3.Connection,
) -> None:
"""record_snapshot_run stores status and detail text."""
run_id = start_snapshot_run(conn, 1700000000)
record_snapshot_run(conn, run_id, "error", "RuntimeError: boom")
row = conn.execute("SELECT status, detail FROM snapshot_runs").fetchone()
assert row == ("error", "RuntimeError: boom")
def test_record_snapshot_run_without_detail(
self,
conn: sqlite3.Connection,
) -> None:
"""record_snapshot_run stores None for detail when omitted."""
run_id = start_snapshot_run(conn, 1700000000)
record_snapshot_run(conn, run_id, "ok")
row = conn.execute("SELECT status, detail FROM snapshot_runs").fetchone()
assert row == ("ok", None)
tests/test_sdk.py
+374
View File
@@ -61,6 +61,8 @@ from mt5cli.sdk import (
terminal_info,
update_history,
update_history_with_config,
update_observability,
update_observability_with_config,
version,
)
from mt5cli.utils import Dataset, IfExists, coerce_login
@@ -2746,3 +2748,375 @@ class TestSubstituteMappingValues:
result = substitute_mapping_values(data, keys={"mt5_login"})
# tuple is returned as-is; inner dict is NOT visited
assert result == {"accounts": ({"mt5_login": "${MT5_LOGIN}"},)}
class TestUpdateObservability:
"""Tests for update_observability and update_observability_with_config."""
@pytest.fixture
def mock_client(self) -> MagicMock:
"""Mock client returning minimal valid frames."""
client = MagicMock()
client.account_info_as_df.return_value = pd.DataFrame([
{
"login": 12345,
"currency": "USD",
"balance": 10000.0,
"equity": 10000.0,
"margin": 0.0,
"margin_free": 10000.0,
"margin_level": 0.0,
"profit": 0.0,
"leverage": 100,
}
])
client.positions_get_as_df.return_value = pd.DataFrame()
client.orders_get_as_df.return_value = pd.DataFrame()
client.terminal_info_as_df.return_value = pd.DataFrame([
{
"name": "MetaTrader 5",
"connected": 1,
"community_account": 0,
"trade_allowed": 1,
"trade_expert": 1,
"path": "/mt5",
"company": "Broker",
"language": "en",
}
])
return client
def test_update_observability_creates_snapshot_tables(
self,
mock_client: MagicMock,
tmp_path: Path,
) -> None:
"""Snapshot tables are created in the output database."""
output = tmp_path / "obs.db"
update_observability(client=mock_client, output=output)
with sqlite3.connect(output) as conn:
tables = {
row[0]
for row in conn.execute(
"SELECT name FROM sqlite_master WHERE type='table'"
).fetchall()
}
assert "snapshot_runs" in tables
assert "account_snapshots" in tables
assert "position_snapshots" in tables
def test_update_observability_records_ok_on_success(
self,
mock_client: MagicMock,
tmp_path: Path,
) -> None:
"""snapshot_runs records 'ok' status on a successful run."""
output = tmp_path / "obs.db"
update_observability(client=mock_client, output=output)
with sqlite3.connect(output) as conn:
row = conn.execute("SELECT status FROM snapshot_runs").fetchone()
assert row == ("ok",)
def test_update_observability_records_error_on_failure(
self,
mock_client: MagicMock,
tmp_path: Path,
) -> None:
"""snapshot_runs records 'error' and re-raises when a snapshot fails."""
mock_client.account_info_as_df.side_effect = RuntimeError("boom")
output = tmp_path / "obs.db"
with pytest.raises(RuntimeError, match="boom"):
update_observability(client=mock_client, output=output)
with sqlite3.connect(output) as conn:
row = conn.execute("SELECT status FROM snapshot_runs").fetchone()
assert row == ("error",)
def test_update_observability_skips_ensure_grafana_schema_when_disabled(
self,
mock_client: MagicMock,
mocker: MockerFixture,
tmp_path: Path,
) -> None:
"""with_grafana_schema=False does not call ensure_grafana_schema."""
spy = mocker.spy(sdk, "ensure_grafana_schema")
update_observability(
client=mock_client,
output=tmp_path / "obs.db",
with_grafana_schema=False,
)
spy.assert_not_called()
def test_update_observability_calls_ensure_grafana_schema_by_default(
self,
mock_client: MagicMock,
mocker: MockerFixture,
tmp_path: Path,
) -> None:
"""with_grafana_schema=True calls ensure_grafana_schema."""
spy = mocker.spy(sdk, "ensure_grafana_schema")
update_observability(
client=mock_client,
output=tmp_path / "obs.db",
with_grafana_schema=True,
)
spy.assert_called_once()
def test_update_observability_skips_account_when_disabled(
self,
mock_client: MagicMock,
tmp_path: Path,
) -> None:
"""include_account=False does not call account_info_as_df."""
update_observability(
client=mock_client,
output=tmp_path / "obs.db",
include_account=False,
)
mock_client.account_info_as_df.assert_not_called()
def test_update_observability_skips_positions_when_disabled(
self,
mock_client: MagicMock,
tmp_path: Path,
) -> None:
"""include_positions=False does not call positions_get_as_df."""
update_observability(
client=mock_client,
output=tmp_path / "obs.db",
include_positions=False,
)
mock_client.positions_get_as_df.assert_not_called()
def test_update_observability_skips_orders_when_disabled(
self,
mock_client: MagicMock,
tmp_path: Path,
) -> None:
"""include_orders=False does not call orders_get_as_df."""
update_observability(
client=mock_client,
output=tmp_path / "obs.db",
include_orders=False,
)
mock_client.orders_get_as_df.assert_not_called()
def test_update_observability_skips_terminal_when_disabled(
self,
mock_client: MagicMock,
tmp_path: Path,
) -> None:
"""include_terminal=False does not call terminal_info_as_df."""
update_observability(
client=mock_client,
output=tmp_path / "obs.db",
include_terminal=False,
)
mock_client.terminal_info_as_df.assert_not_called()
def test_update_observability_with_positions_rows(
self,
mock_client: MagicMock,
tmp_path: Path,
) -> None:
"""Non-empty positions are written to position_snapshots."""
mock_client.positions_get_as_df.return_value = pd.DataFrame([
{
"ticket": 1,
"position_id": 1,
"symbol": "EURUSD",
"type": 0,
"volume": 0.1,
"price_open": 1.1,
"price_current": 1.1,
"profit": 0.0,
"swap": 0.0,
"comment": "",
"magic": 0,
}
])
output = tmp_path / "obs.db"
update_observability(client=mock_client, output=output)
with sqlite3.connect(output) as conn:
count = conn.execute("SELECT COUNT(*) FROM position_snapshots").fetchone()[
0
]
assert count == 1
def test_update_observability_with_order_rows(
self,
mock_client: MagicMock,
tmp_path: Path,
) -> None:
"""Non-empty orders are written to order_snapshots."""
mock_client.orders_get_as_df.return_value = pd.DataFrame([
{
"ticket": 10,
"symbol": "EURUSD",
"type": 2,
"volume_current": 0.1,
"price_open": 1.2,
"price_current": 1.1,
"state": 1,
"comment": "",
"magic": 0,
"time_setup": 1700000000,
}
])
output = tmp_path / "obs.db"
update_observability(client=mock_client, output=output)
with sqlite3.connect(output) as conn:
count = conn.execute("SELECT COUNT(*) FROM order_snapshots").fetchone()[0]
assert count == 1
def test_update_observability_symbol_filter_positions(
self,
tmp_path: Path,
) -> None:
"""Symbol filter fetches all positions in one call and filters client-side."""
client = MagicMock()
client.account_info_as_df.return_value = pd.DataFrame([{"login": 1}])
# All positions; only EURUSD matches the filter
client.positions_get_as_df.return_value = pd.DataFrame([
{"ticket": 1, "symbol": "EURUSD", "volume": 0.1, "profit": 0.0},
{"ticket": 2, "symbol": "USDJPY", "volume": 0.2, "profit": 0.0},
])
client.orders_get_as_df.return_value = pd.DataFrame()
client.terminal_info_as_df.return_value = pd.DataFrame()
output = tmp_path / "obs.db"
update_observability(client=client, output=output, symbols=["EURUSD", "GBPUSD"])
assert client.positions_get_as_df.call_count == 1
with sqlite3.connect(output) as conn:
count = conn.execute("SELECT COUNT(*) FROM position_snapshots").fetchone()[
0
]
assert count == 1
def test_update_observability_symbol_filter_orders(
self,
tmp_path: Path,
) -> None:
"""Symbol filter fetches all orders in one call and filters client-side."""
client = MagicMock()
client.account_info_as_df.return_value = pd.DataFrame([{"login": 1}])
client.positions_get_as_df.return_value = pd.DataFrame()
# All orders; only EURUSD matches the filter
client.orders_get_as_df.return_value = pd.DataFrame([
{"ticket": 10, "symbol": "EURUSD", "volume_current": 0.1},
{"ticket": 11, "symbol": "USDJPY", "volume_current": 0.5},
])
client.terminal_info_as_df.return_value = pd.DataFrame()
output = tmp_path / "obs.db"
update_observability(client=client, output=output, symbols=["EURUSD", "GBPUSD"])
assert client.orders_get_as_df.call_count == 1
with sqlite3.connect(output) as conn:
count = conn.execute("SELECT COUNT(*) FROM order_snapshots").fetchone()[0]
assert count == 1
def test_update_observability_symbol_filter_no_symbol_col(
self,
tmp_path: Path,
) -> None:
"""Symbol filter is skipped when positions df has no symbol column."""
client = MagicMock()
client.account_info_as_df.return_value = pd.DataFrame([{"login": 1}])
# No symbol column in positions — all rows pass through unfiltered
client.positions_get_as_df.return_value = pd.DataFrame([
{"ticket": 1, "volume": 0.1},
])
client.orders_get_as_df.return_value = pd.DataFrame()
client.terminal_info_as_df.return_value = pd.DataFrame()
output = tmp_path / "obs.db"
update_observability(client=client, output=output, symbols=["EURUSD"])
with sqlite3.connect(output) as conn:
count = conn.execute("SELECT COUNT(*) FROM position_snapshots").fetchone()[
0
]
assert count == 1
def test_update_observability_account_none_login(
self,
tmp_path: Path,
) -> None:
"""Account row with no login key returns None login for downstream helpers."""
client = MagicMock()
client.account_info_as_df.return_value = pd.DataFrame([{"balance": 10000.0}])
client.positions_get_as_df.return_value = pd.DataFrame()
client.orders_get_as_df.return_value = pd.DataFrame()
client.terminal_info_as_df.return_value = pd.DataFrame()
output = tmp_path / "obs.db"
update_observability(client=client, output=output)
with sqlite3.connect(output) as conn:
row = conn.execute("SELECT login FROM account_snapshots").fetchone()
assert row is not None
assert row[0] is None
def test_update_observability_empty_account_logs_warning(
self,
mock_client: MagicMock,
tmp_path: Path,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Empty account_info_as_df logs a warning and does not write account row."""
mock_client.account_info_as_df.return_value = pd.DataFrame()
with caplog.at_level(logging.WARNING, logger="mt5cli.sdk"):
update_observability(client=mock_client, output=tmp_path / "obs.db")
assert "account_info_as_df returned empty frame" in caplog.text
with sqlite3.connect(tmp_path / "obs.db") as conn:
count = conn.execute("SELECT COUNT(*) FROM account_snapshots").fetchone()[0]
assert count == 0
def test_update_observability_empty_terminal_logs_warning(
self,
mock_client: MagicMock,
tmp_path: Path,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Empty terminal_info_as_df logs a warning and does not write terminal row."""
mock_client.terminal_info_as_df.return_value = pd.DataFrame()
with caplog.at_level(logging.WARNING, logger="mt5cli.sdk"):
update_observability(client=mock_client, output=tmp_path / "obs.db")
assert "terminal_info_as_df returned empty frame" in caplog.text
with sqlite3.connect(tmp_path / "obs.db") as conn:
count = conn.execute("SELECT COUNT(*) FROM terminal_snapshots").fetchone()[
0
]
assert count == 0
def test_update_observability_with_config_opens_and_closes_connection(
self,
mocker: MockerFixture,
tmp_path: Path,
) -> None:
"""update_observability_with_config manages the MT5 connection lifecycle."""
mock_client = MagicMock()
mock_client.account_info_as_df.return_value = pd.DataFrame()
mock_client.positions_get_as_df.return_value = pd.DataFrame()
mock_client.orders_get_as_df.return_value = pd.DataFrame()
mock_client.terminal_info_as_df.return_value = pd.DataFrame()
mocker.patch("mt5cli.sdk.Mt5DataClient", return_value=mock_client)
update_observability_with_config(output=tmp_path / "obs.db")
mock_client.initialize_and_login_mt5.assert_called_once()
mock_client.shutdown.assert_called_once()
def test_update_observability_with_config_passes_symbols(
self,
mocker: MockerFixture,
tmp_path: Path,
) -> None:
"""update_observability_with_config forwards symbols to update_observability."""
mock_client = MagicMock()
mock_client.account_info_as_df.return_value = pd.DataFrame()
mock_client.positions_get_as_df.return_value = pd.DataFrame()
mock_client.orders_get_as_df.return_value = pd.DataFrame()
mock_client.terminal_info_as_df.return_value = pd.DataFrame()
mocker.patch("mt5cli.sdk.Mt5DataClient", return_value=mock_client)
spy = mocker.patch("mt5cli.sdk.update_observability")
update_observability_with_config(
output=tmp_path / "obs.db",
symbols=["EURUSD"],
include_account=False,
)
spy.assert_called_once()
call_kwargs = spy.call_args.kwargs
assert call_kwargs["symbols"] == ["EURUSD"]
assert call_kwargs["include_account"] is False