37 lines
1.3 KiB
Python
37 lines
1.3 KiB
Python
from datetime import datetime
|
|
from mercurymesh.contracts import MarketSignal, AggregatedSignal
|
|
from mercurymesh.pipeline import aggregate_signals
|
|
from mercurymesh.provenance import merkle_proof_for_signal, verify_provenance
|
|
|
|
|
|
def _make_signal(venue_id: str, symbol: str, t: datetime, f: dict) -> MarketSignal:
|
|
return MarketSignal(venue_id=venue_id, symbol=symbol, timestamp=t, features=f)
|
|
|
|
|
|
def test_aggregate_signals_basic():
|
|
t = datetime.utcnow()
|
|
s1 = _make_signal("venue-a", "ABC", t, {"liquidity_proxy": 1.0, "order_flow_intensity": 0.5})
|
|
s2 = _make_signal("venue-b", "XYZ", t, {"liquidity_proxy": 0.5, "order_flow_intensity": 0.8})
|
|
|
|
agg = aggregate_signals([s1, s2])
|
|
|
|
# Basic sanity checks
|
|
assert isinstance(agg, AggregatedSignal)
|
|
assert len(agg.venues) == 2
|
|
assert "liquidity_proxy" in agg.feature_vector
|
|
assert agg.nonce is not None
|
|
# merkle_proof should be present (deterministic from first signal)
|
|
assert agg.merkle_proof is not None
|
|
|
|
|
|
def test_provenance_roundtrip():
|
|
s = MarketSignal(
|
|
venue_id="venue-a",
|
|
symbol="ABC",
|
|
timestamp=datetime.utcnow(),
|
|
features={"liquidity_proxy": 1.0, "order_flow_intensity": 0.5},
|
|
)
|
|
proof = merkle_proof_for_signal(s)
|
|
# Use verify_provenance to check it matches the first-signal-derived proof
|
|
assert verify_provenance(proof, [s])
|