From ff7a81ceeb81264c1663fc1db6cd4cdf8f3aeb3b Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Tue, 14 Jul 2026 23:32:37 +0200 Subject: [PATCH] feat(D2.1): capture real IBKR fills->exec_fills (strategy-tagged) + ibkr_account_nav snapshot (best-effort, paper-only) Co-Authored-By: Claude Opus 4.8 --- src/fxhnt/adapters/broker/ibkr.py | 28 ++ .../adapters/persistence/cockpit_models.py | 16 ++ .../adapters/persistence/ibkr_account_repo.py | 94 +++++++ src/fxhnt/application/exec_capture.py | 70 +++++ src/fxhnt/application/execution.py | 10 + .../application/multistrat_exec_record.py | 20 +- src/fxhnt/cli.py | 6 +- tests/integration/test_exec_capture.py | 244 ++++++++++++++++++ tests/unit/test_ibkr_account_repo.py | 71 +++++ 9 files changed, 556 insertions(+), 3 deletions(-) create mode 100644 src/fxhnt/adapters/persistence/ibkr_account_repo.py create mode 100644 src/fxhnt/application/exec_capture.py create mode 100644 tests/integration/test_exec_capture.py create mode 100644 tests/unit/test_ibkr_account_repo.py diff --git a/src/fxhnt/adapters/broker/ibkr.py b/src/fxhnt/adapters/broker/ibkr.py index 2ea9c57..0cb4e07 100644 --- a/src/fxhnt/adapters/broker/ibkr.py +++ b/src/fxhnt/adapters/broker/ibkr.py @@ -7,6 +7,7 @@ Bakes in hard-won lessons: paper detection via the DU* account prefix; whole-sha """ from __future__ import annotations +from dataclasses import dataclass, field from types import TracebackType from typing import Any @@ -15,6 +16,16 @@ from fxhnt.ports.broker import AccountState, Order _SUMMARY_TAGS = ("NetLiquidation", "TotalCashValue", "GrossPositionValue") +@dataclass +class AccountSnapshot: + """The REAL DU-paper-account state at a rebalance (D2.1) — captured additively alongside the existing + intended-book return basis, for the `ibkr_account_nav` table.""" + nlv: float + cash: float + gross: float + positions: dict[str, float] = field(default_factory=dict) + + class IbkrBroker: name = "ibkr" supports_fractional = False # IBKR rejects fractional market orders (error 10243) — whole shares only @@ -22,6 +33,7 @@ class IbkrBroker: def __init__(self, host: str = "127.0.0.1", port: int = 4002, client_id: int = 17, timeout: int = 15) -> None: self._host, self._port, self._cid, self._timeout = host, port, client_id, timeout self._ib: Any = None + self._trades: list[Any] = [] # ib-async Trade objects from place_order, popped by pop_trades() def _ensure(self) -> Any: from ib_async import IB @@ -58,8 +70,24 @@ class IbkrBroker: market_order.account = accts[0] trade = ib.placeOrder(contract, market_order) ib.sleep(2) + self._trades.append(trade) # collected by pop_trades() so fills are capturable (D2.1) return str(trade.orderStatus.status) + def pop_trades(self) -> list[Any]: + """Return and clear the ib-async `Trade` objects placed since the last pop (one rebalance's worth). + `exec_capture.capture_fills` reads each trade's `.fills` to persist the realized executions.""" + trades, self._trades = self._trades, [] + return trades + + def account_snapshot(self) -> AccountSnapshot: + """The REAL account state right now (NLV/cash/gross + positions), reusing the same `accountSummary`/ + `positions` reads as `account_state` (D2.1 — captured additively, not the gate-relevant basis).""" + ib = self._ensure() + summ = {v.tag: float(v.value) for v in ib.accountSummary() if v.tag in _SUMMARY_TAGS} + positions = {p.contract.symbol: float(p.position) for p in ib.positions()} + return AccountSnapshot(nlv=summ.get("NetLiquidation", 0.0), cash=summ.get("TotalCashValue", 0.0), + gross=summ.get("GrossPositionValue", 0.0), positions=positions) + def place_combo(self, legs: list[tuple[str, str, int]], net_limit: float) -> str: """Atomic XSP put-credit-spread as a BAG combo. legs = (osi_symbol, side, qty); net_limit<0 = credit. Requires options trading permission (account currently lacks it -> this path ships suspended).""" diff --git a/src/fxhnt/adapters/persistence/cockpit_models.py b/src/fxhnt/adapters/persistence/cockpit_models.py index cab17b9..d02005e 100644 --- a/src/fxhnt/adapters/persistence/cockpit_models.py +++ b/src/fxhnt/adapters/persistence/cockpit_models.py @@ -108,6 +108,7 @@ class ExecFillRow(CockpitBase): __tablename__ = "exec_fills" client_id: Mapped[str] = mapped_column(String(64), primary_key=True) rebalance_id: Mapped[str] = mapped_column(String(48), index=True) + strategy_id: Mapped[str] = mapped_column(String(64), index=True, default="") # "" = back-compat/untagged symbol: Mapped[str] = mapped_column(String(32)) side: Mapped[str] = mapped_column(String(4)) qty: Mapped[float] = mapped_column(Float) @@ -118,6 +119,21 @@ class ExecFillRow(CockpitBase): at: Mapped[dt.datetime] = mapped_column(DateTime) +class IbkrAccountNavRow(CockpitBase): + """Per-run REAL DU-paper-account snapshot (D2.1) — the actual NLV/cash/gross + positions IBKR reports at + each `execute-multistrat` run, captured ADDITIVELY alongside the existing intended-book return basis + (`multistrat_exec_pos`/`forward_record`, unchanged). `run_date` PK — idempotent replace (a same-day re-run + overwrites, never appends). `positions_json` mirrors `ForwardBookStateRow.extra_json`'s JSON-as-text + approach so this table works identically on SQLite (tests) and Postgres.""" + __tablename__ = "ibkr_account_nav" + run_date: Mapped[dt.date] = mapped_column(Date, primary_key=True) + nlv: Mapped[float] = mapped_column(Float) + cash: Mapped[float] = mapped_column(Float) + gross: Mapped[float] = mapped_column(Float) + positions_json: Mapped[str] = mapped_column(Text) + at: Mapped[dt.datetime] = mapped_column(DateTime) + + class PaperPositionRow(CockpitBase): __tablename__ = "paper_positions" run_date: Mapped[dt.date] = mapped_column(Date, primary_key=True) diff --git a/src/fxhnt/adapters/persistence/ibkr_account_repo.py b/src/fxhnt/adapters/persistence/ibkr_account_repo.py new file mode 100644 index 0000000..9a877ee --- /dev/null +++ b/src/fxhnt/adapters/persistence/ibkr_account_repo.py @@ -0,0 +1,94 @@ +"""Repository for the D2.1 REAL DU-paper-account capture: the per-run `ibkr_account_nav` snapshot (NLV/cash/ +gross/positions) plus strategy-tagged reads of the (pre-existing, previously-empty) `exec_fills` table. +Mirrors `ForwardNavRepo`'s engine/StaticPool construction and its guarded idempotent ALTER pattern +(`_ensure_summary_policy_columns`) — `migrate()` never runs heavy DDL on every startup.""" +from __future__ import annotations + +import datetime as dt +import json +import uuid +from typing import Any + +from sqlalchemy import create_engine, select, text +from sqlalchemy.orm import Session +from sqlalchemy.pool import StaticPool + +from fxhnt.adapters.persistence.cockpit_models import CockpitBase, ExecFillRow, IbkrAccountNavRow +from fxhnt.application.exec_capture import FillDTO + + +class IbkrAccountRepo: + def __init__(self, dsn: str) -> None: + # StaticPool keeps a single in-memory SQLite connection alive across sessions; only applied for the + # in-memory SQLite case (tests). StaticPool is NOT used for Postgres. + kw: dict[str, Any] = {"future": True} + if dsn == "sqlite://" or (dsn.startswith("sqlite://") and ":memory:" in dsn): + kw |= {"connect_args": {"check_same_thread": False}, "poolclass": StaticPool} + self._engine = create_engine(dsn, **kw) + self._is_pg = self._engine.dialect.name == "postgresql" + + def migrate(self) -> None: + CockpitBase.metadata.create_all(self._engine) + self._ensure_exec_fills_strategy_column() + + def _ensure_exec_fills_strategy_column(self) -> None: + """Idempotently add `exec_fills.strategy_id` to a table that pre-dates this task (`create_all` does + NOT add columns to an already-existing table on prod Postgres). Postgres has `ADD COLUMN IF NOT + EXISTS`; sqlite does not, so PRAGMA-check the columns first. A brand-new table already has the + column from `create_all` — this is then a no-op on both dialects.""" + with self._engine.begin() as c: + if self._is_pg: + c.execute(text( + "ALTER TABLE exec_fills ADD COLUMN IF NOT EXISTS strategy_id VARCHAR(64) DEFAULT ''")) + else: + cols = {row[1] for row in c.execute(text("PRAGMA table_info(exec_fills)"))} + if "strategy_id" not in cols: + c.execute(text( + "ALTER TABLE exec_fills ADD COLUMN strategy_id VARCHAR(64) DEFAULT ''")) + + def replace_snapshot(self, run_date: str, *, nlv: float, cash: float, gross: float, + positions: dict[str, float], at: dt.datetime) -> None: + """Idempotent per run_date: delete-then-insert in one transaction, so a same-day re-run overwrites + rather than double-books (mirrors `ForwardNavRepo.replace_nav`).""" + rd = dt.date.fromisoformat(run_date) + positions_json = json.dumps({str(k): float(v) for k, v in positions.items()}) + with Session(self._engine) as s: + s.query(IbkrAccountNavRow).filter(IbkrAccountNavRow.run_date == rd).delete() + s.add(IbkrAccountNavRow(run_date=rd, nlv=float(nlv), cash=float(cash), gross=float(gross), + positions_json=positions_json, at=at)) + s.commit() + + def nav_series(self) -> list[tuple[str, float]]: + """The account's (date, NLV) series ascending — the real-account equivalent of `ForwardNavRepo`'s + nav history.""" + with Session(self._engine) as s: + rows = s.scalars(select(IbkrAccountNavRow).order_by(IbkrAccountNavRow.run_date.asc())) + return [(r.run_date.isoformat(), r.nlv) for r in rows] + + def latest_positions(self) -> dict[str, float]: + """The most recently snapshotted real positions, or {} if none captured yet.""" + with Session(self._engine) as s: + row = s.scalars(select(IbkrAccountNavRow) + .order_by(IbkrAccountNavRow.run_date.desc()).limit(1)).first() + if row is None: + return {} + return {str(k): float(v) for k, v in json.loads(row.positions_json).items()} + + def record_fills(self, *, strategy_id: str, rebalance_id: str, fills: list[FillDTO], + at: dt.datetime) -> None: + """Append each captured fill as an `exec_fills` row tagged `strategy_id`. `client_id` (the table's + PK) is synthesized per fill since `FillDTO` carries no execution id of its own.""" + with Session(self._engine) as s: + for f in fills: + s.add(ExecFillRow(client_id=str(uuid.uuid4()), rebalance_id=rebalance_id, + strategy_id=strategy_id, symbol=f.symbol, side=f.side, qty=f.qty, + price=f.price, fee=f.fee, status=f.status, + shortfall_bps=f.shortfall_bps, at=at)) + s.commit() + + def fills_for(self, strategy_id: str, limit: int = 20) -> list[ExecFillRow]: + """The most recent `limit` fills tagged `strategy_id`, newest first. Read-only.""" + stmt = (select(ExecFillRow).where(ExecFillRow.strategy_id == strategy_id) + .order_by(ExecFillRow.at.desc()).limit(limit)) + with Session(self._engine) as s: + return list(s.scalars(stmt)) diff --git a/src/fxhnt/application/exec_capture.py b/src/fxhnt/application/exec_capture.py new file mode 100644 index 0000000..aa81a44 --- /dev/null +++ b/src/fxhnt/application/exec_capture.py @@ -0,0 +1,70 @@ +"""D2.1 — capture the REAL DU-paper-account state produced by an `execute-multistrat` run: the fills that +actually got placed (persisted into the existing `exec_fills` table, now strategy-tagged) and the +post-rebalance account snapshot (`ibkr_account_nav`). This is ADDITIVE to the existing intended-book return +basis (`multistrat_exec_pos`/`forward_record` via `record_exec_track`, unchanged) — a data-layer record of +what the real account actually did, not the gate-relevant return basis. + +BEST-EFFORT throughout: a persistence failure logs and never raises, so a capture bug can never fail or +block an otherwise-successful rebalance (mirrors the existing best-effort ref-persist pattern elsewhere in +the codebase).""" +from __future__ import annotations + +import datetime as dt +import logging +from collections.abc import Iterable +from dataclasses import dataclass +from typing import Any + +log = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class FillDTO: + symbol: str + side: str + qty: float + price: float + fee: float + status: str + shortfall_bps: float + + +def capture_fills(trades: Iterable[Any]) -> list[FillDTO]: + """Map ib-async `Trade` objects (each carrying `.orderStatus`, `.order`, and `.fills` — one per partial/ + full execution) to `FillDTO`s. `shortfall_bps` = (fill price − intended price) / intended price · 1e4, + where the intended price is the order's limit price (`order.lmtPrice`) when a real one was set; whole- + share market orders (the current multistrat execution path) carry none, so `shortfall_bps` is 0.0 for + them today — this is forward-looking for any future limit/combo order path.""" + out: list[FillDTO] = [] + for trade in trades: + status = str(getattr(getattr(trade, "orderStatus", None), "status", "")) + intended = getattr(getattr(trade, "order", None), "lmtPrice", None) + intended_price = float(intended) if intended else None + for fill in getattr(trade, "fills", []): + execution = fill.execution + price = float(execution.price) + report = getattr(fill, "commissionReport", None) + fee = float(getattr(report, "commission", 0.0)) if report is not None else 0.0 + shortfall_bps = ((price - intended_price) / intended_price) * 1e4 if intended_price else 0.0 + out.append(FillDTO(symbol=fill.contract.symbol, side=execution.side, qty=float(execution.shares), + price=price, fee=fee, status=status, shortfall_bps=shortfall_bps)) + return out + + +def persist_execution(repo: Any, *, strategy_id: str, rebalance_id: str, fills: list[FillDTO], + snapshot: Any, at: dt.datetime) -> None: + """Best-effort write of `fills` (tagged `strategy_id`, into `exec_fills`) and `snapshot` (into + `ibkr_account_nav`, keyed by `at`'s date). The two writes are independently guarded — a failure in one + never blocks the other, and NEITHER ever raises: a persistence bug here must never fail an already-placed + rebalance.""" + try: + repo.record_fills(strategy_id=strategy_id, rebalance_id=rebalance_id, fills=fills, at=at) + except Exception: + log.warning("exec capture: failed to persist fills for %s (best-effort, non-blocking)", + strategy_id, exc_info=True) + try: + repo.replace_snapshot(at.date().isoformat(), nlv=snapshot.nlv, cash=snapshot.cash, + gross=snapshot.gross, positions=snapshot.positions, at=at) + except Exception: + log.warning("exec capture: failed to persist account snapshot for %s (best-effort, non-blocking)", + strategy_id, exc_info=True) diff --git a/src/fxhnt/application/execution.py b/src/fxhnt/application/execution.py index 94e6544..e3a9253 100644 --- a/src/fxhnt/application/execution.py +++ b/src/fxhnt/application/execution.py @@ -6,6 +6,8 @@ The safety gates are hard-won from live paper trading. """ from __future__ import annotations +from typing import Any + from pydantic import BaseModel, Field from fxhnt.config import Settings @@ -29,6 +31,11 @@ class RebalancePlan(BaseModel): executed: bool = False blocked: bool = False notes: list[str] = Field(default_factory=list) + rebalance_id: str = "" # caller-assigned; "" = unset (D2.1) + # Raw broker trade objects (e.g. ib-async Trade) placed this run, when the broker exposes `pop_trades()` + # — read by `exec_capture.capture_fills` to persist the REAL fills (D2.1). Empty for any broker that + # doesn't support trade collection (e.g. the test FakeBroker / a fractional broker) or on a dry run. + trades: list[Any] = Field(default_factory=list) class ExecutionService: @@ -130,4 +137,7 @@ class ExecutionService: notes.append(f"order failed {o.symbol}: {str(e)[:80]}") plan.executed = placed > 0 plan.notes = notes + pop_trades = getattr(self._broker, "pop_trades", None) # only IbkrBroker exposes this (D2.1) + if callable(pop_trades): + plan.trades = pop_trades() return plan diff --git a/src/fxhnt/application/multistrat_exec_record.py b/src/fxhnt/application/multistrat_exec_record.py index 0f9be88..8200a6c 100644 --- a/src/fxhnt/application/multistrat_exec_record.py +++ b/src/fxhnt/application/multistrat_exec_record.py @@ -8,8 +8,11 @@ C1 task (Task 3). No new execution engine, no new return store.""" from __future__ import annotations import datetime as dt +import logging from typing import Any +log = logging.getLogger(__name__) + def scaled_weights(within_weights: dict[str, float], envelope: float, nlv: float) -> dict[str, float]: if envelope <= 0.0 or nlv <= 0.0: @@ -66,7 +69,7 @@ def record_exec_track(repo: Any, exec_return: float | None, new_positions: dict[ def plan_and_record(*, repo: Any, exec_svc: Any, within_weights: dict[str, float], prices: dict[str, float], nlv: float, today: str, at: dt.datetime, max_gross: float, execute: bool, paper_envelope: float = 0.0, account_is_paper: bool = True, - venue: str = "") -> tuple[Any, float]: + venue: str = "", account_repo: Any = None, broker: Any = None) -> tuple[Any, float]: """Compute the B2a envelope, size the book to it (flatten on $0), reconcile via exec_svc, and on `execute` (when the plan was NOT blocked by a safety gate) record the observed multistrat_exec return. Returns (plan, envelope). The next run's return basis is the TARGET book we rebalance TO @@ -77,7 +80,12 @@ def plan_and_record(*, repo: Any, exec_svc: Any, within_weights: dict[str, float diverge the executed track from the real account. `paper_envelope > 0` sizes the book to min(paper_envelope, nlv) and does NOT read B2a — a PAPER run that - bypasses the real-capital gate, hard-refused (no orders) unless `account_is_paper`. `venue` is recorded.""" + bypasses the real-capital gate, hard-refused (no orders) unless `account_is_paper`. `venue` is recorded. + + `account_repo` + `broker` (both optional, default None) additionally capture the REAL DU-account state + (D2.1): when both are supplied AND `account_is_paper`, the fills actually placed this run + a fresh + account snapshot are persisted via `exec_capture`. This is ADDITIVE and BEST-EFFORT — it never changes + `positions_after` (the gate-relevant return basis, unchanged above) and a capture failure only logs.""" if paper_envelope > 0.0: if not account_is_paper: raise ValueError("paper-envelope refused: account is not paper (real-capital exposure guard)") @@ -95,4 +103,12 @@ def plan_and_record(*, repo: Any, exec_svc: Any, within_weights: dict[str, float positions_after = {s: (w * float(nlv)) / prices[s] # the target book just placed for s, w in weights.items() if prices.get(s, 0.0) > 0.0} record_exec_track(repo, exec_return, positions_after, prices, envelope, today, at, venue) + if account_is_paper and account_repo is not None and broker is not None: + try: + from fxhnt.application.exec_capture import capture_fills, persist_execution + persist_execution(account_repo, strategy_id="multistrat", + rebalance_id=(plan.rebalance_id or today), fills=capture_fills(plan.trades), + snapshot=broker.account_snapshot(), at=at) + except Exception: # capture is ADDITIVE — never let it fail an already-placed rebalance + log.warning("multistrat exec capture failed (best-effort, non-blocking)", exc_info=True) return plan, envelope diff --git a/src/fxhnt/cli.py b/src/fxhnt/cli.py index c2f8f42..94e65a0 100644 --- a/src/fxhnt/cli.py +++ b/src/fxhnt/cli.py @@ -261,6 +261,7 @@ def execute_multistrat( import datetime as _dt from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo + from fxhnt.adapters.persistence.ibkr_account_repo import IbkrAccountRepo from fxhnt.application.multistrat_exec_record import plan_and_record from fxhnt.domain.strategies import multistrat @@ -294,6 +295,8 @@ def execute_multistrat( repo = ForwardNavRepo(settings.operational_dsn) repo.migrate() + account_repo = IbkrAccountRepo(settings.operational_dsn) # D2.1 — real fills + NAV snapshot capture + account_repo.migrate() today = _dt.datetime.now(_dt.UTC).strftime("%Y-%m-%d") at = _dt.datetime.now(_dt.UTC).replace(tzinfo=None) @@ -312,7 +315,8 @@ def execute_multistrat( plan, envelope = plan_and_record( repo=repo, exec_svc=svc, within_weights=bweights, prices=bprices, nlv=acct.nlv, today=today, at=at, max_gross=max_gross, execute=do_execute, - paper_envelope=paper_env, account_is_paper=acct.is_paper, venue=venue) + paper_envelope=paper_env, account_is_paper=acct.is_paper, venue=venue, + account_repo=account_repo, broker=brk) except ValueError as e: # a deliberate safety refusal (e.g. paper-envelope on a non-paper account) — typer.echo(f"paper-envelope refused: {e}") # distinct from a connectivity error, keep it that way raise typer.Exit(1) from e diff --git a/tests/integration/test_exec_capture.py b/tests/integration/test_exec_capture.py new file mode 100644 index 0000000..3454aaf --- /dev/null +++ b/tests/integration/test_exec_capture.py @@ -0,0 +1,244 @@ +"""D2.1 — capture the REAL DU-account fills + snapshot at an `execute-multistrat` run: capture_fills maps +ib-async Trade objects to FillDTOs; persist_execution writes them (strategy-tagged) + the account snapshot; +both are best-effort (never raise on a broken repo); plan_and_record wires the whole capture end-to-end +when an account_repo + broker are supplied (and is a no-op — no AttributeError — when they are not, so +existing callers that don't pass them are unaffected).""" +from __future__ import annotations + +import datetime as dt +from types import SimpleNamespace + +import pytest + +from fxhnt.adapters.broker.ibkr import AccountSnapshot +from fxhnt.adapters.persistence.ibkr_account_repo import IbkrAccountRepo +from fxhnt.application.exec_capture import capture_fills, persist_execution + + +class _FakeExecution: + def __init__(self, side: str, shares: float, price: float, avg_price: float) -> None: + self.side = side + self.shares = shares + self.price = price + self.avgPrice = avg_price + + +class _FakeCommissionReport: + def __init__(self, commission: float) -> None: + self.commission = commission + + +class _FakeContract: + def __init__(self, symbol: str) -> None: + self.symbol = symbol + + +class _FakeFill: + def __init__(self, symbol: str, side: str, shares: float, price: float, avg_price: float, + commission: float) -> None: + self.contract = _FakeContract(symbol) + self.execution = _FakeExecution(side, shares, price, avg_price) + self.commissionReport = _FakeCommissionReport(commission) + + +class _FakeTrade: + def __init__(self, fills: list[_FakeFill], status: str = "Filled", lmt_price: float | None = None) -> None: + self.fills = fills + self.orderStatus = SimpleNamespace(status=status) + self.order = SimpleNamespace(lmtPrice=lmt_price) if lmt_price is not None else SimpleNamespace() + + +# ---- capture_fills ----------------------------------------------------------------------------------- + +def test_capture_fills_maps_trade_fills_to_dtos(): + trade = _FakeTrade([_FakeFill("SPY", "BOT", 10.0, 452.30, 452.31, 1.25)]) + out = capture_fills([trade]) + assert len(out) == 1 + f = out[0] + assert f.symbol == "SPY" + assert f.side == "BOT" + assert f.qty == 10.0 + assert f.price == 452.30 + assert f.fee == 1.25 + assert f.status == "Filled" + assert f.shortfall_bps == 0.0 # market order — no intended (limit) price to compare against + + +def test_capture_fills_computes_shortfall_when_intended_price_known(): + trade = _FakeTrade([_FakeFill("SPY", "BOT", 10.0, 452.30, 452.31, 1.25)], lmt_price=452.0) + out = capture_fills([trade]) + assert out[0].shortfall_bps == pytest.approx((452.30 - 452.0) / 452.0 * 1e4) + + +def test_capture_fills_handles_multiple_trades_and_fills(): + t1 = _FakeTrade([_FakeFill("SPY", "BOT", 10.0, 452.30, 452.31, 1.25)]) + t2 = _FakeTrade([_FakeFill("IEF", "SLD", 5.0, 99.0, 99.0, 0.50), + _FakeFill("IEF", "SLD", 3.0, 99.1, 99.1, 0.30)]) + out = capture_fills([t1, t2]) + assert len(out) == 3 + assert [f.symbol for f in out] == ["SPY", "IEF", "IEF"] + + +def test_capture_fills_empty_trades_returns_empty(): + assert capture_fills([]) == [] + + +# ---- persist_execution -------------------------------------------------------------------------------- + +def test_persist_execution_writes_fills_and_snapshot(tmp_path): + repo = IbkrAccountRepo(f"sqlite:///{tmp_path}/t.db") + repo.migrate() + trade = _FakeTrade([_FakeFill("SPY", "BOT", 10.0, 452.30, 452.31, 1.25)]) + fills = capture_fills([trade]) + snapshot = AccountSnapshot(nlv=1_000_000.0, cash=500.0, gross=999_500.0, positions={"SPY": 10.0}) + at = dt.datetime(2026, 7, 14, 15, 0) + + persist_execution(repo, strategy_id="multistrat", rebalance_id="2026-07-14", fills=fills, + snapshot=snapshot, at=at) + + stored = repo.fills_for("multistrat", limit=10) + assert len(stored) == 1 + assert stored[0].symbol == "SPY" and stored[0].strategy_id == "multistrat" + assert stored[0].rebalance_id == "2026-07-14" + assert repo.nav_series() == [("2026-07-14", 1_000_000.0)] + assert repo.latest_positions() == {"SPY": 10.0} + + +class _RaisingRepo: + def record_fills(self, **kwargs): + raise RuntimeError("db is down") + + def replace_snapshot(self, *args, **kwargs): + raise RuntimeError("db is down") + + +def test_persist_execution_is_best_effort_never_raises(): + persist_execution(_RaisingRepo(), strategy_id="multistrat", rebalance_id="x", fills=[], + snapshot=AccountSnapshot(nlv=0.0, cash=0.0, gross=0.0, positions={}), + at=dt.datetime(2026, 7, 14)) # must not raise + + +class _FillsOnlyRaisingRepo: + """Fills write raises; the snapshot write must still happen (independent guards).""" + def __init__(self) -> None: + self.snapshot_calls: list[tuple] = [] + + def record_fills(self, **kwargs): + raise RuntimeError("fills write failed") + + def replace_snapshot(self, *args, **kwargs): + self.snapshot_calls.append((args, kwargs)) + + +def test_persist_execution_snapshot_write_survives_a_fills_failure(): + repo = _FillsOnlyRaisingRepo() + persist_execution(repo, strategy_id="multistrat", rebalance_id="x", fills=[], + snapshot=AccountSnapshot(nlv=1.0, cash=1.0, gross=1.0, positions={}), + at=dt.datetime(2026, 7, 14)) + assert len(repo.snapshot_calls) == 1 + + +# ---- broker: account_snapshot + trade collection ------------------------------------------------------- + +def test_ibkr_broker_account_snapshot_reads_summary_and_positions(): + from fxhnt.adapters.broker.ibkr import IbkrBroker + + broker = IbkrBroker() + fake_ib = SimpleNamespace( + isConnected=lambda: True, + accountSummary=lambda: [SimpleNamespace(tag="NetLiquidation", value="1028594.0"), + SimpleNamespace(tag="TotalCashValue", value="500.0"), + SimpleNamespace(tag="GrossPositionValue", value="1000000.0")], + positions=lambda: [SimpleNamespace(contract=SimpleNamespace(symbol="SPY"), position=10.0)]) + broker._ib = fake_ib + snap = broker.account_snapshot() + assert snap.nlv == 1028594.0 + assert snap.cash == 500.0 + assert snap.gross == 1000000.0 + assert snap.positions == {"SPY": 10.0} + + +def test_ibkr_broker_pop_trades_returns_and_clears_placed_trades(): + from fxhnt.adapters.broker.ibkr import IbkrBroker + + broker = IbkrBroker() + assert broker.pop_trades() == [] # nothing placed yet + fake_trade = _FakeTrade([_FakeFill("SPY", "BOT", 1.0, 100.0, 100.0, 0.1)]) + broker._trades.append(fake_trade) + popped = broker.pop_trades() + assert popped == [fake_trade] + assert broker.pop_trades() == [] # cleared after pop + + +# ---- plan_and_record wiring: capture only fires when account_repo + broker are supplied ------------------ + +class _FakePlanWithTrades: + def __init__(self, trades): + self.nlv = 100_000.0 + self.orders: list = [] + self.notes: list = [] + self.blocked = False + self.executed = True + self.trades = trades + self.rebalance_id = "" + + +class _FakeExecSvcWithTrades: + def __init__(self, trades): + self._trades = trades + + def rebalance_weights(self, name, weights, prices, *, max_gross, execute): + return _FakePlanWithTrades(self._trades) + + +class _FakeBrokerWithSnapshot: + def account_snapshot(self): + return AccountSnapshot(nlv=100_000.0, cash=1_000.0, gross=99_000.0, positions={"SPY": 12.0}) + + +def test_plan_and_record_persists_capture_when_repo_and_broker_supplied(tmp_path): + from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo + from fxhnt.application.multistrat_exec_record import plan_and_record + + fwd_repo = ForwardNavRepo(f"sqlite:///{tmp_path}/f.db") + fwd_repo.migrate() + account_repo = IbkrAccountRepo(f"sqlite:///{tmp_path}/a.db") + account_repo.migrate() + at = dt.datetime(2026, 7, 14, 15, 0) + trade = _FakeTrade([_FakeFill("SPY", "BOT", 12.0, 500.0, 500.0, 1.0)]) + + plan, env = plan_and_record( + repo=fwd_repo, exec_svc=_FakeExecSvcWithTrades([trade]), within_weights={"SPY": 1.0}, + prices={"SPY": 500.0}, nlv=100_000.0, today="2026-07-14", at=at, max_gross=2.0, execute=True, + account_is_paper=True, account_repo=account_repo, broker=_FakeBrokerWithSnapshot()) + + stored = account_repo.fills_for("multistrat", limit=10) + assert len(stored) == 1 and stored[0].symbol == "SPY" + assert account_repo.nav_series() == [("2026-07-14", 100_000.0)] + assert not plan.blocked + + +def test_plan_and_record_skips_capture_when_repo_and_broker_not_supplied(tmp_path): + """Existing callers (and `_FakePlan`-style test fakes without a `.trades` attribute) must be unaffected — + no capture attempted, no AttributeError, when account_repo/broker are the default None.""" + from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo + from fxhnt.application.multistrat_exec_record import plan_and_record + + class _PlanNoTrades: + nlv = 100_000.0 + orders: list = [] + notes: list = [] + blocked = False + executed = True + + class _ExecSvcNoTrades: + def rebalance_weights(self, name, weights, prices, *, max_gross, execute): + return _PlanNoTrades() + + fwd_repo = ForwardNavRepo(f"sqlite:///{tmp_path}/f2.db") + fwd_repo.migrate() + at = dt.datetime(2026, 7, 14, 15, 0) + plan, env = plan_and_record( + repo=fwd_repo, exec_svc=_ExecSvcNoTrades(), within_weights={"SPY": 1.0}, prices={"SPY": 500.0}, + nlv=100_000.0, today="2026-07-14", at=at, max_gross=2.0, execute=True, account_is_paper=True) + assert not plan.blocked # no AttributeError from touching plan.trades/plan.rebalance_id diff --git a/tests/unit/test_ibkr_account_repo.py b/tests/unit/test_ibkr_account_repo.py new file mode 100644 index 0000000..b65fbca --- /dev/null +++ b/tests/unit/test_ibkr_account_repo.py @@ -0,0 +1,71 @@ +"""D2.1 — IbkrAccountRepo: per-run REAL DU-account NAV/positions snapshot (idempotent replace) + strategy- +tagged exec_fills reads. Mirrors ForwardNavRepo's migrate/engine construction, tested against SQLite.""" +import datetime as dt + +from fxhnt.adapters.persistence.ibkr_account_repo import IbkrAccountRepo + + +def test_snapshot_roundtrip(tmp_path): + repo = IbkrAccountRepo(f"sqlite:///{tmp_path}/t.db") + repo.migrate() + at = dt.datetime(2026, 7, 14, 14, 35) + repo.replace_snapshot("2026-07-14", nlv=1_028_594.0, cash=500.0, gross=1_000_000.0, + positions={"SPY": 10.0}, at=at) + assert repo.nav_series() == [("2026-07-14", 1_028_594.0)] + assert repo.latest_positions() == {"SPY": 10.0} + # idempotent replace + repo.replace_snapshot("2026-07-14", nlv=1_029_000.0, cash=1.0, gross=1.0, positions={"SPY": 11.0}, at=at) + assert repo.nav_series() == [("2026-07-14", 1_029_000.0)] + + +def test_nav_series_multi_date_ascending(tmp_path): + repo = IbkrAccountRepo(f"sqlite:///{tmp_path}/t2.db") + repo.migrate() + at = dt.datetime(2026, 7, 13, 10, 0) + repo.replace_snapshot("2026-07-13", nlv=1_000_000.0, cash=0.0, gross=1_000_000.0, positions={}, at=at) + repo.replace_snapshot("2026-07-14", nlv=1_010_000.0, cash=0.0, gross=1_010_000.0, positions={}, at=at) + assert repo.nav_series() == [("2026-07-13", 1_000_000.0), ("2026-07-14", 1_010_000.0)] + + +def test_latest_positions_empty_when_no_snapshot(tmp_path): + repo = IbkrAccountRepo(f"sqlite:///{tmp_path}/t3.db") + repo.migrate() + assert repo.nav_series() == [] + assert repo.latest_positions() == {} + + +def test_fills_for_reads_strategy_tagged_exec_fills(tmp_path): + repo = IbkrAccountRepo(f"sqlite:///{tmp_path}/t4.db") + repo.migrate() + at = dt.datetime(2026, 7, 14, 15, 0) + from fxhnt.application.exec_capture import FillDTO + fills = [FillDTO(symbol="SPY", side="BOT", qty=10.0, price=452.3, fee=1.25, status="Filled", + shortfall_bps=0.0)] + repo.record_fills(strategy_id="multistrat", rebalance_id="2026-07-14", fills=fills, at=at) + out = repo.fills_for("multistrat", limit=10) + assert len(out) == 1 + assert out[0].symbol == "SPY" and out[0].strategy_id == "multistrat" + assert repo.fills_for("other_strategy", limit=10) == [] + + +def test_migrate_is_idempotent_and_adds_strategy_id_to_existing_exec_fills(tmp_path): + """A pre-existing exec_fills table (created before this task, no strategy_id column) must gain the + column on migrate without losing data — the guarded ALTER path.""" + dsn = f"sqlite:///{tmp_path}/t5.db" + from sqlalchemy import create_engine, text + engine = create_engine(dsn) + with engine.begin() as c: + c.execute(text( + "CREATE TABLE exec_fills (client_id VARCHAR(64) PRIMARY KEY, rebalance_id VARCHAR(48), " + "symbol VARCHAR(32), side VARCHAR(4), qty FLOAT, price FLOAT, fee FLOAT, status VARCHAR(12), " + "shortfall_bps FLOAT, at DATETIME)")) + c.execute(text( + "INSERT INTO exec_fills VALUES ('c1', 'r1', 'SPY', 'BOT', 1.0, 2.0, 0.0, 'Filled', 0.0, " + "'2026-07-14 00:00:00')")) + engine.dispose() + + repo = IbkrAccountRepo(dsn) + repo.migrate() # must not raise, and must add strategy_id without dropping the pre-existing row + assert repo.fills_for("", limit=10)[0].symbol == "SPY" # pre-existing row defaults to "" + + repo.migrate() # second migrate is a no-op