Python 行情看板同时取最新价和 K 线，怎样拆请求避免接口意大利面

from dataclasses import dataclass
from typing import Iterable

import pandas as pd
from quantdash import QuantDash


@dataclass(frozen=True)
class MarketRequest:
    symbols: list[str]
    period: str = '1d'
    count: int = 100
    adjust: str = 'forward'


@dataclass
class MarketBundle:
    quotes: pd.DataFrame
    klines: pd.DataFrame


def normalize_symbols(symbols: Iterable[str]) -> list[str]:
    cleaned = []
    seen = set()
    for raw in symbols:
        symbol = str(raw).strip().upper()
        if not symbol or symbol in seen:
            continue
        cleaned.append(symbol)
        seen.add(symbol)
    return cleaned


class QuantDashMarketSource:
    def __init__(self) -> None:
        self.qd = QuantDash()

    def fetch_quotes(self, symbols: list[str]) -> pd.DataFrame:
        return self.qd.quotes.get(
            symbols=symbols,
            to_dataframe=True,
        )

    def fetch_klines(
        self,
        symbols: list[str],
        period: str,
        count: int,
        adjust: str,
    ) -> dict[str, pd.DataFrame]:
        return self.qd.klines.batch(
            symbols,
            period=period,
            count=count,
            adjust=adjust,
            to_dataframe=True,
            show_progress=True,
        )

def prepare_quotes(df: pd.DataFrame, symbols: list[str]) -> pd.DataFrame:
    if df is None or df.empty:
        raise ValueError('quotes is empty')
    if 'symbol' not in df.columns:
        raise ValueError('quotes must contain symbol column')

    quotes = df.copy()

    if 'timestamp' in quotes.columns:
        quotes = quotes.sort_values(['symbol', 'timestamp'])

    quotes = quotes.drop_duplicates(subset=['symbol'], keep='last')
    quotes = quotes.set_index('symbol', drop=False)

    missing = sorted(set(symbols) - set(quotes.index))
    if missing:
        raise ValueError(f'missing quote rows: {missing}')

    return quotes.loc[symbols]


def prepare_klines(dfs: dict[str, pd.DataFrame], symbols: list[str]) -> pd.DataFrame:
    frames = []

    for symbol in symbols:
        df = dfs.get(symbol)
        if df is None or df.empty:
            raise ValueError(f'empty kline data: {symbol}')

        one = df.copy()
        time_col = 'trade_time' if 'trade_time' in one.columns else 'trade_date'
        if time_col not in one.columns:
            raise ValueError(f'kline data has no trade_date or trade_time: {symbol}')

        if 'symbol' not in one.columns:
            one['symbol'] = symbol

        one = one.sort_values(time_col)
        one = one.drop_duplicates(subset=['symbol', time_col], keep='last')
        one = one.set_index(['symbol', time_col])
        frames.append(one)

    panel = pd.concat(frames).sort_index()
    return panel

def load_market_bundle(request: MarketRequest) -> MarketBundle:
    symbols = normalize_symbols(request.symbols)
    if not symbols:
        raise ValueError('symbols is empty')

    source = QuantDashMarketSource()

    raw_quotes = source.fetch_quotes(symbols)
    raw_klines = source.fetch_klines(
        symbols=symbols,
        period=request.period,
        count=request.count,
        adjust=request.adjust,
    )

    quotes = prepare_quotes(raw_quotes, symbols)
    klines = prepare_klines(raw_klines, symbols)

    return MarketBundle(quotes=quotes, klines=klines)


if __name__ == '__main__':
    req = MarketRequest(
        symbols=['600519.SH', '000001.SZ'],
        period='1d',
        count=60,
        adjust='forward',
    )
    bundle = load_market_bundle(req)

    print(bundle.quotes.columns.tolist())
    print(bundle.klines.index.names)

def validate_bundle(bundle: MarketBundle, symbols: list[str]) -> None:
    quote_symbols = set(bundle.quotes.index)
    expected_symbols = set(symbols)
    if quote_symbols != expected_symbols:
        raise ValueError('quote symbols are not aligned with request symbols')

    if not bundle.klines.index.is_monotonic_increasing:
        raise ValueError('kline index should be sorted')

    required_price_cols = {'open', 'high', 'low', 'close'}
    missing_price_cols = required_price_cols - set(bundle.klines.columns)
    if missing_price_cols:
        raise ValueError(f'missing kline columns: {sorted(missing_price_cols)}')

    null_close = bundle.klines['close'].isna().sum()
    if null_close > 0:
        raise ValueError(f'close contains null values: {null_close}')