40 lines
1.6 KiB
Python
40 lines
1.6 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import List
|
|
from .dsl import Asset, MarketSignal, PlanDelta, StrategyDelta
|
|
|
|
|
|
class Curator:
|
|
"""Simple central coordinator: enforces cross-venue delta neutrality
|
|
with a naive ADMM-lite style constraint. This is a minimal stub for MVP.
|
|
"""
|
|
|
|
def __init__(self, assets: List[Asset] = None):
|
|
self.assets = assets or []
|
|
|
|
def synthesize_plan(self, signals: List[MarketSignal], objectives: List[StrategyDelta]) -> PlanDelta:
|
|
"""Generate a PlanDelta given per-venue signals and desired strategy blocks.
|
|
|
|
The simplest cross-venue constraint: enforce sum of target_delta across blocks is ~0.
|
|
This is intentionally minimal for MVP demonstration.
|
|
"""
|
|
# naive aggregation: align target deltas to neutralize total delta
|
|
total_target = sum([sd.target_delta for sd in objectives]) if objectives else 0.0
|
|
# If not neutral, scale down deltas proportionally to achieve neutrality if possible
|
|
if total_target != 0:
|
|
factor = -total_target / max(1e-6, len(objectives))
|
|
adjusted = [StrategyDelta(
|
|
asset=sd.asset,
|
|
target_delta=sd.target_delta + factor,
|
|
risk_budget=sd.risk_budget,
|
|
objective=sd.objective,
|
|
timestamp=sd.timestamp,
|
|
) for sd in objectives]
|
|
else:
|
|
adjusted = objectives
|
|
|
|
plan = PlanDelta()
|
|
plan.hedges = adjusted
|
|
plan.total_cost = sum(abs(sd.target_delta) * 0.1 for sd in adjusted) # simplistic cost proxy
|
|
return plan
|