19 lines
651 B
Python
19 lines
651 B
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Optional
|
|
|
|
from ..schema import SignalNode
|
|
import time
|
|
|
|
|
|
@dataclass
|
|
class PriceFeedAdapter:
|
|
asset: str
|
|
venue: str
|
|
def generate_signal(self, price: float, timestamp: Optional[int] = None) -> SignalNode:
|
|
ts = timestamp if timestamp is not None else int(time.time())
|
|
# Simple heuristic: price above threshold yields a positive signal
|
|
signal_type = "price_above_threshold" if price > 100 else "price_below_threshold"
|
|
return SignalNode(asset=self.asset, venue=self.venue, signal_type=signal_type, timestamp=ts, quality=1.0)
|