Подключение стратегии моментума ETF к Interactive Brokers в теневом режиме

# Interactive Brokers Paper Trading Demo


# Interactive Brokers Paper Trading Demo

**Chapter**: 25 - Live Trading Systems
**Section Reference**: 25.2 (Integrating with Interactive Brokers)
**Learning Outcome**: LO2 - Connect ml4t-backtest strategies to live brokers

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

This notebook demonstrates:
1. Connecting to IB TWS paper trading account
2. Querying account information and positions
3. Requesting historical data for strategy warm-up
4. Real-time data feed with tick aggregation
5. Safe order submission in shadow mode

**Learning Objectives**:
- Validate IB connectivity, account state, and warm-up data before a live engine starts.
- Reuse the same backtest strategy inside a shadow-mode live workflow.
- Exercise the loud-fail setup checklist: the notebook does not synthesize live behaviour from history when TWS/Gateway is unreachable or the market is closed.

**Prerequisites**:
- IB TWS or Gateway running with API enabled

- `IB_ACCOUNT` environment variable set to your IB paper account ID (e.g. `DU1234567`); the notebook passes `None` to `IBBroker` if unset, which lets the broker pick the default account on the session.
- Familiarity with the ETF momentum example used throughout the live-trading chapter

```python
"""Connect ml4t strategies to IB with shadow-mode risk controls."""

import os
```

`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
# Production defaults. Papermill may inject overrides for CI.
IB_HOST = "127.0.0.1"
IB_PORT = 7497  # Paper trading port
CLIENT_ID = 10  # Use unique ID per notebook
ACCOUNT = os.environ.get(
    "IB_ACCOUNT"
)  # set IB_ACCOUNT=DU... before running; None picks the session default
MARKET_DATA_TYPE: int | None = 3  # delayed quotes; see above
SYMBOLS = ["SPY", "QQQ", "IWM"]  # ETFs to monitor
WARMUP_DAYS = 10  # Must provide at least lookback + 1 daily closes
LIVE_DURATION_SECONDS = 75
```

```python
import asyncio
import logging
import warnings
from datetime import datetime

os.environ.setdefault("NUMEXPR_MAX_THREADS", "16")

from async_utils import run_async

from utils.paths import display_path, get_output_dir
```

```python
# Set up logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)

# Reduce noise from ib_async
logging.getLogger("ib_async").setLevel(logging.WARNING)
logging.getLogger("ml4t.live.brokers.ib").setLevel(logging.WARNING)
```

```python
# Import ml4t.live components
from ml4t.backtest import OrderSide, Strategy
from ml4t.live import LiveEngine, LiveRiskConfig
from ml4t.live.brokers.ib import IBBroker
from ml4t.live.feeds import BarAggregator, IBDataFeed
from ml4t.live.safety import SafeBroker

print("[OK] ml4t.live components imported successfully")
```

The import check runs before any Interactive Brokers connection is attempted, so an environment
that is missing a package fails differently from a session that cannot reach TWS. Collapsing the
two into one opaque error is what makes a live deployment slow to debug.

## 1. Connect to Interactive Brokers

TWS/Gateway must be running with:
- API access enabled (Edit > Global Configuration > API > Settings)
- Socket port set (default: 7497 for paper)
- "Enable ActiveX and Socket Clients" checked

```python
# Create IBBroker instance
broker = IBBroker(
    host=IB_HOST,
    port=IB_PORT,
    client_id=CLIENT_ID,
    account=ACCOUNT,
    market_data_type=MARKET_DATA_TYPE,
)

print(f"IBBroker configured for {IB_HOST}:{IB_PORT}")
print(f"Account selection: {'configured paper account' if ACCOUNT else 'session default'}")
print(f"Client ID: {CLIENT_ID}")
```

Host, port, account and client ID are printed rather than left as constants in a cell above,
because an account or session mismatch is one of the most common live failures and is invisible
until something has already been submitted to the wrong place.


```python
async def connect_to_ib():
    """Connect to IB and show account summary."""
    print("\n" + "=" * 60)
    print("CONNECTING TO INTERACTIVE BROKERS")
    print("=" * 60)

    await broker.connect()

    account = str(broker._account or "")
    if not account.startswith("DU"):
        await broker.disconnect()
        raise RuntimeError("Refusing to continue: connected IB account is not a paper account")

    print("\n[OK] Connected to IB")
    print("   Paper account identity: [OK]")
    # Get account values
    nlv = await broker.get_account_value_async()
    cash = await broker.get_cash_async()

    print("\nACCOUNT READINESS")
    print("   Account values received: [OK]")

    # Get positions
    positions = await broker.get_positions_async()
    print(f"   Open positions received: {len(positions)}")

    return nlv, cash, positions
```

```python
# Run connection
try:
    warnings.filterwarnings("ignore", category=DeprecationWarning, module=r"nest_asyncio")
    nlv, cash, positions = run_async(connect_to_ib())
except Exception as exc:
    print()
    print("=" * 60)
    print("ERROR: IB paper session unreachable")
    print("=" * 60)
    print(f"Could not connect to {IB_HOST}:{IB_PORT}")
    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 'Read-Only API' unchecked.")
    print("  3. Confirm no other client is using this client_id.")
    print()
    print("Re-run this notebook once TWS is reachable.")
    raise RuntimeError("IB paper session unreachable") from exc
```

When TWS is unreachable the connection block prints a checklist and stops, rather than
substituting placeholder data and letting the rest of the notebook read as if it had traded.
A disconnected broker is an operational state, which is why the check comes before the signal
loop rather than inside it.

## 2. Request Historical Data

Before running a strategy live, we need historical data to:
- Calculate initial indicator values (moving averages, etc.)
- Establish baseline for position sizing
- Verify data quality

IB provides historical bars via `reqHistoricalData`.

```python
from ib_async import Stock


async def get_historical_data(symbol: str, days: int = 5) -> list[dict]:
    """Request historical daily bars from IB and return OHLCV records."""
    contract = Stock(symbol, "SMART", "USD")
    qualified = await broker.ib.qualifyContractsAsync(contract)
    if not qualified:
        raise RuntimeError(f"IB could not qualify the {symbol} contract")

    # `timeout=0` disables ib_async's internal `asyncio.wait_for` wrapper,
    # which fails under nest_asyncio on Python 3.14.
    bars = await broker.ib.reqHistoricalDataAsync(
        contract,
        endDateTime="",
        durationStr=f"{days} D",
        barSizeSetting="1 day",
        whatToShow="TRADES",
        useRTH=True,
        timeout=0,
    )

    result = []
    for bar in bars:
        result.append(
            {
                "timestamp": bar.date,
                "open": bar.open,
                "high": bar.high,
                "low": bar.low,
                "close": bar.close,
                "volume": bar.volume,
            }
        )

    if not result:
        raise RuntimeError(f"IB returned no warm-up bars for {symbol}")
    return result
```

```python
# Request historical data for our symbols
print("\n" + "=" * 60)
print("HISTORICAL DATA (Warm-up)")
print("=" * 60)

historical_data = {}
try:
    for symbol in SYMBOLS:
        warnings.filterwarnings("ignore", category=DeprecationWarning, module=r"nest_asyncio")
        bars = run_async(get_historical_data(symbol, WARMUP_DAYS))
        if len(bars) < 6:
            raise RuntimeError(f"{symbol} returned {len(bars)} bars; at least 6 are required")
        historical_data[symbol] = bars
        latest = bars[-1]
        print(f"\n{symbol}: {len(bars)} days of data")
        print(
            f"   Latest: {latest['timestamp']} - Close: ${latest['close']:.2f}, Volume: {latest['volume']:,}"
        )
except Exception:
    warnings.filterwarnings("ignore", category=DeprecationWarning, module=r"nest_asyncio")
    run_async(broker.disconnect())
    raise

assert set(historical_data) == set(SYMBOLS)
```

The warm-up prints the exact bars the indicators are initialized from. Without a deterministic
warm-up path, the first minutes after a reconnect or an open size positions from indicators that
are still filling.

## 3. Strategy Definition

This strategy is **identical** to what we use in backtesting.
The `on_data` signature works with both `ml4t.backtest.Engine` and `ml4t.live.LiveEngine`.

```python
class MomentumStrategy(Strategy):
    """Five-day ETF momentum strategy shared by backtest and live engines."""

    def __init__(self, lookback: int = 5, threshold: float = 0.02):
        self.lookback = lookback
        self.threshold = threshold
        self.prices: dict[str, list[float]] = {}
        self.signals: list[dict] = []

    def on_start(self, broker):
        """Called when engine starts."""
        logger.info(f"Strategy started: Momentum({self.lookback}, {self.threshold})")
        for symbol, bars in historical_data.items():
            self.prices[symbol] = [bar["close"] for bar in bars]
            logger.info(f"  {symbol}: Loaded {len(self.prices[symbol])} historical prices")

    def on_data(self, timestamp: datetime, data: dict, context: dict, broker):
        """Update trailing momentum and route threshold crossings."""
        for symbol, bar in data.items():
            prices = self.prices.setdefault(symbol, [])
            close = bar["close"]
            prices.append(close)
            if len(prices) <= self.lookback:
                continue
            momentum = (close - prices[-self.lookback - 1]) / prices[-self.lookback - 1]
            position = broker.get_position(symbol)
            has_position = position is not None and position.quantity > 0
            side = None
            if momentum > self.threshold and not has_position:
                side = OrderSide.BUY
            elif momentum < -self.threshold and has_position:
                side = OrderSide.SELL
            if side is None:
                continue
            action = side.value.upper()
            self.signals.append(
                {"timestamp": timestamp, "symbol": symbol, "action": action, "momentum": momentum}
            )
            logger.info(f"{action} {symbol}: momentum {momentum:.2%}")
            broker.submit_order(symbol, 100, side=side)

    def on_end(self, broker):
        """Called when engine stops."""
        logger.info(f"Strategy ended. Signals generated: {len(self.signals)}")
```

## 4. Safe Broker Configuration

Before going live, we wrap the broker with `SafeBroker` which provides:
- Shadow mode (virtual orders routed through `VirtualPortfolio`, never to IB)
- Position and order value caps, rate limiting, kill switch
- Persisted `RiskState`, so the daily-loss counter is read back after a restart
- Startup reconciliation: `safe_broker.connect()` diffs the persisted snapshot from the previous run
  against the broker's current positions and pending orders

```python
RISK_STATE_PATH = get_output_dir(25, "ib_paper_demo") / "risk_state.json"
risk_config = LiveRiskConfig(
    shadow_mode=True,  # CRITICAL: Virtual orders only!
    max_position_value=50_000.0,
    max_order_value=10_000.0,
    max_orders_per_minute=5,
    max_daily_loss=2_500.0,
    max_data_staleness_seconds=60,
    dedup_window_seconds=0.0,  # Disable for demo
    state_file=str(RISK_STATE_PATH),
)

safe_broker = SafeBroker(broker, risk_config)
warnings.filterwarnings("ignore", category=DeprecationWarning, module=r"nest_asyncio")
run_async(safe_broker.connect())

print("\n" + "=" * 60)
print("RISK CONFIGURATION (SHADOW MODE)")
print("=" * 60)
print("   Shadow Mode: [OK] ENABLED (no real orders)")
print(f"   Max Position Value: ${risk_config.max_position_value:,.0f}")
print(f"   Max Order Value: ${risk_config.max_order_value:,.0f}")
print(f"   Max Daily Loss: ${risk_config.max_daily_loss:,.0f}")
print(f"   Max Data Staleness: {risk_config.max_data_staleness_seconds}s")
print(f"   Rate Limit: {risk_config.max_orders_per_minute}/minute")
print(f"   Risk State: {display_path(RISK_STATE_PATH)}")
report = safe_broker.reconciliation_report
print(f"   Startup Reconciliation: {'clean' if report and report['clean'] else 'review report'}")
```

## 5. Real-Time Data Feed

`IBDataFeed` subscribes to live tick data from IB; `BarAggregator` rolls those
ticks into minute bars for the strategy. A live notebook is only meaningful
while the market is open. If the run starts after the close, the
notebook fails loudly rather than substituting historical bars (a backtest
wearing a live disguise).

```python
def _is_rth_now(zone: str = "America/New_York") -> bool:
    """Return True iff wall-clock now is inside US-equity regular trading hours."""
    from zoneinfo import ZoneInfo

    now = datetime.now(ZoneInfo(zone))
    if now.weekday() >= 5:
        return False
    open_t = now.replace(hour=9, minute=30, second=0, microsecond=0)
    close_t = now.replace(hour=16, minute=0, second=0, microsecond=0)
    return open_t <= now <= close_t
```

The feed, aggregator, and engine are assembled in one helper so the bounded
run below focuses only on lifecycle and cleanup.

```python
def build_live_stack() -> tuple[BarAggregator, LiveEngine]:
    """Build the IB feed, minute aggregator, and shadow execution engine."""
    ib_feed = IBDataFeed(ib=broker.ib, symbols=SYMBOLS, tick_throttle_ms=1000)
    feed = BarAggregator(source_feed=ib_feed, bar_size_minutes=1, assets=SYMBOLS)
    strategy = MomentumStrategy(lookback=5, threshold=0.02)
    engine = LiveEngine(strategy=strategy, broker=safe_broker, feed=feed)
    return feed, engine
```

A fixed-duration RTH run proves connectivity without leaving an unattended
strategy loop behind.

```python
async def run_live_demo(duration_seconds: int = 30):
    """Run strategy with live IB data for specified duration."""
    print("\n" + "=" * 60)
    print("LIVE TRADING DEMO (Shadow Mode)")
    print("=" * 60)

    if not _is_rth_now():
        print()
        print("ERROR: Market is closed. This notebook requires an open RTH session.")
        print("US equity RTH: 09:30-16:00 America/New_York, Mon-Fri.")
        print("Re-run during RTH; the notebook will not synthesize live behaviour from history.")
        raise RuntimeError("IB live-feed gate requires an open US-equity RTH session")

    feed, engine = build_live_stack()
    print(f"\nStarting live engine for {duration_seconds} seconds...")
    print("   Watching: " + ", ".join(SYMBOLS))

    engine_task: asyncio.Task | None = None
    try:
        await engine.connect()

        # Run for a fixed duration. `asyncio.wait_for` is unreliable under
        # nest_asyncio in notebook kernels, so we drive the timeout manually.
        engine_task = asyncio.create_task(engine.run())
        await asyncio.sleep(duration_seconds)
        print(f"\nDemo duration ({duration_seconds}s) reached")
    finally:
        if engine_task is not None and not engine_task.done():
            engine_task.cancel()
            try:
                await engine_task
            except asyncio.CancelledError:
                pass
        feed.stop()
        print("\nEngine Statistics:")
        for key, value in engine.stats.items():
            print(f"   {key}: {value}")


# The live workflow runs after the shadow-order routine is defined so one
# outer `finally` block can always release the IB session.
```

The strategy and the risk configuration are the same whether or not the live feed is active;
only the transport changes. Keeping that boundary is what makes the stack testable, and it means
a change of feed is never a change of trading rule.

## 6. Order Submission Demo

Let's demonstrate order submission in shadow mode.
Orders are tracked virtually but never sent to IB.

```python
async def fetch_ib_snapshot(symbol: str) -> float | None:
    """Fetch one delayed top-of-book snapshot from IB for `symbol` and return mid price."""
    contract = Stock(symbol, "SMART", "USD")
    qualified = await broker.ib.qualifyContractsAsync(contract)
    if not qualified:
        return None
    ticker = broker.ib.reqMktData(qualified[0], "", snapshot=True, regulatorySnapshot=False)
    deadline = datetime.now().timestamp() + 5.0
    while datetime.now().timestamp() < deadline:
        bid = float(ticker.bid) if ticker.bid and ticker.bid > 0 else None
        ask = float(ticker.ask) if ticker.ask and ticker.ask > 0 else None
        last = float(ticker.last) if ticker.last and ticker.last > 0 else None
        close = float(ticker.close) if ticker.close and ticker.close > 0 else None
        if bid and ask:
            return (bid + ask) / 2
        if last:
            return last
        if close:
            return close
        await asyncio.sleep(0.2)
    return None
```

Every order below is preceded by a real snapshot quote pulled from IB and recorded on
`SafeBroker`, never a research-time price and never an in-memory mock. That ordering is the
operational discipline the chapter is teaching. Delayed data is enough to demonstrate it;
whether the account carries live subscriptions is an account-level question.

The two quantities are sized to fit `max_order_value` at current SPY and QQQ levels, so the
demo shows a virtual fill rather than the cap rejection. Every other control still applies to
each leg.

```python
async def demonstrate_order_submission():
    """Show order submission in shadow mode using live IB snapshot quotes."""
    print("\n" + "=" * 60)
    print("ORDER SUBMISSION DEMO (Shadow Mode)")
    print("=" * 60)

    print("\nFetching delayed snapshot quotes from IB...")
    snapshot_prices = {}
    for symbol in SYMBOLS:
        price = await fetch_ib_snapshot(symbol)
        if price is None:
            raise RuntimeError(f"No IB snapshot quote received for {symbol} within 5 seconds")
        safe_broker.record_market_snapshot(symbol, price)
        snapshot_prices[symbol] = price
        print(f"   {symbol}: ${price:,.2f}")
    assert set(snapshot_prices) == set(SYMBOLS)

    orders = [("SPY", 10), ("QQQ", 12)]

    for symbol, qty in orders:
        print(f"\nSubmitting virtual BUY order: {qty} shares {symbol}")
        order = await safe_broker.submit_order_async(symbol, qty, side=OrderSide.BUY)
        print(f"   Order ID: {order.order_id}")
        print(f"   Status: {order.status.value}")
        print(f"   Side: {order.side.value}")
        print(f"   Quantity: {order.quantity}")

    vp = safe_broker._virtual_portfolio
    print("\nFinal Virtual Portfolio:")
    print(f"   Cash: ${vp.cash:,.2f}")
    for symbol, pos in vp.positions.items():
        value = pos.quantity * (pos.current_price or pos.entry_price)
        print(f"   {symbol}: {pos.quantity} shares @ ${pos.entry_price:.2f} = ${value:,.2f}")
```

One outer cleanup boundary covers both the live-feed demonstration and the
virtual-order demonstration.

```python
async def run_shadow_workflow() -> None:
    """Run the live-feed and shadow-order gates with guaranteed cleanup."""
    try:
        await run_live_demo(duration_seconds=LIVE_DURATION_SECONDS)
        await demonstrate_order_submission()
    finally:
        await safe_broker.disconnect()


warnings.filterwarnings("ignore", category=DeprecationWarning, module=r"nest_asyncio")
run_async(run_shadow_workflow())
```

The shadow portfolio is printed after each submission, so the path from intent to inventory is
visible without capital at risk. That path is where production bugs surface first, because the
strategy, the broker adapter and the risk guard all interact on it at once.

## 7. Clean Shutdown

Always disconnect cleanly to avoid connection issues on the next run. A disciplined shutdown sequence is
part of live reliability because stale sessions and dangling subscriptions are operational bugs too.

```python
print("\n[OK] Shadow workflow completed and disconnected from IB")
```

## Summary

This notebook demonstrated the complete IB integration workflow:

1. **Connection**: Connect to TWS/Gateway paper trading
2. **Account Info**: Query NLV, cash, positions
3. **Historical Data**: Request bars for strategy warm-up
4. **Real-Time Feed**: Subscribe to tick data, aggregate to bars
5. **Safe Trading**: Use SafeBroker in shadow mode
6. **Order Submission**: Virtual orders tracked in VirtualPortfolio

### Key Takeaways

- **Same Strategy class** works in backtest and live
- **Shadow mode first** - always test before real trading
- **SafeBroker** provides 8 layers of protection
- **IBBroker** handles all IB-specific details

### Next Steps

1. Run in shadow mode for 1-2 weeks
2. Verify signals match backtest expectations
3. Enable paper trading (`shadow_mode=False`)
4. Monitor for 2-4 weeks on paper
5. Gradually transition to live with small positions

```python
print("\n" + "=" * 60)
print("DEMO COMPLETE")
print("=" * 60)
print("Paper account readiness: verified")
print(f"Symbols Monitored: {', '.join(SYMBOLS)}")
print("Shadow Mode: ENABLED [OK]")
print("\nThe same MomentumStrategy that runs in backtest")
print("completed the IB shadow workflow without sending an order to the venue.")
```

## Key Takeaways

IB carries more connectivity and market-structure complexity than Alpaca, and the execution
architecture above absorbs all of it without the strategy knowing. Once connectivity, warm-up
and the shadow-mode controls are standard, moving a strategy between brokers is a configuration
change rather than a rewrite.

**Next**: Compare the simpler REST-style path in `04_alpaca_paper_trading_demo`, then move to
`08_pipeline_verification` to check that research outputs and live inputs still match.