"""
benchmark_compare.py
---------------------
Self-contained benchmark - does not require MT5 to be running. It writes
synthetic tick batches into local memory using the exact same Arrow
columnar layout ArrowBufferWriter.mqh produces, then times two read paths
at increasing batch sizes:

  1. zero_copy_read()  - pyarrow C Data Interface import (this article)
  2. legacy_struct_read() - struct.unpack per tick into a list of tuples,
                             the approach used by the earlier flat shared-
                             memory IPC bridge article

This quantifies the "zero-copy" claim instead of just asserting it.
"""

#+-------------------------------------------------------------------+
#| Imports                                                           |
#+-------------------------------------------------------------------+
import ctypes
import struct
import time

import pandas as pd
import pyarrow as pa

from arrow_tick_reader import (
    ARROW_TICK_COLUMNS, COLUMN_NAMES, COLUMN_IS_INT64,
    HEADER_SIZE, layout_for_capacity, ArrowSchema, ArrowArray, _Pinned,
    SCHEMA_RELEASE_FN, ARRAY_RELEASE_FN, _release_schema, _release_array,
)

#+-------------------------------------------------------------------+
#| Benchmark configuration                                           |
#+-------------------------------------------------------------------+
BATCH_SIZES = [64, 256, 1024, 4096]  # ticks per batch, swept for the table
REPEATS = 200                        # repeats per batch size, for a stable median


#+-------------------------------------------------------------------+
#| make_synthetic_columns                                            |
#| Six columns of plausible XAUUSD tick values, as flat              |
#| Python lists - no MT5 connection required.                        |
#+-------------------------------------------------------------------+
def make_synthetic_columns(n: int):
    base_time = 1_700_000_000_000
    time_msc = [base_time + i for i in range(n)]
    bid = [2400.00 + 0.01 * (i % 50) for i in range(n)]
    ask = [b + 0.03 for b in bid]
    last = [0.0] * n
    volume_real = [1.0 + 0.1 * (i % 7) for i in range(n)]
    flags = [6] * n  # TICK_FLAG_BID | TICK_FLAG_ASK
    return [time_msc, bid, ask, last, volume_real, flags]


#+-------------------------------------------------------------------+
#| build_arrow_column_buffer                                         |
#| Writes one column's data buffer exactly as                        |
#| ArrowBufferWriter.mqh would (validity bitmap all-1s,              |
#| then packed 8-byte values).                                       |
#+-------------------------------------------------------------------+
def build_arrow_column_buffer(values, is_int64: int, layout) -> bytearray:
    buf = bytearray(layout["column_block_bytes"])
    n = len(values)
    for i in range(n):
        byte_idx, bit_idx = divmod(i, 8)
        buf[byte_idx] |= (1 << bit_idx)
    fmt = "<q" if is_int64 else "<d"
    off = layout["validity_bytes"]
    for i, v in enumerate(values):
        struct.pack_into(fmt, buf, off + i * 8, int(v) if is_int64 else float(v))
    return buf


#+-------------------------------------------------------------------+
#| zero_copy_read                                                    |
#| Imports all six columns via the Arrow C Data Interface,           |
#| pointing straight at a local bytearray instead of shared          |
#| memory - isolates import cost from IPC cost.                      |
#+-------------------------------------------------------------------+
def zero_copy_read(column_buffers: list, n: int) -> pa.RecordBatch:
    pins = []
    arrays = []
    for c in range(ARROW_TICK_COLUMNS):
        buf = column_buffers[c]
        #--- pointer straight into the bytearray's own memory - no copy.
        addr = ctypes.addressof((ctypes.c_char * len(buf)).from_buffer(buf))
        fmt = "l" if COLUMN_IS_INT64[c] else "g"

        buffers = (ctypes.c_void_p * 2)(None, addr)
        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

        pins.append(_Pinned(
            schema=schema, array=array, buffers=buffers, format_buf=format_buf,
            schema_release_cb=schema_release_cb, array_release_cb=array_release_cb,
        ))
        arrays.append(pa.Array._import_from_c(ctypes.addressof(array), ctypes.addressof(schema)))

    batch = pa.RecordBatch.from_arrays(arrays, names=COLUMN_NAMES)
    return batch, pins  # pins returned so caller controls their lifetime


#+-------------------------------------------------------------------+
#| legacy_struct_read                                                |
#| Mirrors the earlier CreateFileMapping bridge article: unpack      |
#| every field of every tick by hand into Python tuples, then        |
#| build a DataFrame. The O(n) path zero-copy import avoids.         |
#+-------------------------------------------------------------------+
def legacy_struct_read(column_buffers: list, n: int, layout) -> pd.DataFrame:
    rows = []
    offsets = [layout["validity_bytes"] + i * 8 for i in range(n)]
    for i in range(n):
        row = []
        for c in range(ARROW_TICK_COLUMNS):
            buf = column_buffers[c]
            off = offsets[i]
            if COLUMN_IS_INT64[c]:
                (v,) = struct.unpack_from("<q", buf, off)
            else:
                (v,) = struct.unpack_from("<d", buf, off)
            row.append(v)
        rows.append(tuple(row))
    return pd.DataFrame(rows, columns=COLUMN_NAMES)


#+-------------------------------------------------------------------+
#| run                                                               |
#| Times both read paths across BATCH_SIZES and prints the           |
#| comparison table shown in the article.                            |
#+-------------------------------------------------------------------+
def run():
    print(f"{'batch':>8} | {'zero-copy (ms)':>16} | {'legacy struct (ms)':>20} | speedup")
    print("-" * 62)
    for n in BATCH_SIZES:
        layout = layout_for_capacity(n)
        cols = make_synthetic_columns(n)
        column_buffers = [
            build_arrow_column_buffer(cols[c], COLUMN_IS_INT64[c], layout)
            for c in range(ARROW_TICK_COLUMNS)
        ]

        #--- warm up import machinery once before timing
        _batch, _pins = zero_copy_read(column_buffers, n)

        t0 = time.perf_counter()
        for _ in range(REPEATS):
            batch, pins = zero_copy_read(column_buffers, n)
            _ = batch.num_rows  # touch the result, exclude to_pandas() cost
        t_arrow = (time.perf_counter() - t0) / REPEATS * 1000

        t0 = time.perf_counter()
        for _ in range(REPEATS):
            df = legacy_struct_read(column_buffers, n, layout)
        t_legacy = (time.perf_counter() - t0) / REPEATS * 1000

        speedup = t_legacy / t_arrow if t_arrow > 0 else float("inf")
        print(f"{n:>8} | {t_arrow:>16.4f} | {t_legacy:>20.4f} | {speedup:>6.1f}x")


if __name__ == "__main__":
    run()