75 lines
2.4 KiB
Python
75 lines
2.4 KiB
Python
import json
|
|
try:
|
|
from mercurymesh_federated_reproducible_marke.core import (
|
|
MarketStateSnapshot,
|
|
SharedSignals,
|
|
PlanDelta,
|
|
AuditLog,
|
|
FederatedCoordinator,
|
|
)
|
|
except ModuleNotFoundError:
|
|
# Fallback: load module directly from file path if package import fails in the environment
|
|
import importlib.util
|
|
import pathlib
|
|
core_path = pathlib.Path(__file__).resolve().parents[1] / "mercurymesh_federated_reproducible_marke" / "core.py"
|
|
spec = importlib.util.spec_from_file_location("mm_core", str(core_path))
|
|
if spec is None or spec.loader is None:
|
|
raise ImportError("Could not load core module from path")
|
|
mm_core = importlib.util.module_from_spec(spec)
|
|
# Ensure the module is discoverable via sys.modules for dataclass decoration
|
|
import sys
|
|
sys.modules[spec.name] = mm_core # type: ignore
|
|
spec.loader.exec_module(mm_core) # type: ignore
|
|
MarketStateSnapshot = mm_core.MarketStateSnapshot
|
|
SharedSignals = mm_core.SharedSignals
|
|
PlanDelta = mm_core.PlanDelta
|
|
AuditLog = mm_core.AuditLog
|
|
FederatedCoordinator = mm_core.FederatedCoordinator
|
|
|
|
|
|
def test_market_state_serialization_roundtrip():
|
|
m = MarketStateSnapshot(
|
|
assets=["AAPL", "MSFT"],
|
|
bids={"AAPL": 150.0, "MSFT": 300.0},
|
|
offers={"AAPL": 151.0, "MSFT": 301.0},
|
|
last_trade_times={"AAPL": "2026-01-01T00:00:00Z", "MSFT": "2026-01-01T00:00:01Z"},
|
|
liquidity_metrics={"AAPL": 0.8, "MSFT": 0.9},
|
|
timestamp="2026-01-01T00:00:00Z",
|
|
version=1,
|
|
)
|
|
d = m.to_dict()
|
|
s = json.dumps(d)
|
|
json.loads(s)
|
|
|
|
|
|
def test_shared_signals_roundtrip():
|
|
sig = SharedSignals(
|
|
metric="liquidity",
|
|
value=0.75,
|
|
distribution=[0.1, 0.3, 0.6],
|
|
confidence=0.95,
|
|
version=1,
|
|
source="simulated",
|
|
)
|
|
assert sig.to_json() # basic serialization
|
|
|
|
|
|
def test_audit_log_roundtrip():
|
|
log = AuditLog(
|
|
entry="delta_applied",
|
|
signer="tester",
|
|
timestamp="2026-01-01T00:00:00Z",
|
|
contract_id="root",
|
|
version=1,
|
|
)
|
|
assert log.to_dict()["entry"] == "delta_applied"
|
|
assert log.to_json()
|
|
|
|
|
|
def test_federated_coordinator_aggregation():
|
|
s1 = {"A": 1.0, "B": -0.5}
|
|
s2 = {"A": 3.0, "B": 0.0}
|
|
plan = FederatedCoordinator.aggregate_signals([s1, s2])
|
|
assert abs(plan.delta["A"] - 2.0) < 1e-6
|
|
assert abs(plan.delta["B"] - (-0.25)) < 1e-6
|