Merge pull request #55 from vgreg/25-add-support-for-itch-41-format-2

Author: vgreg

samples/itch41/Step0_ExtractSymbols.py

@@ -0,0 +1,61 @@
+"""Listing Symbols in an ITCH 4.1 File
+
+This script shows how to extract all available symbols from an ITCH 4.1 file.
+Based on the pattern from the MeatPy documentation notebooks.
+"""
+
+from pathlib import Path
+from meatpy.itch41 import ITCH41MessageReader
+
+# Define the path to our sample data file
+script_dir = Path(__file__).parent
+data_dir = script_dir / "data"
+file_path = data_dir / "S083012-v41.txt.gz"
+
+print(f"Reading ITCH 4.1 file: {file_path}")
+if file_path.exists():
+    file_size_mb = file_path.stat().st_size / (1024**2)
+    print(f"File size: {file_size_mb:.2f} MB")
+else:
+    print("⚠️  Sample file not found - this is expected in most setups")
+    print("You can download ITCH 4.1 sample files or use your own data")
+    exit(1)
+
+symbols = set()
+message_count = 0
+
+print("Reading ITCH 4.1 file to extract symbols...")
+
+with ITCH41MessageReader(file_path) as reader:
+    for message in reader:
+        message_count += 1
+
+        # Stock Directory messages (type 'R') contain symbol information
+        if message.type == b"R":
+            symbol = message.stock.decode().strip()
+            symbols.add(symbol)
+
+        # For ITCH 4.1, we can break early since stock directory messages
+        # typically appear at the beginning of the file
+        if message_count >= 50000:
+            break
+
+print(f"Found {len(symbols)} symbols after processing {message_count:,} messages")
+
+symbols = sorted(symbols)
+
+# Display first 20 symbols
+print("\nFirst 20 symbols:")
+for symbol in symbols[:20]:
+    print(f"  {symbol}")
+
+if len(symbols) > 20:
+    print(f"  ... and {len(symbols) - 20} more")
+
+# Save symbols to file
+output_file = data_dir / "itch41_symbols.txt"
+with open(output_file, "w") as f:
+    for symbol in symbols:
+        f.write(f"{symbol}\n")
+
+print(f"\n✅ Symbols saved to: {output_file}")

samples/itch41/Step1_Parsing.py

@@ -0,0 +1,64 @@
+"""Extracting Specific Symbols from ITCH 4.1 File
+
+This script demonstrates how to create a new ITCH 4.1 file containing only data
+for specific symbols of interest. Based on the pattern from MeatPy documentation.
+"""
+
+from pathlib import Path
+
+from meatpy.itch41 import ITCH41MessageReader, ITCH41Writer
+
+# Define paths
+data_dir = Path("data")
+input_file = data_dir / "S083012-v41.txt.gz"
+output_file = data_dir / "S083012-v41-AAPL-SPY.itch41.gz"
+
+# Symbols we want to extract
+target_symbols = ["AAPL", "SPY"]
+
+print(f"Input file: {input_file}")
+if input_file.exists():
+    input_size_mb = input_file.stat().st_size / (1024**2)
+    print(f"Input file size: {input_size_mb:.2f} MB")
+else:
+    print("⚠️  Sample file not found - this is expected in most setups")
+    print("You can download ITCH 4.1 sample files or use your own data")
+    exit(1)
+
+print(f"Extracting symbols: {target_symbols}")
+print(f"Output file: {output_file}")
+
+# Process the file and filter for target symbols
+message_count = 0
+with (
+    ITCH41MessageReader(input_file) as reader,
+    ITCH41Writer(output_file, symbols=target_symbols) as writer,
+):
+    for message in reader:
+        message_count += 1
+        writer.process_message(message)
+
+        # Progress update every 10,000 messages
+        if message_count % 10000 == 0:
+            print(f"Processed {message_count:,} messages...")
+
+print(f"Total messages processed: {message_count:,}")
+
+# # Check the filtered file
+# if output_file.exists():
+#     new_message_count = 0
+#     with ITCH41MessageReader(output_file) as reader:
+#         for message in reader:
+#             print(message)
+#             new_message_count += 1
+
+#     print(f"Total messages in filtered file: {new_message_count:,}")
+#     output_size_mb = output_file.stat().st_size / (1024**2)
+#     print(f"Output file size: {output_size_mb:.2f} MB")
+
+#     size_reduction = (1 - output_size_mb / input_size_mb) * 100
+#     print(f"Size reduction: {size_reduction:.1f}%")
+
+#     print(f"\n✅ Filtered file created: {output_file}")
+# else:
+#     print("❌ Failed to create output file")

samples/itch41/Step2_Processing.py

@@ -0,0 +1,48 @@
+import datetime
+from pathlib import Path
+
+from meatpy.event_handlers.lob_recorder import LOBRecorder
+from meatpy.itch41 import ITCH41MarketProcessor, ITCH41MessageReader
+from meatpy.lob import ExecutionPriorityExceptionList
+from meatpy.writers.parquet_writer import ParquetWriter
+
+# Define paths and parameters
+data_dir = Path("data")
+
+file_path = data_dir / "S083012-v41-AAPL-SPY.itch41.gz"
+outfile_path = data_dir / "spy_tob.parquet"
+book_date = datetime.datetime(2012, 8, 30)
+
+
+with ITCH41MessageReader(file_path) as reader, ParquetWriter(outfile_path) as writer:
+    processor = ITCH41MarketProcessor("SPY", book_date)
+
+    # We only care about the top of book
+    tob_recorder = LOBRecorder(writer=writer, max_depth=1)
+    # Generate a list of timedeltas from 9:30 to 16:00 (inclusive) in 1-minute increments
+    market_open = book_date + datetime.timedelta(hours=9, minutes=30)
+    market_close = book_date + datetime.timedelta(hours=16, minutes=0)
+    record_timestamps = [
+        market_open + datetime.timedelta(minutes=i)
+        for i in range(int((market_close - market_open).total_seconds() // 60) + 1)
+    ]
+    tob_recorder.record_timestamps = record_timestamps
+
+    # Attach the recorders to the processor
+    processor.handlers.append(tob_recorder)
+
+    message_types = set()
+
+    for i, message in enumerate(reader):
+        # if i % 10_000 == 0:
+        #     print(f"Processing message {i:,}...")
+        if message.type != b"T":
+            print(message)
+        try:
+            processor.process_message(message)
+        except ExecutionPriorityExceptionList as e:
+            print(f"Execution priority exception: {e}")
+        message_types.add(message.type)
+
+    print(f"Processed {i + 1:,} messages.")
+    print(f"Message types: {message_types}")

src/meatpy/__init__.py

@@ -62,3 +62,18 @@
     "TradeRef",
     "Qualifiers",
 ]
+
+# ITCH format imports (available when format-specific modules are imported)
+try:
+    from . import itch41  # noqa: F401
+
+    __all__.extend(["itch41"])
+except ImportError:
+    pass
+
+try:
+    from . import itch50  # noqa: F401
+
+    __all__.extend(["itch50"])
+except ImportError:
+    pass

src/meatpy/itch41/__init__.py

@@ -0,0 +1,60 @@
+"""ITCH 4.1 market data subpackage.
+
+This package provides message types, parsers, processors, and recorders for handling ITCH 4.1 market data in MeatPy.
+"""
+
+from .itch41_exec_trade_recorder import ITCH41ExecTradeRecorder
+from .itch41_market_message import (
+    AddOrderMessage,
+    AddOrderMPIDMessage,
+    BrokenTradeMessage,
+    CrossTradeMessage,
+    MarketParticipantPositionMessage,
+    OrderCancelMessage,
+    OrderDeleteMessage,
+    OrderExecutedMessage,
+    OrderExecutedPriceMessage,
+    OrderReplaceMessage,
+    RegSHOMessage,
+    SecondsMessage,
+    StockDirectoryMessage,
+    StockTradingActionMessage,
+    SystemEventMessage,
+    TradeMessage,
+)
+from .itch41_market_processor import ITCH41MarketProcessor
+from .itch41_message_reader import ITCH41MessageReader
+from .itch41_ofi_recorder import ITCH41OFIRecorder
+from .itch41_order_event_recorder import ITCH41OrderEventRecorder
+from .itch41_top_of_book_message_recorder import (
+    ITCH41TopOfBookMessageRecorder,
+)
+from .itch41_writer import ITCH41Writer
+
+__all__ = [
+    "ITCH41ExecTradeRecorder",
+    "ITCH41MarketMessage",
+    "ITCH41MarketProcessor",
+    "ITCH41MessageParser",
+    "ITCH41MessageReader",
+    "ITCH41Writer",
+    "ITCH41OFIRecorder",
+    "ITCH41OrderEventRecorder",
+    "ITCH41TopOfBookMessageRecorder",
+    "AddOrderMessage",
+    "AddOrderMPIDMessage",
+    "BrokenTradeMessage",
+    "CrossTradeMessage",
+    "MarketParticipantPositionMessage",
+    "OrderCancelMessage",
+    "OrderDeleteMessage",
+    "OrderExecutedMessage",
+    "OrderExecutedPriceMessage",
+    "OrderReplaceMessage",
+    "RegSHOMessage",
+    "SecondsMessage",
+    "StockDirectoryMessage",
+    "StockTradingActionMessage",
+    "SystemEventMessage",
+    "TradeMessage",
+]

src/meatpy/itch41/itch41_exec_trade_recorder.py

@@ -0,0 +1,132 @@
+"""ITCH 4.1 execution trade recorder for limit order books.
+
+This module provides the ITCH41ExecTradeRecorder class, which records trade
+executions from ITCH 4.1 market data and exports them to CSV files.
+"""
+
+from typing import Any
+
+from ..market_event_handler import MarketEventHandler
+from .itch41_market_message import (
+    OrderExecutedMessage,
+    OrderExecutedPriceMessage,
+    TradeMessage,
+)
+
+
+class ITCH41ExecTradeRecorder(MarketEventHandler):
+    """Records trade executions from ITCH 4.1 market data.
+
+    This recorder detects and records trade executions, including both
+    visible and hidden trades, and exports them to CSV format.
+
+    Attributes:
+        records: List of recorded trade execution records
+    """
+
+    def __init__(self) -> None:
+        """Initialize the ITCH41ExecTradeRecorder."""
+        self.records: list[Any] = []
+
+    def message_event(
+        self,
+        market_processor,
+        timestamp,
+        message,
+    ) -> None:
+        """Detect messages that represent trade executions and record them.
+
+        Args:
+            market_processor: The market processor instance
+            timestamp: The timestamp of the message
+            message: The market message to process
+        """
+        lob = market_processor.lob
+        if lob is None:
+            return
+
+        if isinstance(message, OrderExecutedMessage):
+            # An executed order will ALWAYS be against top of book
+            # because of price priority, so record.
+            if lob.ask_order_on_book(message.order_ref):
+                record = {
+                    "MessageType": "Exec",
+                    "Volume": message.shares,
+                    "OrderID": message.order_ref,
+                }
+                record["Queue"] = "Ask"
+                record["Price"] = lob.ask_levels[0].price
+                try:
+                    (queue, i, j) = lob.find_order(message.order_ref)
+                    record["OrderTimestamp"] = queue[i].queue[j].timestamp
+                except Exception:
+                    record["OrderTimestamp"] = ""
+                self.records.append((timestamp, record))
+            elif lob.bid_order_on_book(message.order_ref):
+                record = {
+                    "MessageType": "Exec",
+                    "Volume": message.shares,
+                    "OrderID": message.order_ref,
+                }
+                record["Queue"] = "Bid"
+                record["Price"] = lob.bid_levels[0].price
+                try:
+                    (queue, i, j) = lob.find_order(message.order_ref)
+                    record["OrderTimestamp"] = queue[i].queue[j].timestamp
+                except Exception:
+                    record["OrderTimestamp"] = ""
+                self.records.append((timestamp, record))
+        elif isinstance(message, TradeMessage):
+            if message.side == b"S":
+                record = {
+                    "MessageType": "ExecHid",
+                    "Volume": message.shares,
+                    "OrderID": "",
+                    "OrderTimestamp": "",
+                }
+                record["Queue"] = "Ask"
+                record["Price"] = message.price
+                self.records.append((timestamp, record))
+            elif message.side == b"B":
+                record = {
+                    "MessageType": "ExecHid",
+                    "Volume": message.shares,
+                    "OrderID": "",
+                    "OrderTimestamp": "",
+                }
+                record["Queue"] = "Bid"
+                record["Price"] = message.price
+                self.records.append((timestamp, record))
+        elif isinstance(message, OrderExecutedPriceMessage):
+            if len(lob.ask_levels) > 0 and lob.ask_levels[0].order_on_book(
+                message.order_ref
+            ):
+                record = {
+                    "MessageType": "ExecPrice",
+                    "Queue": "Ask",
+                    "Volume": message.shares,
+                    "OrderID": message.order_ref,
+                    "Price": message.price,
+                }
+                try:
+                    (queue, i, j) = lob.find_order(message.order_ref)
+                    record["OrderTimestamp"] = queue[i].queue[j].timestamp
+                except Exception:
+                    record["OrderTimestamp"] = ""
+                self.records.append((timestamp, record))
+            elif len(lob.bid_levels) > 0 and lob.bid_levels[0].order_on_book(
+                message.order_ref
+            ):
+                record = {
+                    "MessageType": "ExecPrice",
+                    "Queue": "Bid",
+                    "Volume": message.shares,
+                    "OrderID": message.order_ref,
+                    "Price": message.price,
+                }
+                try:
+                    (queue, i, j) = lob.find_order(message.order_ref)
+                    record["OrderTimestamp"] = queue[i].queue[j].timestamp
+                except Exception:
+                    record["OrderTimestamp"] = ""
+                self.records.append((timestamp, record))