fix: added types

src/nasdaq_protocols/fix/__init__.py

@@ -0,0 +1 @@
+from .types import *

src/nasdaq_protocols/fix/types.py

@@ -0,0 +1,162 @@
+from typing import Callable
+
+from nasdaq_protocols.common import TypeDefinition
+
+
+__all__ = [
+    'FixString',
+    'FixBool',
+    'FixInt',
+    'FixFloat',
+    'FixChar',
+    'FixQuantity',
+    'FixPrice',
+    'FixPriceOffset',
+    'FixAmount',
+    'FixMultipleValueString',
+    'FixCurrency',
+    'FixExchange',
+    'FixUTCTimeStamp',
+    'FixUTCTime',
+    'FixLocalMktDate',
+    'FixTzTime',
+    'FixTzTimestamp',
+    'FixDayOfMonth',
+]
+
+
+def _pack_str(str_: str) -> tuple[int, bytes]:
+    return len(str_), str_.encode('ascii')
+
+
+def _unpack_str(data: bytes) -> tuple[int, str]:
+    return len(data), data.decode('ascii')
+
+
+def _pack_int(val: int) -> tuple[int, bytes]:
+    return _pack_str(str(val))
+
+
+def _unpack_int(val: bytes) -> tuple[int, int]:
+    len_, str_ = _unpack_str(val)
+    return len_, int(str_)
+
+
+def _pack_float(val: int | float) -> tuple[int, bytes]:
+    return _pack_str(str(val))
+
+
+def _unpack_float(val: bytes) -> tuple[int, float]:
+    len_, str_ = _unpack_str(val)
+    return len_, float(str_)
+
+
+@TypeDefinition.add_type('fix_string')
+class FixString(TypeDefinition):
+    to_str: Callable[[str], str] = str
+    from_str: Callable[[str], str] = str
+    to_bytes: Callable[[str], tuple[int, bytes]] = _pack_str
+    from_bytes: Callable[[bytes], tuple[int, str]] = _unpack_str
+    hint = 'str'
+    type_cls = str
+    default_value = ''
+
+
+@TypeDefinition.add_type('fix_bool')
+class FixBool(TypeDefinition):
+    to_str: Callable[[bool], str] = lambda x: 'Y' if x else 'N'
+    from_str: Callable[[str], bool] = lambda x: x == 'Y' or x == 'y'
+    to_bytes: Callable[[bool], tuple[int, bytes]] = lambda x: (1, b'Y') if x else (1, b'N')
+    from_bytes: Callable[[bytes], tuple[int, bool]] = lambda x: (len(x), x == b'Y')
+    hint = 'bool'
+    type_cls = bool
+
+
+@TypeDefinition.add_type('fix_bool')
+class FixInt(TypeDefinition):
+    to_str: Callable[[int], str] = str
+    from_str: Callable[[str], int] = int
+    to_bytes: Callable[[int], tuple[int, bytes]] = _pack_int
+    from_bytes: Callable[[bytes], tuple[int, int]] = _unpack_int
+    hint = 'int'
+    type_cls = int
+
+
+@TypeDefinition.add_type('fix_float')
+class FixFloat(TypeDefinition):
+    to_str: Callable[[float], str] = str
+    from_str: Callable[[str], float] = float
+    to_bytes: Callable[[float | int], tuple[int, bytes]] = _pack_float
+    from_bytes: Callable[[bytes], tuple[int, float]] = _unpack_float
+    hint = 'float'
+    type_cls = float
+
+
+@TypeDefinition.add_type('fix_char')
+class FixChar(FixString):
+    ...
+
+
+@TypeDefinition.add_type('fix_quantity')
+class FixQuantity(FixFloat):
+    ...
+
+
+@TypeDefinition.add_type('fix_price')
+class FixPrice(FixFloat):
+    ...
+
+
+@TypeDefinition.add_type('fix_price_offset')
+class FixPriceOffset(FixFloat):
+    ...
+
+
+@TypeDefinition.add_type('fix_amount')
+class FixAmount(FixFloat):
+    ...
+
+
+@TypeDefinition.add_type('fix_multiple_value_string')
+class FixMultipleValueString(FixString):
+    ...
+
+
+@TypeDefinition.add_type('fix_currency')
+class FixCurrency(FixString):
+    ...
+
+
+@TypeDefinition.add_type('fix_exchange')
+class FixExchange(FixString):
+    ...
+
+
+@TypeDefinition.add_type('fix_utc_timestamp')
+class FixUTCTimeStamp(FixString):
+    ...
+
+
+@TypeDefinition.add_type('fix_utc_time')
+class FixUTCTime(FixString):
+    ...
+
+
+@TypeDefinition.add_type('fix_local_mkt_date')
+class FixLocalMktDate(FixString):
+    ...
+
+
+@TypeDefinition.add_type('fix_tx_time')
+class FixTzTime(FixString):
+    ...
+
+
+@TypeDefinition.add_type('fix_tz_timestamp')
+class FixTzTimestamp(FixString):
+    ...
+
+
+@TypeDefinition.add_type('fix_day_of_month')
+class FixDayOfMonth(FixInt):
+    ...

tests/test_fix_types.py

@@ -0,0 +1,100 @@
+import pytest
+
+from nasdaq_protocols.fix import *
+
+
+def test__fix_bool__to_str():
+    assert types.FixBool.to_str(True) is 'Y'
+    assert types.FixBool.to_str(False) is 'N'
+
+
+def test__fix_bool__from_str():
+    assert types.FixBool.from_str('Y') is True
+    assert types.FixBool.from_str('y') is True
+    assert types.FixBool.from_str('N') is False
+    assert types.FixBool.from_str('n') is False
+    assert types.FixBool.from_str('something') is False
+
+
+def test__fix_bool__to_bytes():
+    assert types.FixBool.to_bytes(True) == (1, b'Y')
+    assert types.FixBool.to_bytes(False) == (1, b'N')
+
+
+def test__fix_bool__from_bytes():
+    assert types.FixBool.from_bytes(b'Y') == (1, True)
+    assert types.FixBool.from_bytes(b'N') == (1, False)
+    assert types.FixBool.from_bytes(b'invalid') == (len('invalid'), False)
+
+
+def test__fix_string__to_str():
+    assert types.FixString.to_str('test') == 'test'
+
+
+def test__fix_string__from_str():
+    assert types.FixString.from_str('test') == 'test'
+
+
+def test__fix_string__to_bytes():
+    assert FixString.to_bytes('test') == (len('test'), b'test')
+    assert FixString.to_bytes('t') == (len('t'), b't')
+
+
+def test__fix_string__from_bytes():
+    assert FixString.from_bytes(b'test') == (len('test'), 'test')
+    assert FixString.from_bytes(b't') == (1, 't')
+
+
+def test__fix_int__to_str():
+    assert FixInt.to_str(0) == '0'
+    assert FixInt.to_str(1) == '1'
+    assert FixInt.to_str(12345) == '12345'
+
+
+def test__fix_int__from_str():
+    assert FixInt.from_str('0') == 0
+    assert FixInt.from_str('1') == 1
+    assert FixInt.from_str('12345') == 12345
+
+
+def test__fix_int__to_bytes():
+    assert FixInt.to_bytes(0) == (1, b'0')
+    assert FixInt.to_bytes(1) == (1, b'1')
+    assert FixInt.to_bytes(12345) == (5, b'12345')
+
+
+def test__fix_int__from_bytes():
+    assert FixInt.from_bytes(b'0') == (1, 0)
+    assert FixInt.from_bytes(b'1') == (1, 1)
+    assert FixInt.from_bytes(b'12345') == (5, 12345)
+    with pytest.raises(ValueError):
+        assert FixInt.from_bytes(b'a')
+
+
+##
+def test__fix_float__to_str():
+    assert FixFloat.to_str(0.1) == '0.1'
+    assert FixFloat.to_str(1.0) == '1.0'
+    assert FixFloat.to_str(12.345) == '12.345'
+
+
+def test__fix_float__from_str():
+    assert FixFloat.from_str('0.1') == 0.1
+    assert FixFloat.from_str('1.0') == 1.0
+    assert FixFloat.from_str('12.345') == 12.345
+
+
+def test__fix_float__to_bytes():
+    assert FixFloat.to_bytes(0.1) == (3, b'0.1')
+    assert FixFloat.to_bytes(1.0) == (3, b'1.0')
+    assert FixFloat.to_bytes(12.345) == (6, b'12.345')
+    assert FixFloat.to_bytes(1) == (1, b'1')
+
+
+def test__fix_float__from_bytes():
+    assert FixFloat.from_bytes(b'0.1') == (3, 0.1)
+    assert FixFloat.from_bytes(b'1.0') == (3, 1.0)
+    assert FixFloat.from_bytes(b'12.345') == (6, 12.345)
+    assert FixFloat.from_bytes(b'2') == (1, 2.0)
+    with pytest.raises(ValueError):
+        assert FixFloat.from_bytes(b'a')