wingfoil-python 8.0.0

//! Python bindings for the FIX protocol adapter.

use crate::py_element::PyElement;
use crate::py_stream::PyStream;
use crate::types::{DICT_INSERT_INFALLIBLE, INTO_PY_INFALLIBLE, LIST_NEW_INFALLIBLE};

use pyo3::prelude::*;
use pyo3::types::{PyDict, PyList};
use wingfoil::StreamOperators;
use wingfoil::adapters::fix::{
    FixMessage, FixPollMode, FixSessionStatus, fix_accept, fix_connect, fix_connect_tls,
};

/// Connect to a FIX acceptor and return `(data_stream, status_stream)`.
///
/// Args:
///     host: FIX acceptor hostname
///     port: FIX acceptor port
///     sender_comp_id: SenderCompID for this session
///     target_comp_id: TargetCompID of the counterparty
///
/// Returns:
///     A tuple of `(data_stream, status_stream)` where data_stream yields
///     lists of message dicts and status_stream yields session status strings.
#[pyfunction]
pub fn py_fix_connect(
    host: String,
    port: u16,
    sender_comp_id: String,
    target_comp_id: String,
) -> (PyStream, PyStream) {
    let (data, status) = fix_connect(
        &host,
        port,
        &sender_comp_id,
        &target_comp_id,
        FixPollMode::Threaded,
    );

    let py_data = data.map(|burst| {
        Python::attach(|py| {
            let items: Vec<Py<PyAny>> = burst.into_iter().map(|msg| msg_to_py(py, &msg)).collect();
            PyElement::new(
                PyList::new(py, items)
                    .expect(LIST_NEW_INFALLIBLE)
                    .into_any()
                    .unbind(),
            )
        })
    });

    let py_status = status.map(|burst| {
        Python::attach(|py| {
            let items: Vec<Py<PyAny>> = burst.into_iter().map(|s| status_to_py(py, &s)).collect();
            PyElement::new(
                PyList::new(py, items)
                    .expect(LIST_NEW_INFALLIBLE)
                    .into_any()
                    .unbind(),
            )
        })
    });

    (PyStream(py_data), PyStream(py_status))
}

/// Connect to a TLS-secured FIX acceptor (e.g. LMAX) and return
/// `(data_stream, status_stream, sender)`.
///
/// The sender is a `FixSender` whose `send()` method sends a FIX message dict:
/// ```python
/// sender.send({"msg_type": "V", "fields": [(262, "req1"), (263, "1")]})
/// ```
///
/// Args:
///     host: FIX acceptor hostname
///     port: FIX acceptor port (typically 443 for TLS)
///     sender_comp_id: SenderCompID (usually your username)
///     target_comp_id: TargetCompID of the counterparty
///     password: Optional password (tag 554 in Logon)
#[pyfunction]
#[pyo3(signature = (host, port, sender_comp_id, target_comp_id, password=None))]
pub fn py_fix_connect_tls(
    host: String,
    port: u16,
    sender_comp_id: String,
    target_comp_id: String,
    password: Option<String>,
) -> (PyStream, PyStream, Py<PyAny>) {
    let fix = fix_connect_tls(
        &host,
        port,
        &sender_comp_id,
        &target_comp_id,
        password.as_deref(),
    );

    let py_data = fix.data.map(|burst| {
        Python::attach(|py| {
            let items: Vec<Py<PyAny>> = burst.into_iter().map(|msg| msg_to_py(py, &msg)).collect();
            PyElement::new(
                PyList::new(py, items)
                    .expect(LIST_NEW_INFALLIBLE)
                    .into_any()
                    .unbind(),
            )
        })
    });

    let py_status = fix.status.map(|burst| {
        Python::attach(|py| {
            let items: Vec<Py<PyAny>> = burst.into_iter().map(|s| status_to_py(py, &s)).collect();
            PyElement::new(
                PyList::new(py, items)
                    .expect(LIST_NEW_INFALLIBLE)
                    .into_any()
                    .unbind(),
            )
        })
    });

    let sender_obj = Python::attach(|py| {
        let sender_cls = PyFixSender {
            sender: fix.sender(),
        };
        Py::new(py, sender_cls)
            .expect("invariant: allocating a PyFixSender pyclass cannot fail")
            .into_any()
    });

    (PyStream(py_data), PyStream(py_status), sender_obj)
}

/// Bind a FIX acceptor and return `(data_stream, status_stream)`.
///
/// Args:
///     port: Port to listen on
///     sender_comp_id: SenderCompID for this session
///     target_comp_id: Expected TargetCompID of the connecting initiator
#[pyfunction]
pub fn py_fix_accept(
    port: u16,
    sender_comp_id: String,
    target_comp_id: String,
) -> (PyStream, PyStream) {
    let (data, status) = fix_accept(
        port,
        &sender_comp_id,
        &target_comp_id,
        FixPollMode::Threaded,
    );

    let py_data = data.map(|burst| {
        Python::attach(|py| {
            let items: Vec<Py<PyAny>> = burst.into_iter().map(|msg| msg_to_py(py, &msg)).collect();
            PyElement::new(
                PyList::new(py, items)
                    .expect(LIST_NEW_INFALLIBLE)
                    .into_any()
                    .unbind(),
            )
        })
    });

    let py_status = status.map(|burst| {
        Python::attach(|py| {
            let items: Vec<Py<PyAny>> = burst.into_iter().map(|s| status_to_py(py, &s)).collect();
            PyElement::new(
                PyList::new(py, items)
                    .expect(LIST_NEW_INFALLIBLE)
                    .into_any()
                    .unbind(),
            )
        })
    });

    (PyStream(py_data), PyStream(py_status))
}

/// Python wrapper around [`FixSender`] for sending outbound FIX messages.
#[pyclass(unsendable)]
struct PyFixSender {
    sender: wingfoil::adapters::fix::FixSender,
}

#[pymethods]
impl PyFixSender {
    /// Send a FIX message on the session.
    ///
    /// Args:
    ///     msg: A dict with "msg_type" (str) and "fields" (list of [tag, value] pairs).
    fn send(&self, msg: &Bound<'_, PyDict>) -> PyResult<()> {
        let msg_type = msg
            .get_item("msg_type")?
            .ok_or_else(|| pyo3::exceptions::PyKeyError::new_err("missing 'msg_type'"))?
            .extract::<String>()?;

        let fields_obj = msg
            .get_item("fields")?
            .ok_or_else(|| pyo3::exceptions::PyKeyError::new_err("missing 'fields'"))?;
        let fields_list = fields_obj.cast::<PyList>()?;

        let mut fields = Vec::new();
        for item in fields_list.iter() {
            let pair = item.cast::<PyList>()?;
            if pair.len() != 2 {
                return Err(pyo3::exceptions::PyValueError::new_err(
                    "each field must be [tag, value]",
                ));
            }
            let tag: u32 = pair.get_item(0)?.extract()?;
            let value: String = pair.get_item(1)?.extract()?;
            fields.push((tag, value));
        }

        self.sender
            .send(FixMessage {
                msg_type,
                seq_num: 0,
                sending_time: wingfoil::NanoTime::ZERO,
                fields,
            })
            .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))
    }
}

fn msg_to_py(py: Python<'_>, msg: &FixMessage) -> Py<PyAny> {
    let dict = PyDict::new(py);
    dict.set_item("msg_type", &msg.msg_type)
        .expect(DICT_INSERT_INFALLIBLE);
    dict.set_item("seq_num", msg.seq_num)
        .expect(DICT_INSERT_INFALLIBLE);
    let fields: Vec<(u32, &str)> = msg.fields.iter().map(|(t, v)| (*t, v.as_str())).collect();
    dict.set_item("fields", fields)
        .expect(DICT_INSERT_INFALLIBLE);
    dict.into_any().unbind()
}

fn status_to_py(py: Python<'_>, status: &FixSessionStatus) -> Py<PyAny> {
    match status {
        FixSessionStatus::Disconnected => "disconnected"
            .into_pyobject(py)
            .expect(INTO_PY_INFALLIBLE)
            .into_any()
            .unbind(),
        FixSessionStatus::LoggingIn => "logging_in"
            .into_pyobject(py)
            .expect(INTO_PY_INFALLIBLE)
            .into_any()
            .unbind(),
        FixSessionStatus::LoggedIn => "logged_in"
            .into_pyobject(py)
            .expect(INTO_PY_INFALLIBLE)
            .into_any()
            .unbind(),
        FixSessionStatus::LoggedOut(reason) => {
            let dict = PyDict::new(py);
            dict.set_item("status", "logged_out")
                .expect(DICT_INSERT_INFALLIBLE);
            dict.set_item("reason", reason.as_deref().unwrap_or(""))
                .expect(DICT_INSERT_INFALLIBLE);
            dict.into_any().unbind()
        }
        FixSessionStatus::Error(msg) => {
            let dict = PyDict::new(py);
            dict.set_item("status", "error")
                .expect(DICT_INSERT_INFALLIBLE);
            dict.set_item("message", msg.as_str())
                .expect(DICT_INSERT_INFALLIBLE);
            dict.into_any().unbind()
        }
    }
}