1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
|
"""Tests for shared event types."""
from decimal import Decimal
from datetime import datetime, timezone
def make_candle():
from shared.models import Candle
return Candle(
symbol="BTCUSDT",
timeframe="1m",
open_time=datetime(2024, 1, 1, tzinfo=timezone.utc),
open=Decimal("50000"),
high=Decimal("51000"),
low=Decimal("49500"),
close=Decimal("50500"),
volume=Decimal("100"),
)
def make_signal():
from shared.models import Signal, OrderSide
return Signal(
strategy="test",
symbol="BTCUSDT",
side=OrderSide.BUY,
price=Decimal("50000"),
quantity=Decimal("0.01"),
reason="test signal",
)
def test_candle_event_serialize():
"""Test CandleEvent serializes to dict correctly."""
from shared.events import CandleEvent, EventType
candle = make_candle()
event = CandleEvent(data=candle)
d = event.to_dict()
assert d["type"] == EventType.CANDLE
assert d["data"]["symbol"] == "BTCUSDT"
assert d["data"]["timeframe"] == "1m"
def test_candle_event_deserialize():
"""Test CandleEvent round-trips through to_dict/from_raw."""
from shared.events import CandleEvent, EventType
candle = make_candle()
event = CandleEvent(data=candle)
d = event.to_dict()
restored = CandleEvent.from_raw(d)
assert restored.type == EventType.CANDLE
assert restored.data.symbol == "BTCUSDT"
assert restored.data.close == Decimal("50500")
def test_signal_event_serialize():
"""Test SignalEvent serializes to dict correctly."""
from shared.events import SignalEvent, EventType
signal = make_signal()
event = SignalEvent(data=signal)
d = event.to_dict()
assert d["type"] == EventType.SIGNAL
assert d["data"]["symbol"] == "BTCUSDT"
assert d["data"]["strategy"] == "test"
def test_event_from_dict_dispatch():
"""Test Event.from_dict dispatches to correct class."""
from shared.events import Event, CandleEvent, SignalEvent
candle = make_candle()
event = CandleEvent(data=candle)
d = event.to_dict()
restored = Event.from_dict(d)
assert isinstance(restored, CandleEvent)
signal = make_signal()
s_event = SignalEvent(data=signal)
sd = s_event.to_dict()
restored_s = Event.from_dict(sd)
assert isinstance(restored_s, SignalEvent)
|
