nautilus_interactive_brokers/gateway/
dockerized.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Dockerized IB Gateway management.
17
18#[cfg(feature = "gateway")]
19use std::{collections::HashMap, fmt::Debug, time::Duration};
20
21#[cfg(feature = "gateway")]
22use anyhow::Context;
23#[cfg(feature = "gateway")]
24use bollard::Docker;
25#[cfg(feature = "gateway")]
26use bollard::container::LogOutput;
27#[cfg(feature = "gateway")]
28use bollard::models::{
29    ContainerCreateBody, ContainerCreateResponse, HostConfig, PortBinding, RestartPolicy,
30    RestartPolicyNameEnum,
31};
32#[cfg(feature = "gateway")]
33use bollard::query_parameters::{
34    CreateContainerOptions, ListContainersOptions, LogsOptions, RemoveContainerOptions,
35    StartContainerOptions, StopContainerOptions,
36};
37#[cfg(feature = "gateway")]
38use futures_util::StreamExt;
39#[cfg(feature = "gateway")]
40use serde::{Deserialize, Serialize};
41
42#[cfg(feature = "gateway")]
43use crate::config::DockerizedIBGatewayConfig;
44#[cfg(feature = "gateway")]
45
46/// Container status enumeration.
47#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
48#[cfg_attr(
49    feature = "python",
50    pyo3::pyclass(
51        module = "nautilus_trader.core.nautilus_pyo3.interactive_brokers",
52        from_py_object
53    )
54)]
55pub enum ContainerStatus {
56    /// No container exists.
57    NoContainer = 1,
58    /// Container has been created but not started.
59    ContainerCreated = 2,
60    /// Container is starting.
61    ContainerStarting = 3,
62    /// Container has stopped.
63    ContainerStopped = 4,
64    /// Container is running but not logged in.
65    NotLoggedIn = 5,
66    /// Container is ready (running and logged in).
67    Ready = 6,
68    /// Unknown container status.
69    Unknown = 7,
70}
71
72/// Dockerized IB Gateway manager.
73///
74/// This struct manages the lifecycle of Interactive Brokers Gateway Docker containers,
75/// including creation, starting, stopping, and status checking.
76#[derive(Clone)]
77#[cfg_attr(
78    feature = "python",
79    pyo3::pyclass(
80        module = "nautilus_trader.core.nautilus_pyo3.interactive_brokers",
81        from_py_object
82    )
83)]
84#[cfg(feature = "gateway")]
85pub struct DockerizedIBGateway {
86    /// Configuration for the gateway.
87    config: DockerizedIBGatewayConfig,
88    /// Docker client.
89    pub(crate) docker: Docker,
90    /// Username for IB account.
91    username: String,
92    /// Password for IB account.
93    password: String,
94    /// Host address (always 127.0.0.1).
95    host: String,
96    /// Port for the gateway.
97    port: u16,
98    /// Container name.
99    container_name: String,
100}
101
102#[cfg(feature = "gateway")]
103impl Debug for DockerizedIBGateway {
104    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
105        f.debug_struct(stringify!(DockerizedIBGateway))
106            .field("host", &self.host)
107            .field("port", &self.port)
108            .field("container_name", &self.container_name)
109            .field("trading_mode", &self.config.trading_mode)
110            .finish_non_exhaustive()
111    }
112}
113
114#[cfg(feature = "gateway")]
115impl DockerizedIBGateway {
116    /// Base container name.
117    pub const CONTAINER_NAME: &'static str = "nautilus-ib-gateway";
118
119    /// Host API ports by trading mode.
120    pub const HOST_PORTS: &'static [(&'static str, u16)] = &[("Paper", 4002), ("Live", 4001)];
121
122    /// Container API ports exposed by the IB Gateway image.
123    pub const CONTAINER_PORTS: &'static [(&'static str, u16)] = &[("Paper", 4004), ("Live", 4003)];
124
125    /// Internal VNC port.
126    pub const VNC_PORT_INTERNAL: u16 = 5900;
127
128    fn host_port_for_mode(trading_mode: crate::config::TradingMode) -> u16 {
129        match trading_mode {
130            crate::config::TradingMode::Paper => 4002,
131            crate::config::TradingMode::Live => 4001,
132        }
133    }
134
135    fn container_port_for_mode(trading_mode: crate::config::TradingMode) -> u16 {
136        match trading_mode {
137            crate::config::TradingMode::Paper => 4004,
138            crate::config::TradingMode::Live => 4003,
139        }
140    }
141
142    fn logs_indicate_ready(logs: &str) -> bool {
143        logs.contains("Login has completed")
144            || logs.contains("Configuration tasks completed")
145            || logs.contains("Logged in to")
146            || logs.contains("Login successful")
147    }
148
149    /// Create a new DockerizedIBGateway from configuration.
150    ///
151    /// # Arguments
152    ///
153    /// * `config` - Configuration for the gateway
154    ///
155    /// # Errors
156    ///
157    /// Returns an error if:
158    /// - Username or password is not provided and not available in environment variables
159    /// - Docker client creation fails
160    pub fn new(config: DockerizedIBGatewayConfig) -> anyhow::Result<Self> {
161        // Load username from config or environment (clone to avoid partial move)
162        let username = config
163            .username
164            .clone()
165            .or_else(|| std::env::var("TWS_USERNAME").ok())
166            .ok_or_else(|| anyhow::anyhow!("username not set nor available in env TWS_USERNAME"))?;
167
168        // Load password from config or environment (clone to avoid partial move)
169        let password = config
170            .password
171            .clone()
172            .or_else(|| std::env::var("TWS_PASSWORD").ok())
173            .ok_or_else(|| anyhow::anyhow!("password not set nor available in env TWS_PASSWORD"))?;
174
175        // Connect to Docker
176        let docker = Docker::connect_with_local_defaults().context(
177            "Failed to connect to the local Docker daemon. Ensure Docker is running and the local Docker socket is available",
178        )?;
179
180        // Determine port based on trading mode
181        let mode_str = match config.trading_mode {
182            crate::config::TradingMode::Paper => "Paper",
183            crate::config::TradingMode::Live => "Live",
184        };
185        let port = Self::host_port_for_mode(config.trading_mode);
186
187        // Generate container name
188        let container_name = format!("{}-{}", Self::CONTAINER_NAME, mode_str).to_lowercase();
189
190        Ok(Self {
191            config,
192            docker,
193            username,
194            password,
195            host: "127.0.0.1".to_string(),
196            port,
197            container_name,
198        })
199    }
200
201    /// Get the container name.
202    pub fn container_name(&self) -> &str {
203        &self.container_name
204    }
205
206    /// Get the host address.
207    pub fn host(&self) -> &str {
208        &self.host
209    }
210
211    /// Get the port.
212    pub fn port(&self) -> u16 {
213        self.port
214    }
215
216    /// Check if the container is logged in by examining logs.
217    ///
218    /// # Arguments
219    ///
220    /// * `container_id` - The container ID to check
221    ///
222    /// # Errors
223    ///
224    /// Returns an error if log retrieval fails.
225    pub async fn is_logged_in(&self, container_id: &str) -> anyhow::Result<bool> {
226        let logs_options = LogsOptions {
227            stdout: true,
228            stderr: true,
229            ..Default::default()
230        };
231
232        let mut logs_stream = self.docker.logs(container_id, Some(logs_options));
233
234        let mut logged_in = false;
235
236        while let Some(log_result) = logs_stream.next().await {
237            let log_output = log_result.context("Failed to read log chunk")?;
238            // Handle LogOutput enum variants
239            let log_bytes = match log_output {
240                LogOutput::StdOut { message } | LogOutput::StdErr { message } => message,
241                LogOutput::StdIn { message } | LogOutput::Console { message } => message,
242            };
243            let log_string = String::from_utf8_lossy(&log_bytes);
244            if Self::logs_indicate_ready(&log_string) {
245                logged_in = true;
246                break;
247            }
248        }
249
250        Ok(logged_in)
251    }
252
253    /// Get the current container status.
254    ///
255    /// # Errors
256    ///
257    /// Returns an error if container inspection fails.
258    pub async fn container_status(&self) -> anyhow::Result<ContainerStatus> {
259        let list_options = ListContainersOptions {
260            all: true,
261            ..Default::default()
262        };
263        let containers = self
264            .docker
265            .list_containers(Some(list_options))
266            .await
267            .context("Failed to list containers")?;
268
269        let container = containers.iter().find(|c| {
270            c.names
271                .as_ref()
272                .and_then(|names| names.first())
273                .map(|name| name.trim_start_matches('/') == self.container_name)
274                .unwrap_or(false)
275        });
276
277        let Some(container) = container else {
278            return Ok(ContainerStatus::NoContainer);
279        };
280
281        let state = container
282            .state
283            .as_ref()
284            .map(|state| state.as_ref())
285            .unwrap_or("unknown");
286
287        match state {
288            "running" => {
289                let container_id = container
290                    .id
291                    .as_ref()
292                    .ok_or_else(|| anyhow::anyhow!("Container ID missing"))?;
293
294                if self.is_logged_in(container_id).await.unwrap_or(false) {
295                    Ok(ContainerStatus::Ready)
296                } else {
297                    Ok(ContainerStatus::ContainerStarting)
298                }
299            }
300            "stopped" | "exited" => Ok(ContainerStatus::ContainerStopped),
301            "created" => Ok(ContainerStatus::ContainerCreated),
302            _ => Ok(ContainerStatus::Unknown),
303        }
304    }
305
306    /// Start the gateway container.
307    ///
308    /// # Arguments
309    ///
310    /// * `wait` - Optional wait time in seconds (overrides config timeout)
311    ///
312    /// # Errors
313    ///
314    /// Returns an error if container creation or startup fails.
315    pub async fn start(&mut self, wait: Option<u64>) -> anyhow::Result<()> {
316        tracing::debug!("Ensuring gateway is running");
317
318        let status = self.container_status().await?;
319
320        let broken_statuses = [
321            ContainerStatus::NotLoggedIn,
322            ContainerStatus::ContainerStopped,
323            ContainerStatus::ContainerCreated,
324            ContainerStatus::Unknown,
325        ];
326
327        match status {
328            ContainerStatus::NoContainer => {
329                tracing::debug!("No container, starting");
330            }
331            status if broken_statuses.contains(&status) => {
332                tracing::debug!("Status {:?}, removing existing container", status);
333                self.stop().await?;
334            }
335            ContainerStatus::Ready | ContainerStatus::ContainerStarting => {
336                tracing::debug!("Status {:?}, using existing container", status);
337                return Ok(());
338            }
339            _ => {}
340        }
341
342        tracing::debug!("Starting new container");
343
344        // Determine port mappings
345        let host_port = Self::host_port_for_mode(self.config.trading_mode);
346        let container_port = Self::container_port_for_mode(self.config.trading_mode);
347
348        let mut port_bindings = HashMap::new();
349        port_bindings.insert(
350            format!("{}/tcp", container_port),
351            Some(vec![PortBinding {
352                host_ip: Some(self.host.clone()),
353                host_port: Some(host_port.to_string()),
354            }]),
355        );
356
357        if let Some(vnc_port) = self.config.vnc_port {
358            port_bindings.insert(
359                format!("{}/tcp", Self::VNC_PORT_INTERNAL),
360                Some(vec![PortBinding {
361                    host_ip: Some(self.host.clone()),
362                    host_port: Some(vnc_port.to_string()),
363                }]),
364            );
365        }
366
367        // Prepare environment variables
368        let mode_str = match self.config.trading_mode {
369            crate::config::TradingMode::Paper => "paper",
370            crate::config::TradingMode::Live => "live",
371        };
372        let env = vec![
373            format!("TWS_USERID={}", self.username),
374            format!("TWS_PASSWORD={}", self.password),
375            format!("TRADING_MODE={}", mode_str),
376            format!(
377                "READ_ONLY_API={}",
378                if self.config.read_only_api {
379                    "yes"
380                } else {
381                    "no"
382                }
383            ),
384            "EXISTING_SESSION_DETECTED_ACTION=primary".to_string(),
385        ];
386
387        // Create container configuration
388        let container_config = ContainerCreateBody {
389            image: Some(self.config.container_image.clone()),
390            hostname: Some(self.container_name.clone()),
391            host_config: Some(HostConfig {
392                port_bindings: Some(port_bindings),
393                restart_policy: Some(RestartPolicy {
394                    name: Some(RestartPolicyNameEnum::ALWAYS),
395                    maximum_retry_count: None,
396                }),
397                ..Default::default()
398            }),
399            env: Some(env),
400            ..Default::default()
401        };
402
403        // Create container
404        let create_options = CreateContainerOptions {
405            name: Some(self.container_name.clone()),
406            ..Default::default()
407        };
408
409        let create_response: ContainerCreateResponse = self
410            .docker
411            .create_container(Some(create_options), container_config)
412            .await
413            .context("Failed to create container")?;
414
415        let container_id = create_response.id;
416
417        // Start container
418        self.docker
419            .start_container(&container_id, None::<StartContainerOptions>)
420            .await
421            .context("Failed to start container")?;
422
423        tracing::debug!(
424            "Container `{}` starting, waiting for ready",
425            self.container_name
426        );
427
428        // Wait for container to be ready
429        let wait_time = wait.unwrap_or(self.config.timeout);
430        let mut waited = 0u64;
431
432        while waited < wait_time {
433            if self.is_logged_in(&container_id).await.unwrap_or(false) {
434                tracing::debug!(
435                    "Gateway `{}` ready. VNC port is {:?}",
436                    self.container_name,
437                    self.config.vnc_port
438                );
439                return Ok(());
440            }
441
442            tracing::debug!("Waiting for IB Gateway to start");
443            tokio::time::sleep(Duration::from_secs(1)).await;
444            waited += 1;
445        }
446
447        anyhow::bail!(
448            "Gateway `{}` not ready after {} seconds",
449            self.container_name,
450            wait_time
451        )
452    }
453
454    /// Safely start the gateway, handling container already exists errors.
455    ///
456    /// # Arguments
457    ///
458    /// * `wait` - Optional wait time in seconds
459    ///
460    /// # Errors
461    ///
462    /// Returns an error if startup fails (other than container exists).
463    pub async fn safe_start(&mut self, wait: Option<u64>) -> anyhow::Result<()> {
464        match self.start(wait).await {
465            Ok(()) => Ok(()),
466            Err(e) if e.to_string().contains("already exists") => {
467                tracing::warn!("Container already exists, continuing");
468                Ok(())
469            }
470            Err(e) => Err(e),
471        }
472    }
473
474    /// Stop and remove the gateway container.
475    ///
476    /// # Errors
477    ///
478    /// Returns an error if container stop or removal fails.
479    pub async fn stop(&self) -> anyhow::Result<()> {
480        let list_options = ListContainersOptions {
481            all: true,
482            ..Default::default()
483        };
484        let containers = self
485            .docker
486            .list_containers(Some(list_options))
487            .await
488            .context("Failed to list containers")?;
489
490        let container = containers.iter().find(|c| {
491            c.names
492                .as_ref()
493                .and_then(|names| names.first())
494                .map(|name| name.trim_start_matches('/') == self.container_name)
495                .unwrap_or(false)
496        });
497
498        if let Some(container) = container {
499            if let Some(container_id) = &container.id {
500                // Stop container if running
501                if matches!(
502                    container.state.as_ref().map(|state| state.as_ref()),
503                    Some("running")
504                ) {
505                    self.docker
506                        .stop_container(container_id, None::<StopContainerOptions>)
507                        .await
508                        .context("Failed to stop container")?;
509                }
510
511                // Remove container
512                let remove_options = RemoveContainerOptions {
513                    force: true,
514                    ..Default::default()
515                };
516
517                self.docker
518                    .remove_container(container_id, Some(remove_options))
519                    .await
520                    .context("Failed to remove container")?;
521
522                tracing::debug!("Stopped and removed container `{}`", self.container_name);
523            }
524        }
525
526        Ok(())
527    }
528}
529
530/// Stub implementation when gateway feature is disabled.
531#[cfg(not(feature = "gateway"))]
532#[derive(Debug)]
533pub struct DockerizedIBGateway;
534
535#[cfg(not(feature = "gateway"))]
536impl DockerizedIBGateway {
537    /// # Errors
538    ///
539    /// Returns an error if the Dockerized IB Gateway cannot be created or started.
540    pub fn new(_config: crate::config::DockerizedIBGatewayConfig) -> anyhow::Result<Self> {
541        anyhow::bail!("Gateway feature is not enabled. Build with --features gateway")
542    }
543}
544
545#[cfg(all(test, feature = "gateway"))]
546mod tests {
547    use rstest::rstest;
548
549    use super::DockerizedIBGateway;
550    use crate::config::TradingMode;
551
552    #[rstest]
553    #[case(TradingMode::Paper, 4002)]
554    #[case(TradingMode::Live, 4001)]
555    fn host_port_matches_trading_mode(#[case] trading_mode: TradingMode, #[case] expected: u16) {
556        assert_eq!(
557            DockerizedIBGateway::host_port_for_mode(trading_mode),
558            expected
559        );
560    }
561
562    #[rstest]
563    #[case(TradingMode::Paper, 4004)]
564    #[case(TradingMode::Live, 4003)]
565    fn container_port_matches_trading_mode(
566        #[case] trading_mode: TradingMode,
567        #[case] expected: u16,
568    ) {
569        assert_eq!(
570            DockerizedIBGateway::container_port_for_mode(trading_mode),
571            expected
572        );
573    }
574
575    #[rstest]
576    #[case(TradingMode::Paper, 4002)]
577    #[case(TradingMode::Live, 4001)]
578    fn new_reports_the_host_api_port(#[case] trading_mode: TradingMode, #[case] expected: u16) {
579        let gateway = DockerizedIBGateway::new(
580            crate::config::DockerizedIBGatewayConfig::builder()
581                .username("test-user".to_string())
582                .password("test-password".to_string())
583                .trading_mode(trading_mode)
584                .build(),
585        )
586        .unwrap();
587
588        assert_eq!(gateway.port(), expected);
589    }
590
591    #[rstest]
592    #[case("Forking ::: Starting IBC Gateway", false)]
593    #[case("Started IB Gateway", false)]
594    #[case("Login has completed", true)]
595    #[case("Configuration tasks completed", true)]
596    #[case("Logged in to backend", true)]
597    #[case("Login successful", true)]
598    fn ready_log_markers_are_strict(#[case] logs: &str, #[case] expected: bool) {
599        assert_eq!(DockerizedIBGateway::logs_indicate_ready(logs), expected);
600    }
601}
nautilus_interactive_brokers/gateway/dockerized.rs

nautilus_interactive_brokers/gateway/
dockerized.rs
