台湾股票行情API接入教程

想用程序获取台股行情,很多人第一步会搜“台股资料 API 推荐”,结果发现选择不少:证交所有免费接口,Yahoo Finance 也能查,券商还有自己的 API。真正写代码时才发现,有的数据只到收盘,有的延迟 20 分钟,有的必须先开户。

本文先梳理台湾行情数据的几种获取渠道,再用 Python 从零搭一个可以运行的台股数据脚本:查代码清单、取实时报价和盘口、拉历史K线、做复权、用 WebSocket 接收实时推送,最后结合台积电 ADR 和新台币汇率计算跨市场溢价。

1. 台湾行情数据API有哪些渠道

渠道时效接入门槛适合场景
证交所 / 柜买中心 OpenAPI盘后数据免费,无需 Key日频研究、数据对账
Yahoo Finance(yfinance)延迟约 20 分钟免费,yfinance 为非官方库学习、非实时分析
券商 API,如永丰 Shioaji实时需要券商账户并申请 API Key已在该券商交易的程序化用户
台股行情 API,如富果实时申请金钥,按方案限速只做台股的应用
多市场行情 API实时注册取得 API Key,付费使用需要同时使用台股、美股、外汇等数据
直接向证交所申请授权实时费用高,需要签约有一定规模的机构

直接申请授权的成本最容易被低估。按证交所的收费标准,非券商和期货商申请者的传输授权费是每月 6 万新台币;如果通过公网向 0 至 5000 户提供自动更新的报价,信息费是每月 10 万新台币。起步就是每月 16 万新台币,而且不包括柜买中心的数据。

选择的思路其实很简单:只做日频分析,先用官方 OpenAPI;需要盘中实时数据,再看券商 API 或第三方接口;如果台股只是多个市场之一,用一套多市场接口可以少维护几份代码。

本文的示例基于 Infoway API,属于多市场这一类。后面的台股规则、数据对齐和踩坑经验并不依赖某个接口,换成其他数据源时同样适用。

2. 写代码前要了解的台股规则

以下规则主要来自证交所的交易制度说明,每一条都会影响数据处理:

项目规则对代码的影响
交易时间周一至周五 09:00–13:30(UTC+8),无午休不要沿用 A 股、港股的午休判断
集合竞价开盘 08:30–09:00,收盘 13:25–13:301 分钟K线在 13:24 之后会跳到 13:30
涨跌幅上市、上柜股票为 ±10%兴柜股票不适用,不能一律按 10% 校验
交易单位1 张 = 1000 股接口的成交量按股返回,展示时要换算
价格档位随价格变化,1000 元及以上为 5 元一档不能用固定的最小变动价计算价差
币种新台币(TWD)和美股 ADR 比较时要做汇率换算

3. 准备工作

安装依赖(Python 3.9 及以上):

pip install requests pandas websockets

把 API Key 和请求逻辑封装到一个文件里,后面的脚本都会用到它。这里有两点值得注意:业务错误通常是 HTTP 200 加上非 200 的ret,只检查 HTTP 状态码会漏掉错误;另外,所有接口共用同一个频率限制,所以在这一层统一限速。

# tw_api.py
import time

import requests

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://data.infoway.io"
MIN_INTERVAL = 1.1  # 请求间隔(秒),可按自己套餐的频率调整

session = requests.Session()
session.headers.update({"apiKey": API_KEY})
_last_request = 0.0


class ApiError(Exception):
    pass


def _request(method, path, **kwargs):
    global _last_request
    wait = MIN_INTERVAL - (time.monotonic() - _last_request)
    if wait > 0:
        time.sleep(wait)
    _last_request = time.monotonic()

    response = session.request(method, BASE_URL + path, timeout=30, **kwargs)
    response.raise_for_status()  # 处理 401、429 等 HTTP 错误

    body = response.json()
    if body["ret"] != 200:  # 处理 HTTP 200 中的业务错误
        raise ApiError(f"ret={body['ret']} msg={body['msg']} traceId={body['traceId']}")
    return body["data"]


def api_get(path, params=None):
    return _request("GET", path, params=params)


def api_post(path, payload):
    return _request("POST", path, json=payload)

示例中用到了这几个标的:

标的代码接口路径WebSocket 通道
台积电(上市)2330.TW/taiwanbusiness=taiwan
环球晶(上柜)6488.TW/taiwanbusiness=taiwan
台积电 ADRTSM.US/stockbusiness=stock
美元/新台币USDTWD/commonbusiness=common

如果使用免费试用,台股行情只能查询2330.TW,TSM.US和USDTWD都不在免费范围内。第 5、7 节的多标的示例和第 8 节的跨市场示例需要付费套餐;只用免费试用的话,把第 5、7 节的代码改成只查2330.TW即可,第 8 节则无法运行。

4. 查询台股代码清单

台股代码统一使用.TW后缀,上市和上柜都一样。这和 Yahoo Finance 不同,Yahoo 的上柜股票用.TWO。从 yfinance 迁移时,6488.TWO要改成6488.TW,否则会查不到。

下面的脚本拉取全部代码,再按交易所分类。exchange字段只区分TWSE和TPEX,而TPEX里同时有上柜和兴柜股票,所以这里再用柜买中心的兴柜清单区分一次。

# list_symbols.py
from collections import Counter

import requests

from tw_api import api_get

symbols = api_get("/common/basic/symbols", {"type": "STOCK_TW"})
codes = [item["symbol"] for item in symbols]

exchange_of = {}
for i in range(0, len(codes), 100):
    batch = ",".join(codes[i:i + 100])
    for info in api_get("/common/basic/symbols/info", {"type": "STOCK_TW", "symbols": batch}):
        exchange_of[info["symbol"]] = info["exchange"]

emerging_rows = requests.get(
    "https://www.tpex.org.tw/openapi/v1/tpex_esb_latest_statistics", timeout=30
).json()
emerging = {row["SecuritiesCompanyCode"] for row in emerging_rows}


def board(code):
    if exchange_of.get(code) == "TWSE":
        return "上市"
    return "兴柜" if code.split(".")[0] in emerging else "上柜"


print(f"共 {len(codes)} 只")
print(Counter(board(code) for code in codes))

运行结果:

共 2356 只
Counter({'上市': 1094, '上柜': 895, '兴柜': 367})

这个分类有实际意义:兴柜股票没有 ±10% 的涨跌幅限制,流动性也和上市、上柜股票差别很大。如果要写全市场扫描,最好先把它们分开。

5. 查询实时报价和盘口

成交和盘口接口都支持批量查询,一次最多 100 只,代码用英文逗号分隔。返回顺序不保证和请求一致,要按字段s匹配,不要按下标取值。

# quote.py
from datetime import datetime
from zoneinfo import ZoneInfo

from tw_api import api_get

CODES = "2330.TW,6488.TW"
TAIPEI = ZoneInfo("Asia/Taipei")

trades = {item["s"]: item for item in api_get(f"/taiwan/batch_trade/{CODES}")}
depths = {item["s"]: item for item in api_get(f"/taiwan/batch_depth/{CODES}")}

for code in CODES.split(","):
    trade = trades[code]
    depth = depths[code]
    ask, ask_shares = float(depth["a"][0][0]), float(depth["a"][1][0])
    bid, bid_shares = float(depth["b"][0][0]), float(depth["b"][1][0])
    trade_time = datetime.fromtimestamp(trade["t"] / 1000, TAIPEI)

    print(
        f"{code} 最新成交 {trade['p']}({trade_time:%H:%M:%S})"
        f" | 卖一 {ask:g} × {ask_shares / 1000:g} 张"
        f" | 买一 {bid:g} × {bid_shares / 1000:g} 张"
        f" | 价差 {ask - bid:g}"
    )

收盘后运行的结果:

2330.TW 最新成交 2235.00(13:24:59) | 卖一 2240 × 1019 张 | 买一 2235 × 92 张 | 价差 5
6488.TW 最新成交 563.00(13:24:59) | 卖一 563 × 16 张 | 买一 560 × 5 张 | 价差 3

几个容易出错的字段:

字段说明
p成交价,字符串类型,计算前要转成数字
v成交量,单位是股
vw成交额,不是成交均价(VWAP)
t成交和盘口都是毫秒时间戳
td交易方向,1 为买、2 为卖、0 为默认值。本文测试的台股成交都返回 0,不建议用它判断主动买卖
a / b卖盘 / 买盘,结构是[[价格], [数量]],台股目前只有一档

收盘价不要取最新成交。 上面的结果里,最新成交时间停在 13:24:59,因为成交明细不包含 13:25 开始的收盘竞价。这个价格可能和真正的收盘价不同:同一天鸿海(2317.TW)的最后一笔成交是 240 元,收盘价却是 239 元。收盘价要从K线取。

6. 获取K线和历史数据

台股K线只支持 POST。常用的klineType包括:1 为 1 分钟,2 为 5 分钟,3 为 15 分钟,4 为 30 分钟,5 为 1 小时,8 为日K,9 为周K,10 为月K。

下面把请求封装成函数,返回按时间升序排列的 DataFrame。第 8 节也会用到它。

# candles.py
import pandas as pd

from tw_api import api_get, api_post

COLUMNS = {"o": "open", "h": "high", "l": "low", "c": "close", "v": "volume", "vw": "amount"}


def get_candles(market, code, kline_type=8, num=100, end_ts=None):
    payload = {"klineType": kline_type, "klineNum": num, "codes": code}
    if end_ts is not None:
        payload["timestamp"] = end_ts

    data = api_post(f"/{market}/v2/batch_kline", payload)
    rows = data[0]["respList"] if data else []
    if not rows:
        return pd.DataFrame()

    df = pd.DataFrame(rows).rename(columns=COLUMNS)
    df["ts"] = df["t"].astype(int)
    df["time"] = pd.to_datetime(df["ts"], unit="s", utc=True)
    for column in COLUMNS.values():
        df[column] = pd.to_numeric(df[column])
    columns = ["ts", "time", *COLUMNS.values()]
    return df[columns].sort_values("time").reset_index(drop=True)


def get_history(market, code, kline_type=8, max_pages=20):
    """从最新数据开始向前翻页,每页最多 500 根"""
    pages = []
    end_ts = None
    for _ in range(max_pages):
        page = get_candles(market, code, kline_type, 500, end_ts)
        if page.empty:
            break
        pages.append(page)
        end_ts = int(page["ts"].iloc[0]) - 1
        if len(page) < 500:
            break
    return pd.concat(pages).sort_values("time").reset_index(drop=True)


def get_forward_factors(code, market, begin_day, end_day):
    """单次查询的跨度不能超过一年,所以按 365 天分段"""
    factors = {}
    day = pd.Timestamp(begin_day)
    last = pd.Timestamp(end_day)
    while day <= last:
        stop = min(day + pd.Timedelta(days=364), last)
        rows = api_get(
            "/common/basic/symbols/adjustment_factors",
            {
                "symbol": code,
                "market": market,
                "beginDay": day.strftime("%Y%m%d"),
                "endDay": stop.strftime("%Y%m%d"),
            },
        )
        factors.update({row["trade_date"]: row["forward_factor"] for row in rows})
        day = stop + pd.Timedelta(days=1)
    return factors

6.1 查询最近的日K

# recent_daily.py
from candles import get_candles

df = get_candles("taiwan", "2330.TW", kline_type=8, num=5)
df["date"] = df["time"].dt.tz_convert("Asia/Taipei").dt.strftime("%Y-%m-%d")
df["lots"] = df["volume"] / 1000  # 换算成张

print(df[["date", "open", "high", "low", "close", "lots"]].to_string(index=False))

运行结果:

      date   open   high    low  close      lots
2026-11-03 2300.0 2335.0 2280.0 2295.0 44826.336
2026-11-04 2320.0 2320.0 2235.0 2260.0 53552.371
2026-11-05 2250.0 2285.0 2235.0 2240.0 36397.000
2026-11-06 2020.0 2195.0 2020.0 2135.0 96920.000
2026-11-09 2210.0 2290.0 2190.0 2235.0 41343.000

日K的时间戳是当地时间的 0 点,所以必须先转成Asia/Taipei再取日期。如果按 UTC 取日期,每根K线都会早一天。

查询多只股票时要注意,每只最多只返回 2 根K线,请求更大的klineNum也一样。要拉多只股票的历史,只能逐只循环。

6.2 翻页拉取完整历史并做复权

接口单次最多返回 500 根K线,数据从新到旧排列。把本页最早一根的时间减 1 秒,作为下一页的timestamp,就能继续向前查询。日K及以上周期可以这样拉取较长的历史,台积电的日K可以一直追溯到 1994 年 9 月上市首日;分钟和小时周期的历史范围则由套餐决定。

K线返回的是未复权价格,和证交所公布的价格一致,适合展示。回测通常需要复权,否则除息日会出现假的下跌。前复权价格等于原始价格乘以当天的forward_factor:

# backfill.py
from candles import get_forward_factors, get_history

CODE = "2330.TW"

df = get_history("taiwan", CODE, kline_type=8)
df["date"] = df["time"].dt.tz_convert("Asia/Taipei").dt.strftime("%Y%m%d")

factors = get_forward_factors(CODE, "TW", df["date"].iloc[0], df["date"].iloc[-1])
df["factor"] = df["date"].map(factors).ffill()  # 最新交易日的因子可能还没生成
df["adj_close"] = (df["close"] * df["factor"]).round(2)

df.to_csv("2330_daily.csv", index=False)

print(f"共 {len(df)} 根,{df['date'].iloc[0]} 至 {df['date'].iloc[-1]}")
print(f"有复权因子的最早日期:{df.loc[df['factor'].notna(), 'date'].iloc[0]}")
sample = df[df["date"].between("20260914", "20260917")]
print(sample[["date", "close", "factor", "adj_close"]].to_string(index=False))

运行结果:

共 7841 根,19940905 至 20261109
有复权因子的最早日期:20000104
    date  close   factor  adj_close
20260914 2380.0 0.997065    2373.01
20260915 2385.0 0.997065    2378.00
20260916 2380.0 1.000000    2380.00
20260917 2425.0 1.000000    2425.00

看 9 月 16 日这一行:原始收盘价比前一天跌了 5 元,但这一天复权因子发生了变化,复权后实际上涨了 2 元。如果直接用原始价格回测,就会把除息当成下跌。

另外,2330.TW的复权因子最早到 2000 年 1 月,更早的adj_close是空值,长周期回测时要留意。

7. 用 WebSocket 接收实时推送

轮询适合低频查询,如果要做实时看盘或信号监控,用 WebSocket 更合适。台股的连接地址要使用business=taiwan,并通过 URL 参数传入apikey。

数据订阅应答推送
成交100001000110002
盘口100031000410005
K线100061000710008

一个能长期运行的客户端要处理好四件事:

  1. 每 30 秒发送一次心跳(协议号 10010),只收行情不发心跳,连接可能被关闭。
  2. 断线后自动重连,并重新发送订阅。服务端不会保留上一个连接的订阅。
  3. 同一根K线会被多次推送,且没有“已收盘”标记,要按s、ty、t更新记录,不能每次都追加。
  4. 部分消息是纯文本,比如订阅失败时的Subscribe fail: ...,不能直接按 JSON 解析。
# stream.py
import asyncio
import json
import uuid

import websockets

API_KEY = "YOUR_API_KEY"
WS_URL = f"wss://data.infoway.io/ws?business=taiwan&apikey={API_KEY}"
CODES = "2330.TW,6488.TW"

candles = {}  # (代码, 周期, 时间) -> 最新的K线


def frame(code, data=None):
    message = {"code": code, "trace": uuid.uuid4().hex}
    if data is not None:
        message["data"] = data
    return json.dumps(message)


async def subscribe(ws):
    await ws.send(frame(10000, {"codes": CODES}))
    await ws.send(frame(10003, {"codes": CODES}))
    await ws.send(frame(10006, {"arr": [{"type": 1, "codes": CODES}]}))


async def heartbeat(ws):
    while True:
        await asyncio.sleep(30)
        await ws.send(frame(10010))


def handle(raw):
    try:
        message = json.loads(raw)
    except json.JSONDecodeError:
        print(f"文本消息:{raw}")
        return

    code, data = message.get("code"), message.get("data")
    if code == 10002:
        print(f"成交 {data['s']} {data['p']} × {float(data['v']) / 1000:g} 张")
    elif code == 10005:
        print(f"盘口 {data['s']} 卖一 {data['a'][0][0]} 买一 {data['b'][0][0]}")
    elif code == 10008:
        candles[(data["s"], data["ty"], data["t"])] = data
        print(f"K线 {data['s']} {data['t']} 最新 {data['c']},已缓存 {len(candles)} 根")
    else:
        print(f"其他消息:{message}")


async def main():
    while True:
        try:
            async with websockets.connect(WS_URL) as ws:
                await subscribe(ws)
                keepalive = asyncio.create_task(heartbeat(ws))
                try:
                    async for raw in ws:
                        handle(raw)
                finally:
                    keepalive.cancel()
        except (OSError, asyncio.TimeoutError, websockets.exceptions.WebSocketException) as error:
            print(f"连接异常:{error}")

        print("连接断开,10 秒后重连")
        await asyncio.sleep(10)


if __name__ == "__main__":
    asyncio.run(main())

在非交易时段运行,只会看到连接成功和订阅应答:

其他消息:{'code': 200, 'msg': 'ws connect success'}
其他消息:{'code': 10001, 'trace': '4315d96a3be84293ad6c96c09b7a8d56', 'msg': 'ok'}
其他消息:{'code': 10004, 'trace': '8700f75f585f4df0a6a8d224756231db', 'msg': 'ok'}
其他消息:{'code': 10007, 'trace': '5dc55eac211943b4beebc09e6627512f', 'msg': 'ok'}

这是正常现象。msg: ok只表示订阅请求已被接收,不代表马上有数据,要到交易时段才会有推送。

另外,同一个连接的订阅、取消订阅和心跳,每分钟合计不能超过 60 次,超过会被断开。需要订阅很多标的时,把代码合并成一次请求,不要逐只发送。

8. 实战:计算台积电 ADR 溢价

台积电同时在台北和纽约交易,1 股 ADR 对应 5 股台股普通股,两边的价格经常不一致。计算溢价需要三份数据,正好可以把前面的内容串起来:

ADR 溢价 = (ADR 价格 × 美元兑新台币汇率 ÷ 5) ÷ 台股价格 − 1

难点在于时间对齐。三类数据的日K时间戳不一样:台股是台北时间 0 点,美股是纽约时间 0 点,外汇是 UTC 0 点。如果直接按日期合并,就会把还没发生的美股收盘价算进当天的台股。

台股和美股的常规交易时段并不重叠。正确做法是先算出每根K线的实际收盘时间,再用merge_asof为每个台股收盘时点匹配“当时已知的最近一个收盘价”:

# adr_premium.py
import pandas as pd

from candles import get_candles

ADR_RATIO = 5
N = 30


def closes(market, code, close_at):
    df = get_candles(market, code, kline_type=8, num=N)
    return pd.DataFrame({"at": close_at(df["time"]), code: df["close"]}).sort_values("at")


def local_close(tz, clock):
    def convert(times):
        local_day = times.dt.tz_convert(tz).dt.normalize()
        return (local_day + pd.Timedelta(clock)).dt.tz_convert("UTC")
    return convert


tw = closes("taiwan", "2330.TW", local_close("Asia/Taipei", "13:30:00"))
adr = closes("stock", "TSM.US", local_close("America/New_York", "16:00:00"))
# 外汇日K的时间是当天 0 点(UTC),向后加一天,避免用到当时还未收盘的汇率
fx = closes("common", "USDTWD", lambda times: times + pd.Timedelta(days=1))

merged = pd.merge_asof(tw, adr, on="at")
merged = pd.merge_asof(merged, fx, on="at").dropna()

implied = merged["TSM.US"] * merged["USDTWD"] / ADR_RATIO
merged["implied_twd"] = implied.round(1)
merged["premium_pct"] = ((implied / merged["2330.TW"] - 1) * 100).round(2)
merged["date"] = merged["at"].dt.tz_convert("Asia/Taipei").dt.strftime("%Y-%m-%d")

columns = ["date", "2330.TW", "TSM.US", "USDTWD", "implied_twd", "premium_pct"]
print(merged[columns].tail(5).to_string(index=False))
print(f"近 {len(merged)} 个交易日平均溢价:{merged['premium_pct'].mean():.2f}%")

运行结果:

      date  2330.TW  TSM.US  USDTWD  implied_twd  premium_pct
2026-11-03   2295.0  398.63 32.5669       2596.4        13.13
2026-11-04   2260.0  404.80 32.7705       2653.1        17.39
2026-11-05   2240.0  395.75 32.7710       2593.8        15.80
2026-11-06   2135.0  353.10 33.1584       2341.6         9.68
2026-11-09   2235.0  375.44 32.9121       2471.3        10.57
近 26 个交易日平均溢价:15.38%

implied_twd是把 ADR 价格换算成每股新台币后的价格。这个结果可以直接用于监控:把它写入数据库,再设一个阈值,就是一个简单的跨市场提醒。需要注意的是,这里比较的是前一晚的美股收盘价和当天的台股收盘价,中间至少隔了 8 个小时,周一的数据还隔了一个周末。溢价的波动里有一部分来自这个时差,不完全是定价差异。

9. 常见问题

9.1 为什么接口的成交量和证交所不一样?

我们抽取了2330.TW在 2026 年 3 月和 11 月共 14 个交易日的日K,和证交所个股日成交资料对比,开高低收全部一致,但成交量常常偏低。以其中 5 天为例:

日期证交所成交股数接口v差异
2026-11-06110,714,58796,920,000-12.5%
2026-11-0545,175,14536,397,000-19.4%
2026-11-0453,879,37153,552,371-0.6%
2026-11-0346,137,33644,826,336-2.8%
2026-11-0241,165,06839,025,068-5.2%

差异的具体构成,接口文档没有说明。如果你的策略依赖精确的日成交量,比如量比或成交额因子,建议用官方的盘后数据校准。价格类指标不受影响。

官方数据可以这样获取,注意Date是民国纪年,1151106表示 2026 年 11 月 6 日,而且盘后才会更新:

# official_check.py
import requests

rows = requests.get(
    "https://openapi.twse.com.tw/v1/exchangeReport/STOCK_DAY_ALL", timeout=30
).json()
row = next(item for item in rows if item["Code"] == "2330")

roc_date = row["Date"]
date = f"{int(roc_date[:3]) + 1911}-{roc_date[3:5]}-{roc_date[5:]}"
print(date, row["ClosingPrice"], row["TradeVolume"])

9.2 yfinance 的数据为什么和证交所有差异?

常见原因有三个:Yahoo 的台股行情有约 20 分钟延迟;yfinance 的history()默认返回复权后的价格(auto_adjust=True),和证交所的原始价格本来就不同;上柜股票要用.TWO后缀,写错后缀可能查不到。

9.3 能查加权指数吗?

需要注意,指数类的TAIEX代码并不是证交所的发行量加权股价指数。它在产品清单中的名称是“台湾 RIC Capped 指数”,2026 年 11 月 6 日的点位约为 3417,而证交所公布的加权指数当天收在 39,701.64,周末也会产生K线。

如果需要官方的加权指数收盘值,可以使用证交所 OpenAPI 的/v1/exchangeReport/MI_INDEX。

9.4 能查 ETF 吗?

目前不能。0050.TW这样的代码会返回ret=508(产品不存在)。如果需要跟踪台湾市场的 ETF,只能查询在其他市场上市的产品,比如美股的EWT.US。

批量查询时,不存在的代码会被静默过滤,只有全部无效才返回 508。所以不要假设请求多少只就会返回多少只,最好核对一下返回的s。

9.5 为什么 WebSocket 订阅成功却没有数据?

先排除三种情况:当前是否在交易时段;business是否为taiwan,通道和标的不匹配时可能照样返回ok,却不会推送数据;代码是否和产品清单完全一致,代码区分大小写。

9.6 分钟K线能查多久以前的?

分钟和小时周期的历史范围由套餐决定,超出范围会返回ret=513,并在msg中给出可查询的最早时间戳。日K及以上周期不做这项限制。

10. 小结

接入台湾行情数据,代码本身并不复杂,容易出问题的是这些细节:成交量单位是股而不是张,收盘价要从K线取,上柜代码的后缀和 Yahoo 不同,兴柜股票不受 ±10% 限制,跨市场计算要按实际收盘时间对齐。

完整的字段说明和频率限制可以查阅接口文档。建议先用2330.TW跑通 HTTP 查询,再接 WebSocket,最后加入自己的存储和业务逻辑。

0
0
0
0
评论
未登录
暂无评论