48 lines
1.8 KiB
Python
48 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
"""Toy adapters to bootstrap interoperability between venues and the core engine.
|
|
|
|
- PriceFeedAdapter: simulates a price feed (e.g., equities, spot, etc.).
|
|
- OptionsVenueAdapter: simulates an options venue with basic option data.
|
|
|
|
These are deliberately simple and deterministic to serve as MVP building blocks.
|
|
"""
|
|
|
|
from typing import Dict, List
|
|
from .dsl import Asset, MarketSignal
|
|
|
|
|
|
class PriceFeedAdapter:
|
|
def __init__(self, assets: List[Asset]):
|
|
self.assets = assets
|
|
# deterministic seed values per asset
|
|
self.prices: Dict[str, float] = {a.symbol: 100.0 for a in assets}
|
|
|
|
def update(self) -> List[MarketSignal]:
|
|
# simple random walk with fixed seed for deterministic tests
|
|
signals: List[MarketSignal] = []
|
|
for a in self.assets:
|
|
price = self.prices[a.symbol] * 1.0005 # tiny drift
|
|
self.prices[a.symbol] = price
|
|
signals.append(MarketSignal(asset=a, signal_type="price", value=price))
|
|
return signals
|
|
|
|
|
|
class OptionsVenueAdapter:
|
|
def __init__(self, underlying: Asset, strikes: List[float], maturities: List[float]):
|
|
self.underlying = underlying
|
|
self.strikes = strikes
|
|
self.maturities = maturities
|
|
|
|
def quote(self) -> List[MarketSignal]:
|
|
# toy: simple intrinsic value proxy for options across strikes
|
|
signals: List[MarketSignal] = []
|
|
for k in self.strikes:
|
|
delta = max(0.0, self.underlying_asset_spot() - k)
|
|
signals.append(MarketSignal(asset=self.underlying, signal_type=f"option_delta_{k}", value=delta))
|
|
return signals
|
|
|
|
def underlying_asset_spot(self) -> float:
|
|
# deterministic spot via a fixed function of symbol hash
|
|
return float(sum(ord(c) for c in self.underlying.symbol)) % 200 + 50
|