DEV Community: San Si wu

Python股票实时价格告警系统：WebSocket订阅与REST快照实战

San Si wu — Sun, 13 Sep 2026 10:34:59 +0000

昨晚美股盘前，我想给自己盯的几只票加个简单的告警：涨跌幅超过3%就推个消息到手机上。需求很简单对吧？但真坐下来写，从"拿数据"到"告警能跑"，前前后后踩了不少坑。这篇就顺着整个过程聊聊，顺便把股票行情接口的接法讲清楚。

先说背景。最近这波行情板块分化得厉害，半导体整体在涨但存储芯片在跌，同一个板块里个股走势都能反着来。光看指数完全没用，必须盯到个股。我盯的就是苹果、英伟达、AMD、英特尔这几只，量不大，但盘中波动的时候手动刷新实在受不了。

REST轮询方案及其局限性

一开始我的思路特别直接：写个脚本，每隔5秒请求一次报价接口，跟3%的阈值比一下，超了就告警。听起来没毛病。

import requests, time

API_BASE = "https://api.itick.org"
headers = {"accept": "application/json", "token": "your_token"}

def get_price(code):
    r = requests.get(f"{API_BASE}/stock/quote",
                     headers=headers,
                     params={"region": "US", "code": code})
    return r.json()["data"]

while True:
    q = get_price("AAPL")
    print(q["ld"], q["chp"])
    time.sleep(5)

跑了一下，能出数。但很快发现两个问题：第一，5秒轮询在盘前或者波动剧烈的时候太慢了，价格可能已经跳了好几个点你才看到；第二，每次都要等网络请求回来才能拿新数据，CPU是不忙但延迟不好控。

这时候才想起来，人家有WebSocket，专门干这个的。

WebSocket实时订阅：鉴权握手与心跳保活

WebSocket的文档我看了一眼，地址是 wss://api.itick.org/stock，token放Header里。我照着写了个最小连接：

import websocket, json

ws = websocket.WebSocketApp(
    "wss://api.itick.org/stock",
    header=["token: your_token"],
    on_open=lambda ws: ws.send(json.dumps({
        "ac": "subscribe",
        "params": "AAPL$US",
        "types": "quote"
    })),
    on_message=lambda ws, msg: print(msg),
)
ws.run_forever()

结果一跑，服务端回了个 cannot be resolved action。我当时就懵了——格式看着跟文档一样啊。

翻了文档才看到那个容易被忽略的细节：连接建立成功和可以订阅是两回事。服务端先回一句 Connected Successfully，这时候你什么都不能做；必须等它再推一条 resAc: "auth" 且 code: 1 的消息，才代表鉴权通过，这时候发订阅才有效。

我之前写了太多"连上就能用"的WebSocket接口，差点忘了这个服务端是要做鉴权握手的。改了一下：

authenticated = False

def on_open(ws):
    print("连接已建立")  # 注意：这时候还不能订阅！

def on_message(ws, message):
    global authenticated
    payload = json.loads(message)

    # 必须等这条鉴权成功消息
    if payload.get("resAc") == "auth" and payload.get("code") == 1 and not authenticated:
        authenticated = True
        ws.send(json.dumps({
            "ac": "subscribe",
            "params": "AAPL$US,NVDA$US",
            "types": "quote"
        }))
        return

这下终于收到数据了。但没高兴两秒，连接过了一分钟就断了。

又是文档里写了但我没仔细看的：这个服务端要求30秒发一次心跳，超过1分钟不发就踢人。加个线程专门发ping：

import threading, time

def heartbeat(ws):
    while True:
        time.sleep(30)
        ws.send(json.dumps({
            "ac": "ping",
            "params": str(int(time.time() * 1000))
        }))

# 鉴权成功后启动
threading.Thread(target=heartbeat, args=(ws,), daemon=True).start()

到这一步，实时报价终于稳定跑起来了。

返回字段解析与交易状态处理

数据能收到了，但解析的时候又踩了坑。返回的JSON长这样：

{
  "code": 1,
  "data": {
    "s": "AAPL",
    "ld": 225.215,
    "o": 226.27,
    "p": 226.27,
    "h": 226.92,
    "l": 224.44,
    "ch": -3.24,
    "chp": -1.17,
    "v": 16742235,
    "ts": 0,
    "type": "quote"
  }
}

全是缩写。我第一反应是 ld 是什么？chp 又是什么？翻文档才搞清楚：

ld = 最新价（last price）
p = 前收盘价（previous close）
ch = 涨跌额（change）
chp = 涨跌幅百分比（change percent）
ts = 交易状态，0正常、1停牌、2退市、3熔断

这里有个特别坑的地方：我一开始算涨跌幅的时候，偷懒用了 (ld - o) / o，也就是拿最新价跟开盘价比。跑出来发现告警触发得特别频繁——因为开盘后前几分钟波动本来就大。后来才反应过来，涨跌幅应该跟前收盘比，服务端已经在 chp 里算好了，直接用就行。

还有一个 ts 字段。有次我挂着告警跑了一晚上，第二天发现某只停牌的票一直在触发告警——因为 ld 停留在昨天的价格，而我的脚本没有检查 ts。加个判断就好了：

if data.get("ts") == 1:
    continue  # 停牌，跳过

REST历史K线查询与技术指标计算

告警的事搞完了，又想多做一步：打开脚本的时候先显示一下这几只票近30天的走势，让我对当前位置有个感觉。这就需要拉历史K线。

K线接口跟报价是分开的，路径是 /stock/kline，注意是单数 stock 不是复数 stocks——这个我也踩了，一开始写 /stocks/kline 直接404。

def get_kline(region, code, k_type=8, limit=30):
    """
    k_type: 8=日K
    其他: 1=1分 2=5分 5=1时 9=周 10=月
    """
    r = requests.get(
        f"{API_BASE}/stock/kline",
        headers=headers,
        params={
            "region": region,
            "code": code,
            "kType": k_type,
            "limit": limit
        }
    )
    return r.json()["data"]

klines = get_kline("US", "AAPL", k_type=8, limit=30)
for bar in klines[-5:]:
    # 每根K线: t=时间戳 o/h/l/c/v/tu
    print(bar["t"], bar["o"], bar["c"])

返回的是一个数组，每根K线就七个字段，干净利落。用这个算个简单的均线或者波动率，比从行情软件截图方便多了。

完整告警系统：REST初始化与WebSocket增量推送

把REST初始化和WebSocket拼一起，就是最终能跑的告警脚本：

import json, threading, time, requests, websocket

API_BASE = "https://api.itick.org"
WS_URL = "wss://api.itick.org/stock"
TOKEN = "your_token_here"
headers = {"accept": "application/json", "token": TOKEN}

watchlist = ["AAPL", "NVDA", "AMD", "INTC"]
THRESHOLD = 3.0  # 涨跌幅超过3%告警

# 启动时先拉一遍快照做基线
print("=== 盘前快照 ===")
for code in watchlist:
    q = requests.get(f"{API_BASE}/stock/quote",
                      headers=headers,
                      params={"region": "US", "code": code}).json()["data"]
    print(f"{code}: 前收={q['p']} 最新={q['ld']}")

# WebSocket盯盘
authenticated = False

def on_message(ws, message):
    global authenticated
    payload = json.loads(message)

    if payload.get("resAc") == "auth" and payload.get("code") == 1 and not authenticated:
        authenticated = True
        ws.send(json.dumps({
            "ac": "subscribe",
            "params": ",".join(f"{c}$US" for c in watchlist),
            "types": "quote"
        }))
        threading.Thread(target=heartbeat, args=(ws,), daemon=True).start()
        return

    data = payload.get("data") or {}
    if data.get("type") != "quote":
        return
    if data.get("ts") != 0:  # 非交易状态不告警
        return

    chp = data.get("chp", 0)
    if abs(chp) >= THRESHOLD:
        print(f"[告警] {data['s']}: 涨跌幅 {chp}% 超过阈值!")

def heartbeat(ws):
    while True:
        time.sleep(30)
        ws.send(json.dumps({"ac": "ping", "params": str(int(time.time() * 1000))}))

ws = websocket.WebSocketApp(WS_URL, header=[f"token: {TOKEN}"], on_message=on_message)
ws.run_forever()

跑起来之后，开盘前先看到一组基线，盘中有股票涨跌超过3%就立刻打印告警。整个过程没用到任何框架，依赖就两个。

生产部署注意事项

回头看，整个接入过程的技术点其实不多，但容易踩的坑全在细节里：连接不等于鉴权、必须等auth消息才能订阅、30秒心跳不能忘、字段全是缩写别想当然、交易状态要检查。这些东西文档里都写了，但第一次看的时候很容易扫过去，真跑起来才会一个个撞上。

做监控工具这种东西，稳定性比功能多重要。我现在挂着这个脚本跑了一阵，WebSocket断了会自动重连（websocket-client的run_forever自带），告警逻辑也加了去重，基本上不用管了。

最近这行情波动大，有个自己写的监控脚本比来回切行情软件省心多了。接口文档可以在这里看，有其他市场的接入方式也类似。官网地址https://itick.org

Python实时外汇行情接入实战：WebSocket与REST K线查询

San Si wu — Fri, 11 Sep 2026 20:34:44 +0000

最近外汇市场的消息面非常密集。欧洲央行刚结束议息会议，如期加息25个基点，把存款机制利率推到2.50%，同时上调了全年通胀预期——能源价格上涨是背后的主要推手。另一边，美国CPI数据刚落地，虽然整体符合预期，但核心通胀环比略超市场预期，美联储下周议息会议的加息概率从六成跳到了七成以上。

这两件事凑在一起，欧元兑美元的波动明显加大。ECB加息当天欧元不涨反跌，因为市场注意力立刻转向"美联储会不会跟着加"——如果两边都加，利差不变，欧元自然没有方向性行情。这种央行政策博弈期，汇率价格跳动快、消息面密集，做行情看板或者交易辅助工具的开发者，对实时数据接口的需求就特别迫切。

这篇文章从技术角度聊聊，怎么用Python从零接入一套实时外汇行情——包括WebSocket实时报价推送、REST历史K线查询，以及把两者组合起来做一个轻量监控脚本。不依赖任何量化框架，标准库加 requests 和 websocket-client 就能跑。

为什么要分开用WebSocket和REST

很多初学者一开始会问：直接定时轮询REST不行吗，为什么还要搞WebSocket？

实际用过就知道差别很大。像最近ECB决议和美联储议息这种事件，汇率反应往往在几秒钟内完成。用REST轮询，哪怕你每5秒请求一次，也会错过中间的波动轨迹。WebSocket适合盯实时跳动——价格一变服务端就推过来，延迟在毫秒级，做盘中监控和事件触发很顺手。

但WebSocket不适合拉历史数据，你不能让它"把过去90天的日K线给我"。REST反过来，拉历史K线、查最新快照都很方便，但频繁请求既浪费带宽又有速率限制。

所以正确的用法是：REST拉历史数据做初始化和回测，WebSocket订阅实时数据做增量更新。下面分别说。

REST：拉取历史K线做初始化

假设我们要做一个欧元兑美元的行情图表，首先需要拉一段历史日K线作为底图，观察近期的走势结构。接口的结构很规整：基础地址是 https://api.itick.org，外汇相关的REST路径都在 /forex/ 下面。

认证用HTTP Header token，不是常见的 X-API-Key 或 Authorization: Bearer，这个坑要注意。请求参数是 region 加 code 的组合——region 是市场代码（外汇用 GB），code 是货币对代码（如 EURUSD）。K线周期用 kType 数字编码：8代表日K，1代表1分钟，5代表1小时。

import requests

API_BASE = "https://api.itick.org"
TOKEN = "your_token_here"
headers = {
    "accept": "application/json",
    "token": TOKEN
}


def get_forex_kline(region, code, k_type=8, limit=90):
    """
    获取外汇历史K线
    region: 市场代码，外汇固定用 GB
    code: 货币对代码，如 EURUSD、GBPUSD、USDJPY
    k_type: K线周期编码
            1=1分钟  2=5分钟  3=15分钟  4=30分钟
            5=1小时  8=日K    9=周K     10=月K
    limit: 返回条数，最大500
    """
    resp = requests.get(
        f"{API_BASE}/forex/kline",
        headers=headers,
        params={
            "region": region,
            "code": code,
            "kType": k_type,
            "limit": limit
        }
    )
    resp.raise_for_status()
    return resp.json()["data"]


# 拉取EUR/USD近90根日K线
kline = get_forex_kline("GB", "EURUSD", k_type=8, limit=90)

# 返回的每根K线字段是缩写
# t=时间戳(毫秒) o=开 h=高 l=低 c=收 v=成交量 tu=成交额
latest = kline[-1]
print(f"最新日K: 开={latest['o']} 高={latest['h']} "
      f"低={latest['l']} 收={latest['c']} 量={latest['v']}")

这里有个容易踩的坑：返回字段名是缩写的。价格字段用 o/h/l/c，成交量是 v，时间戳是 t（毫秒级）。不是 open/high/low/close/volume 那种长名。写代码的时候先打印一次返回值确认字段，别猜。

拉完日K线之后，你可以自己算一下近期的波动幅度。比如EUR/USD最近在1.16附近来回震荡，如果你用日K的最高价减最低价再除以收盘价，算出来的日均波幅大概在0.4%左右——这个数字对你后面设计告警阈值很有用。

WebSocket：订阅实时报价

历史K线拉完之后，需要把最新价格实时推过来。像下周美联储议息这种事件，价格跳动可能在几分钟内完成，用REST轮询完全跟不上。外汇WebSocket的地址是 wss://api.itick.org/forex，跟REST同一个域名，只是协议换成了WebSocket。

连接的时候把token放在HTTP Header里：

import json
import threading
import time
import websocket

WS_URL = "wss://api.itick.org/forex"
TOKEN = "your_token_here"

authenticated = False
heartbeat_timer = None


def start_heartbeat(ws):
    """服务端要求每30秒发一次心跳，否则可能断开"""
    def ping():
        while True:
            time.sleep(30)
            ws.send(json.dumps({
                "ac": "ping",
                "params": str(int(time.time() * 1000))
            }))
    global heartbeat_timer
    heartbeat_timer = threading.Thread(target=ping, daemon=True)
    heartbeat_timer.start()


def subscribe(ws):
    """
    订阅格式：{"ac": "subscribe", "params": "EURUSD$GB", "types": "quote"}
    params 用 标的$市场 的格式，多标的用逗号分隔
    types 可选 quote/tick/depth/kline@1 等
    """
    ws.send(json.dumps({
        "ac": "subscribe",
        "params": "EURUSD$GB,GBPUSD$GB",
        "types": "quote"
    }))

WebSocket的状态机有个关键点：不能连上就发订阅。连接刚建立时服务端会先回一条 {"code":1,"msg":"Connected Successfully"}，这时候还不能订阅。必须等到收到鉴权成功消息 {"resAc":"auth","code":1,"msg":"authenticated"} 之后，才能发subscribe。很多人第一次接的时候卡在这里——连接成功了但订阅一直报错，就是因为没等鉴权。

def on_open(ws):
    print("连接已建立，等待鉴权...")


def on_message(ws, message):
    global authenticated
    payload = json.loads(message)

    # 等鉴权成功后再订阅，同时启动心跳
    if payload.get("resAc") == "auth" and payload.get("code") == 1 and not authenticated:
        authenticated = True
        print("鉴权成功，开始订阅行情")
        subscribe(ws)
        start_heartbeat(ws)
        return

    # 心跳响应直接忽略
    if payload.get("resAc") == "pong":
        return

    # 处理实时报价推送
    data = payload.get("data") or {}
    if data.get("type") == "quote":
        # 推送字段和REST一致：s=标的 ld=最新价 o/h/l=当日开高低 t=毫秒时间戳
        symbol = data.get("s")
        price = data.get("ld")
        ts = time.strftime("%H:%M:%S", time.localtime(data.get("t", 0) / 1000))
        print(f"[{ts}] {symbol}: {price}")


ws = websocket.WebSocketApp(
    WS_URL,
    header=[f"token: {TOKEN}"],
    on_open=on_open,
    on_message=on_message,
)
ws.run_forever()

整个流程梳理一下：建立连接 → 等鉴权 → 发订阅 → 收推送 → 持续心跳。顺序不能乱，否则订阅会失败。

如果你想在美联储决议公布那一刻收到提醒，可以在 on_message 里加一个价格变动阈值判断——比如当EUR/USD在10秒内波动超过30个点就触发告警。这种用WebSocket做的实时监控，比轮询方案的响应速度快一个数量级。

REST查最新报价：补一个快照接口

WebSocket是增量推送，但有些场景你只需要一个当前快照，不想维护长连接。比如你写个定时任务，每小时记录一次主要货币对的最新价，这种就没必要一直挂着WebSocket。REST的报价接口刚好合适：

def get_forex_quote(region, code):
    """获取实时报价快照"""
    resp = requests.get(
        f"{API_BASE}/forex/quote",
        headers=headers,
        params={"region": region, "code": code}
    )
    return resp.json()["data"]


# 一次查多个主要货币对
for pair in ["EURUSD", "GBPUSD", "USDJPY"]:
    q = get_forex_quote("GB", pair)
    # ld=最新价 chp=涨跌幅 ch=涨跌额 o/h/l=当日开高低
    print(f"{pair}: 最新={q['ld']} 涨跌幅={q['chp']}%")

报价接口的返回字段比K线多几个：ch 是涨跌额，chp 是涨跌幅（百分比），ts 是交易状态。这些字段做仪表盘显示够用了。

最近EUR/USD就在1.16附近窄幅波动，ECB加息后欧元不但没涨反而小跌，说明市场当前更关注美联储那边的政策预期。这种"利差变化"类的分析，你可以用报价接口同时拉EURUSD和USDJPY，对比看美元指数的整体强弱。

组合起来：一个轻量行情监控脚本

把REST初始化和WebSocket实时推送结合，就是一个最基本的行情监控器——启动时先拉历史K线作为基线，然后订阅实时报价做增量更新。这个模式在央行决议日特别有用：你可以在决议公布前把脚本挂着，价格一动就能立刻看到。

import json
import threading
import time
import requests
import websocket

API_BASE = "https://api.itick.org"
WS_URL = "wss://api.itick.org/forex"
TOKEN = "your_token_here"
headers = {"accept": "application/json", "token": TOKEN}

# 启动时先拉历史日K作为基线
def init_kline(code):
    resp = requests.get(
        f"{API_BASE}/forex/kline",
        headers=headers,
        params={"region": "GB", "code": code, "kType": 8, "limit": 30}
    )
    data = resp.json()["data"]
    recent = [bar["c"] for bar in data[-5:]]
    print(f"{code} 近5日收盘价: {recent}")

# WebSocket部分
authenticated = False
alerts = []  # 可以在这里加自己的告警逻辑

def on_message(ws, message):
    global authenticated
    payload = json.loads(message)
    if payload.get("resAc") == "auth" and payload.get("code") == 1 and not authenticated:
        authenticated = True
        ws.send(json.dumps({
            "ac": "subscribe",
            "params": "EURUSD$GB",
            "types": "quote"
        }))
        return
    data = payload.get("data") or {}
    if data.get("type") == "quote":
        price = data['ld']
        # 这里可以加自己的告警判断，比如价格突破某个区间
        print(f"实时更新 {data['s']}: {price}")

init_kline("EURUSD")
ws = websocket.WebSocketApp(
    WS_URL,
    header=[f"token: {TOKEN}"],
    on_message=on_message,
)
ws.run_forever()

这个脚本虽然简单，但骨架已经完整：REST负责历史数据初始化，WebSocket负责实时增量，两者字段格式统一，切换的时候不需要做数据格式转换。

几个实际踩过的坑

Header字段名是 token，不是 X-API-Key 也不是 Authorization。我第一次接的时候用了 X-API-Key，一直返回鉴权失败，翻了文档才发现。

必须等鉴权成功再订阅。刚连上就发subscribe会被服务端拒绝，错误消息是 cannot be resolved action。用 resAc == "auth" 判断鉴权状态最靠谱。

K线周期是数字编码，不是字符串 "1d" 或 "1m"。8是日K，5是1小时，1是1分钟。这个没有默认值，必须传对。

WebSocket心跳不能省。30秒一次ping，超过1分钟不发服务端会主动断开。用一个daemon线程跑心跳就行，别阻塞主消息循环。

返回字段是缩写。ld 是最新价，chp 是涨跌幅，v 是量。不要用 last_price、change_percent 这种直觉字段名，会直接KeyError。

小结

外汇行情接入的技术栈其实不复杂：REST拉历史，WebSocket收实时，统一用token鉴权。关键是把状态机走对——连接、鉴权、订阅、心跳这四个环节顺序不能乱，字段名对照文档用缩写，剩下的就是业务逻辑了。

最近这个阶段，欧洲央行刚加完息，美联储下周要议息，汇率市场的消息密度会持续比较高。有一套顺手的实时数据接入方案，不管是做监控工具还是自己看盘，都能少走很多弯路。

接口的完整文档可以在这里查阅，支持REST、WebSocket和FIX三种接入方式，除了外汇还有股票、指数、期货、基金等产品。

油价破百、央行决议、通胀数据扎堆：这种行情密集日该怎么盯盘

San Si wu — Thu, 10 Sep 2026 12:40:59 +0000

做量化这几年，我总结出一个规律：市场最容易出大行情的，不是单边趋势，而是一堆宏观事件挤在同一天的时候。

最近就是这种状态。原油站上100美元，美股连跌三天，欧洲央行要公布利率决议，美国PPI和CPI数据接踵而至，比特币在8万美元门口反复试探。这种日子里，信息处理速度直接决定了你能不能抓住机会、避开风险。

这篇文章我想做两件事：一是把最近市场上最值得关注的几条主线捋清楚，说说它们之间怎么互相影响；二是聊聊在这种行情密集的日子里，我自己是怎么把"找数据"这件事的时间压缩到最短的——毕竟数据获取效率上去了，才有更多精力留给判断。

最近市场的几条核心主线

1. 原油站稳100美元上方，中东局势是根因

最近一个交易日，布伦特原油期货收于101美元上方，涨幅超过3%；WTI原油也摸到96美元附近。两者都创了几个月以来的新高。

背后的推手很明确：美伊冲突升级。伊朗方面称在霍尔木兹海峡附近攻击了多艘船只，革命卫队警告将加大反制力度。美方则公开表示可能打击伊朗核设施。霍尔木兹海峡承担着全球约五分之一的原油运输，这个通道一旦出问题，油价的风险溢价就下不来。

油价破百的影响远不止能源板块。它直接推升通胀预期，进而影响央行的货币政策路径，最终传导到股市、债市、黄金和加密货币的定价上。可以说，最近所有资产的波动，多多少少都能追溯到油价这根线。

2. 美股连跌三天，但板块分化比指数更值得看

最近美股收盘，道指跌约0.77%，标普500跌约0.48%，纳指跌约0.64%。连跌三天，听起来市场很弱。

但拆开看，内部结构其实很有意思。能源板块逆势上涨约1%——油价涨，能源股直接受益。Meta暴涨超过6%，因为发布了个人AI智能体产品，市场给了非常正面的反馈。而工业和非必需消费板块领跌，逻辑也很直接：高油价挤压企业利润和消费者钱包。

盘前半导体和存储芯片股也出现集体下挫。这种"指数跌、个股冰火两重天"的行情，对做行业轮动和选股的人来说是机会，但前提是你得有及时、准确的数据支撑判断，而不是靠新闻标题拍脑袋。

3. 欧洲央行利率决议落地，加息预期已被充分定价

欧洲央行最近公布最新利率决议。市场此前几乎一致预期加息25个基点，将存款机制利率从2.25%上调至2.50%，这将是年内第二次加息。

加息本身不是重点，因为市场早就price in了。真正的变量在央行行长的新闻发布会：会不会暗示年内还有第三次加息？最新的宏观经济预测怎么调——增长预测可能小幅上调，但通胀预测大概率因为能源价格而上修？对中东冲突和油价的影响怎么看？

欧元兑美元目前在1.16附近震荡。如果发布会措辞偏鹰，欧元可能向上测试前高；如果偏鸽，欧元短线会承压。做外汇的朋友，这种决议日的波动往往几分钟内完成，盯盘工具的响应速度很关键。

4. 美国PPI和CPI接连公布，通胀走向决定美联储下一步

美国最新PPI数据近期公布，市场预期同比涨幅从4.7%升至5.2%左右，环比从0%升至0.4%。这个预期本身就不低，说明市场已经在计入油价带来的上游通胀压力。

如果PPI超预期，美债收益率会继续上冲，美元走强，黄金和加密货币承压；如果低于预期，风险资产可能迎来一波喘息。紧接着的CPI数据才是终极大考，再加上即将到来的FOMC会议，这段时间的宏观数据密度非常高。

目前市场对美联储下次会议加息的概率定价在60%左右。PPI和CPI出来之后，这个概率会大幅修正。对交易者来说，这种数据公布前后的行情，既危险也充满机会。

5. 比特币守在7.8万美元附近，ETF资金面出现分化

BTC最近在78000美元上下震荡，24小时小幅下跌，始终没能重新站上8万美元这个心理关口。以太坊在2460-2500美元区间。

资金面有一个值得注意的信号：美国现货比特币ETF连续两个交易日出现净流出，两天合计超过1.6亿美元。ARK旗下产品单日流出居首，Grayscale也有明显流出。

但与此同时，以太坊、XRP和Solana的ETF却在吸引资金流入。这种"比特币流出、山寨币流入"的分化，说明机构资金在加密市场内部做轮动，而不是全面撤离。全市场加密货币总市值在2.2-2.3万亿美元区间，恐慌贪婪指数处于偏低位置，市场情绪偏谨慎。

6. 黄金4400美元关口反复，白银跌得更狠

现货黄金最近在4370-4420美元之间来回波动。多头的逻辑是地缘冲突、美元走弱、央行持续购金；空头的逻辑是油价推升通胀预期→美联储更鹰→实际利率上行→压制黄金。两边都有道理，价格就在4400美元这个位置反复掰手腕。

白银的表现比黄金弱很多，伦敦银现大跌超过2%。金银比在走阔，说明市场当前更偏防御性配置，而不是押注通胀交易。

有一个反直觉的现象值得注意：中东打仗了，按道理黄金应该涨，但这次金价反应平平。原因就是前面说的传导链——油价涨→通胀黏性变强→美联储更没理由降息→加息预期变硬→反而压制黄金。这种时候，简单的"打仗买黄金"逻辑是会失效的，得靠数据去验证每一环。

7. A股港股跟跌，北证50重挫近3%

A股最近也没能独善其身。上证指数跌约0.43%，深证成指跌约0.77%，创业板指跌约0.49%。最惨的是北证50，跌了近3%，科创50也跌约0.7%。

港股更弱一些，恒生指数跌约1.27%，恒生科技指数跌超2%。中概股普遍回调，纳斯达克中国金龙指数跌约2%。

外围风险偏好下降加上油价高企带来的通胀担忧，是A股港股承压的主要外部因素。不过国内政策面也有积极信号：相关部门最近举行金融规划发布会，透露境外主体持有境内人民币金融资产已超11万亿元，80多个国家和地区将人民币纳入外汇储备。中长期看，人民币国际化的趋势还在推进。

数据获取的一些实操经验

上面这些分析，每一条都需要数据验证。我平时获取行情数据主要靠API接口，覆盖股票、外汇、指数、贵金属、期货、基金和加密货币，最近提到的所有资产类别都能在一个接口里拿到，不用在七八个行情网站之间来回切。

下面分享几个我自己常用的脚本，都是在这种行情密集日里真正能省时间的。

央行决议夜，用WebSocket盯外汇实时跳动

像欧洲央行利率决议这种事件，外汇市场的反应往往在几分钟内完成。用REST轮询太慢，我会用WebSocket订阅欧元兑美元的实时报价。连接建立后服务端会自动用请求头里的token做鉴权，鉴权通过后才能发订阅消息，这个顺序不能搞反。

import json
import threading
import time
import websocket

# 外汇实时行情 WebSocket 接入
WS_URL = "wss://api.itick.org/forex"
TOKEN = "your_token_here"  # 替换为你的 API Token

authenticated = False
heartbeat_timer = None


def start_heartbeat(ws):
    """每30秒发送一次心跳，保持连接活跃"""
    def ping():
        while True:
            time.sleep(30)
            ws.send(json.dumps({
                "ac": "ping",
                "params": str(int(time.time() * 1000))
            }))
    global heartbeat_timer
    heartbeat_timer = threading.Thread(target=ping, daemon=True)
    heartbeat_timer.start()


def subscribe(ws):
    """鉴权成功后订阅行情，params 格式为 标的$市场"""
    ws.send(json.dumps({
        "ac": "subscribe",
        "params": "EURUSD$GB,GBPUSD$GB",
        "types": "quote"
    }))


def on_open(ws):
    print("WebSocket 已连接，等待鉴权...")


def on_message(ws, message):
    global authenticated
    payload = json.loads(message)

    # 等待鉴权成功后再订阅
    if payload.get("resAc") == "auth" and payload.get("code") == 1 and not authenticated:
        authenticated = True
        print("鉴权成功，开始订阅行情")
        subscribe(ws)
        start_heartbeat(ws)
        return

    if payload.get("resAc") == "pong":
        return

    # 处理实时报价消息
    data = payload.get("data") or {}
    if data.get("type") == "quote":
        symbol = data.get("s")
        price = data.get("ld")
        ts = time.strftime("%H:%M:%S", time.localtime(data.get("t", 0) / 1000))
        print(f"[{ts}] {symbol}: 最新价 {price}")


ws = websocket.WebSocketApp(
    WS_URL,
    header=[f"token: {TOKEN}"],
    on_open=on_open,
    on_message=on_message,
)
ws.run_forever()

决议公布的那一刻，欧元的跳动你能第一时间看到，不用手动刷新网页。对于做事件驱动交易的人来说，这几秒的差距可能就是盈亏的差距。

验证原油突破有效性，拉一段日K线看量价

最近油价破100美元，我想确认这是真突破还是假突破。最直接的办法就是拉一段原油期货的日K线，看看成交量和之前几次冲击关键价位时有什么不同。K线接口的参数用 kType 控制周期，8代表日K，返回的每根K线包含开高低收和成交量。

import requests

# 获取期货K线数据
API_BASE = "https://api.itick.org"
TOKEN = "your_token_here"
headers = {
    "accept": "application/json",
    "token": TOKEN
}


def get_future_kline(region, code, k_type=8, limit=90):
    """
    获取期货K线数据
    region: 市场代码，如 US（美国期货）
    code: 品种代码，如 CL（WTI原油）、NQ（纳指期货）
    k_type: K线周期，1=1分钟 2=5分钟 5=1小时 8=日K 9=周K 10=月K
    limit: 返回K线数量，最大500
    """
    resp = requests.get(
        f"{API_BASE}/future/kline",
        headers=headers,
        params={
            "region": region,
            "code": code,
            "kType": k_type,
            "limit": limit
        }
    )
    resp.raise_for_status()
    return resp.json()["data"]


# 拉取WTI原油近90根日K线，分析突破有效性
cl_kline = get_future_kline(region="US", code="CL", k_type=8, limit=90)

# 计算近20日平均成交量，对比最新一根K线的量能
volumes = [bar["v"] for bar in cl_kline[-20:]]
avg_vol = sum(volumes) / len(volumes)
latest = cl_kline[-1]

print(f"WTI原油最新收盘价: {latest['c']}")
print(f"最新K线成交量: {latest['v']}")
print(f"近20日平均成交量: {avg_vol:.0f}")
print(f"量能放大倍数: {latest['v'] / avg_vol:.2f}x")

如果突破日成交量明显放大（比如1.5倍以上），说明有真实资金在推动，突破的可信度就高；如果缩量突破，就要小心是假突破。这种分析靠手动去各个网站翻数据，半小时就没了，用API几秒钟搞定。

一个脚本同时监控全球主要资产

这种宏观数据密集的日子，我最需要的是一个"仪表盘"——同时看美股期货、外汇、原油、黄金、比特币的实时状态。不同资产类别走不同的REST路径，外汇是 /forex/quote，加密货币是 /crypto/quote，期货是 /future/quote，指数是 /indices/quote，统一用 region + code 定位品种。

import requests

API_BASE = "https://api.itick.org"
TOKEN = "your_token_here"
headers = {
    "accept": "application/json",
    "token": TOKEN
}

# 重点监控的品种，按产品类型分组
# 每项格式: (产品路径, region, code, 显示名称)
watchlist = [
    ("/indices/quote", "US", "SPX",  "标普500"),
    ("/indices/quote", "US", "NDX",  "纳斯达克100"),
    ("/future/quote",  "US", "CL",   "WTI原油"),
    ("/forex/quote",   "GB", "EURUSD", "欧元兑美元"),
    ("/forex/quote",   "GB", "USDJPY", "美元兑日元"),
    ("/crypto/quote",  "BA", "BTCUSDT", "比特币"),
    ("/crypto/quote",  "BA", "ETHUSDT", "以太坊"),
]


def get_quote(path, region, code):
    """获取单个品种的实时报价"""
    resp = requests.get(
        f"{API_BASE}{path}",
        headers=headers,
        params={"region": region, "code": code}
    )
    return resp.json()["data"]


# 一键拉取所有关注品种的行情
print(f"{'品种':<12} {'最新价':>14} {'涨跌幅':>10}")
print("-" * 40)
for path, region, code, name in watchlist:
    q = get_quote(path, region, code)
    price = q.get("ld", "N/A")
    chp = q.get("chp", "N/A")
    sign = "+" if isinstance(chp, (int, float)) and chp >= 0 else ""
    print(f"{name:<12} {price:>14} {sign}{chp:>9}%")

跑一次这个脚本，全球主要资产的状态一目了然。我一般在重要数据公布前把这个挂着，数据一出就能立刻看到各个市场的联动反应。

最后

最近的市场节奏很快：原油破百、央行决议、通胀数据、加密货币震荡，所有事情挤在一起。这种日子里，谁能更快地拿到数据、验证逻辑，谁就更有优势。

我的做法还是老一套：先把信息串起来理解传导链条，然后用数据去验证每一环，最后等市场自己给答案。工具的作用就是帮你把"获取数据"这件事的时间压缩到最短，把更多精力留给思考和判断。

上面用到的接口文档可以在这里找到，有需要的朋友可以自行查阅。

一套 API 对接全球跨资产行情：股票、外汇、指数、贵金属、加密货币统一接入实战

San Si wu — Mon, 07 Sep 2026 21:50:47 +0000

前阵子要做个跨资产的监控小工具，把港股、美股、外汇、黄金还有比特币放一块儿看。原以为很简单，真开工才发现光数据源就够折腾：股票要接 A 源、外汇要接 B 源、加密货币又是另一套，每个源的鉴权方式、字段结构、K线周期定义全都不一样。数据接回来还得自己统一成一套格式，光字段映射就写了一百多行。

后来换了个思路，找一套能把多资产统一起来的行情 API，花了两个晚上把整个管道跑通了。这篇文章把接入过程写下来，代码都是能直接跑的，重点讲怎么用一个统一模型去接股票、外汇、指数这些不同资产，避免在数据格式转换上重复造轮子。

先搞清楚核心：多资产接口到底统一在哪

市面上多数行情源是按资产类别各做一套 API，字段名、时间戳单位、周期枚举全都对不上。这也是跨资产开发最头疼的地方。

我这次用的这套 API 核心逻辑是：所有资产类别走同一种 endpoint 结构、同一种鉴权、同一种 K 线数据结构。你只需要按资产类别换一下路径前缀，比如股票是 stock、外汇是 forex、指数是 indices，参数区就是三件套——region（市场代码）、code（产品代码）、kType（K线周期）。

举个例子，这是官网文档里的标准调用，拿港股腾讯控股（700）的 1 分钟线：

import requests

url = "https://api.itick.org/stock/kline?region=HK&code=700&kType=1"
headers = {
    "accept": "application/json",
    "token": "你的token"
}

resp = requests.get(url, headers=headers).json()
print(resp["code"], resp["msg"])

返回的 K 线是标准的 OHLCV 结构：o/h/l/c 是开高低收，v 是成交量，t 是毫秒时间戳，tu 是成交额。这个结构在外汇、指数上完全一致。

一个函数打通多资产：股票、外汇、指数

既然结构统一，那干脆写一个通用函数，资产类别做成参数。外汇这边市场代码有点特别，EURUSD 的 region 是 GB，别照搬股票那套 US/HK 逻辑。指数像标普 500 用 SPX，region 同样是 GB。

import requests

TOKEN = "你的token"

def fetch_kline(asset: str, region: str, code: str, ktype: int, limit: int = 10):
    """统一的K线拉取函数：asset 传 stock/forex/indices 等资产类别"""
    url = f"https://api.itick.org/{asset}/kline"
    params = {"region": region, "code": code, "kType": ktype, "limit": limit}
    resp = requests.get(url, params=params, headers={"accept": "application/json", "token": TOKEN})
    data = resp.json()
    if data["code"] != 0:
        raise RuntimeError(f"接口返回异常: {data['msg']}")
    return data["data"]

# 港股日线（kType=8 是日线）
hk_700 = fetch_kline("stock", "HK", "700", 8, 30)

# 外汇 EURUSD 5分钟线（kType=2）
eurusd = fetch_kline("forex", "GB", "EURUSD", 2, 30)

# 标普500 指数 1小时线（kType=5）
spx = fetch_kline("indices", "GB", "SPX", 5, 30)

print(f"港股700 最近收盘: {[k['c'] for k in hk_700][-1]}")
print(f"EURUSD 最近收盘: {[k['c'] for k in eurusd][-1]}")
print(f"SPX 最近收盘: {[k['c'] for k in spx][-1]}")

kType 的映射建议对着文档确认，不要凭感觉猜：常见的 1 是 1 分钟、2 是 5 分钟、5 是 1 小时、8 是日线、9 是周线、10 是月线。不同资产类别的周期档位略有差异，外汇还有 2 小时、4 小时这些档位。

这样写的好处很明显：以后要加新的资产，只改一行 asset 参数，不用再写一套映射逻辑。

用统一结构算跨资产相关性

数据结构统一了，后面做分析就顺了。我把拉回来的几类资产收盘价塞进 pandas，算一下日收益率的相关性矩阵，看看港股、美股、黄金、比特币之间到底联动强不强：

import pandas as pd
import numpy as np

def closes_to_series(klines, name):
    df = pd.DataFrame(klines)
    df["dt"] = pd.to_datetime(df["t"], unit="ms")
    df = df.set_index("dt").sort_index()
    return df["c"].rename(name)

# 拉几类资产的日线（这里用前面封装的函数）
assets = {
    "HK700": fetch_kline("stock", "HK", "700", 8, 60),
    "SPX": fetch_kline("indices", "GB", "SPX", 8, 60),
    "XAUUSD": fetch_kline("forex", "GB", "XAUUSD", 8, 60),
    "BTCUSDT": fetch_kline("crypto", "GB", "BTCUSDT", 8, 60),
}

# 合并成一张表，计算日收益率
px = pd.concat(
    [closes_to_series(kl, name) for name, kl in assets.items()],
    axis=1
)
ret = px.pct_change().dropna()

# 相关性矩阵
corr = ret.corr().round(3)
print(corr)

这里有两个点要留意。第一，不同资产的时间戳是按各自市场撮合时间记录的，合并前 sort_index 加 dropna 是必须的，否则对不齐。第二，加密货币是 7×24 小时交易，股票周末没行情，时间序列天然就不对齐，算相关性前先确认口径是"按交易日对齐"还是"按自然日对齐"，不然结果会偏。

数据质量验证：别拿到数据就信

跨资产管道最容易被忽略的是数据质量。我一般拿到 K 线会做三道检查：

def validate_klines(klines):
    """基础数据质量检查"""
    if not klines:
        return "空数据"
    df = pd.DataFrame(klines)
    # 1. OHLC 逻辑校验：high 必须 >= open 和 close
    bad_ohlc = df[(df["h"] < df[["o", "c"]].max(axis=1))].shape[0]
    # 2. 时间戳单调递增
    monotonic = df["t"].is_monotonic_increasing
    # 3. 周期完整性：相邻K线时间戳间隔应该一致
    gaps = df["t"].diff().dropna().unique()
    return {
        "bad_ohlc_rows": bad_ohlc,
        "timestamps_monotonic": bool(monotonic),
        "unique_intervals": len(gaps),
    }

print(validate_klines(hk_700))

做过一次就会知道，这种检查不是走形式——有的数据源在停牌日会返回重复时间戳，有的在极速行情下会出现 high 小于 close 的脏数据。统一结构 + 统一校验函数，换资产类别时检查逻辑一次复用。

几个踩过的坑

外汇的 region 别套股票逻辑：EURUSD、XAUUSD 这些 region 是 GB，一开始我按股票习惯填了别的市场代码，直接返回空。不同资产类别的 region 取值，看文档的枚举最靠谱。
kType 档位不是全资产通用：股票、指数、外汇的周期枚举不完全一样，比如外汇多 2 小时、4 小时档。写通用函数时，周期档位要做成配置，别硬编码。
时间戳单位要确认：返回的时间戳是毫秒，pd.to_datetime(..., unit="ms") 别漏了 unit 参数，漏了会把日期算到 1970 年去。
免费档有调用频率限制：REST 接口每分钟有次数上限，批量拉多资产历史数据时记得加 time.sleep() 控制节奏，不然会被限流。
token 放环境变量：代码里别写死，尤其是要推到 GitHub 的时候，这个坑我踩过不止一次。

小结

跨资产行情接入这件事，选对数据源能省掉一大半工作量。核心就看三点：数据结构是否统一、寻址方式是否一致、鉴权是否简单。这套 API 把股票、外汇、指数、贵金属、加密货币都收敛到同一个 OHLCV 模型下，对我来说最大的收益是少写了上百行字段映射代码。

如果你也在做跨资产的监控、回测或者数据管道，建议先拿免费档把两三个资产类别的 K 线拉通，重点验证数据结构和时间戳对齐这两个环节，跑通了再往上层加分析逻辑。详细的接口字段和 kType 映射，可以参考官方文档，Python/Java/Go 的 SDK 都有现成的示例，官网在这里。

iTick API 实时行情接入实战：用免费档搭一个多市场股票预警机器人

San Si wu — Mon, 31 Aug 2026 13:49:46 +0000

前阵子朋友让我帮他盯几只港美股，说是开盘时间他总在开会，错过了好几次加减仓的窗口。我嘴上答应，心里其实有点犯怵——总不能每五分钟刷一次行情软件吧，我也有自己的事。

于是花了一个周末，用 iTick 的免费档搭了个简单的行情预警机器人，WebSocket 长连接收实时推送，价格一碰到我设的阈值就推消息到手机上。这篇文章把整套接入过程写下来，代码都是能直接跑的，给想自己折腾实时行情接入的朋友当个参考。

为什么用 WebSocket 而不是轮询

一开始我也想过用 REST 接口定时去拉，写起来简单嘛。但算了一下账就放弃了：iTick 免费档 REST 接口是 5 次/分钟，如果盯 20 只股票，每只每 5 分钟拉一次就得 4 次/分钟，一个预警机器人就快把配额吃光了，更别说以后加标的。

WebSocket 是另一套逻辑，建一条长连接，服务端把行情主动推过来，你只需要订阅感兴趣的标的，不占 REST 的调用次数。免费档虽然 WebSocket 只给 1 条连接、3 个订阅，但对个人盯盘来说，3 个订阅刚好够我盯 3 只重点股票，REST 的 5 次/分钟配额还能留着做别的查询。

第一步：拉一把历史数据做参考

预警逻辑需要先有个基准价，比如昨收价或者最近 N 根 K 线的均价。我先用 REST 接口把腾讯控股（700）的日线拉出来，算一下过去 20 天的简单移动平均，拿这个当判断趋势的参考线。

iTick 的鉴权很简单，注册后在控制台生成一个 token，请求时放 header 里就行：

import requests

url = "https://api.itick.org/stock/klines?region=HK&codes=700&kType=8&limit=20"
headers = {
    "accept": "application/json",
    "token": "你的token"
}

resp = requests.get(url, headers=headers).json()
print(resp["code"], resp["msg"])

# 取出20根日线收盘价
closes = [k["c"] for k in resp["data"]["700"]]
sma20 = sum(closes) / len(closes)
print(f"700 最近20日收盘均价: {sma20:.2f}")

返回的每根 K 线是标准的 OHLCV 结构，o/h/l/c 是开高低收，v 是成交量，t 是毫秒时间戳。kType=8 是日线，想拉分钟线就改 1（1分钟）、2（5分钟）这些，映射关系文档里写得很清楚。

第二步：WebSocket 订阅实时行情

历史参考线有了，接下来接实时推送。iTick 官方有 Python SDK，装一下：

pip install itick-sdk

然后建连接、注册回调、订阅标的：

from itick.sdk import Client
import time

token = "你的token"
client = Client(token)

def on_message(message):
    # 收到的每一条实时行情都走这里
    print(f"[{time.strftime('%H:%M:%S')}] {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(
    '{"action": "subscribe", "region": "HK", "codes": ["700", "9988"], "types": ["quote"]}'
)

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    client.close_websocket()

SDK 内部把心跳和重连都处理好了：默认 30 秒发一次心跳，网络断开后按 5 秒间隔自动重连，最多重试 10 次，重连成功后订阅关系会自动恢复。这一点省了我很多事，自己手写 WebSocket 的话，心跳、断线重连、订阅恢复这三块至少要折腾大半天。

推送回来的 quote 消息里有最新价、涨跌幅、买一卖一这些字段，够做基础的盘中判断了。

第三步：加阈值预警逻辑

只打印消息没意义，得让它按我的规则提醒我。我在回调里加了简单的判断：最新价相对参考线（刚才算的 SMA20）的偏离超过 ±3% 就告警，并且用状态去重，同一只股票在告警解除前不重复提醒。

from itick.sdk import Client
import time
import json

token = "你的token"
client = Client(token)

# 参考价：从REST拿的SMA20
REF_PRICE = {"700": 535.0, "9988": 140.0}
THRESHOLD = 0.03  # 3%
alerted = {}  # 去重：记录每只股票是否在告警中

def on_message(message):
    try:
        data = json.loads(message)
    except Exception:
        return

    # 只处理实时报价
    if data.get("type") != "quote":
        return

    code = data.get("code")
    last = data.get("last", data.get("ld"))
    if not code or last is None:
        return

    ref = REF_PRICE.get(code)
    if not ref:
        return

    pct = (last - ref) / ref
    if abs(pct) >= THRESHOLD and not alerted.get(code):
        alerted[code] = True
        direction = "突破" if pct > 0 else "跌破"
        print(f"⚠️ {code} {direction}参考线 {abs(pct)*100:.1f}%，最新价 {last}")
        # 这里可以接推送服务，比如 Server酱、企业微信机器人

    if abs(pct) < THRESHOLD * 0.5:
        alerted[code] = False  # 回到安全区间，解除告警

client.set_message_handler(on_message)

client.connect_stock_websocket()
client.send_websocket_message(
    '{"action": "subscribe", "region": "HK", "codes": ["700", "9988"], "types": ["quote"]}'
)

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    client.close_websocket()

告警触发后怎么通知自己，看个人习惯：想收到手机推送，可以在那个 print 的位置换成 Server酱或者企业微信群机器人的 webhook；想落库做后续分析，就写进 SQLite。逻辑都一样，改一行的事。

第四步：多市场怎么扩展

盯完港股，想顺手把美股也加上，不用另起一套代码。iTick 的接口对多市场是统一的，region 参数切换市场：US 是美股、HK 是港股、SH/SZ 是沪深。订阅消息里把 region 换掉就行：

# 订阅美股
client.send_websocket_message(
    '{"action": "subscribe", "region": "US", "codes": ["AAPL", "TSLA"], "types": ["quote"]}'
)

免费档限制是 1 条连接、3 个订阅，所以同一时间只能盯一个市场的 3 个标的。想同时盯港股和美股，要么升级套餐加连接数，要么按时间段切换——比如 A 股收盘后再切到美股。个人用的话，我觉得按时间段切换其实够用。

几个踩过的坑

整个接入不算复杂，但有几个地方当时卡了我一阵，写出来免得大家重复踩：

region 代码别记混：美股是 US，港股是 HK，一开始我写了个 HKG，返回空数据，排查了半天才发现是市场代码的问题。
订阅参数格式要对：不同接口版本对订阅消息的字段名要求不完全一样，我用的 SDK 版本里 codes 要传数组，传成字符串会静默失败，没有任何报错，容易误以为连上了。
告警要去重：实时行情一秒推好几条，不做状态去重的话，价格一过阈值手机能给你震到没电。
免费档 REST 有频率限制：5 次/分钟，写历史数据拉取的循环时记得加 time.sleep(12)，不然会被限流返回错误。
token 别写死在代码里：放环境变量，尤其是要把代码推到 GitHub 的时候，血的教训。

小结

这套预警机器人从零到跑通，大概花了我一个周末，其中大半时间花在研究订阅格式和调试告警去重上，纯接入的部分其实很快。WebSocket 收实时行情 + REST 拉历史参考线，两条路配合，个人盯盘、辅助决策完全够用。

如果你是刚开始接触实时行情接入，或者想给自家产品加个行情监控功能，建议先拿免费档把 WebSocket 这条链路跑通，重点看数据延迟和推送稳定性是否符合预期，再决定要不要上更高级的规格。详细的接口字段和订阅格式，可以翻 iTick 的官方文档，Python/Java/Go 的 SDK 都有现成的示例。

有问题欢迎在评论区交流，我踩过的坑，希望能帮你少走点弯路。

全球股票行情数据怎么接？我用 iTick API 一周搭好了跨市场量化数据管道

San Si wu — Sun, 23 Aug 2026 21:42:13 +0000

全球股票行情数据怎么接？我用 iTick API 一周搭好了跨市场量化数据管道

做量化这几年，我踩过最多的坑不是策略，而是数据。

最早接美股的时候用的某家老牌数据商，文档写得像天书，SDK 只支持 Python 2，光调通一个历史K线接口就花了我三天。后来业务扩展到港股和A股，又分别接了两个不同的数据源，三套接口、三种鉴权方式、两套字段命名，每天光维护数据同步脚本就要占掉半天。

上个月团队要做一个覆盖全球主要市场的选股策略，我实在不想再维护第四套数据接入了，花了一周时间调研，最后选了 iTick。这篇文章记录一下整个接入过程，给同样在折腾多市场数据的朋友一个参考。

为什么是 iTick

选数据源我主要看三点：覆盖市场够不够广、接口是不是统一、延迟稳不稳定。

iTick 覆盖的市场确实全，美股、港股、A股、台湾、新加坡、日本、印度、泰国、德国、英国、荷兰这些主流市场都有，外加外汇、指数、期货、基金、加密货币，基本上一个 API 就能搞定大部分场景。详细的支持列表可以看他们的官方文档。

接口统一这点对我来说是最大的痛点解决。不管是股票还是外汇，REST API 的调用方式、参数结构、返回字段都是一套规范，不用每个市场都重新学一遍。WebSocket 也是统一的发布订阅模式，连上之后订阅什么品种就收什么数据。

延迟方面我实测了一周，美股盘口数据推送基本在毫秒级，99.99% 的可用性标称不是吹的，至少我这一周没遇到过断连。

第一步：REST API 拉历史K线

先从最简单的开始。iTick 的 REST API 用 header 传 token 鉴权，注册账号后在控制台生成一个 token 就能用。

拉港股腾讯控股（700）和阿里巴巴（9988）最近5根5分钟K线，直接一个 GET 请求：

import requests

url = "https://api.itick.org/stock/klines?region=HK&codes=700,9988&kType=2&limit=5"
headers = {
    "accept": "application/json",
    "token": "你的token"
}

response = requests.get(url, headers=headers)
print(response.json())

参数说几个关键点：

region 是市场代码，HK是港股，US是美股，SH/SZ是沪深，支持的市场列表文档里有
codes 支持批量，逗号分隔，免费版最多同时查10个
kType 是K线周期，1=1分钟，2=5分钟，3=15分钟，4=30分钟，5=1小时，8=日线，9=周线，10=月线

返回的数据结构是这样的：

{
  "code": 0,
  "msg": null,
  "data": {
    "700": [
      {
        "o": 535,
        "h": 536,
        "l": 534.5,
        "c": 534.5,
        "v": 104799385,
        "tu": 56119888070.5,
        "t": 1741239000000
      }
    ],
    "9988": [
      {
        "o": 139.9,
        "h": 140.3,
        "l": 139.8,
        "c": 140.1,
        "v": 538602171,
        "tu": 75404622753.1,
        "t": 1741239000000
      }
    ]
  }
}

标准的 OHLCV 结构，t 是毫秒级时间戳，tu 是成交额。拿到数据后直接丢进 pandas 就能算指标了，不用做任何字段映射。

我对比了一下，同样的日线数据，iTick 的复权处理是对的，拆股和分红都自动调整过了，不用自己再算复权因子。

第二步：WebSocket 订阅实时行情

历史数据搞定了，接下来是实时行情。做策略的都知道，轮询 REST API 既浪费调用次数又有延迟，WebSocket 才是正道。

iTick 的 WebSocket 是发布订阅模式，连上之后发一条订阅消息就行。我用他们的 Python SDK 来演示，比自己手写 websocket 连接省事很多：

from itick.sdk import Client
import time

# 初始化客户端
token = "你的token"
client = Client(token)

# 设置消息回调
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('{"action": "subscribe", "codes": ["700"]}')

# 保持连接
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    client.close_websocket()

SDK 内置了自动重连和心跳维护，断网后会自动重连并恢复订阅，不用自己写重连逻辑。重连间隔5秒，最多重试10次，心跳30秒一次，这些参数对大多数场景都够用了。

实时推送的数据包含最新价、买卖盘口、成交量这些，做盘中策略完全够用。我同时订阅了20只股票，推送频率很稳，没有遇到丢数据的情况。

第三步：用 SDK 还是直接调 API？

iTick 提供了 Python SDK 和 Java SDK，我两个都试了一下。

如果你的项目是 Python，强烈建议直接用 SDK，pip install itick-sdk 一行装好，封装了所有 REST 接口和 WebSocket 连接：

from itick.sdk import Client

client = Client("你的token")

# 股票实时报价
quote = client.get_stock_quote("US", "AAPL")
print(f"苹果最新价: {quote['ld']}")

# 历史K线
kline = client.get_stock_kline("US", "AAPL", kType=8, limit=30)

# 外汇实时tick
forex_tick = client.get_forex_tick("GB", "EURUSD")

# 加密货币深度
crypto_depth = client.get_crypto_depth("BA", "BTCUSDT")

方法命名很直观，get_stock_quote、get_stock_kline、get_forex_tick，看名字就知道干嘛的，不用翻文档。

如果你用的语言没有官方 SDK，直接调 REST API 也很简单，就是标准的 HTTP 请求加 header 鉴权，没有什么花里胡哨的签名算法。

数据质量这块多说几句

做量化的对数据质量都很敏感，我重点验证了几个方面：

复权处理：美股拆股、港股供股、A股分红，这些场景下的前复权数据都是对的，我抽了几只历史上有拆股的股票对比过，没有出现价格断层。

时间戳对齐：不同市场的交易时间不一样，iTick 返回的时间戳都是 UTC 毫秒，自己转成本地时间就行，不会出现时区混乱的问题。

缺失数据：停牌期间的K线会跳过，不会用0或者上一个价格填充，这点比有些数据商厚道，不用自己再过滤异常值。

盘口深度：免费版能拿到5档盘口，付费版档位更多，做市商策略可能需要高级版，普通趋势策略5档够用了。

成本怎么选

定价这块他们分了四档，我列一下实际使用中的感受：

免费版：5次/分钟 REST 调用，1个WebSocket连接，3个订阅。个人学习、回测够用，实盘肯定不够。
Base版 $79/月：120次/分钟，3个连接，200个订阅。个人交易者够用了。
Professional版 $159/月：600次/分钟，6个连接，500个订阅。小团队或者多策略适用。
Premium版 $319/月：1200次/分钟，12个连接，2000个订阅。机构级用量。

我现在用的是 Professional 版，同时跑3个策略，订阅了大概300只股票，调用量和连接数都还有余量。

几个踩坑提醒

最后说几个我接入时遇到的小问题，帮大家省点时间：

region 参数别写错：美股是 US 不是 USA，港股是 HK 不是 HKG，刚开始我写错了返回空数据，查了半天才发现。
WebSocket 订阅 codes 是数组：别传字符串，要传 ["700", "9988"] 这种数组格式。
免费版有调用频率限制：5次/分钟，写循环的时候记得加 sleep，不然会被限流。
历史数据深度：不同市场的历史数据深度不一样，美股能到十几年，一些小市场可能只有几年，回测前先确认数据量够不够。
token 别硬编码：放环境变量里，代码上传到 Git 之前检查一下，别把 token 提交上去了。

总结

用了 iTick 一个月，最大的感受就是省时间。以前接三个市场的数据要维护三套代码，现在一个 SDK 全搞定，省下来的时间能多研究几个策略。

如果你也在做多市场量化，或者正在为数据源的事情头疼，可以去 itick.org 注册个免费账号试试，免费版够你评估数据质量和接口易用性了。详细的接口文档在 docs.itick.org，写得还算清楚，照着例子抄基本不会出错。

有问题也可以在评论区交流，我踩过的坑能帮你少走点弯路。

土耳其股市数据踩坑记：itick Turkey Stock API 实际使用分享

San Si wu — Wed, 19 Aug 2026 19:16:45 +0000

做全球市场配置的朋友最近老问我土耳其股市的数据怎么拿。里拉这几年波动大，BIST（伊斯坦布尔证券交易所）反而成了不少宏观策略和新兴市场基金盯的池子。但说实话，土耳其市场的数据接口不像美股、A股那么好找，免费能用的更少。我自己试了一圈，最后用 itick 跑通了，这里把实际用法和踩过的坑记一下。

准备工作

pip install itick-sdk

去 itick.org 注册个账号，后台拿 token。土耳其市场的 region 代码是 TR，这个记一下，后面所有接口都要带。

from itick.sdk import Client
token = "your_api_token"
client = Client(token)

多周期K线交叉验证

土耳其股市波动比A股大，做技术分析的时候我习惯同时看几个周期交叉验证。itick 支持的周期还挺全的，从1分钟到月线都有。拿土耳其航空（THYAO）举个例子，8个周期各取5根对比一下：

# THYAO 土耳其航空，多周期K线
periods = {
    1: "1分钟",
    2: "5分钟",
    3: "15分钟",
    4: "30分钟",
    5: "1小时",
    8: "1天",
    9: "1周",
    10: "1月",
}
for k_type, label in periods.items():
    kline = client.get_stock_kline("TR", "THYAO", k_type, 5)
    print(f"{label}K线: {kline}")

分钟级的我一般用来做短线择时验证，日线周线看中长期趋势。这个多周期交叉验证的思路其实哪个市场都能用，不只是土耳其。

批量拉取权重股K线

如果要同时盯好几只权重股，一只一只请求太浪费调用次数了。itick 有批量接口，KCHOL（Koç Holding）、EREGL（Erdemir钢铁）、THYAO、GARAN（Garanti银行）这几只可以一次拉回来：

# 一次性拉取多只土耳其权重股的日K线
codes = ["KCHOL", "EREGL", "THYAO", "GARAN"]
klines_batch = client.get_stock_klines("TR", codes, 8, 30)
print("批量K线结果:", klines_batch)

免费套餐每分钟只有5次调用限额，批量接口基本是刚需——不然光拉四只股票的日K就把额度造完了。做板块轮动分析或者多标的对比图表的时候尤其好用。

WebSocket实时订阅：quote + depth + tick

做实时行情得用 WebSocket。itick 的 SDK 封装了连接逻辑，quote（报价）、depth（盘口）、tick（逐笔成交）三种类型可以同时订阅：

import time
def on_message(msg):
    print("推送:", msg)
def on_error(err):
    print("错误:", err)
client.set_message_handler(on_message)
client.set_error_handler(on_error)
client.connect_stock_websocket()
# 四只权重股，同时订阅报价、盘口、逐笔成交三种类型
client.send_websocket_message(
    '{"ac":"subscribe","params":"KCHOL$TR,EREGL$TR,THYAO$TR,GARAN$TR","types":"quote,depth,tick"}'
)
time.sleep(30)
print("连接状态:", client.is_websocket_connected())
client.close_websocket()

这里有个坑：三种数据类型同时订阅时，服务端推送的消息结构不一样，得用返回里的 type 字段区分。我一开始没注意，把所有消息都按报价处理，盘口数据进来直接报错。建议消息处理函数按类型分发：

def on_message(msg):
    import json
    data = json.loads(msg) if isinstance(msg, str) else msg
    msg_type = data.get("data", {}).get("type")
    if msg_type == "quote":
        handle_quote(data)
    elif msg_type == "depth":
        handle_depth(data)
    elif msg_type == "tick":
        handle_tick(data)
def handle_quote(data):
    print("报价更新:", data["data"]["ld"])
def handle_depth(data):
    print("盘口更新，买一价:", data["data"]["b"][0]["p"])
def handle_tick(data):
    print("成交:", data["data"]["ld"], "方向:", data["data"].get("d"))

quote 里的 ld 是最新价，depth 的 b[0].p 是买一价，tick 除了价格还有成交方向 d。字段名记得看清楚，我第一次把买一卖一搞反了，盯了半天才发现。

上线前要注意的几件事

限流保护：免费和基础套餐的 REST 调用频率有限制，客户端最好加一层简单的请求节流，短时间内打爆了会返回限流错误，影响线上服务。

异常降级：WebSocket 即使 SDK 会自动重连，重连的那几秒前端最好用最后一次收到的数据兜底，别让界面直接空白报错。

时区处理：土耳其跟北京有5小时时差（夏令时的时候是4小时），K线数据里的时间戳都是 UTC 毫秒，前端展示的时候记得按当地时区转换，不然会出现"收盘时间对不上"的困惑——我第一次就踩了这个坑。

多周期缓存：日线、周线这种变化频率低的数据，客户端可以做适当缓存，没必要每次都重新请求，减少不必要的调用消耗。

写在最后

土耳其市场因为波动性和汇率关联性，在数据消费场景上比其他市场稍微复杂一点，但 itick 的接口逻辑跟它支持的其他30多个市场是完全一致的，就是 region=TR 加上 BIST 的股票代码就行。免费套餐先跑通流程，需要更高频率和更多标的的时候再考虑升级。

如果你也在做新兴市场的数据接入，这个方案可以先试试。

项目地址：https://github.com/itick-org

做A股量化，一套Python SDK搞定A股行情数据沪深实时+历史行情

San Si wu — Mon, 17 Aug 2026 21:00:47 +0000

写量化小工具、个人盯盘脚本的时候，相信很多朋友跟我一样：最难的往往不是写策略逻辑，而是搞定靠谱的行情数据源。

之前踩过不少坑：用爬虫抓取网页行情，网站一改接口直接全部失效；高频请求就触发IP限流；不同市场返回字段五花八门，写一堆适配代码；想要实时推送，还要自己折腾WebSocket心跳、断线重连，一堆底层细节耗掉大把时间。

最近在做个人A股监控小项目，试了 itick 的 Python SDK，体验挺舒服，不需要复杂封装，REST拿历史K线，WebSocket接收实时推送，沪深两市一套代码就能搞定，在这里分享下实战踩坑与完整示例。

一个关键认知：A股要区分沪市SH、深市SZ

和美股港股不一样，A股接口里，上海交易所用region=SH，深圳交易所用region=SZ，这是最容易踩的第一个坑。

举两个典型标的：

贵州茅台 600519 → 沪市 SH
平安银行 000001 → 深市 SZ

⚠️注意：股票代码直接传纯数字，不要带 .SH / .SZ 后缀，region参数负责区分交易所，后缀加上反而查不到数据。如果拿不准某只股票归属哪个市场，可以调用get_symbol_list标的列表接口确认。

快速安装初始化

安装SDK，填入自己申请的token即可完成初始化：

pip install itick-sdk

from itick.sdk import Client

# 替换成你的token
token = "your_api_token"
client = Client(token)

REST接口：获取实时快照 + 历史K线

普通的最新报价、历史K线数据，直接调用REST接口就可以，同步调用简单直接，适合回测、定时拉取数据场景。

# 获取沪市贵州茅台实时报价
quote_sh = client.get_stock_quote("SH", "600519")
print("茅台实时报价:", quote_sh)

# 获取深市平安银行实时报价
quote_sz = client.get_stock_quote("SZ", "000001")
print("平安银行实时报价:", quote_sz)

# 获取茅台最近60根日线K线
# kType:1=1分钟,2=5分钟,3=15分钟,4=30分钟,5=1小时,8=日线,9=周线,10=月线
kline = client.get_stock_kline("SH", "600519", kType=8, count=60)
print("茅台日K线:", kline)

返回字段是统一规范：
o开盘、h最高、l最低、c收盘、v成交量、tu成交额。A股、港股、美股字段命名保持一致，做多市场项目的时候，不用反复写不同字段适配代码，这点很省心。

WebSocket实时推送：同时订阅沪深多只股票

如果要做盘中盯盘、预警脚本，轮询接口效率太低，优先使用 WebSocket 推送。

比较友好的一点：同一个连接可以同时订阅沪市、深市的标的，不用分别新建两条连接，监控沪深300一篮子股票的时候会省事很多。SDK内部已经封装好了心跳、断线自动重连，断线后最多重试10次，重连成功自动恢复订阅关系，不用自己手写重连逻辑。

示例代码：

import time

# 收到行情消息回调
def on_message(message):
    print("收到行情推送：", message)

# 异常回调
def on_error(error):
    print("连接异常：", error)

# 注册回调函数
client.set_message_handler(on_message)
client.set_error_handler(on_error)

# 建立websocket连接
client.connect_stock_websocket()

# 订阅标的：格式 代码$交易所，同时订阅quote快照、tick逐笔
sub_msg = '{"ac":"subscribe","params":"600519$SH,000001$SZ,300750$SZ","types":"quote,tick"}'
client.send_websocket_message(sub_msg)

# 保持连接30秒接收数据
time.sleep(30)

print("连接状态：", client.is_websocket_connected())
# 关闭连接
client.close_websocket()

💡小提示：A股不是7×24小时市场，非交易时段不会有tick、quote推送。如果连接成功但是收不到数据，先确认是否开盘；排查连接是否正常，可以尝试订阅kline类型做验证。

开发A股工具，这几个细节一定要留意

交易状态字段 ts
quote接口返回ts字段代表股票状态：0正常交易、1停牌、2退市、3熔断。写监控、预警程序务必判断这个字段，否则停牌、涨跌停股票的数据容易造成逻辑误判。
T+1 是交易规则，和行情接口无关
SDK只返回行情数据，不涉及交易下单。A股T+1属于券商交易层面规则，不会影响行情数据获取。
常见报错排查
cannot be resolved action：大概率订阅消息里股票代码、SH/SZ格式写错，核对代码$region写法；
完全收不到推送：优先检查当前是否A股交易时间，其次核对token权限。

简单总结

个人做量化小项目的时候，优先避开不稳定的网页爬虫，选择成熟SDK可以节省大量底层开发时间。

这套方案的优势总结：

统一SDK，A股区分SH/SZ，多市场字段统一；
历史K线用REST，实时行情用WebSocket，分工清晰；
WebSocket内置心跳、自动重连，减少造轮子；
有免费套餐可以用来做原型验证，跑通脚本之后，再按需升级扩大标的数量。

完整的字段说明、批量接口可以去官方文档中心查阅。

希望这篇实战可以帮到正在折腾Python量化的小伙伴。如果大家有好用的行情工具，也欢迎评论区一起交流。

免责声明：本文仅为技术开发实战分享，不构成任何投资建议，行情接口仅供程序学习研究使用。

参考文档：https://docs.itick.org/rest-api/stocks/stock-kline
GitHub：https://github.com/itick-org/

港股数据接口开发教程 Python 实现实时报价、逐笔、盘口深度推送

San Si wu — Sat, 15 Aug 2026 17:32:54 +0000

做港股量化、写小行情工具也有大半年了，期间换过好几个行情数据源，踩过不少坑：要么实时推送延迟高、要么历史K线残缺、免费额度抠门到没法测试，直到最近上手iTick，才算把港股数据对接流程理顺，分享一套实测能用的实操方案，给同样做港股数据需求的朋友避坑。

一、先说下港股数据采集的难点

和A股、美股不一样，港交所原生行情授权门槛很高，个人开发者很难拿到一手直连权限，市面上很多第三方工具存在几个通病：

实时行情推送卡顿，盘口逐笔数据断断续续，做短线回测完全不准；
历史分钟线、分时线接口收费贵，免费额度只给少量日线，回测根本不够用；
SDK封装差，大多只提供原生http请求，WebSocket实时订阅要自己手写心跳、重连，调试成本极高；
标的代码规则混乱，有的要加0前缀、有的要后缀，多股票批量查询容易报错。

对比下来iTick对个人开发者友好度会高很多，一套SDK同时覆盖REST静态历史数据+WebSocket实时推送，港股、美股、A股切换只改地区参数，不用重复写多套对接逻辑，小项目、个人量化工具完全够用。

二、零门槛上手，新手也能快速搭环境

1. 安装依赖

直接pip一键安装官方封装好的SDK，不用自己处理http请求、websocket连接底层逻辑：

pip install itick-sdk

2. 获取权限

官网注册账号就能领取免费token，不用企业资质，个人开发者直接可用：
免费套餐权益（日常测试足够）：

REST历史接口每分钟5次调用
WebSocket单条连接可订阅3只个股像腾讯700、阿里9988、美团3690这类热门标的，拿来做短期测试、小策略回测完全没问题，后期多标的批量监控再升级付费套餐就行。

三、两个高频场景实操代码（直接复制运行）

场景1：拉取港股历史K线，做策略回测

支持1分钟、5分钟、小时线、日线、周线、月线全周期，返回开高低收、成交量、成交额、时间戳完整字段，不用自己二次清洗时间。

from itick.sdk import Client
# 替换成你自己的token
token = "你的API_TOKEN"
client = Client(token)

# 1. 腾讯控股700，获取最近30根日线
stock_day = client.get_stock_kline("HK", "700", 8, 30)
print("腾讯日线数据：", stock_day)

# 2. 阿里巴巴9988，获取20根60分钟K线
stock_60m = client.get_stock_kline("HK", "9988", 5, 20)
print("阿里小时线数据：", stock_60m)

周期编码对照表，按需替换第三个参数：
1=1分钟｜2=5分钟｜3=15分钟｜4=30分钟｜5=1小时｜8=日线｜9=周线｜10=月线
返回字段简单说明：o开盘、h最高、l最低、c收盘、v成交量、tu成交额、t毫秒时间戳，标准化格式，pandas直接导入做回测。

场景2：WebSocket实时订阅盘口、逐笔、实时报价

做盯盘工具、实时信号预警必备，内置心跳、断线自动重连，不用自己写定时心跳、断线重试逻辑，省心很多。

import time
from itick.sdk import Client

token = "你的API_TOKEN"
client = Client(token)

# 接收推送回调
def on_message(msg):
    print("实时行情推送：", msg)

# 异常捕获回调
def on_error(err):
    print("连接异常：", err)

# 绑定回调函数
client.set_message_handler(on_message)
client.set_error_handler(on_error)

# 建立长连接
client.connect_stock_websocket()
# 批量订阅多只港股：实时报价+逐笔成交+五档盘口深度
sub_msg = '{"ac":"subscribe","params":"700$HK,9988$HK,3690$HK","types":"quote,tick,depth"}'
client.send_websocket_message(sub_msg)

# 持续接收20秒数据测试
time.sleep(20)
# 关闭连接
client.close_websocket()

订阅成功会返回提示subscribe Successfully，如果代码/地区填错会提示解析失败，检查股票代码和HK地区标识即可。
额外实用细节：港股逐笔tick数据自带成交方向d字段（1卖出、2买入），做资金流向、逐笔成交分析非常方便，目前支持A股/港股/美股三个市场。

四、内置省心功能，省掉大量开发工作量

自动心跳+断线重连
官方SDK已经封装30秒自动心跳，断线后5秒间隔重试，最多10次重连，重连后自动恢复订阅，不用自己维护定时任务，跑隔夜策略不用怕断流丢数据。
批量行情接口，高效多标的监控
自带get_stock_quotes批量报价接口，一次性查询多只股票实时价格，不用循环单条请求，大幅减少接口调用次数，节省额度。
多市场统一调用逻辑
港股region填HK，切换A股、美股只修改地区参数，接口名称、传参逻辑完全一致，同时做多市场策略不用维护多套对接代码。

五、实测踩坑总结，提前避坑

港股代码纯数字即可，比如腾讯直接传700，0700也能识别，但官方示例统一不带前导0，建议统一格式避免后期出兼容问题；
免费套餐WebSocket仅支持3个标的订阅，批量盯盘多只个股建议升级付费版；
1分钟K线实时推送（WebSocket kline）仅高级套餐开放，普通用户拉分钟历史数据走REST接口更稳定；
接口调用有频率限制，批量循环拉取数据记得加延时，避免超限被限制访问。

六、适合哪些人用？

个人量化交易者：写策略回测、实时信号提醒工具；
小型资讯/盯盘小工具开发者：需要稳定港股实时+历史数据；
多市场交易者：同时覆盖A股、港股、美股，想要统一SDK降低开发成本。

整体用下来对比之前试的几款数据源，itick最大优势是对个人开发者友好，文档清晰、SDK封装完整，不用耗费大量时间处理底层连接、数据格式化，入门门槛低，免费额度足够前期测试，有港股数据需求的朋友可以自行注册试一下。

温馨提示：本文仅供代码参考，不构成任何投资建议。市场有风险，投资需谨慎

参考文档：https://docs.itick.org/rest-api/stocks/stock-kline
GitHub：https://github.com/itick-org/

越南股市数据获取实践：用 Python 轻松接入 HOSE 实时行情

San Si wu — Sun, 09 Aug 2026 14:06:44 +0000

胡志明证券交易所（HOSE）的日均成交额在 2025 年已突破 20 万亿越南盾，外资净买入额连年攀升。Vinamilk（VNM）、Vingroup（VIC）这些标的不仅在本土有影响力，也逐渐进入全球新兴市场投资者的视野。但相比美股、港股，越南市场的行情数据接口在国内讨论得并不多——要么依赖国际券商提供的有限 API，要么自己爬取网页，稳定性堪忧。

最近我在调研东南亚市场的数据方案时，发现一套比较顺手的工具链，借着这篇文章把实践过程记录下来，希望能给对越南股市感兴趣的朋友节省一些踩坑时间。

数据接口的选型思路

最初我考虑过直接用 requests 爬取 Vietstock 或 CafeF 的页面，但很快就遇到两个问题：一是反爬策略越来越严，二是历史 K 线需要自己拼接计算，分钟级数据几乎拿不到。后来转向专业数据服务，对比了几家国际行情提供商，最终选了 itick，理由很朴素：

它对越南市场的覆盖以 HOSE 主流标的为主，正好是我需要的范围；
Python SDK 封装得比较干净，WebSocket 自动重连和心跳保活都内置了，省去自己处理断线重连的麻烦；
免费套餐足够跑通流程，确认数据字段后再决定是否升级。

当然，如果你只需要收盘价做回测，用付费的普通 REST 接口也够；但如果要搭建实时看盘或策略监控，WebSocket 推送是更合适的方式。

环境准备与初次调用

安装 SDK 只需要一行：

pip install itick-sdk

注册账号后在后台拿到 token，就可以开始调用了。越南市场的 region 参数固定为 "VN"，股票代码沿用 HOSE 的惯例（如 VNM、VIC、VCB）。

from itick.sdk import Client

client = Client("your_api_token")

# 获取 VNM 实时报价
quote = client.get_stock_quote("VN", "VNM")
print(quote)

返回的字段结构和其他新兴市场保持一致，核心字段包括：

ld —— 最新价
ch / chp —— 涨跌额 / 涨跌幅
o / h / l —— 开 / 高 / 低
ts —— 交易状态（正常、熔断、停牌等）

这里特别提醒：越南股市有涨跌停板（HOSE 为 ±7%），如果 ts 字段显示熔断或停牌，一定不能把旧价格当作实时价展示，否则会误导用户。

历史 K 线获取与常用周期

对于回测或技术分析，历史 K 线是刚需。itick 的 get_stock_kline 接口支持从 1 分钟到月线多种周期，kType 参数对应关系如下：

kType	周期
1	1分钟
2	5分钟
3	15分钟
4	30分钟
5	1小时
8	日线
9	周线
10	月线

以下代码获取 Vingroup（VIC）近 60 个交易日的日线数据：

kline = client.get_stock_kline("VN", "VIC", 8, 60)
for bar in kline:
    print(f"{bar['t']} O:{bar['o']} H:{bar['h']} L:{bar['l']} C:{bar['c']}")

返回的时间戳是 Unix 毫秒级，转换成北京时间需要注意时差（越南比北京慢 1 小时）。另外，越南盾面值较大，价格字段本身是整数或浮点数，前端渲染时记得做千分位格式化，避免一长串零影响阅读。

WebSocket 实时推送与分钟级 K 线

如果你需要做盘中实时监控或动态图表，REST 轮询的效率不够，WebSocket 是更好的选择。itick 的 SDK 已经封装好了连接、心跳和自动重连逻辑（5 秒间隔，最多 10 次重试），我们只需关注消息处理：

def on_message(msg):
    print("收到推送:", msg)

def on_error(err):
    print("连接错误:", err)

client.set_message_handler(on_message)
client.set_error_handler(on_error)

client.connect_stock_websocket()

# 订阅 VNM 和 VIC 的实时报价 + 1分钟 K 线推送
client.send_websocket_message(
    '{"ac":"subscribe","params":"VNM$VN,VIC$VN","types":"quote,kline@1"}'
)

这里有个实际成本考量：kline@1（1 分钟 K 线实时推送）在 itick 的免费和基础套餐里是不可用的，需要 Premium 及以上或者单独购买股票数据包。如果你的产品对分钟级数据实时性要求不高，完全可以退而求其次，用 REST 接口每 60 秒拉取一次最新 K 线，成本会低很多，且逻辑足够简单。

对于日线级别的展示，REST 拉取已经绰绰有余，WebSocket 主要优势体现在需要捕捉盘中异动的场景。

越南市场特有的几个注意事项

交易所区分：HOSE 是胡志明市主板，HNX 是河内交易所（类似创业板）。itick 目前主要覆盖 HOSE 标的，如果需要查询某只代码是否支持，可以先调用 get_symbol_list("VN") 遍历确认。
节假日与交易时段：越南股市交易时间为周一至周五的 9:00-11:30 和 13:00-14:45（越南时间），比 A 股早收盘。做定时任务时要注意时区和节假日日历，避免无效请求。
流动性分层：HOSE 上头部股票（VNM、VIC、VCB、HPG）流动性很好，但一些小盘股日内成交稀疏，tick 数据可能跳变较大，回测时需注意处理非活跃时段的数据。

如果扩展到其他东南亚市场

这套代码的可移植性是我比较满意的地方。itick 对印尼（ID）、菲律宾（PH）、马来西亚（MY）等市场也有支持，后续如果要做跨市场对比，只需要把 region 参数换成对应的国家代码，其余字段结构基本一致，开发成本非常低。

当然，前提是先确认目标市场在 itick 的产品目录中是否有 coverage——这一点建议直接查阅官方文档的 symbol 列表，比盲目尝试更高效。

结语

从零开始接入越南行情数据，其实没想象中复杂。一套 SDK + 几行代码就能拿到可用的实时报价和历史 K 线，剩下的精力可以放在策略逻辑和前端交互上。至于选哪家数据服务商，我觉得关键看三点：市场覆盖是否匹配、接口稳定性和维护成本、以及定价与自身需求的契合度。itick 在我这次调研中表现合格，特别是免费套餐让前期验证几乎没有门槛，算是比较务实的选择。

如果你也在研究东南亚市场，欢迎交流。数据是投资的起点，但远不是终点——祝大家都能在胡志明指数的起伏中找到自己的节奏。

参考文档：https://docs.itick.org/rest-api/stocks/stock-kline
GitHub：https://github.com/itick-org/

How to Access Swiss Stock Market Data via API – Live Prices & Historical Candlesticks for SMI Top Stocks (NESN, NOVN, ROG)

San Si wu — Sun, 02 Aug 2026 11:49:37 +0000

Recently I have been building a lightweight global asset monitoring dashboard. I quickly solved the data interfaces for US stocks, A-shares, and Hong Kong stocks, but the market data for the Swiss Exchange (SIX) was the one that kept getting stuck.

Anyone building global asset allocation or personal wealth management tools knows the Swiss market hides many heavyweight blue-chips — Nestlé, Novartis, Roche. These are classic holdings in a global portfolio. Most free data sources do not cover SIX, and paid APIs are too expensive. Before, I had to make do with static data and couldn’t support real-time monitoring or historical trend review.

After testing multiple financial data APIs, I finally used iTick’s Python SDK to successfully fetch Swiss market real-time quotes, historical K-lines, and WebSocket push data. The integration process is very simple, requires no complex configuration, and the free quota is enough for personal projects. Here I record the full implementation process so others can avoid pitfalls and get up and running quickly.

1. Why You Must Integrate Swiss Market Data

Many personal quant and asset dashboard projects tend to ignore the Swiss market, but SIX’s core names are highly valuable:

Nestlé (NESN): global consumer food leader, defensive asset benchmark
Novartis (NOVN), Roche (ROG): global pharma giants, core healthcare allocation
UBS Group (UBSG): major international financial institution

These low-volatility, high-stability overseas blue chips are essential for diversification. If you build global asset visualization, cross-border backtesting, or personal holdings monitoring, missing Swiss market data leaves the asset allocation picture incomplete.

I used static end-of-day data for a long time. That prevented me from seeing intraday fluctuations or automatically pulling K-lines for trend analysis. The user experience was terrible. After integrating a real-time API, I finally closed the loop for mainstream global market coverage.

2. Preparation: Account Registration and SDK Installation

The integration is very lightweight and does not require complicated approvals. Individual developers can get started quickly:

Go to the iTick official website (https://itick.org) to register an account. In the personal console you can get an API Token. One token works for both REST and WebSocket APIs, so no separate application is needed.
Install the official Python SDK locally in one command, compatible with mainstream Python 3 versions:

pip install itick-sdk

New users receive free usage quota. That is enough for personal dashboards, small-scale data review, and daily market monitoring without needing to upgrade to paid service.

3. Example 1: Fetch Nestlé Real-Time Quote in Python

iTick uses a unified region coding scheme across global markets. The Swiss market is identified as CH, and stock symbols use the official SIX codes directly, so no secondary mapping is needed.

Just a few lines of code fetch Nestlé real-time market data, including current price, open, high, low, and other core fields:

from itick.sdk import Client

# Replace with the real token from your personal console
token = "your_api_token"
client = Client(token)

# Get Nestlé (NESN) real-time quote on the Swiss market
quote = client.get_stock_quote("CH", "NESN")

# Practical Guide: Accessing US Stock Quant Data — Low-Cost Real-Time Quotes and Historical K-Lines

Environment Setup and Data Source Selection

First, install the SDK:

pip install itick-sdk

Then register an account at https://itick.org and obtain an API token from the dashboard. You will need this token for every request.

Why choose iTick? There are many excellent financial data APIs, but iTick's free tier already covers major markets including US, HK, and A-share real-time quotes and historical K-lines. The REST API allows 5 calls per minute, and WebSocket supports 1 connection with subscriptions for 3 symbols on the free tier. This quota is sufficient for learning and prototype testing, and latency is acceptable in practice. If you need higher frequency, paid plans offer configurations such as 600 calls/minute and multiple connections.

1. Fetching Real-Time Quotes (REST)

from itick.sdk import Client

token = "your_api_token"
client = Client(token)

# Use "US" for the US market
quote = client.get_stock_quote("US", "AAPL")
print("Latest Apple quote:", quote)

tick = client.get_stock_tick("US", "TSLA")
print("Tesla tick:", tick)

Set region to "US"; the SDK will build the full request. Returned fields include ld (latest price), o/h/l (open/high/low), ch/chp (price change and percent change), which match typical broker app fields.

2. Fetching Historical K-Lines

US stocks support multiple intervals from 1-minute to monthly. The kType values are:

kType	Meaning
1	1 minute
2	5 minutes
3	15 minutes
4	30 minutes
5	1 hour
8	1 day
9	1 week
10	1 month

Pull the last 10 five-minute K-lines for Apple:

kline = client.get_stock_kline("US", "AAPL", kType=2, limit=10)
print("K-line data:", kline)

Each K-line includes t (timestamp), o/h/l/c (open/high/low/close), v (volume), and tu (turnover). That’s sufficient for basic charting.

3. WebSocket Real-Time Push

Polling REST endpoints can hit rate limits and is inefficient for live monitoring. Use WebSocket so the server pushes updates. The iTick SDK wraps connection handling, heartbeat, and auto-reconnect.

Example subscribing to Apple and Tesla quotes and order book depth:

import time

def on_message(message):
    print(f"Received push: {message}")

def on_error(error):
    print(f"WebSocket error: {error}")

client.set_message_handler(on_message)
client.set_error_handler(on_error)

# Establish WebSocket connection
client.connect_stock_websocket()

# Subscribe to AAPL (US) and TSLA (NASDAQ on US market)
subscribe_msg = '{"ac":"subscribe","params":"AAPL$US,TSLA$NASDAQ$US","types":"quote,depth"}'
client.send_websocket_message(subscribe_msg)

# Keep the connection for a while; replace with an event loop in production
time.sleep(30)

print("WebSocket connected:", client.is_websocket_connected())
client.close_websocket()

Subscription parameters:

params format: SYMBOL$REGION. If you need to disambiguate exchanges within a market, use SYMBOL$EXCHANGE$REGION.
types can include tick (trade-by-trade), quote (real-time quote), depth (top-10 order book), and kline (K-line push; note 1-minute kline@1 may be restricted to premium plans).

The SDK provides a default heartbeat (ping every 30s) and automatic reconnection on network blips (retry every 5s up to 10 times). Subscriptions are restored after reconnect.

FAQ

Q: Is the free tier enough for daily development?

A: For learning, prototyping, or low-frequency strategies, 5 REST calls/min and 1 WebSocket connection are usually sufficient. Upgrade when you need to monitor more symbols or higher-frequency data.

Q: How is latency?

A: iTick reports millisecond-level push latency. In my tests, delay is small and suitable for monitoring and real-time computation; actual latency depends on server location and network.

Q: How many symbols can one WebSocket connection subscribe to?

A: A single connection supports up to 500 symbols. For thousands of symbols, split across connections or request higher limits from support.

Conclusion

This article demonstrates accessing US real-time quotes and historical K-lines using the iTick Python SDK: use REST for on-demand queries and WebSocket for continuous real-time feeds. Together they support stock selection, monitoring, and backtesting workflows. From registration to receiving the first WebSocket push, you can get up and running in about 30 minutes.

For more field details, batch endpoints, and SDKs in Java/Go/Node.js, see the iTick docs: https://docs.itick.org

美股量化数据接入实战：低成本获取实时行情与历史K线

San Si wu — Sun, 02 Aug 2026 11:36:06 +0000

环境准备与数据源选择

先装 SDK：

pip install itick-sdk

然后去 itick 官网注册账号，在后台拿到 API Token。所有请求都会用到它。

为什么选 itick？其实社区里还有很多优秀的金融数据 API，但 itick 的免费套餐已经覆盖了美股、港股、A股等主流市场的实时数据和历史K线。REST 接口每分钟可以调用 5 次，WebSocket 允许 1 条连接、订阅 3 个标的。这个额度对学习、原型测试基本够用，实测下来延迟也在可接受范围内。如果你需要更高频的查询，官方也提供了 600 次/分钟、6 条连接等更高配置的付费方案，按需升级就好。

1. 获取实时报价（REST）

from itick.sdk import Client

token = "your_api_token"
client = Client(token)

# region 填 US 代表美股市场
quote = client.get_stock_quote("US", "AAPL")
print("苹果最新报价：", quote)

tick = client.get_stock_tick("US", "TSLA")
print("特斯拉 Tick：", tick)

region 参数传 "US" 即可，SDK 底层会自动拼接成完整请求。返回的字段里，ld 是最新价，o/h/l 是开、高、低，ch/chp 是涨跌额和涨跌幅，和券商 App 看盘时的数据基本一致。

2. 获取历史 K 线

美股支持从 1 分钟到月线的多种周期，参数 kType 的取值整理如下：

kType	含义
1	1 分钟
2	5 分钟
3	15 分钟
4	30 分钟
5	1 小时
8	1 天
9	1 周
10	1 月

拉取苹果最近 10 根 5 分钟 K 线的代码：

kline = client.get_stock_kline("US", "AAPL", kType=2, limit=10)
print("K线数据：", kline)

每条 K 线包含 t（时间戳）、o/h/l/c（开高低收）、v（成交量）和 tu（成交额），画个简单的走势图完全够用。

3. WebSocket 实时推送

盯盘、量化实盘等场景下，反复轮询 REST 接口很容易触碰频率上限，而且效率不高。更好的方式是使用 WebSocket，让服务端主动推送数据。itick 的 SDK 已经封装好了连接、心跳和自动重连。

下面是一个完整的订阅例子，同时接收苹果和特斯拉的报价与盘口数据：

import time

def on_message(message):
    print(f"收到推送: {message}")

def on_error(error):
    print(f"WebSocket 出错: {error}")

client.set_message_handler(on_message)
client.set_error_handler(on_error)

# 建立 WebSocket 连接
client.connect_stock_websocket()

# 订阅 AAPL（US 市场）和 TSLA（NASDAQ 交易所，US 市场）
subscribe_msg = '{"ac":"subscribe","params":"AAPL$US,TSLA$NASDAQ$US","types":"quote,depth"}'
client.send_websocket_message(subscribe_msg)

# 保持连接一段时间，实际项目中可以改成事件循环
time.sleep(30)

print("当前连接状态：", client.is_websocket_connected())
client.close_websocket()

关于订阅参数：

params 格式为 代码$地区。如果同一市场内有多个交易所需要区分，可以写成 代码$交易所$地区。
types 可选 tick（逐笔成交）、quote（实时报价）、depth（十档盘口）、kline（K线推送，其中 1 分钟粒度的 kline@1 目前仅对 Premium 及以上套餐开放）。

SDK 还内置了心跳机制（默认每 30 秒 ping 一次），一旦因网络抖动断线，会自动按 5 秒间隔重连，最多尝试 10 次。重连成功后，之前的订阅关系也会自动恢复。这个小设计在搭建实时监控时省了不少事。

常见问题

Q：免费套餐能支撑日常开发吗？

如果只是个人学习、搭建原型或者跑一些低频策略，每分钟 5 次 REST 调用、1 条 WebSocket 连接基本足够。当策略需要监控更多标的或更高频的数据时，再考虑升级到更高级别的套餐。

Q：数据延迟表现怎么样？

itick 官方给出的延迟参考是毫秒级推送。我在本地测试下来，做盯盘工具和实时计算的延迟感知很小，符合一般量化的要求。当然实际体验也和你的服务器位置、网络环境有关。

Q：一条 WebSocket 连接可以订阅多少个标的？

单条连接最多支持 500 个标的。如果你监控上千只美股，可以按逻辑拆分成多条连接，或者联系客服调整连接数上限。

结语

本文用 itick 的 Python SDK 演示了美股实时行情和历史 K 线的基本接入流程：REST 负责一次性或低频查询，WebSocket 负责持续的实时推送。两条链路配合使用，基本能覆盖选股、盯盘、策略回测等场景。整个流程从注册、拿到 Token 到跑通第一条 WebSocket 推送，通常半小时内能搞定。

更详细的字段说明、批量请求接口以及 Java/Go/Node.js 等语言的 SDK，可以查阅 itick 文档中心。希望这篇文章能帮你少走一点弯路，快速把数据层搭起来。

免责声明：本文仅用于技术交流与学习，不构成任何投资建议。