导言 / TL;DR 在全球化资产配置中，跨市场策略（如中美科技股轮动、AH溢价套利）需要同时获取 A股、港股和美股的数据。然而，不同市场之间代码后缀混乱、休市日历错位和字段差异常常让开发者写出大量的 if-else 地狱代码[4]。本文向您展示如何通过 QuantDash 客户端一站式对齐跨市场多资产数据。 技术痛点拆解（传统写法的“硬伤”） 反面教材 (Before)：“万国码”拼接地狱 当采用多套开源或商业数据源组合跨市场回测时，代码很容易写成一团乱麻： # 传统写法：调用不同数据源与处理各异的代码后缀 import yfinance as yf import efinance as ef # 标的代码格式各异：A股、港股、美股各有一套规则 a_stock = "600519" # efinance 使用纯数字 us_stock = "AAPL" # yfinance # 痛点 1：API 接口完全不统一 df_a = ef.stock.get_quote_history(a_stock) # 返回的是带有中文列名的 DataFrame df_us = yf.download(us_stock) # 返回英文 MultiIndex # 痛点 2：时间列名称不一致，导致后续的多股 Join 合并极易产生错位 生产级解决方案（基于 QuantDash SDK） QuantDash 统一了底层多个市场的字段格式，支持国际通用的 {代码}.{交易所后缀} 标准（如 AAPL.US, 00700.HK, 600519.SH）[4][5]，极大地减轻了数据对齐成本。 输入: {"A股": "600519.SH", "港股": "00700.HK", "美股": "AAPL.US"} │ ▼ [QuantDash 统一行情客户端] ──> 标准时间戳对齐 │ ▼ [统一标准化标准列名 DataFrame] 优雅实现 (After)：一套标准接口秒级跨市场对齐 import pandas as pd from quantdash import QuantDash # 实例化官方客户端 (沙盒公共 Token) qd = QuantDash(api_key="demo_public_token") def fetch_and_align_global_assets(): targets = { "A_Share": "600519.SH", "HK_Stock": "00700.HK", "US_Stock": "AAPL.US" } aligned_df = pd.DataFrame() for market, symbol in targets.items(): try: # 修正 1：将原先的 qfq 改为官方严格支持的标准值 "forward" df = qd.klines.get( symbol=symbol, period="1d", adjust="forward", # 必须使用 "forward"、"backward" 或 "none" to_dataframe=True ) if df is not None and not df.empty: df['trade_date'] = pd.to_datetime(df['trade_date']) # 提取收盘价，以交易日期对齐 df_close = df[['trade_date', 'close']].rename(columns={'close': f'{market}_close'}) if aligned_df.empty: aligned_df = df_close else: # 使用外连接（Outer Join）保留所有交易日 aligned_df = pd.merge(aligned_df, df_close, on='trade_date', how='outer') else: print(f"[警告] {symbol} 返回数据为空") except Exception as e: print(f"[错误] 获取 {symbol} 失败: {e}") # 按时间排序并前向填充缺失值（解决不同市场假期不一致问题） if not aligned_df.empty: aligned_df = aligned_df.sort_values('trade_date').ffill() return aligned_df if __name__ == "__main__": aligned_portfolio = fetch_and_align_global_assets() print("\n--- 跨市场多资产收盘价对齐结果 ---") print(aligned_portfolio.tail()) DataFrame 输出 --- 跨市场多资产收盘价对齐结果 --- trade_date A_Share_close HK_Stock_close US_Stock_close 101 2026-07-20 1327.50 477.8 326.59 102 2026-07-21 1308.00 474.0 327.74 103 2026-07-22 1305.00 440.6 325.89 104 2026-07-23 1292.01 445.2 321.66 105 2026-07-24 1297.41 434.6 333.02 Cursor / DeepSeek AI 编程助手专享 Prompt 我需要使用 Python 编写一个跨市场 ETF 轮动选股策略。 请写一段基于 `quantdash` SDK 的代码，导入客户端类（from quantdash import QuantDash; qd = QuantDash(api_key="sk_xxxxx")），分别获取 SPY.US、02800.HK 和 510300.SH 过去 30 天的前复权日 K 线（period="1d", adjust="qfq", to_dataframe=True）。提取收盘价并合并成一个以 trade_date 为对齐标准的 DataFrame。 总结与“三步走”落地指引 第一步：获取完整源码。访问官方开源托管仓库获取该跨市场时序对齐的完整代码模板：https://github.com/quantdash-net/QuantDash（请认准 quantdash-net）。 第二步：申请专属密钥。注册获取您的个人 API Key：https://quantdash.net/。 第三步：查阅开发细节。更多有关美股和港股高频行情接口参数请参考：https://docs.quantdash.net/。 导言 / TL;DR 在量化多因子选股或全市场异动复盘中，顺序拉取几百只股票的历史K线往往因网络时延累加而导致程序运行极度缓慢。本文将分享如何利用 Python 内置的线程池（ThreadPoolExecutor），结合 QuantDash 官方客户端搭建一套高并发、带指数退避重试机制的多股行情下载流[2]，在避免被服务商频控的同时将耗时降低 80%。 技术痛点拆解（传统写法的“硬伤”） 反面教材 (Before)：低效的串行 for 循环与硬编码延时 很多开发者在调用 AkShare、Tushare 或爬虫接口时，通常会写出如下的串行循环代码[2][3]： # 传统低效写法：串行拉取 import time import akshare as ak # 常见开源爬虫 symbols = ["600519", "000858", "600809"] # 假设有数百只 data_list = [] for symbol in symbols: try: # 爬虫接口容易因为访问频繁被封禁，只能被迫加入 sleep 降低速度 df = ak.stock_zh_a_hist(symbol=symbol, period="daily") data_list.append(df) time.sleep(1.5) # 致命痛点：500只股票需要等待 750 秒！ except Exception as e: print(f"获取 {symbol} 失败: {e}") 传统硬伤**：** 网络时滞累加：每次网络请求的往返时间（RTT）叠加，拉取500只股票耗时长达十几分钟。 缺乏流控与错误退避：一旦遇到网络抖动造成单只股票拉取失败，直接导致整个任务中断。 生产级解决方案（基于 QuantDash SDK） 我们可以使用 ThreadPoolExecutor 建立一个并发管道。QuantDash 的轻量客户端原生的 klines.get 接口非常契合并发场景。 [股票列表: 500只] ────> [ThreadPoolExecutor] ├── 线程 1 ──> qd.klines.get() (QuantDash) ──> DataFrame ├── 线程 2 ──> qd.klines.get() (QuantDash) ──> DataFrame └── 线程 3 (遇异常触发退避重试) ──> 自动重试 ──> DataFrame 优雅实现 (After)：并发与指数退避重试机制 import time from concurrent.futures import ThreadPoolExecutor, as_completed import pandas as pd from quantdash import QuantDash # 初始化官方客户端 (沙盒测试 Token，生产密钥请替换为您的 sk_xxxxx) qd = QuantDash(api_key="demo_public_token") def fetch_single_stock_with_retry(symbol: str, max_retries: int = 3, base_delay: float = 1.0): """带指数退避重试机制的单股行情拉取""" for attempt in range(max_retries): try: # 修正 1：QuantDash 官方 SDK 前复权参数值为 "forward"，不复权为 "none" df = qd.klines.get( symbol=symbol, period="1d", adjust="forward", # 必须使用 "forward" to_dataframe=True ) if df is not None and not df.empty: return symbol, df else: raise ValueError("API 返回了空的数据集(Empty DataFrame)") except Exception as e: # 修正 2：绝不静默吞掉异常，打印真实的底层报错以供定位 print(f"[警告] 拉取 {symbol} 失败 (第 {attempt+1}/{max_retries} 次尝试): {e}") if attempt < max_retries - 1: time.sleep(base_delay * (2 ** attempt)) return symbol, None def parallel_fetch_klines(symbol_list: list, max_workers: int = 3): """多线程池并发拉取""" all_data = {} with ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_symbol = { executor.submit(fetch_single_stock_with_retry, sym): sym for sym in symbol_list } for future in as_completed(future_to_symbol): symbol = future_to_symbol[future] try: sym, df = future.result() if df is not None: all_data[sym] = df else: print(f"[错误] {symbol} 在重试 {max_retries} 次后仍未成功获取数据。") except Exception as e: print(f"[线程崩溃] 线程执行异常 {symbol}: {e}") return all_data if __name__ == "__main__": # 测试标的列表 test_symbols = ["600519.SH", "00700.HK", "AAPL.US"] start_time = time.time() results = parallel_fetch_klines(test_symbols, max_workers=3) print(f"\n拉取任务结束！总耗时: {round(time.time() - start_time, 2)} 秒") # 安全检查：确保 results 中确实存在该 key 再打印 if "600519.SH" in results: print("\n--- 贵州茅台 (600519.SH) 标准日 K 线数据 ---") print(results["600519.SH"].head()) else: print("\n[错误] 未能成功拉取 600519.SH 的数据，请检查上方打印的异常日志。") DataFrame 输出： 拉取任务结束！总耗时: 1.94 秒 --- 贵州茅台 (600519.SH) 标准日 K 线数据 --- symbol name timestamp trade_date trade_time open high low close volume amount 0 600519.SH 贵州茅台 1772380800000 2026-03-02 2026-03-02 00:00:00 1416.358649 1423.196243 1403.328150 1406.698107 35454 5.115064e+09 1 600519.SH 贵州茅台 1772467200000 2026-03-03 2026-03-03 00:00:00 1406.688339 1419.162063 1389.135259 1393.101064 45891 6.565382e+09 2 600519.SH 贵州茅台 1772553600000 2026-03-04 2026-03-04 00:00:00 1382.170682 1389.985075 1359.792215 1368.671319 48014 6.743267e+09 3 600519.SH 贵州茅台 1772640000000 2026-03-05 2026-03-05 00:00:00 1373.379490 1380.178012 1362.634701 1366.580969 30505 4.276226e+09 4 600519.SH 贵州茅台 1772726400000 2026-03-06 2026-03-06 00:00:00 1362.634701 1374.844689 1355.797107 1369.472294 29154 4.072329e+09 Cursor / DeepSeek AI 编程助手专享 Prompt 我正在使用 Python 编写一个批量拉取股票 K 线数据的脚本。请基于 `quantdash` SDK，帮我实现一个多线程并发拉取器。 1. SDK 初始化：from quantdash import QuantDash; qd = QuantDash(api_key="your-key") 2. 行情接口：qd.klines.get(symbol=symbol, period="1d", adjust="qfq", to_dataframe=True) 3. 要求使用 concurrent.futures.ThreadPoolExecutor 实现并发。 4. 加入指数退避重试逻辑，当遇到连接异常时自动等待并重试，最高重试3次。 总结与“三步走”落地指引 第一步：获取完整源码。访问官方开源托管仓库获取多线程并发脚本及更复杂的队列流控制机制：https://github.com/quantdash-net/QuantDash（欢迎 Star 支持）。 第二步：申请专属密钥。注册获取您的个人 API Key，摆脱公共测试配额限制：https://quantdash.net/。 第三步：查阅开发细节。更详尽的 K 线周期及限流说明请参考官方文档：https://docs.quantdash.net/。 一只原本值得继续研究的股票已经出现明显变化，你到盘后才从新闻或朋友转发里看到。 问题可能不在指标。它从来没有进入你的观察池，均线、量比、估值和预警自然都看不见它。你花了半小时调整参数，实际上只是把一个偏小的范围筛得更细。 所以，多市场选股的第一步不是继续加指标，而是先回答： 该看的市场和标的，是否已经进入同一张观察清单？ TickDB 在这里解决的也不是“替你找出会涨的股票”。它能把 A 股、港股、美股等对象放进同一条行情研究路径：先找到对象，再批量取得当前行情、研究指标和历史 K 线，最后由你自己的规则决定先研究谁。 先看这次真实运行的结果。 三个市场，先放进同一条研究链 本次运行时间为 2026年7月24日09:11:36 UTC。共发出 8 次请求，3 个跨市场样本全部通过。 市场与样本 当次目录股票数 现价 5日变化 10日变化 量比 日K线 A股 600519.SH 贵州茅台 7059 1297.41 3.54% 7.67% 0.48 20根 港股 700.HK 腾讯控股 3156 434.6 -5.85% -5.56% 0.58 20根 美股 AAPL.US Apple Inc. 12587 321.66 -3.48% 1.72% 0.78 20根 三只股票只是容易识别的接口验收样本，不代表推荐。表里的目录数量、价格、变化率和量比，也只对应这次运行。 这张表真正有用的地方，不是比较三家公司的好坏，更不是把不同币种的股价放在一起排名。它证明了一件更基础的事：三个市场的对象能够先进入同一条研究链，再用一致的步骤核对当前状态和历史数据。 脚本还按“五日绝对涨跌幅”排了一个人工复核顺序： 700.HK > 600519.SH > AAPL.US 这个顺序只表示本次三个样本里先检查谁，不是买卖信号，也不代表未来收益。 完整代码：从对象目录跑到日K线 下面是本次真实运行的完整代码。Key 只从环境变量读取，不会打印，也不会写入输出文件。 复现前先在本机设置： export TICKDB_API_KEY="你自己的 Key" 然后运行： #!/usr/bin/env python3 """真实运行 TickDB 的 A股、港股、美股最小研究链。 运行前： export TICKDB_API_KEY="你自己的 Key" 密钥只从环境变量读取，不打印、不写入文件。 """ from __future__ import annotations import json import os import subprocess import sys import urllib.parse from datetime import datetime, timezone from pathlib import Path from typing import Any BASE_URL = "https://api.tickdb.ai" RUN_ID = "F-W30-01R-THS-EV13-LIVE-20260724" SAMPLES = [ ("A股", "CN", "600519.SH"), ("港股", "HK", "700.HK"), ("美股", "US", "AAPL.US"), ] OUTPUT_ROOT = Path(__file__).resolve().parents[1] def save_json(path: Path, value: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text( json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) def get_json( evidence_id: str, path: str, params: dict[str, Any], api_key: str, ) -> dict[str, Any]: query = urllib.parse.urlencode(params) url = f"{BASE_URL}{path}?{query}" result = subprocess.run( [ "curl", "--location", "--silent", "--show-error", "--max-time", "30", "--header", f"X-API-Key: {api_key}", "--header", "Accept: application/json", "--write-out", "\nHTTP_STATUS:%{http_code}", url, ], check=False, capture_output=True, text=True, timeout=35, ) body, marker, status = result.stdout.rpartition("\nHTTP_STATUS:") http_status = int(status.strip()) if marker and status.strip().isdigit() else None if result.returncode != 0 or http_status != 200: raise RuntimeError( f"request failed: path={path}, " f"curl_exit={result.returncode}, http={http_status}" ) payload = json.loads(body) if payload.get("code") != 0: raise RuntimeError( f"API rejected request: path={path}, code={payload.get('code')}, " f"message={payload.get('message')}" ) record = { "evidence_id": evidence_id, "evidence_run_id": RUN_ID, "retrieved_at_utc": datetime.now(timezone.utc).isoformat(timespec="seconds"), "method": "GET", "base_url": BASE_URL, "path": path, "query": params, "http_status": http_status, "response": payload, "sanitization": "Authentication header and API key are excluded.", } save_json(OUTPUT_ROOT / "raw" / f"{evidence_id}.json", record) return payload def rows_by_symbol(payload: dict[str, Any]) -> dict[str, dict[str, Any]]: return { row["symbol"]: row for row in payload["data"] if isinstance(row, dict) and "symbol" in row } def pct(value: Any) -> str: return f"{float(value) * 100:.2f}%" def main() -> int: api_key = os.getenv("TICKDB_API_KEY") if not api_key: print("请先在环境变量中设置 TICKDB_API_KEY", file=sys.stderr) return 2 retrieved_at = datetime.now(timezone.utc).isoformat(timespec="seconds") catalog_totals: dict[str, int] = {} for _, market, _ in SAMPLES: payload = get_json( f"catalog_{market.lower()}_stock", "/v1/symbols/available", {"market": market, "type": "stock", "limit": 3, "offset": 0}, api_key, ) catalog_totals[market] = payload["data"]["pagination"]["total"] symbols = ",".join(symbol for _, _, symbol in SAMPLES) ticker_payload = get_json( "ticker_three_markets", "/v1/market/ticker", {"symbols": symbols, "type": "stock"}, api_key, ) metrics_payload = get_json( "metrics_three_markets", "/v1/market/calc-index", {"symbols": symbols, "type": "stock"}, api_key, ) ticker = rows_by_symbol(ticker_payload) metrics = rows_by_symbol(metrics_payload) lines = [ "$ python public_tickdb_cross_market_check.py", f"evidence_run_id: {RUN_ID}", f"retrieved_at_utc: {retrieved_at}", "api_key: loaded from environment (not printed)", "flow: 3 catalogs + batch ticker + batch metrics + 3 daily K-lines", "", ] results: list[dict[str, Any]] = [] review_order: list[tuple[float, str]] = [] for market_name, catalog_market, symbol in SAMPLES: kline_payload = get_json( f"kline_{symbol.lower().replace('.', '_')}_1d_20", "/v1/market/kline", {"symbol": symbol, "type": "stock", "interval": "1d", "limit": 20}, api_key, ) bars = kline_payload["data"]["klines"] quote = ticker[symbol] indicator = metrics[symbol] review_order.append((abs(float(indicator["five_day_change_rate"])), symbol)) row = { "market": market_name, "symbol": symbol, "name": quote["name"], "catalog_stock_total": catalog_totals[catalog_market], "last_price": quote["last_price"], "price_change_percent_24h": quote["price_change_percent_24h"], "ticker_timestamp": quote["timestamp"], "five_day_change_rate": pct(indicator["five_day_change_rate"]), "ten_day_change_rate": pct(indicator["ten_day_change_rate"]), "volume_ratio": indicator["volume_ratio"], "daily_kline_count": len(bars), "latest_kline_time": bars[-1]["time"], "latest_close": bars[-1]["close"], "status": "PASS" if len(bars) == 20 else "FAIL", } results.append(row) lines.extend( [ f"[{market_name}] {symbol} {row['name']}", ( f"目录股票数={row['catalog_stock_total']} | " f"现价={row['last_price']} | 24h={row['price_change_percent_24h']}%" ), ( f"5日={row['five_day_change_rate']} | " f"10日={row['ten_day_change_rate']} | " f"量比={row['volume_ratio']}" ), ( f"日K线={row['daily_kline_count']}根 | " f"最近收盘={row['latest_close']} | " f"时间={row['latest_kline_time']} | {row['status']}" ), "", ] ) review_order.sort(reverse=True) lines.extend( [ "人工复核顺序（仅按5日绝对涨跌幅）：" + " > ".join(symbol for _, symbol in review_order), "SUMMARY: 3/3 samples passed", "BOUNDARY: 本次三个样本、单次运行；不是荐股、收益排名或SLA证明。", ] ) summary = { "schema_version": "tickdb-live-research-acceptance-summary-1.0", "task_id": "F-W30-01R-THS", "platform": "tonghuashun", "evidence_run_id": RUN_ID, "retrieved_at_utc": retrieved_at, "credential": "loaded from environment; value not saved", "request_count": 8, "passed_samples": sum(row["status"] == "PASS" for row in results), "total_samples": len(results), "results": results, "manual_review_order_by_absolute_5d_move": [ symbol for _, symbol in review_order ], "boundary": ( "This single run proves only the stated requests, samples and time. " "It does not prove completeness, long-term stability, latency, " "accuracy, SLA or investment performance." ), } save_json( OUTPUT_ROOT / "summary" / "live_research_acceptance_summary.json", summary, ) transcript = "\n".join(lines) + "\n" terminal_path = OUTPUT_ROOT / "terminal" / "public_code_run_terminal.txt" terminal_path.parent.mkdir(parents=True, exist_ok=True) terminal_path.write_text(transcript, encoding="utf-8") print(transcript, end="") return 0 if summary["passed_samples"] == len(results) else 1 if __name__ == "__main__": raise SystemExit(main()) 这段代码按下面的顺序工作： 查询 CN、HK、US 三个股票目录，确认对象范围； 一次批量取得三只样本的当前行情； 一次批量取得五日、十日变化率和量比； 分别取得三只样本的 20 根日 K 线； 保存脱敏原始响应、汇总和终端文本； 只按五日绝对涨跌幅排人工复核顺序。 同一次运行的完整终端输出 $ python public_tickdb_cross_market_check.py evidence_run_id: F-W30-01R-THS-EV13-LIVE-20260724 retrieved_at_utc: 2026-07-24T09:11:36+00:00 api_key: loaded from environment (not printed) flow: 3 catalogs + batch ticker + batch metrics + 3 daily K-lines [A股] 600519.SH 贵州茅台 目录股票数=7059 | 现价=1297.41 | 24h=0.42% 5日=3.54% | 10日=7.67% | 量比=0.48 日K线=20根 | 最近收盘=1297.41 | 时间=1784822400000 | PASS [港股] 700.HK 腾讯控股 目录股票数=3156 | 现价=434.6 | 24h=-2.38% 5日=-5.85% | 10日=-5.56% | 量比=0.58 日K线=20根 | 最近收盘=434.6 | 时间=1784822400000 | PASS [美股] AAPL.US Apple Inc. 目录股票数=12587 | 现价=321.66 | 24h=-1.30% 5日=-3.48% | 10日=1.72% | 量比=0.78 日K线=20根 | 最近收盘=321.66 | 时间=1784779200000 | PASS 人工复核顺序（仅按5日绝对涨跌幅）：700.HK > 600519.SH > AAPL.US SUMMARY: 3/3 samples passed BOUNDARY: 本次三个样本、单次运行；不是荐股、收益排名或SLA证明。 8 次请求均返回 HTTP 200、接口 code 0，3/3 样本通过。这只能证明本次三个样本和这些请求跑通，不能外推为全市场完整、长期稳定、准确率、低延迟或 SLA。 TickDB 的六类核心价值，放到投资任务里怎么看 产品能力不能停在“接口能返回数据”。对投资者来说，更重要的是它能把哪一步做得更顺。 核心价值 投资者能得到什么 本文如何体现 看得更广 先把更多市场和对象放进观察范围 三个股票目录与统一 symbol 规则，让 A 股、港股、美股样本进入同一研究链 筛得更快 对池内对象排出先研究谁 批量行情和指标提供研究输入；排序规则由使用者定义 及时感知变化 让观察池的变化进入后续监控 当前产品能力可继续接入实时订阅；本轮没有实跑 WebSocket 判断更稳 先核对对象、类型、时间和数据状态 symbol、type、时间戳、K 线数量和返回状态都能成为检查项 少做重复接入 扩展市场时复用一套请求、鉴权和保存逻辑 同一脚本处理三个市场；本文不据此承诺具体成本降幅 做成研究工具或 AI 工作流 把结构化行情接入扫描器、研究助手或内部工具 结构化结果可以继续加工；本轮没有实跑 AI 判断 这六类价值不是六个互不相干的口号。它们更适合按任务组合。 四套更实用的能力组合 1. 普通投资者的跨市场观察池 组合：看得更广 + 筛得更快 + 判断更稳 先按市场、行业或产业链找到对象，再批量取得当前状态和历史 K 线。排序只负责告诉你“先检查谁”，最终判断仍由人完成。 适合的任务是：每天从多个市场中缩小人工阅读范围，又不想一开始就漏掉对象。 2. 量化研究的统一输入链 组合：看得更广 + 筛得更快 + 判断更稳 + 少做重复接入 先统一对象标识和数据资格检查，再把行情、指标和 K 线保存到研究环境。后面无论做因子、回测还是横截面比较，都从同一批经过检查的输入开始。 本文没有验证任何因子收益，也没有给出策略结论。这里讨论的是研究输入，不是收益结果。 3. 从扫描到持续监控 组合：筛得更快 + 及时感知变化 + 判断更稳 先用批量数据缩小范围，再把留下的对象放进持续监控。实际落地时还需要自行处理交易时段、权限、断线恢复、补查、去重和保存。 本次只跑了 REST 请求，没有把实时订阅写成已经验证的结论。 4. 研究工具或 AI 助手 组合：看得更广 + 判断更稳 + 少做重复接入 + 做成工具或 AI 工作流 对象、行情、指标和 K 线都变成结构化输入后，可以继续做内部扫描器、研究看板或 AI 研究助手。AI 负责整理和提问，人负责核对来源、时间与风险。 这条组合的价值在于减少数据准备的断点，不代表 AI 可以替代投资判断。 三个容易混淆的问题 1. 为什么代码里是 600519.SH、700.HK 和 AAPL.US？ 后缀用于区分市场。统一 symbol 让程序知道对象属于哪个市场，也能减少重名和代码歧义。统一写法不代表三个市场的字段、权限、深度、交易时段和历史范围完全一样。 2. 当前行情和日 K 线为什么都要取？ 当前行情回答“现在是什么状态”，日 K 线回答“此前怎样走到这里”。前者适合扫描，后者适合核对历史路径。只看其中一个，研究上下文都不够完整。 3. 能不能直接把人工复核顺序当成选股结果？ 不能。本文只是用五日绝对涨跌幅演示如何在三个样本中安排检查顺序。它没有使用公司基本面、估值、流动性、交易成本、汇率、组合风险等条件，也没有验证与未来收益的关系。 想自己试，先做这四步 把自己的 TickDB Key 放进环境变量，不要写死在代码里； 把三个示例 symbol 换成自己的观察对象； 先核对对象、类型、返回时间和 K 线数量； 再加入自己的筛选条件，并把“排序”和“买卖决定”分开。 如果这四步跑通，你得到的不是一张神奇股票名单，而是一条更完整、可检查、可重复的研究输入链。 观察池决定你有机会研究谁，指标决定你怎样缩小范围。次序反了，指标越精细，也可能只是在一个过窄的池子里反复优化。 引言：破解“收盘即迷茫”的困局 很多新手投资者常有这种困扰：每天下午三点收盘后，看着红红绿绿的涨跌幅榜，心中却一片茫然。明天该买什么？热点还能持续吗？这种“收盘即迷茫”的根源，在于你根本没有建立起一套科学的“复盘”体系。 作为投资者，你必须明白：真正的复盘绝非简单地查看账面盈亏，而是一套洞察资金动向、拆解市场逻辑的严密程序。通过深度复盘，你才能从纷乱的波动中理清思路，变被动为主动。今天，我把这套极具操作性的“五步复盘法”教给你，这不仅是学知识，更是帮你找回在市场中生存的底气。 第一步：指数共振——给市场“量体温”的艺术 复盘的第一步，你必须先看大局。每天收盘后，你要做的第一件事就是查看各大主要指数的表现。 **●**关注共振与分化： 几大指数是同步上扬（共振），还是各走各路（分化）？这决定了当前市场的合力程度。 **●**识别关键位： 盯紧指数是否处于重要压力位、支撑位或整数关口。 ●判断资金“温度”： 如果指数处于高位且成交异常放大，那是“发烧”过热，是贪婪的信号；如果指数持续阴跌触及支撑，那是“低烧”恐慌，往往孕育机会。 **●**风格研判： 通过对比，看资金是在追逐大盘股维稳，还是在中小盘、创业板里掘金。 “对指数的复盘能让你对当前行情的短期趋势，以及资金的情绪和风格偏向，有一个大致的判断。” 第二步：板块梳理——捕捉资金的“主战场” 看完了大盘，你得知道资金都在哪儿打架。这是从宏观进入中观的关键一步。 你必须将当天的题材板块和行业板块按涨幅排序。不要只看谁涨得好，你得去跟踪一段时间，看它的持续性如何。真正的强势板块绝不是“一日游”，你需要从技术面、基本面和消息面多个维度去综合观察对比。只有那些经过多维度验证、能持续走强的板块，才是你接下来的重点伏击区。 第三步：涨停个股——市场情绪的“晴雨表” 涨停股是市场的“先锋队”，它们最能直观反映资金的真实情绪。你必须从以下三个维度盯防涨停股： **●**涨停数量： 数量多代表市场情绪热烈，数量缩减则代表资金开始观望。 **●一字板的加速（一字板——**即开盘即涨停）： 反映了资金极度的饥渴与一致性预期。 **●涨停开板的频率（开板——**即涨停被砸开）： 这反映了多头力量的衰竭和抛压的动态变化。 通过这三个维度的观察，结合板块和消息面，你就能锁定当下最受资金认可的交易风向。 第四步：深挖逻辑——新闻消息的“续航力” 消息和政策是板块行情的催化剂，但你千万不能看到新闻就盲目冲进去。 复盘消息面，你必须深究背后的推动逻辑。为什么有的利好出来能涨一周，有的利好出来却是“见光死”？这取决于消息是否解决了行业的内核痛点，或者是否超出了市场预期。这种对“持续性”的判断，需要你通过大量的复盘实战去积累经验，学会分辨哪些是短期脉冲，哪些是能支撑趋势的硬逻辑。 第五步：心理建设——把“军令状”刻进交易纪律 复盘到最后，最关键的是如何将认知转化为行动。很多投资者拍着胸脯说“我记住了”，结果第二天开盘就全忘了。好记性不如烂笔头！ 你必须建立起复盘的“仪式感”： **●**记录与收藏： 将复盘的心得和看好的标的记录下来，这不仅是笔记，更是你与自己签下的心理契约。 ●立下“军令状” 就像我常说的，在评论区留下“红火”二字，这不仅是祈愿，更是在内心深处刻下纪律——符合复盘条件的标的就买，不符合就等。 “你克制住了冲动，财富自然就来了。” 这种物理上的记录和心理上的承诺，能帮你对抗盘中的贪婪与恐惧。 总结：从筛选到跟踪，完成交易闭环 复盘的终点是筛选标的。 这是一个从宏观（指数决定仓位）到中观（板块决定方向），再到微观（个股决定武器）的递进过程。只有走完这五步，你的复盘才算真正转化为了生产力。长期坚持系统化复盘、留存各类盘面数据，可借助 9db交割单 整理汇总行情参考素材。 当财神（市场红利）来敲门时，你是大门紧闭、一无所知，还是已经做好了万全的复盘准备？ 在这个充满波动的市场里，没有随随便便的成功。从今晚收盘开始，拿起笔，建立你的复盘笔记。记住，机会永远只留给那些在红利到来前，就已经做好了所有功课的人。 引言：揭开技术指标的“面纱” 在变幻莫测的股市战场上，普通投资者最容易陷入的泥潭就是：被纷乱的价格波动牵着鼻子走。你是否经历过看着股价翻红便匆忙追入，结果买在最高点？或者在股价阴跌时盲目抄底，却发现下面还有地狱？ 作为交易者，我们需要一套逻辑自洽的系统来剥开市场的迷雾。而 MACD（平滑异同移动平均线）凭借其对趋势的精准捕捉，被誉为“趋势指标之王”。它不仅仅是简单的买卖信号，更是一把衡量市场能量的“标尺”。今天，我将带你拆解这个经典指标，从复杂的参数中读懂最直观的交易逻辑。 核心逻辑：不仅仅是两条线，它是“聚合与发散”的艺术 很多股民看了多年 MACD，却不知道这四个字母背后的深意。MACD 的逻辑并非凭空捏造，它是对均线（MA）原理的深度升华。 “MACD 可以拆分为 MA、C、D。MA 是均线，C 是聚合（Convergence），D 是发散（Divergence）。这个指标表示的是移动平均线与价格聚合、分散的关系。” 为了更精准地刻画市场，我们要关注其核心组件和标准参数（12, 26, 9）： ●DIF（快线）： 默认为 12 日 EMA 减去 26 日 EMA。它代表了短期与长期趋势之间的差距。当 DIF 为正值，市场由多头主导；为负值，则空头占优。 ●DEA（慢线）： 它是 DIF 的 9 日平滑移动平均线。作为快线的“影子”，它的移动更为稳健，起到过滤噪音的作用。 **●**能量柱（柱状图）： 这是 MACD 的精髓。柱状图的本质是 DIF 与 DEA 之间的差值。当两条线之间的距离拉大，意味着能量正在“发散”，趋势在加强；当距离缩小，意味着能量“聚合”，趋势可能衰减。 Takeaway 1: 零轴之上是晴天，零轴之下是寒冬 在专业交易员的眼中，MACD 的零轴不仅是一条线，它更像是“水面”，是划分市场阴阳的分界线。 生存法则：永远不要忽视那条“零轴” “零轴就是中间这条线，是一个多空分界线。” **●**水上世界（零轴上方）： 当 DIF 和 DEA 稳居零轴上方时，说明大环境处于多头市场。此时红柱放大的每一次跃动，都是多头能量的喷薄。在“水上”操作，顺水行舟，胜率天然更高。 **●**水下世界（零轴下方）： 当指标深陷零轴下方且向下发散，意味着空头势力猖獗。此时即便是反弹，往往也只是寒冬里的短暂暖阳，存在进一步下跌风险。 Takeaway 2: 并非所有的“金叉”都值得你孤注一掷 当 DIF 向上穿过 DEA，能量柱由绿转红时，我们称之为“金叉”。但作为策略师，我必须提醒你：金叉的“含金量”取决于它长在什么地方。 水上与水下：识别高含金量的买入信号 **●**高含金量信号：水上金叉。 如果 DIF 在零轴上方回踩 DEA 后再次向上勾头，这种“水上金叉”代表了多头趋势的二次加速。特别是当低位连续出现两次金叉，往往预示着一波大级别的上涨行情。 **●**低含金量信号：水下金叉。 虽然水下金叉（绿柱缩短）也意味着空头减弱，但它更多是超跌后的反弹，上涨力度有限。 ●警惕信号：死叉（DIF 下穿 DEA）。 当红柱转绿，无论在水上还是水下，这都是动能衰减的警告。水上死叉需警惕深度回调，水下死叉则意味着下跌趋势的延续，必须果断规避。 Takeaway 3: 背离——市场给你的“最后通牒” 当价格走势与指标方向出现“性格不合”时，这就是技术分析中最具预警价值的信号——背离。 背离：当指标开始**“拒绝”**承认股价的新高 **●**顶背离： 股价一浪高过一浪，创出新高，但 MACD 的快线峰值却一浪低于一浪。这意味着股价的上涨纯粹是“强弩之末”，内在动能已枯竭，是极度危险的见顶信号。 **●**底背离： 股价不断下挫创出新低，但 MACD 指标却开始逐级抬升。这预示着卖盘动能耗尽，市场正在悄悄积蓄反转的力量。 提示： 背离虽不是立刻反转的绝对指令，但它是最重要的“先行信号”。当背离出现，请务必结合成交量或其他指标，做好撤退或进场的准备。 Takeaway 4: 隐藏的指标——克制与纪律 在交易的世界里，再完美的工具也无法拯救没有灵魂的操作。原文提到的“留评论、立军令状”，其实是一种深刻的交易心理学应用。 **终极武器：技术之外的“军令状” 为什么要留下“红火”二字？这不仅仅是为了讨个吉利，而是一种自我暗示的“意志算法”。 **●**立下军令状： 每一个字都是在与自己达成契约——从此告别冲动，严格执行“符合条件就买，不符合就等”的纪律。 **●**筛选思维模型： 当你主动关注“干货”内容，系统也会优化你的信息流。同样地，当你的大脑习惯了逻辑筛选而非情绪驱动，你的交易账户才会真正开始“红火”。 记住，技术是剑，而克制与纪律才是挥剑的手。 结语：通往财富红利的阶梯 MACD 是一个集合了趋势判断、动能衡量和风险预警的综合工具。理解了零轴的定性、金叉的层级以及背离的预警，你便掌握了股市中一套逻辑严密的作战地图。 然而，工具本身并不能带来财富。真正的财富红利，只会眷顾那些对规则有敬畏心、对纪律有执行力的人。 互动思考： 在下一次金叉出现时，你是会由于肾上腺素飙升而冲动满仓，还是会冷静地先看一眼：它究竟是发生在“晴朗的水上”，还是“寒冷的水下”？ 最近我专门针对 Supermind 平台的AI 量化代码生成平台进行了优化改进，现在效果比市面上的 DS、豆包等工具好很多。 👉 SuperMind AI量化代码生成平台 这个工具最大的特点是直接和 AI 对话就能生成完整可运行的Supermind量化策略代码。你不需要懂 Python、C# 或策略 API，只要用自然语言描述你的交易逻辑，比如：“当5日均线向上突破20日均线时买入，反向时卖出。” AI 就会自动帮你生成完整策略代码，并能直接在平台上运行。 相比于通用大模型的输出，这个平台针对量化交易进行了专门优化生成的代码结构更清晰，逻辑更准确，对策略逻辑的理解更接近量化开发者的思路，并且可用作 API 查询或策略自动生成工具 之前上线后，很多朋友反馈代码质量和可运行性都非常高，几乎不需要再手动修改。现在我们的AI量化代码生成平台已经全面支持 Supermind，你可以直接体验。如果你之前在用 DS、豆包等平台，不妨试试看这个版本，可能会刷新你对AI 写量化策略的想象。 股票Level2和港股行情数据 做量化或者单纯想研究市场微观结构的朋友，经常头疼数据从哪里找。今天不聊策略，就单纯聊聊我最近在用的一个数据源，看看里面到底有什么东西，给需要的人一个参考。 数据主要分两大块，一块是A股的Level2行情，另一块是港股的分钟线和逐笔tick。数据包是压缩好的，下载下来解压就能用，格式主要是csv，用起来比较直接。 先看看A股的Level2数据有啥 这个Level2数据不是那种简单的分时图，里面的东西要细得多。我主要用来看盘口的变化和资金流向。从CMES金融数据库中下载了一个数据文件里通常包含这些字段： 时间戳：这个不用说，精确到秒，有时候是毫秒级的，看具体的数据包。 股票代码：比如 600519.SH 这种格式。 最新价：就是当前那一笔成交的价格。 成交量：当前这一笔成交了多少股。 成交额：这一笔成交了多少钱，这个算资金流的时候挺有用的。 买卖盘口：这是重点。一般会有买一价到买十价，买一量到买十量，同样卖盘也是十档。有了这个才能看到真实的委托堆积情况，而不是只看一个五档。 总委买量/总委卖量：所有买盘和卖盘挂单的总和，有时候用来判断压力支撑。 加权平均委买/委卖价格：这个指标我自己用得不多，但数据里有提供。 这些字段基本上能把一个时刻的盘面情况还原个七七八八。比如你想算主力资金净流入，或者监控大单的动向，没有十档行情还真不好弄。我之前试过用普通行情算，误差挺大的，后来换了Level2数据感觉靠谱不少。 为了验证一些想法，我通常会写个简单的脚本来读取数据做初步处理。比如用Python的pandas，几行代码就能把数据拉出来看看结构。 # 示例：使用CMES金融数据库的行情接口读取Level2数据文件 # 注意：确保已安装pandas，并且数据文件路径正确 # pip install pandas import pandas as pd # 假设数据已经下载并解压，这里读取一个CSV文件 # 具体字段名需要根据实际数据文件调整，这里只是示例 data_path = 'path/to/your/level2_data.csv' try: df = pd.read_csv(data_path, encoding='gbk') # 有时数据是gbk编码 print("数据列名：", df.columns.tolist()) print("前几行数据：\n", df.head()) except FileNotFoundError: print("文件没找到，检查一下路径吧。") except Exception as e: print(f"读取数据时出了点问题：{e}") 港股行情：分钟线和逐笔Tick 港股的数据包是分开的，分钟线一个包，逐笔tick一个包。分钟线数据对做日间或者稍短周期策略的够用了，逐笔数据就更细了，适合做高频或者订单流分析。 分钟线数据 一般包含这些： 时间、股票代码、开盘价、最高价、最低价、收盘价、成交量、成交额。和A股的分时数据差不多，但频率是每分钟一根K线。 逐笔成交数据（Tick） 就厉害多了，每一笔真实的成交记录都有。字段大概是这样： 成交时间：精确到毫秒甚至更细。 股票代码 成交价格 成交数量 成交金额 买卖方向：这个很重要，标识这一笔是主动买还是主动卖。但有些数据源可能不直接提供，需要自己用盘口数据去推断。 用逐笔数据可以还原出非常细致的交易图谱，谁在买，谁在卖，大单小单分布，都能看出来。不过数据量也很大，处理起来对硬件有点要求。 几种数据的对比和怎么选 简单列一下区别，方便大家按需选择： 数据类别 数据粒度 主要用途 数据量大小 处理难度 A股Level2 3秒/笔？不等 盘口分析、资金流、主力监控 比较大 中等，需要整理 港股分钟线 1分钟/K线 技术分析、中短期策略回测 相对较小 容易，类似日线 港股逐笔Tick 每笔成交 高频策略、订单流分析、微观结构研究 非常大 较高，需要专门处理 选哪个完全看你的需求。如果只是做普通的回测，分钟线可能就够了。如果想研究盘口博弈或者做T+0相关的策略，那Level2或者逐笔数据几乎是必须的。我自己的经验是，刚开始不用追求最细的数据，先从分钟线跑通策略逻辑，再上更细的数据做优化，不然数据处理这块就可能卡住很久。 对了，这些数据在CMES金融数据库的下载页面上是分门别类放好的，需要哪个下哪个就行，不用一次性全搞下来。我之前就犯过傻，一口气下了好几个G的逐笔数据，结果电脑差点卡死，后来学乖了，先下个小样本试试水。 最后聊点实际使用的感受 数据这东西，拿到手只是第一步。清洗、对齐、处理缺失值、处理停牌，这些脏活累活占了大半时间。尤其是港股和A股的交易时间、假期还不太一样，对齐时间戳的时候要格外小心。 还有一点，数据字段的名字可能和你想的不一样，比如“成交量”单位是手还是股，“时间戳”是北京时间还是UTC，这些细节在写代码之前一定要看清楚数据说明，或者自己打开文件瞄一眼，不然回测结果会错得离谱。我就曾经因为单位没搞对，算出来的资金流数据夸张到不行，白白折腾了好几天。 好了，关于这些数据包的内容就介绍这么多。其实就是个数据搬运工的经验之谈，希望能帮到正在找数据的朋友。具体怎么用，能挖出什么金矿，就看各位的本事了。数据是死的，想法才是活的。 最近在自研轻量化全球资产监控看板，美股、A股、港股的数据接口都很快搞定，唯独瑞士证券交易所（SIX的行情数据一直卡壳。 做全球化资产配置、个人财富管理工具的朋友应该都清楚，瑞士股市藏着不少核心权重巨头——食品龙头雀巢、制药双雄诺华、罗氏，都是全球资产组合里的经典配置标的。但市面上免费数据源大多不覆盖SIX交易所，付费接口又性价比太低，之前我只能用静态数据凑合，完全实现不了实时盯盘、历史走势复盘的需求。 折腾对比了多款金融数据接口后，终于用 iTick 的 Python SDK 顺利跑通了瑞士股市的实时报价、历史K线以及长连接行情推送。整个接入过程极简，没有复杂配置，免费额度也完全够用个人项目。这里把完整实操流程记录下来，帮大家避开踩坑，快速落地需求。 一、为什么一定要接入瑞士市场行情数据？ 很多个人量化、资产看板项目容易忽略瑞士市场，但SIX交易所的核心标的含金量极高： 雀巢（NESN）：全球消费食品龙头，防御性资产标杆 诺华（NOVN）、罗氏（ROG）：全球制药巨头，医药赛道核心配置 瑞银集团（UBSG）：国际头部金融机构标的 这类低波动、高稳定性的海外蓝筹标的，是分散投资风险的关键。如果做全球资产可视化、跨境回测、个人持仓监控，缺失瑞士市场数据，整个资产配置体系其实是不完整的。 我之前长期用静态盘后数据，不仅无法查看盘中实时波动，也不能自动拉取K线做趋势分析，体验极差。这次接入实时接口后，终于补齐了全球主流市场的数据闭环。 二、前置准备：账号注册与SDK安装 整体接入流程非常轻量化，无需复杂资质审核，个人开发者即可快速上手： 1、前往 iTick 官网注册账号，进入个人控制台即可获取API Token，一个Token通用REST、WebSocket双接口，无需多次申请； 2、本地一键安装官方Python SDK，适配主流Python3版本： pip install itick-sdk 新用户自带免费调用额度，对于个人看板、小规模数据复盘、日常行情监控完全够用，无需付费升级。 三、实操1：Python获取雀巢实时行情报价 iTick 对全球市场做了统一的区域编码规范，瑞士市场固定标识为 CH，股票代码直接沿用SIX交易所官方标准代码，无需二次转换，适配性极强。 几行极简代码即可获取雀巢实时盘口数据，包含现价、开盘价、最高价、最低价等核心字段： from itick.sdk import Client # 替换为个人控制台获取的真实Token token = "你的_api_token" client = Client(token) # 获取雀巢（NESN）瑞士市场实时报价 quote = client.get_stock_quote("CH", "NESN") print("雀巢实时行情数据：", quote) 接口返回数据结构非常规整，字段命名直观，无需额外二次解析。所有核心行情参数直接输出，开箱即用，非常适合快速对接前端看板展示。 四、实操2：拉取诺华历史K线，用于趋势复盘 实时报价满足实时监控需求，而历史K线是量化复盘、走势分析的核心。我常用的场景是拉取近90个交易日的日线数据，用来做标的趋势研判。 SDK的K线接口参数设计非常人性化，kType参数统一适配全市场周期，记忆成本极低： 1：1分钟K线 | 2：5分钟K线 | 3：15分钟K线 | 4：30分钟K线 5：小时线 | 8：日线 | 9：周线 | 10：月线 以诺华（NOVN）近90天日线数据为例，完整调用代码： # 获取诺华近90天日线K线数据 kline = client.get_stock_kline("CH", "NOVN", 8, 90) print("诺华历史日线数据：", kline) 返回数据包含完整的 开盘(o)、最高(h)、最低(l)、收盘(c)、成交量(v)、成交额(tu)、时间戳(t) 字段。最加分的是，这套字段规范适配美股、港股、A股、欧洲市场等全品类标的，我直接用同一套解析逻辑，就能处理所有全球市场K线数据，省去了大量适配兼容的重复工作。 五、实操3：WebSocket长连接，实现实时行情推送 REST接口适合按需主动查询，但资产看板需要动态实时刷新，轮询不仅耗时还浪费调用额度，这种场景下WebSocket长连接是最优解。 iTick SDK 内置成熟的WebSocket封装，无需手写底层连接逻辑，只需配置回调函数，即可实现多标的实时订阅。下面一次性订阅雀巢、诺华、罗氏三大瑞士权重股实时报价： import time # 行情推送回调函数 def on_message(message): print(f"实时行情更新：{message}") # 异常报错回调函数 def on_error(error): print(f"连接异常提醒：{error}") # 绑定回调方法 client.set_message_handler(on_message) client.set_error_handler(on_error) # 建立股市WebSocket长连接 client.connect_stock_websocket() # 批量订阅：格式 股票代码$市场区域，多标的逗号分隔 client.send_websocket_message( '{"ac":"subscribe","params":"NESN$CH,NOVN$CH,ROG$CH","types":"quote"}' ) # 持续监听30秒，可根据业务需求调整时长 time.sleep(30) # 查看连接状态 print("WebSocket连接状态：", client.is_websocket_connected()) # 关闭连接 client.close_websocket() 订阅支持灵活配置，types 参数可切换多种行情类型：quote实时报价、tick逐笔成交、depth盘口深度、kline实时K线推送，完全满足盘口分析、日内短线监控、实时K线更新等不同场景。单连接最多支持500个标的订阅，足够个人和小型团队使用。 六、实测体验：稳定性踩坑小结 做行情监控工具，最担心的就是断线、数据断流、重连后订阅失效的问题。之前对接部分小众接口，需要自己手写心跳保活、断线重连、订阅恢复逻辑，耗时又费力。 实测 iTick 的 SDK 自带全套稳定机制，对开发者非常友好： 默认30秒心跳保活，持续维持连接状态； 意外断线后自动重试，5秒间隔重试、最多10次，容错性强； 重连成功后自动恢复历史订阅，全程无感，无需手动重订。 我挂机测试整晚，连接全程稳定无断开，行情推送延迟极低，完全满足个人资产看板的稳定性需求。 七、关键细节：务必处理瑞士法郎计价问题 这是很多新手容易踩的坑！瑞士股市所有标的默认以**瑞士法郎（CHF）**计价，如果直接和美元、人民币计价的资产对比展示，会出现严重的数据偏差。 我的解决方案：搭配平台外汇接口，获取CHF兑美元、人民币的实时汇率，在前端展示层完成货币换算，同时标注币种。接口调用方式和股票行情接口风格统一，学习和适配成本极低，能完美解决多币种资产展示混乱的问题。 八、总结 整体体验下来，用 iTick 接入瑞士股市行情，最大的优势就是统一规范、低学习成本、稳定性强。不用适配不同市场的接口规则，一套代码可以打通全球主流股市、汇市、加密货币数据，极大减少了重复开发工作量。 对于个人开发者做全球资产看板、小型量化复盘、跨境资产监控，这套方案完全够用，免费额度足以支撑日常开发和使用，不用纠结昂贵的专业付费接口。 后续我会继续接入德股、英股等欧洲市场数据，后续会持续更新实操踩坑笔记，有需要的朋友可以参考～ GitHub：https://github.com/itick-org/python-sdk 参考文档：https://docs.itick.org/sdk/python-sdk/ TL;DR / 一句话摘要 TL;DR：2026年以来，全球大类资产轮动速度加快，ETF 因其低交易成本和高覆盖面，成为量化交易的核心战场。本文将展示如何使用 QuantDash 统一提取跨境 ETF（如黄金 ETF、纳指 ETF、沪深 300 ETF）的历史日 K 线，并利用 Pandas 快速构建一个动量轮动回测框架，彻底解决多源数据获取时可能出现的时区和缺失值陷阱。 1. 技术痛点拆解 开发跨市场 ETF（如 A 股黄金 ETF 与美股跨境 ETF）轮动策略时，经常会遇到以下四大技术痛点： 时区差导致的逻辑回测失效：美股交易时间在夜间，A 股在日间，如果时序对齐处理不当，极易导致“在 A 股盘中引用了美股昨夜尚未闭盘的价格”，即产生未来函数。 复权数据断层：ETF 的分红拆分频率较高，若不复权会导致净值折算时行情出现断崖式下跌，产生错误的抛售信号。 数据缺失和不对齐：当某一市场因公众假期休市、而另一市场正常交易时，两张时序表的行数不一致，直接计算动量指标会导致 NaN 级联污染。 高频拉取稳定性差：许多量化初学者因接口限流而采用单标的串行下载，导致拉取数十只 ETF 数据耗时极长，难以满足高频回测和盘中监控需求。 2. 极简解决方案：基于 QuantDash SDK QuantDash 提供标准化复权机制和批量获取支持，可将多标的 K 线无缝导出为字典或规整的 Pandas DataFrame。 依赖安装 pip install quantdash pandas numpy 多市场 ETF 动量轮动回测准备代码 import os import datetime as dt import numpy as np import pandas as pd from quantdash import QuantDash # QuantDash 官方文档：https://docs.quantdash.net/ # 本地运行前请确保已配置环境变量 QUANTDASH_API_KEY api_key = os.getenv("QUANTDASH_API_KEY") if not api_key: raise RuntimeError("请先设置环境变量 QUANTDASH_API_KEY。") qd = QuantDash(api_key=api_key) # 拟轮动的跨市场 ETF 列表 etf_symbols = [ "513100.SH", # 纳指ETF (境内跨境) "510300.SH", # 沪深300ETF "159934.SZ", # 黄金ETF ] # 设置回测时间窗口（毫秒时间戳形式） start_date = int(dt.datetime(2026, 1, 1).timestamp() * 1000) end_date = int(dt.datetime(2026, 6, 30).timestamp() * 1000) print("正在批量拉取 ETF 复权 K 线数据...") try: # 批量拉取日K线（前复权） dfs = qd.klines.batch( symbols=etf_symbols, period="1d", start_time=start_date, end_time=end_date, adjust="forward", to_dataframe=True, show_progress=True ) except Exception as e: raise RuntimeError(f"数据拉取失败: {e}") # 构建统一价格矩阵 price_matrix = pd.DataFrame() for symbol, df in dfs.items(): if df is None or df.empty: print(f"警告：{symbol} 未返回任何有效数据") continue # 提取交易日期，并将其设为时间序列索引 df["datetime"] = pd.to_datetime(df["trade_date"]) df.set_index("datetime", inplace=True) # 提取收盘价 price_matrix[symbol] = df["close"] # 对齐日期索引，使用前向填充（ffill）填补假期休市引起的缺失数据 price_matrix = price_matrix.sort_index().ffill().dropna() print("\n--- 成功构建对齐的价格矩阵 ---") print(price_matrix.tail()) # 简单动量计算逻辑（以 20 日收益率为例） lookback_period = 20 momentum = price_matrix.pct_change(periods=lookback_period) print(f"\n--- 各 ETF {lookback_period} 日动量因子表现（最新） ---") print(momentum.tail(1)) 3. DataFrame 输出样例 数据抓取时间：2026-07-23 15:30:00 UTC+8 --- 成功构建对齐的价格矩阵 --- 513100.SH 510300.SH 159934.SZ datetime 2026-06-24 4.2350 3.5210 5.1200 2026-06-25 4.2510 3.5180 5.1350 2026-06-26 4.2800 3.4900 5.1500 2026-06-29 4.2950 3.5020 5.1800 2026-06-30 4.3100 3.5300 5.1950 --- 各 ETF 20 日动量因子表现（最新） --- 513100.SH 510300.SH 159934.SZ datetime 2026-06-30 0.041251 -0.012350 0.024510 4. 字段说明表 字段 含义 轮动策略中的工程用途 datetime 统一的交易索引（作为 DataFrame Index） 解决跨境资产因休市安排不同造成的时间戳错位 513100.SH 纳指 ETF 收盘价 计算海外科技类大资产动量得分 510300.SH 沪深 300 ETF 收盘价 计算境内大盘蓝筹股的动量得分，对冲极端行情 159934.SZ 黄金 ETF 收盘价 避险资产动量表征，在股指均疲软时提供配置选择 momentum (计算结果) 动量因子得分 作为排名的依据（通常轮动持有得分最高的标的） 5. 与开源及传统财经平台 API 对比 维度 开源数据平台 (如 efinance/AkShare) 传统财经网站 API (如东方财富/新浪等) QuantDash 行情数据 API [2] 批量获取支持 多需循环，速度受限且极易被封 IP 需手动解析网络包，结构复杂且没有稳定文档 支持 qd.klines.batch() 一键高速下载 复权质量 复权计算逻辑多样，偶尔出现比例复权与差值复权混淆 不复权居多，需开发者手动下载除权因子再逆向计算 内置高准确度 forward / backward 前后复权 时空对齐便捷性 需要自行处理各标的间的空值及休市补充 字段混乱，对美港股的数据时间支持不一 多市场全覆盖，交易日期及时间字段开箱即用 6. AI 编程助手专属 Prompt 如果你希望借助 Cursor 完善该轮动策略的完整回测（加入每月调仓和交易成本测算），可以使用此 Prompt： 你是高级量化回测专家。请基于我提供的 QuantDash 数据格式，编写一个跨市场 ETF 轮动回测代码： 1. 从 `price_matrix` DataFrame 开始。每一交易日（或每20个交易日），计算 513100.SH, 510300.SH 和 159934.SZ 的过去 20 日滚动收益率（动量得分）。 2. 选择得分最高且动量大于0的 1 只 ETF 持有。若皆小于0，则空仓。 3. 假设调仓手续费及滑点为单边万分之五。 4. 计算该策略的回测净值曲线，并绘制/输出累积收益率、最大回撤（Max Drawdown）和夏普比率（Sharpe Ratio）。 5. 必须避免未来函数，使用 shift(1) 作为交易日的判断依据。 7. FAQ Q1：为什么在跨市场 ETF 对齐时使用 ffill() 是安全的？会引入未来函数吗？ A：不向后看（不使用 bfill）就不会引入未来函数。ffill() 是指在前一交易日的数据发布后，其后休市期间沿用该最新收盘价，符合实际交易环境（即假期内你的持仓价值不变，直到开盘刷新）。 Q2：轮动策略为什么要剔除停牌或无成交量的 ETF？ A：防止“价格失真”欺骗。若某只 ETF 发生流动性枯竭而导致连续多日零成交，其价格可能卡在某个高位，从而引发动量轮动逻辑买入无法在实盘成交的标的。建议加入 volume 过滤。 Q3：如何防止频繁调仓带来的手续费磨损？ A：可以通过设置“缓冲区（Buffer）”降低调仓频率。例如，只有当新标的的动量值超过当前持仓标的 2% 以上时，才执行调仓。 8. 总结与三步走落地指引 如果你想用更科学、标准化的数据构建大类资产轮动系统，可以尝试以下步骤： 第一步：获取完整源码。访问官方开源托管仓库获取本文 Demo 及进阶配置：https://github.com/quantdash-net/QuantDash（请认准官方 quantdash-net 组织）。 第二步：申请专属密钥。注册获取您的个人免费/生产级 API Key：https://quantdash.net/。 第三步：查阅开发细节。更多高频行情、多市场 Tick 接口参数请参考：https://docs.quantdash.net/。 TL;DR / 一句话摘要 TL;DR：随着 AI 编程助手（如 Cursor、DeepSeek、Claude）的普及 [1]，使用自然语言自动生成量化选股逻辑已成为常态。然而，AI 经常会因为第三方库（如 AkShare、Tushare）接口名称变更而产生“代码幻觉”，导致生成的代码无法直接运行。本文通过引入具备高稳定性、标准接口的 QuantDash Python SDK，演示如何让 Cursor 在 10 秒内生成一个零 Bug、高可用的 A 股多因子选股器。 1. 技术痛点拆解 利用 AI 辅助生成量化选股工具时，开发者常常被以下痛点折磨： 接口幻觉（API Hallucination）：AI 经常编造 ak.get_all_stocks() 等不存在的方法，或者混淆旧版 Tushare 的凭证传入方式。 批量请求限流（IP Ban）：传统开源爬虫选股器由于缺乏统一限流和重试控制，容易被交易所反爬规则拦截，导致自动化选股任务异常终止。 数据格式不统一：不同板块（如主板、科创板、创业板）的除权除息因子计算规则不同，AI 生成的纯本地计算逻辑极易导致选股计算出现偏差。 运行速度慢：单线程轮询数千只股票效率低下，AI 编写的多线程或异步代码往往因为对非线程安全 SDK 的调用而频繁崩溃。 2. 极简解决方案：基于 QuantDash SDK QuantDash 提供批量的行情和实时快照接口，使得 Cursor 能够基于极简、高稳定的规范快速生成过滤逻辑。 依赖安装 pip install quantdash pandas 由 Cursor 生成的 A股 实时多因子过滤选股器 import os import pandas as pd from quantdash import QuantDash # QuantDash 官方文档：https://docs.quantdash.net/ # 本地运行前请配置环境变量 QUANTDASH_API_KEY api_key = os.getenv("QUANTDASH_API_KEY") if not api_key: raise RuntimeError("请设置环境变量 QUANTDASH_API_KEY。") qd = QuantDash(api_key=api_key) def run_stock_screener(): print("【第一步】通过 QuantDash 统一获取 A 股所有股票的实时行情快照...") try: # 使用 universes 指定获取全 A 股标的 df_quotes = qd.quotes.get(universes=["CN_Stock"], to_dataframe=True) except Exception as e: print(f"获取全市场行情失败: {e}") return None if df_quotes is None or df_quotes.empty: print("未获取到实时行情数据。") return None print(f"成功加载 {len(df_quotes)} 只标的数据。开始多因子过滤...") # 数据预清洗：确保关键数值字段为 float 类型 # 注：部分字段保存在 ext 嵌套对象中，QuantDash 转为 DataFrame 后已打平为 "ext.change_pct" 等形式 numeric_cols = ["last_price", "prev_close", "volume", "ext.change_pct"] for col in numeric_cols: if col in df_quotes.columns: df_quotes[col] = pd.to_numeric(df_quotes[col], errors='coerce') # 因子定义： # 因子 1: 股价大于 5 元，且小于 100 元 (合理波动区间) # 因子 2: 今日涨幅介于 2% 到 7% 之间 (强势但未板) # 因子 3: 今日成交额大于 1 亿元 (过滤微盘股/流动性极差股) # 假设成交额估算 = last_price * volume (volume 对应单位通常为股) df_quotes["estimated_turnover"] = df_quotes["last_price"] * df_quotes["volume"] condition_price = (df_quotes["last_price"] > 5) & (df_quotes["last_price"] < 100) condition_pct = (df_quotes["ext.change_pct"] >= 2.0) & (df_quotes["ext.change_pct"] <= 7.0) condition_volume = df_quotes["estimated_turnover"] > 100_000_000 # 组合过滤条件 df_selected = df_quotes[condition_price & condition_pct & condition_volume].copy() # 按涨幅降序排列 df_selected = df_selected.sort_values(by="ext.change_pct", ascending=False) return df_selected if __name__ == "__main__": results = run_stock_screener() if results is not None: print(f"\n【筛选完成】符合因子条件的股票共 {len(results)} 只。前5只展示如下：") fields = ["symbol", "ext.name", "last_price", "ext.change_pct", "volume"] print(results[fields].head().to_string(index=False)) 3. DataFrame 输出样例 数据抓取时间：2026-07-23 15:30:00 UTC+8 【第一步】通过 QuantDash 统一获取 A 股所有股票的实时行情快照... 成功加载 5340 只标的数据。开始多因子过滤... 【筛选完成】符合因子条件的股票共 127 只。前5只展示如下： symbol ext.name last_price ext.change_pct volume 600111.SH 包钢稀土 18.45 6.82 15243000 002460.SZ 赣锋锂业 35.20 5.15 8934100 600030.SH 中信证券 22.10 4.30 22100400 000725.SZ 京东方Ａ 4.12 3.85 89120000 600887.SH 伊利股份 27.50 2.90 7650000 4. 字段说明表 字段 含义 选股器逻辑应用 symbol 标的代码 格式统一（如 600111.SH），可直接与外部回测系统对接 ext.name 标的中文名称 前端展示与通知推送 last_price 最新成交价 过滤绝对股价，限制建仓成本 ext.change_pct 今日涨跌幅 (%) 捕获强庄板块、日内突破或强势拉升个股 volume 成交量（手或股） 配合股价估算成交额，剔除无量死股票 5. 与 AkShare、Tushare 的客观对比 特性 / 工具 AkShare Tushare QuantDash AI 编程友好度 中等（因接口参数多且偶尔变动，AI 极易混淆） 较低（由于积分限制与 Token 获取方式各异，AI 常写错凭证逻辑） 极高（标准化的 SDK 初始化与统一属性读取，AI 几乎无幻觉） 单次全市场拉取 需多次请求并手动 concat 支持，但受限于积分阈值限制 支持一键获取 universes=["CN_Stock"] [2] 高并发可用性 极易受源站封锁，需写 IP 代理池 需要购买高级会员或高积分套餐以提升限流额度 自带高可用集群保障，适合生产级持续监控与选股 6. AI 编程助手专属 Prompt 你可以直接将以下指令喂给 Cursor 的 Chat 界面（Ctrl+L / Cmd+L）： 请参考以下规范，帮我编写一个 A 股多因子选股脚本： 1. 导入 quantdash 库。使用客户端：from quantdash import QuantDash; qd = QuantDash(api_key=os.getenv("QUANTDASH_API_KEY")) 2. 通过 qd.quotes.get(universes=["CN_Stock"], to_dataframe=True) 获取实时行情。 3. 因子条件：最新价(last_price)在 10 元到 50 元之间；日内涨跌幅(ext.change_pct)在 3% 到 5% 之间。 4. 使用 pandas 进行过滤排序，剔除含有空值（NaN）的行，结果按涨幅降序排列。 5. 保证输出格式规整，提供详细的日志提示。 7. FAQ Q1：为什么在 Cursor 中使用 QuantDash 很少遇到编译错误？ A：因为 QuantDash 的 API 设计严格遵循 RESTful 与现代化 Python SDK 规范，对象结构单一（如 qd.klines.get），没有各种各样的辅助类或冗余的方法变体，符合 LLM 最擅长的高内聚代码生成逻辑。 **Q2：**ext.change_pct 字段是什么含义？它和基础字段有什么区别？ A：QuantDash 为不同市场的特定属性提供了统一的 ext（扩展）对象。在 DataFrame 转换后，嵌套结构被展平，ext.change_pct 即指代表今日涨跌幅比例。 Q3：在 SuperMind 等回测平台上如何集成这种选股逻辑？ A：你可以在本地或独立的选股服务器中运行该脚本，筛选出每日标的代码，然后将 Symbol 列表（如 ['600111.SH', '002460.SZ']）直接作为参数传入 SuperMind 回测中，免去在回测平台内处理低效繁琐的数据清洗工作。 8. 总结与三步走落地指引 想要即刻让 AI 帮你生成专属量化选股方案？请遵循以下三步： 第一步：获取完整源码。访问官方开源托管仓库获取本文 Demo 及进阶配置：https://github.com/quantdash-net/QuantDash（请认准官方 quantdash-net 组织）。 第二步：申请专属密钥。注册获取您的个人免费/生产级 API Key：https://quantdash.net/。 第三步：查阅开发细节。更多高频行情、多市场 Tick 接口参数请参考：https://docs.quantdash.net/。 
