Upload realtime_data.py

realtime_data.py

@@ -0,0 +1,253 @@
+"""Real-Time Data Streaming - WebSocket feeds and live processing"""
+import numpy as np
+import pandas as pd
+import threading
+import time
+import json
+from typing import Dict, List, Optional, Callable
+from collections import deque
+from datetime import datetime, timedelta
+import warnings
+warnings.filterwarnings('ignore')
+
+
+class LiveDataBuffer:
+    """Thread-safe circular buffer for real-time market data"""
+    
+    def __init__(self, max_len=10000):
+        self.max_len = max_len
+        self._data = {}
+        self._lock = threading.Lock()
+    
+    def append(self, ticker, timestamp, open_p, high, low, close, volume, **kwargs):
+        with self._lock:
+            if ticker not in self._data:
+                self._data[ticker] = deque(maxlen=self.max_len)
+            self._data[ticker].append({
+                'timestamp': timestamp, 'Open': open_p, 'High': high,
+                'Low': low, 'Close': close, 'Volume': volume, **kwargs
+            })
+    
+    def get_dataframe(self, ticker, last_n=None):
+        with self._lock:
+            if ticker not in self._data:
+                return pd.DataFrame()
+            data = list(self._data[ticker])
+            if last_n:
+                data = data[-last_n:]
+            return pd.DataFrame(data).set_index('timestamp')
+    
+    def latest(self, ticker):
+        with self._lock:
+            if ticker not in self._data or len(self._data[ticker]) == 0:
+                return None
+            return self._data[ticker][-1]
+
+
+class AlpacaStreamer:
+    """Alpaca Markets WebSocket - free real-time IEX data"""
+    
+    def __init__(self, api_key, secret_key, buffer=None, paper=True):
+        self.api_key = api_key
+        self.secret_key = secret_key
+        self.paper = paper
+        self.buffer = buffer or LiveDataBuffer()
+        self._running = False
+        self._callbacks = []
+    
+    def subscribe(self, tickers):
+        self._running = True
+        url = "wss://paper-api.alpaca.markets/stream" if self.paper else "wss://data.alpaca.markets/stream"
+        
+        def on_message(ws, message):
+            data = json.loads(message)
+            if data.get('T') in ('b', 't'):
+                self.buffer.append(
+                    ticker=data['S'], timestamp=pd.Timestamp(data['t']),
+                    open_p=data.get('o', data.get('p', 0)),
+                    high=data.get('h', data.get('p', 0)),
+                    low=data.get('l', data.get('p', 0)),
+                    close=data.get('c', data.get('p', 0)),
+                    volume=data.get('v', data.get('s', 0)),
+                    vwap=data.get('vw', data.get('p', 0))
+                )
+                for cb in self._callbacks:
+                    cb(data)
+        
+        def on_open(ws):
+            ws.send(json.dumps({"action": "auth", "key": self.api_key, "secret": self.secret_key}))
+            time.sleep(1)
+            for t in tickers:
+                ws.send(json.dumps({"action": "subscribe", "bars": [t]}))
+        
+        import websocket
+        self._ws = websocket.WebSocketApp(url, on_open=on_open, on_message=on_message)
+        self._thread = threading.Thread(target=self._ws.run_forever, daemon=True)
+        self._thread.start()
+    
+    def on_data(self, callback):
+        self._callbacks.append(callback)
+    
+    def unsubscribe(self):
+        self._running = False
+        if hasattr(self, '_ws'):
+            self._ws.close()
+
+
+class NewsStreamAggregator:
+    """Aggregate news from Yahoo Finance + FMP + custom RSS"""
+    
+    def __init__(self, tickers, sentiment_model=None):
+        self.tickers = tickers
+        self.sentiment_model = sentiment_model
+        self.news_buffer = deque(maxlen=5000)
+        self._running = False
+    
+    def fetch_yahoo_news(self, ticker):
+        import yfinance as yf
+        try:
+            tk = yf.Ticker(ticker)
+            news = tk.news or []
+            articles = []
+            for item in news:
+                articles.append({
+                    'ticker': ticker,
+                    'timestamp': pd.Timestamp(item.get('publish_time', 0), unit='s') if 'publish_time' in item else pd.Timestamp.now(),
+                    'title': item.get('title', ''),
+                    'source': item.get('publisher', 'yahoo'),
+                    'text': item.get('title', ''),
+                })
+            return articles
+        except Exception as e:
+            return []
+    
+    def fetch_fmp_news(self, api_key='demo'):
+        import requests
+        try:
+            url = "https://financialmodelingprep.com/api/v3/stock_news"
+            resp = requests.get(url, params={'tickers': ','.join(self.tickers), 'limit': 100, 'apikey': api_key}, timeout=10)
+            if resp.status_code == 200:
+                articles = []
+                for item in resp.json():
+                    articles.append({
+                        'ticker': item.get('symbol', ''),
+                        'timestamp': pd.Timestamp(item.get('publishedDate', '')),
+                        'title': item.get('title', ''),
+                        'source': 'fmp',
+                        'text': item.get('text', item.get('title', '')),
+                    })
+                return articles
+        except:
+            pass
+        return []
+    
+    def process_sentiment(self, articles):
+        df = pd.DataFrame(articles)
+        if len(df) == 0 or self.sentiment_model is None:
+            if len(df) > 0:
+                df['sentiment_score'] = 0.0
+            return df
+        texts = (df['title'].fillna('') + ' ' + df.get('text', pd.Series(['']*len(df))).fillna('')).tolist()
+        sentiments = self.sentiment_model.analyze_batch(texts)
+        df['sentiment_score'] = [s['sentiment_score'] for s in sentiments]
+        df['sentiment_label'] = [s['label'] for s in sentiments]
+        return df
+    
+    def start_streaming(self, poll_seconds=300):
+        self._running = True
+        def poll():
+            while self._running:
+                all_articles = []
+                for t in self.tickers:
+                    all_articles.extend(self.fetch_yahoo_news(t))
+                all_articles.extend(self.fetch_fmp_news())
+                if all_articles:
+                    df = self.process_sentiment(all_articles)
+                    for _, row in df.iterrows():
+                        self.news_buffer.append(row.to_dict())
+                time.sleep(poll_seconds)
+        self._thread = threading.Thread(target=poll, daemon=True)
+        self._thread.start()
+    
+    def stop(self):
+        self._running = False
+    
+    def get_latest_sentiment(self, ticker, hours=24):
+        if len(self.news_buffer) == 0:
+            return pd.DataFrame()
+        df = pd.DataFrame(list(self.news_buffer))
+        if 'ticker' not in df.columns:
+            return df
+        cutoff = pd.Timestamp.now() - timedelta(hours=hours)
+        return df[(df['ticker'] == ticker) & (df['timestamp'] > cutoff)]
+
+
+class OrderFlowEstimator:
+    """Estimate order flow imbalance from tick data"""
+    
+    def __init__(self, lookback=1000):
+        self.lookback = lookback
+        self._trades = {}
+    
+    def process_trade(self, ticker, price, volume, timestamp=None, side='auto'):
+        if ticker not in self._trades:
+            self._trades[ticker] = deque(maxlen=self.lookback)
+        if side == 'auto' and len(self._trades[ticker]) > 0:
+            last_price = self._trades[ticker][-1]['price']
+            side = 'buy' if price > last_price else ('sell' if price < last_price else 'neutral')
+        self._trades[ticker].append({'price': price, 'volume': volume, 'timestamp': timestamp or pd.Timestamp.now(), 'side': side})
+    
+    def get_imbalance(self, ticker, window=100):
+        if ticker not in self._trades:
+            return {'ofi': 0, 'buy_volume': 0, 'sell_volume': 0}
+        df = pd.DataFrame(list(self._trades[ticker])[-window:])
+        buy_vol = df[df['side'] == 'buy']['volume'].sum()
+        sell_vol = df[df['side'] == 'sell']['volume'].sum()
+        total = buy_vol + sell_vol
+        ofi = (buy_vol - sell_vol) / total if total > 0 else 0
+        return {'ofi': ofi, 'buy_volume': int(buy_vol), 'sell_volume': int(sell_vol)}
+
+
+class RealtimeFeatureEngine:
+    """Real-time feature computation pipeline"""
+    
+    def __init__(self, tickers, data_source='yahoo', api_key='', secret_key='',
+                 include_sentiment=True, sentiment_model=None):
+        self.tickers = tickers
+        self.buffer = LiveDataBuffer(max_len=50000)
+        self.order_flow = OrderFlowEstimator()
+        self.news = NewsStreamAggregator(tickers, sentiment_model)
+        self._source = data_source
+        self._api_key = api_key
+        self._secret_key = secret_key
+    
+    def start(self, interval='1m', poll_seconds=60):
+        self.news.start_streaming()
+        if self._source == 'alpaca':
+            self.streamer = AlpacaStreamer(self._api_key, self._secret_key, self.buffer)
+            self.streamer.subscribe(self.tickers)
+        else:
+            import yfinance as yf
+            def poll():
+                while True:
+                    for t in self.tickers:
+                        try:
+                            hist = yf.Ticker(t).history(period='1d', interval=interval)
+                            for idx, row in hist.iterrows():
+                                self.buffer.append(t, idx, row['Open'], row['High'], row['Low'], row['Close'], row['Volume'])
+                        except:
+                            pass
+                    time.sleep(poll_seconds)
+            self._thread = threading.Thread(target=poll, daemon=True)
+            self._thread.start()
+    
+    def stop(self):
+        if hasattr(self, 'streamer') and hasattr(self.streamer, 'unsubscribe'):
+            self.streamer.unsubscribe()
+        self.news.stop()
+    
+    def get_latest(self, ticker, lookback=60):
+        return self.buffer.get_dataframe(ticker, last_n=lookback)
+    
+    def get_sentiment(self, ticker):
+        return self.news.get_latest_sentiment(ticker)