A股实时看板接入批量行情前，先解决三类工程问题

import os
import pandas as pd
from quantdash import QuantDash

# 依赖建议：Python 3.10+，quantdash==0.1.0，pandas 2.x
# 环境变量名：QUANTDASH_API_KEY
qd = QuantDash(api_key=os.getenv('QUANTDASH_API_KEY'))

watch_symbols = pd.Index([
    '600519.SH',
    '000001.SZ',
    '920047.BJ',
], name='symbol')

quotes = qd.quotes.get(
    symbols=watch_symbols.tolist(),
    to_dataframe=True,
)

print(quotes.columns.tolist())
print(quotes.head())

def align_to_watchlist(quotes: pd.DataFrame, watch_symbols: pd.Index) -> pd.DataFrame:
    if 'symbol' not in quotes.columns:
        raise ValueError('quotes must contain symbol column')

    df = quotes.copy()
    df['symbol'] = df['symbol'].astype(str)
    df = df.drop_duplicates(subset=['symbol'], keep='last')
    df = df.set_index('symbol')

    aligned = pd.DataFrame(index=watch_symbols).join(df, how='left')
    aligned['is_missing_quote'] = aligned['last_price'].isna() if 'last_price' in aligned else True
    return aligned.reset_index()

aligned_quotes = align_to_watchlist(quotes, watch_symbols)
print(aligned_quotes[['symbol', 'is_missing_quote']])

import time
from dataclasses import dataclass
from typing import Optional

@dataclass
class Snapshot:
    data: pd.DataFrame
    fetched_at_ms: int
    ok: bool
    error: Optional[str] = None


def now_ms() -> int:
    return int(time.time() * 1000)


def fetch_once(qd: QuantDash, symbols: list[str]) -> pd.DataFrame:
    df = qd.quotes.get(symbols=symbols, to_dataframe=True)
    if not isinstance(df, pd.DataFrame):
        raise TypeError('quotes.get(to_dataframe=True) should return a DataFrame')
    return df


def refresh_loop(qd: QuantDash, symbols: list[str], interval_sec: float = 5.0):
    last_snapshot: Optional[Snapshot] = None

    while True:
        start = time.monotonic()
        try:
            df = fetch_once(qd, symbols)
            last_snapshot = Snapshot(data=df, fetched_at_ms=now_ms(), ok=True)
            yield last_snapshot
        except Exception as exc:
            stale = last_snapshot.data if last_snapshot is not None else pd.DataFrame()
            yield Snapshot(
                data=stale,
                fetched_at_ms=now_ms(),
                ok=False,
                error=type(exc).__name__,
            )

        elapsed = time.monotonic() - start
        sleep_sec = max(0.0, interval_sec - elapsed)
        time.sleep(sleep_sec)

def normalize_realtime_quotes(df: pd.DataFrame) -> pd.DataFrame:
    required = {'symbol', 'last_price', 'prev_close'}
    missing_cols = required - set(df.columns)
    if missing_cols:
        raise ValueError(f'missing required columns: {sorted(missing_cols)}')

    out = df.copy()
    out['symbol'] = out['symbol'].astype(str)
    out = out.drop_duplicates(subset=['symbol'], keep='last')

    for col in ['last_price', 'prev_close', 'volume', 'amount']:
        if col in out.columns:
            out[col] = pd.to_numeric(out[col], errors='coerce')

    out['quote_error'] = ''
    out.loc[out['last_price'].isna(), 'quote_error'] += '|missing_last_price'
    out.loc[out['prev_close'].isna(), 'quote_error'] += '|missing_prev_close'
    out.loc[out['last_price'] <= 0, 'quote_error'] += '|non_positive_last_price'
    out.loc[out['prev_close'] <= 0, 'quote_error'] += '|non_positive_prev_close'

    valid_base = out['prev_close'] > 0
    out['calc_change_pct'] = pd.NA
    out.loc[valid_base, 'calc_change_pct'] = (
        out.loc[valid_base, 'last_price'] / out.loc[valid_base, 'prev_close'] - 1
    )

    if 'timestamp' in out.columns:
        out['timestamp'] = pd.to_numeric(out['timestamp'], errors='coerce')
        out.loc[out['timestamp'].isna(), 'quote_error'] += '|missing_timestamp'

    out['is_valid_for_alert'] = out['quote_error'].eq('')
    return out

quality_checked = normalize_realtime_quotes(aligned_quotes)
print(quality_checked[['symbol', 'last_price', 'prev_close', 'calc_change_pct', 'is_valid_for_alert', 'quote_error']])

def mark_stale_quotes(df: pd.DataFrame, max_age_ms: int = 30_000) -> pd.DataFrame:
    out = df.copy()
    if 'timestamp' not in out.columns:
        out['is_stale'] = True
        return out

    current = now_ms()
    out['timestamp'] = pd.to_numeric(out['timestamp'], errors='coerce')
    out['quote_age_ms'] = current - out['timestamp']
    out['is_stale'] = out['quote_age_ms'].isna() | (out['quote_age_ms'] > max_age_ms)
    return out