Release v0.6.0

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## v0.6.0
+- Added one-row-per-account balance snapshots.
+- Snapshot reports now return the latest snapshots for one account as a plain array.
+- Snapshot valuation now uses `crypto.get_price` at capture time.
+- Removed the stale Bitstamp live-price cache from snapshot valuation.
+- Order completion now triggers snapshot capture through the shared status update path.
+
 ## v0.5.0
 - Removed the Bitstamp nonce/auth instability that caused recurring 403 failures.
 - Exchange communication is stable again.

README.md

@@ -1,14 +1,14 @@
-# exec-mcp v0.5.0
+# exec-mcp v0.6.0
 
 ## Release notes
 
-This release reflects the execution layer becoming stable and production-usable:
+This release adds real balance snapshot history and live crypto-priced account valuation:
 
-- nonce handling no longer produces the earlier 403 auth failures
-- exchange communication is stable again
-- real fee data is used for sizing on the trader side
-- order placement, query, and cancel paths are reliable
-- account and order recovery behaviour is more robust after reconnects and restarts
+- account balance snapshots are stored as one record per capture
+- snapshot reports return the latest snapshots for one account as a plain array
+- snapshot valuation uses `crypto.get_price` at capture time
+- stale Bitstamp live-price cache usage was removed from snapshot valuation
+- order completion now triggers snapshot capture through the shared status path
 
 Execution MCP for Trader27.
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "exec-mcp"
-version = "0.5.0"
+version = "0.6.0"
 description = "Execution MCP for Trader27"
 readme = "README.md"
 requires-python = ">=3.11"

src/exec_mcp/bitstamp_ws.py

@@ -1,92 +0,0 @@
-from __future__ import annotations
-
-import asyncio
-import json
-from datetime import datetime, timezone
-
-import websockets
-
-from . import repo
-
-WS_URL = "wss://ws.bitstamp.net"
-WS_RECONNECT_SECONDS = 5
-WS_HEARTBEAT_SECONDS = 15
-QUOTE_CURRENCY = "usd"
-
-
-async def ws_main(stop_event: asyncio.Event) -> None:
-    while not stop_event.is_set():
-        try:
-            await _run_once(stop_event)
-        except asyncio.CancelledError:
-            raise
-        except Exception:
-            await asyncio.sleep(WS_RECONNECT_SECONDS)
-
-
-def get_watched_markets() -> list[str]:
-    accounts = repo.list_accounts(enabled_only=True)
-    assets = set()
-    for account in accounts:
-        account_id = account["id"]
-        # Prefer normalized balances if present; fall back to raw payload later.
-        # This stays exchange-specific to Bitstamp.
-        info = None
-        try:
-            from .services_bitstamp import fetch_account_balance
-            info = fetch_account_balance(account_id)
-        except Exception:
-            continue
-        for item in info.get("balances", []):
-            asset = str(item.get("asset_code", "")).lower()
-            if asset and asset != QUOTE_CURRENCY:
-                assets.add(asset)
-    return sorted(f"{asset}{QUOTE_CURRENCY}" for asset in assets)
-
-
-async def _run_once(stop_event: asyncio.Event) -> None:
-    async with websockets.connect(WS_URL, ping_interval=None) as ws:
-        markets = get_watched_markets()
-        for market in markets:
-            await ws.send(json.dumps({"event": "bts:subscribe", "data": {"channel": f"live_trades_{market}"}}))
-
-        last_heartbeat = asyncio.get_event_loop().time()
-        while not stop_event.is_set():
-            try:
-                message = await asyncio.wait_for(ws.recv(), timeout=WS_HEARTBEAT_SECONDS)
-            except asyncio.TimeoutError:
-                await ws.send(json.dumps({"event": "bts:heartbeat"}))
-                last_heartbeat = asyncio.get_event_loop().time()
-                continue
-
-            payload = json.loads(message)
-            _handle_message(payload)
-
-
-def _handle_message(payload: dict) -> None:
-    event = payload.get("event")
-    data = payload.get("data") or {}
-    if event != "trade":
-        return
-
-    channel = str(payload.get("channel", ""))
-    market = channel.replace("live_trades_", "")
-    price = data.get("price")
-    if not market or price is None:
-        return
-
-    captured_at = datetime.now(timezone.utc).isoformat()
-    from .storage import get_connection
-    with get_connection() as conn:
-        conn.execute(
-            """
-            INSERT INTO bitstamp_live_prices (market, price, payload_json, captured_at)
-            VALUES (?, ?, ?, ?)
-            ON CONFLICT(market) DO UPDATE SET
-                price=excluded.price,
-                payload_json=excluded.payload_json,
-                captured_at=excluded.captured_at
-            """,
-            (market, str(price), json.dumps(payload), captured_at),
-        )
-        conn.commit()

src/exec_mcp/crypto_client.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import threading
+from typing import Any
+
+from mcp import ClientSession
+from mcp.client.sse import sse_client
+
+
+CRYPTO_MCP_SSE_URL = os.getenv("CRYPTO_MCP_SSE_URL", "http://localhost:8505/mcp/sse")
+
+
+async def get_price(symbol: str, base_url: str | None = None) -> dict[str, Any]:
+    url = base_url or CRYPTO_MCP_SSE_URL
+    async with sse_client(url) as (read_stream, write_stream):
+        async with ClientSession(read_stream, write_stream) as session:
+            await session.initialize()
+            result = await session.call_tool("get_price", {"symbol": symbol})
+            content = getattr(result, "content", None)
+            if not content:
+                return {"error": "EMPTY_RESULT", "symbol": symbol}
+            first = content[0]
+            text = getattr(first, "text", None)
+            if text is None and isinstance(first, dict):
+                text = first.get("text")
+            if text is None:
+                return {"error": "UNPARSEABLE_RESULT", "symbol": symbol}
+            try:
+                return json.loads(text)
+            except Exception:
+                return {"raw": text, "symbol": symbol}
+
+
+def get_price_sync(symbol: str, base_url: str | None = None) -> dict[str, Any]:
+    try:
+        asyncio.get_running_loop()
+    except RuntimeError:
+        return asyncio.run(get_price(symbol, base_url=base_url))
+
+    result: dict[str, Any] = {}
+    error: list[BaseException] = []
+
+    def _runner() -> None:
+        try:
+            result.update(asyncio.run(get_price(symbol, base_url=base_url)))
+        except BaseException as exc:  # pragma: no cover - defensive bridge
+            error.append(exc)
+
+    thread = threading.Thread(target=_runner, daemon=True)
+    thread.start()
+    thread.join()
+    if error:
+        raise error[0]
+    return result

src/exec_mcp/repo.py

@@ -181,17 +181,6 @@ def cache_delete(cache_key: str) -> None:
         conn.commit()
 
 
-def get_latest_price(market: str) -> float | None:
-    with get_connection() as conn:
-        row = conn.execute("SELECT price FROM bitstamp_live_prices WHERE market = ?", (market.lower(),)).fetchone()
-    if row is None:
-        return None
-    try:
-        return float(row["price"])
-    except (TypeError, ValueError):
-        return None
-
-
 def has_account_balance_snapshot(*, account_id: str, source_kind: str, source_ref: str) -> bool:
     with get_connection() as conn:
         row = conn.execute(

src/exec_mcp/services_bitstamp.py

@@ -9,8 +9,8 @@ except ModuleNotFoundError:  # allows test runs without the optional dependency
     bitstamp = None  # type: ignore
 
 from . import repo
+from .crypto_client import get_price_sync as get_crypto_price
 from .bitstamp import BitstampClient
-from .bitstamp_fx import load_eur_usd
 
 BITSTAMP_BASE_URL = "https://www.bitstamp.net"
 # Live trading reads should refresh quickly, but not so often that Bitstamp trips a breaker.
@@ -196,15 +196,17 @@ def _price_for_asset(asset_code: str) -> float | None:
         return None
     if asset == "usd":
         return 1.0
-    if asset == "eur":
-        fx = load_eur_usd()
-        if fx and fx.get("sell") is not None:
-            try:
-                return float(fx["sell"])
-            except (TypeError, ValueError):
-                return None
+    payload = get_crypto_price(asset)
+    if not isinstance(payload, dict):
         return None
-    return repo.get_latest_price(f"{asset}usd")
+    price = payload.get("price")
+    try:
+        price_value = float(price)
+    except (TypeError, ValueError):
+        return None
+    if price_value <= 0:
+        return None
+    return price_value
 
 
 def _value_account_balances(balance: dict) -> tuple[list[dict], float]:

src/exec_mcp/storage.py

@@ -141,13 +141,6 @@ def init_db() -> None:
                 PRIMARY KEY (kind, item_key)
             );
 
-            CREATE TABLE IF NOT EXISTS bitstamp_live_prices (
-                market TEXT PRIMARY KEY,
-                price TEXT NOT NULL,
-                payload_json TEXT NOT NULL,
-                captured_at TEXT NOT NULL
-            );
-
             CREATE TABLE IF NOT EXISTS bitstamp_fx_rates (
                 pair TEXT PRIMARY KEY,
                 buy TEXT NOT NULL,

tests/test_reporting_tools.py

@@ -5,7 +5,7 @@ import json
 import pytest
 
 from exec_mcp import repo, storage
-from exec_mcp import services_orders
+from exec_mcp import services_bitstamp, services_orders
 from exec_mcp.server import report_balance_snapshots, report_recent_trades
 
 
@@ -247,3 +247,49 @@ def test_record_order_status_update_persists_raw_json_and_triggers_snapshot_once
         account_id=account_id,
     )
     assert len(snapshot_calls) == 1
+
+
+def test_capture_account_balance_snapshot_uses_crypto_price(temp_db, monkeypatch):
+    account_id = _create_account()
+
+    monkeypatch.setattr(
+        services_bitstamp,
+        "fetch_account_balance",
+        lambda _account_id: {
+            "source": "bitstamp",
+            "cached": False,
+            "balances": [
+                {
+                    "account_id": _account_id,
+                    "asset_code": "XRP",
+                    "available": 1.0,
+                    "reserved": 0.0,
+                    "total": 2.0,
+                }
+            ],
+            "payload": [{"currency": "xrp", "total": "2.0", "available": "1.0", "reserved": "0.0"}],
+        },
+    )
+    monkeypatch.setattr(
+        services_bitstamp,
+        "get_crypto_price",
+        lambda symbol: {"symbol": symbol.upper(), "price": 1.4109, "timestamp": 1779039021},
+    )
+
+    result = services_bitstamp.capture_account_balance_snapshot(
+        account_id,
+        source_kind="order_finished",
+        source_ref="order-crypto-price",
+    )
+
+    assert result["ok"] is True
+    snapshot = result["snapshot"]
+    assert snapshot["balances"][0]["price_usd"] == 1.4109
+    with storage.get_connection() as conn:
+        row = conn.execute(
+            "SELECT snapshot_json FROM account_balance_snapshots WHERE account_id = ?",
+            (account_id,),
+        ).fetchone()
+
+    stored = json.loads(row["snapshot_json"])
+    assert stored["balances"][0]["price_usd"] == 1.4109