63 lines
2.3 KiB
Python
63 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import List, Dict, Optional
|
|
from datetime import datetime
|
|
|
|
|
|
class Asset:
|
|
"""Canonical Asset representation used by DeltaForge DSL.
|
|
|
|
Supports two constructor styles for compatibility:
|
|
- Modern: Asset(id=..., type=..., symbol=...)
|
|
- Legacy: Asset(symbol=..., asset_type=..., venue=...)
|
|
"""
|
|
def __init__(self, id: str | None = None, type: str | None = None, symbol: str | None = None, venue: str | None = None, **kwargs):
|
|
# Preferred explicit constructor
|
|
if id is not None and type is not None and symbol is not None:
|
|
self.id = id
|
|
self.type = type
|
|
self.symbol = symbol
|
|
else:
|
|
# Legacy constructor path: expect symbol and asset_type (and optionally venue)
|
|
self.symbol = kwargs.get("symbol")
|
|
self.type = kwargs.get("asset_type")
|
|
self.id = kwargs.get("id")
|
|
# If legacy path didn't supply id, derive a simple one if possible
|
|
if self.id is None and self.symbol is not None and self.type is not None:
|
|
self.id = f"{self.type}-{self.symbol}"
|
|
if self.symbol is None or self.type is None:
|
|
raise ValueError("Asset requires either (id, type, symbol) or (symbol, asset_type)")
|
|
|
|
# Validate type
|
|
if self.type not in {"equity", "option", "futures"}:
|
|
raise ValueError("type must be one of 'equity', 'option', or 'futures'")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MarketSignal:
|
|
asset: Asset
|
|
price: float
|
|
timestamp: float = field(default_factory=lambda: 0.0)
|
|
delta: Optional[float] = None # local delta proxy if available
|
|
meta: Dict[str, float] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass
|
|
class StrategyDelta:
|
|
id: str
|
|
assets: List[Asset]
|
|
objectives: Dict
|
|
timestamp: float = field(default_factory=lambda: 0.0)
|
|
# kept minimal; can extend with per-block budgets if needed
|
|
|
|
|
|
@dataclass
|
|
class PlanDelta:
|
|
delta: List = field(default_factory=list) # list of hedge/trade actions (dicts)
|
|
actions: List[Dict] = field(default_factory=list) # provenance and trade actions
|
|
total_cost: float = 0.0
|
|
timestamp: float = field(default_factory=lambda: 0.0)
|
|
author: Optional[str] = None
|
|
contract_id: Optional[str] = None
|