deltaforge-real-time-cross-.../deltaforge_mvp/adapters/options_feed.py

21 lines
953 B
Python

from __future__ import annotations
import time
from deltaforge_mvp.core import Asset, MarketSignal
def generate_signal(underlying: str, strike: float, expiry: str, price: float, timestamp: float | None = None) -> MarketSignal:
"""Tiny options feed adapter: returns a MarketSignal for an option asset."""
a = Asset(type="option", underlying=underlying, strike=strike, expires=expiry, symbol=None)
ts = timestamp if timestamp is not None else time.time()
return MarketSignal(asset=a, price=price, volatility=0.25, liquidity=1.0, timestamp=ts)
class OptionsFeedAdapter:
"""Lightweight adapter wrapper to expose a stable interface used by tests."""
def __init__(self, name: str = "options"):
self.name = name
def get_signal(self, underlying: str, strike: float, expiry: str, price: float, timestamp: float | None = None) -> MarketSignal:
return generate_signal(underlying, strike, expiry, price, timestamp)