59 lines
1.8 KiB
Python
59 lines
1.8 KiB
Python
from __future__ import annotations
|
|
from dataclasses import dataclass, field
|
|
from typing import List, Optional, Any
|
|
|
|
def _validate_non_empty(name: str, value: Any) -> None:
|
|
if value is None or (isinstance(value, (str, list, dict)) and len(value) == 0):
|
|
raise ValueError(f"{name} must be non-empty")
|
|
|
|
@dataclass
|
|
class Asset:
|
|
id: str
|
|
type: str # "equity" | "option" | "future"
|
|
symbol: str
|
|
|
|
def __post_init__(self):
|
|
_validate_non_empty("Asset.id", self.id)
|
|
_validate_non_empty("Asset.type", self.type)
|
|
_validate_non_empty("Asset.symbol", self.symbol)
|
|
|
|
@dataclass
|
|
class MarketSignal:
|
|
asset: Asset
|
|
timestamp: float
|
|
price: float
|
|
liquidity: float = 1.0
|
|
extra: dict = field(default_factory=dict)
|
|
|
|
def __post_init__(self):
|
|
_validate_non_empty("MarketSignal.asset", self.asset)
|
|
if self.timestamp is None or self.price is None:
|
|
raise ValueError("MarketSignal timestamp/price required")
|
|
|
|
@dataclass
|
|
class PlanDelta:
|
|
delta: List[dict] # A list of actions, simplified
|
|
timestamp: float
|
|
author: str
|
|
contract_id: Optional[str] = None
|
|
privacy_budget: Optional[dict] = None
|
|
|
|
def __post_init__(self):
|
|
_validate_non_empty("PlanDelta.delta", self.delta)
|
|
if self.timestamp is None:
|
|
raise ValueError("PlanDelta.timestamp required")
|
|
|
|
@dataclass
|
|
class StrategyDelta:
|
|
id: str
|
|
assets: List[Asset]
|
|
objectives: dict
|
|
constraints: dict = field(default_factory=dict)
|
|
version: int = 1
|
|
|
|
def __post_init__(self):
|
|
_validate_non_empty("StrategyDelta.id", self.id)
|
|
_validate_non_empty("StrategyDelta.assets", self.assets)
|
|
if not isinstance(self.assets, list) or len(self.assets) == 0:
|
|
raise ValueError("StrategyDelta.assets must be a non-empty list")
|