Conciliación y controles de riesgo para una cesta de acciones en Interactive Brokers

# Interactive Brokers Basket Rebalance Demo


# Interactive Brokers Basket Rebalance Demo

**Chapter 25: Live Trading Systems**
**Section Reference**: 25.2 (Integrating with Interactive Brokers)

**Docker image**: `ml4t` (requires IB TWS/Gateway running on host port 7497)

This notebook extends `03_ib_paper_trading_demo` from a single-order workflow to a full daily-rebalance
workflow against a fixed 20-name US large-cap universe. Publication execution is planning-only;
the opt-in paper path demonstrates three mechanics that a single-order
demo cannot:

1. **Reconciliation.** Compare current IB positions against model-target positions and compute the
   order basket required to close the gap. In production this is the *only* place position drift gets
   corrected, and without it every divergence compounds.
2. **Basket planning and optional submission.** Build the complete delta basket. Only an explicitly
   authorized paper run routes it concurrently through `SafeBroker`.
3. **Post-fill attribution contract.** The optional paper path compares fills with the last completed
   close. Publication mode reports planned notional and makes no broker mutation.

**Universe**: 20 fixed US large-cap names (AAPL, MSFT, GOOGL, AMZN, META, NVDA, JPM, V, JNJ, PG, UNH, HD,
DIS, MA, BAC, KO, PEP, XOM, CVX, WMT).

**Cross-References**
- Chapter 25.2: Interactive Brokers integration overview
- Chapter 25.5: Order lifecycle management (state machine used by `SafeBroker`)
- Chapter 25.7: Operational readiness (kill switch, risk limits wrapped around the basket)

**Learning Objectives**
- Diff current broker positions against model targets and express the gap as a concrete order list.
- Inspect a no-mutation order plan with per-order risk inputs intact.
- Understand how an explicitly authorized paper run reconciles fills and estimates execution cost.

**Prerequisites**
- Chapter 25.2 for IB connectivity; Chapter 25.5 for the order state machine.
- IB TWS or Gateway running with paper-account API access enabled. This is required: the
  notebook fails loudly when no session is reachable and never substitutes a mock.

## 1. Setup and Configuration

The parameters cell fixes the deployment surface: which host/port to reach, how much notional to deploy
per rebalance, and how many names the long leg spans. Production overrides flow through Papermill.

```python
"""IB Basket Rebalance Demo: daily rebalance of a 20-name US large-cap universe via IB paper."""

import asyncio
import hashlib
import json
import logging
import warnings
from datetime import UTC, datetime

import matplotlib.pyplot as plt
import polars as pl
from async_utils import run_async
from ib_async import Stock
from ml4t.backtest import OrderSide, OrderType
from ml4t.live import LiveRiskConfig, SafeBroker
from ml4t.live.brokers.ib import IBBroker

from utils.paths import display_path, get_output_dir
from utils.style import COLORS, add_message_title, show_with_alt


def run_demo(awaitable):
    """Run an async demo while suppressing only nest_asyncio's Python 3.14 deprecation."""
    with warnings.catch_warnings():
        warnings.filterwarnings("ignore", category=DeprecationWarning, module=r"nest_asyncio")
        return run_async(awaitable)
```

`MARKET_DATA_TYPE` decides which quote TWS returns. Leaving it at `None` keeps whatever the
session is configured for, which is the right choice when the paper account carries live Level 1
subscriptions. A paper account without them rejects MARKET orders with "No market data
available...", so this notebook asks for delayed quotes instead. The four values TWS accepts are
1 for real time, 2 for frozen, 3 for delayed and 4 for delayed frozen.

```python
IB_HOST = "127.0.0.1"
IB_PORT = 7497
CLIENT_ID = 12
ACCOUNT = ""
UNIVERSE = [
    "AAPL",
    "MSFT",
    "GOOGL",
    "AMZN",
    "META",
    "NVDA",
    "JPM",
    "V",
    "JNJ",
    "PG",
    "UNH",
    "HD",
    "DIS",
    "MA",
    "BAC",
    "KO",
    "PEP",
    "XOM",
    "CVX",
    "WMT",
]
TOP_K_LONG = 5
TARGET_NOTIONAL_USD = 50_000
MAX_POSITION_USD = 15_000
MAX_ORDER_USD = 12_000
MAX_DAILY_LOSS_USD = 5_000
WARMUP_DAYS = 60
OUTSIDE_RTH = False  # set True to permit extended-hours execution on market orders
MARKET_DATA_TYPE: int | None = 3  # delayed quotes; see above
SUBMIT_PAPER_ORDERS = False  # explicit opt-in only; publication execution is read-only
```

```python
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("basket_demo")
logging.getLogger("ib_async").setLevel(logging.WARNING)
logging.getLogger("ml4t.live.brokers.ib").setLevel(logging.WARNING)

print("=" * 60)
print("IB BASKET REBALANCE DEMO")
print("=" * 60)
print(f"Universe: {len(UNIVERSE)} names ({', '.join(UNIVERSE[:5])}, ...)")
print(f"Long leg: top {TOP_K_LONG} by signal")
print(f"Target notional per leg: ${TARGET_NOTIONAL_USD:,}")
print(f"Position cap: ${MAX_POSITION_USD:,}  |  Order cap: ${MAX_ORDER_USD:,}")
print(f"Host/port: {IB_HOST}:{IB_PORT}  client_id={CLIENT_ID}")
print(f"Order submission: {'ENABLED' if SUBMIT_PAPER_ORDERS else 'DISABLED (planning only)'}")
```

The configuration banner makes the deployment envelope explicit. Position and order caps are
risk controls rather than guidelines: `SafeBroker` enforces them, so exceeding one raises before
the order reaches IB.

## 2. Connect to IB Paper Session

The notebook requires a reachable IB paper session and has no silent fallback. If TWS or IB Gateway
is not running on the configured host and port, the cell prints actionable setup instructions and exits.
Demos that quietly substitute mock state when the real broker is missing are inconsistent with the
operational discipline this chapter is teaching.

```python
OUTPUT_DIR = get_output_dir(25, "ib_basket_rebalance_demo")
STATE_FILE = OUTPUT_DIR / "risk_state.json"
STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
```

The connection gate rejects unreachable services and any managed account that is not an IB paper account.

```python
async def open_ib_session() -> IBBroker:
    """Connect to the IB paper session. Raises on failure with actionable instructions."""
    broker = IBBroker(
        host=IB_HOST,
        port=IB_PORT,
        client_id=CLIENT_ID,
        account=ACCOUNT or None,
        market_data_type=MARKET_DATA_TYPE,
    )
    try:
        await asyncio.wait_for(broker.connect(), timeout=10.0)
    except (TimeoutError, ConnectionRefusedError, OSError) as exc:
        print()
        print("=" * 60)
        print("ERROR: IB paper session unreachable")
        print("=" * 60)
        print(f"Could not connect to {IB_HOST}:{IB_PORT} (client_id={CLIENT_ID})")
        print(f"Underlying error: {type(exc).__name__}: {exc}")
        print()
        print("Setup checklist:")
        print("  1. Start TWS or IB Gateway and log into a paper account.")
        print("  2. Edit -> Global Configuration -> API -> Settings:")
        print("     - 'Enable ActiveX and Socket Clients' must be checked.")
        print(f"     - Socket port must be {IB_PORT} (TWS paper=7497, Gateway paper=4002).")
        print("     - 127.0.0.1 must be in 'Trusted IPs', or unchecked 'Read-Only API'.")
        print(f"  3. Confirm no other client is using client_id={CLIENT_ID}.")
        print()
        print("Re-run this notebook once TWS is reachable.")
        raise SystemExit(1) from exc
    accounts = [str(account) for account in broker.ib.managedAccounts()]
    if not accounts or any(not account.upper().startswith("DU") for account in accounts):
        await broker.disconnect()
        raise RuntimeError(
            "Refusing non-paper IB session: every managed account must start with 'DU'"
        )
    logger.info("Connected to IB paper session at %s:%s", IB_HOST, IB_PORT)
    return broker
```

```python
broker = run_demo(open_ib_session())
print(f"\nVerified IB paper session connected at {IB_HOST}:{IB_PORT}")
```

Reaching this line means the paper session is real. Every subsequent order, position
snapshot, and reconciliation diff reflects the broker's authoritative state, not an in-process
simulation.

## 3. Wrap in SafeBroker and Reconcile Persisted State

`SafeBroker` enforces position and order caps, persists kill-switch state across runs, and on
`connect()` diffs the persisted snapshot from the previous session against the broker's current
positions and pending orders. A non-clean report means something changed between sessions
(uncleared after-hours order, manual close, partial fill landed after the last persist), and the
notebook must stop and let the operator investigate before submitting a basket.

```python
risk_config = LiveRiskConfig(
    max_position_value=MAX_POSITION_USD,
    max_order_value=MAX_ORDER_USD,
    max_daily_loss=MAX_DAILY_LOSS_USD,
    max_data_staleness_seconds=120,
    shadow_mode=not SUBMIT_PAPER_ORDERS,
    state_file=str(STATE_FILE),
)
safe_broker = SafeBroker(broker, risk_config)
run_demo(safe_broker.connect())

report = safe_broker.reconciliation_report
if report is None or not report["clean"]:
    print()
    print("=" * 60)
    print("ERROR: Startup reconciliation is not clean")
    print("=" * 60)
    print(f"State file: {STATE_FILE}")
    if report is not None:
        for key in (
            "missing_positions",
            "unexpected_positions",
            "quantity_mismatches",
            "missing_pending_orders",
            "unexpected_pending_orders",
        ):
            value = report.get(key)
            if value:
                print(f"  {key}: {value}")
    print()
    print("Investigate the divergence in TWS before re-running the basket loop.")
    print(f"If the divergence is expected (manual flatten, etc.), delete {STATE_FILE} to reset.")
    raise SystemExit(1)

print(
    f"SafeBroker configured with max_position=${MAX_POSITION_USD:,}, "
    f"max_order=${MAX_ORDER_USD:,}, max_daily_loss=${MAX_DAILY_LOSS_USD:,}"
)
print(f"Reconciliation: clean (state file {display_path(STATE_FILE)})")
```

`SafeBroker.connect()` is the single place where stale-session damage gets caught.
A previous run that left orders pending after-hours, a manual flatten in the GUI, or a partial fill
that landed after the last persist all show up as a non-clean reconciliation report, and the
notebook refuses to launch the basket until the operator resolves the divergence. The position and order
caps that follow apply uniformly because every leg flows through `safe_broker`.

## 4. Fetch Warmup Bars from IB

The rebalance ranks names against a 20-day momentum signal, so the notebook needs roughly three months
of recent daily history. A live-trading notebook must source warmup bars from the same broker session
that will execute the orders. Using a research-time loader risks ranking names on stale prices and
computing position sizes against historical levels that no longer reflect the live tape. We pull
`WARMUP_DAYS` of daily bars per universe symbol via `reqHistoricalDataAsync`, parallelising the 20
requests through `asyncio.gather`.

```python
IB_BAR_AUDIT: list[dict] = []
```

Raw IB bars are hashed and counted before parsing so the broker-data boundary is reproducible.

```python
def audit_ib_bars(symbol: str, bars: list) -> dict:
    """Return raw-record identity and duplicate-date diagnostics."""
    canonical = [
        {
            "date": str(bar.date),
            "open": float(bar.open),
            "high": float(bar.high),
            "low": float(bar.low),
            "close": float(bar.close),
            "volume": float(bar.volume or 0),
        }
        for bar in bars
    ]
    dates = [row["date"] for row in canonical]
    payload = json.dumps(canonical, separators=(",", ":"), sort_keys=True).encode()
    return {
        "symbol": symbol,
        "payload_sha256": hashlib.sha256(payload).hexdigest(),
        "raw_rows": len(canonical),
        "duplicate_dates": len(dates) - len(set(dates)),
    }
```

Each symbol parser preserves identity and converts one raw IB bar to one canonical row.

```python
async def fetch_one_symbol_bars(ib_app: object, symbol: str, days: int) -> list[dict]:
    """Return daily OHLCV bars for `symbol` from IB. Returns [] when IB has no data for the contract."""
    contract = Stock(symbol, "SMART", "USD")
    qualified = await ib_app.qualifyContractsAsync(contract)
    if not qualified:
        logger.warning("IB could not qualify %s; skipping", symbol)
        return []
    raw_bars = await ib_app.reqHistoricalDataAsync(
        qualified[0],
        endDateTime="",
        durationStr=f"{days} D",
        barSizeSetting="1 day",
        whatToShow="TRADES",
        useRTH=True,
        # `timeout=0` disables ib_async's internal `asyncio.wait_for` wrapper,
        # which fails under nest_asyncio on Python 3.14. The IB request keeps
        # its own network timeout.
        timeout=0,
    )
    audit = audit_ib_bars(symbol, raw_bars)
    rows = []
    for bar in raw_bars:
        ts = bar.date
        if not isinstance(ts, datetime):
            ts = datetime.combine(ts, datetime.min.time())
        rows.append(
            {
                "symbol": symbol,
                "timestamp": ts,
                "open": float(bar.open),
                "high": float(bar.high),
                "low": float(bar.low),
                "close": float(bar.close),
                "volume": float(bar.volume or 0),
            }
        )
    audit["parsed_rows"] = len(rows)
    IB_BAR_AUDIT.append(audit)
    return rows
```

Fan the per-symbol fetch out across the universe via `asyncio.gather`,
then concatenate the rows into a single OHLCV panel for downstream use.

```python
async def fetch_warmup_bars(ib_app: object, universe: list[str], days: int) -> pl.DataFrame:
    """Pull daily bars for every name in `universe` from IB in parallel."""
    per_symbol = await asyncio.gather(*[fetch_one_symbol_bars(ib_app, s, days) for s in universe])
    rows = [r for symbol_rows in per_symbol for r in symbol_rows]
    if not rows:
        raise RuntimeError(
            "IB returned zero bars for the entire universe; check market-data subscriptions"
        )
    return pl.DataFrame(rows).sort(["symbol", "timestamp"])
```

```python
bars = run_demo(fetch_warmup_bars(broker.ib, UNIVERSE, WARMUP_DAYS))
ib_audit_frame = pl.DataFrame(IB_BAR_AUDIT)
if len(ib_audit_frame) == 0:
    raise RuntimeError("No IB bar payload reached the parser audit")
assert (ib_audit_frame["raw_rows"] == ib_audit_frame["parsed_rows"]).all()
assert ib_audit_frame["duplicate_dates"].sum() == 0
```

The decision panel excludes the current UTC date and reconciles that exclusion against raw counts.

```python
bars = bars.filter(pl.col("timestamp").dt.date() < datetime.now(UTC).date())
if len(bars) == 0:
    raise RuntimeError("IB returned no completed daily bars before the current UTC date")

completed_counts = bars.group_by("symbol").len().rename({"len": "completed_rows"})
ib_audit_frame = (
    ib_audit_frame.join(completed_counts, on="symbol", how="left")
    .with_columns(pl.col("completed_rows").fill_null(0))
    .with_columns(current_or_future_rows_excluded=pl.col("parsed_rows") - pl.col("completed_rows"))
)
assert (ib_audit_frame["current_or_future_rows_excluded"] >= 0).all()
print(
    ib_audit_frame.select(
        "symbol",
        "payload_sha256",
        "raw_rows",
        "completed_rows",
        "current_or_future_rows_excluded",
    )
)

print(
    f"\nWarmup bars: {len(bars):,} rows over {bars['symbol'].n_unique()} symbols (last close: {bars['timestamp'].max().date()})"
)
latest_close = (
    bars.group_by("symbol")
    .agg(
        pl.col("close").last().alias("last_close"),
        pl.col("timestamp").last().alias("timestamp"),
    )
    .sort("symbol")
)
print(f"Latest close table: {len(latest_close)} symbols")
```

## 5. Signal Computation and Target Basket

The signal is a simple 20-day momentum proxy, adequate for a live-mechanics demonstration even though
Chapter 6 establishes that naïve momentum on its own is insufficient as a research-grade factor. The
target basket longs the top `TOP_K_LONG` by signal and leaves everything else flat.

```python
def compute_target_basket(bars_frame: pl.DataFrame, top_k: int, notional: float) -> pl.DataFrame:
    """Return a frame with `symbol`, `signal`, `target_qty` for the current rebalance."""
    log_close = pl.col("close").log()
    with_momentum = bars_frame.with_columns(
        momentum_20d=(log_close - log_close.shift(20)).over("symbol"),
        last_close=pl.col("close").last().over("symbol"),
    )
    latest = (
        with_momentum.group_by("symbol", maintain_order=True)
        .agg(
            pl.col("momentum_20d").last().alias("signal"),
            pl.col("last_close").last().alias("last_close"),
        )
        .sort("signal", descending=True)
    )
    longs = latest.head(top_k).with_columns(
        target_notional=pl.lit(notional / top_k),
        target_qty=(pl.lit(notional / top_k) / pl.col("last_close")).round(0).cast(pl.Float64),
    )
    flats = latest.tail(len(latest) - top_k).with_columns(
        target_notional=pl.lit(0.0), target_qty=pl.lit(0.0)
    )
    return pl.concat([longs, flats], how="vertical").sort("signal", descending=True)


target = compute_target_basket(bars, TOP_K_LONG, TARGET_NOTIONAL_USD)
print("\nTarget basket (top 5 by signal):")
target.head(TOP_K_LONG)
```

The ranked momentum view shows exactly why each name enters or misses the
target basket before any broker mutation is possible.

```python
plot_target = target.sort("signal")
target_symbols = set(target.head(TOP_K_LONG)["symbol"].to_list())
bar_colors = [
    COLORS["positive"] if symbol in target_symbols else COLORS["neutral"]
    for symbol in plot_target["symbol"].to_list()
]
fig, ax = plt.subplots(figsize=(9, 6))
ax.barh(
    plot_target["symbol"].to_list(),
    plot_target["signal"].to_list(),
    color=bar_colors,
)
ax.axvline(0, linestyle="--", color=COLORS["neutral"], linewidth=1)
ax.set(xlabel="20-day log return (decimal)", ylabel="US equity")
add_message_title(
    ax,
    "The strongest 20-day momentum names form the target basket",
    subtitle="Completed IB daily bars; green identifies the planned long leg",
)
show_with_alt(
    fig,
    "Horizontal bar chart of the 20-day log return for each name in the universe, sorted by "
    "signal, with a dashed line at zero. Colour marks membership: the names taken as the "
    "long leg of the target basket are drawn in one colour and the names left flat in another.",
)
```

Printing the head of the target basket before any order is sent makes the intended
portfolio state auditable. In production the same table is logged and stored as the rebalance intent,
so that post-fill state can be compared against it.

## 6. Reconciliation: Current Positions versus Target

Reconciliation is the step that distinguishes a real live-trading loop from a fire-and-forget signal
pipeline. The `current` frame comes from the broker; the `target` frame comes from the model. The
`delta` frame is the order basket required to move from one to the other.

```python
async def fetch_current_positions(active_broker: object, universe: list[str]) -> pl.DataFrame:
    """Return current broker positions as a frame keyed by symbol, with every universe name represented."""
    raw = await active_broker.get_positions_async()
    rows = []
    for sym in universe:
        pos = raw.get(sym)
        rows.append(
            {
                "symbol": sym,
                "current_qty": float(pos.quantity) if pos else 0.0,
                "entry_price": float(pos.entry_price) if pos else 0.0,
            }
        )
    return pl.DataFrame(rows)


current_positions = run_demo(fetch_current_positions(broker, UNIVERSE))
```

`reconcile` joins current and target positions and emits a `delta_qty`
column, the per-symbol order instruction that drives the basket
submission step.

```python
def reconcile(current: pl.DataFrame, target_frame: pl.DataFrame) -> pl.DataFrame:
    """Join current and target positions, emit a `delta_qty` column as the order instruction per symbol."""
    joined = current.join(target_frame, on="symbol", how="full", coalesce=True).with_columns(
        pl.col("target_qty").fill_null(0.0),
        pl.col("current_qty").fill_null(0.0),
    )
    return joined.with_columns(
        delta_qty=pl.col("target_qty") - pl.col("current_qty"),
    ).filter(pl.col("delta_qty").abs() > 0)


orders_needed = reconcile(current_positions, target)
print(f"\nReconciliation: {len(orders_needed)} orders needed to reach target")
print(orders_needed.select(["symbol", "current_qty", "target_qty", "delta_qty", "last_close"]))
```

The delta table is the audit surface. Each row justifies exactly one order; rows that
vanish (delta already zero) are implicit parity confirmations. A production run would checkpoint this
table before any submission, so a mid-rebalance crash can resume from the same intent.

## 6b. Cache Reference Prices for SafeBroker

`SafeBroker.submit_order_async` rejects orders when no recent price snapshot is cached for the asset.
We seed the cache with the latest completed daily close per symbol, using the same
`reqHistoricalDataAsync` data that fed the signal in section 4. The notebook excludes the current
UTC date before selecting these observations, so an unfinished daily bar cannot leak into sizing.

A continuous-loop deployment would replace this seed with a streaming `IBDataFeed` driven by
`LiveEngine`, so every rebalance is preceded by a continuously refreshed quote. For this one-shot
planning demonstration, the completed close provides a reproducible reference for the safety checks.

```python
for row in latest_close.iter_rows(named=True):
    safe_broker.record_market_snapshot(
        row["symbol"], float(row["last_close"]), timestamp=row["timestamp"]
    )
print(f"Cached {len(latest_close)} IB-sourced reference prices for SafeBroker staleness guard")
```

## 7. Basket Submission

Submitting the basket through `asyncio.gather` parallelises I/O without changing the per-order
semantics. Every leg still flows through `SafeBroker`, so caps and the kill switch apply uniformly.

```python
def dry_run_basket(basket: pl.DataFrame) -> list[dict]:
    """Convert order deltas to auditable intents without broker calls."""
    return [
        {
            "symbol": row["symbol"],
            "side": "buy" if row["delta_qty"] > 0 else "sell",
            "quantity": abs(row["delta_qty"]),
            "delta_qty": row["delta_qty"],
            "order_id": None,
            "last_close": row["last_close"],
            "status": "dry_run",
        }
        for row in basket.to_dicts()
    ]
```

One submitted leg carries its close as a risk-check hint; IB still receives a market order.

```python
async def submit_one_basket_order(active_broker: SafeBroker, row: dict) -> dict:
    """Submit one delta through `SafeBroker` and return its audit record."""
    qty = abs(row["delta_qty"])
    side = OrderSide.BUY if row["delta_qty"] > 0 else OrderSide.SELL
    record = {
        "symbol": row["symbol"],
        "side": side.value,
        "quantity": qty,
        "delta_qty": row["delta_qty"],
        "order_id": None,
        "last_close": row["last_close"],
    }
    try:
        order = await active_broker.submit_order_async(
            asset=row["symbol"],
            quantity=qty,
            side=side,
            order_type=OrderType.MARKET,
            limit_price=float(row["last_close"]),
            outsideRth=OUTSIDE_RTH,
        )
        record.update(status="submitted", order_id=getattr(order, "order_id", None))
    except Exception as exc:  # noqa: BLE001
        record["status"] = f"rejected: {exc}"
    return record
```

Basket submission is an explicit switch: publication mode returns intents, while an authorized paper
run submits all legs concurrently through the risk wrapper.

```python
async def submit_basket(active_broker: SafeBroker, basket: pl.DataFrame) -> list[dict]:
    """Plan or submit every non-zero rebalance delta."""
    if not SUBMIT_PAPER_ORDERS:
        return dry_run_basket(basket)
    tasks = [submit_one_basket_order(active_broker, row) for row in basket.to_dicts()]
    return await asyncio.gather(*tasks)
```

```python
submissions = run_demo(submit_basket(safe_broker, orders_needed))
submissions_frame = pl.DataFrame(submissions) if submissions else pl.DataFrame()
if not SUBMIT_PAPER_ORDERS:
    assert all(record["status"] == "dry_run" for record in submissions)
label = "submitted" if SUBMIT_PAPER_ORDERS else "planned"
print(f"\nBasket {label}: {len(submissions_frame)} orders")
if len(submissions_frame):
    print(submissions_frame)
```

The fills frame is the first place an execution problem becomes visible. A row with a `rejected`
status, or a large gap between `fill_price` and `last_close`, is something the operator has to
investigate before the next rebalance.

## 8. Wait for Fills and Reconcile Post-Submission State

`submit_order_async` returns as soon as the order is queued; fills arrive asynchronously via IB
callbacks. A short sleep lets routing complete, after which re-fetching positions reflects the
executed basket. A residual delta after this wait means an unfilled or rejected order, which is
what the operator needs to see before the next rebalance cycle.

```python
async def wait_for_basket_to_settle(
    active_broker: object, expected_symbols: list[str], timeout_s: float = 15.0
) -> dict:
    """Poll the broker for position updates until every expected symbol is reflected, or timeout."""
    deadline = asyncio.get_event_loop().time() + timeout_s
    while asyncio.get_event_loop().time() < deadline:
        if hasattr(active_broker, "_sync_positions"):
            await active_broker._sync_positions()
        positions = await active_broker.get_positions_async()
        present = {s for s, p in positions.items() if p.quantity != 0}
        missing = [s for s in expected_symbols if s not in present]
        if not missing:
            return positions
        await asyncio.sleep(1.0)
    return await active_broker.get_positions_async()


if SUBMIT_PAPER_ORDERS:
    expected_long = orders_needed.filter(pl.col("delta_qty") > 0)["symbol"].to_list()
    raw_post = run_demo(wait_for_basket_to_settle(broker, expected_long, timeout_s=15.0))
    post_positions = run_demo(fetch_current_positions(broker, UNIVERSE))
    post_delta = reconcile(post_positions, target)
    print(f"\nPost-submission residual deltas: {len(post_delta)}")
    if len(post_delta):
        print(post_delta.select(["symbol", "current_qty", "target_qty", "delta_qty"]))
    else:
        print("All positions reconciled to target.")
else:
    raw_post = run_demo(broker.get_positions_async())
    post_positions = current_positions
    post_delta = orders_needed
    print("\nPost-submission reconciliation skipped: planning mode made no broker mutations.")
```

**Finding:** A non-empty post-submission residual is not automatically a bug. It can also mean a
kill-switch or risk-cap triggered mid-basket and blocked a leg. Distinguishing the two requires the
per-order status from step 7, which is why both are printed side by side.

## 9. Execution Cost and Expected-versus-Realised P\&L

The final check estimates execution cost per leg and aggregates it against the notional deployed. In
production this number feeds Chapter 26 (MLOps and Governance): sustained drift between `fill_price`
and `last_close` is the first signal that venue or router assumptions have changed.

```python
def execution_summary(submissions: pl.DataFrame, raw_positions: dict) -> dict:
    """Cross-reference submitted orders with post-fill positions and report slippage vs last_close.

    For a new long position, the post-fill `entry_price` is the weighted-average fill price for that
    symbol. We compare it to the `last_close` at submission time to estimate intraday slippage in bps.
    """
    if not len(submissions):
        return {"orders": 0, "notional": 0.0, "slippage_bps": 0.0, "buys": 0, "sells": 0}

    def fill_price_for(symbol: str) -> float | None:
        pos = raw_positions.get(symbol)
        return float(pos.entry_price) if pos and pos.quantity != 0 else None

    enriched = submissions.with_columns(
        fill_price=pl.col("symbol").map_elements(fill_price_for, return_dtype=pl.Float64),
    ).with_columns(
        slippage_bps=(
            (pl.col("fill_price") - pl.col("last_close")) / pl.col("last_close") * 10_000.0
        ).cast(pl.Float64),
        notional=(pl.col("fill_price") * pl.col("quantity")).cast(pl.Float64),
    )
    filled = enriched.filter(pl.col("fill_price").is_not_null())
    total_notional = float(filled["notional"].sum()) if len(filled) else 0.0
    vwap_slip = (
        float((filled["slippage_bps"] * filled["notional"]).sum() / total_notional)
        if total_notional
        else 0.0
    )
    return {
        "orders_submitted": len(submissions),
        "orders_filled": len(filled),
        "notional": total_notional,
        "slippage_bps_vwap": vwap_slip,
        "buys": int((submissions["side"] == "buy").sum()),
        "sells": int((submissions["side"] == "sell").sum()),
    }
```

```python
if SUBMIT_PAPER_ORDERS:
    summary = execution_summary(submissions_frame, raw_post)
else:
    summary = {
        "orders_planned": len(submissions_frame),
        "broker_mutations": 0,
        "planned_notional": (
            float((submissions_frame["quantity"] * submissions_frame["last_close"]).sum())
            if len(submissions_frame)
            else 0.0
        ),
    }
print("\nExecution summary:")
for key, value in summary.items():
    if key in {"notional", "planned_notional"}:
        print(f"  {key:18s}: ${value:,.2f}")
    elif key == "slippage_bps_vwap":
        print(f"  {key:18s}: {value:+.2f} bps")
    else:
        print(f"  {key:18s}: {value}")
```

## 10. End-of-Day Flatten via Market-On-Close

A daily basket strategy that opens positions at start of day and re-evaluates
at the next rebalance leaves overnight exposure unless the operator explicitly
closes out. The clean primitive for this is a Market-On-Close (MOC) order:
IB queues it for the closing auction and targets the official session close.
Exchange cutoffs, halts, and broker rejection still prevent any guarantee.

This cell submits opposing-side MOC orders only for positions opened by this
notebook run. Existing account positions and orders remain outside its
mutation boundary. In a continuous
loop deployment, the same MOC step would be conditional on the next-rebalance
decision (hold overnight if the signal still ranks the name long, flatten via
MOC otherwise). The operator must use the current broker and exchange cutoff;
late orders may be rejected.

`submit_eod_flatten` issues one MOC order per position opened by this run. The
MOC primitive routes each leg to the closing auction; the side is
opposite the current position.

```python
async def submit_eod_flatten(
    active_broker: SafeBroker, symbols_with_qty: list[tuple[str, float]]
) -> list[dict]:
    """Submit one MOC order per non-zero position to flatten via the closing auction."""

    async def one_flatten(symbol: str, qty: float) -> dict:
        side = OrderSide.SELL if qty > 0 else OrderSide.BUY
        try:
            order = await active_broker.submit_order_async(
                asset=symbol,
                quantity=abs(qty),
                side=side,
                order_type=OrderType.MOC,
            )
            return {
                "symbol": symbol,
                "side": side.value,
                "quantity": abs(qty),
                "order_id": getattr(order, "order_id", None),
                "status": "submitted",
            }
        except Exception as exc:  # noqa: BLE001
            return {
                "symbol": symbol,
                "side": side.value,
                "quantity": abs(qty),
                "order_id": None,
                "status": f"rejected: {exc}",
            }

    tasks = [one_flatten(s, q) for s, q in symbols_with_qty]
    return await asyncio.gather(*tasks)
```

`confirm_orders_accepted` waits for IB to acknowledge every MOC order
before disconnect. Without this gate, a fast teardown can race the
acceptance and queued orders never reach the closing-auction queue.

```python
async def confirm_orders_accepted(
    active_ib: object, expected_count: int, timeout_s: float = 5.0
) -> list[dict]:
    """Wait until IB acknowledges all expected open orders (status != 'PendingSubmit')."""
    deadline = asyncio.get_event_loop().time() + timeout_s
    while asyncio.get_event_loop().time() < deadline:
        trades = active_ib.openTrades()
        statuses = [t.orderStatus.status for t in trades]
        if len(trades) >= expected_count and all(s != "PendingSubmit" for s in statuses):
            break
        await asyncio.sleep(0.5)
    return [
        {
            "symbol": t.contract.symbol,
            "order_type": t.order.orderType,
            "tif": t.order.tif,
            "ib_status": t.orderStatus.status,
        }
        for t in active_ib.openTrades()
    ]
```

```python
if SUBMIT_PAPER_ORDERS:
    opened_symbols = {
        row["symbol"]
        for row in submissions
        if row["status"] == "submitted" and row["delta_qty"] > 0
    }
    post_flatten_positions = run_demo(fetch_current_positions(broker, UNIVERSE))
    to_flatten = [
        (row["symbol"], row["current_qty"])
        for row in post_flatten_positions.iter_rows(named=True)
        if row["symbol"] in opened_symbols and row["current_qty"] != 0
    ]
    if to_flatten:
        flatten_records = run_demo(submit_eod_flatten(safe_broker, to_flatten))
        flatten_frame = pl.DataFrame(flatten_records)
        print(f"\nEnd-of-day flatten: submitted {len(flatten_frame)} MOC order(s)")
        print(flatten_frame)

        ack = run_demo(confirm_orders_accepted(broker.ib, expected_count=len(to_flatten)))
        print(f"\nIB acknowledgement ({len(ack)} orders open at IB):")
        for row in ack:
            print(
                f"  {row['symbol']}: orderType={row['order_type']} "
                f"tif={row['tif']} status={row['ib_status']}"
            )
    else:
        print("\nEnd-of-day flatten: this run opened no positions.")
else:
    print("\nEnd-of-day flatten skipped: planning mode created no positions or orders.")
```

**Finding:** The MOC primitive turns end-of-day flattening into a single library call
per leg. Without it, a basket strategy must either submit market orders ahead of the
16:00 ET clock or accept overnight exposure. The MOC path is explicit and
auditable, but exchange acceptance and the broker acknowledgement remain gates.

**Operator note:** TWS's `Global Configuration → API → Settings → Auto-cancel API
orders on disconnect` defaults to *enabled* on most paper accounts. With that
setting, IB cancels these MOC orders the moment the notebook disconnects, defeating
the close-out. An operator who explicitly enables paper submission must reconcile
that account-level setting before relying on queued DAY orders.

## 11. Teardown

```python
run_demo(broker.disconnect())
print(f"\nIB paper session closed at {datetime.now(UTC).isoformat(timespec='seconds')}")
```

## Key Takeaways

1. **Reconciliation is the loop.** A daily rebalance is not "submit these orders"; it is "diff current
   against target, then submit only the delta." The diff table is the audit artefact.
2. **Basket I/O is parallelised in `asyncio.gather`; order controls stay in `SafeBroker`.**
   Parallelising submission is a latency optimisation and enforcing caps is a risk control, and
   separating the two layers keeps one from overriding the other.
3. **Post-fill reconciliation is a separate step.** The same reconciliation routine used to *plan*
   the basket is used to *verify* the basket. Residual deltas name the specific failure modes.
4. **Execution cost is a monitoring signal, not a KPI.** When paper submission is enabled, drift in
   average slippage across successive rebalances is the Chapter 26 monitoring input.
5. **Programmatic close-out is part of the loop, not an operator chore.** A daily basket strategy
   that opens positions at start of day needs an explicit close-out path. `OrderType.MOC`
   targets each newly opened leg for the closing auction, subject to exchange acceptance
   and cutoff rules. Planning mode creates no exposure to flatten.
6. **Live-trading data sources come from the broker.** Warmup bars and pre-submission quotes both
   come from IB in this notebook. A research-time loader can carry a different cutoff date, a
   different survivor universe or a different vendor, so the training pipeline reads from one and
   the live execution path does not.
   Chapter 25.6's feature-parity discussion applies: when the live and training data sources differ,
   the signal ranks names the model was never validated on.

**Next:** See `10_safety_risk_demo` for the kill-switch and state-persistence behaviour that `SafeBroker`
enforces on every leg of this basket.
![notebook output](figures/p1_1.png)