59 lines
1.7 KiB
Python
59 lines
1.7 KiB
Python
from datetime import datetime, timedelta
|
|
|
|
from mercurymesh.ir import (
|
|
LocalMarketContext,
|
|
AggregatedSignalIR,
|
|
PlanDeltaIR,
|
|
DualVariables,
|
|
PrivacyBudgetEntry,
|
|
AuditLogEntry,
|
|
TimeRound,
|
|
GoCRegistryEntry,
|
|
)
|
|
|
|
|
|
def test_local_market_context_and_roundtrip():
|
|
now = datetime.utcnow()
|
|
ctx = LocalMarketContext(
|
|
venue_id="venue-a",
|
|
symbol="XYZ",
|
|
timeframe="1m",
|
|
objective="liquidity",
|
|
timestamp=now,
|
|
)
|
|
assert ctx.venue_id == "venue-a"
|
|
assert ctx.timestamp == now
|
|
|
|
|
|
def test_aggregated_signal_ir_and_plan_delta():
|
|
now = datetime.utcnow()
|
|
agg = AggregatedSignalIR(
|
|
version="v1",
|
|
venues=["venue-a", "venue-b"],
|
|
feature_vector={"liquidity_proxy": 0.7},
|
|
privacy_budget=0.05,
|
|
nonce="n-1",
|
|
)
|
|
pd = PlanDeltaIR(
|
|
contract_id="c1",
|
|
author="agent-1",
|
|
timestamp=now,
|
|
actions={"adjust": "scale", "scale": 0.5},
|
|
)
|
|
assert "liquidity_proxy" in agg.feature_vector
|
|
assert pd.contract_id == "c1"
|
|
|
|
|
|
def test_misc_ir_types():
|
|
now = datetime.utcnow()
|
|
dv = DualVariables(multipliers={"alpha": 1.0})
|
|
pb = PrivacyBudgetEntry(signal_class="liquidity", budget=1.0, expiry=now)
|
|
al = AuditLogEntry(entry="created", signer="auditor", timestamp=now, contract_id="c1")
|
|
tr = TimeRound(round_id="r1", start_ts=now, end_ts=now + timedelta(seconds=60))
|
|
reg = GoCRegistryEntry(adapter_id="a1", supported_domains=["feeds"], contract_version="v1")
|
|
assert dv.multipliers["alpha"] == 1.0
|
|
assert pb.budget == 1.0
|
|
assert al.signer == "auditor"
|
|
assert tr.end_ts > tr.start_ts
|
|
assert reg.contract_version == "v1"
|