QuantDinger/backend_api_python/tests/test_runtime_io_optimizations.py

from datetime import datetime, timezone
import json

from app.data_sources.us_stock import USStockDataSource
from app.services.strategy_runtime import state
from app.services.strategy_runtime.timeframes import completed_bar_token
from app.services.trading_executor import TradingExecutor
from app.services.trading_executor import _latest_frame_timestamp


def test_completed_bar_token_changes_only_after_next_bar_boundary():
    first = completed_bar_token(
        "1m", datetime(2026, 8, 31, 12, 0, 5, tzinfo=timezone.utc)
    )
    same = completed_bar_token(
        "1m", datetime(2026, 8, 31, 12, 0, 59, tzinfo=timezone.utc)
    )
    following = completed_bar_token(
        "1m", datetime(2026, 8, 31, 12, 1, 0, tzinfo=timezone.utc)
    )

    assert first == same
    assert following == first + 1


def test_runtime_state_coalesces_snapshots_and_flushes_latest(monkeypatch):
    executions = []

    class Cursor:
        def execute(self, _sql, params):
            executions.append(params)

        def close(self):
            pass

    class Connection:
        def __enter__(self):
            return self

        def __exit__(self, *_args):
            return False

        def cursor(self):
            return Cursor()

        def commit(self):
            pass

    clock = iter([100.0, 101.0, 102.0])
    monkeypatch.setattr(state, "get_db_connection", Connection)
    monkeypatch.setattr(state.time, "monotonic", lambda: next(clock))
    store = state.RuntimeStateStore(strategy_id=9, strategy_run_id=2)

    assert store.save({"step": 1}, min_interval_seconds=5) is True
    assert store.save({"step": 2}, min_interval_seconds=5) is False
    assert store.flush() is True

    assert len(executions) == 2
    assert json.loads(executions[-1][-1]) == {"step": 2}


def test_us_stock_quote_batch_deduplicates_and_caches(monkeypatch):
    source = USStockDataSource.__new__(USStockDataSource)
    source.finnhub_client = None
    source.clear_quote_cache()
    calls = []

    def fetch(symbol):
        calls.append(symbol)
        return {"symbol": symbol, "last": 100.0}

    monkeypatch.setattr(source, "_fetch_ticker", fetch)
    monkeypatch.setattr(source, "_fetch_yfinance_batch_quotes", lambda _symbols: {})
    monkeypatch.setenv("US_STOCK_QUOTE_BATCH_WINDOW_MS", "0")
    first = source.get_tickers(["msft", "AAPL", "MSFT"])
    second = source.get_tickers(["AAPL", "MSFT"])

    assert set(first) == {"AAPL", "MSFT"}
    assert set(second) == {"AAPL", "MSFT"}
    assert calls == ["AAPL", "MSFT"]


def test_us_stock_quote_batch_prefers_one_bulk_request(monkeypatch):
    source = USStockDataSource.__new__(USStockDataSource)
    source.finnhub_client = None
    source.clear_quote_cache()
    batches = []

    def fetch_batch(symbols):
        batches.append(symbols)
        return {symbol: {"last": 100.0} for symbol in symbols}

    monkeypatch.setattr(source, "_fetch_yfinance_batch_quotes", fetch_batch)
    monkeypatch.setenv("US_STOCK_QUOTE_BATCH_WINDOW_MS", "0")
    monkeypatch.setattr(
        source,
        "_fetch_ticker",
        lambda _symbol: (_ for _ in ()).throw(AssertionError("fallback not expected")),
    )

    quotes = source.get_tickers(["MSFT", "AAPL", "NVDA"])

    assert batches == [["AAPL", "MSFT", "NVDA"]]
    assert set(quotes) == {"MSFT", "AAPL", "NVDA"}


def test_concurrent_us_stock_quotes_join_one_batch(monkeypatch):
    from concurrent.futures import ThreadPoolExecutor

    source = USStockDataSource.__new__(USStockDataSource)
    source.finnhub_client = None
    source.clear_quote_cache()
    batches = []

    def fetch_batch(symbols):
        batches.append(symbols)
        return {symbol: {"last": 100.0} for symbol in symbols}

    monkeypatch.setattr(source, "_fetch_yfinance_batch_quotes", fetch_batch)
    monkeypatch.setenv("US_STOCK_QUOTE_BATCH_WINDOW_MS", "50")
    with ThreadPoolExecutor(max_workers=3) as pool:
        results = list(pool.map(source.get_ticker, ["MSFT", "AAPL", "NVDA"]))

    assert batches == [["AAPL", "MSFT", "NVDA"]]
    assert all(result["last"] == 100.0 for result in results)


def test_latest_frame_timestamp_detects_candle_advancement():
    import pandas as pd

    first = pd.DataFrame(
        {"close": [1.0]},
        index=pd.DatetimeIndex(["2026-08-31T12:00:00Z"]),
    )
    second = pd.DataFrame(
        {"close": [1.0, 2.0]},
        index=pd.DatetimeIndex([
            "2026-08-31T12:00:00Z",
            "2026-08-31T12:01:00Z",
        ]),
    )

    assert _latest_frame_timestamp(second) > _latest_frame_timestamp(first)


def test_latest_frame_timestamp_reads_strategy_instrument_panel():
    import pandas as pd

    panel = {
        "Crypto:BTC/USDT@swap": pd.DataFrame(
            {"close": [1.0, 2.0]},
            index=pd.DatetimeIndex([
                "2026-08-31T12:00:00Z",
                "2026-08-31T12:01:00Z",
            ]),
        ),
        "Crypto:ETH/USDT@swap": pd.DataFrame(
            {"close": [3.0]},
            index=pd.DatetimeIndex(["2026-08-31T12:02:00Z"]),
        ),
    }

    assert _latest_frame_timestamp(panel) == pd.Timestamp("2026-08-31T12:02:00Z")
    assert _latest_frame_timestamp({}) is None


def test_finnhub_429_blocks_followup_requests_across_symbols(monkeypatch):
    class Client:
        def __init__(self):
            self.calls = []

        def quote(self, symbol):
            self.calls.append(symbol)
            raise RuntimeError("429 rate limit")

    source = USStockDataSource.__new__(USStockDataSource)
    source.finnhub_client = Client()
    source.clear_quote_cache()
    monkeypatch.setenv("FINNHUB_QUOTE_MIN_INTERVAL_SEC", "0")

    assert source._fetch_finnhub_quote("MSFT") == {}
    assert source._fetch_finnhub_quote("AAPL") == {}
    assert source.finnhub_client.calls == ["MSFT"]


def test_live_prices_fetches_each_market_group_as_one_batch(monkeypatch):
    batches = []

    def get_tickers(market, symbols, exchange_id=None, market_type=None):
        batches.append((market, symbols, exchange_id, market_type))
        return {symbol: {"last": index + 100.0} for index, symbol in enumerate(symbols)}

    from app.services import trading_executor

    monkeypatch.setattr(trading_executor.DataSourceFactory, "get_tickers", get_tickers)
    prices = TradingExecutor._live_prices([
        {"key": "USStock:MSFT", "market": "USStock", "symbol": "MSFT"},
        {"key": "USStock:AAPL", "market": "USStock", "symbol": "AAPL"},
    ])

    assert len(batches) == 1
    assert batches[0][1] == ["MSFT", "AAPL"]
    assert prices == {"USStock:MSFT": 100.0, "USStock:AAPL": 101.0}