ParallelLLC/algorithmic_trading

1import logging2import random3import threading4import time5from typing import Any, Callable, Dict, List, Optional6 7import pandas as pd8 9logger = logging.getLogger(__name__)10 11_INTERVAL_MAP = {12    '1min': '1m',13    '1m': '1m',14    '5min': '5m',15    '5m': '5m',16    '15min': '15m',17    '15m': '15m',18    '30min': '30m',19    '30m': '30m',20    '1H': '1h',21    '1h': '1h',22    '60min': '1h',23    '1D': '1d',24    '1d': '1d',25    '1day': '1d',26}27 28# Yahoo lookback limits by interval. Requesting more returns empty or errors.29_MAX_LOOKBACK = {30    '1m': pd.Timedelta(days=7),31    '2m': pd.Timedelta(days=60),32    '5m': pd.Timedelta(days=60),33    '15m': pd.Timedelta(days=60),34    '30m': pd.Timedelta(days=60),35    '60m': pd.Timedelta(days=730),36    '90m': pd.Timedelta(days=60),37    '1h': pd.Timedelta(days=730),38    '1d': None,39    '5d': None,40    '1wk': None,41    '1mo': None,42    '3mo': None,43}44 45# How long each bar covers. Used to tell a finished bar from the one still46# forming right now -- see _drop_incomplete.47_INTERVAL_DURATION = {48    '1m': pd.Timedelta(minutes=1),49    '2m': pd.Timedelta(minutes=2),50    '5m': pd.Timedelta(minutes=5),51    '15m': pd.Timedelta(minutes=15),52    '30m': pd.Timedelta(minutes=30),53    '60m': pd.Timedelta(hours=1),54    '90m': pd.Timedelta(minutes=90),55    '1h': pd.Timedelta(hours=1),56    '1d': pd.Timedelta(days=1),57    '5d': pd.Timedelta(days=5),58    '1wk': pd.Timedelta(weeks=1),59}60 61# A daily-or-slower bar that moves more than this is almost always an62# unadjusted split rather than a real move (NVDA's 2024 10:1 shows up as -90%).63_SPLIT_SUSPECT_MOVE = 0.3564 65 66class YahooDataStream:67    """68    Market data from Yahoo Finance via yfinance.69 70    Yahoo has no public equities WebSocket. This polls OHLCV bars.71    Quotes are typically delayed (~15 minutes for US equities).72    Unofficial API: rate limits and schema changes are expected failure modes.73    """74 75    def __init__(self, config: Dict[str, Any]):76        self.config = config77        trading = config.get('trading', {})78        if trading.get('symbols'):79            self.symbols = list(trading['symbols'])80        elif trading.get('symbol'):81            self.symbols = [trading['symbol']]82        else:83            self.symbols = ['AAPL']84        yahoo_cfg = config.get('yahoo', {})85        self.poll_interval = int(yahoo_cfg.get('poll_interval_seconds', 60))86        # Adjusted by default. With auto_adjust off, Yahoo returns raw Close and87        # every split reads as a crash: NVDA's June 2024 10:1 becomes a -90% bar.88        self.auto_adjust = bool(yahoo_cfg.get('auto_adjust', True))89        self.emit_incomplete_bars = bool(yahoo_cfg.get('emit_incomplete_bars', False))90        self.max_backoff = int(yahoo_cfg.get('max_backoff_seconds', 900))91        self.interval = self._map_interval(config.get('trading', {}).get('timeframe', '1d'))92        self._consecutive_failures = 093        self.data_callbacks: List[Callable] = []94        self.is_connected = False95        self.data_buffer: Dict[str, Dict[str, Any]] = {}96        self._stop_event = threading.Event()97        self._poll_thread: Optional[threading.Thread] = None98        self._last_bar_ts: Dict[str, pd.Timestamp] = {}99        self._buffer_size = int(config.get('realtime_data', {}).get('buffer_size', 100))100 101        for symbol in self.symbols:102            self.data_buffer[symbol] = {103                'trades': [],104                'quotes': [],105                'bars': [],106                'latest_bar': None,107            }108 109        if not self.auto_adjust:110            logger.warning(111                "yahoo.auto_adjust is false: prices are NOT split- or dividend-adjusted. "112                "Every split will appear as a large single-bar loss and any backtest "113                "spanning one will be wrong."114            )115 116        logger.info(117            "Initialized YahooDataStream symbols=%s interval=%s poll_interval=%ss "118            "auto_adjust=%s emit_incomplete_bars=%s",119            self.symbols,120            self.interval,121            self.poll_interval,122            self.auto_adjust,123            self.emit_incomplete_bars,124        )125 126    @staticmethod127    def _map_interval(timeframe: str) -> str:128        mapped = _INTERVAL_MAP.get(str(timeframe), None)129        if mapped is None:130            logger.warning("Unknown timeframe %s, defaulting to 1d", timeframe)131            return '1d'132        return mapped133 134    def connect(self) -> None:135        """Start polling Yahoo for new bars."""136        if self.is_connected:137            logger.info("Yahoo data stream already connected")138            return139 140        self._stop_event.clear()141        # Seed the backoff from the first attempt: if we are already being142        # throttled, the loop should start backed off rather than hammering.143        self._consecutive_failures = 0 if self._poll_once() else 1144        self._poll_thread = threading.Thread(target=self._poll_loop, name='yahoo-poll', daemon=True)145        self._poll_thread.start()146        self.is_connected = True147        logger.info("Yahoo data stream polling started")148 149    def disconnect(self) -> None:150        self._stop_event.set()151        self.is_connected = False152        if self._poll_thread and self._poll_thread.is_alive():153            self._poll_thread.join(timeout=min(5, self.poll_interval + 1))154        logger.info("Disconnected from Yahoo data stream")155 156    def is_streaming(self) -> bool:157        return self.is_connected and self._poll_thread is not None and self._poll_thread.is_alive()158 159    def add_data_callback(self, callback: Callable) -> None:160        self.data_callbacks.append(callback)161 162    def get_latest_data(self, symbol: str) -> Dict[str, Any]:163        if symbol not in self.data_buffer:164            return {}165        buffer = self.data_buffer[symbol]166        return {167            'latest_trade': buffer['trades'][-1] if buffer['trades'] else None,168            'latest_quote': buffer['quotes'][-1] if buffer['quotes'] else None,169            'latest_bar': buffer['latest_bar'],170            'recent_trades': buffer['trades'][-10:] if buffer['trades'] else [],171            'recent_quotes': buffer['quotes'][-10:] if buffer['quotes'] else [],172        }173 174    def get_historical_data(self, symbol: str, start_date: str, end_date: str) -> pd.DataFrame:175        start, end = self._clamp_window(start_date, end_date, self.interval)176        try:177            raw = self._download(symbol, start=start, end=end, interval=self.interval)178            df = self._drop_incomplete(self._normalize_ohlcv(raw))179            self._warn_if_unadjusted(symbol, df)180            if df.empty:181                logger.warning("No Yahoo historical data for %s between %s and %s", symbol, start, end)182            else:183                logger.info("Loaded %s Yahoo bars for %s (%s to %s)", len(df), symbol, start, end)184            return df185        except Exception as e:186            logger.error("Error fetching Yahoo historical data for %s: %s", symbol, e, exc_info=True)187            return pd.DataFrame()188 189    def test_connection(self) -> bool:190        try:191            symbol = self.symbols[0] if self.symbols else 'AAPL'192            df = self._download(symbol, period='5d', interval='1d')193            if df is None or df.empty:194                logger.warning("Yahoo connection test returned no data for %s", symbol)195                return False196            logger.info("Yahoo connection test succeeded for %s (%s rows)", symbol, len(df))197            return True198        except Exception as e:199            logger.error("Yahoo connection test failed: %s", e)200            return False201 202    def get_connection_status(self) -> Dict[str, Any]:203        return {204            'is_connected': self.is_connected,205            'provider': 'yahoo',206            'interval': self.interval,207            'poll_interval_seconds': self.poll_interval,208            'symbols': self.symbols,209            'data_buffers': {210                symbol: len(buffer['bars']) for symbol, buffer in self.data_buffer.items()211            },212        }213 214    def generate_simulated_data(self, symbol: str) -> Dict[str, Any]:215        latest_data = self.get_latest_data(symbol)216        base_price = 150.0217        if latest_data.get('latest_bar'):218            base_price = latest_data['latest_bar']['close']219        elif latest_data.get('latest_trade'):220            base_price = latest_data['latest_trade']['price']221 222        price_change = random.uniform(-0.01, 0.01) * base_price223        new_price = base_price + price_change224        simulated_bar = {225            'symbol': symbol,226            'open': base_price,227            'high': max(base_price, new_price),228            'low': min(base_price, new_price),229            'close': new_price,230            'volume': random.randint(100, 1000),231            'timestamp': int(time.time() * 1_000_000),232        }233        self._store_bar(symbol, simulated_bar, emit=False)234        return simulated_bar235 236    def _poll_loop(self) -> None:237        delay = self.poll_interval238        while not self._stop_event.wait(delay):239            try:240                succeeded = self._poll_once()241            except Exception as e:242                logger.error("Yahoo poll loop error: %s", e, exc_info=True)243                succeeded = False244            self._consecutive_failures = 0 if succeeded else self._consecutive_failures + 1245            delay = self._next_delay()246 247    def _next_delay(self) -> float:248        """Poll interval, backed off exponentially while Yahoo is refusing us.249 250        Yahoo rate-limits aggressively and an unofficial API gives no251        Retry-After, so a fixed interval just keeps you throttled. Jitter stops252        several symbols (or several deployments) resynchronising after an outage.253        """254        if self._consecutive_failures == 0:255            base = float(self.poll_interval)256        else:257            base = min(258                self.poll_interval * (2 ** self._consecutive_failures),259                float(self.max_backoff),260            )261            logger.warning(262                "Yahoo poll failed %s time(s) in a row; next attempt in ~%.0fs",263                self._consecutive_failures,264                base,265            )266        return max(1.0, base * random.uniform(0.8, 1.2))267 268    def _poll_once(self) -> bool:269        """Fetch and ingest one round of bars. Returns True if any symbol succeeded."""270        any_success = False271        for symbol in self.symbols:272            try:273                raw = self._download(symbol, period='5d', interval=self.interval)274                df = self._normalize_ohlcv(raw)275                if df.empty:276                    logger.warning("Yahoo poll returned no bars for %s", symbol)277                    continue278                self._ingest_new_bars(symbol, df)279                any_success = True280            except Exception as e:281                logger.error("Yahoo poll failed for %s: %s", symbol, e)282        return any_success283 284    def _warn_if_unadjusted(self, symbol: str, df: pd.DataFrame) -> int:285        """Flag single-bar moves that look like unadjusted corporate actions.286 287        This is a backstop rather than the fix -- the fix is auto_adjust. But a288        split slipping through silently corrupts every downstream number, so it289        is worth naming the dates rather than letting a strategy trade them.290        Returns the number of suspicious bars found.291        """292        duration = _INTERVAL_DURATION.get(self.interval)293        if df.empty or len(df) < 2 or duration is None or duration < pd.Timedelta(days=1):294            return 0295        moves = df['close'].pct_change()296        suspects = df.loc[moves.abs() > _SPLIT_SUSPECT_MOVE, 'timestamp']297        if len(suspects):298            dates = ', '.join(str(pd.Timestamp(t).date()) for t in suspects.head(5))299            logger.warning(300                "%s has %s bar(s) moving more than %.0f%% (%s). On a liquid name that is "301                "usually an unadjusted split, not a real move — check yahoo.auto_adjust.",302                symbol,303                len(suspects),304                _SPLIT_SUSPECT_MOVE * 100,305                dates,306            )307        return int(len(suspects))308 309    def _drop_incomplete(self, df: pd.DataFrame) -> pd.DataFrame:310        """Remove the bar that is still forming.311 312        Yahoo returns the in-progress period as an ordinary row. Emitting it313        would hand the strategy a close that has not happened yet, and because314        the watermark advances past it, the finished version never arrives.315        """316        if self.emit_incomplete_bars or df.empty:317            return df318        duration = _INTERVAL_DURATION.get(self.interval)319        if duration is None:320            return df321        now = pd.Timestamp.now(tz='UTC').tz_convert(None)322        complete = df[df['timestamp'] + duration <= now]323        dropped = len(df) - len(complete)324        if dropped:325            logger.debug("Dropped %s in-progress %s bar(s)", dropped, self.interval)326        return complete327 328    def _ingest_new_bars(self, symbol: str, df: pd.DataFrame) -> None:329        rows = self._drop_incomplete(df)330        last_ts = self._last_bar_ts.get(symbol)331        if last_ts is not None:332            rows = rows[rows['timestamp'] > last_ts]333        if rows.empty:334            return335 336        for _, row in rows.iterrows():337            ts = pd.Timestamp(row['timestamp'])338            bar = {339                'symbol': symbol,340                'open': float(row['open']),341                'high': float(row['high']),342                'low': float(row['low']),343                'close': float(row['close']),344                'volume': float(row['volume']),345                'timestamp': int(ts.timestamp() * 1_000_000),346            }347            self._store_bar(symbol, bar, emit=True)348            self._last_bar_ts[symbol] = ts349 350    def _store_bar(self, symbol: str, bar: Dict[str, Any], emit: bool) -> None:351        buffer = self.data_buffer[symbol]352        buffer['bars'].append(bar)353        buffer['latest_bar'] = bar354        if len(buffer['bars']) > self._buffer_size:355            buffer['bars'] = buffer['bars'][-self._buffer_size:]356        if emit:357            self._notify_callbacks('bar', bar)358 359    def _notify_callbacks(self, data_type: str, data: Dict[str, Any]) -> None:360        for callback in self.data_callbacks:361            try:362                callback(data_type, data)363            except Exception as e:364                logger.error("Error in data callback: %s", e)365 366    def _clamp_window(self, start_date: str, end_date: str, interval: str) -> tuple:367        start = pd.to_datetime(start_date, utc=True).tz_convert(None)368        end = pd.to_datetime(end_date, utc=True).tz_convert(None)369        max_lookback = _MAX_LOOKBACK.get(interval)370        if max_lookback is not None:371            earliest = pd.Timestamp.now(tz='UTC').tz_convert(None) - max_lookback372            if start < earliest:373                logger.warning(374                    "Yahoo %s bars only cover ~%s; clamping start from %s to %s",375                    interval,376                    max_lookback,377                    start.date(),378                    earliest.date(),379                )380                start = earliest381        if end < start:382            end = start + pd.Timedelta(days=1)383        return start.strftime('%Y-%m-%d'), end.strftime('%Y-%m-%d')384 385    def _download(386        self,387        symbol: str,388        start: Optional[str] = None,389        end: Optional[str] = None,390        period: Optional[str] = None,391        interval: str = '1d',392    ) -> pd.DataFrame:393        import yfinance as yf394 395        kwargs: Dict[str, Any] = {396            'tickers': symbol,397            'interval': interval,398            'auto_adjust': self.auto_adjust,399            'progress': False,400            'threads': False,401        }402        if period:403            kwargs['period'] = period404        else:405            kwargs['start'] = start406            kwargs['end'] = end407        return yf.download(**kwargs)408 409    @staticmethod410    def _normalize_ohlcv(df: pd.DataFrame) -> pd.DataFrame:411        if df is None or df.empty:412            return pd.DataFrame(columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])413 414        out = df.copy()415        if isinstance(out.columns, pd.MultiIndex):416            out.columns = [str(col[0]).strip().lower() for col in out.columns]417        else:418            out.columns = [str(c).strip().lower() for c in out.columns]419        rename = {}420        if 'datetime' in out.columns:421            rename['datetime'] = 'timestamp'422        out = out.rename(columns=rename)423 424        if 'timestamp' not in out.columns:425            out = out.reset_index()426            time_col = out.columns[0]427            out = out.rename(columns={time_col: 'timestamp'})428 429        out['timestamp'] = pd.to_datetime(out['timestamp'], utc=True).dt.tz_localize(None)430 431        required = ['timestamp', 'open', 'high', 'low', 'close', 'volume']432        missing = [c for c in required if c not in out.columns]433        if missing:434            logger.error("Yahoo response missing columns: %s", missing)435            return pd.DataFrame(columns=required)436 437        out = out[required].dropna()438        out = out.sort_values('timestamp').drop_duplicates(subset=['timestamp'])439        return out.reset_index(drop=True)440