""" arrow_tick_reader.py --------------------- Attaches to the named shared-memory segment written by ArrowTickStreamer.mq5 / ArrowBufferWriter.mqh and reconstructs the current tick batch as a zero-copy pyarrow.RecordBatch, by building real ArrowArray / ArrowSchema C Data Interface structs in this process's own memory and pointing their buffer fields directly at offsets inside the mmap - no memcpy, no per-tick Python loop. Layout constants below MUST mirror ArrowBufferWriter.mqh exactly. This is the Python side of the series' three-way schema contract: the mqh compile-time ARROW_TICK_COLUMNS, the writer's per-tick column loop, and COLUMN_NAMES/COLUMN_FORMATS here must all agree, or reads will silently misinterpret bytes. Requires: pyarrow, polars (optional, for the zero-copy DataFrame demo) """ #+-------------------------------------------------------------------+ #| Imports | #+-------------------------------------------------------------------+ import ctypes import mmap import time from dataclasses import dataclass import pyarrow as pa #+-------------------------------------------------------------------+ #| Schema contract - must mirror ArrowBufferWriter.mqh exactly | #+-------------------------------------------------------------------+ ARROW_TICK_COLUMNS = 6 # column count, MT5 <-> Python COLUMN_NAMES = ["time_msc", "bid", "ask", "last", "volume_real", "flags"] # MT5 -> Python order COLUMN_FORMATS = ["l", "g", "g", "g", "g", "l"] # Arrow format codes: int64 / float64 COLUMN_IS_INT64 = [1, 0, 0, 0, 0, 1] # 1 = int64 storage, 0 = float64 storage HEADER_SIZE = 64 # fixed bytes reserved for the slot header ELEM_SIZE = 8 # every column stored as one 8-byte word ALIGN = 64 # Arrow spec: buffers padded to 64-byte boundaries #+-------------------------------------------------------------------+ #| _align64 | #| Rounds a byte count up to the next 64-byte boundary, | #| mirroring Align64() in ArrowBufferWriter.mqh. | #+-------------------------------------------------------------------+ def _align64(n: int) -> int: return ((n + ALIGN - 1) // ALIGN) * ALIGN #+-------------------------------------------------------------------+ #| layout_for_capacity | #| Reproduces the exact byte layout ArrowBufferWriter.mqh | #| computed on the MT5 side, from capacity alone. | #+-------------------------------------------------------------------+ def layout_for_capacity(capacity: int): #--- no pointers are ever serialized into shared memory, so both #--- sides just need to agree on this formula, not on an address. validity_bytes = _align64((capacity + 7) // 8) data_bytes = _align64(capacity * ELEM_SIZE) column_block_bytes = validity_bytes + data_bytes slot_size = HEADER_SIZE + ARROW_TICK_COLUMNS * column_block_bytes total_size = slot_size * 2 return { "validity_bytes": validity_bytes, "data_bytes": data_bytes, "column_block_bytes": column_block_bytes, "slot_size": slot_size, "total_size": total_size, } #+-------------------------------------------------------------------+ #| SlotHeader | #| ctypes mirror of ArrowSlotHeader (ArrowBufferWriter.mqh) - | #| field order and sizes must stay byte-identical. | #+-------------------------------------------------------------------+ class SlotHeader(ctypes.LittleEndianStructure): _pack_ = 1 _fields_ = [ ("generation", ctypes.c_int64), # odd = write in progress, even = stable ("batch_seq", ctypes.c_int64), # monotonically increasing batch id ("tick_count", ctypes.c_int32), # ticks actually populated this batch ("column_count", ctypes.c_int32), # must equal ARROW_TICK_COLUMNS ("write_timestamp_msc", ctypes.c_int64), # EA-side wall clock at publish time ("capacity", ctypes.c_int32), # max ticks this slot can hold ("reserved0", ctypes.c_int32), # padding ("reserved1", ctypes.c_uint8 * 24), # padding out to 64 bytes total ] assert ctypes.sizeof(SlotHeader) == HEADER_SIZE #+-------------------------------------------------------------------+ #| ArrowSchema / ArrowArray | #| C Data Interface structs, per the Apache Arrow spec. | #| Field order and types are fixed by the spec, not by us. | #+-------------------------------------------------------------------+ class ArrowSchema(ctypes.Structure): pass class ArrowArray(ctypes.Structure): pass ArrowSchema._fields_ = [ ("format", ctypes.c_char_p), ("name", ctypes.c_char_p), ("metadata", ctypes.c_char_p), ("flags", ctypes.c_int64), ("n_children", ctypes.c_int64), ("children", ctypes.POINTER(ctypes.POINTER(ArrowSchema))), ("dictionary", ctypes.POINTER(ArrowSchema)), ("release", ctypes.c_void_p), ("private_data", ctypes.c_void_p), ] ArrowArray._fields_ = [ ("length", ctypes.c_int64), ("null_count", ctypes.c_int64), ("offset", ctypes.c_int64), ("n_buffers", ctypes.c_int64), ("n_children", ctypes.c_int64), ("buffers", ctypes.POINTER(ctypes.c_void_p)), ("children", ctypes.POINTER(ctypes.POINTER(ArrowArray))), ("dictionary", ctypes.POINTER(ArrowArray)), ("release", ctypes.c_void_p), ("private_data", ctypes.c_void_p), ] #+-------------------------------------------------------------------+ #| SCHEMA_RELEASE_FN / ARRAY_RELEASE_FN | #| ctypes function-pointer types for the release callbacks below. | #+-------------------------------------------------------------------+ SCHEMA_RELEASE_FN = ctypes.CFUNCTYPE(None, ctypes.POINTER(ArrowSchema)) ARRAY_RELEASE_FN = ctypes.CFUNCTYPE(None, ctypes.POINTER(ArrowArray)) #+-------------------------------------------------------------------+ #| _release_schema | #| Marks an ArrowSchema as released. No heap memory to free - | #| buffers live in the mmap, not on this struct - but a real | #| callback must exist or pyarrow refuses the import. | #+-------------------------------------------------------------------+ def _release_schema(schema_ptr): #--- per the Arrow spec, release() must set its own struct's release #--- field to NULL once done; this is exactly the error you get if #--- you skip this step (consumers treat NULL as "already released"). schema_ptr.contents.release = ctypes.cast(None, ctypes.c_void_p) #+-------------------------------------------------------------------+ #| _release_array | #| Marks an ArrowArray as released. See _release_schema. | #+-------------------------------------------------------------------+ def _release_array(array_ptr): array_ptr.contents.release = ctypes.cast(None, ctypes.c_void_p) #+-------------------------------------------------------------------+ #| _Pinned | #| Keeps every ctypes object an imported Array points at alive | #| for as long as the Array itself is alive. Losing a reference | #| here is the most common way to get a use-after-free segfault. | #+-------------------------------------------------------------------+ @dataclass class _Pinned: schema: ArrowSchema array: ArrowArray buffers: object format_buf: object schema_release_cb: object = None array_release_cb: object = None #+-------------------------------------------------------------------+ #| ArrowTickReader | #| Mirror image of CArrowTickBridge on the Python side - attaches | #| to the named shared memory segment and reconstructs tick | #| batches as zero-copy pyarrow.RecordBatch objects. | #+-------------------------------------------------------------------+ class ArrowTickReader: #+---------------------------------------------------------------+ #| __init__ | #| Attaches to the named mapping and computes this process's | #| own base address for it. | #+---------------------------------------------------------------+ def __init__(self, mapping_name: str, capacity: int): self.mapping_name = mapping_name self.capacity = capacity self.layout = layout_for_capacity(capacity) #--- on Windows, tagname with fileno=-1 attaches to an existing #--- named mapping if one already exists under that name - this #--- is how we join the segment the EA created, without #--- re-creating or resizing it. self.mm = mmap.mmap(-1, self.layout["total_size"], tagname=mapping_name) self.base_addr = ctypes.addressof(ctypes.c_char.from_buffer(self.mm)) self._pins = [] # keeps imported batches' backing structs alive #+---------------------------------------------------------------+ #| _slot_offset | #| Converts a slot index (0 or 1) into its byte offset, | #| mirroring SlotOffset() in ArrowBufferWriter.mqh. | #+---------------------------------------------------------------+ def _slot_offset(self, slot_index: int) -> int: return slot_index * self.layout["slot_size"] #+---------------------------------------------------------------+ #| _read_header | #| Reads one slot's 64-byte header without any consistency | #| checking - callers use _consistent_header() instead. | #+---------------------------------------------------------------+ def _read_header(self, slot_index: int) -> SlotHeader: off = self._slot_offset(slot_index) hdr = SlotHeader.from_buffer_copy(self.mm, off) return hdr #+---------------------------------------------------------------+ #| _consistent_header | #| Seqlock read: retry until generation is even and unchanged | #| across the read. Mirrors the write-side odd/even flip in | #| CArrowTickBridge::WriteBatch(). | #+---------------------------------------------------------------+ def _consistent_header(self, slot_index: int, retries: int = 50) -> SlotHeader: for _ in range(retries): h1 = self._read_header(slot_index) if h1.generation % 2 != 0: time.sleep(0) # writer mid-publish, yield and retry continue h2 = self._read_header(slot_index) if h1.generation == h2.generation: return h1 raise TimeoutError( f"Slot {slot_index} did not stabilise after {retries} retries - " "writer may be publishing faster than this reader can poll." ) #+---------------------------------------------------------------+ #| latest_slot | #| Picks whichever of the two slots has the higher stable | #| batch_seq. | #+---------------------------------------------------------------+ def latest_slot(self) -> int: h0 = self._consistent_header(0) h1 = self._consistent_header(1) return 0 if h0.batch_seq >= h1.batch_seq else 1 #+---------------------------------------------------------------+ #| read_batch | #| Assembles a full six-column zero-copy RecordBatch from | #| whichever slot is currently stable and newest. | #+---------------------------------------------------------------+ def read_batch(self) -> "pa.RecordBatch": slot = self.latest_slot() hdr = self._consistent_header(slot) if hdr.tick_count == 0: return pa.RecordBatch.from_arrays( [pa.array([], type=pa.int64())] * 2 + [pa.array([], type=pa.float64())] * 4, names=COLUMN_NAMES, ) if hdr.column_count != ARROW_TICK_COLUMNS: raise RuntimeError( f"Schema contract violated: header says column_count=" f"{hdr.column_count}, reader expects {ARROW_TICK_COLUMNS}" ) slot_base = self.base_addr + self._slot_offset(slot) col_base = slot_base + HEADER_SIZE n = hdr.tick_count arrays = [] for c in range(ARROW_TICK_COLUMNS): data_ptr = col_base + self.layout["validity_bytes"] arr = self._import_column(data_ptr, n, COLUMN_FORMATS[c], COLUMN_NAMES[c]) arrays.append(arr) col_base += self.layout["column_block_bytes"] return pa.RecordBatch.from_arrays(arrays, names=COLUMN_NAMES) #+---------------------------------------------------------------+ #| _import_column | #| Builds real ArrowSchema/ArrowArray C structs pointing at a | #| single column's data inside the mmap, and imports them into | #| pyarrow with zero copies. | #+---------------------------------------------------------------+ def _import_column(self, data_ptr: int, n: int, fmt: str, name: str) -> pa.Array: #--- null_count=0 lets us hand Arrow a NULL validity buffer even #--- though ArrowBufferWriter.mqh still physically writes one - #--- see the article's note on why this is spec-legal. buffers = (ctypes.c_void_p * 2)(None, data_ptr) format_buf = ctypes.create_string_buffer(fmt.encode("ascii")) schema_release_cb = SCHEMA_RELEASE_FN(_release_schema) array_release_cb = ARRAY_RELEASE_FN(_release_array) schema = ArrowSchema() schema.format = ctypes.cast(format_buf, ctypes.c_char_p) schema.name = None schema.metadata = None schema.flags = 0 schema.n_children = 0 schema.children = None schema.dictionary = None schema.release = ctypes.cast(schema_release_cb, ctypes.c_void_p) schema.private_data = None array = ArrowArray() array.length = n array.null_count = 0 array.offset = 0 array.n_buffers = 2 array.n_children = 0 array.buffers = ctypes.cast(buffers, ctypes.POINTER(ctypes.c_void_p)) array.children = None array.dictionary = None array.release = ctypes.cast(array_release_cb, ctypes.c_void_p) array.private_data = None #--- every object referenced above must outlive the returned #--- Array, or pyarrow later dereferences freed memory. pin = _Pinned( schema=schema, array=array, buffers=buffers, format_buf=format_buf, schema_release_cb=schema_release_cb, array_release_cb=array_release_cb, ) self._pins.append(pin) # kept alive for this reader's lifetime schema_ptr = ctypes.addressof(schema) array_ptr = ctypes.addressof(array) return pa.Array._import_from_c(array_ptr, schema_ptr) #+---------------------------------------------------------------+ #| close | #| Releases the mmap. Any RecordBatch derived from this reader | #| must not be touched after this call - see Edge Cases. | #+---------------------------------------------------------------+ def close(self): self.mm.close() #+-------------------------------------------------------------------+ #| demo | #| Minimal end-to-end usage example: read one batch and hand | #| it to pandas and (optionally) polars. | #+-------------------------------------------------------------------+ def demo(mapping_name: str = "Local\\ArrowTickBridge_XAUUSD", capacity: int = 256): reader = ArrowTickReader(mapping_name, capacity) try: batch = reader.read_batch() print(f"Imported {batch.num_rows} rows, {batch.num_columns} columns, zero-copy.") print(batch.to_pandas().head()) #--- zero-copy handoff into polars - no re-materialization of #--- values, only Arrow buffer pointers change ownership #--- bookkeeping. try: import polars as pl df = pl.from_arrow(batch) print(df.tail(3)) except ImportError: print("polars not installed - skipping DataFrame demo.") finally: reader.close() if __name__ == "__main__": demo()