deltatrace-deterministic-re.../adapters/fix_feed_adapter.py

29 lines
841 B
Python

"""Minimal FIX feed simulator adapter.
This module provides a tiny, test-friendly FIX-like feed generator that
produces LocalEvent payloads compatible with the core types.
"""
from __future__ import annotations
import time
from deltatrace.core import LocalEvent
class FIXFeedSimulator:
def __init__(self, instrument: str = "EURUSD", source_id: str = "FIXFIX") -> None:
self.instrument = instrument
self.source_id = source_id
self._counter = 0
def tick(self) -> LocalEvent:
self._counter += 1
evt = LocalEvent(
instrument=self.instrument,
timestamp=time.time(),
data_hash=f"hash-{self._counter}",
source_id=self.source_id,
version="0.1",
payload={"type": "MDTick", "seq": self._counter},
)
return evt