balances snapshots

AGENTS.md

@@ -28,7 +28,7 @@
 - Bitstamp client instances are cached per account, and nonce generation must remain monotonic per credential scope.
 - Background tasks refresh Bitstamp metadata and EUR/USD data, and the private websocket updates local order records.
 - Public discovery tools should stay read-only; order placement and cancel paths are the execution surface.
-- Reporting tools are snapshot-only: recent trades default to `filled`, support `cancelled` and `all`, and balance reporting reads stored `balance_snapshots`.
+- Reporting tools are snapshot-only: recent trades default to `finished`, support `cancelled` and `all`, and balance reporting reads stored account balance snapshots.
 
 ## Editing Rules
 

PROJECT.md

@@ -57,6 +57,9 @@ We have a clear separation of concerns:
 - Trader = strategy logic and UI/control surface
 - Exec MCP = account and order execution
 
+The recent-trades reporter accepts `all`, `open`, `cancelled`, or `finished` status filters.
+The balance snapshot reporter returns the latest snapshots for one account as a plain array.
+
 The open design question is mostly implementation detail, not architecture viability.
 This appears sane, feasible, and worth further analysis.
 

README.md

@@ -104,4 +104,6 @@ These are stored in SQLite for reuse across restarts.
 - `client_id` is optional on `place_order` and is stored on the local order record.
 - `get_open_orders(account_id, client_id=None)` filters by `client_id` when supplied.
 - `cancel_all_orders(account_id, client_id=None)` cancels all open orders for the account, and if an exchange order is already missing it is marked locally so it no longer appears open.
+- `report_recent_trades(account_id, status="finished", client_id=None, limit=50)` accepts `all`, `open`, `cancelled`, or `finished`.
+- `report_balance_snapshots(account_id, limit=50)` returns the latest balance snapshots for that account as a plain array.
 - `expire_time` is optional, no `expire_time` means a normal order, and `expire_time` means a GTD order.

src/exec_mcp/bitstamp_private_ws.py

@@ -2,12 +2,11 @@ from __future__ import annotations
 
 import asyncio
 import json
-from datetime import datetime, timezone
 
 import websockets
 
 from .services_bitstamp import get_bitstamp_client
-from .storage import get_connection
+from .services_orders import _record_order_status_update
 
 WS_URL = "wss://ws.bitstamp.net"
 WS_RECONNECT_SECONDS = 5
@@ -57,10 +56,10 @@ def _handle_message(payload: dict) -> None:
     order_id = data.get("id") or data.get("order_id")
     if not order_id:
         return
-    captured_at = datetime.now(timezone.utc).isoformat()
-    with get_connection() as conn:
-        conn.execute(
-            "UPDATE order_records SET status = ?, payload_json = ?, updated_at = ? WHERE bitstamp_order_id = ?",
-            (str(event), json.dumps(payload), captured_at, str(order_id)),
-        )
-        conn.commit()
+    status = data.get("status") or data.get("order_status") or data.get("order_state") or payload.get("status") or payload.get("order_status") or payload.get("order_state")
+    if status is None:
+        return
+    try:
+        _record_order_status_update(bitstamp_order_id=str(order_id), status=str(status), raw_json=payload)
+    except Exception:
+        return

src/exec_mcp/repo.py

@@ -192,22 +192,46 @@ def get_latest_price(market: str) -> float | None:
         return None
 
 
-def save_balance_snapshot(*, account_id: str, asset_code: str, balance: float, balance_value: float | None = None, value_currency: str | None = None) -> str:
+def has_account_balance_snapshot(*, account_id: str, source_kind: str, source_ref: str) -> bool:
+    with get_connection() as conn:
+        row = conn.execute(
+            """
+            SELECT 1
+            FROM account_balance_snapshots
+            WHERE account_id = ? AND source_kind = ? AND source_ref = ?
+            """,
+            (account_id, source_kind, source_ref),
+        ).fetchone()
+    return row is not None
+
+
+def save_account_balance_snapshot(*, account_id: str, source_kind: str, source_ref: str = "", snapshot: dict) -> str:
     snapshot_id = str(uuid4())
     captured_at = utc_now_iso()
     with get_connection() as conn:
         conn.execute(
             """
-            INSERT INTO balance_snapshots (id, account_id, asset_code, balance, balance_value, value_currency, captured_at)
-            VALUES (?, ?, ?, ?, ?, ?, ?)
+            INSERT OR IGNORE INTO account_balance_snapshots
+            (id, account_id, source_kind, source_ref, snapshot_json, captured_at)
+            VALUES (?, ?, ?, ?, ?, ?)
             """,
-            (snapshot_id, account_id, asset_code, balance, balance_value, value_currency, captured_at),
-            )
+            (snapshot_id, account_id, source_kind, source_ref, json.dumps(snapshot), captured_at),
+        )
+        row = conn.execute(
+            """
+            SELECT id
+            FROM account_balance_snapshots
+            WHERE account_id = ? AND source_kind = ? AND source_ref = ?
+            """,
+            (account_id, source_kind, source_ref),
+        ).fetchone()
         conn.commit()
-    return snapshot_id
+    if row is None:
+        return snapshot_id
+    return str(row["id"])
 
 
-def list_recent_orders(*, account_id: str, client_id: str | None = None, status: str = "filled", limit: int = 50) -> list[dict]:
+def list_recent_orders(*, account_id: str, client_id: str | None = None, status: str = "finished", limit: int = 50) -> list[dict]:
     query = """
         SELECT id, account_id, market, side, order_type, amount, price, expire_time, status,
                bitstamp_order_id, client_id, client_order_id, raw_json, created_at, updated_at
@@ -218,7 +242,13 @@ def list_recent_orders(*, account_id: str, client_id: str | None = None, status:
     if client_id is not None:
         query += " AND client_id = ?"
         params.append(client_id)
-    if status != "all":
+    if status == "open":
+        query += " AND lower(status) IN ('open', 'new', 'partially_filled')"
+    elif status == "cancelled":
+        query += " AND lower(status) = 'cancelled'"
+    elif status == "finished":
+        query += " AND lower(status) = 'finished'"
+    elif status != "all":
         query += " AND lower(status) = ?"
         params.append(status)
     query += " ORDER BY created_at DESC, id DESC LIMIT ?"
@@ -230,12 +260,12 @@ def list_recent_orders(*, account_id: str, client_id: str | None = None, status:
     return [dict(row) for row in rows]
 
 
-def list_balance_snapshots(*, account_id: str, limit: int = 50) -> list[dict]:
+def list_account_balance_snapshots(*, account_id: str, limit: int = 50) -> list[dict]:
     with get_connection() as conn:
         rows = conn.execute(
             """
-            SELECT id, account_id, asset_code, balance, balance_value, value_currency, captured_at
-            FROM balance_snapshots
+            SELECT id, account_id, source_kind, source_ref, snapshot_json, captured_at
+            FROM account_balance_snapshots
             WHERE account_id = ?
             ORDER BY captured_at DESC, id DESC
             LIMIT ?
@@ -243,4 +273,9 @@ def list_balance_snapshots(*, account_id: str, limit: int = 50) -> list[dict]:
             (account_id, limit),
         ).fetchall()
 
-    return [dict(row) for row in rows]
+    snapshots = []
+    for row in rows:
+        item = dict(row)
+        item["snapshot"] = json.loads(item.pop("snapshot_json"))
+        snapshots.append(item)
+    return snapshots

src/exec_mcp/server.py

@@ -60,13 +60,13 @@ async def lifespan(_: FastAPI):
 app = FastAPI(title="exec-mcp", lifespan=lifespan)
 app.mount("/mcp", mcp.http_app(path="/sse", transport="sse"))
 SUPPORTED_VENUES = {"bitstamp"}
-REPORT_ORDER_STATUSES = {"filled", "cancelled", "all"}
+REPORT_ORDER_STATUSES = {"all", "open", "cancelled", "finished"}
 
 
 def _normalize_report_status(status: str) -> str:
     value = str(status or "").strip().lower()
     if value not in REPORT_ORDER_STATUSES:
-        raise HTTPException(status_code=400, detail="status must be one of filled, cancelled, all")
+        raise HTTPException(status_code=400, detail="status must be one of all, open, cancelled, finished")
     return value
 
 
@@ -357,12 +357,12 @@ def cancel_all_orders(account_id: str, client_id: str | None = None) -> dict:
 
 
 @mcp.tool()
-def report_recent_trades(account_id: str, status: str = "filled", client_id: str | None = None, limit: int = 50) -> dict:
-    """report_recent_trades(account_id, status="filled", client_id=None, limit=50)
+def report_recent_trades(account_id: str, status: str = "finished", client_id: str | None = None, limit: int = 50) -> dict:
+    """report_recent_trades(account_id, status="finished", client_id=None, limit=50)
 
     Return recent order snapshots for reporting, ordered newest first. Status
-    may be `filled`, `cancelled`, or `all`. `client_id` is optional and filters
-    to one reporting stream when provided.
+    may be `all`, `open`, `cancelled`, or `finished`. `client_id` is optional
+    and filters to one reporting stream when provided.
     """
     repo.get_account(account_id)
     status = _normalize_report_status(status)
@@ -380,21 +380,14 @@ def report_recent_trades(account_id: str, status: str = "filled", client_id: str
 
 
 @mcp.tool()
-def report_balance_snapshots(account_id: str, limit: int = 50) -> dict:
+def report_balance_snapshots(account_id: str, limit: int = 50) -> list[dict]:
     """report_balance_snapshots(account_id, limit=50)
 
-    Return stored wallet balance snapshots for reporting, ordered newest first.
+    Return stored wallet balance snapshots for the account, ordered newest first.
     """
     repo.get_account(account_id)
     limit = _normalize_report_limit(limit)
-    snapshots = repo.list_balance_snapshots(account_id=account_id, limit=limit)
-    return {
-        "ok": True,
-        "account_id": account_id,
-        "limit": limit,
-        "count": len(snapshots),
-        "snapshots": snapshots,
-    }
+    return repo.list_account_balance_snapshots(account_id=account_id, limit=limit)
 
 
 def main() -> None:

src/exec_mcp/services_bitstamp.py

@@ -190,6 +190,45 @@ def _normalize_account_balances_payload(payload: list[dict], account_id: str) ->
     return balances
 
 
+def _price_for_asset(asset_code: str) -> float | None:
+    asset = str(asset_code or "").lower()
+    if not asset:
+        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
+        return None
+    return repo.get_latest_price(f"{asset}usd")
+
+
+def _value_account_balances(balance: dict) -> tuple[list[dict], float]:
+    valued_balances: list[dict] = []
+    total_value_usd = 0.0
+    for item in balance.get("balances", []):
+        asset_code = str(item.get("asset_code") or "").upper()
+        total = float(item.get("total") or 0)
+        price_usd = _price_for_asset(asset_code)
+        value_usd = total * price_usd if price_usd is not None else None
+        if value_usd is not None:
+            total_value_usd += value_usd
+        valued_balances.append(
+            {
+                **item,
+                "asset_code": asset_code,
+                "price_usd": price_usd,
+                "value_currency": "USD",
+                "value_usd": value_usd,
+            }
+        )
+    return valued_balances, total_value_usd
+
+
 def fetch_account_balance(account_id: str) -> dict:
     cache_key = f"bitstamp:account_balance:{account_id}"
     cached = repo.cache_get(cache_key)
@@ -230,6 +269,39 @@ def fetch_account_balance(account_id: str) -> dict:
         return result
 
 
+def capture_account_balance_snapshot(account_id: str, *, source_kind: str, source_ref: str = "") -> dict:
+    if source_ref and repo.has_account_balance_snapshot(account_id=account_id, source_kind=source_kind, source_ref=source_ref):
+        return {"ok": True, "account_id": account_id, "source_kind": source_kind, "source_ref": source_ref, "snapshot_id": None, "snapshot": None}
+
+    account = repo.get_account(account_id)
+    invalidate_balance_cache(account_id)
+    balance = fetch_account_balance(account_id)
+    valued_balances, total_value_usd = _value_account_balances(balance)
+    snapshot = {
+        "account_id": account["id"],
+        "display_name": account["display_name"],
+        "venue": account["venue"],
+        "venue_account_ref": account["venue_account_ref"],
+        "balances": valued_balances,
+        "total_value_usd": total_value_usd,
+        "raw_balance": balance["payload"],
+    }
+    snapshot_id = repo.save_account_balance_snapshot(
+        account_id=account_id,
+        source_kind=source_kind,
+        source_ref=source_ref,
+        snapshot=snapshot,
+    )
+    return {
+        "ok": True,
+        "account_id": account_id,
+        "source_kind": source_kind,
+        "source_ref": source_ref,
+        "snapshot_id": snapshot_id,
+        "snapshot": snapshot,
+    }
+
+
 def fetch_account_info(account_id: str) -> dict:
     cache_key = f"bitstamp:account_info:{account_id}"
     cached = repo.cache_get(cache_key)
@@ -258,26 +330,7 @@ def fetch_account_info(account_id: str) -> dict:
             _cache_error(cache_key, str(exc))
             raise
 
-        valued_balances = []
-        total_value_usd = 0.0
-        for item in balance["balances"]:
-            asset = item["asset_code"].lower()
-            total = float(item["total"])
-            if asset == "usd":
-                value_usd = total
-            else:
-                value_usd = None
-                market = f"{asset}usd"
-                price = repo.get_latest_price(market)
-                if price is not None:
-                    value_usd = total * price
-                elif asset == "eur":
-                    fx = load_eur_usd()
-                    if fx and fx.get("sell") is not None:
-                        value_usd = total * float(fx["sell"])
-            if value_usd is not None:
-                total_value_usd += value_usd
-            valued_balances.append({**item, "value_currency": "USD", "value_usd": value_usd})
+        valued_balances, total_value_usd = _value_account_balances(balance)
 
         result = {
             "id": account["id"],

src/exec_mcp/services_orders.py

@@ -11,12 +11,13 @@ from uuid import uuid4
 from fastapi import HTTPException
 
 from .bitstamp import BitstampClient, BitstampError
-from .services_bitstamp import get_bitstamp_client, clear_bitstamp_trading_client, invalidate_account_cache
+from .services_bitstamp import capture_account_balance_snapshot, get_bitstamp_client, clear_bitstamp_trading_client, invalidate_account_cache
 from .bitstamp_metadata import load_market_by_symbol
 from .storage import get_connection
 
 
 OPEN_ORDER_STATUSES = {"open", "new", "partially_filled"}
+FINISHED_ORDER_STATUS = "finished"
 _CANCEL_BREAKER_LOCK = threading.Lock()
 _CANCEL_BREAKER_NEXT_ALLOWED: dict[str, float] = {}
 _CANCEL_BREAKER_SECONDS = 3.0
@@ -78,10 +79,65 @@ def _normalize_client_id(client_id: str | None) -> str | None:
 
 
 def _normalize_status(status) -> str:
-    return str(status or "unknown").strip().lower().replace(" ", "_")
+    value = str(status or "unknown").strip().lower().replace(" ", "_")
+    if value in OPEN_ORDER_STATUSES or value in {"cancelled", "cancel_failed", "missing", FINISHED_ORDER_STATUS}:
+        return value
+    return FINISHED_ORDER_STATUS
+
+
+def _load_order_record(*, bitstamp_order_id: str) -> dict | None:
+    with get_connection() as conn:
+        row = conn.execute(
+            "SELECT id, account_id, status FROM order_records WHERE bitstamp_order_id = ?",
+            (bitstamp_order_id,),
+        ).fetchone()
+    return dict(row) if row is not None else None
+
+
+def _encode_raw_json(raw_json) -> str:
+    if isinstance(raw_json, str):
+        return raw_json
+    return json.dumps(raw_json)
+
+
+def _record_order_status_update(*, bitstamp_order_id: str, status: str, raw_json, account_id: str | None = None) -> dict:
+    record = _load_order_record(bitstamp_order_id=bitstamp_order_id)
+    if record is None:
+        raise HTTPException(status_code=404, detail="order not found")
+
+    if account_id is None:
+        account_id = str(record["account_id"])
+
+    previous_status = str(record["status"]) if record is not None else None
+    normalized_status = _normalize_status(status)
+    captured_at = _utc_now()
+    with get_connection() as conn:
+        conn.execute(
+            "UPDATE order_records SET status = ?, raw_json = ?, updated_at = ? WHERE bitstamp_order_id = ?",
+            (normalized_status, _encode_raw_json(raw_json), captured_at, bitstamp_order_id),
+        )
+        conn.commit()
+
+    if previous_status != FINISHED_ORDER_STATUS and normalized_status == FINISHED_ORDER_STATUS:
+        try:
+            capture_account_balance_snapshot(account_id, source_kind="order_finished", source_ref=bitstamp_order_id)
+        except Exception:
+            pass
+
+    invalidate_account_cache(account_id)
+    return {
+        "ok": True,
+        "account_id": account_id,
+        "bitstamp_order_id": bitstamp_order_id,
+        "status": normalized_status,
+        "previous_status": previous_status,
+    }
 
 
 def _set_local_order_status(*, bitstamp_order_id: str, status: str) -> None:
+    record = _load_order_record(bitstamp_order_id=bitstamp_order_id)
+    if record is None:
+        return
     with get_connection() as conn:
         conn.execute(
             "UPDATE order_records SET status = ?, updated_at = ? WHERE bitstamp_order_id = ?",
@@ -219,8 +275,12 @@ def get_open_orders(*, account_id: str, client_id: str | None = None) -> dict:
             result = order_status_v2(order_id=str(bitstamp_order_id), omit_transactions=True)
             status = _normalize_status(result.get("status", "unknown"))
             if status not in OPEN_ORDER_STATUSES:
-                _set_local_order_status(bitstamp_order_id=str(bitstamp_order_id), status=status)
-                invalidate_account_cache(account_id)
+                _record_order_status_update(
+                    bitstamp_order_id=str(bitstamp_order_id),
+                    status=status,
+                    raw_json=result,
+                    account_id=account_id,
+                )
                 continue
             order["status"] = status
             order["raw_json"] = json.dumps(result)
@@ -283,14 +343,12 @@ def query_order(*, account_id: str, order_id, client_order_id: str | None = None
             _invalidate_client(account_id)
         raise HTTPException(status_code=400, detail=detail) from exc
 
-    with get_connection() as conn:
-        conn.execute(
-            "UPDATE order_records SET status = ?, raw_json = ?, updated_at = ? WHERE bitstamp_order_id = ?",
-            (_normalize_status(result.get("status", "unknown")), json.dumps(result), _utc_now(), order_id),
-        )
-        conn.commit()
-
-    invalidate_account_cache(account_id)
+    _record_order_status_update(
+        bitstamp_order_id=order_id,
+        status=result.get("status", "unknown"),
+        raw_json=result,
+        account_id=account_id,
+    )
 
     return {"ok": True, "order_id": order_id, "raw": result}
 
@@ -310,13 +368,11 @@ def cancel_order(*, account_id: str, order_id) -> dict:
         raise HTTPException(status_code=400, detail=detail) from exc
 
     status = "cancelled" if result else "cancel_failed"
-    with get_connection() as conn:
-        conn.execute(
-            "UPDATE order_records SET status = ?, updated_at = ? WHERE bitstamp_order_id = ?",
-            (status, _utc_now(), order_id),
-        )
-        conn.commit()
-
-    invalidate_account_cache(account_id)
+    _record_order_status_update(
+        bitstamp_order_id=order_id,
+        status=status,
+        raw_json=result,
+        account_id=account_id,
+    )
 
     return {"ok": bool(result), "order_id": order_id, "raw": result}

src/exec_mcp/storage.py

@@ -92,17 +92,22 @@ def init_db() -> None:
                 FOREIGN KEY(account_id) REFERENCES accounts(id) ON DELETE CASCADE
             );
 
-            CREATE TABLE IF NOT EXISTS balance_snapshots (
+            DROP TABLE IF EXISTS balance_snapshots;
+
+            CREATE TABLE IF NOT EXISTS account_balance_snapshots (
                 id TEXT PRIMARY KEY,
                 account_id TEXT NOT NULL,
-                asset_code TEXT NOT NULL,
-                balance REAL NOT NULL,
-                balance_value REAL,
-                value_currency TEXT,
+                source_kind TEXT NOT NULL,
+                source_ref TEXT NOT NULL DEFAULT '',
+                snapshot_json TEXT NOT NULL,
                 captured_at TEXT NOT NULL,
-                FOREIGN KEY(account_id) REFERENCES accounts(id) ON DELETE CASCADE
+                FOREIGN KEY(account_id) REFERENCES accounts(id) ON DELETE CASCADE,
+                UNIQUE(account_id, source_kind, source_ref)
             );
 
+            CREATE INDEX IF NOT EXISTS idx_account_balance_snapshots_account_captured_at
+                ON account_balance_snapshots(account_id, captured_at DESC, id DESC);
+
             CREATE TABLE IF NOT EXISTS order_records (
                 id TEXT PRIMARY KEY,
                 account_id TEXT NOT NULL,

tests/test_reporting_tools.py

@@ -5,6 +5,7 @@ import json
 import pytest
 
 from exec_mcp import repo, storage
+from exec_mcp import services_orders
 from exec_mcp.server import report_balance_snapshots, report_recent_trades
 
 
@@ -21,15 +22,16 @@ def _create_account() -> str:
     return repo.list_accounts(enabled_only=False)[0]["id"]
 
 
-def test_report_recent_trades_defaults_to_filled_and_sorts_newest_first(temp_db):
+def test_report_recent_trades_defaults_to_finished_and_sorts_newest_first(temp_db):
     account_id = _create_account()
 
     with storage.get_connection() as conn:
         rows = [
-            ("rec-1", "filled", "client-a", "1994292738961401", "2026-04-09T08:20:50+00:00"),
+            ("rec-1", "finished", "client-a", "1994292738961401", "2026-04-09T08:20:50+00:00"),
             ("rec-2", "cancelled", "client-a", "1994292738961402", "2026-04-09T09:20:50+00:00"),
-            ("rec-3", "filled", "client-b", "1994292738961403", "2026-04-09T10:20:50+00:00"),
+            ("rec-3", "finished", "client-b", "1994292738961403", "2026-04-09T10:20:50+00:00"),
             ("rec-4", "open", "client-a", "1994292738961404", "2026-04-09T11:20:50+00:00"),
+            ("rec-5", "new", "client-a", "1994292738961405", "2026-04-09T12:20:50+00:00"),
         ]
         for record_id, status, client_id, order_id, stamp in rows:
             conn.execute(
@@ -60,20 +62,67 @@ def test_report_recent_trades_defaults_to_filled_and_sorts_newest_first(temp_db)
 
     result = report_recent_trades(account_id=account_id)
     assert result["ok"] is True
-    assert result["status"] == "filled"
+    assert result["status"] == "finished"
     assert result["count"] == 2
     assert [row["bitstamp_order_id"] for row in result["orders"]] == ["1994292738961403", "1994292738961401"]
 
 
+def test_report_recent_trades_supports_open_status_and_client_filter(temp_db):
+    account_id = _create_account()
+
+    with storage.get_connection() as conn:
+        for idx, (status, client_id, stamp) in enumerate(
+            [
+                ("open", "client-a", "2026-04-09T08:20:50+00:00"),
+                ("new", "client-a", "2026-04-09T09:20:50+00:00"),
+                ("partially_filled", "client-b", "2026-04-09T10:20:50+00:00"),
+                ("cancelled", "client-a", "2026-04-09T11:20:50+00:00"),
+            ],
+            start=1,
+        ):
+            order_id = f"19942927389614{idx}"
+            conn.execute(
+                """
+                INSERT INTO order_records
+                (id, account_id, market, side, order_type, amount, price, expire_time, status, bitstamp_order_id, client_id, client_order_id, raw_json, created_at, updated_at)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    f"rec-{idx}",
+                    account_id,
+                    "xrpusd",
+                    "sell",
+                    "limit",
+                    "10",
+                    "2.00",
+                    None,
+                    status,
+                    order_id,
+                    client_id,
+                    None,
+                    json.dumps({"id": order_id, "status": status}),
+                    stamp,
+                    stamp,
+                ),
+            )
+        conn.commit()
+
+    result = report_recent_trades(account_id=account_id, status="open", client_id="client-a", limit=10)
+    assert result["ok"] is True
+    assert result["status"] == "open"
+    assert result["client_id"] == "client-a"
+    assert [row["status"] for row in result["orders"]] == ["new", "open"]
+
+
 def test_report_recent_trades_supports_all_statuses_and_client_filter(temp_db):
     account_id = _create_account()
 
     with storage.get_connection() as conn:
         for idx, (status, client_id, stamp) in enumerate(
             [
-                ("filled", "client-a", "2026-04-09T08:20:50+00:00"),
+                ("finished", "client-a", "2026-04-09T08:20:50+00:00"),
                 ("cancelled", "client-a", "2026-04-09T09:20:50+00:00"),
-                ("filled", "client-b", "2026-04-09T10:20:50+00:00"),
+                ("finished", "client-b", "2026-04-09T10:20:50+00:00"),
             ],
             start=1,
         ):
@@ -85,7 +134,7 @@ def test_report_recent_trades_supports_all_statuses_and_client_filter(temp_db):
                 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                 """,
                 (
-                    f"rec-{idx}",
+                    f"rec-all-{idx}",
                     account_id,
                     "xrpusd",
                     "sell",
@@ -108,16 +157,93 @@ def test_report_recent_trades_supports_all_statuses_and_client_filter(temp_db):
     assert result["ok"] is True
     assert result["status"] == "all"
     assert result["client_id"] == "client-a"
-    assert [row["status"] for row in result["orders"]] == ["cancelled", "filled"]
+    assert [row["status"] for row in result["orders"]] == ["cancelled", "finished"]
 
 
 def test_report_balance_snapshots_returns_history_newest_first(temp_db):
     account_id = _create_account()
 
-    repo.save_balance_snapshot(account_id=account_id, asset_code="BTC", balance=1.25, balance_value=11250.0, value_currency="USD")
-    repo.save_balance_snapshot(account_id=account_id, asset_code="USD", balance=500.0, balance_value=500.0, value_currency="USD")
+    repo.save_account_balance_snapshot(
+        account_id=account_id,
+        source_kind="order_finished",
+        source_ref="order-1",
+        snapshot={"balances": [{"asset_code": "BTC"}], "total_value_usd": 11250.0},
+    )
+    repo.save_account_balance_snapshot(
+        account_id=account_id,
+        source_kind="order_finished",
+        source_ref="order-2",
+        snapshot={"balances": [{"asset_code": "USD"}], "total_value_usd": 500.0},
+    )
 
     result = report_balance_snapshots(account_id=account_id)
-    assert result["ok"] is True
-    assert result["count"] == 2
-    assert [row["asset_code"] for row in result["snapshots"]] == ["USD", "BTC"]
+    assert len(result) == 2
+    assert [row["source_ref"] for row in result] == ["order-2", "order-1"]
+    assert [row["snapshot"]["balances"][0]["asset_code"] for row in result] == ["USD", "BTC"]
+
+
+def test_record_order_status_update_persists_raw_json_and_triggers_snapshot_once(temp_db, monkeypatch):
+    account_id = _create_account()
+
+    with storage.get_connection() as conn:
+        conn.execute(
+            """
+            INSERT INTO order_records
+            (id, account_id, market, side, order_type, amount, price, expire_time, status, bitstamp_order_id, client_id, client_order_id, raw_json, created_at, updated_at)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            (
+                "rec-1",
+                account_id,
+                "xrpusd",
+                "buy",
+                "limit",
+                "10",
+                "2.00",
+                None,
+                "open",
+                "1994292738961401",
+                "client-a",
+                None,
+                json.dumps({"status": "open"}),
+                "2026-04-09T08:20:50+00:00",
+                "2026-04-09T08:20:50+00:00",
+            ),
+        )
+        conn.commit()
+
+    snapshot_calls: list[dict] = []
+    monkeypatch.setattr(
+        services_orders,
+        "capture_account_balance_snapshot",
+        lambda account_id, *, source_kind, source_ref="": snapshot_calls.append(
+            {"account_id": account_id, "source_kind": source_kind, "source_ref": source_ref}
+        ),
+    )
+
+    services_orders._record_order_status_update(
+        bitstamp_order_id="1994292738961401",
+        status="finished",
+        raw_json={"id": "1994292738961401", "status": "finished"},
+        account_id=account_id,
+    )
+
+    with storage.get_connection() as conn:
+        row = conn.execute(
+            "SELECT status, raw_json FROM order_records WHERE bitstamp_order_id = ?",
+            ("1994292738961401",),
+        ).fetchone()
+
+    assert row["status"] == "finished"
+    assert json.loads(row["raw_json"]) == {"id": "1994292738961401", "status": "finished"}
+    assert snapshot_calls == [
+        {"account_id": account_id, "source_kind": "order_finished", "source_ref": "1994292738961401"}
+    ]
+
+    services_orders._record_order_status_update(
+        bitstamp_order_id="1994292738961401",
+        status="finished",
+        raw_json={"id": "1994292738961401", "status": "finished"},
+        account_id=account_id,
+    )
+    assert len(snapshot_calls) == 1