python量化获取分钟级别数据的几种方式

def get_minute_data(stock_code, period):    data = xtdata.get_market_data(        field_list=["time", "open", "high", "low", "close", "volume"],        stock_list=[stock_code],        period=period,        fill_data=True  # 填充缺失数据    )    return data

    raw_data = tq.get_market_data(                stock_list=stock_list,                start_time=start_time,                end_time=end_time,                period='1m',                dividend_type='front',                field_list=['Open', 'High', 'Low', 'Close', 'Amount'],                fill_data=True            )

"""eltdx 分钟K线数据演示 (Streamlit)=================================演示如何用 eltdx 通达信数据接口拉取个股「分钟级」行情:    client.bars.get("sz000001", period="1m", count=240)参数说明--------    code    —— 带市场前缀的证券代码(如 sz000001 / sh600519)    period  —— 周期: 1m / 5m / 15m / 30m / 60m    count   —— 取最近多少根; 1m 时 240 ≈ 一个交易日(4 小时)    adjust  —— 复权: qfq 前复权 / hfq 后复权 / None 不复权(分钟数据一般不复权)返回结构--------    client.bars.get(...) -> KlineSeries, 其 .bars 为 KlineBar 元组, 每根含:        time          带时区(+08:00) 的 datetime        open/high/low/close   OHLC        volume_lots   成交量(单位: 手)        amount        成交额(单位: 元)运行----    streamlit run eltdx_minute_demo.py"""import loggingimport pandas as pdimport plotly.graph_objects as gofrom plotly.subplots import make_subplotsimport streamlit as stfrom eltdx import TdxClient# ====================# 页面与日志配置# ====================st.set_page_config(page_title="eltdx 分钟K线", page_icon="📈", layout="wide")logging.basicConfig(level=logging.INFO, format="%(asctime)s  %(levelname)s  %(message)s")logger = logging.getLogger(__name__)# A 股配色: 红涨绿跌COLOR_UP = "#d62728"COLOR_DOWN = "#2ca02c"PERIODS = {"1 分钟": "1m", "5 分钟": "5m", "15 分钟": "15m", "30 分钟": "30m", "60 分钟": "60m"}ADJUSTS = {"不复权": None, "前复权(qfq)": "qfq", "后复权(hfq)": "hfq"}DEFAULT_CODE = "sz000001"  # 平安银行# ====================# eltdx 数据层# ====================@st.cache_data(show_spinner="正在拉取证券代码表...", ttl=600)def build_name_map() -> dict[str, str]:    """拉取沪深北三市证券代码表, 构建 full_code -> name 映射。缓存 10 分钟。"""    name_map: dict[str, str] = {}    with TdxClient(timeout=10) as client:        for market in ("sh", "sz", "bj"):            try:                for sec in client.codes.all(market):                    name_map[sec.full_code] = sec.name            except Exception as e:  # noqa: BLE001                logger.warning("拉取 %s 代码表失败: %s", market, e)    return name_mapdef normalize_code(code: str) -> str:    """把用户输入规整为 full_code(如 sh600519)。    支持:        · 已带前缀: sh/sz/bj + 6 位        · 6 位纯数字: 按首位判市场(60/68/9→sh, 00/30→sz, 4/8→bj)        · 名称: 查代码表反查    """    code = str(code).strip().lower()    for pref in ("sh", "sz", "bj"):        if code.startswith(pref) and len(code) == len(pref) + 6:            return code    if code.isdigit() and len(code) == 6:        head = code[0]        if head in ("6", "9"):            return f"sh{code}"        if head in ("0", "3"):            return f"sz{code}"        if head in ("4", "8"):            return f"bj{code}"        raise ValueError(f"无法识别代码前缀: {code}")    # 名称反查    nm = build_name_map()    for fc, name in nm.items():        if code == fc or code in str(name).lower():            return fc    raise ValueError(f"无法解析代码/名称: {code}")@st.cache_data(show_spinner="正在拉取分钟K线...", ttl=60)def fetch_minute_bars(full_code: str, period: str, count: int, adjust) -> pd.DataFrame:    """★ 核心调用 ★ 用 client.bars.get 拉取分钟K线。    返回 DataFrame, 列:        time / open / high / low / close / volume(手) / amount(元)        pct_chg(%, 相邻根) / vwap(分时均价 元/股, 截至该根)    时间升序。    """    with TdxClient(timeout=10) as client:        series = client.bars.get(            full_code, period=period, count=count, adjust=adjust, kind="stock",        )    if not series or not series.bars:        raise RuntimeError(f"拉取 {full_code}{period} K线失败, 返回为空(可能非交易时段/停牌)")    bars = sorted(series.bars, key=lambda b: b.time)    df = pd.DataFrame([{        "time": b.time,        "open": b.open, "high": b.high, "low": b.low, "close": b.close,        "volume": b.volume_lots,   # 手        "amount": b.amount,        # 元    } for b in bars])    df["pct_chg"] = (df["close"].pct_change().fillna(0) * 100).round(3)    # 分时均价 VWAP: 累计成交额(元) / 累计成交量(股)。volume 单位是「手」, ×100 换算成股    cum_shares = (df["volume"] * 100).cumsum().replace(0, pd.NA)    df["vwap"] = (df["amount"].cumsum() / cum_shares).round(3)    return df# ====================# 绘图# ====================def render_chart(df: pd.DataFrame, title: str) -> go.Figure:    """画「价格K线 + VWAP」与「成交量」双子图。A 股配色: 红涨绿跌。"""    fig = make_subplots(        rows=2, cols=1, shared_xaxes=True,        vertical_spacing=0.08, row_width=[0.72, 0.28],        subplot_titles=("价格 K线 + 分时均价 VWAP", "成交量(手)"),    )    up = df["close"] >= df["open"]    colors = [COLOR_UP if u else COLOR_DOWN for u in up]    # —— 价格: K线 + VWAP ——    fig.add_trace(go.Candlestick(        x=df["time"], open=df["open"], high=df["high"], low=df["low"], close=df["close"],        increasing_line=dict(color=COLOR_UP, width=1),        decreasing_line=dict(color=COLOR_DOWN, width=1),        increasing_fillcolor=COLOR_UP, decreasing_fillcolor=COLOR_DOWN,        name="K线", showlegend=False, whiskerwidth=0.4,    ), row=1, col=1)    fig.add_trace(go.Scatter(        x=df["time"], y=df["vwap"], mode="lines",        line=dict(color="#1f77b4", width=1.4, dash="dot"),        name="VWAP", connectgaps=True,    ), row=1, col=1)    # —— 成交量 ——    fig.add_trace(go.Bar(        x=df["time"], y=df["volume"], marker_color=colors,        name="成交量", showlegend=False,    ), row=2, col=1)    fig.update_layout(        title=dict(text=title, x=0.5, xanchor="center"),        height=620,        template="plotly_white",        xaxis_rangeslider=dict(visible=False),        margin=dict(l=50, r=20, t=70, b=40),        hovermode="x unified",    )    fig.update_yaxes(title_text="价格", row=1, col=1)    fig.update_yaxes(title_text="成交量(手)", row=2, col=1)    fig.update_xaxes(        rangeslider=dict(visible=False),        row=2, col=1,        rangeselector=dict(            buttons=[                dict(count=30, label="30分", step="minute", stepmode="backward"),                dict(count=60, label="1小时", step="minute", stepmode="backward"),                dict(count=120, label="2小时", step="minute", stepmode="backward"),                dict(label="全部", step="all"),            ],            x=0.0, xanchor="left",        ),    )    return fig# ====================# Streamlit 界面# ====================def main() -> None:    st.title("📈 eltdx 分钟K线数据演示")    st.caption(        "核心调用：`client.bars.get(code, period, count, adjust, kind='stock')`"        "　·　数据源：通达信 (eltdx)"    )    # —— 侧边栏参数 ——    with st.sidebar:        st.header("⚙️ 查询参数")        code_input = st.text_input(            "股票代码 / 名称", value=DEFAULT_CODE,            help="6 位数字、带前缀(sz000001) 或名称(平安银行)皆可",        )        period_label = st.selectbox("周期 period", list(PERIODS.keys()), index=0)        count = st.slider(            "根数 count", 30, 1200, 240, step=10,            help="1m 时 240 ≈ 一个交易日(4 小时)",        )        adjust_label = st.selectbox("复权 adjust", list(ADJUSTS.keys()), index=0)        run_btn = st.button("🚀 拉取数据", type="primary", use_container_width=True)        st.divider()        st.markdown("**关于 eltdx**")        st.caption("通达信行情接口的 Python 封装，分钟K线来自 `TdxClient().bars.get()`。")    period = PERIODS[period_label]    adjust = ADJUSTS[adjust_label]    st.code(        f'client.bars.get("{code_input}", period="{period}", count={count}, '        f'adjust={adjust!r}, kind="stock")',        language="python",    )    # 首次进入给出提示    if not run_btn and "minute_df" not in st.session_state:        st.info("👈 在左侧填好参数，点 **拉取数据** 开始。")        return    # —— 解析代码 ——    try:        full_code = normalize_code(code_input)    except Exception as e:  # noqa: BLE001        st.error(f"代码解析失败：{e}")        return    name = build_name_map().get(full_code, "")    st.session_state["query"] = (full_code, name, period, count, adjust)    # —— 拉取 ——    try:        df = fetch_minute_bars(full_code, period, count, adjust)    except Exception as e:  # noqa: BLE001        st.error(f"拉取失败：{e}")        return    st.session_state["minute_df"] = df    if df.empty:        st.warning("返回数据为空（可能非交易时段 / 停牌 / 代码不存在）。")        return    # —— 顶部摘要 ——    first_open = float(df["open"].iloc[0])    last_close = float(df["close"].iloc[-1])    hi = float(df["high"].max())    lo = float(df["low"].min())    rng_pct = (last_close - first_open) / first_open * 100 if first_open else 0.0    tot_vol = float(df["volume"].sum())    tot_amt = float(df["amount"].sum())    c1, c2, c3, c4, c5, c6 = st.columns(6)    c1.metric("收盘", f"{last_close:.2f}", f"{rng_pct:+.2f}%")    c2.metric("区间最高", f"{hi:.2f}")    c3.metric("区间最低", f"{lo:.2f}")    c4.metric("总成交量", f"{tot_vol/1e4:.1f}万手")    c5.metric("总成交额", f"{tot_amt/1e8:.2f}亿")    c6.metric("根数", f"{len(df)}")    t0, t1 = df["time"].iloc[0], df["time"].iloc[-1]    st.caption(f"⏱️ 数据区间：`{t0:%Y-%m-%d %H:%M}` ~ `{t1:%Y-%m-%d %H:%M}`")    # —— 图表 ——    title = f"{full_code}{name}{period_label}K线　({len(df)} 根)"    fig = render_chart(df, title)    st.plotly_chart(fig, use_container_width=True)    # —— 数据表 ——    st.subheader("📋 原始数据")    show_df = df.copy()    show_df["time"] = show_df["time"].dt.strftime("%Y-%m-%d %H:%M")    st.dataframe(        show_df,        use_container_width=True,        hide_index=True,        column_config={            "open": st.column_config.NumberColumn(format="%.2f"),            "high": st.column_config.NumberColumn(format="%.2f"),            "low": st.column_config.NumberColumn(format="%.2f"),            "close": st.column_config.NumberColumn(format="%.2f"),            "vwap": st.column_config.NumberColumn(format="%.3f"),            "pct_chg": st.column_config.NumberColumn(format="%.3f%%"),            "amount": st.column_config.NumberColumn(format="%d"),        },    )    # 导出    st.download_button(        "⬇️ 导出 CSV",        df.to_csv(index=False).encode("utf-8-sig"),        file_name=f"{full_code}_{period}_{len(df)}.csv",        mime="text/csv",    )if __name__ == "__main__":    main()