28 lines
1.2 KiB
Python
28 lines
1.2 KiB
Python
from datetime import datetime
|
|
|
|
from idea159_arbsphere_federated_cross.solver import PlanDelta
|
|
from idea159_arbsphere_federated_cross.energi_bridge import EnergiBridge
|
|
from idea159_arbsphere_federated_cross.core import LocalArbProblem, SharedSignals
|
|
|
|
|
|
def test_merge_deltas_deduplicates_actions_and_supports_replay():
|
|
base = PlanDelta(actions=[
|
|
{"venue_from": "NYSE", "venue_to": "CROSS-VENUE", "instrument": "AAPL/GOOG", "size": 100, "time": "t1"}
|
|
], timestamp=datetime(2020, 1, 1))
|
|
|
|
# New delta with a duplicate action and a new one
|
|
new = PlanDelta(actions=[
|
|
{"venue_from": "NYSE", "venue_to": "CROSS-VENUE", "instrument": "AAPL/GOOG", "size": 100, "time": "t1"},
|
|
{"venue_from": "NYSE", "venue_to": "CROSS-VENUE", "instrument": "AAPL/GOOG", "size": 50, "time": "t2"},
|
|
], timestamp=datetime(2020, 1, 2))
|
|
|
|
merged = EnergiBridge.merge_deltas(base, new)
|
|
# Deduplication should keep 2 unique actions total (one duplicate removed)
|
|
assert isinstance(merged, PlanDelta)
|
|
assert len(merged.actions) == 2
|
|
|
|
# Deterministic replay should fold the same two deltas into the same final delta
|
|
replayed = EnergiBridge.replay_deltas([base, new])
|
|
assert isinstance(replayed, PlanDelta)
|
|
assert len(replayed.actions) == 2
|