Python 批量采集台股 TWSE 全市场行情指数 + 个股同步方案

摘要

跨境金融系统开发中，台股 TWSE 上市市场、TPEx 上柜市场数据是亚太行情模块的重要组成部分。本文基于 Python requests 实现台股行情批量采集，覆盖台股加权指数、OTC 指数以及多只上市个股，支持 HTTP 轮询、K 线历史数据获取，附带异常重试、日志记录，可直接部署为定时采集服务。

一、业务背景

台股市场分为 TWSE 上市与 TPEx 上柜两大板块，核心指数包括台股加权指数、OTC 指数，覆盖水泥、电子、金融、航运等众多行业。很多项目对接台股时容易踩几个坑：

单只个股循环请求，效率低，浪费接口次数；
没有 Gzip 压缩，数据传输体积大；
没有重试机制，网络波动直接导致同步失败；
脚本散落在 crontab 里，缺乏统一日志。

二、品种编码说明

脉动行情数据 API 台股市场覆盖：

指数：TWII（台股加权指数）、TWOII（OTC 指数）
个股：1101 台泥、1102 亚泥、1216 统一、1301 台塑、2330 台积电、2317 鸿海、2454 联发科、2882 国泰金、2886 兆丰金等
接口地址：http://39.107.99.235:1008/getQuote.php
K 线接口：http://39.107.99.235:1008/redis.php

三、依赖安装

pip install requests

四、完整 Python 实现代码

import requests
import time
import logging
from typing import List, Dict, Any

# 日志配置
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s"
)
logger = logging.getLogger(__name__)

# 接口配置
QUOTE_URL = "http://39.107.99.235:1008/getQuote.php"
KLINE_URL = "http://39.107.99.235:1008/redis.php"
HEADERS = {"Accept-Encoding": "gzip"}
TIMEOUT = 8
RETRY_TIMES = 3


class TwStockFetcher:
    """台股行情批量采集器"""

    def __init__(self, codes: List[str]):
        self.codes = ",".join(codes)
        self.session = requests.Session()
        self.session.headers.update(HEADERS)

    def _request_with_retry(self, url: str, params: Dict) -> Any:
        """带重试的请求"""
        for i in range(RETRY_TIMES):
            try:
                resp = self.session.get(url, params=params, timeout=TIMEOUT)
                if resp.status_code == 200:
                    return resp.json()
                logger.warning("请求异常状态码: %s", resp.status_code)
            except Exception as e:
                logger.error("第%d次请求失败: %s", i + 1, str(e))
                time.sleep(1)
        return None

    def get_batch_quote(self) -> List[Dict]:
        """批量获取台股实时行情"""
        data = self._request_with_retry(QUOTE_URL, {"code": self.codes})
        if not data or data.get("code") != 200:
            logger.error("接口返回错误")
            return []
        return data["data"]["body"]

    def get_kline(self, code: str, cycle: str = "1d", rows: int = 100) -> str:
        """获取单标的K线数据"""
        params = {"code": code, "time": cycle, "rows": rows}
        data = self._request_with_retry(KLINE_URL, params)
        return data if data else ""

    def run(self):
        """执行一次采集"""
        logger.info("开始采集台股行情: %s", self.codes)
        items = self.get_batch_quote()
        for item in items:
            print(
                f"[{item.get('Time')}] "
                f"{item.get('StockCode')} "
                f"最新价:{item.get('Price')} "
                f"涨跌:{item.get('Diff')} "
                f"涨跌幅:{item.get('DiffRate')}%"
            )
        logger.info("本次采集完成，共%d条", len(items))


if __name__ == "__main__":
    # 批量采集：指数 + 多只个股
    codes = [
        "TWII", "TWOII",
        "1101", "1102", "1216",
        "2330", "2317", "2454",
        "2882", "2886"
    ]
    fetcher = TwStockFetcher(codes)
    fetcher.run()

    # 单独获取台积电日K线
    kline = fetcher.get_kline("2330", "1d", 50)
    print("\n台积电2330日K线数据:")
    print(kline)

五、返回字段说明

表格

字段	说明
StockCode	标的编码
Price	最新价
Open	开盘价
LastClose	昨收价
High	日内最高
Low	日内最低
Diff	涨跌额
DiffRate	涨跌幅
TotalVol	成交量
Time	更新时间

六、定时部署方案

生产环境配合 APScheduler 实现定时采集：

from apscheduler.schedulers.blocking import BlockingScheduler

if __name__ == "__main__":
    scheduler = BlockingScheduler(timezone="Asia/Taipei")
    scheduler.add_job(fetcher.run, "interval", seconds=30)
    logger.info("台股行情采集服务启动")
    scheduler.start()

七、开发注意事项

多个标的用英文逗号拼接在一次请求里，减少 HTTP 连接次数；
使用 requests.Session 复用 TCP 连接，比每次新建请求性能好；
台股交易时段为台北时间，建议定时任务时区设置为 Asia/Taipei；
非交易时段价格不更新，业务层可根据时段降低采集频率；
批量采集结果是数组，遍历解析即可，不要当成单对象。

八、总结

Python 轻量脚本非常适合快速搭建行情采集服务，requests 库简洁易读。本文实现台股指数与个股的批量采集、重试、定时部署，修改标的列表即可扩展到更多 TWSE、TPEx 市场个股，适合跨境行情系统、量化监控类项目。

Python 批量采集台股 TWSE 全市场行情 指数 + 个股同步方案

摘要