盘中候选股如何批量取行情并用 Pandas 本地排序

from quantdash import QuantDash

qd = QuantDash()  # 默认读取环境变量 QUANTDASH_API_KEY

symbols = [
    "600519.SH",
    "000001.SZ",
    "00700.HK",
    "AAPL.US",
]

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

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

import pandas as pd


def normalize_quotes(df: pd.DataFrame) -> pd.DataFrame:
    """
    将实时行情快照整理成候选股排序需要的字段。
    该函数不假设所有扩展字段一定存在。
    """
    if df is None or df.empty:
        return pd.DataFrame()

    data = df.copy()

    required = ["symbol", "last_price", "prev_close", "volume"]
    missing = [col for col in required if col not in data.columns]
    if missing:
        raise ValueError(f"行情数据缺少必要字段: {missing}")

    rename_map = {}
    if "ext.name" in data.columns:
        rename_map["ext.name"] = "name"
    if "ext.change_pct" in data.columns:
        rename_map["ext.change_pct"] = "change_pct"
    if "ext.turnover_rate" in data.columns:
        rename_map["ext.turnover_rate"] = "turnover_rate"
    if "ext.amplitude" in data.columns:
        rename_map["ext.amplitude"] = "amplitude"

    data = data.rename(columns=rename_map)

    numeric_cols = [
        "last_price",
        "prev_close",
        "open",
        "high",
        "low",
        "volume",
        "amount",
        "change_pct",
        "turnover_rate",
        "amplitude",
    ]
    for col in numeric_cols:
        if col in data.columns:
            data[col] = pd.to_numeric(data[col], errors="coerce")

    data = data.drop_duplicates(subset=["symbol"], keep="last")

    if "change_pct" not in data.columns:
        valid = data["prev_close"].notna() & (data["prev_close"] != 0)
        data.loc[valid, "change_pct"] = (
            data.loc[valid, "last_price"] / data.loc[valid, "prev_close"] - 1
        )

    keep_cols = [
        "symbol",
        "name",
        "last_price",
        "prev_close",
        "change_pct",
        "volume",
        "amount",
        "turnover_rate",
        "amplitude",
        "timestamp",
    ]
    keep_cols = [col for col in keep_cols if col in data.columns]

    return data[keep_cols].reset_index(drop=True)

def filter_candidates(
    quotes: pd.DataFrame,
    min_volume: float = 100_000,
    min_amount: float | None = None,
    min_change_pct: float = -0.03,
    max_change_pct: float = 0.08,
) -> pd.DataFrame:
    """
    对候选股行情做本地过滤。
    change_pct 使用小数口径，例如 0.03 表示 3%。
    """
    if quotes is None or quotes.empty:
        return pd.DataFrame()

    data = quotes.copy()

    mask = (
        data["symbol"].notna()
        & data["last_price"].notna()
        & data["prev_close"].notna()
        & (data["prev_close"] > 0)
        & data["change_pct"].notna()
        & data["change_pct"].between(min_change_pct, max_change_pct)
        & data["volume"].fillna(0).ge(min_volume)
    )

    if min_amount is not None and "amount" in data.columns:
        mask = mask & data["amount"].fillna(0).ge(min_amount)

    return data.loc[mask].reset_index(drop=True)

def rank_candidates(
    candidates: pd.DataFrame,
    top_n: int = 20,
    weight_change: float = 0.5,
    weight_liquidity: float = 0.4,
    weight_activity: float = 0.1,
) -> pd.DataFrame:
    """
    对过滤后的候选股打分排序。
    分数越高，排序越靠前。
    """
    if candidates is None or candidates.empty:
        return pd.DataFrame()

    data = candidates.copy()

    liquidity_col = "amount" if "amount" in data.columns else "volume"

    data["rank_change"] = data["change_pct"].rank(pct=True, method="average")
    data["rank_liquidity"] = data[liquidity_col].rank(pct=True, method="average")

    if "amplitude" in data.columns:
        data["rank_activity"] = data["amplitude"].rank(pct=True, method="average")
    else:
        data["rank_activity"] = 0.0
        weight_activity = 0.0

    total_weight = weight_change + weight_liquidity + weight_activity
    if total_weight <= 0:
        raise ValueError("排序权重之和必须大于 0")

    data["score"] = (
        weight_change * data["rank_change"]
        + weight_liquidity * data["rank_liquidity"]
        + weight_activity * data["rank_activity"]
    ) / total_weight

    data["change_pct_display"] = data["change_pct"] * 100

    sort_cols = ["score", "change_pct", liquidity_col]
    data = data.sort_values(sort_cols, ascending=[False, False, False])

    display_cols = [
        "symbol",
        "name",
        "last_price",
        "change_pct_display",
        liquidity_col,
        "score",
    ]
    display_cols = [col for col in display_cols if col in data.columns]

    return data[display_cols].head(top_n).reset_index(drop=True)

from quantdash import QuantDash
import pandas as pd


def refresh_intraday_ranking(symbols: list[str], top_n: int = 20) -> pd.DataFrame:
    qd = QuantDash()

    raw_quotes = qd.quotes.get(
        symbols=symbols,
        to_dataframe=True,
    )

    normalized = normalize_quotes(raw_quotes)

    filtered = filter_candidates(
        normalized,
        min_volume=100_000,
        min_amount=None,
        min_change_pct=-0.03,
        max_change_pct=0.08,
    )

    ranked = rank_candidates(filtered, top_n=top_n)

    return ranked


if __name__ == "__main__":
    candidate_symbols = [
        "600519.SH",
        "000001.SZ",
        "00700.HK",
        "AAPL.US",
    ]

    result = refresh_intraday_ranking(candidate_symbols, top_n=10)
    print(result)

import time


def safe_fetch_quotes(qd: QuantDash, symbols: list[str], max_retries: int = 3) -> pd.DataFrame:
    last_error = None

    for attempt in range(max_retries):
        try:
            return qd.quotes.get(symbols=symbols, to_dataframe=True)
        except Exception as exc:
            last_error = exc
            sleep_seconds = min(2 ** attempt, 10)
            time.sleep(sleep_seconds)

    raise RuntimeError("多次获取行情失败") from last_error

from datetime import datetime, timezone


def add_local_snapshot_time(df: pd.DataFrame) -> pd.DataFrame:
    data = df.copy()
    data["snapshot_collected_at"] = datetime.now(timezone.utc)
    return data