66 lines
1.4 KiB
Python
66 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import List, Dict, Optional
|
|
|
|
|
|
@dataclass(eq=True)
|
|
class Asset:
|
|
symbol: str
|
|
asset_type: str # e.g., 'equity', 'option', etc.
|
|
|
|
def __repr__(self) -> str:
|
|
return f"Asset(symbol={self.symbol!r}, asset_type={self.asset_type!r})"
|
|
|
|
|
|
@dataclass
|
|
class MarketSignal:
|
|
asset: Asset
|
|
signal_type: str
|
|
value: float
|
|
|
|
|
|
@dataclass
|
|
class RiskState:
|
|
asset: Asset
|
|
pnl_exposure: float
|
|
delta_exposure: float
|
|
gamma_exposure: float
|
|
volatility_shift: float
|
|
|
|
|
|
@dataclass
|
|
class HedgePlanDelta:
|
|
assets: List[Asset] = field(default_factory=list)
|
|
hedges: Dict[str, float] = field(default_factory=dict)
|
|
author: Optional[str] = None
|
|
|
|
def __post_init__(self):
|
|
# Simple normalization guard to avoid accidental None in assets
|
|
if self.assets is None:
|
|
self.assets = []
|
|
if self.hedges is None:
|
|
self.hedges = {}
|
|
|
|
|
|
class GraphOfContracts:
|
|
"""Tiny in-process registry for adapters and data contracts."""
|
|
|
|
def __init__(self) -> None:
|
|
self.registry: Dict[str, object] = {}
|
|
|
|
def register(self, name: str, contract: object) -> None:
|
|
self.registry[name] = contract
|
|
|
|
def get(self, name: str) -> Optional[object]:
|
|
return self.registry.get(name)
|
|
|
|
|
|
__all__ = [
|
|
"Asset",
|
|
"MarketSignal",
|
|
"RiskState",
|
|
"HedgePlanDelta",
|
|
"GraphOfContracts",
|
|
]
|