diff --git a/tests/test_drift_comparator.py b/tests/test_drift_comparator.py new file mode 100644 index 0000000..5ff2f59 --- /dev/null +++ b/tests/test_drift_comparator.py @@ -0,0 +1,108 @@ +"""Tests for DriftComparator pushdown logic.""" + +import time +from unittest.mock import patch + +from trading_system.drift_comparator import DriftComparator + + +class TestShouldFullFetch: + def test_cold_start_always_fetches(self): + comp = DriftComparator() + assert comp.should_full_fetch("BTC", 100.0) is True + + def test_no_fetch_when_price_flat(self): + comp = DriftComparator(price_drift_pct=0.005) + comp.update("BTC", 100.0, {"current_price": 100.0}, {}) + # 0.1% move — below 0.5% threshold + assert comp.should_full_fetch("BTC", 100.1) is False + + def test_fetch_when_price_drifts(self): + comp = DriftComparator(price_drift_pct=0.005) + comp.update("BTC", 100.0, {"current_price": 100.0}, {}) + # 1% move — above 0.5% threshold + assert comp.should_full_fetch("BTC", 101.0) is True + + def test_fetch_on_price_drop(self): + comp = DriftComparator(price_drift_pct=0.005) + comp.update("BTC", 100.0, {"current_price": 100.0}, {}) + assert comp.should_full_fetch("BTC", 99.0) is True + + def test_high_vol_tighter_threshold(self): + comp = DriftComparator(price_drift_pct=0.005, high_vol_drift_pct=0.002) + comp.update("BTC", 100.0, {"current_price": 100.0}, {}) + # 0.3% move — below default but above high_vol threshold + assert comp.should_full_fetch("BTC", 100.3, regime="MEDIUM_VOL") is False + assert comp.should_full_fetch("BTC", 100.3, regime="HIGH_VOL") is True + + def test_staleness_forces_fetch(self): + comp = DriftComparator(max_stale_minutes=0.01) # ~0.6 seconds + comp.update("BTC", 100.0, {"current_price": 100.0}, {}) + time.sleep(0.7) + # Price hasn't moved but data is stale + assert comp.should_full_fetch("BTC", 100.0) is True + + +class TestCachedMetrics: + def test_returns_none_for_unknown_symbol(self): + comp = DriftComparator() + assert comp.get_cached_metrics("BTC", 100.0) is None + + def test_patches_current_price(self): + comp = DriftComparator() + original = {"current_price": 100.0, "30d_high": 110.0, "30d_low": 90.0} + comp.update("BTC", 100.0, original, {}) + patched = comp.get_cached_metrics("BTC", 105.0) + assert patched["current_price"] == 105.0 + assert patched["30d_high"] == 110.0 + assert patched["30d_low"] == 90.0 + + def test_does_not_mutate_stored_metrics(self): + comp = DriftComparator() + original = {"current_price": 100.0} + comp.update("BTC", 100.0, original, {}) + comp.get_cached_metrics("BTC", 999.0) + # Original stored state should be unchanged + assert comp._state["BTC"].last_metrics["current_price"] == 100.0 + + +class TestGetSymbolsNeedingFetch: + def test_returns_all_on_cold_start(self): + comp = DriftComparator() + quotes = { + "BTC": {"price": 100.0}, + "SPY": {"price": 500.0}, + } + result = comp.get_symbols_needing_fetch(quotes) + assert set(result) == {"BTC", "SPY"} + + def test_filters_to_drifted_only(self): + comp = DriftComparator(price_drift_pct=0.005) + comp.update("BTC", 100.0, {}, {}) + comp.update("SPY", 500.0, {}, {}) + quotes = { + "BTC": {"price": 100.01}, # 0.01% — no drift + "SPY": {"price": 510.0}, # 2% — drifted + } + result = comp.get_symbols_needing_fetch(quotes) + assert result == ["SPY"] + + def test_skips_missing_quotes(self): + comp = DriftComparator() + quotes = { + "BTC": {"price": 100.0}, + "SPY": None, + } + result = comp.get_symbols_needing_fetch(quotes) + assert "SPY" not in result + + +class TestHasState: + def test_false_before_update(self): + comp = DriftComparator() + assert comp.has_state("BTC") is False + + def test_true_after_update(self): + comp = DriftComparator() + comp.update("BTC", 100.0, {}, {}) + assert comp.has_state("BTC") is True diff --git a/tests/test_drift_comparator_feed.py b/tests/test_drift_comparator_feed.py new file mode 100644 index 0000000..5e5a494 --- /dev/null +++ b/tests/test_drift_comparator_feed.py @@ -0,0 +1,416 @@ +""" +Integration test: DriftComparator against allocation-engine-2.0 market data feed. + +Tests that the comparator can consume quotes from the Netlify blob stores +(market-quotes, options-chain) used by allocation-engine-2.0's Alpaca pipeline. + +Runs in two modes: + 1. Offline (default): Uses realistic mock data matching the blob format. + 2. Live (NETLIFY_API_TOKEN + NETLIFY_SITE_ID set): Reads actual blobs. +""" + +import json +import os +import time +from unittest.mock import patch + +import pytest +import requests + +from trading_system.drift_comparator import DriftComparator + +# --------------------------------------------------------------------------- +# Blob store client (reusable outside tests) +# --------------------------------------------------------------------------- + +NETLIFY_BLOBS_URL = "https://api.netlify.com/api/v1/blobs" + + +class AllocationFeedClient: + """Reads from allocation-engine-2.0 Netlify blob stores.""" + + def __init__(self, token=None, site_id=None): + self.token = token or os.getenv("NETLIFY_API_TOKEN") + self.site_id = site_id or os.getenv("NETLIFY_SITE_ID") + + @property + def available(self): + return bool(self.token and self.site_id) + + def _headers(self): + return { + "Authorization": f"Bearer {self.token}", + "Accept": "application/json", + } + + def list_blobs(self, store_name): + """List blob keys in a store.""" + url = f"{NETLIFY_BLOBS_URL}/{self.site_id}/{store_name}" + resp = requests.get(url, headers=self._headers(), timeout=10) + resp.raise_for_status() + data = resp.json() + return [b["key"] for b in data.get("blobs", [])] + + def get_blob(self, store_name, key): + """Fetch a single blob by key.""" + url = f"{NETLIFY_BLOBS_URL}/{self.site_id}/{store_name}/{key}" + resp = requests.get(url, headers=self._headers(), timeout=10) + resp.raise_for_status() + return resp.json() + + def get_latest_blob(self, store_name): + """Fetch the most recent blob (keys are ISO timestamps, lexicographic sort).""" + keys = self.list_blobs(store_name) + if not keys: + return None + latest_key = sorted(keys)[-1] + return self.get_blob(store_name, latest_key) + + def get_market_quotes(self): + """Get latest market quotes from the market-quotes store. + + Returns dict of {symbol: {price, bid, ask, spread_bps, ...}}. + """ + blob = self.get_latest_blob("market-quotes") + if not blob: + return {} + return blob.get("latest_quotes", blob) + + def get_options_chain(self, symbol=None): + """Get latest options chain data. + + Blob format: {timestamp, underlying, blob_key, latest_chain: {contract_symbol: {...}}} + If symbol provided, fetches blobs keyed under that underlying. + """ + if symbol: + # Options chain blobs are keyed as "SYMBOL/timestamp" + keys = self.list_blobs("options-chain") + symbol_keys = sorted([k for k in keys if k.startswith(f"{symbol}/")]) + if not symbol_keys: + return {} + return self.get_blob("options-chain", symbol_keys[-1]) + blob = self.get_latest_blob("options-chain") + if not blob: + return {} + return blob + + +def normalize_feed_quotes(feed_quotes): + """Convert allocation-engine-2.0 quote format to drift comparator format. + + Feed format (Alpaca via Redis / Netlify blobs): + {symbol: {bid, ask, mid, spread_bps, timestamp, ...}} + — OR legacy — + {symbol: {bid_price, ask_price, midpoint_price, ...}} + + Comparator format: + {symbol: {price: float}} + """ + normalized = {} + for symbol, q in feed_quotes.items(): + if not q or not isinstance(q, dict): + continue + if symbol == "_meta": + continue + # Prefer mid/midpoint_price, fall back to bid/ask average, then price + price = ( + q.get("mid") + or q.get("midpoint_price") + or q.get("price") + or ( + (q.get("bid", 0) + q.get("ask", 0)) / 2 + if q.get("bid") and q.get("ask") + else None + ) + or ( + (q.get("bid_price", 0) + q.get("ask_price", 0)) / 2 + if q.get("bid_price") and q.get("ask_price") + else None + ) + ) + if price and float(price) > 0: + normalized[symbol] = {"price": float(price)} + return normalized + + +# --------------------------------------------------------------------------- +# Realistic mock data matching allocation-engine-2.0 blob format +# --------------------------------------------------------------------------- + +MOCK_MARKET_QUOTES = { + "timestamp": "2026-03-23T14:30:00", + "blob_key": "2026-03-23T14-30-00", + "latest_quotes": { + "BTC": { + "bid_price": 67450.00, + "ask_price": 67460.00, + "midpoint_price": 67455.00, + "bid_size": 0.5, + "ask_size": 0.3, + "spread_bps": 1.48, + "timestamp": "2026-03-23T14:29:58Z", + }, + "SPY": { + "bid_price": 520.10, + "ask_price": 520.15, + "midpoint_price": 520.125, + "bid_size": 100, + "ask_size": 200, + "spread_bps": 0.96, + "timestamp": "2026-03-23T14:29:59Z", + }, + "QQQ": { + "bid_price": 440.50, + "ask_price": 440.55, + "midpoint_price": 440.525, + "bid_size": 50, + "ask_size": 75, + "spread_bps": 1.13, + "timestamp": "2026-03-23T14:29:59Z", + }, + "IWM": { + "bid_price": 205.30, + "ask_price": 205.35, + "midpoint_price": 205.325, + "bid_size": 150, + "ask_size": 120, + "spread_bps": 2.43, + "timestamp": "2026-03-23T14:29:58Z", + }, + }, + "history_count": 48, +} + +MOCK_OPTIONS_CHAIN = { + "timestamp": "2026-03-23T14:30:05", + "IWM": { + "contracts": [ + { + "symbol": "IWM260417C00210000", + "type": "call", + "strike": 210.0, + "expiration": "2026-04-17", + "last_trade": {"price": 3.45, "size": 10}, + "quote": {"bid": 3.40, "ask": 3.50, "midpoint": 3.45}, + "greeks": { + "delta": 0.42, + "gamma": 0.03, + "theta": -0.08, + "vega": 0.15, + "rho": 0.02, + "iv": 0.22, + }, + }, + { + "symbol": "IWM260417P00200000", + "type": "put", + "strike": 200.0, + "expiration": "2026-04-17", + "last_trade": {"price": 2.10, "size": 5}, + "quote": {"bid": 2.05, "ask": 2.15, "midpoint": 2.10}, + "greeks": { + "delta": -0.35, + "gamma": 0.028, + "theta": -0.07, + "vega": 0.14, + "rho": -0.015, + "iv": 0.24, + }, + }, + ] + }, +} + + +# --------------------------------------------------------------------------- +# Tests: Offline (mock data format validation) +# --------------------------------------------------------------------------- + + +class TestNormalizeQuotes: + """Verify quote normalization handles all allocation-feed formats.""" + + def test_midpoint_preferred(self): + quotes = normalize_feed_quotes(MOCK_MARKET_QUOTES["latest_quotes"]) + assert quotes["BTC"]["price"] == 67455.00 + assert quotes["SPY"]["price"] == 520.125 + + def test_fallback_to_bid_ask_average(self): + raw = {"AAPL": {"bid_price": 180.0, "ask_price": 182.0}} + quotes = normalize_feed_quotes(raw) + assert quotes["AAPL"]["price"] == 181.0 + + def test_fallback_to_price_field(self): + raw = {"AAPL": {"price": 180.5}} + quotes = normalize_feed_quotes(raw) + assert quotes["AAPL"]["price"] == 180.5 + + def test_skips_none_entries(self): + raw = {"AAPL": None, "BTC": {"midpoint_price": 67000}} + quotes = normalize_feed_quotes(raw) + assert "AAPL" not in quotes + assert quotes["BTC"]["price"] == 67000 + + +class TestComparatorWithFeedFormat: + """Drift comparator works with allocation-engine-2.0 quote data.""" + + def test_cold_start_all_symbols_need_fetch(self): + comp = DriftComparator() + quotes = normalize_feed_quotes(MOCK_MARKET_QUOTES["latest_quotes"]) + needs = comp.get_symbols_needing_fetch(quotes) + assert set(needs) == {"BTC", "SPY", "QQQ", "IWM"} + + def test_no_drift_after_update(self): + comp = DriftComparator(price_drift_pct=0.005) + quotes = normalize_feed_quotes(MOCK_MARKET_QUOTES["latest_quotes"]) + + # Simulate first cycle — update all symbols + for sym, q in quotes.items(): + comp.update(sym, q["price"], {"current_price": q["price"]}, {}) + + # Same quotes again — no drift + needs = comp.get_symbols_needing_fetch(quotes) + assert needs == [] + + def test_detects_drift_on_price_move(self): + comp = DriftComparator(price_drift_pct=0.005) + quotes = normalize_feed_quotes(MOCK_MARKET_QUOTES["latest_quotes"]) + + for sym, q in quotes.items(): + comp.update(sym, q["price"], {"current_price": q["price"]}, {}) + + # BTC jumps 1%, others flat + quotes["BTC"]["price"] = 67455.00 * 1.01 # +1% + needs = comp.get_symbols_needing_fetch(quotes) + assert needs == ["BTC"] + + def test_cached_metrics_patch_price_from_feed(self): + comp = DriftComparator() + metrics = {"current_price": 67455.0, "30d_high": 70000.0, "vol_30d": 0.45} + comp.update("BTC", 67455.0, metrics, {}) + + # New feed price, but no drift trigger + patched = comp.get_cached_metrics("BTC", 67480.0) + assert patched["current_price"] == 67480.0 + assert patched["30d_high"] == 70000.0 # preserved + + def test_high_vol_regime_tighter_threshold(self): + comp = DriftComparator(price_drift_pct=0.005, high_vol_drift_pct=0.002) + quotes = normalize_feed_quotes(MOCK_MARKET_QUOTES["latest_quotes"]) + + for sym, q in quotes.items(): + comp.update(sym, q["price"], {"current_price": q["price"]}, {}) + + # SPY moves 0.3% — below default, above high-vol + quotes["SPY"]["price"] = 520.125 * 1.003 + assert comp.get_symbols_needing_fetch(quotes, regime="MEDIUM_VOL") == [] + assert comp.get_symbols_needing_fetch(quotes, regime="HIGH_VOL") == ["SPY"] + + +class TestOptionsChainIntegration: + """Verify options chain data can enrich comparator decisions.""" + + def test_iv_from_options_can_inform_regime(self): + """Options IV could drive regime selection for drift thresholds.""" + chain = MOCK_OPTIONS_CHAIN["IWM"]["contracts"] + avg_iv = sum(c["greeks"]["iv"] for c in chain) / len(chain) + # avg_iv = 0.23 — moderate + regime = "HIGH_VOL" if avg_iv > 0.30 else "MEDIUM_VOL" + assert regime == "MEDIUM_VOL" + assert avg_iv == pytest.approx(0.23, abs=0.01) + + def test_high_iv_triggers_tighter_thresholds(self): + """When options IV is elevated, comparator uses tighter thresholds.""" + comp = DriftComparator(price_drift_pct=0.005, high_vol_drift_pct=0.002) + comp.update("IWM", 205.325, {"current_price": 205.325}, {}) + + # Simulate elevated IV from options chain + mock_iv = 0.35 # elevated + regime = "HIGH_VOL" if mock_iv > 0.30 else "MEDIUM_VOL" + + # 0.3% move — only triggers with HIGH_VOL regime + new_price = 205.325 * 1.003 + assert comp.should_full_fetch("IWM", new_price, regime="MEDIUM_VOL") is False + assert comp.should_full_fetch("IWM", new_price, regime=regime) is True + + +class TestRefreshOpenOrders: + """Test pattern for refreshing metrics on symbols with open orders.""" + + def test_force_fetch_for_open_order_symbols(self): + comp = DriftComparator(price_drift_pct=0.005) + quotes = normalize_feed_quotes(MOCK_MARKET_QUOTES["latest_quotes"]) + + # Warm up all symbols + for sym, q in quotes.items(): + comp.update(sym, q["price"], {"current_price": q["price"]}, {}) + + # No drift, but IWM has an open order — force it into the fetch set + open_order_symbols = {"IWM"} + needs = comp.get_symbols_needing_fetch(quotes) + fetch_set = set(needs) | open_order_symbols # union with open orders + + assert "IWM" in fetch_set + assert "BTC" not in fetch_set + + +# --------------------------------------------------------------------------- +# Tests: Live (only runs with Netlify credentials) +# --------------------------------------------------------------------------- + +needs_netlify = pytest.mark.skipif( + not (os.getenv("NETLIFY_API_TOKEN") and os.getenv("NETLIFY_SITE_ID")), + reason="NETLIFY_API_TOKEN and NETLIFY_SITE_ID required", +) + + +@needs_netlify +class TestLiveFeed: + """Integration tests against real Netlify blob stores.""" + + def setup_method(self): + self.client = AllocationFeedClient() + self.comp = DriftComparator() + + def test_market_quotes_blob_readable(self): + quotes = self.client.get_market_quotes() + assert len(quotes) > 0, "market-quotes store is empty" + # Filter out _meta key + symbol_quotes = {k: v for k, v in quotes.items() if k != "_meta"} + assert len(symbol_quotes) > 0, "No symbol quotes found" + sample = next(iter(symbol_quotes.values())) + assert any( + k in sample + for k in ("mid", "midpoint_price", "price", "bid", "bid_price") + ), f"Unexpected quote format: {list(sample.keys())}" + + def test_comparator_with_live_quotes(self): + raw = self.client.get_market_quotes() + quotes = normalize_feed_quotes(raw) + assert len(quotes) > 0 + + # Cold start — all should need fetch + needs = self.comp.get_symbols_needing_fetch(quotes) + assert len(needs) == len(quotes) + + # Update all + for sym, q in quotes.items(): + self.comp.update(sym, q["price"], {"current_price": q["price"]}, {}) + + # Same quotes — none should need fetch + needs = self.comp.get_symbols_needing_fetch(quotes) + assert needs == [] + + def test_options_chain_blob_readable(self): + chain = self.client.get_options_chain() + assert chain is not None, "options-chain store is empty" + + def test_list_available_stores(self): + """Informational: show what's in each blob store.""" + for store in ("market-quotes", "options-chain", "order-book"): + try: + keys = self.client.list_blobs(store) + print(f"\n [{store}] {len(keys)} blobs, latest: {sorted(keys)[-1] if keys else 'none'}") + except Exception as e: + print(f"\n [{store}] Error: {e}") diff --git a/trading_system/drift_comparator.py b/trading_system/drift_comparator.py new file mode 100644 index 0000000..c5004f7 --- /dev/null +++ b/trading_system/drift_comparator.py @@ -0,0 +1,129 @@ +""" +Drift Comparator — pushdown layer for conditional market data fetching. + +Caches per-symbol price + metrics state. On each cycle, a cheap multi-quote +call checks current prices. Full OHLCV fetches only happen when a symbol's +price has drifted past threshold or data has gone stale. +""" + +import time +from dataclasses import dataclass, field +from typing import Dict, List, Optional + + +@dataclass +class SymbolState: + """Cached state for one symbol after a full OHLCV fetch.""" + last_price: float + last_metrics: Dict + last_market_data: Dict # raw {'intraday': [...], 'daily': [...]} + last_full_fetch_ts: float # time.time() of last full fetch + + +class DriftComparator: + """Determines whether a symbol needs a full market data refresh. + + Usage in the trading loop: + comparator = DriftComparator() + quotes = provider.get_multi_quote(symbols) # 1 API call + + for sym in symbols: + price = quotes[sym]['price'] + if comparator.should_full_fetch(sym, price, regime): + data = fetch_market_data(sym) # 2 API calls + metrics = calculate_metrics(sym, data) + comparator.update(sym, price, metrics, data) + else: + metrics = comparator.get_cached_metrics(sym, price) + # ... execute strategy with metrics + """ + + def __init__( + self, + price_drift_pct: float = 0.005, + high_vol_drift_pct: float = 0.002, + max_stale_minutes: float = 30.0, + ): + self.price_drift_pct = price_drift_pct + self.high_vol_drift_pct = high_vol_drift_pct + self.max_stale_seconds = max_stale_minutes * 60 + self._state: Dict[str, SymbolState] = {} + + def should_full_fetch( + self, + symbol: str, + current_price: float, + regime: str = "MEDIUM_VOL", + ) -> bool: + """Return True if this symbol needs a fresh OHLCV pull. + + Triggers on: + 1. Symbol has never been fetched (cold start). + 2. Price moved more than drift threshold from last baseline. + 3. Cached data exceeded max staleness window. + """ + if symbol not in self._state: + return True + + st = self._state[symbol] + + # Staleness check + age = time.time() - st.last_full_fetch_ts + if age >= self.max_stale_seconds: + return True + + # Price drift check — tighter threshold in high-vol regime + threshold = ( + self.high_vol_drift_pct + if regime == "HIGH_VOL" + else self.price_drift_pct + ) + if st.last_price > 0: + pct_change = abs(current_price - st.last_price) / st.last_price + if pct_change >= threshold: + return True + + return False + + def update( + self, + symbol: str, + price: float, + metrics: Dict, + market_data: Dict, + ) -> None: + """Store baseline state after a full OHLCV fetch.""" + self._state[symbol] = SymbolState( + last_price=price, + last_metrics=dict(metrics), + last_market_data=market_data, + last_full_fetch_ts=time.time(), + ) + + def get_cached_metrics( + self, symbol: str, current_price: float + ) -> Optional[Dict]: + """Return cached metrics with current_price patched in. + + Returns None if the symbol has no cached state. + """ + if symbol not in self._state: + return None + patched = dict(self._state[symbol].last_metrics) + patched["current_price"] = current_price + return patched + + def get_symbols_needing_fetch( + self, + quotes: Dict[str, Dict], + regime: str = "MEDIUM_VOL", + ) -> List[str]: + """Given a dict of symbol -> quote, return symbols that need full fetch.""" + return [ + sym + for sym, q in quotes.items() + if q and q.get("price") and self.should_full_fetch(sym, q["price"], regime) + ] + + def has_state(self, symbol: str) -> bool: + return symbol in self._state diff --git a/trading_system/main.py b/trading_system/main.py index 4865dca..c21bc0d 100644 --- a/trading_system/main.py +++ b/trading_system/main.py @@ -21,6 +21,7 @@ from trading_system.strategies.momentum_dca_strategy import MomentumDcaLongStrategy # noqa: E402 from trading_system.state.state_manager import StateManager # noqa: E402 from trading_system.state.blob_logger import log_state_to_blob # noqa: E402 +from trading_system.drift_comparator import DriftComparator # noqa: E402 from trading_system.market_indicators import fetch_and_write_indicators # noqa: E402 from trading_system.utils.slack import send_slack_alert # noqa: E402 from trading_system.entities.OrderType import OrderSide # noqa: E402 @@ -93,6 +94,8 @@ def __init__(self, twelve_data_api_key: str, symbols: List[str], except Exception as e: print(f" [exec-layer] Execution quality layer init failed (proceeding without): {e}") + self.drift_comparator = DriftComparator() + if strategy_name == 'momentum_dca_long': self.strategy = MomentumDcaLongStrategy(symbols) else: @@ -653,6 +656,21 @@ def run_once(self): print(f" Created: {order['created_at']}") print() + # -- Drift-comparator pushdown: cheap quote check before full fetch -- + # Compute drift metrics early (reads local file cache, 0 API calls) + drift_metrics = self._compute_drift_metrics() + current_regime = (drift_metrics or {}).get("regime", "MEDIUM_VOL") + + # 1 API call for all symbols via multi-quote + quotes = self.data_provider.get_multi_quote(self.symbols) + needs_fetch = self.drift_comparator.get_symbols_needing_fetch( + {s: quotes.get(s) for s in self.symbols}, regime=current_regime + ) + if needs_fetch: + print(f" [drift] Full fetch for: {', '.join(needs_fetch)}") + else: + print(f" [drift] No drift detected — using cached metrics") + for symbol in self.symbols: if self.verbose: print(f"\n{'#'*70}") @@ -660,34 +678,55 @@ def run_once(self): print(f"{'#'*70}\n") try: - # 1. Fetch market data - market_data = self.fetch_market_data(symbol) + quote = quotes.get(symbol) + quote_price = quote.get("price") if quote else None + + if symbol in needs_fetch or not quote_price: + # Full OHLCV fetch path (cold start, drift, or quote failure) + market_data = self.fetch_market_data(symbol) + metrics = self.calculate_metrics(symbol, market_data) + if quote_price: + self.drift_comparator.update( + symbol, quote_price, metrics, market_data + ) + else: + # Cheap path — cached metrics with patched price + metrics = self.drift_comparator.get_cached_metrics( + symbol, quote_price + ) + if metrics is None: + # Fallback: no cached state yet + market_data = self.fetch_market_data(symbol) + metrics = self.calculate_metrics(symbol, market_data) + self.drift_comparator.update( + symbol, quote_price, metrics, market_data + ) + else: + # Update state manager with cached metrics + self.state_manager.update_metrics(symbol, metrics) - # 2. Calculate metrics - metrics = self.calculate_metrics(symbol, market_data) if self.verbose: print(self.metrics_calculator.format_metrics(symbol, metrics)) - # 3. Execute strategy + # Execute strategy signal = self.execute_strategy(symbol, metrics, open_orders) - # 4. Process signal + # Process signal self.process_signal(symbol, signal, open_orders) except Exception as e: print(f"Error processing {symbol}: {e}") continue - # Rate limiting between symbols - time.sleep(1) + # Rate limiting between symbols (only matters when full fetches happen) + if symbol in needs_fetch: + time.sleep(1) # Log state: local file in dry-run, Netlify Blobs when live symbols_filter = None if self.verbose else self.symbols portfolio_data = self.trading_bot.get_portfolio_summary(symbols=symbols_filter) - # Compute drift metrics from cached daily bars (if available) - drift_metrics = self._compute_drift_metrics() - + # drift_metrics already computed above (before symbol loop) log_state_to_blob( self.state_manager, live=not self.dry_run, @@ -754,6 +793,52 @@ def run_once(self): # Send oncall Slack summary every cycle self._send_oncall_summary(open_orders, portfolio_data) + def refresh_open_order_metrics( + self, open_orders: List[Dict] = None + ) -> Dict[str, Dict]: + """Fetch fresh metrics for symbols with open orders. + + Uses multi_quote (1 API call) + selective full fetch only for + symbols whose price has drifted past the comparator threshold. + Can be called independently of run_once(). + """ + if open_orders is None: + open_orders = self.trading_bot.get_open_orders() + + order_symbols = list(set( + o["symbol"] + for o in open_orders + if o.get("symbol") in self.symbols + )) + if not order_symbols: + return {} + + quotes = self.data_provider.get_multi_quote(order_symbols) + results: Dict[str, Dict] = {} + + for symbol in order_symbols: + quote = quotes.get(symbol) + if not quote or not quote.get("price"): + continue + price = quote["price"] + + if self.drift_comparator.should_full_fetch(symbol, price): + market_data = self.fetch_market_data(symbol) + metrics = self.calculate_metrics(symbol, market_data) + self.drift_comparator.update(symbol, price, metrics, market_data) + else: + metrics = self.drift_comparator.get_cached_metrics(symbol, price) + if metrics is None: + market_data = self.fetch_market_data(symbol) + metrics = self.calculate_metrics(symbol, market_data) + self.drift_comparator.update(symbol, price, metrics, market_data) + + results[symbol] = metrics + + if results: + print(f" [orders] Refreshed metrics for: {', '.join(results.keys())}") + return results + def _compute_drift_metrics(self) -> dict: """Compute rolling drift metrics from cached BTC daily bars.