module fix_tester

FIX Protocol Unit Tester.

Global Variables

  • nan

class FIXTester

FIX protocol tester.

Attributes:

  • registered_orders: registered orders (sent via FIXTester)
  • schema: fix schema for validation (optional)
  • conn_init: fix connection initiator
  • conn_accept: fix virtual acceptor (simulated by FIXTester)
  • initiator_sent: messages sent by initiator
  • acceptor_rcv_que: raw messages received by (simulated acceptor)
  • acceptor_sent: FIXMessages sent by simulated acceptor

method __init__

__init__(
    schema: FIXSchema | None = None,
    connection: AsyncFIXConnection | None = None
)

Initialize FIXTester.

Args:

  • schema: (optional) FIXSchema for validating incoming/outgoing messages
  • connection: (optional) fix initiator connection

method acceptor_sent_query

acceptor_sent_query(
    tags: tuple[FTag | str | int] | None = None,
    index: int = -1
) → dict[FTag | str, str]

Query message sent from FIXTester to initiator.

Args:

  • tags: tuple of tag numbers
  • index: index of self.acceptor_sent list, default -1

Returns:

  • dict {FTag.MsgSeqNum: "34", "12382": "some value"}

method fix_cxl_request

fix_cxl_request(o: FIXNewOrderSingle) → FIXMessage

Generates FIXNewOrderSingle cancel request + optional validation.

Returns: cancel FIXMessage

method fix_cxlrep_reject_msg

fix_cxlrep_reject_msg(cxl_req: FIXMessage, ord_status: FOrdStatus) → FIXMessage

Generates synthetic ORDERCANCELREJECT message.

Args:

  • cxl_req: cancel / replace request FIXMessage (from initiator)
  • ord_status: new order status generated by this reply

Returns: ORDERCANCELREJECT FIXMessage

method fix_exec_report_msg

fix_exec_report_msg(
    order: FIXNewOrderSingle,
    clord_id: str,
    exec_type: FExecType,
    ord_status: FOrdStatus,
    cum_qty: float = nan,
    leaves_qty: float = nan,
    last_qty: float = nan,
    price: float = nan,
    order_qty: float = nan,
    orig_clord_id: str = None,
    avg_price: float = 0.0
) → FIXMessage

Generates synthetic EXECUTIONREPORT.

Args:

  • order: FIXNewOrderSingle to report
  • clord_id: report ClOrdID tag
  • exec_type: report ExecType
  • ord_status: report FOrdStatus
  • cum_qty: filled qty
  • leaves_qty: remaining qty
  • last_qty: last trade qty
  • price: (ONLY IN REPLACE) price of replaced order
  • order_qty: (ONLY IN REPLACE) qty of replaced order
  • orig_clord_id: report OrigClOrdID
  • avg_price: average fill price

Returns: EXECUTIONREPORT FIXMessage

method fix_rep_request

fix_rep_request(
    o: FIXNewOrderSingle,
    price: float = nan,
    qty: float = nan
) → FIXMessage

Generates FIXNewOrderSingle replace request + optional validating.

Args:

  • o: FIXNewOrderSingle
  • price: new price, nan - to skip
  • qty: new quantity, nan - to skip

Returns: replace FIXMessage

method initiator_sent_query

initiator_sent_query(
    tags: tuple[FTag | str | int] | None = None,
    index: int = -1
) → dict[FTag | str, str]

Query message sent from initiator to FixTester.

Args:

  • tags: tuple of tag numbers
  • index: index of self.initiator_sent list, default -1

Returns:

  • dict {FTag.MsgSeqNum: "34", "12382": "some value"}

method msg_heartbeat

msg_heartbeat(test_req_id=None) → FIXMessage

Generates message HEARTBEAT + schema validation.

Args:

  • test_req_id: if Heartbeat in reply to TESTREQUEST

method msg_logon

msg_logon(tags: dict | None = None) → FIXMessage

Generates message LOGON + schema validation.

Args:

  • tags: extra tags

Returns: FIXMessage

method msg_logout

msg_logout() → FIXMessage

Generates message LOGOUT + schema validation.

method msg_resend_request

msg_resend_request(begin_seq_no, end_seq_no='0') → FIXMessage

Generates message RESENDREQUEST + schema validation.

Args:

  • begin_seq_no: seq no to start with
  • end_seq_no: end seq no, "0" - means all

method msg_sequence_reset

msg_sequence_reset(
    msg_seq_num: int,
    new_seq_no: int,
    is_gap_fill: bool = False
) → FIXMessage

Generates message SEQUENCERESET + schema validation.

Args:

  • msg_seq_num: FTag.MsgSeqNum value
  • new_seq_no: FTag.NewSeqNo value
  • is_gap_fill: FTag.GapFillFlag "Y"/"N"

method msg_test_request

msg_test_request(test_req_id) → FIXMessage

Generates message TESTREQUEST + schema validation.

Args:

  • test_req_id: unique test request id

method order_register_single

order_register_single(o: FIXNewOrderSingle)

Registers FIXNewOrderSingle.

method process_msg_acceptor

process_msg_acceptor(index=None)

Processes messages queued by initiator.send_msg().

Args:

  • index: None - processes all messages in que, number - only one at that index

method reply

reply(msg: FIXMessage)

Manually reply to the initiator with arbitrary FIXMessage.

Args:

  • msg: arbitrary FIXMessage

method reset_messages

reset_messages()

Reset messages queues of initiator and acceptor.

method set_next_num

set_next_num(num_in=None, num_out=None)

Set expected session seq nums for simulated acceptor.

Args:

  • num_in: next num in expected by simulated acceptor
  • num_out: next num out sent by simulated acceptor

This file was automatically generated via lazydocs.