API Reference

Use this page for signatures and public entry points. Use the User Guide for workflows, rollout order, and broker/feed selection.

Recommended Imports

from ml4t.live import (
    AlpacaBroker,
    AlpacaDataFeed,
    AsyncBrokerProtocol,
    BarAggregator,
    BarBuffer,
    BrokerProtocol,
    BrokerOrderContractError,
    AcceptedOrderPersistenceError,
    AuditJournalError,
    CanonicalOrderRequest,
    ConcurrentStateWriterError,
    CorruptStateError,
    DataFeedProtocol,
    FeedContinuityError,
    FeedContractError,
    FeedOverflowError,
    FeedQueueSnapshot,
    IBBroker,
    IBDataFeed,
    LiveEngine,
    LiveRiskConfig,
    OKXFundingFeed,
    OrderReplacementGapError,
    OrderValidationError,
    PersistenceSafetyError,
    ReconciliationMismatchError,
    RiskLimitError,
    RiskState,
    RuntimeCleanupError,
    RuntimeErrorContext,
    RuntimeFailureError,
    RuntimeState,
    RuntimeTransition,
    SafeBroker,
    ThreadSafeBrokerWrapper,
    UnsafePersistencePathError,
    VirtualPortfolio,
    runtime_error_context,
)

Public Surface At A Glance

Group	Primary symbols
Engine	LiveEngine, RuntimeState, RuntimeTransition, RuntimeErrorContext, runtime_error_context, RuntimeFailureError, RuntimeCleanupError
Brokers	IBBroker, AlpacaBroker
Stable-supported feeds	OKXFundingFeed
Experimental feeds	AlpacaDataFeed, IBDataFeed, DataBentoFeed, CryptoFeed, ExperimentalFeedError, ExperimentalFeedWarning
Feed helpers	BarAggregator, BarBuffer, FeedContractError, FeedContinuityError, FeedOverflowError, FeedQueueSnapshot
Orders	CanonicalOrderRequest, OrderValidationError, BrokerOrderContractError
Safety	LiveRiskConfig, SafeBroker, RiskState, RiskLimitError, PersistenceSafetyError, AuditJournalError, AcceptedOrderPersistenceError, VirtualPortfolio
Sync/async bridge	ThreadSafeBrokerWrapper
Protocols	BrokerProtocol, AsyncBrokerProtocol, DataFeedProtocol

Engine

LiveEngine

LiveEngine(
    strategy,
    broker,
    feed,
    *,
    on_error=None,
    feed_silence_seconds=None,
    watchdog_poll_seconds=1.0,
    halt_on_unhealthy=False,
    auto_recover=False,
    recovery_cooldown_seconds=5.0,
    max_recovery_attempts=3,
    max_event_age_seconds=None,
    on_health_change=None,
    strategy_callback_timeout_seconds=5.0,
    lifecycle_version=V1,
    execution_policy=None,
    strategy_config=None,
)

Async live trading engine.

Bridges async infrastructure with sync Strategy.on_data().

Initialize LiveEngine.

Parameters:

Name	Type	Description	Default
strategy	Strategy	Strategy instance to execute.	required
broker	AsyncBrokerProtocol	Async broker implementation.	required
feed	DataFeedProtocol	Data feed providing timestamp, data, context tuples.	required
on_error	Callable[[Exception, datetime, dict], None] | None	Custom error handler callback.	None
feed_silence_seconds	float | None	Optional threshold for degraded feed reporting.	None
watchdog_poll_seconds	float	Poll interval for runtime health monitoring.	1.0
halt_on_unhealthy	bool	Stop the engine when watchdog detects a degraded state.	False
auto_recover	bool	Attempt reconnect/restart when watchdog detects a recoverable state.	False
recovery_cooldown_seconds	float	Delay between recovery attempts.	5.0
max_recovery_attempts	int	Maximum recovery attempts before stopping.	3
max_event_age_seconds	float | None	Maximum provider-event age before dispatch. When omitted, use the supported feed's declared limit if present.	None
on_health_change	Callable[[str, dict[str, Any]], None] | None	Optional callback invoked when runtime health changes.	None
strategy_callback_timeout_seconds	float	Maximum callback duration. A callback that exceeds this duration is allowed to become quiescent before a typed timeout aborts the run.	5.0
lifecycle_version	LifecycleVersion | str	Portable strategy lifecycle version.	V1
execution_policy	ExecutionPolicy | None	Explicit live execution capabilities and behavior.	None
strategy_config	BacktestConfig | None	Backtest strategy configuration supplied to on_prepare.	None

Source code in src/ml4t/live/engine.py

def __init__(
    self,
    strategy: Strategy,
    broker: AsyncBrokerProtocol,
    feed: DataFeedProtocol,
    *,
    on_error: Callable[[Exception, datetime, dict], None] | None = None,
    feed_silence_seconds: float | None = None,
    watchdog_poll_seconds: float = 1.0,
    halt_on_unhealthy: bool = False,
    auto_recover: bool = False,
    recovery_cooldown_seconds: float = 5.0,
    max_recovery_attempts: int = 3,
    max_event_age_seconds: float | None = None,
    on_health_change: Callable[[str, dict[str, Any]], None] | None = None,
    strategy_callback_timeout_seconds: float = 5.0,
    lifecycle_version: LifecycleVersion | str = LifecycleVersion.V1,
    execution_policy: ExecutionPolicy | None = None,
    strategy_config: BacktestConfig | None = None,
) -> None:
    """Initialize LiveEngine.

    Args:
        strategy: Strategy instance to execute.
        broker: Async broker implementation.
        feed: Data feed providing timestamp, data, context tuples.
        on_error: Custom error handler callback.
        feed_silence_seconds: Optional threshold for degraded feed reporting.
        watchdog_poll_seconds: Poll interval for runtime health monitoring.
        halt_on_unhealthy: Stop the engine when watchdog detects a degraded state.
        auto_recover: Attempt reconnect/restart when watchdog detects a recoverable state.
        recovery_cooldown_seconds: Delay between recovery attempts.
        max_recovery_attempts: Maximum recovery attempts before stopping.
        max_event_age_seconds: Maximum provider-event age before dispatch. When omitted, use
            the supported feed's declared limit if present.
        on_health_change: Optional callback invoked when runtime health changes.
        strategy_callback_timeout_seconds: Maximum callback duration. A callback that exceeds
            this duration is allowed to become quiescent before a typed timeout aborts the run.
        lifecycle_version: Portable strategy lifecycle version.
        execution_policy: Explicit live execution capabilities and behavior.
        strategy_config: Backtest strategy configuration supplied to ``on_prepare``.
    """
    negotiated_version = negotiate_lifecycle_version(lifecycle_version)
    self._validate_strategy_lifecycle(strategy)
    self._validate_runtime_configuration(
        feed_silence_seconds=feed_silence_seconds,
        watchdog_poll_seconds=watchdog_poll_seconds,
        recovery_cooldown_seconds=recovery_cooldown_seconds,
        max_recovery_attempts=max_recovery_attempts,
    )
    self.strategy = strategy
    self.broker = broker
    self.feed = feed
    self.on_error = on_error or self._default_error_handler
    self.feed_silence_seconds = feed_silence_seconds
    self.watchdog_poll_seconds = watchdog_poll_seconds
    self.halt_on_unhealthy = halt_on_unhealthy
    self.auto_recover = auto_recover
    self.recovery_cooldown_seconds = recovery_cooldown_seconds
    self.max_recovery_attempts = max_recovery_attempts
    self.max_event_age_seconds = (
        max_event_age_seconds
        if max_event_age_seconds is not None
        else getattr(feed, "max_event_age_seconds", None)
    )
    self._validate_event_age(self.max_event_age_seconds)
    self.on_health_change = on_health_change
    self.strategy_callback_timeout_seconds = strategy_callback_timeout_seconds
    self.lifecycle_version = negotiated_version
    self.execution_policy = execution_policy or default_live_execution_policy()
    self.strategy_config = strategy_config or BacktestConfig()
    self.strategy_runtime = LiveStrategyRuntime(
        broker,
        self.execution_policy,
        negotiated_version,
    )
    self.lifecycle_dispatcher = LiveLifecycleDispatcher(
        strategy,
        LIFECYCLE_V1,
        callback_timeout_seconds=strategy_callback_timeout_seconds,
        event_recorder=self._record_runtime_event,
    )

    self._running = False
    self._shutdown_event = asyncio.Event()
    self._loop: asyncio.AbstractEventLoop | None = None
    self._wrapped_broker: ThreadSafeBrokerWrapper | None = None
    self._watchdog_task: asyncio.Task | None = None
    self._signals_installed = False
    self._signal_shutdown_task: asyncio.Task[None] | None = None
    self._run_in_progress = False
    self._run_done_event = asyncio.Event()
    self._runtime_lock = asyncio.Lock()
    self._cleanup_lock = asyncio.Lock()
    self._runtime_state = RuntimeState.STOPPED
    self._runtime_transitions: list[RuntimeTransition] = []
    self._operational_events: deque[dict[str, Any]] = deque(
        maxlen=RETAINED_OPERATIONAL_EVENT_LIMIT
    )
    self._operational_event_count = 0
    self._operational_event_forwarded_count = 0
    self._broker_connect_attempted = False
    self._feed_start_attempted = False
    self._feed_close_required = False
    self._terminal_failure_reason: str | None = None
    self._stop_requested_reason: str | None = None
    self._run_event_count = 0
    self._last_cleanup_result: dict[str, str] | None = None
    self._release_failures: dict[str, str] = {}
    self._continuity = EventContinuityTracker()

    self._bar_count = 0
    self._event_count = 0
    self._event_kind_counts = dict.fromkeys(MarketEventKind, 0)
    self._error_count = 0
    self._last_bar_time: datetime | None = None
    self._last_event_time: datetime | None = None
    self._last_bar_received_at: datetime | None = None
    self._last_health = "stopped"
    self._recovery_requested_reason: str | None = None
    self._recovery_attempts = 0

runtime_state property

runtime_state

Return the current transactional runtime state.

runtime_transitions property

runtime_transitions

Return retained state transitions in occurrence order.

operational_events property

operational_events

Return redacted structured runtime diagnostics.

stats property

stats

Get engine statistics and runtime health.

connect async

connect()

Acquire the broker and feed transactionally and become ready.

Source code in src/ml4t/live/engine.py

async def connect(self) -> None:
    """Acquire the broker and feed transactionally and become ready."""
    async with self._runtime_lock:
        if self._runtime_state is RuntimeState.READY:
            return
        if self._run_in_progress or self._runtime_state not in {
            RuntimeState.STOPPED,
            RuntimeState.FAILED,
        }:
            raise RuntimeError(
                f"LiveEngine.connect() is invalid while {self._runtime_state.value}"
            )

        self._terminal_failure_reason = None
        self._stop_requested_reason = None
        self._last_cleanup_result = None
        self._release_failures.clear()
        logger.info("LiveEngine: Connecting...")
        try:
            self._transition(RuntimeState.PREFLIGHT, reason="connect_requested")
            await self._acquire_runtime(ready_state=RuntimeState.READY)
            self._loop = asyncio.get_running_loop()
            self._wrapped_broker = ThreadSafeBrokerWrapper(
                self.broker,
                self._loop,
                self.strategy_runtime,
            )
            if not self._signals_installed:
                self._signals_installed = self._install_signal_handlers()
        except BaseException as error:
            self._annotate_connect_failure(error)
            self._terminal_failure_reason = f"startup:{type(error).__name__}"
            cleanup = await self._finalize_runtime(
                terminal_state=RuntimeState.FAILED,
                reason=self._terminal_failure_reason,
            )
            self._add_cleanup_note(error, cleanup)
            raise

        logger.info("LiveEngine: Connected and ready")

run async

run()

Main async loop - receives bars and dispatches to strategy.

Source code in src/ml4t/live/engine.py

async def run(self) -> None:
    """Main async loop - receives bars and dispatches to strategy."""
    if self._wrapped_broker is None or self._runtime_state is not RuntimeState.READY:
        raise RuntimeError("Call connect() before run()")
    if self._run_in_progress:
        raise RuntimeError("LiveEngine.run() is already active")

    self._run_in_progress = True
    self._run_done_event.clear()
    self._running = True
    self._recovery_requested_reason = None
    self._recovery_attempts = 0
    self._terminal_failure_reason = None
    self._stop_requested_reason = None
    self._run_event_count = 0
    self._shutdown_event.clear()
    logger.info("LiveEngine: Starting main loop")

    lifecycle_started = False
    callback_baseline = self.lifecycle_dispatcher.callback_counts
    failure: BaseException | None = None
    try:
        self._transition(RuntimeState.STARTING_STRATEGY, reason="strategy_start")
        lifecycle_started = True
        await self._dispatch_strategy(
            LifecyclePhase.RUN_START,
            self._wrapped_broker,
        )
        await self._dispatch_strategy(
            LifecyclePhase.CAUSAL_INITIALIZATION,
            self._wrapped_broker,
            self.strategy_config,
        )
        self._transition(RuntimeState.RUNNING, reason="strategy_started")
        self._watchdog_task = asyncio.create_task(
            self._watchdog_loop(),
            name="ml4t-live-watchdog",
        )
        while not self._shutdown_event.is_set():
            async for item in self._feed_items():
                if self._shutdown_event.is_set():
                    logger.info("LiveEngine: Shutdown requested")
                    break
                if self._current_broker_connected() is False:
                    self._terminal_failure_reason = "broker_disconnected"
                    self._shutdown_event.set()
                    raise RuntimeError(
                        "Broker disconnected before the next strategy event dispatch"
                    )

                processing_time = datetime.now(UTC)
                typed_event = item if isinstance(item, MarketEvent) else None
                if typed_event is not None:
                    validate_event_timing(
                        typed_event,
                        processing_time=processing_time,
                        max_age_seconds=self.max_event_age_seconds,
                    )
                    disposition = self._continuity.validate(typed_event)
                    if disposition is ContinuityDisposition.DUPLICATE:
                        self._record_runtime_event(
                            "feed_duplicate_skipped",
                            source=typed_event.source,
                            asset=typed_event.asset,
                            kind=typed_event.kind.value,
                            provider_sequence=typed_event.provider_sequence,
                            event_time=typed_event.event_time.isoformat(),
                        )
                        continue
                    timestamp, data, context = strategy_input(
                        typed_event,
                        processing_time=processing_time,
                    )
                    event_kind = typed_event.kind
                else:
                    timestamp, data, context = self._validate_legacy_feed_item(item)
                    event_kind = MarketEventKind.BAR

                self._event_count += 1
                self._run_event_count += 1
                self._event_kind_counts[event_kind] += 1
                self._last_event_time = timestamp
                if event_kind is MarketEventKind.BAR:
                    self._bar_count += 1
                    self._last_bar_time = timestamp
                self._last_bar_received_at = processing_time

                complete_or_non_bar = (
                    typed_event is None
                    or event_kind is not MarketEventKind.BAR
                    or typed_event.completion is EventCompletion.COMPLETE
                )
                record_market_data = getattr(self.broker, "_record_market_data", None)
                if callable(record_market_data) and complete_or_non_bar:
                    record_market_data(timestamp, data, context)

                try:
                    if event_kind is MarketEventKind.BAR and complete_or_non_bar:
                        try:
                            await self.strategy_runtime.process_market_event(
                                timestamp,
                                data,
                                context,
                            )
                        except BaseException as error:
                            _attach_runtime_error_context(
                                error,
                                component="strategy_runtime",
                                operation="process_market_event",
                                runtime_state=self._runtime_state,
                                recovery_action=(
                                    "reconcile portable strategy state before restarting"
                                ),
                            )
                            raise
                    await self._dispatch_strategy(
                        LifecyclePhase.MARKET_EVENT,
                        timestamp,
                        data,
                        context,
                        self._wrapped_broker,
                        event_time=timestamp,
                    )
                except Exception as error:
                    self._error_count += 1
                    try:
                        self.on_error(error, timestamp, data)
                    except BaseException as handler_error:
                        error.add_note(
                            "on_error also failed: "
                            f"{type(handler_error).__name__}: "
                            f"{redact_sensitive(str(handler_error))}"
                        )
                    self._shutdown_event.set()
                    raise

            if self._shutdown_event.is_set():
                break

            if self._recovery_requested_reason is None:
                self._recovery_requested_reason = "feed_terminated"
                self._transition(RuntimeState.DEGRADED, reason="feed_terminated")
                logger.warning("LiveEngine: Feed terminated")

            if not self.auto_recover:
                logger.warning(
                    "LiveEngine: Runtime degraded (%s) and auto recovery disabled",
                    self._recovery_requested_reason,
                )
                break

            recovered = await self._attempt_recovery(self._recovery_requested_reason)
            if not recovered:
                break
    except BaseException as error:
        failure = error
        if isinstance(error, FeedContractError | FeedOverflowError):
            _attach_runtime_error_context(
                error,
                component="feed",
                operation="validate",
                runtime_state=self._runtime_state,
                recovery_action=("restore the feed and establish continuity before restarting"),
            )
        elif not isinstance(error, asyncio.CancelledError):
            _attach_runtime_error_context(
                error,
                component="engine",
                operation="run",
                runtime_state=self._runtime_state,
                recovery_action="inspect runtime diagnostics before restarting",
            )
        if (
            not isinstance(error, asyncio.CancelledError)
            and self._terminal_failure_reason is None
        ):
            self._terminal_failure_reason = f"runtime:{type(error).__name__}"
        if isinstance(error, FeedContinuityError | FeedOverflowError):
            if self._runtime_state is RuntimeState.RUNNING:
                try:
                    self._transition(
                        RuntimeState.DEGRADED,
                        reason=f"feed_safety:{type(error).__name__}",
                    )
                except BaseException as transition_error:
                    error.add_note(
                        "feed safety transition also failed: "
                        f"{type(transition_error).__name__}: "
                        f"{redact_sensitive(str(transition_error))}"
                    )
            try:
                self._record_runtime_event(
                    "feed_safety_halt",
                    detail=error.to_dict(),
                )
            except BaseException as audit_error:
                error.add_note(
                    "feed safety audit also failed: "
                    f"{type(audit_error).__name__}: "
                    f"{redact_sensitive(str(audit_error))}"
                )
    finally:
        self._running = False
        try:
            await self._cancel_watchdog()
        except BaseException as finalization_error:
            failure = self._retain_failure(
                failure,
                finalization_error,
                context="watchdog cancellation",
            )
            self._terminal_failure_reason = (
                f"watchdog_cancel:{type(finalization_error).__name__}"
            )
        try:
            self._emit_health_transition(self.runtime_status())
        except BaseException as finalization_error:
            failure = self._retain_failure(
                failure,
                finalization_error,
                context="health finalization",
            )
            self._terminal_failure_reason = (
                f"health_finalization:{type(finalization_error).__name__}"
            )
        if lifecycle_started:
            try:
                await self._dispatch_strategy(
                    LifecyclePhase.RUN_END,
                    self._wrapped_broker,
                )
            except BaseException as finalization_error:
                self._terminal_failure_reason = (
                    f"strategy_end:{type(finalization_error).__name__}"
                )
                failure = self._retain_failure(
                    failure,
                    finalization_error,
                    context="on_end",
                )
        if failure is None:
            try:
                self.lifecycle_dispatcher.validate_completed_run(
                    self._run_event_count,
                    baseline=callback_baseline,
                )
            except BaseException as finalization_error:
                failure = finalization_error
                self._terminal_failure_reason = (
                    f"lifecycle_validation:{type(finalization_error).__name__}"
                )
        try:
            self.lifecycle_dispatcher.close()
        except BaseException as finalization_error:
            failure = self._retain_failure(
                failure,
                finalization_error,
                context="strategy worker shutdown",
            )
            self._terminal_failure_reason = (
                f"strategy_worker:{type(finalization_error).__name__}"
            )
        terminal_state = (
            RuntimeState.FAILED
            if self._terminal_failure_reason is not None
            else RuntimeState.STOPPED
        )
        cleanup = await self._finalize_runtime(
            terminal_state=terminal_state,
            reason=(
                self._terminal_failure_reason or self._stop_requested_reason or "run_completed"
            ),
        )
        if failure is not None:
            self._add_cleanup_note(failure, cleanup)
        elif any(value.startswith("failed:") for value in cleanup.values()):
            failure = RuntimeCleanupError(cleanup)
        elif self._terminal_failure_reason is not None:
            failure = RuntimeFailureError(self._terminal_failure_reason)
        self._run_in_progress = False
        self._run_done_event.set()
        logger.info(
            "LiveEngine: Stopped. Bars: %s, Errors: %s",
            self._bar_count,
            self._error_count,
        )

    if failure is not None:
        raise failure.with_traceback(failure.__traceback__)

stop async

stop()

Request shutdown and release resources exactly once.

Source code in src/ml4t/live/engine.py

async def stop(self) -> None:
    """Request shutdown and release resources exactly once."""
    logger.info("LiveEngine: Stopping...")
    self._stop_requested_reason = self._stop_requested_reason or "stop_requested"
    self._shutdown_event.set()
    await self._cancel_watchdog()
    if self._run_in_progress:
        self._stop_feed_once()
        await self._run_done_event.wait()
        cleanup = self._last_cleanup_result or {}
        if any(value.startswith("failed:") for value in cleanup.values()):
            raise RuntimeCleanupError(cleanup)
    else:
        async with self._runtime_lock:
            cleanup = await self._finalize_runtime(
                terminal_state=RuntimeState.STOPPED,
                reason=self._stop_requested_reason,
            )
        if any(value.startswith("failed:") for value in cleanup.values()):
            raise RuntimeCleanupError(cleanup)
    logger.info("LiveEngine: Stopped")

runtime_status

runtime_status(now=None)

Return engine runtime health and session context.

Source code in src/ml4t/live/engine.py

def runtime_status(self, now: datetime | None = None) -> dict[str, Any]:
    """Return engine runtime health and session context."""
    reference_now = self._normalize_utc(now or datetime.now(UTC))
    session = self._equity_session_status(reference_now)
    broker_connected = self._current_broker_connected()
    feed_stats = getattr(self.feed, "stats", None)
    if not isinstance(feed_stats, dict):
        feed_stats = None

    last_bar_age_seconds: float | None = None
    if self._last_bar_received_at is not None:
        last_bar_age_seconds = max(
            0.0,
            (reference_now - self._last_bar_received_at).total_seconds(),
        )

    if self._runtime_state is RuntimeState.FAILED:
        health = "failed"
    elif self._runtime_state is RuntimeState.READY:
        health = "ready"
    elif not self._running:
        health = "stopped"
    elif broker_connected is False:
        health = "broker_disconnected"
    elif session["market"] not in {"open", "continuous"}:
        health = "idle_market_closed"
    elif last_bar_age_seconds is None:
        health = "waiting_for_data"
    elif (
        self.feed_silence_seconds is not None
        and last_bar_age_seconds > self.feed_silence_seconds
    ):
        health = "feed_silent"
    else:
        health = "ok"

    return {
        "running": self._running,
        "runtime_state": self._runtime_state.value,
        "terminal_failure_reason": self._terminal_failure_reason,
        "event_count": self._event_count,
        "event_kind_counts": {
            kind.value: count for kind, count in self._event_kind_counts.items()
        },
        "bar_count": self._bar_count,
        "error_count": self._error_count,
        "last_event_time": self._last_event_time,
        "last_bar_time": self._last_bar_time,
        "last_bar_received_at": self._last_bar_received_at,
        "last_bar_age_seconds": last_bar_age_seconds,
        "broker_connected": broker_connected,
        "session_state": session["market"],
        "next_session_boundary": session["next_boundary"],
        "tracked_symbols": session["tracked_symbols"],
        "health": health,
        "halt_on_unhealthy": self.halt_on_unhealthy,
        "auto_recover": self.auto_recover,
        "max_event_age_seconds": self.max_event_age_seconds,
        "feed": feed_stats,
        "continuity": self._continuity.snapshot(),
        "recovery_requested": self._recovery_requested_reason,
        "recovery_attempts": self._recovery_attempts,
        "lifecycle_version": self.lifecycle_version.value,
        "execution_policy": self.execution_policy.to_dict(),
        "target_intent_count": len(self.strategy_runtime.targets),
        "position_rule_state_count": len(self.strategy_runtime.position_rule_states),
        "callback_counts": {
            phase.value: count
            for phase, count in self.lifecycle_dispatcher.callback_counts.items()
        },
        "diagnostics": {
            "callback_invocations_total": self.lifecycle_dispatcher.invocation_count,
            "callback_invocations_retained": len(self.lifecycle_dispatcher.invocations),
            "callback_invocations_dropped": (
                self.lifecycle_dispatcher.dropped_invocation_count
            ),
            "operational_events_total": self._operational_event_count,
            "operational_events_forwarded": self._operational_event_forwarded_count,
            "operational_events_retained": len(self._operational_events),
            "operational_events_dropped": max(
                0,
                self._operational_event_count - len(self._operational_events),
            ),
        },
        "last_cleanup_result": self._last_cleanup_result,
    }

RuntimeState

Bases: StrEnum

One explicit phase of engine resource and strategy ownership.

RuntimeTransition dataclass

RuntimeTransition(
    occurred_at,
    previous,
    current,
    reason,
    attempt,
    last_known_sequence,
)

Structured evidence for one runtime state change.

RuntimeErrorContext dataclass

RuntimeErrorContext(
    component,
    operation,
    runtime_state,
    recovery_action,
    root_cause_type,
)

Redacted operator context attached to a runtime exception.

to_dict

to_dict()

Return machine-readable context without exception text.

Source code in src/ml4t/live/engine.py

def to_dict(self) -> dict[str, str]:
    """Return machine-readable context without exception text."""
    return {
        "component": self.component,
        "operation": self.operation,
        "runtime_state": self.runtime_state.value,
        "recovery_action": self.recovery_action,
        "root_cause_type": self.root_cause_type,
    }

runtime_error_context

runtime_error_context(error)

Return structured runtime context when the engine attached it.

Source code in src/ml4t/live/engine.py

def runtime_error_context(error: BaseException) -> RuntimeErrorContext | None:
    """Return structured runtime context when the engine attached it."""
    context = getattr(error, "runtime_context", None)
    return context if isinstance(context, RuntimeErrorContext) else None

RuntimeFailureError

RuntimeFailureError(reason)

Bases: RuntimeError

Raised when an asynchronous runtime failure reaches a terminal state.

Source code in src/ml4t/live/engine.py

def __init__(self, reason: str) -> None:
    self.reason = reason
    super().__init__(reason)
    _attach_runtime_error_context(
        self,
        component="engine",
        operation="recover",
        runtime_state=RuntimeState.FAILED,
        recovery_action="inspect recovery events and restore the failed dependency before restart",
    )

RuntimeCleanupError

RuntimeCleanupError(cleanup_result)

Bases: RuntimeError

Raised when runtime finalization cannot release every acquired resource.

Source code in src/ml4t/live/engine.py

def __init__(self, cleanup_result: dict[str, str]) -> None:
    self.cleanup_result = dict(cleanup_result)
    super().__init__(f"runtime cleanup failed: {self.cleanup_result}")
    _attach_runtime_error_context(
        self,
        component="runtime_resources",
        operation="release",
        runtime_state=RuntimeState.FAILED,
        recovery_action="correct the reported release failure, then call stop() again",
    )

Safety And Rollout

CanonicalOrderRequest dataclass

CanonicalOrderRequest(
    asset,
    quantity,
    side,
    order_type,
    limit_price=None,
    stop_price=None,
)

One unsigned venue request used unchanged for checks and submission.

validate_result

validate_result(order)

Require the adapter result to describe this exact request.

Source code in src/ml4t/live/orders.py

def validate_result(self, order: Order) -> None:
    """Require the adapter result to describe this exact request."""
    if not isinstance(order, Order):
        raise BrokerOrderContractError("broker submission did not return an Order")
    if (
        order.asset.upper() != self.asset
        or order.side is not self.side
        or order.order_type is not self.order_type
        or not math.isclose(float(order.quantity), self.quantity, rel_tol=0, abs_tol=1e-12)
        or order.limit_price != self.limit_price
        or order.stop_price != self.stop_price
    ):
        raise BrokerOrderContractError("broker result differs from the canonical order request")
    if not order.order_id or not isinstance(order.status, OrderStatus):
        raise BrokerOrderContractError("broker result has no identifier or valid status")
    if order.status is OrderStatus.PENDING and (
        not isinstance(order.created_at, datetime) or order.created_at.utcoffset() is None
    ):
        raise BrokerOrderContractError(
            "pending broker result must have a timezone-aware creation time"
        )
    if not math.isfinite(float(order.filled_quantity)) or not (
        0 <= order.filled_quantity <= order.quantity
    ):
        raise BrokerOrderContractError("broker result has an invalid cumulative fill")
    if order.status is OrderStatus.FILLED and (
        not math.isclose(order.filled_quantity, order.quantity, rel_tol=0, abs_tol=1e-12)
        or order.filled_price is None
        or not math.isfinite(order.filled_price)
        or order.filled_price <= 0
        or not isinstance(order.filled_at, datetime)
        or order.filled_at.utcoffset() is None
    ):
        raise BrokerOrderContractError("filled broker result has incomplete fill evidence")

LiveRiskConfig dataclass

LiveRiskConfig(
    max_position_value=50000.0,
    max_position_shares=1000.0,
    max_total_exposure=200000.0,
    max_positions=20,
    max_order_value=10000.0,
    max_order_shares=500.0,
    max_orders_per_minute=10,
    max_daily_loss=5000.0,
    max_drawdown_pct=0.05,
    max_price_deviation_pct=0.05,
    max_data_staleness_seconds=60.0,
    dedup_window_seconds=1.0,
    allowed_assets=set(),
    blocked_assets=set(),
    shadow_mode=False,
    execution_mode=None,
    kill_switch_enabled=False,
    allow_reducing_risk_when_killed=True,
    halt_on_reducing_risk_failure=True,
    fail_on_reconciliation_mismatch=False,
    state_file=".ml4t_risk_state.json",
    journal_file=None,
    fail_on_journal_error=True,
)

Risk configuration for live trading.

Multiple layers of protection. Set a limit to None to disable that specific check. NaN and infinity are always invalid.

Example

Conservative configuration

config = LiveRiskConfig( max_position_value=25_000.0, max_daily_loss=2_000.0, execution_mode="shadow", )

Disable a specific check explicitly

config = LiveRiskConfig( max_position_value=None, max_daily_loss=10_000.0, # Only daily loss limit )

Safety Recommendations

1. Always start with execution_mode="shadow"
2. Graduate to paper trading
3. Use small positions when going live
4. Set conservative risk limits

__post_init__

__post_init__()

Validate configuration parameters.

Source code in src/ml4t/live/safety.py

def __post_init__(self) -> None:
    """Validate configuration parameters."""
    if isinstance(self.execution_mode, str):
        try:
            self.execution_mode = ExecutionMode(self.execution_mode.lower())
        except ValueError as error:
            raise ExecutionModeError(
                "execution_mode must be one of: shadow, paper, live"
            ) from error
    if self.execution_mode is None:
        if self.shadow_mode:
            self.execution_mode = ExecutionMode.SHADOW
    else:
        if self.shadow_mode and self.execution_mode is not ExecutionMode.SHADOW:
            raise ExecutionModeError(
                "shadow_mode=True conflicts with a non-shadow execution_mode"
            )
        self.shadow_mode = self.execution_mode is ExecutionMode.SHADOW

    positive_numbers = (
        "max_position_value",
        "max_position_shares",
        "max_total_exposure",
        "max_order_value",
        "max_order_shares",
        "max_daily_loss",
        "max_data_staleness_seconds",
    )
    positive_counts = (
        "max_positions",
        "max_orders_per_minute",
    )
    percentages = ("max_drawdown_pct", "max_price_deviation_pct")
    for name in positive_numbers:
        self._validate_optional_number(name, minimum=0.0, inclusive=False)
    for name in positive_counts:
        self._validate_optional_count(name)
    for name in percentages:
        self._validate_optional_number(name, minimum=0.0, maximum=1.0, inclusive=False)
    self._validate_optional_number("dedup_window_seconds", minimum=0.0, inclusive=True)

    # Validate asset restrictions
    if self.allowed_assets and self.blocked_assets:
        overlap = self.allowed_assets & self.blocked_assets
        if overlap:
            raise ValueError(f"Assets cannot be in both allowed and blocked lists: {overlap}")

    # Validate state file path
    if not self.state_file:
        raise ValueError("state_file cannot be empty")
    if self.journal_file is not None and not self.journal_file:
        raise ValueError("journal_file cannot be empty when provided")
    if self.journal_file is not None:
        state_path = Path(self.state_file).absolute()
        journal_path = Path(self.journal_file).absolute()
        if journal_path in {
            state_path,
            state_path.with_name(f"{state_path.name}.lock"),
        }:
            raise ValueError("journal_file must not overlap state_file or its lock")

require_execution_mode

require_execution_mode()

Return the explicit mode or reject ambiguous external execution.

Source code in src/ml4t/live/safety.py

def require_execution_mode(self) -> ExecutionMode:
    """Return the explicit mode or reject ambiguous external execution."""
    if self.execution_mode is None:
        raise ExecutionModeError(
            "execution_mode must be explicitly set to shadow, paper, or live; "
            "shadow_mode=False is ambiguous"
        )
    try:
        mode = ExecutionMode(self.execution_mode)
    except (TypeError, ValueError) as error:
        raise ExecutionModeError(
            "execution_mode must be one of: shadow, paper, live"
        ) from error
    if self.shadow_mode != (mode is ExecutionMode.SHADOW):
        raise ExecutionModeError("shadow_mode conflicts with execution_mode")
    self.execution_mode = mode
    return mode

SafeBroker

SafeBroker(broker, config)

Risk-controlled wrapper with state persistence.

Safety Features: 1. Pre-trade validation against all risk limits 2. Order rate limiting 3. Drawdown monitoring with kill switch 4. Fat finger protection (price deviation check) 5. Stale data protection 6. Duplicate order filter 7. Shadow mode with VirtualPortfolio (realistic paper trading) 8. Owner-only, versioned state and chained audit persistence across restarts

Example

broker = IBBroker() await broker.connect()

safe = SafeBroker( broker=broker, config=LiveRiskConfig( max_position_value=25000, execution_mode="shadow", ) )

Use safe in strategy

engine = LiveEngine(strategy, safe, feed)

Initialize SafeBroker.

Parameters:

Name	Type	Description	Default
broker	AsyncBrokerProtocol	Async broker implementation (IBBroker, AlpacaBroker, etc.)	required
config	LiveRiskConfig	Risk configuration	required

Source code in src/ml4t/live/safety.py

def __init__(self, broker: AsyncBrokerProtocol, config: LiveRiskConfig) -> None:
    """Initialize SafeBroker.

    Args:
        broker: Async broker implementation (IBBroker, AlpacaBroker, etc.)
        config: Risk configuration
    """
    self.execution_mode = config.require_execution_mode()
    self._broker = broker
    self.config = config
    self._state_store = SecureStateStore(config.state_file)
    self._state_generation = 0
    self._state_error: PersistenceSafetyError | None = None
    self._journal_error: AuditJournalError | None = None
    self._state_migration_required = False
    self._virtual_portfolio = VirtualPortfolio(initial_cash=100_000.0)
    self._state_store.acquire_writer()

    try:
        self._state = self._load_state()
        if (
            self._state.execution_mode is not None
            and self._state.execution_mode != self.execution_mode.value
        ):
            raise ExecutionModeError(
                "persisted execution mode does not match the configured execution_mode"
            )
        self._state.execution_mode = self.execution_mode.value
        if config.shadow_mode and self._state.shadow_portfolio:
            self._virtual_portfolio.restore_state(self._state.shadow_portfolio)
        self._audit_journal = SecureAuditJournal(self._journal_path())
        self._validate_journal()
        if self._state_migration_required:
            self._save_state()
            self._state_migration_required = False
    except BaseException:
        self._state_store.release_writer()
        raise

    # Rate limiting
    self._order_timestamps: list[float] = []
    self._order_operation_lock = asyncio.Lock()

    # Duplicate detection
    self._recent_orders: list[tuple[float, str, float]] = []  # (time, asset, qty)

    # Latest market reference per asset, populated by LiveEngine
    self._latest_market_data: dict[str, MarketSnapshot] = {}
    self._last_reconciliation_report: dict[str, Any] | None = None

    self._execution_identity_validated = self.execution_mode is ExecutionMode.SHADOW
    logger.info("SafeBroker initialized. Execution mode: %s", self.execution_mode.value)
    disabled_controls = self._disabled_safety_controls()
    if disabled_controls:
        logger.warning("Disabled safety controls: %s", ", ".join(disabled_controls))
    if self._state.kill_switch_activated:
        logger.warning(
            f"Kill switch was previously activated: {self._state.kill_switch_reason}"
        )

positions property

positions

Get current positions.

In shadow mode, returns virtual positions. In live mode, returns broker positions.

pending_orders property

pending_orders

Get pending orders.

reconciliation_report property

reconciliation_report

Return the latest startup reconciliation report.

replacement_gaps property

replacement_gaps

Return unresolved cancel-and-resubmit operations.

persistence_status property

persistence_status

Return the current state and journal health without secret values.

is_connected property

is_connected

Check if broker is connected.

execution_capabilities property

execution_capabilities

Return capabilities declared by the wrapped venue.

close_persistence

close_persistence()

Release the exclusive state-writer lease for this broker instance.

Source code in src/ml4t/live/safety.py

def close_persistence(self) -> None:
    """Release the exclusive state-writer lease for this broker instance."""
    self._state_store.release_writer()

assert_paper_trading

assert_paper_trading()

Fail unless this wrapper and its provider are configured for paper execution.

Source code in src/ml4t/live/safety.py

def assert_paper_trading(self) -> None:
    """Fail unless this wrapper and its provider are configured for paper execution."""
    if self.execution_mode is not ExecutionMode.PAPER:
        raise ExecutionModeError("SafeBroker is not configured for paper execution")
    self._validate_execution_identity()

assert_live_trading

assert_live_trading()

Fail unless this wrapper and its provider are configured for live execution.

Source code in src/ml4t/live/safety.py

def assert_live_trading(self) -> None:
    """Fail unless this wrapper and its provider are configured for live execution."""
    if self.execution_mode is not ExecutionMode.LIVE:
        raise ExecutionModeError("SafeBroker is not configured for live execution")
    self._validate_execution_identity()

load_portable_strategy_state

load_portable_strategy_state()

Return a copy of persisted target and position-rule state.

Source code in src/ml4t/live/safety.py

def load_portable_strategy_state(self) -> dict[str, Any]:
    """Return a copy of persisted target and position-rule state."""
    return json.loads(json.dumps(self._state.portable_strategy_state))

save_portable_strategy_state

save_portable_strategy_state(state)

Persist target and position-rule state with the safety state.

Source code in src/ml4t/live/safety.py

def save_portable_strategy_state(self, state: dict[str, Any]) -> None:
    """Persist target and position-rule state with the safety state."""
    self._state.portable_strategy_state = json.loads(json.dumps(state))
    self._save_state()

get_position

get_position(asset)

Get position for specific asset.

Parameters:

Name	Type	Description	Default
asset	str	Asset symbol	required

Returns:

Type	Description
Position | None	Position object or None

Source code in src/ml4t/live/safety.py

def get_position(self, asset: str) -> Position | None:
    """Get position for specific asset.

    Args:
        asset: Asset symbol

    Returns:
        Position object or None
    """
    if self.config.shadow_mode:
        return self._virtual_portfolio.positions.get(asset)
    return self._broker.positions.get(asset)

get_positions

get_positions()

Get all current positions through the portable strategy facade.

Source code in src/ml4t/live/safety.py

def get_positions(self) -> dict[str, Position]:
    """Get all current positions through the portable strategy facade."""
    return self.positions

get_account_value_async async

get_account_value_async()

Get total account value (async).

Returns:

Type	Description
float	Total account value in base currency

Source code in src/ml4t/live/safety.py

async def get_account_value_async(self) -> float:
    """Get total account value (async).

    Returns:
        Total account value in base currency
    """
    if self.config.shadow_mode:
        return self._virtual_portfolio.account_value
    return await self._broker.get_account_value_async()

get_cash_async async

get_cash_async()

Get available cash (async).

Returns:

Type	Description
float	Available cash in base currency

Source code in src/ml4t/live/safety.py

async def get_cash_async(self) -> float:
    """Get available cash (async).

    Returns:
        Available cash in base currency
    """
    if self.config.shadow_mode:
        return self._virtual_portfolio.cash
    return await self._broker.get_cash_async()

cancel_order_async async

cancel_order_async(order_id)

Cancel pending order.

Parameters:

Name	Type	Description	Default
order_id	str	ID of order to cancel	required

Returns:

Type	Description
bool	True if cancel request submitted

Source code in src/ml4t/live/safety.py

async def cancel_order_async(self, order_id: str) -> bool:
    """Cancel pending order.

    Args:
        order_id: ID of order to cancel

    Returns:
        True if cancel request submitted
    """
    async with self._order_operation_lock:
        return await self._cancel_order_unlocked(order_id)

close_position_async async

close_position_async(asset)

Close entire position.

Close positions bypass normal limits (safety feature).

Parameters:

Name	Type	Description	Default
asset	str	Asset symbol to close	required

Returns:

Type	Description
Order | None	Order object if position exists

Source code in src/ml4t/live/safety.py

async def close_position_async(self, asset: str) -> Order | None:
    """Close entire position.

    Close positions bypass normal limits (safety feature).

    Args:
        asset: Asset symbol to close

    Returns:
        Order object if position exists
    """
    position = self.get_position(asset)
    if position is None or position.quantity == 0:
        return None
    return await self.reduce_position_async(
        asset,
        abs(position.quantity),
        reason="close_position",
        idempotency_key=(
            f"close:{asset}:"
            f"{position.entry_time.isoformat() if position.entry_time else 'unknown-entry'}"
        ),
    )

reduce_position_async async

reduce_position_async(
    asset,
    quantity,
    *,
    reason,
    idempotency_key,
    fill_price=None,
)

Submit an explicitly reducing order under the configured kill-switch policy.

Source code in src/ml4t/live/safety.py

async def reduce_position_async(
    self,
    asset: str,
    quantity: float,
    *,
    reason: str,
    idempotency_key: str,
    fill_price: float | None = None,
) -> Order:
    """Submit an explicitly reducing order under the configured kill-switch policy."""
    async with self._order_operation_lock:
        return await self._reduce_position_unlocked(
            asset,
            quantity,
            reason=reason,
            idempotency_key=idempotency_key,
            fill_price=fill_price,
        )

replace_order_async async

replace_order_async(
    order_id,
    *,
    quantity=None,
    limit_price=None,
    stop_price=None,
)

Replace a pending order via cancel-and-resubmit.

Source code in src/ml4t/live/safety.py

async def replace_order_async(
    self,
    order_id: str,
    *,
    quantity: float | None = None,
    limit_price: float | None = None,
    stop_price: float | None = None,
) -> Order:
    """Replace a pending order via cancel-and-resubmit."""
    async with self._order_operation_lock:
        return await self._replace_order_unlocked(
            order_id,
            quantity=quantity,
            limit_price=limit_price,
            stop_price=stop_price,
        )

submit_order_async async

submit_order_async(
    asset,
    quantity,
    side=None,
    order_type=MARKET,
    limit_price=None,
    stop_price=None,
    **kwargs,
)

Submit order with full risk validation.

Parameters:

Name	Type	Description	Default
asset	str	Asset symbol	required
quantity	float	Signed shares/contracts when side is omitted; positive unsigned shares/contracts when side is provided	required
side	OrderSide | None	Order side (BUY/SELL), inferred from signed quantity if omitted	None
order_type	OrderType	Type of order	MARKET
limit_price	float | None	Limit price for LIMIT/STOP_LIMIT orders	None
stop_price	float | None	Stop price for STOP/STOP_LIMIT orders	None
**kwargs	Any	Additional broker-specific parameters	{}

Returns:

Type	Description
Order	Order object

Raises:

Type	Description
RiskLimitError	If order violates any risk limit

Source code in src/ml4t/live/safety.py

async def submit_order_async(
    self,
    asset: str,
    quantity: float,
    side: OrderSide | None = None,
    order_type: OrderType = OrderType.MARKET,
    limit_price: float | None = None,
    stop_price: float | None = None,
    **kwargs: Any,
) -> Order:
    """Submit order with full risk validation.

    Args:
        asset: Asset symbol
        quantity: Signed shares/contracts when side is omitted; positive unsigned
            shares/contracts when side is provided
        side: Order side (BUY/SELL), inferred from signed quantity if omitted
        order_type: Type of order
        limit_price: Limit price for LIMIT/STOP_LIMIT orders
        stop_price: Stop price for STOP/STOP_LIMIT orders
        **kwargs: Additional broker-specific parameters

    Returns:
        Order object

    Raises:
        RiskLimitError: If order violates any risk limit
    """
    async with self._order_operation_lock:
        return await self._submit_order_unlocked(
            asset,
            quantity,
            side,
            order_type,
            limit_price,
            stop_price,
            **kwargs,
        )

record_market_snapshot

record_market_snapshot(asset, price, timestamp=None)

Cache a single price observation for the staleness guard.

The supported way to keep the cache fresh is the streaming path: a Feed (e.g. IBDataFeed) emits ticks, LiveEngine shuttles them into _record_market_data on every bar, and the cache stays current automatically. Use that for any continuous-loop deployment.

This method is the non-streaming escape hatch for one-shot flows that legitimately have no tick stream in front of the broker:

• A CLI flatten tool that takes a position list and submits MOC/MARKET closeouts.
• A REST-only broker adapter that fetches quotes synchronously per request rather than via a streaming feed.
• A test harness setting up controlled state.

It is not the right tool inside a notebook that already runs a live engine - there the streaming path covers staleness implicitly. Reaching for record_market_snapshot from inside a tick-driven flow is a code smell: it usually means the engine isn't actually wired up, and the snapshot will go stale silently.

Parameters:

Name	Type	Description	Default
asset	str	Symbol to record.	required
price	float	Reference price (e.g. last close, mid quote, snapshot top-of-book mid). Must be > 0.	required
timestamp	datetime | None	Bar/quote timestamp. Defaults to now (UTC). The observed_at field is always set to now so freshness checks measure wall-clock age since this call, not since the quote was generated.	None

Source code in src/ml4t/live/safety.py

def record_market_snapshot(
    self,
    asset: str,
    price: float,
    timestamp: datetime | None = None,
) -> None:
    """Cache a single price observation for the staleness guard.

    **The supported way to keep the cache fresh is the streaming path:**
    a ``Feed`` (e.g. ``IBDataFeed``) emits ticks, ``LiveEngine`` shuttles
    them into ``_record_market_data`` on every bar, and the cache stays
    current automatically. Use that for any continuous-loop deployment.

    This method is the **non-streaming escape hatch** for one-shot flows
    that legitimately have no tick stream in front of the broker:

    - A CLI flatten tool that takes a position list and submits MOC/MARKET
      closeouts.
    - A REST-only broker adapter that fetches quotes synchronously per
      request rather than via a streaming feed.
    - A test harness setting up controlled state.

    It is **not** the right tool inside a notebook that already runs a
    live engine - there the streaming path covers staleness implicitly.
    Reaching for ``record_market_snapshot`` from inside a tick-driven
    flow is a code smell: it usually means the engine isn't actually
    wired up, and the snapshot will go stale silently.

    Args:
        asset: Symbol to record.
        price: Reference price (e.g. last close, mid quote, snapshot
            top-of-book mid). Must be > 0.
        timestamp: Bar/quote timestamp. Defaults to now (UTC). The
            ``observed_at`` field is always set to now so freshness
            checks measure wall-clock age since this call, not since the
            quote was generated.
    """
    if not (price > 0):
        raise ValueError(f"record_market_snapshot: price must be > 0, got {price}")
    observed_at = datetime.now(UTC)
    ts = timestamp if timestamp is not None else observed_at
    if ts.tzinfo is None:
        ts = ts.replace(tzinfo=UTC)
    else:
        ts = ts.astimezone(UTC)
    asset_key = asset.upper()
    self._latest_market_data[asset_key] = MarketSnapshot(
        timestamp=ts,
        observed_at=observed_at,
        price=float(price),
    )
    self._virtual_portfolio.update_prices({asset_key: float(price)})

enable_kill_switch

enable_kill_switch(reason='Manual')

Manually enable kill switch.

Parameters:

Name	Type	Description	Default
reason	str	Reason for activation (default: "Manual")	'Manual'

Source code in src/ml4t/live/safety.py

def enable_kill_switch(self, reason: str = "Manual") -> None:
    """Manually enable kill switch.

    Args:
        reason: Reason for activation (default: "Manual")
    """
    self._activate_kill_switch(reason)

disable_kill_switch

disable_kill_switch()

Manually disable kill switch (use with caution!).

Source code in src/ml4t/live/safety.py

def disable_kill_switch(self) -> None:
    """Manually disable kill switch (use with caution!)."""
    logger.warning("Kill switch DISABLED - proceed with caution!")
    self._state.kill_switch_activated = False
    self._state.kill_switch_reason = ""
    self.config.kill_switch_enabled = False
    self._save_state()
    self.record_event("kill_switch_disabled")

close_all_positions async

close_all_positions()

Emergency close all positions.

Returns:

Type	Description
list[Order]	List of close orders

Source code in src/ml4t/live/safety.py

async def close_all_positions(self) -> list[Order]:
    """Emergency close all positions.

    Returns:
        List of close orders
    """
    logger.warning("EMERGENCY: Closing ALL positions")
    orders = []
    for asset in list(self.positions.keys()):
        order = await self.close_position_async(asset)
        if order:
            orders.append(order)
    return orders

record_event

record_event(event, **payload)

Append a structured runtime event to the execution journal.

Source code in src/ml4t/live/safety.py

def record_event(self, event: str, **payload: Any) -> None:
    """Append a structured runtime event to the execution journal."""
    entry = {
        "timestamp": datetime.now(UTC).isoformat(),
        "event": event,
        "shadow_mode": self.config.shadow_mode,
        "execution_mode": self.execution_mode.value,
        "disabled_safety_controls": self._disabled_safety_controls(),
        "kill_switch": self._state.kill_switch_activated,
        "orders_placed": self._state.orders_placed,
        "daily_loss": self._state.daily_loss,
        "payload": redact_sensitive(self._json_safe(payload)),
    }

    try:
        self._audit_journal.append(entry)
        self._journal_error = None
    except AuditJournalError as error:
        self._journal_error = error
        logger.error("Audit journal append failed: %s", type(error).__name__)
        if self.config.fail_on_journal_error:
            raise

preview_reconciliation_async async

preview_reconciliation_async()

Build the current runtime reconciliation report without mutating state.

Source code in src/ml4t/live/safety.py

async def preview_reconciliation_async(self) -> dict[str, Any]:
    """Build the current runtime reconciliation report without mutating state."""
    live_positions, live_pending_orders = await self._capture_runtime_snapshot_async()
    return self._build_reconciliation_report(live_positions, live_pending_orders)

preflight_async async

preflight_async()

Probe broker reachability and startup reconciliation without persisting state.

Source code in src/ml4t/live/safety.py

async def preflight_async(self) -> dict[str, Any]:
    """Probe broker reachability and startup reconciliation without persisting state."""
    self._assert_persistence_ready()
    await self._broker.connect()
    try:
        self._validate_execution_identity()
        account_value = await self.get_account_value_async()
        cash = await self.get_cash_async()
        report = await self.preview_reconciliation_async()
        connected = await self._broker.is_connected_async()
        if not isinstance(connected, bool):
            raise BrokerSnapshotError("broker connectivity snapshot must be boolean")
        result = {
            "broker_connected": connected,
            "account_value": account_value,
            "cash": cash,
            "kill_switch_activated": self._state.kill_switch_activated,
            "kill_switch_reason": self._state.kill_switch_reason,
            "reconciliation": report,
            "journal_file": str(self._journal_path()),
            "state_file": self.config.state_file,
            "passed": connected
            and not self._state.kill_switch_activated
            and (report["clean"] or not self.config.fail_on_reconciliation_mismatch),
        }
        self.record_event(
            "preflight_completed",
            passed=result["passed"],
            reconciliation_clean=report["clean"],
            broker_connected=connected,
        )
        return result
    finally:
        await self._broker.disconnect()
        if self.execution_mode is not ExecutionMode.SHADOW:
            self._execution_identity_validated = False

connect async

connect()

Connect to broker and reconcile persisted state.

Source code in src/ml4t/live/safety.py

async def connect(self) -> None:
    """Connect to broker and reconcile persisted state."""
    self._assert_persistence_ready()
    await self._broker.connect()
    try:
        self._validate_execution_identity()
        connected = await self._broker.is_connected_async()
        if not isinstance(connected, bool) or not connected:
            raise BrokerSnapshotError("broker did not report a connected state")
        report = await self.preview_reconciliation_async()
    except BaseException:
        await self._broker.disconnect()
        if self.execution_mode is not ExecutionMode.SHADOW:
            self._execution_identity_validated = False
        raise
    self._last_reconciliation_report = report
    for original_id in report["resolved_replacement_gaps"]:
        self._state.replacement_gaps.pop(original_id, None)
    self._log_reconciliation_report(report)
    if report["clean"]:
        self.record_event("reconciliation_clean")
    else:
        self.record_event("reconciliation_mismatch", report=report)

    if (
        not report["clean"]
        and self.config.fail_on_reconciliation_mismatch
        and not self.config.shadow_mode
    ):
        self.record_event("reconciliation_blocked_startup", report=report)
        await self._broker.disconnect()
        raise ReconciliationMismatchError("Startup reconciliation mismatch blocked connect()")

    live_positions, live_pending_orders = await self._capture_runtime_snapshot_async()
    self._set_state_snapshot(live_positions, live_pending_orders)
    self._save_state()
    self.record_event("broker_connected")

disconnect async

disconnect()

Disconnect from broker and save state.

Source code in src/ml4t/live/safety.py

async def disconnect(self) -> None:
    """Disconnect from broker and save state."""
    try:
        try:
            live_positions, live_pending_orders = await self._capture_runtime_snapshot_async()
            self._set_state_snapshot(live_positions, live_pending_orders)
        except Exception as error:
            logger.warning(
                "Failed to capture broker snapshot during disconnect: %s",
                type(error).__name__,
            )
            self._refresh_state_snapshot_from_cache()
        self._save_state()
        await self._broker.disconnect()
        if self.execution_mode is not ExecutionMode.SHADOW:
            self._execution_identity_validated = False
        self.record_event("broker_disconnected")
    finally:
        self.close_persistence()

is_connected_async async

is_connected_async()

Check if connected (async).

Source code in src/ml4t/live/safety.py

async def is_connected_async(self) -> bool:
    """Check if connected (async)."""
    return await self._broker.is_connected_async()

get_positions_async async

get_positions_async()

Get all positions (async).

Returns:

Type	Description
dict[str, Position]	Dictionary mapping asset symbol to Position

Source code in src/ml4t/live/safety.py

async def get_positions_async(self) -> dict[str, Position]:
    """Get all positions (async).

    Returns:
        Dictionary mapping asset symbol to Position
    """
    if self.config.shadow_mode:
        return self._virtual_portfolio.positions
    return await self._broker.get_positions_async()

get_pending_orders_async async

get_pending_orders_async(asset=None)

Get pending orders (async).

Returns:

Type	Description
list[Order]	List of pending orders

Source code in src/ml4t/live/safety.py

async def get_pending_orders_async(self, asset: str | None = None) -> list[Order]:
    """Get pending orders (async).

    Returns:
        List of pending orders
    """
    if asset is None:
        return await self._broker.get_pending_orders_async()
    return await self._broker.get_pending_orders_async(asset)

get_position_async async

get_position_async(asset)

Get position (async).

Parameters:

Name	Type	Description	Default
asset	str	Asset symbol	required

Returns:

Type	Description
Position | None	Position object or None

Source code in src/ml4t/live/safety.py

async def get_position_async(self, asset: str) -> Position | None:
    """Get position (async).

    Args:
        asset: Asset symbol

    Returns:
        Position object or None
    """
    if self.config.shadow_mode:
        return self._virtual_portfolio.positions.get(asset)
    return await self._broker.get_position_async(asset)

ThreadSafeBrokerWrapper

ThreadSafeBrokerWrapper(
    async_broker, loop, strategy_runtime=None
)

Wraps an async broker for use from sync strategy code.

This wrapper is passed to Strategy.on_data() instead of the raw broker. It bridges the sync/async boundary by scheduling coroutines on the main event loop and blocking the worker thread until they complete.

Thread Safety: - Every portable strategy callback runs on one dedicated worker thread - Broker methods run on main event loop - run_coroutine_threadsafe() handles the cross-thread communication

Timeouts (from design review): - Getters (get_cash, get_account_value): 5s - Order operations (submit, cancel, close): 30s

Example

LiveEngine creates this wrapper

loop = asyncio.get_running_loop() wrapped = ThreadSafeBrokerWrapper(ib_broker, loop)

Strategy uses it like a normal sync broker

order = wrapped.submit_order('AAPL', 100, OrderSide.BUY)

Note

This class implements BrokerProtocol but does not inherit from it. It provides a sync interface backed by async operations.

Initialize thread-safe wrapper.

Parameters:

Name	Type	Description	Default
async_broker	AsyncBrokerProtocol	Async broker implementation (IBBroker, etc.)	required
loop	AbstractEventLoop	Main event loop (from asyncio.get_running_loop())	required

Source code in src/ml4t/live/wrappers.py

def __init__(
    self,
    async_broker: AsyncBrokerProtocol,
    loop: asyncio.AbstractEventLoop,
    strategy_runtime: LiveStrategyRuntime | None = None,
) -> None:
    """Initialize thread-safe wrapper.

    Args:
        async_broker: Async broker implementation (IBBroker, etc.)
        loop: Main event loop (from asyncio.get_running_loop())
    """
    self._broker = async_broker
    self._loop = loop
    self._loop_thread_id = threading.get_ident()
    self._strategy_runtime = strategy_runtime

positions property

positions

Get current positions (thread-safe read).

Returns:

Type	Description
dict[str, Position]	Dictionary mapping asset symbol to Position

pending_orders property

pending_orders

Get pending orders (thread-safe read).

Returns:

Type	Description
list[Order]	List of pending Order objects

is_connected property

is_connected

Check if broker is connected.

Returns:

Type	Description
bool	True if connected and ready to trade

get_position

get_position(asset)

Get position for specific asset.

Parameters:

Name	Type	Description	Default
asset	str	Asset symbol (e.g., "AAPL")	required

Returns:

Type	Description
Position | None	Position object if holding position, None otherwise

Raises:

Type	Description
TimeoutError	If operation times out
RuntimeError	If broker error occurs

Source code in src/ml4t/live/wrappers.py

def get_position(self, asset: str) -> Position | None:
    """Get position for specific asset.

    Args:
        asset: Asset symbol (e.g., "AAPL")

    Returns:
        Position object if holding position, None otherwise

    Raises:
        TimeoutError: If operation times out
        RuntimeError: If broker error occurs
    """
    # Can use positions property since it returns a copy
    return self.positions.get(asset)

get_positions

get_positions()

Get all current positions through the portable strategy facade.

Source code in src/ml4t/live/wrappers.py

def get_positions(self) -> dict[str, Position]:
    """Get all current positions through the portable strategy facade."""
    return self.positions

get_account_value

get_account_value()

Get total account value (cash + positions).

Returns:

Type	Description
float	Total account value in base currency

Raises:

Type	Description
TimeoutError	If operation times out (5s)
RuntimeError	If broker error occurs

Source code in src/ml4t/live/wrappers.py

def get_account_value(self) -> float:
    """Get total account value (cash + positions).

    Returns:
        Total account value in base currency

    Raises:
        TimeoutError: If operation times out (5s)
        RuntimeError: If broker error occurs
    """
    return self._run_sync(self._broker.get_account_value_async(), timeout=5.0)

get_cash

get_cash()

Get available cash balance.

Returns:

Type	Description
float	Available cash in base currency

Raises:

Type	Description
TimeoutError	If operation times out (5s)
RuntimeError	If broker error occurs

Source code in src/ml4t/live/wrappers.py

def get_cash(self) -> float:
    """Get available cash balance.

    Returns:
        Available cash in base currency

    Raises:
        TimeoutError: If operation times out (5s)
        RuntimeError: If broker error occurs
    """
    return self._run_sync(self._broker.get_cash_async(), timeout=5.0)

get_pending_orders

get_pending_orders(asset=None)

Get pending orders, optionally filtered by asset.

Source code in src/ml4t/live/wrappers.py

def get_pending_orders(self, asset: str | None = None) -> list[Order]:
    """Get pending orders, optionally filtered by asset."""
    operation = (
        self._broker.get_pending_orders_async()
        if asset is None
        else self._broker.get_pending_orders_async(asset)
    )
    return self._run_sync(operation, timeout=5.0)

submit_order

submit_order(
    asset,
    quantity,
    side=None,
    order_type=MARKET,
    limit_price=None,
    stop_price=None,
    **kwargs,
)

Submit order for execution.

Parameters:

Name	Type	Description	Default
asset	str	Asset symbol (e.g., "AAPL")	required
quantity	float	Signed shares/contracts when side is omitted; positive unsigned shares/contracts when side is provided	required
side	OrderSide | None	Order side (BUY/SELL), inferred from signed quantity if omitted	None
order_type	OrderType	Type of order (MARKET, LIMIT, STOP, etc.)	MARKET
limit_price	float | None	Limit price for LIMIT/STOP_LIMIT orders	None
stop_price	float | None	Stop price for STOP/STOP_LIMIT orders	None
**kwargs	Any	Additional broker-specific parameters	{}

Returns:

Type	Description
Order	Order object with order_id and initial status

Raises:

Type	Description
TimeoutError	If operation times out (30s)
ValueError	If order parameters are invalid
RuntimeError	If broker is not connected or error occurs

Source code in src/ml4t/live/wrappers.py

def submit_order(
    self,
    asset: str,
    quantity: float,
    side: OrderSide | None = None,
    order_type: OrderType = OrderType.MARKET,
    limit_price: float | None = None,
    stop_price: float | None = None,
    **kwargs: Any,
) -> Order:
    """Submit order for execution.

    Args:
        asset: Asset symbol (e.g., "AAPL")
        quantity: Signed shares/contracts when side is omitted; positive unsigned
            shares/contracts when side is provided
        side: Order side (BUY/SELL), inferred from signed quantity if omitted
        order_type: Type of order (MARKET, LIMIT, STOP, etc.)
        limit_price: Limit price for LIMIT/STOP_LIMIT orders
        stop_price: Stop price for STOP/STOP_LIMIT orders
        **kwargs: Additional broker-specific parameters

    Returns:
        Order object with order_id and initial status

    Raises:
        TimeoutError: If operation times out (30s)
        ValueError: If order parameters are invalid
        RuntimeError: If broker is not connected or error occurs
    """
    request = CanonicalOrderRequest.from_input(
        asset,
        quantity,
        side,
        order_type,
        limit_price,
        stop_price,
        capabilities=getattr(self._broker, "execution_capabilities", ()),
    )
    order = self._run_sync(
        self._broker.submit_order_async(
            request.asset,
            request.quantity,
            request.side,
            request.order_type,
            request.limit_price,
            request.stop_price,
            **kwargs,
        ),
        timeout=30.0,  # Orders need longer timeout
    )
    request.validate_result(order)
    if self._strategy_runtime is not None:
        self._strategy_runtime.observe_strategy_order(order)
    return order

cancel_order

cancel_order(order_id)

Cancel pending order.

Parameters:

Name	Type	Description	Default
order_id	str	ID of order to cancel	required

Returns:

Type	Description
bool	True if cancel request submitted, False if order not found

Raises:

Type	Description
TimeoutError	If operation times out (30s)
RuntimeError	If broker error occurs

Source code in src/ml4t/live/wrappers.py

def cancel_order(self, order_id: str) -> bool:
    """Cancel pending order.

    Args:
        order_id: ID of order to cancel

    Returns:
        True if cancel request submitted, False if order not found

    Raises:
        TimeoutError: If operation times out (30s)
        RuntimeError: If broker error occurs
    """
    return self._run_sync(self._broker.cancel_order_async(order_id), timeout=30.0)

replace_order

replace_order(
    order_id,
    *,
    quantity=None,
    limit_price=None,
    stop_price=None,
)

Replace a pending order with updated parameters.

Source code in src/ml4t/live/wrappers.py

def replace_order(
    self,
    order_id: str,
    *,
    quantity: float | None = None,
    limit_price: float | None = None,
    stop_price: float | None = None,
) -> Order:
    """Replace a pending order with updated parameters."""
    return self._run_sync(
        self._broker.replace_order_async(
            order_id,
            quantity=quantity,
            limit_price=limit_price,
            stop_price=stop_price,
        ),
        timeout=30.0,
    )

close_position

close_position(asset)

Close entire position in asset.

Convenience method that submits a closing order.

Parameters:

Name	Type	Description	Default
asset	str	Asset symbol to close	required

Returns:

Type	Description
Order | None	Order object if position exists, None if no position

Raises:

Type	Description
TimeoutError	If operation times out (30s)
RuntimeError	If broker error occurs

Source code in src/ml4t/live/wrappers.py

def close_position(self, asset: str) -> Order | None:
    """Close entire position in asset.

    Convenience method that submits a closing order.

    Args:
        asset: Asset symbol to close

    Returns:
        Order object if position exists, None if no position

    Raises:
        TimeoutError: If operation times out (30s)
        RuntimeError: If broker error occurs
    """
    return self._run_sync(self._broker.close_position_async(asset), timeout=30.0)

register_target_intent

register_target_intent(intent, *, position_rules=None)

Register a persistent target intent during a causal callback.

Source code in src/ml4t/live/wrappers.py

def register_target_intent(
    self,
    intent: CanonicalTargetIntent,
    *,
    position_rules: PositionRule | None = None,
) -> CanonicalTargetIntent:
    """Register a persistent target intent during a causal callback."""
    return self._runtime().register_target_intent(intent, position_rules=position_rules)

register_position_rule_policy

register_position_rule_policy(policy_id, rules)

Bind a portable position-rule policy to its client implementation.

Source code in src/ml4t/live/wrappers.py

def register_position_rule_policy(self, policy_id: str, rules: PositionRule) -> None:
    """Bind a portable position-rule policy to its client implementation."""
    self._runtime().register_position_rule_policy(policy_id, rules)

get_target_intents

get_target_intents()

Return registered portable target intents.

Source code in src/ml4t/live/wrappers.py

def get_target_intents(self) -> tuple[CanonicalTargetIntent, ...]:
    """Return registered portable target intents."""
    return self._runtime().targets

get_child_order_intents

get_child_order_intents()

Return lowered child-order intents.

Source code in src/ml4t/live/wrappers.py

def get_child_order_intents(self) -> tuple[CanonicalChildOrderIntent, ...]:
    """Return lowered child-order intents."""
    return self._runtime().children

get_intent_reconciliations

get_intent_reconciliations()

Return retained target execution evidence.

Source code in src/ml4t/live/wrappers.py

def get_intent_reconciliations(self) -> tuple[IntentReconciliation, ...]:
    """Return retained target execution evidence."""
    return self._runtime().reconciliations

export_target_intent_state

export_target_intent_state()

Return restart-safe target and position-rule state.

Source code in src/ml4t/live/wrappers.py

def export_target_intent_state(self) -> dict[str, Any]:
    """Return restart-safe target and position-rule state."""
    return self._runtime().to_state()

set_position_rules

set_position_rules(rules, asset=None)

Set client-evaluated position rules globally or for one asset.

Source code in src/ml4t/live/wrappers.py

def set_position_rules(self, rules: PositionRule | None, asset: str | None = None) -> None:
    """Set client-evaluated position rules globally or for one asset."""
    self._runtime().set_position_rules(rules, asset=asset)

clear_position_rules

clear_position_rules(asset=None)

Clear client-evaluated position rules globally or for one asset.

Source code in src/ml4t/live/wrappers.py

def clear_position_rules(self, asset: str | None = None) -> None:
    """Clear client-evaluated position rules globally or for one asset."""
    self._runtime().clear_position_rules(asset=asset)

update_position_context

update_position_context(asset, context)

Merge portable context used by position-rule evaluation.

Source code in src/ml4t/live/wrappers.py

def update_position_context(self, asset: str, context: dict[str, Any]) -> None:
    """Merge portable context used by position-rule evaluation."""
    self._runtime().update_position_context(asset, context)

Brokers

IBBroker

IBBroker(
    host="127.0.0.1",
    port=7497,
    client_id=1,
    account=None,
    market_data_type=None,
)

Interactive Brokers implementation.

Design: - All broker operations are async - Uses asyncio.Lock for thread safety - Event handlers use put_nowait() (non-blocking) - Reconnection handled externally

Connection Ports: - TWS Paper: 7497 - TWS Live: 7496 - Gateway Paper: 4002 - Gateway Live: 4001

Example

broker = IBBroker(port=7497) # Paper trading await broker.connect() positions = await broker.get_positions_async() await broker.disconnect()

Initialize IBBroker.

Parameters:

Name	Type	Description	Default
host	str	IB Gateway/TWS host (default: '127.0.0.1')	'127.0.0.1'
port	int	IB Gateway/TWS port (default: 7497 for paper)	7497
client_id	int	Unique client ID (default: 1)	1
account	str | None	IB account ID (default: use first account)	None
market_data_type	int | None	IB market-data type to request after connect. None (default) leaves TWS at its configured default - use this when the account has live Level 1 subscriptions. 1 = real-time, 2 = frozen, 3 = delayed, 4 = delayed-frozen. Paper accounts without market-data subscriptions typically need 3 to avoid order rejections.	None

Source code in src/ml4t/live/brokers/ib.py

def __init__(
    self,
    host: str = "127.0.0.1",
    port: int = 7497,  # Paper trading default
    client_id: int = 1,
    account: str | None = None,
    market_data_type: int | None = None,
) -> None:
    """Initialize IBBroker.

    Args:
        host: IB Gateway/TWS host (default: '127.0.0.1')
        port: IB Gateway/TWS port (default: 7497 for paper)
        client_id: Unique client ID (default: 1)
        account: IB account ID (default: use first account)
        market_data_type: IB market-data type to request after connect.
            ``None`` (default) leaves TWS at its configured default -
            use this when the account has live Level 1 subscriptions.
            ``1`` = real-time, ``2`` = frozen, ``3`` = delayed,
            ``4`` = delayed-frozen. Paper accounts without market-data
            subscriptions typically need ``3`` to avoid order rejections.
    """
    self._host = host
    self._port = port
    self._client_id = client_id
    self._account = account
    self._configured_account = account
    self._market_data_type = market_data_type

    self.ib = IB()
    self.ib.RequestTimeout = 60
    self.ib.RaiseRequestErrors = True
    self._connected = False

    # Thread-safe state with locks
    self._positions: dict[str, Position] = {}
    self._position_lock = asyncio.Lock()
    self._pending_orders: dict[str, Order] = {}
    self._order_lock = asyncio.Lock()

    # Order tracking
    self._order_counter = 0
    self._ib_order_map: dict[int, tuple[str, float]] = {}  # IB orderId -> (our_id, timestamp)
    self._snapshot_error: RuntimeError | None = None

    # Contract cache
    self._contracts: dict[str, Contract] = {}

execution_capabilities property

execution_capabilities

Return order behaviors implemented by this adapter.

is_connected property

is_connected

Check if connected to IB.

positions property

positions

Return a thread-safe snapshot of positions.

Note: This is called from worker thread via ThreadSafeBrokerWrapper. The lock prevents RuntimeError during dict iteration if IB callback modifies positions concurrently.

Returns:

Type	Description
dict[str, Position]	Dictionary mapping asset symbols to Position objects

pending_orders property

pending_orders

Get list of pending orders.

Returns:

Type	Description
list[Order]	List of pending Order objects

connect async

connect()

Connect to IB Gateway/TWS.

Raises:

Type	Description
RuntimeError	If connection fails
TimeoutError	If connection times out

Source code in src/ml4t/live/brokers/ib.py

async def connect(self) -> None:
    """Connect to IB Gateway/TWS.

    Raises:
        RuntimeError: If connection fails
        asyncio.TimeoutError: If connection times out
    """
    if self._connected:
        logger.info("IBBroker: Already connected")
        return

    logger.info(
        f"IBBroker: Connecting to {self._host}:{self._port} (client_id={self._client_id})"
    )
    self._snapshot_error = None

    callbacks_registered = False
    try:
        # Use outer timeout wrapper like production code
        await asyncio.wait_for(
            self.ib.connectAsync(
                host=self._host,
                port=self._port,
                clientId=self._client_id,
                account=self._account or "",  # Pass account like production
                timeout=IB_CONNECT_HANDSHAKE_TIMEOUT,
            ),
            timeout=IB_CONNECT_TIMEOUT,  # Outer timeout wrapper
        )
    except (TimeoutError, ConnectionRefusedError) as e:
        self.ib.disconnect()
        self._connected = False
        detail = str(redact_sensitive(str(e)))
        logger.error("IBBroker: Connection failed: %s", detail)
        raise RuntimeError(f"IB connection failed: {detail}") from None
    except Exception as e:
        self.ib.disconnect()
        self._connected = False
        detail = str(redact_sensitive(str(e)))
        logger.error("IBBroker: Unexpected connect error: %s", detail)
        raise RuntimeError(f"IB connection failed: {detail}") from None

    try:
        accounts = self.ib.managedAccounts()
        if not isinstance(accounts, list) or not accounts:
            raise RuntimeError("IB returned no managed account identity")
        if self._account is None:
            self._account = accounts[0]
        elif self._account not in accounts:
            raise RuntimeError("Configured IB account is not managed")

        # Subscribe before sync so updates cannot be missed between the snapshot and stream.
        self.ib.orderStatusEvent += self._on_order_status
        self.ib.positionEvent += self._on_position
        callbacks_registered = True

        if self._market_data_type is not None:
            self.ib.reqMarketDataType(self._market_data_type)
            logger.info(
                "IBBroker: market_data_type=%d "
                "(1=realtime, 2=frozen, 3=delayed, 4=delayed-frozen)",
                self._market_data_type,
            )

        # Bounded for the same reason the handshake above is. A Gateway can complete
        # the handshake and then stop answering requests, and reqPositionsAsync waits
        # on a reply that never arrives: the caller gets no error, no log line and no
        # timeout. A notebook run sat 75 minutes here before it was killed, holding the
        # connection and its scheduling slot the whole time. Failing loudly after
        # IB_SNAPSHOT_TIMEOUT reuses the handler below, which removes the callbacks,
        # disconnects and records the snapshot as unavailable.
        await asyncio.wait_for(self._sync_positions(), timeout=IB_SNAPSHOT_TIMEOUT)
        await asyncio.wait_for(self._sync_orders(), timeout=IB_SNAPSHOT_TIMEOUT)
    except Exception:
        if callbacks_registered:
            try:
                self.ib.orderStatusEvent -= self._on_order_status
                self.ib.positionEvent -= self._on_position
            except Exception:
                logger.exception("IBBroker: Failed to remove callbacks after connect failure")
        self.ib.disconnect()
        self._connected = False
        self._snapshot_error = RuntimeError("IB initial broker snapshot is unavailable")
        raise

    self._snapshot_error = None
    self._connected = True
    logger.info("IBBroker: Connected successfully")

disconnect async

disconnect()

Disconnect from IB.

Source code in src/ml4t/live/brokers/ib.py

async def disconnect(self) -> None:
    """Disconnect from IB."""
    if self._connected:
        self.ib.orderStatusEvent -= self._on_order_status
        self.ib.positionEvent -= self._on_position
        self.ib.disconnect()
        self._connected = False
        # Give time for socket cleanup to prevent zombie connections
        await asyncio.sleep(0.1)
        logger.info("IBBroker: Disconnected")

is_connected_async async

is_connected_async()

Return current adapter and vendor connection state.

Source code in src/ml4t/live/brokers/ib.py

async def is_connected_async(self) -> bool:
    """Return current adapter and vendor connection state."""
    return bool(self.is_connected)

assert_paper_trading

assert_paper_trading()

Fail unless the connected endpoint and managed account identify IB paper trading.

Source code in src/ml4t/live/brokers/ib.py

def assert_paper_trading(self) -> None:
    """Fail unless the connected endpoint and managed account identify IB paper trading."""
    if not self.is_connected or self._account is None:
        raise RuntimeError("IB paper identity requires a connected managed account")
    if self._port not in {4002, 7497}:
        raise RuntimeError("IB endpoint is not a standard paper-trading port")
    if re.fullmatch(r"DU[0-9]+", self._account.upper()) is None:
        raise RuntimeError("IB managed account is not identified as a paper account")

assert_live_trading

assert_live_trading()

Fail unless the connected live endpoint matches an explicitly selected account.

Source code in src/ml4t/live/brokers/ib.py

def assert_live_trading(self) -> None:
    """Fail unless the connected live endpoint matches an explicitly selected account."""
    if not self.is_connected or self._account is None:
        raise RuntimeError("IB live identity requires a connected managed account")
    if self._port not in {4001, 7496}:
        raise RuntimeError("IB endpoint is not a standard live-trading port")
    if self._configured_account is None:
        raise RuntimeError("IB live trading requires explicit account selection")
    if self._account.upper() != self._configured_account.upper():
        raise RuntimeError("IB connected account does not match the selected live account")
    if re.fullmatch(r"DU[0-9]+", self._account.upper()) is not None:
        raise RuntimeError("IB managed account is identified as a paper account")

get_position

get_position(asset)

Thread-safe single position access.

Parameters:

Name	Type	Description	Default
asset	str	Asset symbol	required

Returns:

Type	Description
Position | None	Position object if exists, None otherwise

Source code in src/ml4t/live/brokers/ib.py

def get_position(self, asset: str) -> Position | None:
    """Thread-safe single position access.

    Args:
        asset: Asset symbol

    Returns:
        Position object if exists, None otherwise
    """
    return self._positions.get(asset.upper())

get_positions_async async

get_positions_async()

Async thread-safe position access with lock.

Returns:

Type	Description
dict[str, Position]	Dictionary mapping asset symbols to Position objects

Source code in src/ml4t/live/brokers/ib.py

async def get_positions_async(self) -> dict[str, Position]:
    """Async thread-safe position access with lock.

    Returns:
        Dictionary mapping asset symbols to Position objects
    """
    self._raise_snapshot_error()
    async with self._position_lock:
        return dict(self._positions)

get_position_async async

get_position_async(asset)

Return one position from the synchronized adapter snapshot.

Source code in src/ml4t/live/brokers/ib.py

async def get_position_async(self, asset: str) -> Position | None:
    """Return one position from the synchronized adapter snapshot."""
    self._raise_snapshot_error()
    async with self._position_lock:
        return self._positions.get(asset.upper())

get_pending_orders_async async

get_pending_orders_async(asset=None)

Return pending orders, optionally filtered by asset.

Source code in src/ml4t/live/brokers/ib.py

async def get_pending_orders_async(self, asset: str | None = None) -> list[Order]:
    """Return pending orders, optionally filtered by asset."""
    self._raise_snapshot_error()
    orders = list(self._pending_orders.values())
    if asset is None:
        return orders
    normalized = asset.upper()
    return [order for order in orders if order.asset.upper() == normalized]

get_account_value_async async

get_account_value_async()

Get Net Liquidation Value.

Returns:

Type	Description
float	Account net liquidation value in USD

Source code in src/ml4t/live/brokers/ib.py

async def get_account_value_async(self) -> float:
    """Get Net Liquidation Value.

    Returns:
        Account net liquidation value in USD
    """
    for av in self.ib.accountValues():
        if (
            av.tag == "NetLiquidation"
            and av.currency == "USD"
            and (av.account == self._account or self._account is None)
        ):
            return self._validate_account_metric(av.value, "NetLiquidation")
    raise RuntimeError("IB NetLiquidation is unavailable for the configured account")

get_cash_async async

get_cash_async()

Get available funds.

Returns:

Type	Description
float	Available funds in USD

Source code in src/ml4t/live/brokers/ib.py

async def get_cash_async(self) -> float:
    """Get available funds.

    Returns:
        Available funds in USD
    """
    for av in self.ib.accountValues():
        if (
            av.tag == "AvailableFunds"
            and av.currency == "USD"
            and (av.account == self._account or self._account is None)
        ):
            return self._validate_account_metric(av.value, "AvailableFunds")
    raise RuntimeError("IB AvailableFunds is unavailable for the configured account")

submit_order_async async

submit_order_async(
    asset,
    quantity,
    side=None,
    order_type=MARKET,
    limit_price=None,
    stop_price=None,
    **kwargs,
)

Submit order to IB.

TASK-013: Full order submission implementation with IB order tracking.

Parameters:

Name	Type	Description	Default
asset	str	Asset symbol	required
quantity	float	Signed shares/contracts when side is omitted; positive unsigned shares/contracts when side is provided	required
side	OrderSide | None	BUY or SELL, inferred from signed quantity if omitted	None
order_type	OrderType	Market, limit, stop, or stop-limit	MARKET
limit_price	float | None	Limit price for limit orders	None
stop_price	float | None	Stop price for stop orders	None

Returns:

Type	Description
Order	Order object

Raises:

Type	Description
RuntimeError	If not connected
ValueError	If order parameters are invalid

Source code in src/ml4t/live/brokers/ib.py

async def submit_order_async(
    self,
    asset: str,
    quantity: float,
    side: OrderSide | None = None,
    order_type: OrderType = OrderType.MARKET,
    limit_price: float | None = None,
    stop_price: float | None = None,
    **kwargs: Any,
) -> Order:
    """Submit order to IB.

    TASK-013: Full order submission implementation with IB order tracking.

    Args:
        asset: Asset symbol
        quantity: Signed shares/contracts when side is omitted; positive unsigned
            shares/contracts when side is provided
        side: BUY or SELL, inferred from signed quantity if omitted
        order_type: Market, limit, stop, or stop-limit
        limit_price: Limit price for limit orders
        stop_price: Stop price for stop orders

    Returns:
        Order object

    Raises:
        RuntimeError: If not connected
        ValueError: If order parameters are invalid
    """
    if not self.is_connected:
        raise RuntimeError("Not connected to IB")

    request = CanonicalOrderRequest.from_input(
        asset,
        quantity,
        side,
        order_type,
        limit_price,
        stop_price,
        capabilities=self.execution_capabilities,
    )
    asset = request.asset
    quantity = request.quantity
    side = request.side
    order_type = request.order_type
    limit_price = request.limit_price
    stop_price = request.stop_price

    # Get contract
    contract = self._get_contract(asset)

    # Create IB order
    action = "BUY" if side == OrderSide.BUY else "SELL"
    outside_rth = bool(kwargs.get("outsideRth", False))
    order_ref = kwargs.get("order_ref")
    if order_ref is not None and (
        not isinstance(order_ref, str)
        or not order_ref
        or len(order_ref) > 32
        or not order_ref.replace("-", "").replace("_", "").isalnum()
    ):
        raise ValueError("order_ref must be 1-32 letters, numbers, hyphens, or underscores")
    if order_type == OrderType.MOC and outside_rth:
        raise ValueError("IB MOC orders do not support outsideRth=True")
    ib_order = self._create_ib_order(
        action, quantity, order_type, limit_price, stop_price, outside_rth
    )
    if order_ref is not None:
        ib_order.orderRef = order_ref

    # Submit atomically with lock
    async with self._order_lock:
        self._order_counter += 1
        order_id = f"ML4T-{self._order_counter}"

        # Place order with IB
        try:
            trade = self.ib.placeOrder(contract, ib_order)
        except Exception as e:
            # Surface a clear error; nothing has been tracked yet, so state stays consistent.
            detail = str(redact_sensitive(str(e)))
            raise RuntimeError(
                f"IBBroker: failed to place order for {asset}: {detail}"
            ) from None

        # Create our order
        order = Order(
            asset=asset,
            side=side,
            quantity=quantity,
            order_type=order_type,
            limit_price=limit_price,
            stop_price=stop_price,
            order_id=order_id,
            status=OrderStatus.PENDING,
            created_at=datetime.now(UTC),
        )

        # Track order
        self._pending_orders[order_id] = order
        self._ib_order_map[trade.order.orderId] = (order_id, time.time())

    logger.info(f"IBBroker: Order {order_id} submitted: {side.value} {quantity} {asset}")
    return order

cancel_order_async async

cancel_order_async(order_id)

Cancel pending order.

TASK-016: Full order cancellation implementation.

This method finds the IB order ID from our tracking map and cancels the order via the IB API. Handles edge cases like order not found or order already filled.

Parameters:

Name	Type	Description	Default
order_id	str	Order ID to cancel (e.g., 'ML4T-1')	required

Returns:

Type	Description
bool	True if cancellation request sent successfully, False otherwise

Note

The actual cancellation is confirmed via _on_order_status callback when IB sends the 'Cancelled' status update.

Source code in src/ml4t/live/brokers/ib.py

async def cancel_order_async(self, order_id: str) -> bool:
    """Cancel pending order.

    TASK-016: Full order cancellation implementation.

    This method finds the IB order ID from our tracking map and cancels
    the order via the IB API. Handles edge cases like order not found
    or order already filled.

    Args:
        order_id: Order ID to cancel (e.g., 'ML4T-1')

    Returns:
        True if cancellation request sent successfully, False otherwise

    Note:
        The actual cancellation is confirmed via _on_order_status callback
        when IB sends the 'Cancelled' status update.
    """
    # Find IB order ID from our tracking map
    ib_order_id = None
    for ib_id, (our_id, _) in self._ib_order_map.items():
        if our_id == order_id:
            ib_order_id = ib_id
            break

    if ib_order_id is None:
        logger.warning(f"IBBroker: Order {order_id} not found in tracking map")
        return False

    # Find the trade in open trades and cancel
    for trade in self.ib.openTrades():
        if trade.order.orderId == ib_order_id:
            self.ib.cancelOrder(trade.order)
            logger.info(f"IBBroker: Cancellation requested for order {order_id}")
            return True

    # Order not in open trades (possibly already filled or cancelled)
    logger.warning(f"IBBroker: Order {order_id} not found in open trades")
    return False

replace_order_async async

replace_order_async(
    order_id,
    *,
    quantity=None,
    limit_price=None,
    stop_price=None,
    **kwargs,
)

Replace a pending order via cancel-and-resubmit.

Source code in src/ml4t/live/brokers/ib.py

async def replace_order_async(
    self,
    order_id: str,
    *,
    quantity: float | None = None,
    limit_price: float | None = None,
    stop_price: float | None = None,
    **kwargs: Any,
) -> Order:
    """Replace a pending order via cancel-and-resubmit."""
    original = self._pending_orders.get(order_id)
    if original is None:
        raise RuntimeError(f"Order {order_id} not found in pending orders")

    replacement_quantity = original.quantity if quantity is None else quantity
    replacement_limit = original.limit_price if limit_price is None else limit_price
    replacement_stop = original.stop_price if stop_price is None else stop_price

    cancelled = await self.cancel_order_async(order_id)
    if not cancelled:
        raise RuntimeError(f"Failed to cancel order {order_id} before replacement")

    return await self.submit_order_async(
        asset=original.asset,
        quantity=replacement_quantity,
        side=original.side,
        order_type=original.order_type,
        limit_price=replacement_limit,
        stop_price=replacement_stop,
        **kwargs,
    )

close_position_async async

close_position_async(asset)

Close position in asset.

Parameters:

Name	Type	Description	Default
asset	str	Asset symbol	required

Returns:

Type	Description
Order | None	Order object if position exists, None otherwise

Raises:

Type	Description
NotImplementedError	Depends on TASK-013

Source code in src/ml4t/live/brokers/ib.py

async def close_position_async(self, asset: str) -> Order | None:
    """Close position in asset.

    Args:
        asset: Asset symbol

    Returns:
        Order object if position exists, None otherwise

    Raises:
        NotImplementedError: Depends on TASK-013
    """
    pos = self.get_position(asset)
    if not pos or pos.quantity == 0:
        return None

    side = OrderSide.SELL if pos.quantity > 0 else OrderSide.BUY
    return await self.submit_order_async(asset, abs(pos.quantity), side)

AlpacaBroker

AlpacaBroker(api_key, secret_key, paper=True)

Alpaca Markets broker implementation.

Design (matching IBBroker patterns): - All broker operations are async - Uses asyncio.Lock for thread safety - WebSocket stream for real-time order updates - REST API for account/position queries and order submission

Paper vs Live: - paper=True (default): Uses paper trading endpoint - paper=False: Uses live trading endpoint (USE WITH CAUTION)

Example

broker = AlpacaBroker( api_key='PKXXXXXXXX', secret_key='XXXXXXXXXX', paper=True, # Always start with paper trading! ) await broker.connect() positions = await broker.get_positions_async() await broker.disconnect()

Initialize AlpacaBroker.

Parameters:

Name	Type	Description	Default
api_key	str	Alpaca API key (from https://app.alpaca.markets)	required
secret_key	str	Alpaca secret key	required
paper	bool	Use paper trading endpoint (default: True)	True

Source code in src/ml4t/live/brokers/alpaca.py

def __init__(
    self,
    api_key: str,
    secret_key: str,
    paper: bool = True,  # Paper trading by default (SAFETY)
) -> None:
    """Initialize AlpacaBroker.

    Args:
        api_key: Alpaca API key (from https://app.alpaca.markets)
        secret_key: Alpaca secret key
        paper: Use paper trading endpoint (default: True)
    """
    self._api_key = api_key
    self._secret_key = secret_key
    self._paper = paper

    # Clients (created in connect())
    self._trading_client: TradingClient | None = None
    self._trading_stream: TradingStream | None = None
    self._stream_task: asyncio.Task | None = None

    # Connection state
    self._connected = False

    # Thread-safe state with locks (matching IBBroker pattern)
    self._positions: dict[str, Position] = {}
    self._position_lock = asyncio.Lock()
    self._pending_orders: dict[str, Order] = {}
    self._order_lock = asyncio.Lock()

    # Order tracking (matching IBBroker pattern)
    self._order_counter = 0
    # Alpaca order ID (UUID string) -> (our_id, timestamp)
    self._alpaca_order_map: dict[str, tuple[str, float]] = {}
    self._account_id: str | None = None
    self._snapshot_error: RuntimeError | None = None
    self._position_snapshot_poisoned = False

execution_capabilities property

execution_capabilities

Return order behaviors implemented by this adapter.

is_connected property

is_connected

Check if connected to Alpaca.

positions property

positions

Thread-safe position access (shallow copy).

Note: This is called from worker thread via ThreadSafeBrokerWrapper. The shallow copy prevents RuntimeError during dict iteration.

Returns:

Type	Description
dict[str, Position]	Dictionary mapping asset symbols to Position objects

pending_orders property

pending_orders

Get list of pending orders.

Returns:

Type	Description
list[Order]	List of pending Order objects

connect async

connect()

Connect to Alpaca and sync initial state.

Steps: 1. Create TradingClient (REST) 2. Create TradingStream (WebSocket) 3. Verify connection by fetching account 4. Register trade update callback 5. Sync positions and open orders 6. Start WebSocket stream for order updates

Raises:

Type	Description
RuntimeError	If connection fails

Source code in src/ml4t/live/brokers/alpaca.py

async def connect(self) -> None:
    """Connect to Alpaca and sync initial state.

    Steps:
    1. Create TradingClient (REST)
    2. Create TradingStream (WebSocket)
    3. Verify connection by fetching account
    4. Register trade update callback
    5. Sync positions and open orders
    6. Start WebSocket stream for order updates

    Raises:
        RuntimeError: If connection fails
    """
    if self._connected:
        logger.info("AlpacaBroker: Already connected")
        return

    mode = "paper" if self._paper else "LIVE"
    logger.info(f"AlpacaBroker: Connecting ({mode} trading)")
    self._snapshot_error = None
    self._position_snapshot_poisoned = False

    try:
        # Create REST client
        self._trading_client = TradingClient(
            api_key=self._api_key,
            secret_key=self._secret_key,
            paper=self._paper,
        )

        # Verify connection by fetching account
        account = self._trading_client.get_account()
        account_id = getattr(account, "account_number", None)
        if not isinstance(account_id, str) or not account_id.strip():
            raise RuntimeError("Alpaca returned no account identity")
        self._account_id = account_id.strip()
        equity = self._validate_account_metric(account, "equity")
        cash = self._validate_account_metric(account, "cash", allow_negative=True)
        logger.info(
            f"AlpacaBroker: Account verified - equity=${equity:,.2f}, cash=${cash:,.2f}"
        )

        # Create WebSocket stream for order updates
        self._trading_stream = TradingStream(
            api_key=self._api_key,
            secret_key=self._secret_key,
            paper=self._paper,
        )

        # Subscribe to trade updates BEFORE initial sync (IBBroker pattern)
        self._trading_stream.subscribe_trade_updates(self._on_trade_update)

        # Initial sync
        await self._sync_positions()
        await self._sync_orders()

        # Start stream in background task
        self._stream_task = asyncio.create_task(self._run_trading_stream())

        self._connected = True
        logger.info("AlpacaBroker: Connected successfully")

    except Exception as e:
        detail = str(redact_sensitive(str(e)))
        logger.error("AlpacaBroker: Connection failed: %s", detail)
        if self._stream_task is not None:
            self._stream_task.cancel()
        if self._stream_task is not None and self._trading_stream is not None:
            try:
                self._trading_stream.stop()
            except Exception as stop_error:
                logger.error(
                    "AlpacaBroker: Failed to stop stream after connect failure: %s",
                    redact_sensitive(str(stop_error)),
                )
        self._connected = False
        self._snapshot_error = RuntimeError("Alpaca initial broker snapshot is unavailable")
        self._account_id = None
        self._trading_client = None
        self._trading_stream = None
        self._stream_task = None
        raise RuntimeError(f"Failed to connect to Alpaca: {detail}") from None

disconnect async

disconnect()

Disconnect from Alpaca.

Source code in src/ml4t/live/brokers/alpaca.py

async def disconnect(self) -> None:
    """Disconnect from Alpaca."""
    if not self._connected:
        return

    stream_task = self._stream_task
    if self._trading_stream:
        try:
            deadline = time.monotonic() + 2.0
            stream_loop = getattr(self._trading_stream, "_loop", None)
            while (
                stream_task is not None
                and not stream_task.done()
                and stream_loop is None
                and time.monotonic() < deadline
            ):
                await asyncio.sleep(0.01)
                stream_loop = getattr(self._trading_stream, "_loop", None)
            self._trading_stream.stop()
        except Exception as e:
            logger.warning("AlpacaBroker: Error stopping stream: %s", redact_sensitive(str(e)))
    if stream_task and not stream_task.done():
        try:
            await asyncio.wait_for(stream_task, timeout=5.0)
        except TimeoutError:
            stream_task.cancel()
            try:
                await stream_task
            except asyncio.CancelledError:
                pass

    self._connected = False
    self._account_id = None
    self._trading_client = None
    self._trading_stream = None
    self._stream_task = None

    logger.info("AlpacaBroker: Disconnected")

is_connected_async async

is_connected_async()

Return current REST and adapter connection state.

Source code in src/ml4t/live/brokers/alpaca.py

async def is_connected_async(self) -> bool:
    """Return current REST and adapter connection state."""
    return bool(self.is_connected)

assert_paper_trading

assert_paper_trading()

Fail unless the connected client is authenticated through Alpaca's paper endpoint.

Source code in src/ml4t/live/brokers/alpaca.py

def assert_paper_trading(self) -> None:
    """Fail unless the connected client is authenticated through Alpaca's paper endpoint."""
    if not self.is_connected or self._trading_client is None or self._account_id is None:
        raise RuntimeError("Alpaca paper identity requires a connected account")
    if self._paper is not True:
        raise RuntimeError("Alpaca broker is configured for live trading")
    if getattr(self._trading_client, "_sandbox", None) is not True:
        raise RuntimeError("Alpaca client did not confirm sandbox mode")
    if getattr(self._trading_client, "_base_url", None) != BaseURL.TRADING_PAPER:
        raise RuntimeError("Alpaca client is not using the official paper endpoint")

assert_live_trading

assert_live_trading()

Fail unless the connected client is authenticated through Alpaca's live endpoint.

Source code in src/ml4t/live/brokers/alpaca.py

def assert_live_trading(self) -> None:
    """Fail unless the connected client is authenticated through Alpaca's live endpoint."""
    if not self.is_connected or self._trading_client is None or self._account_id is None:
        raise RuntimeError("Alpaca live identity requires a connected account")
    if self._paper is not False:
        raise RuntimeError("Alpaca broker is configured for paper trading")
    if getattr(self._trading_client, "_sandbox", None) is not False:
        raise RuntimeError("Alpaca client did not confirm live mode")
    if getattr(self._trading_client, "_base_url", None) != BaseURL.TRADING_LIVE:
        raise RuntimeError("Alpaca client is not using the official live endpoint")

get_position

get_position(asset)

Thread-safe single position access.

Parameters:

Name	Type	Description	Default
asset	str	Asset symbol (e.g., 'AAPL' or 'BTC/USD')	required

Returns:

Type	Description
Position | None	Position object if exists, None otherwise

Source code in src/ml4t/live/brokers/alpaca.py

def get_position(self, asset: str) -> Position | None:
    """Thread-safe single position access.

    Args:
        asset: Asset symbol (e.g., 'AAPL' or 'BTC/USD')

    Returns:
        Position object if exists, None otherwise
    """
    if self._position_snapshot_poisoned:
        self._raise_snapshot_error()
    return self._positions.get(asset.upper())

get_positions_async async

get_positions_async()

Async thread-safe position access with lock.

Returns:

Type	Description
dict[str, Position]	Dictionary mapping asset symbols to Position objects

Source code in src/ml4t/live/brokers/alpaca.py

async def get_positions_async(self) -> dict[str, Position]:
    """Async thread-safe position access with lock.

    Returns:
        Dictionary mapping asset symbols to Position objects
    """
    self._raise_snapshot_error()
    async with self._position_lock:
        return dict(self._positions)

get_position_async async

get_position_async(asset)

Return one position from the synchronized adapter snapshot.

Source code in src/ml4t/live/brokers/alpaca.py

async def get_position_async(self, asset: str) -> Position | None:
    """Return one position from the synchronized adapter snapshot."""
    self._raise_snapshot_error()
    async with self._position_lock:
        return self._positions.get(asset.upper())

get_pending_orders_async async

get_pending_orders_async(asset=None)

Return pending orders, optionally filtered by asset.

Source code in src/ml4t/live/brokers/alpaca.py

async def get_pending_orders_async(self, asset: str | None = None) -> list[Order]:
    """Return pending orders, optionally filtered by asset."""
    self._raise_snapshot_error()
    orders = list(self._pending_orders.values())
    if asset is None:
        return orders
    normalized = asset.upper()
    return [order for order in orders if order.asset.upper() == normalized]

get_account_value_async async

get_account_value_async()

Get portfolio value (equity).

Returns:

Type	Description
float	Total account equity in USD

Source code in src/ml4t/live/brokers/alpaca.py

async def get_account_value_async(self) -> float:
    """Get portfolio value (equity).

    Returns:
        Total account equity in USD
    """
    if not self._trading_client:
        raise RuntimeError("Alpaca trading client is unavailable")

    account = self._trading_client.get_account()
    return self._validate_account_metric(account, "equity")

get_cash_async async

get_cash_async()

Get the signed cash balance.

Returns:

Type	Description
float	Cash balance in USD. Margin accounts may report a negative balance.

Source code in src/ml4t/live/brokers/alpaca.py

async def get_cash_async(self) -> float:
    """Get the signed cash balance.

    Returns:
        Cash balance in USD. Margin accounts may report a negative balance.
    """
    if not self._trading_client:
        raise RuntimeError("Alpaca trading client is unavailable")

    account = self._trading_client.get_account()
    return self._validate_account_metric(account, "cash", allow_negative=True)

submit_order_async async

submit_order_async(
    asset,
    quantity,
    side=None,
    order_type=MARKET,
    limit_price=None,
    stop_price=None,
    **kwargs,
)

Submit order to Alpaca.

Parameters:

Name	Type	Description	Default
asset	str	Asset symbol (e.g., 'AAPL' or 'BTC/USD')	required
quantity	float	Signed shares/units when side is omitted; positive unsigned shares/units when side is provided	required
side	OrderSide | None	BUY or SELL, inferred from signed quantity if omitted	None
order_type	OrderType	Market, limit, stop, or stop-limit	MARKET
limit_price	float | None	Limit price for limit orders	None
stop_price	float | None	Stop price for stop orders	None
**kwargs	Any	Additional parameters (ignored)	{}

Returns:

Type	Description
Order	Order object

Raises:

Type	Description
RuntimeError	If not connected
ValueError	If order parameters are invalid

Source code in src/ml4t/live/brokers/alpaca.py

async def submit_order_async(
    self,
    asset: str,
    quantity: float,
    side: OrderSide | None = None,
    order_type: OrderType = OrderType.MARKET,
    limit_price: float | None = None,
    stop_price: float | None = None,
    **kwargs: Any,
) -> Order:
    """Submit order to Alpaca.

    Args:
        asset: Asset symbol (e.g., 'AAPL' or 'BTC/USD')
        quantity: Signed shares/units when side is omitted; positive unsigned
            shares/units when side is provided
        side: BUY or SELL, inferred from signed quantity if omitted
        order_type: Market, limit, stop, or stop-limit
        limit_price: Limit price for limit orders
        stop_price: Stop price for stop orders
        **kwargs: Additional parameters (ignored)

    Returns:
        Order object

    Raises:
        RuntimeError: If not connected
        ValueError: If order parameters are invalid
    """
    if not self.is_connected or not self._trading_client:
        raise RuntimeError("Not connected to Alpaca")

    request = CanonicalOrderRequest.from_input(
        asset,
        quantity,
        side,
        order_type,
        limit_price,
        stop_price,
        capabilities=self.execution_capabilities,
    )
    asset = request.asset
    qty = request.quantity
    side = request.side
    order_type = request.order_type
    limit_price = request.limit_price
    stop_price = request.stop_price
    client_order_id = kwargs.get("client_order_id")
    if client_order_id is not None and (
        not isinstance(client_order_id, str)
        or not client_order_id
        or len(client_order_id) > 48
        or not client_order_id.replace("-", "").replace("_", "").isalnum()
    ):
        raise ValueError(
            "client_order_id must be 1-48 letters, numbers, hyphens, or underscores"
        )
    if order_type == OrderType.MOC:
        if kwargs.get("extended_hours"):
            raise ValueError("Alpaca MOC orders do not support extended_hours=True")
        if "/" in asset:
            raise NotImplementedError("Alpaca MOC orders are only supported for US equities")
        if not qty.is_integer():
            raise ValueError("Alpaca MOC orders require whole-share quantities")

    # Create order request
    order_request = self._create_order_request(
        asset,
        qty,
        side,
        order_type,
        limit_price,
        stop_price,
        client_order_id=client_order_id,
    )

    # Submit atomically with lock (IBBroker pattern)
    async with self._order_lock:
        self._order_counter += 1
        order_id = f"ML4T-{self._order_counter}"

        # Submit to Alpaca
        alpaca_order = self._trading_client.submit_order(order_request)

        # Create our order object
        order = Order(
            asset=asset,
            side=side,
            quantity=qty,
            order_type=order_type,
            limit_price=limit_price,
            stop_price=stop_price,
            order_id=order_id,
            status=self._map_order_status(alpaca_order.status),
            created_at=alpaca_order.created_at or datetime.now(UTC),
        )

        if alpaca_order.status in {
            AlpacaOrderStatus.FILLED,
            AlpacaOrderStatus.PARTIALLY_FILLED,
        }:
            if alpaca_order.filled_qty is None or alpaca_order.filled_avg_price is None:
                raise RuntimeError("Alpaca material fill is missing fill evidence")
            try:
                filled_quantity = float(alpaca_order.filled_qty)
                filled_price = float(alpaca_order.filled_avg_price)
            except (TypeError, ValueError) as error:
                raise RuntimeError(
                    "Alpaca material fill contains non-numeric evidence"
                ) from error
            if (
                not math.isfinite(filled_quantity)
                or filled_quantity <= 0
                or filled_quantity > order.quantity
                or (
                    alpaca_order.status is AlpacaOrderStatus.FILLED
                    and filled_quantity != order.quantity
                )
                or not math.isfinite(filled_price)
                or filled_price <= 0
            ):
                raise RuntimeError("Alpaca material fill contains invalid evidence")
            order.filled_quantity = filled_quantity
            order.filled_price = filled_price
            order.filled_at = getattr(alpaca_order, "filled_at", None) or datetime.now(UTC)

        # Track order
        if order.status is OrderStatus.PENDING:
            self._pending_orders[order_id] = order
            self._alpaca_order_map[str(alpaca_order.id)] = (order_id, time.time())

    if order.filled_quantity > 0:
        try:
            await self._sync_positions()
        except Exception as error:
            self._poison_snapshot("position", error)

    logger.info(f"AlpacaBroker: Order {order_id} submitted: {side.value} {qty} {asset}")
    return order

cancel_order_async async

cancel_order_async(order_id)

Cancel pending order.

Parameters:

Name	Type	Description	Default
order_id	str	Order ID to cancel (e.g., 'ML4T-1')	required

Returns:

Type	Description
bool	True if cancellation request sent successfully, False otherwise

Source code in src/ml4t/live/brokers/alpaca.py

async def cancel_order_async(self, order_id: str) -> bool:
    """Cancel pending order.

    Args:
        order_id: Order ID to cancel (e.g., 'ML4T-1')

    Returns:
        True if cancellation request sent successfully, False otherwise
    """
    if not self._trading_client:
        return False

    # Find Alpaca order ID from our tracking map
    alpaca_order_id = None
    for alpaca_id, (our_id, _) in self._alpaca_order_map.items():
        if our_id == order_id:
            alpaca_order_id = alpaca_id
            break

    if alpaca_order_id is None:
        logger.warning(f"AlpacaBroker: Order {order_id} not found in tracking map")
        return False

    try:
        self._trading_client.cancel_order_by_id(alpaca_order_id)
        logger.info(f"AlpacaBroker: Cancellation requested for order {order_id}")
        return True
    except Exception as e:
        logger.warning(
            "AlpacaBroker: Failed to cancel order %s: %s",
            order_id,
            redact_sensitive(str(e)),
        )
        return False

replace_order_async async

replace_order_async(
    order_id,
    *,
    quantity=None,
    limit_price=None,
    stop_price=None,
    **kwargs,
)

Replace a pending order via cancel-and-resubmit.

Source code in src/ml4t/live/brokers/alpaca.py

async def replace_order_async(
    self,
    order_id: str,
    *,
    quantity: float | None = None,
    limit_price: float | None = None,
    stop_price: float | None = None,
    **kwargs: Any,
) -> Order:
    """Replace a pending order via cancel-and-resubmit."""
    original = self._pending_orders.get(order_id)
    if original is None:
        raise RuntimeError(f"Order {order_id} not found in pending orders")

    replacement_quantity = original.quantity if quantity is None else quantity
    replacement_limit = original.limit_price if limit_price is None else limit_price
    replacement_stop = original.stop_price if stop_price is None else stop_price

    cancelled = await self.cancel_order_async(order_id)
    if not cancelled:
        raise RuntimeError(f"Failed to cancel order {order_id} before replacement")

    return await self.submit_order_async(
        asset=original.asset,
        quantity=replacement_quantity,
        side=original.side,
        order_type=original.order_type,
        limit_price=replacement_limit,
        stop_price=replacement_stop,
        **kwargs,
    )

close_position_async async

close_position_async(asset)

Close position in asset.

Parameters:

Name	Type	Description	Default
asset	str	Asset symbol	required

Returns:

Type	Description
Order | None	Order object if position exists, None otherwise

Source code in src/ml4t/live/brokers/alpaca.py

async def close_position_async(self, asset: str) -> Order | None:
    """Close position in asset.

    Args:
        asset: Asset symbol

    Returns:
        Order object if position exists, None otherwise
    """
    pos = self.get_position(asset)
    if not pos or pos.quantity == 0:
        return None

    side = OrderSide.SELL if pos.quantity > 0 else OrderSide.BUY
    return await self.submit_order_async(asset, abs(pos.quantity), side)

Data Feeds And Aggregation

IBDataFeed

IBDataFeed(
    ib,
    symbols,
    *,
    exchange="SMART",
    currency="USD",
    tick_throttle_ms=100,
    queue_capacity=1024,
    experimental=False,
)

Bases: DataFeedProtocol

Experimental real-time market data feed from Interactive Brokers.

Subscribes to tick-by-tick market data for specified symbols. Emits validated trade and quote events.

Data Format

MarketEvent trade and quote snapshots with UTC provider or receipt time.

Note

• IB must be connected before creating feed
• Requires market data subscription for symbols
• Throttles rapid ticks to avoid overwhelming strategy

Example

ib = IB() await ib.connectAsync('127.0.0.1', 7497, clientId=1)

feed = IBDataFeed(ib, symbols=['SPY', 'QQQ', 'IWM'], experimental=True) await feed.start()

Use directly or wrap with BarAggregator

aggregator = BarAggregator(feed, bar_size_minutes=1)

Initialize IB data feed.

Parameters:

Name	Type	Description	Default
ib	IB	Connected IB instance	required
symbols	list[str]	List of symbols to subscribe to	required
exchange	str	IB exchange (default: SMART routing)	'SMART'
currency	str	Currency (default: USD)	'USD'
tick_throttle_ms	int	Minimum milliseconds between tick emissions (prevents overwhelming strategy with rapid ticks)	100
queue_capacity	int	Maximum pending events before a fail-closed overflow.	1024
experimental	bool	Must be true to acknowledge the unsupported feed contract.	False

Source code in src/ml4t/live/feeds/ib_feed.py

def __init__(
    self,
    ib: IB,
    symbols: list[str],
    *,
    exchange: str = "SMART",
    currency: str = "USD",
    tick_throttle_ms: int = 100,  # Min time between emits
    queue_capacity: int = 1_024,
    experimental: bool = False,
) -> None:
    """Initialize IB data feed.

    Args:
        ib: Connected IB instance
        symbols: List of symbols to subscribe to
        exchange: IB exchange (default: SMART routing)
        currency: Currency (default: USD)
        tick_throttle_ms: Minimum milliseconds between tick emissions
            (prevents overwhelming strategy with rapid ticks)
        queue_capacity: Maximum pending events before a fail-closed overflow.
        experimental: Must be true to acknowledge the unsupported feed contract.
    """
    require_experimental_opt_in(
        "IBDataFeed",
        experimental=experimental,
        missing_guarantees=IB_MISSING_GUARANTEES,
    )
    if not symbols or any(
        not isinstance(symbol, str) or not symbol.strip() for symbol in symbols
    ):
        raise ValueError("symbols must contain at least one non-empty symbol")
    if (
        isinstance(tick_throttle_ms, bool)
        or not isinstance(tick_throttle_ms, int | float)
        or not math.isfinite(tick_throttle_ms)
        or tick_throttle_ms < 0
    ):
        raise ValueError("tick_throttle_ms must be finite and non-negative")
    self.ib = ib
    self.symbols = list(symbols)
    self.exchange = exchange
    self.currency = currency
    self.tick_throttle_ms = tick_throttle_ms

    # State
    self.queue_capacity = queue_capacity
    self._queue = BoundedEventQueue(capacity=queue_capacity, feed="interactive_brokers")
    self._running = False
    self._contracts: dict[str, Stock] = {}
    self._tickers: dict[str, Ticker] = {}
    self._last_emit_time = 0.0
    self._callback_registered = False
    self.max_event_age_seconds = 5.0

    # Statistics
    self._tick_count = 0
    self._throttled_count = 0
    self._rejected_count = 0

stats property

stats

Get feed statistics.

Returns:

Type	Description
dict[str, Any]	Dict with keys:
dict[str, Any]	running: bool
dict[str, Any]	tick_count: int - Total ticks received
dict[str, Any]	throttled_count: int - Ticks throttled
dict[str, Any]	symbols: list[str] - Subscribed symbols

start async

start()

Subscribe to market data for all symbols.

Creates contracts and subscribes to real-time tick data.

Raises:

Type	Description
RuntimeError	If IB not connected

Source code in src/ml4t/live/feeds/ib_feed.py

async def start(self) -> None:
    """Subscribe to market data for all symbols.

    Creates contracts and subscribes to real-time tick data.

    Raises:
        RuntimeError: If IB not connected
    """
    if not self.ib.isConnected():
        raise RuntimeError("IB must be connected before starting feed")
    if self._running:
        return

    logger.info(f"IBDataFeed: Starting feed for {len(self.symbols)} symbols")
    self._queue = BoundedEventQueue(
        capacity=self.queue_capacity,
        feed="interactive_brokers",
    )
    self._contracts.clear()
    self._tickers.clear()
    self._running = True

    # Create contracts
    for symbol in self.symbols:
        contract = Stock(symbol, self.exchange, self.currency)
        self._contracts[symbol] = contract

        # Qualify contract (ensure IB recognizes it)
        qualified = await self.ib.qualifyContractsAsync(contract)
        if not qualified:
            logger.warning(f"IBDataFeed: Could not qualify contract for {symbol}")
            continue

        # Request market data
        ticker = self.ib.reqMktData(contract, "", False, False)
        self._tickers[symbol] = ticker

    # Register callback for ticker updates
    self.ib.pendingTickersEvent += self._on_pending_tickers
    self._callback_registered = True

    logger.info(f"IBDataFeed: Subscribed to {len(self._tickers)} symbols")

stop

stop()

Unsubscribe from market data.

Cancels all market data subscriptions and stops feed.

Source code in src/ml4t/live/feeds/ib_feed.py

def stop(self) -> None:
    """Unsubscribe from market data.

    Cancels all market data subscriptions and stops feed.
    """
    logger.info("IBDataFeed: Stopping feed")
    self._running = False

    # Unsubscribe from all tickers
    for symbol, contract in self._contracts.items():
        try:
            self.ib.cancelMktData(contract)
        except Exception as e:
            logger.warning(
                "IBDataFeed: Error canceling %s: %s",
                symbol,
                redact_sensitive(str(e)),
            )

    # Remove callback
    if self._callback_registered:
        self.ib.pendingTickersEvent -= self._on_pending_tickers
        self._callback_registered = False

    # Signal consumer to exit
    self._queue.finish(discard=True)

    logger.info(
        f"IBDataFeed: Stopped. Ticks: {self._tick_count}, Throttled: {self._throttled_count}"
    )

__aiter__ async

__aiter__()

Async iterator yielding market data.

Yields:

Type	Description
AsyncIterator[MarketEvent]	Validated trade and quote events.

Stops when

• stop() is called (None sentinel)
• Feed is not running

Source code in src/ml4t/live/feeds/ib_feed.py

async def __aiter__(self) -> AsyncIterator[MarketEvent]:
    """Async iterator yielding market data.

    Yields:
        Validated trade and quote events.

    Stops when:
        - stop() is called (None sentinel)
        - Feed is not running
    """
    while True:
        item = await self._queue.get()

        # None sentinel signals shutdown
        if item is None:
            break

        yield item

AlpacaDataFeed

AlpacaDataFeed(
    api_key,
    secret_key,
    symbols,
    *,
    data_type="bars",
    feed="iex",
    queue_capacity=1024,
    experimental=False,
)

Bases: DataFeedProtocol

Experimental real-time market data feed from Alpaca Markets.

Subscribes to real-time data for specified symbols. Supports both stocks and crypto.

Data Types

bars: OHLCV minute bars (default, recommended for strategies) quotes: Bid/ask quotes (for spread-sensitive strategies) trades: Individual trades (highest frequency)

Data Feeds

iex: Free tier (limited data) sip: Premium (full market data, requires subscription)

Data Format

Validated MarketEvent bars, quotes, or trades with UTC event and receipt times.

Example

Stocks only

feed = AlpacaDataFeed( api_key='PKXXXXXXXX', secret_key='XXXXXXXXXX', symbols=['AAPL', 'MSFT'], experimental=True, )

Mixed stocks and crypto

feed = AlpacaDataFeed( api_key='PKXXXXXXXX', secret_key='XXXXXXXXXX', symbols=['AAPL', 'BTC/USD', 'ETH/USD'], experimental=True, )

await feed.start()

async for event in feed: consume(event)

Initialize Alpaca data feed.

Parameters:

Name	Type	Description	Default
api_key	str	Alpaca API key	required
secret_key	str	Alpaca secret key	required
symbols	list[str]	List of symbols (e.g., ['AAPL', 'BTC/USD'])	required
data_type	str	Type of data - 'bars' (default), 'quotes', or 'trades'	'bars'
feed	str	Data feed type - 'iex' (free) or 'sip' (premium)	'iex'
queue_capacity	int	Maximum pending events before a fail-closed overflow.	1024
experimental	bool	Must be true to acknowledge the unsupported feed contract.	False

Source code in src/ml4t/live/feeds/alpaca_feed.py

def __init__(
    self,
    api_key: str,
    secret_key: str,
    symbols: list[str],
    *,
    data_type: str = "bars",  # 'bars', 'quotes', 'trades'
    feed: str = "iex",  # 'iex' (free) or 'sip' (premium)
    queue_capacity: int = 1_024,
    experimental: bool = False,
) -> None:
    """Initialize Alpaca data feed.

    Args:
        api_key: Alpaca API key
        secret_key: Alpaca secret key
        symbols: List of symbols (e.g., ['AAPL', 'BTC/USD'])
        data_type: Type of data - 'bars' (default), 'quotes', or 'trades'
        feed: Data feed type - 'iex' (free) or 'sip' (premium)
        queue_capacity: Maximum pending events before a fail-closed overflow.
        experimental: Must be true to acknowledge the unsupported feed contract.
    """
    require_experimental_opt_in(
        "AlpacaDataFeed",
        experimental=experimental,
        missing_guarantees=ALPACA_MISSING_GUARANTEES,
    )
    if not symbols or any(
        not isinstance(symbol, str) or not symbol.strip() for symbol in symbols
    ):
        raise ValueError("symbols must contain at least one non-empty symbol")
    if data_type not in {"bars", "quotes", "trades"}:
        raise ValueError("data_type must be bars, quotes, or trades")
    if feed.lower() not in {"iex", "sip"}:
        raise ValueError("feed must be iex or sip")
    self._api_key = api_key
    self._secret_key = secret_key
    self._data_type = data_type
    self._feed = feed

    # Separate stock and crypto symbols
    self._stock_symbols = [s for s in symbols if not self._is_crypto(s)]
    self._crypto_symbols = [s for s in symbols if self._is_crypto(s)]

    # Streams (created in start())
    self._stock_stream: StockDataStream | None = None
    self._crypto_stream: CryptoDataStream | None = None
    self._stream_tasks: list[asyncio.Task] = []
    self._failure: Exception | None = None
    self._consumer_loop: asyncio.AbstractEventLoop | None = None

    # State
    self.queue_capacity = queue_capacity
    self._queue = BoundedEventQueue(capacity=queue_capacity, feed="alpaca")
    self._running = False
    self.max_event_age_seconds = 120.0 if data_type == "bars" else 30.0

    # Statistics
    self._bar_count = 0
    self._quote_count = 0
    self._trade_count = 0

stats property

stats

Get feed statistics.

Returns:

Type	Description
dict[str, Any]	Dict with keys:
dict[str, Any]	running: bool
dict[str, Any]	bar_count: int
dict[str, Any]	quote_count: int
dict[str, Any]	trade_count: int
dict[str, Any]	stock_symbols: list[str]
dict[str, Any]	crypto_symbols: list[str]

start async

start()

Subscribe to market data for all symbols.

Creates streams and subscribes to real-time data.

Source code in src/ml4t/live/feeds/alpaca_feed.py

async def start(self) -> None:
    """Subscribe to market data for all symbols.

    Creates streams and subscribes to real-time data.
    """
    logger.info(
        f"AlpacaDataFeed: Starting feed for "
        f"{len(self._stock_symbols)} stocks, {len(self._crypto_symbols)} crypto"
    )
    if self._running:
        return
    self._queue = BoundedEventQueue(capacity=self.queue_capacity, feed="alpaca")
    self._stream_tasks.clear()
    self._failure = None
    self._consumer_loop = asyncio.get_running_loop()
    self._running = True

    # Create stock stream if we have stock symbols
    if self._stock_symbols:
        # Convert string feed to DataFeed enum
        feed_enum = DataFeed.IEX if self._feed.lower() == "iex" else DataFeed.SIP
        self._stock_stream = StockDataStream(
            api_key=self._api_key,
            secret_key=self._secret_key,
            feed=feed_enum,
        )

        # Subscribe based on data type
        if self._data_type == "bars":
            self._stock_stream.subscribe_bars(self._on_stock_bar, *self._stock_symbols)
        elif self._data_type == "quotes":
            self._stock_stream.subscribe_quotes(self._on_stock_quote, *self._stock_symbols)
        elif self._data_type == "trades":
            self._stock_stream.subscribe_trades(self._on_stock_trade, *self._stock_symbols)

        # Start stream in background
        task = asyncio.create_task(self._run_stock_stream())
        self._stream_tasks.append(task)

    # Create crypto stream if we have crypto symbols
    if self._crypto_symbols:
        self._crypto_stream = CryptoDataStream(
            api_key=self._api_key,
            secret_key=self._secret_key,
        )

        # Subscribe based on data type
        if self._data_type == "bars":
            self._crypto_stream.subscribe_bars(self._on_crypto_bar, *self._crypto_symbols)
        elif self._data_type == "quotes":
            self._crypto_stream.subscribe_quotes(self._on_crypto_quote, *self._crypto_symbols)
        elif self._data_type == "trades":
            self._crypto_stream.subscribe_trades(self._on_crypto_trade, *self._crypto_symbols)

        # Start stream in background
        task = asyncio.create_task(self._run_crypto_stream())
        self._stream_tasks.append(task)

    logger.info("AlpacaDataFeed: Subscriptions started")

stop

stop()

Stop data feed.

Closes all streams and signals consumer to exit.

Source code in src/ml4t/live/feeds/alpaca_feed.py

def stop(self) -> None:
    """Stop data feed.

    Closes all streams and signals consumer to exit.
    """
    logger.info("AlpacaDataFeed: Stopping feed")
    self._running = False

    # Cancel stream tasks
    for task in self._stream_tasks:
        if not task.done():
            task.cancel()

    # Stop streams
    if self._stock_stream:
        try:
            self._stock_stream.stop()
        except Exception as e:
            logger.warning(
                "AlpacaDataFeed: Error stopping stock stream: %s",
                redact_sensitive(str(e)),
            )

    if self._crypto_stream:
        try:
            self._crypto_stream.stop()
        except Exception as e:
            logger.warning(
                "AlpacaDataFeed: Error stopping crypto stream: %s",
                redact_sensitive(str(e)),
            )

    # Signal consumer to exit
    self._queue.finish(discard=True)

    logger.info(
        f"AlpacaDataFeed: Stopped. "
        f"Bars: {self._bar_count}, Quotes: {self._quote_count}, Trades: {self._trade_count}"
    )

__aiter__ async

__aiter__()

Async iterator yielding market data.

Yields:

Type	Description
AsyncIterator[MarketEvent]	Validated bar, quote, or trade events.

Stops when

• stop() is called (None sentinel)
• Feed is not running

Source code in src/ml4t/live/feeds/alpaca_feed.py

async def __aiter__(self) -> AsyncIterator[MarketEvent]:
    """Async iterator yielding market data.

    Yields:
        Validated bar, quote, or trade events.

    Stops when:
        - stop() is called (None sentinel)
        - Feed is not running
    """
    while True:
        item = await self._queue.get()

        # None sentinel signals shutdown
        if item is None:
            if self._failure is not None:
                raise RuntimeError("Alpaca stream failed") from self._failure
            break

        yield item

__anext__ async

__anext__()

Get next data item.

Returns:

Type	Description
MarketEvent	A validated bar, quote, or trade event.

Raises:

Type	Description
StopAsyncIteration	When feed is stopped

Source code in src/ml4t/live/feeds/alpaca_feed.py

async def __anext__(self) -> MarketEvent:
    """Get next data item.

    Returns:
        A validated bar, quote, or trade event.

    Raises:
        StopAsyncIteration: When feed is stopped
    """
    queue_state = self._queue.snapshot()
    if not self._running and self._queue.empty() and not queue_state.failed:
        raise StopAsyncIteration

    item = await self._queue.get()
    if item is None:
        if self._failure is not None:
            raise RuntimeError("Alpaca stream failed") from self._failure
        raise StopAsyncIteration

    return item

OKXFundingFeed

OKXFundingFeed(
    symbols,
    *,
    timeframe="1H",
    poll_interval_seconds=60.0,
    queue_capacity=256,
)

Bases: DataFeedProtocol

OKX funding rate feed with OHLCV bars.

Combines price data with funding rate information for ML strategies that trade crypto perpetual futures based on funding rate signals.

Data Flow

1. Poll /market/candles for latest OHLCV bar
2. Poll /public/funding-rate for current funding rate
3. Emit each causal record as its own event

Symbol Format

OKX perpetual swaps use format: BTC-USDT-SWAP, ETH-USDT-SWAP

Initialize OKX funding rate feed.

Parameters:

Name	Type	Description	Default
symbols	list[str]	List of perpetual swap symbols (e.g., ['BTC-USDT-SWAP'])	required
timeframe	str	OHLCV bar timeframe ('1m', '1H', '4H', '1D')	'1H'
poll_interval_seconds	float	How often to poll for new data	60.0
queue_capacity	int	Maximum pending events before a fail-closed overflow.	256

Source code in src/ml4t/live/feeds/okx_feed.py

def __init__(
    self,
    symbols: list[str],
    *,
    timeframe: str = "1H",
    poll_interval_seconds: float = 60.0,
    queue_capacity: int = 256,
) -> None:
    """Initialize OKX funding rate feed.

    Args:
        symbols: List of perpetual swap symbols (e.g., ['BTC-USDT-SWAP'])
        timeframe: OHLCV bar timeframe ('1m', '1H', '4H', '1D')
        poll_interval_seconds: How often to poll for new data
        queue_capacity: Maximum pending events before a fail-closed overflow.
    """
    if not symbols or any(
        not isinstance(symbol, str) or not symbol.strip() for symbol in symbols
    ):
        raise ValueError("symbols must contain at least one non-empty symbol")
    if (
        isinstance(poll_interval_seconds, bool)
        or not isinstance(poll_interval_seconds, int | float)
        or not math.isfinite(poll_interval_seconds)
        or poll_interval_seconds <= 0
    ):
        raise ValueError("poll_interval_seconds must be finite and positive")
    self.symbols = list(symbols)
    self.timeframe = timeframe
    self.poll_interval = poll_interval_seconds

    # State
    self.queue_capacity = queue_capacity
    self._queue = BoundedEventQueue(capacity=queue_capacity, feed="okx")
    self._running = False
    self._poll_task: asyncio.Task | None = None
    self._client: httpx.AsyncClient | None = None
    self._failure: Exception | None = None

    self._emitted_bars: set[tuple[str, datetime, EventCompletion]] = set()
    self._emitted_evolving: dict[tuple[str, datetime], BarPayload] = {}
    self._last_complete_bar_time: dict[str, datetime] = {}
    self._emitted_funding: set[tuple[str, str | int | None, float, str | None]] = set()
    self.max_event_age_seconds = self._timeframe_seconds(timeframe) * 2 + poll_interval_seconds

    # Statistics
    self._bar_count = 0
    self._funding_updates = 0
    self._rejected_count = 0
    self._error_count = 0

stats property

stats

Get feed statistics.

start async

start()

Start the OKX data feed.

Begins polling for OHLCV and funding rate data.

Source code in src/ml4t/live/feeds/okx_feed.py

async def start(self) -> None:
    """Start the OKX data feed.

    Begins polling for OHLCV and funding rate data.
    """
    logger.info(f"OKXFundingFeed: Starting feed for {len(self.symbols)} symbols")
    if self._running:
        return
    await self.close()
    self._queue = BoundedEventQueue(capacity=self.queue_capacity, feed="okx")
    self._failure = None
    self._running = True

    # Create async HTTP client
    self._client = httpx.AsyncClient(timeout=30.0)

    # Start polling task
    self._poll_task = asyncio.create_task(self._poll_loop())

    logger.info(f"OKXFundingFeed: Started polling every {self.poll_interval}s")

stop

stop()

Stop the data feed.

Source code in src/ml4t/live/feeds/okx_feed.py

def stop(self) -> None:
    """Stop the data feed."""
    logger.info("OKXFundingFeed: Stopping feed")
    self._running = False

    if self._poll_task:
        self._poll_task.cancel()

    # Signal consumer
    self._queue.finish(discard=True)

    logger.info(
        f"OKXFundingFeed: Stopped. Bars: {self._bar_count}, "
        f"Funding updates: {self._funding_updates}"
    )

close async

close()

Close HTTP client.

Source code in src/ml4t/live/feeds/okx_feed.py

async def close(self) -> None:
    """Close HTTP client."""
    if self._poll_task is not None:
        self._running = False
        if not self._poll_task.done():
            self._poll_task.cancel()
        if self._poll_task is not asyncio.current_task():
            await asyncio.gather(self._poll_task, return_exceptions=True)
        self._poll_task = None
    if self._client:
        await self._client.aclose()
        self._client = None

__aiter__

__aiter__()

Return async iterator.

Source code in src/ml4t/live/feeds/okx_feed.py

def __aiter__(self):
    """Return async iterator."""
    return self

__anext__ async

__anext__()

Get next bar with funding data.

Returns:

Type	Description
MarketEvent	A validated bar or funding event.

Raises:

Type	Description
StopAsyncIteration	When feed stops

Source code in src/ml4t/live/feeds/okx_feed.py

async def __anext__(self) -> MarketEvent:
    """Get next bar with funding data.

    Returns:
        A validated bar or funding event.

    Raises:
        StopAsyncIteration: When feed stops
    """
    item = await self._queue.get()

    if item is None:  # Shutdown sentinel
        if self._failure is not None:
            raise RuntimeError("OKX polling failed") from self._failure
        raise StopAsyncIteration

    return item

BarAggregator

BarAggregator(
    source_feed,
    bar_size_minutes=1,
    assets=None,
    flush_timeout_seconds=2.0,
    queue_capacity=256,
)

Aggregates raw ticks or 5-second bars into minute bars.

Addresses aggregation and finalization requirements: 1. "If IBDataFeed pushes a tick to Strategy.on_data, the strategy might trigger 60x more often than intended." - Buffer incoming data. 2. "The 15:59 bar is never emitted because no 16:00 tick arrives." - Background flush checker emits bars on timeout.

The aggregator buffers incoming data and emits when: - A bar boundary is crossed (new tick arrives in next minute) - OR timeout expires (2s past bar end with no new data)

Example

raw_feed = IBTickFeed(ib, assets=['AAPL']) aggregated_feed = BarAggregator(raw_feed, bar_size_minutes=1)

async for event in aggregated_feed: consume(event)

Initialize BarAggregator.

Parameters:

Name	Type	Description	Default
source_feed	DataFeedProtocol	Raw tick or sub-minute bar feed	required
bar_size_minutes	int	Output bar size in minutes (default: 1)	1
assets	list[str] | None	List of assets to track (default: all from source)	None
flush_timeout_seconds	float	Seconds after bar end before forcing emit (default: 2.0)	2.0
queue_capacity	int	Maximum pending bars before a fail-closed overflow.	256

Source code in src/ml4t/live/feeds/aggregator.py

def __init__(
    self,
    source_feed: "DataFeedProtocol",
    bar_size_minutes: int = 1,
    assets: list[str] | None = None,
    flush_timeout_seconds: float = 2.0,
    queue_capacity: int = 256,
) -> None:
    """Initialize BarAggregator.

    Args:
        source_feed: Raw tick or sub-minute bar feed
        bar_size_minutes: Output bar size in minutes (default: 1)
        assets: List of assets to track (default: all from source)
        flush_timeout_seconds: Seconds after bar end before forcing emit (default: 2.0)
        queue_capacity: Maximum pending bars before a fail-closed overflow.
    """
    if isinstance(bar_size_minutes, bool) or not isinstance(bar_size_minutes, int):
        raise TypeError("bar_size_minutes must be an integer")
    if bar_size_minutes <= 0:
        raise ValueError("bar_size_minutes must be positive")
    if (
        isinstance(flush_timeout_seconds, bool)
        or not isinstance(flush_timeout_seconds, int | float)
        or not math.isfinite(flush_timeout_seconds)
        or flush_timeout_seconds < 0
    ):
        raise ValueError("flush_timeout_seconds must be finite and non-negative")
    self.source = source_feed
    self.bar_size = timedelta(minutes=bar_size_minutes)
    self.assets = assets or []
    self.flush_timeout = flush_timeout_seconds
    self.queue_capacity = queue_capacity

    # Per-asset bar buffers
    self._buffers: dict[str, BarBuffer] = {}
    self._current_bar_start: dict[str, datetime] = {}
    self._source_events: dict[str, MarketEvent] = {}
    self._last_completed_bar: dict[str, datetime] = {}
    self._last_data_time: float = 0  # Track when we last got data

    # Output queue (use None sentinel for shutdown instead of timeout)
    self._queue = BoundedEventQueue(capacity=queue_capacity, feed="bar_aggregator")
    self._running = False
    self._aggregate_task: asyncio.Task | None = None
    self._flush_task: asyncio.Task | None = None
    self._failure: Exception | None = None

stats property

stats

Return aggregation and bounded-queue state.

start async

start()

Start aggregation.

Source code in src/ml4t/live/feeds/aggregator.py

async def start(self) -> None:
    """Start aggregation."""
    if self._running:
        return
    self._failure = None
    self._queue = BoundedEventQueue(capacity=self.queue_capacity, feed="bar_aggregator")
    self._running = True
    await self.source.start()

    # Start aggregation task
    self._aggregate_task = asyncio.create_task(self._aggregate_loop())

stop

stop()

Stop aggregation.

Source code in src/ml4t/live/feeds/aggregator.py

def stop(self) -> None:
    """Stop aggregation."""
    self._running = False
    self.source.stop()
    if self._flush_task:
        self._flush_task.cancel()
    if self._aggregate_task is None:
        self._queue.finish(discard=True)

__aiter__

__aiter__()

Return this feed's asynchronous iterator.

Source code in src/ml4t/live/feeds/aggregator.py

def __aiter__(self) -> AsyncIterator[MarketEvent]:
    """Return this feed's asynchronous iterator."""
    return self

__anext__ async

__anext__()

Return the next validated bar or finish after the queue drains.

Source code in src/ml4t/live/feeds/aggregator.py

async def __anext__(self) -> MarketEvent:
    """Return the next validated bar or finish after the queue drains."""
    item = await self._queue.get()
    if item is None:
        if self._failure is not None:
            raise self._failure
        raise StopAsyncIteration
    return item

BarBuffer dataclass

BarBuffer(
    open=None,
    high=float("-inf"),
    low=float("inf"),
    close=0.0,
    volume=0.0,
)

Accumulates ticks into OHLCV bar.

Attributes:

Name	Type	Description
open	float | None	Opening price (first tick)
high	float	Highest price seen
low	float	Lowest price seen
close	float	Most recent price
volume	float	Total volume accumulated

update

update(price, size=0)

Add a tick to the bar.

Parameters:

Name	Type	Description	Default
price	float	Trade price	required
size	float	Trade size (defaults to 0 for quote ticks)	0

Source code in src/ml4t/live/feeds/aggregator.py

def update(self, price: float, size: float = 0) -> None:
    """Add a tick to the bar.

    Args:
        price: Trade price
        size: Trade size (defaults to 0 for quote ticks)
    """
    if self.open is None:
        self.open = price
    self.high = max(self.high, price)
    self.low = min(self.low, price)
    self.close = price
    self.volume += size

update_bar

update_bar(payload)

Merge one validated OHLCV payload without discarding its range.

Source code in src/ml4t/live/feeds/aggregator.py

def update_bar(self, payload: dict[str, Any]) -> None:
    """Merge one validated OHLCV payload without discarding its range."""
    open_price = float(payload["open"])
    high = float(payload["high"])
    low = float(payload["low"])
    close = float(payload["close"])
    if self.open is None:
        self.open = open_price
    self.high = max(self.high, high)
    self.low = min(self.low, low)
    self.close = close
    self.volume += float(payload.get("volume", 0.0))

to_dict

to_dict()

Convert to OHLCV dict.

Returns:

Type	Description
dict[str, Any]	Dictionary with keys: open, high, low, close, volume
dict[str, Any]	If no ticks received, uses close price as fallback for OHLC

Source code in src/ml4t/live/feeds/aggregator.py

def to_dict(self) -> dict[str, Any]:
    """Convert to OHLCV dict.

    Returns:
        Dictionary with keys: open, high, low, close, volume
        If no ticks received, uses close price as fallback for OHLC
    """
    return {
        "open": self.open or self.close,
        "high": self.high if self.high != float("-inf") else self.close,
        "low": self.low if self.low != float("inf") else self.close,
        "close": self.close,
        "volume": self.volume,
    }

reset

reset()

Reset for next bar.

Source code in src/ml4t/live/feeds/aggregator.py

def reset(self) -> None:
    """Reset for next bar."""
    self.open = None
    self.high = float("-inf")
    self.low = float("inf")
    self.close = 0.0
    self.volume = 0.0

FeedContractError

Bases: ValueError

Raised when provider data cannot cross the portable feed boundary.

FeedContinuityError

FeedContinuityError(reason, event)

Bases: FeedContractError

Raised when event history cannot support another causal decision.

Source code in src/ml4t/live/feeds/events.py

def __init__(self, reason: str, event: MarketEvent) -> None:
    self.reason = reason
    self.source = event.source
    self.asset = event.asset
    self.kind = event.kind.value
    self.provider_sequence = event.provider_sequence
    super().__init__(f"{event.source} {event.asset} {event.kind.value}: {reason}")

FeedOverflowError

FeedOverflowError(*, feed, event, snapshot)

Bases: RuntimeError

Raised when continuing after a full queue would hide lost market data.

Source code in src/ml4t/live/feeds/queue.py

def __init__(
    self,
    *,
    feed: str,
    event: MarketEvent,
    snapshot: FeedQueueSnapshot,
) -> None:
    self.feed = feed
    self.asset = event.asset
    self.kind = event.kind.value
    self.snapshot = snapshot
    self.gap = GapEvidence(
        True,
        f"{feed} queue overflow rejected {event.kind.value} for {event.asset}",
        previous_sequence=f"retained:{feed}",
        current_sequence=f"rejected:{event.source}:{event.asset}",
    )
    super().__init__(
        f"{feed} queue capacity {snapshot.capacity} exceeded by "
        f"{event.kind.value} event for {event.asset}"
    )

FeedQueueSnapshot dataclass

FeedQueueSnapshot(
    capacity,
    occupancy,
    high_watermark,
    overflow_count,
    oldest_event_lag_seconds,
    failed,
    finished,
)

Observable state for one supported feed queue.

Experimental Feed Opt-In

These imports are available for deliberate evaluation. They are not part of the stable support contract and require experimental=True.

from ml4t.live import (
    AlpacaDataFeed,
    CryptoFeed,
    DataBentoFeed,
    ExperimentalFeedError,
    ExperimentalFeedWarning,
    IBDataFeed,
)

ExperimentalFeedError

Bases: RuntimeError

Raised when an experimental feed is constructed without deliberate opt-in.

ExperimentalFeedWarning

Bases: UserWarning

Reports guarantees that do not apply to an experimental feed.

DataBentoFeed

DataBentoFeed(
    client,
    symbols,
    *,
    mode="historical",
    replay_speed=1.0,
    experimental=False,
)

Bases: DataFeedProtocol

Experimental market data feed from DataBento.

Supports selected historical replay and live record shapes after explicit opt-in.

Historical Mode

• Reads from .dbn files (DataBento native format)
• Replays at historical speed or accelerated
• Intended for custom evaluation under the experimental limitations

Real-time Mode

• Streams live market data
• Supports multiple datasets (GLBX, XNAS, OPRA, etc.)
• No stable latency or throughput guarantee

Data Format

Experimental typed MarketEvent bars, trades, and quotes with UTC timestamps.

Example Historical

feed = DataBentoFeed.from_file( 'ES_202401.dbn', symbols=['ES.FUT'], replay_speed=10.0, # 10x speed experimental=True, )

Example Real-time

feed = DataBentoFeed.from_live( api_key=os.getenv('DATABENTO_API_KEY'), dataset='GLBX.MDP3', schema='ohlcv-1s', symbols=['ES.c.0', 'NQ.c.0'], experimental=True, )

Initialize DataBento feed.

Parameters:

Name	Type	Description	Default
client	_HistoricalRecordSource | _LiveRecordSource	DataBento client (Historical or Live)	required
symbols	list[str]	List of symbols to subscribe to	required
mode	str	'historical' or 'live'	'historical'
replay_speed	float	Playback speed multiplier (historical only) 1.0 = real-time, 10.0 = 10x speed	1.0
experimental	bool	Must be true to acknowledge the unsupported feed contract.	False

Source code in src/ml4t/live/feeds/databento_feed.py

def __init__(
    self,
    client: _HistoricalRecordSource | _LiveRecordSource,
    symbols: list[str],
    *,
    mode: str = "historical",
    replay_speed: float = 1.0,
    experimental: bool = False,
) -> None:
    """Initialize DataBento feed.

    Args:
        client: DataBento client (Historical or Live)
        symbols: List of symbols to subscribe to
        mode: 'historical' or 'live'
        replay_speed: Playback speed multiplier (historical only)
            1.0 = real-time, 10.0 = 10x speed
        experimental: Must be true to acknowledge the unsupported feed contract.
    """
    require_experimental_opt_in(
        "DataBentoFeed",
        experimental=experimental,
        missing_guarantees=DATABENTO_MISSING_GUARANTEES,
    )
    if not DATABENTO_AVAILABLE:
        raise ImportError("DataBentoFeed requires the ml4t-live[experimental] package extra")
    _validate_symbols(symbols)
    if mode not in {"historical", "live"}:
        raise ValueError("mode must be 'historical' or 'live'")
    _validate_replay_speed(replay_speed)

    self.client = client
    self.symbols = list(symbols)
    self.mode = mode
    self.replay_speed = replay_speed

    # State
    self._queue: asyncio.Queue[MarketEvent | None] = asyncio.Queue()
    self._running = False
    self._replay_task: asyncio.Task | None = None
    self._failure: RuntimeError | None = None

    # Statistics
    self._record_count = 0

stats property

stats

Get feed statistics.

from_file classmethod

from_file(
    file_path,
    symbols,
    *,
    replay_speed=1.0,
    experimental=False,
)

Create feed from historical .dbn file.

Parameters:

Name	Type	Description	Default
file_path	str | Path	Path to .dbn file	required
symbols	list[str]	Non-empty symbol selection	required
replay_speed	float	Playback speed (1.0 = real-time)	1.0
experimental	bool	Must be true to acknowledge the unsupported feed contract.	False

Returns:

Type	Description
DataBentoFeed	DataBentoFeed configured for historical replay

Source code in src/ml4t/live/feeds/databento_feed.py

@classmethod
def from_file(
    cls,
    file_path: str | Path,
    symbols: list[str],
    *,
    replay_speed: float = 1.0,
    experimental: bool = False,
) -> "DataBentoFeed":
    """Create feed from historical .dbn file.

    Args:
        file_path: Path to .dbn file
        symbols: Non-empty symbol selection
        replay_speed: Playback speed (1.0 = real-time)
        experimental: Must be true to acknowledge the unsupported feed contract.

    Returns:
        DataBentoFeed configured for historical replay
    """
    if experimental is not True:
        require_experimental_opt_in(
            "DataBentoFeed",
            experimental=experimental,
            missing_guarantees=DATABENTO_MISSING_GUARANTEES,
        )
    if not DATABENTO_AVAILABLE:
        raise ImportError("DataBentoFeed requires the ml4t-live[experimental] package extra")
    _validate_symbols(symbols)
    _validate_replay_speed(replay_speed)

    # Read file
    store = db.DBNStore.from_file(file_path)

    return cls(
        client=store,
        symbols=symbols,
        mode="historical",
        replay_speed=replay_speed,
        experimental=experimental,
    )

from_live classmethod

from_live(
    api_key, dataset, schema, symbols, *, experimental=False
)

Create feed for real-time streaming.

Parameters:

Name	Type	Description	Default
api_key	str	DataBento API key	required
dataset	str	Dataset code (e.g., 'GLBX.MDP3', 'XNAS.ITCH')	required
schema	str	Data schema (e.g., 'ohlcv-1s', 'mbp-10', 'trades')	required
symbols	list[str]	Symbols to subscribe to	required
experimental	bool	Must be true to acknowledge the unsupported feed contract.	False

Returns:

Type	Description
DataBentoFeed	DataBentoFeed configured for live streaming

Source code in src/ml4t/live/feeds/databento_feed.py

@classmethod
def from_live(
    cls,
    api_key: str,
    dataset: str,
    schema: str,
    symbols: list[str],
    *,
    experimental: bool = False,
) -> "DataBentoFeed":
    """Create feed for real-time streaming.

    Args:
        api_key: DataBento API key
        dataset: Dataset code (e.g., 'GLBX.MDP3', 'XNAS.ITCH')
        schema: Data schema (e.g., 'ohlcv-1s', 'mbp-10', 'trades')
        symbols: Symbols to subscribe to
        experimental: Must be true to acknowledge the unsupported feed contract.

    Returns:
        DataBentoFeed configured for live streaming
    """
    if experimental is not True:
        require_experimental_opt_in(
            "DataBentoFeed",
            experimental=experimental,
            missing_guarantees=DATABENTO_MISSING_GUARANTEES,
        )
    if not DATABENTO_AVAILABLE:
        raise ImportError("DataBentoFeed requires the ml4t-live[experimental] package extra")
    _validate_symbols(symbols)

    client = db.Live(key=api_key)

    # Configure subscription
    client.subscribe(
        dataset=dataset,
        schema=schema,
        symbols=symbols,
    )

    return cls(
        client=client,
        symbols=symbols,
        mode="live",
        experimental=experimental,
    )

start async

start()

Start data feed.

Historical mode: Begins replay task Live mode: Starts streaming subscription

Source code in src/ml4t/live/feeds/databento_feed.py

async def start(self) -> None:
    """Start data feed.

    Historical mode: Begins replay task
    Live mode: Starts streaming subscription
    """
    if self._running:
        return
    logger.info(f"DataBentoFeed: Starting {self.mode} feed for {len(self.symbols)} symbols")
    self._queue = asyncio.Queue()
    self._failure = None
    self._running = True

    if self.mode == "historical":
        # Start replay task
        self._replay_task = asyncio.create_task(self._replay_historical())
    elif self.mode == "live":
        # Start live streaming task
        self._replay_task = asyncio.create_task(self._stream_live())

    logger.info("DataBentoFeed: Feed started")

stop

stop()

Stop data feed.

Source code in src/ml4t/live/feeds/databento_feed.py

def stop(self) -> None:
    """Stop data feed."""
    logger.info("DataBentoFeed: Stopping feed")
    self._running = False

    # Cancel replay task
    if self._replay_task:
        self._replay_task.cancel()

    self._signal_stop()

    logger.info(f"DataBentoFeed: Stopped. Records: {self._record_count}")

__aiter__ async

__aiter__()

Async iterator yielding market data.

Yields:

Type	Description
AsyncIterator[MarketEvent]	Typed experimental market event.

Source code in src/ml4t/live/feeds/databento_feed.py

async def __aiter__(self) -> AsyncIterator[MarketEvent]:
    """Async iterator yielding market data.

    Yields:
        Typed experimental market event.
    """
    while True:
        item = await self._queue.get()

        if item is None:  # Shutdown sentinel
            if self._failure is not None:
                raise self._failure
            break

        yield item

CryptoFeed

CryptoFeed(
    exchange,
    symbols,
    *,
    timeframe="1m",
    stream_trades=False,
    stream_ohlcv=True,
    api_key=None,
    api_secret=None,
    api_passphrase=None,
    experimental=False,
)

Bases: DataFeedProtocol

Experimental cryptocurrency market data feed via asynchronous CCXT.

Supports async REST polling and uses CCXT Pro websocket methods when available. No exchange, overload, reconnect, or performance guarantee is included in the stable support contract.

Data Format

Experimental typed MarketEvent bars and trades with UTC timestamps.

Exchange Symbols

• Binance: 'BTC/USDT', 'ETH/USDT'
• Coinbase: 'BTC-USD', 'ETH-USD'
• Kraken: 'BTC/USD', 'ETH/USD'

Timeframes

'1m', '5m', '15m', '1h', '4h', '1d'

Example WebSocket (Real-time): feed = CryptoFeed( exchange='binance', symbols=['BTC/USDT', 'ETH/USDT'], stream_trades=True, # Stream trades (fastest) experimental=True, )

Example OHLCV Bars

feed = CryptoFeed( exchange='binance', symbols=['BTC/USDT'], timeframe='1m', stream_ohlcv=True, experimental=True, )

Example Authenticated

feed = CryptoFeed( exchange='binance', symbols=['BTC/USDT'], api_key='your-key', api_secret='your-secret', experimental=True, )

Initialize crypto feed.

Parameters:

Name	Type	Description	Default
exchange	str	Exchange ID (e.g., 'binance', 'coinbasepro', 'kraken')	required
symbols	list[str]	Trading pairs (e.g., ['BTC/USDT', 'ETH/USDT'])	required
timeframe	str	OHLCV timeframe ('1m', '5m', '1h', etc.)	'1m'
stream_trades	bool	Stream trade ticks (faster updates)	False
stream_ohlcv	bool	Stream OHLCV candles	True
api_key	str | None	API key (for authenticated endpoints)	None
api_secret	str | None	API secret	None
api_passphrase	str | None	API passphrase (Coinbase only)	None
experimental	bool	Must be true to acknowledge the unsupported feed contract.	False

Source code in src/ml4t/live/feeds/crypto_feed.py

def __init__(
    self,
    exchange: str,
    symbols: list[str],
    *,
    timeframe: str = "1m",
    stream_trades: bool = False,
    stream_ohlcv: bool = True,
    api_key: str | None = None,
    api_secret: str | None = None,
    api_passphrase: str | None = None,
    experimental: bool = False,
) -> None:
    """Initialize crypto feed.

    Args:
        exchange: Exchange ID (e.g., 'binance', 'coinbasepro', 'kraken')
        symbols: Trading pairs (e.g., ['BTC/USDT', 'ETH/USDT'])
        timeframe: OHLCV timeframe ('1m', '5m', '1h', etc.)
        stream_trades: Stream trade ticks (faster updates)
        stream_ohlcv: Stream OHLCV candles
        api_key: API key (for authenticated endpoints)
        api_secret: API secret
        api_passphrase: API passphrase (Coinbase only)
        experimental: Must be true to acknowledge the unsupported feed contract.
    """
    require_experimental_opt_in(
        "CryptoFeed",
        experimental=experimental,
        missing_guarantees=CRYPTO_MISSING_GUARANTEES,
    )
    if not CCXT_AVAILABLE:
        raise ImportError(
            "ccxt package required. Install ml4t-live with its locked dependencies"
        )
    if not isinstance(exchange, str) or not exchange.strip():
        raise ValueError("exchange must be a non-empty string")
    if not symbols or any(
        not isinstance(symbol, str) or not symbol.strip() for symbol in symbols
    ):
        raise ValueError("symbols must contain at least one non-empty symbol")

    self.exchange_id = exchange
    self.symbols = list(symbols)
    self.timeframe = timeframe
    self.stream_trades = stream_trades
    self.stream_ohlcv = stream_ohlcv

    # Create exchange instance
    exchange_class = getattr(ccxt, exchange, None)
    if not callable(exchange_class):
        raise ValueError(f"CCXT exchange is unavailable: {exchange}")
    config = {
        "enableRateLimit": True,
    }

    if api_key:
        config["apiKey"] = api_key
    if api_secret:
        config["secret"] = api_secret
    if api_passphrase:
        config["password"] = api_passphrase

    self.exchange = exchange_class(config)

    # State
    self._queue: asyncio.Queue[MarketEvent | None] = asyncio.Queue()
    self._running = False
    self._stream_tasks: list[asyncio.Task] = []
    self._failure: RuntimeError | None = None
    self._completed_candles: set[tuple[str, int]] = set()
    self._evolving_candles: dict[tuple[str, int], BarPayload] = {}

    # Statistics
    self._tick_count = 0
    self._trade_count = 0
    self._candle_count = 0

stats property

stats

Get feed statistics.

start async

start()

Start streaming market data.

Initiates WebSocket subscriptions for all symbols.

Source code in src/ml4t/live/feeds/crypto_feed.py

async def start(self) -> None:
    """Start streaming market data.

    Initiates WebSocket subscriptions for all symbols.
    """
    if self._running:
        return
    logger.info(f"CryptoFeed: Starting {self.exchange_id} feed for {len(self.symbols)} symbols")
    self._queue = asyncio.Queue()
    self._stream_tasks.clear()
    self._failure = None
    self._running = True

    # Load markets
    try:
        await self.exchange.load_markets()
    except BaseException:
        self._running = False
        raise

    # Start streaming tasks
    for symbol in self.symbols:
        if self.stream_trades:
            task = asyncio.create_task(self._stream_trades_for_symbol(symbol))
            self._stream_tasks.append(task)

        if self.stream_ohlcv:
            task = asyncio.create_task(self._stream_ohlcv_for_symbol(symbol))
            self._stream_tasks.append(task)

    logger.info(f"CryptoFeed: Started {len(self._stream_tasks)} stream(s)")

stop

stop()

Stop streaming; use close to release the exchange connection.

Source code in src/ml4t/live/feeds/crypto_feed.py

def stop(self) -> None:
    """Stop streaming; use ``close`` to release the exchange connection."""
    logger.info("CryptoFeed: Stopping feed")
    self._running = False

    # Cancel all streaming tasks
    for task in self._stream_tasks:
        task.cancel()

    # Signal consumer
    self._signal_stop()

    logger.info(
        f"CryptoFeed: Stopped. "
        f"Ticks: {self._tick_count}, Trades: {self._trade_count}, "
        f"Candles: {self._candle_count}"
    )

__aiter__ async

__aiter__()

Async iterator yielding market data.

Yields:

Type	Description
AsyncIterator[MarketEvent]	Typed experimental market event.

Source code in src/ml4t/live/feeds/crypto_feed.py

async def __aiter__(self) -> AsyncIterator[MarketEvent]:
    """Async iterator yielding market data.

    Yields:
        Typed experimental market event.
    """
    while True:
        item = await self._queue.get()

        if item is None:  # Shutdown sentinel
            if self._failure is not None:
                raise self._failure
            break

        yield item

close async

close()

Close exchange connection.

Should be called in finally block.

Source code in src/ml4t/live/feeds/crypto_feed.py

async def close(self) -> None:
    """Close exchange connection.

    Should be called in finally block.
    """
    await self.exchange.close()

Protocols

BrokerProtocol

Bases: Protocol

Synchronous broker protocol for Strategy.on_data().

This is the interface strategies interact with. It must be synchronous because Strategy.on_data() is synchronous (matches backtest behavior).

ThreadSafeBrokerWrapper implements this protocol by wrapping an AsyncBrokerProtocol and using run_coroutine_threadsafe().

Example

class MyStrategy(Strategy): def on_data(self, timestamp, data, context, broker: BrokerProtocol): # broker is sync - no async/await needed pos = broker.get_position("AAPL") if pos is None: broker.submit_order("AAPL", 100)

positions property

positions

Get all current positions.

Returns:

Type	Description
dict[str, Position]	Dictionary mapping asset symbol to Position

pending_orders property

pending_orders

Get all pending orders.

Returns:

Type	Description
list[Order]	List of pending Order objects

is_connected property

is_connected

Check if broker is connected.

Returns:

Type	Description
bool	True if connected and ready to trade

get_position

get_position(asset)

Get position for specific asset.

Parameters:

Name	Type	Description	Default
asset	str	Asset symbol (e.g., "AAPL")	required

Returns:

Type	Description
Position | None	Position object if holding position, None otherwise

Source code in src/ml4t/live/protocols.py

def get_position(self, asset: str) -> Position | None:
    """Get position for specific asset.

    Args:
        asset: Asset symbol (e.g., "AAPL")

    Returns:
        Position object if holding position, None otherwise
    """
    ...

get_positions

get_positions()

Get all current positions using the portable backtest-facing method.

Source code in src/ml4t/live/protocols.py

def get_positions(self) -> dict[str, Position]:
    """Get all current positions using the portable backtest-facing method."""
    ...

get_account_value

get_account_value()

Get total account value (cash + positions).

Returns:

Type	Description
float	Total account value in base currency

Source code in src/ml4t/live/protocols.py

def get_account_value(self) -> float:
    """Get total account value (cash + positions).

    Returns:
        Total account value in base currency
    """
    ...

get_cash

get_cash()

Get available cash balance.

Returns:

Type	Description
float	Available cash in base currency

Source code in src/ml4t/live/protocols.py

def get_cash(self) -> float:
    """Get available cash balance.

    Returns:
        Available cash in base currency
    """
    ...

get_pending_orders

get_pending_orders(asset=None)

Get pending orders, optionally filtered by asset.

Source code in src/ml4t/live/protocols.py

def get_pending_orders(self, asset: str | None = None) -> list[Order]:
    """Get pending orders, optionally filtered by asset."""
    ...

submit_order

submit_order(
    asset,
    quantity,
    side=None,
    order_type=MARKET,
    limit_price=None,
    stop_price=None,
    **kwargs,
)

Submit order for execution.

Parameters:

Name	Type	Description	Default
asset	str	Asset symbol (e.g., "AAPL")	required
quantity	float	Signed shares/contracts when side is omitted; positive unsigned shares/contracts when side is provided	required
side	OrderSide | None	Order side (BUY/SELL), inferred from signed quantity if omitted	None
order_type	OrderType	Type of order (MARKET, LIMIT, STOP, etc.)	MARKET
limit_price	float | None	Limit price for LIMIT/STOP_LIMIT orders	None
stop_price	float | None	Stop price for STOP/STOP_LIMIT orders	None
**kwargs	Any	Additional broker-specific parameters	{}

Returns:

Type	Description
Order	Order object with order_id and initial status

Raises:

Type	Description
ValueError	If order parameters are invalid
RuntimeError	If broker is not connected

Source code in src/ml4t/live/protocols.py

def submit_order(
    self,
    asset: str,
    quantity: float,
    side: OrderSide | None = None,
    order_type: OrderType = OrderType.MARKET,
    limit_price: float | None = None,
    stop_price: float | None = None,
    **kwargs: Any,
) -> Order:
    """Submit order for execution.

    Args:
        asset: Asset symbol (e.g., "AAPL")
        quantity: Signed shares/contracts when side is omitted; positive unsigned
            shares/contracts when side is provided
        side: Order side (BUY/SELL), inferred from signed quantity if omitted
        order_type: Type of order (MARKET, LIMIT, STOP, etc.)
        limit_price: Limit price for LIMIT/STOP_LIMIT orders
        stop_price: Stop price for STOP/STOP_LIMIT orders
        **kwargs: Additional broker-specific parameters

    Returns:
        Order object with order_id and initial status

    Raises:
        ValueError: If order parameters are invalid
        RuntimeError: If broker is not connected
    """
    ...

cancel_order

cancel_order(order_id)

Cancel pending order.

Parameters:

Name	Type	Description	Default
order_id	str	ID of order to cancel	required

Returns:

Type	Description
bool	True if cancel request submitted, False if order not found

Source code in src/ml4t/live/protocols.py

def cancel_order(self, order_id: str) -> bool:
    """Cancel pending order.

    Args:
        order_id: ID of order to cancel

    Returns:
        True if cancel request submitted, False if order not found
    """
    ...

replace_order

replace_order(
    order_id,
    *,
    quantity=None,
    limit_price=None,
    stop_price=None,
)

Replace a pending order with updated parameters.

Source code in src/ml4t/live/protocols.py

def replace_order(
    self,
    order_id: str,
    *,
    quantity: float | None = None,
    limit_price: float | None = None,
    stop_price: float | None = None,
) -> Order:
    """Replace a pending order with updated parameters."""
    ...

close_position

close_position(asset)

Close entire position in asset.

Convenience method that submits a closing order.

Parameters:

Name	Type	Description	Default
asset	str	Asset symbol to close	required

Returns:

Type	Description
Order | None	Order object if position exists, None if no position

Source code in src/ml4t/live/protocols.py

def close_position(self, asset: str) -> Order | None:
    """Close entire position in asset.

    Convenience method that submits a closing order.

    Args:
        asset: Asset symbol to close

    Returns:
        Order object if position exists, None if no position
    """
    ...

register_target_intent

register_target_intent(intent, *, position_rules=None)

Register an idempotent target for a causal opening phase.

Source code in src/ml4t/live/protocols.py

def register_target_intent(
    self,
    intent: CanonicalTargetIntent,
    *,
    position_rules: PositionRule | None = None,
) -> CanonicalTargetIntent:
    """Register an idempotent target for a causal opening phase."""
    ...

register_position_rule_policy

register_position_rule_policy(policy_id, rules)

Bind a portable policy identity to its client rule implementation.

Source code in src/ml4t/live/protocols.py

def register_position_rule_policy(self, policy_id: str, rules: PositionRule) -> None:
    """Bind a portable policy identity to its client rule implementation."""
    ...

get_target_intents

get_target_intents()

Return accepted canonical target intents.

Source code in src/ml4t/live/protocols.py

def get_target_intents(self) -> tuple[CanonicalTargetIntent, ...]:
    """Return accepted canonical target intents."""
    ...

get_child_order_intents

get_child_order_intents()

Return canonical child intents derived from accepted targets.

Source code in src/ml4t/live/protocols.py

def get_child_order_intents(self) -> tuple[CanonicalChildOrderIntent, ...]:
    """Return canonical child intents derived from accepted targets."""
    ...

get_intent_reconciliations

get_intent_reconciliations()

Return retained child fill and remainder evidence.

Source code in src/ml4t/live/protocols.py

def get_intent_reconciliations(self) -> tuple[IntentReconciliation, ...]:
    """Return retained child fill and remainder evidence."""
    ...

export_target_intent_state

export_target_intent_state()

Serialize accepted target, child, reconciliation, and rule state.

Source code in src/ml4t/live/protocols.py

def export_target_intent_state(self) -> dict[str, Any]:
    """Serialize accepted target, child, reconciliation, and rule state."""
    ...

set_position_rules

set_position_rules(rules, asset=None)

Set client-evaluated position rules globally or for one asset.

Source code in src/ml4t/live/protocols.py

def set_position_rules(
    self,
    rules: PositionRule | None,
    asset: str | None = None,
) -> None:
    """Set client-evaluated position rules globally or for one asset."""
    ...

clear_position_rules

clear_position_rules(asset=None)

Clear client-evaluated position rules globally or for one asset.

Source code in src/ml4t/live/protocols.py

def clear_position_rules(self, asset: str | None = None) -> None:
    """Clear client-evaluated position rules globally or for one asset."""
    ...

update_position_context

update_position_context(asset, context)

Update persistent context used by position-rule evaluation.

Source code in src/ml4t/live/protocols.py

def update_position_context(self, asset: str, context: dict[str, Any]) -> None:
    """Update persistent context used by position-rule evaluation."""
    ...

AsyncBrokerProtocol

Bases: Protocol

Asynchronous broker protocol for broker implementations.

All broker implementations (IBBroker, AlpacaBroker, etc.) must implement this protocol. The async design enables efficient I/O without blocking.

ThreadSafeBrokerWrapper wraps this protocol to provide BrokerProtocol for strategies.

Example

class IBBroker: async def connect(self): await self.ib.connectAsync(...)

async def submit_order_async(self, asset, quantity, ...):
    # Async I/O to IB
    trade = await self.ib.placeOrderAsync(...)
    return order

positions property

positions

Return an independent snapshot of known positions.

pending_orders property

pending_orders

Return an independent snapshot of known pending orders.

is_connected property

is_connected

Return the cached provider connection state without blocking.

execution_capabilities property

execution_capabilities

Return the order capabilities implemented by the adapter.

connect async

connect()

Connect to broker and sync initial state.

This should: 1. Establish connection to broker API 2. Sync current positions 3. Sync pending orders 4. Register event callbacks

Source code in src/ml4t/live/protocols.py

async def connect(self) -> None:
    """Connect to broker and sync initial state.

    This should:
    1. Establish connection to broker API
    2. Sync current positions
    3. Sync pending orders
    4. Register event callbacks
    """
    ...

disconnect async

disconnect()

Disconnect from broker gracefully.

Source code in src/ml4t/live/protocols.py

async def disconnect(self) -> None:
    """Disconnect from broker gracefully."""
    ...

is_connected_async async

is_connected_async()

Check if broker is connected.

Returns:

Type	Description
bool	True if connected and ready

Source code in src/ml4t/live/protocols.py

async def is_connected_async(self) -> bool:
    """Check if broker is connected.

    Returns:
        True if connected and ready
    """
    ...

assert_paper_trading

assert_paper_trading()

Fail unless the connected endpoint and account are paper-only.

Source code in src/ml4t/live/protocols.py

def assert_paper_trading(self) -> None:
    """Fail unless the connected endpoint and account are paper-only."""
    ...

assert_live_trading

assert_live_trading()

Fail unless the connected endpoint and explicitly selected account are live.

Source code in src/ml4t/live/protocols.py

def assert_live_trading(self) -> None:
    """Fail unless the connected endpoint and explicitly selected account are live."""
    ...

get_positions_async async

get_positions_async()

Get all positions (async version).

Source code in src/ml4t/live/protocols.py

async def get_positions_async(self) -> dict[str, Position]:
    """Get all positions (async version)."""
    ...

get_pending_orders_async async

get_pending_orders_async(asset=None)

Get pending orders, optionally filtered by asset.

Source code in src/ml4t/live/protocols.py

async def get_pending_orders_async(self, asset: str | None = None) -> list[Order]:
    """Get pending orders, optionally filtered by asset."""
    ...

get_position_async async

get_position_async(asset)

Get position for asset (async version).

Source code in src/ml4t/live/protocols.py

async def get_position_async(self, asset: str) -> Position | None:
    """Get position for asset (async version)."""
    ...

get_account_value_async async

get_account_value_async()

Get total account value (async version).

Source code in src/ml4t/live/protocols.py

async def get_account_value_async(self) -> float:
    """Get total account value (async version)."""
    ...

get_cash_async async

get_cash_async()

Get available cash (async version).

Source code in src/ml4t/live/protocols.py

async def get_cash_async(self) -> float:
    """Get available cash (async version)."""
    ...

submit_order_async async

submit_order_async(
    asset,
    quantity,
    side=None,
    order_type=MARKET,
    limit_price=None,
    stop_price=None,
    **kwargs,
)

Submit an order using the same quantity contract as submit_order.

Source code in src/ml4t/live/protocols.py

async def submit_order_async(
    self,
    asset: str,
    quantity: float,
    side: OrderSide | None = None,
    order_type: OrderType = OrderType.MARKET,
    limit_price: float | None = None,
    stop_price: float | None = None,
    **kwargs: Any,
) -> Order:
    """Submit an order using the same quantity contract as ``submit_order``."""
    ...

cancel_order_async async

cancel_order_async(order_id)

Cancel order (async version).

Source code in src/ml4t/live/protocols.py

async def cancel_order_async(self, order_id: str) -> bool:
    """Cancel order (async version)."""
    ...

replace_order_async async

replace_order_async(
    order_id,
    *,
    quantity=None,
    limit_price=None,
    stop_price=None,
)

Replace a pending order with updated parameters (async version).

Source code in src/ml4t/live/protocols.py

async def replace_order_async(
    self,
    order_id: str,
    *,
    quantity: float | None = None,
    limit_price: float | None = None,
    stop_price: float | None = None,
) -> Order:
    """Replace a pending order with updated parameters (async version)."""
    ...

close_position_async async

close_position_async(asset)

Close position (async version).

Source code in src/ml4t/live/protocols.py

async def close_position_async(self, asset: str) -> Order | None:
    """Close position (async version)."""
    ...

DataFeedProtocol

Bases: Protocol

Protocol for real-time data feeds.

Stable-supported feeds provide an async iterator of validated MarketEvent objects. The legacy tuple member of FeedItem is temporary compatibility for experimental feeds. The feed handles: - Subscribing to market data - Aggregating ticks to bars (if needed) - Emitting data on schedule (e.g., every minute)

Example

class IBDataFeed: async def start(self): await self._subscribe_to_ticks()

async def __aiter__(self):
    return self

async def __anext__(self):
    return await self._event_queue.get()

def stop(self):
    self._running = False

start async

start()

Start the data feed.

This should: 1. Subscribe to market data 2. Start internal aggregation/buffering 3. Begin emitting data

Source code in src/ml4t/live/protocols.py

async def start(self) -> None:
    """Start the data feed.

    This should:
    1. Subscribe to market data
    2. Start internal aggregation/buffering
    3. Begin emitting data
    """
    ...

stop

stop()

Stop the data feed gracefully.

Should be non-blocking. The feed should stop after the current iteration completes.

Source code in src/ml4t/live/protocols.py

def stop(self) -> None:
    """Stop the data feed gracefully.

    Should be non-blocking. The feed should stop after
    the current iteration completes.
    """
    ...

__aiter__

__aiter__()

Return async iterator.

Yields:

Type	Description
AsyncIterator[FeedItem]	A validated market event, or a temporary legacy tuple from an experimental feed.

Source code in src/ml4t/live/protocols.py

def __aiter__(
    self,
) -> AsyncIterator[FeedItem]:
    """Return async iterator.

    Yields:
        A validated market event, or a temporary legacy tuple from an experimental feed.
    """
    ...

__anext__ async

__anext__()

Get next market event.

Returns:

Type	Description
FeedItem	A validated market event, or a temporary legacy tuple from an experimental feed.

Raises:

Type	Description
StopAsyncIteration	When feed ends

Source code in src/ml4t/live/protocols.py

async def __anext__(self) -> FeedItem:
    """Get next market event.

    Returns:
        A validated market event, or a temporary legacy tuple from an experimental feed.

    Raises:
        StopAsyncIteration: When feed ends
    """
    ...

Safety Types

RiskLimitError

Bases: Exception

Raised when an order violates risk limits.

ReconciliationMismatchError

Bases: RuntimeError

Raised when startup reconciliation is configured to fail closed.

RiskState dataclass

RiskState(
    date,
    daily_loss=0.0,
    orders_placed=0,
    high_water_mark=0.0,
    session_start_equity=None,
    persisted_positions=dict(),
    persisted_pending_orders=list(),
    portable_strategy_state=dict(),
    replacement_gaps=dict(),
    shadow_portfolio=dict(),
    execution_mode=None,
    kill_switch_activated=False,
    kill_switch_reason="",
)

Persisted risk state - survives restarts.

This state is validated and written in a versioned integrity envelope after every accepted order and on shutdown.

Example

state = RiskState(date="2023-10-15", daily_loss=1500.0) state.kill_switch_activated = True state.kill_switch_reason = "Max daily loss exceeded"

Note

The kill switch state persists across restarts and must be reset explicitly.

from_dict classmethod

from_dict(data)

Create RiskState from dictionary (for JSON loading).

Parameters:

Name	Type	Description	Default
data	dict	Dictionary with RiskState fields	required

Returns:

Type	Description
RiskState	RiskState instance

Source code in src/ml4t/live/safety.py

@classmethod
def from_dict(cls, data: dict) -> "RiskState":
    """Create RiskState from dictionary (for JSON loading).

    Args:
        data: Dictionary with RiskState fields

    Returns:
        RiskState instance
    """
    if not isinstance(data, dict):
        raise CorruptStateError("risk state payload must be an object")
    allowed = {
        "date",
        "daily_loss",
        "orders_placed",
        "high_water_mark",
        "session_start_equity",
        "persisted_positions",
        "persisted_pending_orders",
        "portable_strategy_state",
        "replacement_gaps",
        "shadow_portfolio",
        "execution_mode",
        "kill_switch_activated",
        "kill_switch_reason",
    }
    unknown = set(data) - allowed
    if unknown:
        raise CorruptStateError(
            f"risk state payload contains unsupported fields: {sorted(unknown)}"
        )
    state = cls(**data)
    state.validate()
    state.portable_strategy_state, _ = migrate_portable_strategy_state(
        state.portable_strategy_state,
        position_quantities=state._persisted_position_quantities(),
    )
    return state

validate

validate()

Validate persisted field types and numeric boundaries.

Source code in src/ml4t/live/safety.py

def validate(self) -> None:
    """Validate persisted field types and numeric boundaries."""
    try:
        date.fromisoformat(self.date)
    except (TypeError, ValueError) as error:
        raise CorruptStateError("risk state date must be an ISO calendar date") from error
    self._non_negative_number(self.daily_loss, "daily_loss")
    self._non_negative_number(self.high_water_mark, "high_water_mark")
    if self.session_start_equity is not None:
        self._non_negative_number(self.session_start_equity, "session_start_equity")
    if isinstance(self.orders_placed, bool) or not isinstance(self.orders_placed, int):
        raise CorruptStateError("risk state orders_placed must be a non-negative integer")
    if self.orders_placed < 0:
        raise CorruptStateError("risk state orders_placed must be a non-negative integer")
    if not isinstance(self.kill_switch_activated, bool) or not isinstance(
        self.kill_switch_reason, str
    ):
        raise CorruptStateError("risk state kill-switch fields have invalid types")
    if self.execution_mode is not None:
        try:
            ExecutionMode(self.execution_mode)
        except (TypeError, ValueError) as error:
            raise CorruptStateError("risk state execution_mode is invalid") from error
    if not isinstance(self.persisted_positions, dict):
        raise CorruptStateError("risk state persisted_positions must be an object")
    for asset, quantity in self.persisted_positions.items():
        if not isinstance(asset, str) or not asset:
            raise CorruptStateError("persisted position assets must be non-empty strings")
        self._finite_number(quantity, "persisted position quantity")
    if not isinstance(self.persisted_pending_orders, list):
        raise CorruptStateError("risk state persisted_pending_orders must be a list")
    for order in self.persisted_pending_orders:
        self._validate_pending_order(order)
    for name in (
        "portable_strategy_state",
        "replacement_gaps",
        "shadow_portfolio",
    ):
        value = getattr(self, name)
        if not isinstance(value, dict):
            raise CorruptStateError(f"risk state {name} must be an object")
        self._validate_json_tree(value, name)

to_dict

to_dict()

Convert to dictionary (for JSON saving).

Returns:

Type	Description
dict[str, Any]	Dictionary with all fields

Source code in src/ml4t/live/safety.py

def to_dict(self) -> dict[str, Any]:
    """Convert to dictionary (for JSON saving).

    Returns:
        Dictionary with all fields
    """
    data = {
        "date": self.date,
        "daily_loss": self.daily_loss,
        "orders_placed": self.orders_placed,
        "high_water_mark": self.high_water_mark,
        "kill_switch_activated": self.kill_switch_activated,
        "kill_switch_reason": self.kill_switch_reason,
    }
    if self.session_start_equity is not None:
        data["session_start_equity"] = self.session_start_equity
    if self.persisted_positions:
        data["persisted_positions"] = self.persisted_positions
    if self.persisted_pending_orders:
        data["persisted_pending_orders"] = self.persisted_pending_orders
    if self.portable_strategy_state:
        data["portable_strategy_state"] = self.portable_strategy_state
    if self.replacement_gaps:
        data["replacement_gaps"] = self.replacement_gaps
    if self.shadow_portfolio:
        data["shadow_portfolio"] = self.shadow_portfolio
    if self.execution_mode is not None:
        data["execution_mode"] = self.execution_mode
    return data

save_atomic staticmethod

save_atomic(state, filepath)

Save state with atomic write (write to .tmp then os.replace).

This prevents corruption if process dies mid-write.

Parameters:

Name	Type	Description	Default
state	RiskState	RiskState to save	required
filepath	str	Path to save to	required

Raises:

Type	Description
OSError	If write fails

Source code in src/ml4t/live/safety.py

@staticmethod
def save_atomic(state: "RiskState", filepath: str) -> None:
    """Save state with atomic write (write to .tmp then os.replace).

    This prevents corruption if process dies mid-write.

    Args:
        state: RiskState to save
        filepath: Path to save to

    Raises:
        OSError: If write fails
    """
    state.validate()
    SecureStateStore(filepath).save(state.to_dict(), expected_generation=None)

load staticmethod

load(filepath)

Load state from file.

Parameters:

Name	Type	Description	Default
filepath	str	Path to load from	required

Returns:

Type	Description
RiskState | None	RiskState if the file exists, otherwise None

Raises:

Type	Description
PersistenceSafetyError	If the file is unsafe, corrupt, or incompatible

Source code in src/ml4t/live/safety.py

@staticmethod
def load(filepath: str) -> "RiskState | None":
    """Load state from file.

    Args:
        filepath: Path to load from

    Returns:
        RiskState if the file exists, otherwise None

    Raises:
        PersistenceSafetyError: If the file is unsafe, corrupt, or incompatible
    """
    path = Path(filepath)
    if not path.exists():
        return None

    snapshot = SecureStateStore(filepath).load()
    if snapshot is None:
        return None
    return RiskState.from_dict(snapshot.payload)

create_for_today staticmethod

create_for_today()

Create new state for today's date.

Returns:

Type	Description
RiskState	RiskState with today's date and default values

Source code in src/ml4t/live/safety.py

@staticmethod
def create_for_today() -> "RiskState":
    """Create new state for today's date.

    Returns:
        RiskState with today's date and default values
    """
    return RiskState(date=datetime.now().strftime("%Y-%m-%d"))

VirtualPortfolio

VirtualPortfolio(initial_cash=100000.0)

Manages internal accounting for Shadow Mode (Paper Trading).

Tracks prior virtual fills so repeated shadow orders observe current positions

Problem: In shadow mode, returning fake Order objects without updating position state causes strategies to keep buying forever because get_position() always returns None.

Solution: Track shadow positions locally. When shadow_mode=True: - submit_order() updates this virtual portfolio - positions/get_position() return from this portfolio - Strategy sees realistic position state

Handles: - New positions - Position increases (weighted avg cost basis) - Position decreases (partial close) - Position close (quantity = 0) - Position flip (long -> short or vice versa)

Example

portfolio = VirtualPortfolio(initial_cash=100_000.0)

Simulate buy order fill

order = Order( asset="AAPL", side=OrderSide.BUY, quantity=100, filled_price=150.0, filled_quantity=100, ... ) portfolio.process_fill(order)

Check position

pos = portfolio.positions.get("AAPL") assert pos.quantity == 100 assert pos.entry_price == 150.0

Simulate sell order fill (close)

sell_order = Order( asset="AAPL", side=OrderSide.SELL, quantity=100, filled_price=155.0, filled_quantity=100, ... ) portfolio.process_fill(sell_order) assert "AAPL" not in portfolio.positions

Initialize virtual portfolio.

Parameters:

Name	Type	Description	Default
initial_cash	float	Starting cash balance (default: 100,000)	100000.0

Source code in src/ml4t/live/safety.py

def __init__(self, initial_cash: float = 100_000.0) -> None:
    """Initialize virtual portfolio.

    Args:
        initial_cash: Starting cash balance (default: 100,000)
    """
    self._initial_cash = initial_cash
    self._cash = initial_cash
    self._positions: dict[str, Position] = {}

positions property

positions

Get current positions (returns copy for safety).

Returns:

Type	Description
dict[str, Position]	Dictionary mapping asset symbol to Position

cash property

cash

Get current cash balance.

Returns:

Type	Description
float	Available cash

account_value property

account_value

Get total account value (cash + position market value).

Returns:

Type	Description
float	Total account value

process_fill

process_fill(order)

Update state based on filled shadow order.

Handles: - Weighted average cost basis for position increases - Position flipping (long -> short or vice versa) - Partial and full closes

Parameters:

Name	Type	Description	Default
order	Order	Filled Order object (must have filled_quantity and filled_price)	required

Source code in src/ml4t/live/safety.py

def process_fill(self, order: Order) -> None:
    """Update state based on filled shadow order.

    Handles:
    - Weighted average cost basis for position increases
    - Position flipping (long -> short or vice versa)
    - Partial and full closes

    Args:
        order: Filled Order object (must have filled_quantity and filled_price)
    """
    if not order.filled_quantity or not order.filled_price:
        logger.warning(f"VirtualPortfolio: Order {order.order_id} has no fill info")
        return

    asset = order.asset
    fill_qty = order.filled_quantity
    fill_price = order.filled_price
    transaction_value = fill_qty * fill_price

    # Cash impact
    if order.side == OrderSide.BUY:
        self._cash -= transaction_value
        signed_qty = fill_qty
    else:
        self._cash += transaction_value
        signed_qty = -fill_qty

    current = self._positions.get(asset)

    if current is None:
        # New position
        self._positions[asset] = Position(
            asset=asset,
            quantity=signed_qty,
            entry_price=fill_price,
            entry_time=datetime.now(),
            current_price=fill_price,
        )
        logger.info(
            f"Shadow: Opened {asset} {'LONG' if signed_qty > 0 else 'SHORT'} {abs(signed_qty)}"
        )

    else:
        old_qty = current.quantity
        new_qty = old_qty + signed_qty

        if new_qty == 0:
            # Position closed
            del self._positions[asset]
            logger.info(f"Shadow: Closed {asset}")

        elif (old_qty > 0 and new_qty < 0) or (old_qty < 0 and new_qty > 0):
            # Position flipped (e.g., Long 100 -> Sell 200 -> Short 100)
            self._positions[asset] = Position(
                asset=asset,
                quantity=new_qty,
                entry_price=fill_price,  # Reset basis on flip
                entry_time=datetime.now(),
                current_price=fill_price,
            )
            logger.info(f"Shadow: Flipped {asset} to {new_qty}")

        elif abs(new_qty) > abs(old_qty):
            # Increasing position - weighted average cost basis
            total_old = old_qty * current.entry_price
            total_new = signed_qty * fill_price
            new_avg = (total_old + total_new) / new_qty
            current.quantity = new_qty
            current.entry_price = abs(new_avg)
            current.current_price = fill_price
            logger.info(
                f"Shadow: Increased {asset} to {new_qty}, basis ${current.entry_price:.2f}"
            )

        else:
            # Decreasing position (partial close) - basis unchanged
            current.quantity = new_qty
            current.current_price = fill_price
            logger.info(f"Shadow: Reduced {asset} to {new_qty}")

update_prices

update_prices(prices)

Update current prices for accurate account value.

Parameters:

Name	Type	Description	Default
prices	dict[str, float]	Dictionary mapping asset symbol to current price	required

Source code in src/ml4t/live/safety.py

def update_prices(self, prices: dict[str, float]) -> None:
    """Update current prices for accurate account value.

    Args:
        prices: Dictionary mapping asset symbol to current price
    """
    for asset, price in prices.items():
        if asset in self._positions:
            self._positions[asset].current_price = price

to_state

to_state()

Return restart-safe shadow cash and position state.

Source code in src/ml4t/live/safety.py

def to_state(self) -> dict[str, Any]:
    """Return restart-safe shadow cash and position state."""
    return {
        "cash": self._cash,
        "positions": [
            {
                "asset": position.asset,
                "quantity": position.quantity,
                "entry_price": position.entry_price,
                "entry_time": position.entry_time.isoformat(),
                "current_price": position.current_price,
                "multiplier": position.multiplier,
                "context": position.context,
            }
            for position in self._positions.values()
        ],
    }

restore_state

restore_state(state)

Restore shadow state before runtime reconciliation.

Source code in src/ml4t/live/safety.py

def restore_state(self, state: dict[str, Any]) -> None:
    """Restore shadow state before runtime reconciliation."""
    self._cash = float(state.get("cash", self._cash))
    self._positions.clear()
    for raw in state.get("positions", ()):
        entry_time = datetime.fromisoformat(raw["entry_time"])
        self._positions[raw["asset"]] = Position(
            asset=raw["asset"],
            quantity=float(raw["quantity"]),
            entry_price=float(raw["entry_price"]),
            entry_time=entry_time,
            current_price=float(raw["current_price"]),
            multiplier=float(raw.get("multiplier", 1.0)),
            context=dict(raw.get("context", {})),
        )

Related Guides

• Backtest to Live
• Brokers
• Data Feeds
• Risk Controls
• Book Guide