51 lines
1.3 KiB
Python
51 lines
1.3 KiB
Python
from __future__ import annotations
|
|
from dataclasses import dataclass, field
|
|
from typing import List, Dict, Optional
|
|
|
|
|
|
@dataclass
|
|
class Asset:
|
|
symbol: str
|
|
asset_type: str # e.g., 'equity' or 'option'
|
|
|
|
def __post_init__(self):
|
|
if not self.symbol:
|
|
raise ValueError("Asset.symbol must be non-empty")
|
|
if self.asset_type not in {"equity", "option", "future"}:
|
|
raise ValueError("asset_type must be 'equity', 'option', or 'future'")
|
|
|
|
|
|
@dataclass
|
|
class MarketSignal:
|
|
asset: Asset
|
|
price: float
|
|
timestamp: float
|
|
|
|
|
|
@dataclass
|
|
class StrategyDelta:
|
|
# A high-level delta that would be applied to a local problem
|
|
asset: Asset
|
|
delta: float
|
|
theta: Optional[float] = None # optional auxiliary metric
|
|
|
|
|
|
@dataclass
|
|
class SharedSignals:
|
|
prices: Dict[str, float] # symbol -> price
|
|
greeks: Dict[str, Dict[str, float]] # symbol -> greek-name -> value
|
|
liquidity: Dict[str, float] # symbol -> liquidity proxy
|
|
|
|
|
|
@dataclass
|
|
class DualVariables:
|
|
coupling_multipliers: Dict[str, float] # venue_id -> multiplier
|
|
|
|
|
|
@dataclass
|
|
class PlanDelta:
|
|
# A delta plan that can be applied to move a local problem toward feasibility
|
|
actions: List[StrategyDelta] = field(default_factory=list)
|
|
timestamp: float = 0.0
|
|
safety_tags: List[str] = field(default_factory=list)
|