新加坡股票API接入指南:新交所(SGX)实时行情与历史数据获取

  1. iTick
  2. 教程
新加坡股票API接入指南:新交所(SGX)实时行情与历史数据获取 - iTick
新加坡股票API接入指南:新交所(SGX)实时行情与历史数据获取

新加坡作为亚洲重要的金融中心,其股市汇聚了众多东南亚龙头企业。从星展银行到新加坡电信,从扬子江船业到凯德集团,新加坡交易所(SGX)的上市公司覆盖金融、电信、房地产、航运等多个领域,成为全球投资者布局东南亚市场的重要门户。

本文将为您详细讲解如何通过iTick API一站式接入新加坡股票市场,获取实时行情、历史 K 线及深度盘口数据,并提供完整的 Python 实战代码,助您快速构建新加坡股市数据分析系统。

一、为什么选择 iTick 接入新加坡股市?

在众多数据服务商中,iTick 对开发者非常友好,特别适合需要对接新加坡(SG)等多市场的量化团队 。

市场覆盖全面:除了新加坡,还支持美国、香港、台湾、日本、印度、马来西亚、泰国等多个国家和地区的股票,一个 API 搞定全球数据。

协议选择灵活:提供 RESTful API(适合单次查询)、WebSocket(适合实时推送)和 FIX API(适合机构级高频),满足不同场景需求。

开发者体验友好:提供免费套餐,新用户注册即可获取免费 Token;文档简洁,并附有 Python、Java、Go 等多语言示例。

数据质量专业:数据延迟低至毫秒级,支持成交(tick)、报价(quote)、盘口(depth)等多种数据类型,满足量化交易的不同需求。。

二、环境准备

在开始接入之前,需要完成以下准备工作:

1. 注册 iTick 账号并获取 API Token

访问 iTick 官网 注册账号,30 秒完成注册,无需信用卡。在控制台即可找到您的专属 API Token。

2. 安装 Python 依赖库

本教程需要安装requestswebsocket-client库:

      pip install requests websocket-client pandas matplotlib

    

三、新加坡股票实时行情接入(REST API)

对于不需要实时推送的场景,REST API 是最简单直接的获取方式。以下代码示例获取新加坡三大代表性公司的实时行情:

  • 星展银行(D05) - 东南亚最大银行
  • 华侨银行(O39) - 新加坡主要金融机构
  • 新加坡电信(Z74) - 新加坡领先电信运营商
      import requests
import datetime

# 配置您的API Token
API_TOKEN = "your_api_token_here"  # 替换为您的实际Token
BASE_URL = "https://api.itick.org"

def get_singapore_stock_quote(symbol):
    """
    获取新加坡股票实时报价
    :param symbol: 股票代码,如"D05"(星展银行)
    """
    url = f"{BASE_URL}/stock/quote"
    params = {
        "region": "SG",      # 新加坡市场代码
        "code": symbol       # 股票代码
    }
    headers = {
        "accept": "application/json",
        "token": API_TOKEN
    }

    try:
        response = requests.get(url, params=params, headers=headers, timeout=5)
        response.raise_for_status()
        result = response.json()

        if result.get("code") == 0:  # 0表示成功
            data = result.get("data", {})
            print(f"📊 股票名称: {data.get('n', 'N/A')}")
            print(f"股票代码: {data.get('s', 'N/A')}")
            print(f"最新价: {data.get('ld', 'N/A')} SGD")
            print(f"开盘价: {data.get('o', 'N/A')} SGD")
            print(f"最高价: {data.get('h', 'N/A')} SGD")
            print(f"最低价: {data.get('l', 'N/A')} SGD")
            print(f"成交量: {data.get('v', 'N/A')} 股")
            print(f"涨跌幅: {data.get('chp', 'N/A')}%")

            # 转换时间戳
            timestamp = data.get('t', 0) / 1000
            if timestamp > 0:
                dt = datetime.datetime.fromtimestamp(timestamp)
                print(f"数据时间: {dt.strftime('%Y-%m-%d %H:%M:%S')} (新加坡时间)")
            return data
        else:
            print(f"❌ API错误: {result.get('msg', 'Unknown error')}")
            return None

    except Exception as e:
        print(f"❌ 请求异常: {str(e)}")
        return None

# 批量获取多只新加坡股票实时行情
def get_multiple_quotes(symbols):
    """批量获取多只股票实时行情"""
    for symbol in symbols:
        print(f"\n{'='*50}")
        print(f"获取 {symbol} 实时行情...")
        get_singapore_stock_quote(symbol)

# 调用示例:获取星展银行实时行情
print("🔍 获取星展银行(D05)实时行情:")
get_singapore_stock_quote("D05")

# 批量获取多只新加坡股票
singapore_stocks = ["D05", "O39", "Z74"]  # 星展银行、华侨银行、新电信
get_multiple_quotes(singapore_stocks)

    

执行上述代码,即可获取新加坡股票的实时交易数据。返回的 JSON 结构包含完整的报价信息,可直接用于量化策略的输入。

四、获取历史 K 线数据(策略回测必备)

历史数据是量化回测的基础。以下示例获取新加坡电信(代码:Z74)的日线历史数据,并进行简单可视化:

      import requests
import pandas as pd
import matplotlib.pyplot as plt
from datetime import datetime, timedelta

def get_singapore_stock_kline(symbol, ktype=8, limit=100):
    """
    获取新加坡股票历史K线数据
    :param symbol: 股票代码,如"Z74"(新加坡电信)
    :param ktype: K线类型 (1:1分钟, 2:5分钟, 3:15分钟, 4:30分钟, 5:60分钟, 8:日线, 9:周线, 10:月线)
    :param limit: 获取K线数量
    :return: DataFrame格式的K线数据
    """
    url = f"{BASE_URL}/stock/kline"
    params = {
        "region": "SG",      # 新加坡市场代码
        "code": symbol,
        "kType": ktype,
        "limit": limit
    }
    headers = {
        "accept": "application/json",
        "token": API_TOKEN
    }

    try:
        response = requests.get(url, params=params, headers=headers)
        result = response.json()

        if result.get("code") == 0:
            kline_data = result.get("data", [])

            if not kline_data:
                print(f"❌ 未获取到{symbol}的数据")
                return None

            # 转换为Pandas DataFrame便于分析
            df = pd.DataFrame(kline_data)

            # 转换时间戳并设置索引
            df['datetime'] = pd.to_datetime(df['t'], unit='ms')
            # 转换为新加坡时间(UTC+8)
            df['datetime'] = df['datetime'].dt.tz_localize('UTC').dt.tz_convert('Asia/Singapore')
            df.set_index('datetime', inplace=True)

            # 重命名列
            df.rename(columns={
                'o': 'open', 'h': 'high', 'l': 'low',
                'c': 'close', 'v': 'volume'
            }, inplace=True)

            # 转换为数值类型
            for col in ['open', 'high', 'low', 'close', 'volume']:
                df[col] = pd.to_numeric(df[col])

            print(f"✅ 获取到 {len(df)} 条K线数据")
            print(f"数据范围: {df.index[0]}{df.index[-1]}")

            # 显示最新5条数据
            print("\n📊 最新5条K线数据:")
            print(df[['open', 'high', 'low', 'close', 'volume']].tail())

            return df
        else:
            print(f"❌ API错误: {result.get('msg', '未知错误')}")
            return None

    except Exception as e:
        print(f"❌ 请求异常: {str(e)}")
        return None

def plot_stock_data(df, symbol):
    """绘制股票K线收盘价走势"""
    if df is None or df.empty:
        return

    plt.figure(figsize=(14, 7))
    plt.plot(df.index, df['close'], 'b-', linewidth=1.5, label='收盘价')
    plt.title(f'{symbol} 收盘价走势图(新加坡市场)', fontsize=16)
    plt.xlabel('日期')
    plt.ylabel('价格 (SGD)')
    plt.grid(True, alpha=0.3)
    plt.legend()

    # 格式化x轴日期
    plt.gcf().autofmt_xdate()

    # 添加简单的移动平均线
    df['MA20'] = df['close'].rolling(window=20).mean()
    plt.plot(df.index, df['MA20'], 'r--', linewidth=1, label='20日均线')
    plt.legend()

    plt.tight_layout()
    plt.show()

# 获取新加坡电信最近100个交易日数据
print("\n🔍 获取新加坡电信(Z74)历史数据...")
singtel_df = get_singapore_stock_kline("Z74", ktype=8, limit=100)

if singtel_df is not None:
    # 绘制走势图
    plot_stock_data(singtel_df, "Z74 (新加坡电信)")

    # 计算简单统计指标
    print(f"\n📈 统计指标:")
    print(f"最新收盘价: {singtel_df['close'].iloc[-1]:.3f} SGD")
    print(f"期间最高价: {singtel_df['high'].max():.3f} SGD")
    print(f"期间最低价: {singtel_df['low'].min():.3f} SGD")
    print(f"平均成交量: {singtel_df['volume'].mean():.0f} 股")
    print(f"期间涨跌幅: {((singtel_df['close'].iloc[-1] / singtel_df['close'].iloc[0] - 1) * 100):.2f}%")

    

这段代码不仅获取了历史数据,还展示了如何使用 Pandas 进行数据处理和简单可视化,方便直接用于策略回测。

五、WebSocket 实时行情订阅(低延迟策略)

对于需要实时监控的高频交易策略,WebSocket 是更好的选择。以下示例展示如何订阅新加坡股票实时行情:

      import websocket
import json
import threading
import time

WS_URL = "wss://api.itick.org/stock"
API_TOKEN = "your_api_token_here"  # 替换为您的实际Token

def on_message(ws, message):
    """处理接收到的消息"""
    data = json.loads(message)

    # 处理连接成功的消息
    if data.get("code") == 1 and data.get("msg") == "Connected Successfully":
        print("✅ WebSocket连接成功,等待认证...")

    # 处理认证结果
    elif data.get("resAc") == "auth":
        if data.get("code") == 1:
            print("✅ 认证成功,准备订阅数据...")
            subscribe(ws)
        else:
            print("❌ 认证失败")
            ws.close()

    # 处理订阅结果
    elif data.get("resAc") == "subscribe":
        if data.get("code") == 1:
            print("✅ 订阅成功")
        else:
            print(f"❌ 订阅失败: {data.get('msg')}")

    # 处理市场数据
    elif data.get("data"):
        market_data = data["data"]
        data_type = market_data.get("type")
        symbol = market_data.get("s")

        if data_type == "quote":
            print(f"[{symbol}] 最新价: {market_data.get('ld')} SGD | "
                  f"涨跌幅: {market_data.get('chp')}% | "
                  f"成交量: {market_data.get('v')}")
        elif data_type == "tick":
            print(f"[{symbol}] 成交: {market_data.get('ld')} SGD | "
                  f"时间: {time.strftime('%H:%M:%S', time.localtime(market_data.get('t')/1000))}")
        elif data_type == "depth":
            bids = market_data.get("b", [])[:3]  # 买三档
            asks = market_data.get("a", [])[:3]  # 卖三档
            print(f"[{symbol}] 买盘: {bids} | 卖盘: {asks}")

def on_error(ws, error):
    """处理错误"""
    print(f"❌ WebSocket错误: {error}")

def on_close(ws, close_status_code, close_msg):
    """连接关闭回调"""
    print(f"🔌 WebSocket连接关闭: {close_msg}")

def on_open(ws):
    """连接建立后的回调"""
    print("🌐 WebSocket连接已打开")

def subscribe(ws):
    """订阅新加坡股票行情数据"""
    subscribe_msg = {
        "ac": "subscribe",
        "params": "D05$SG,O39$SG,Z74$SG",  # 新加坡股票代码:星展银行、华侨银行、新电信
        "types": "tick,quote,depth"          # 订阅成交、报价和盘口数据
    }
    ws.send(json.dumps(subscribe_msg))
    print(f"📤 订阅请求已发送: {subscribe_msg['params']}")

def send_ping(ws):
    """定期发送心跳包保持连接"""
    while True:
        time.sleep(30)  # 每30秒发送一次心跳
        ping_msg = {
            "ac": "ping",
            "params": str(int(time.time() * 1000))
        }
        ws.send(json.dumps(ping_msg))
        print("💓 心跳已发送")

if __name__ == "__main__":
    # 创建 WebSocket 连接,通过header传递token
    ws = websocket.WebSocketApp(
        WS_URL,
        header={"token": API_TOKEN},
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )

    # 在单独的线程中启动心跳机制
    ping_thread = threading.Thread(target=send_ping, args=(ws,))
    ping_thread.daemon = True
    ping_thread.start()

    print("🚀 启动WebSocket连接...")
    # 启动 WebSocket 连接
    ws.run_forever()

    

运行这段代码后,您将实时接收到星展银行、华侨银行和新加坡电信的报价更新和盘口变化,延迟低于 50ms。

六、获取 Tick 数据(逐笔成交)

对于需要精细化分析的量化策略,Tick 数据(逐笔成交)是重要输入。以下示例获取新加坡股票的实时 Tick 数据:

      def get_singapore_tick_data(symbol):
    """
    获取新加坡股票实时Tick数据(逐笔成交)
    :param symbol: 股票代码,如"D05"
    """
    url = f"{BASE_URL}/stock/tick"
    params = {
        "region": "SG",      # 新加坡市场代码
        "code": symbol       # 股票代码
    }
    headers = {
        "accept": "application/json",
        "token": API_TOKEN
    }

    try:
        response = requests.get(url, params=params, headers=headers)
        result = response.json()

        if result.get("code") == 0:
            tick_data = result.get("data", {})
            print(f"📊 股票: {tick_data.get('s', 'N/A')}")
            print(f"最新成交价: {tick_data.get('ld', 'N/A')} SGD")
            print(f"成交量: {tick_data.get('v', 'N/A')} 股")

            # 转换时间戳
            timestamp = tick_data.get('t', 0) / 1000
            if timestamp > 0:
                dt = datetime.datetime.fromtimestamp(timestamp)
                print(f"成交时间: {dt.strftime('%Y-%m-%d %H:%M:%S')} (新加坡时间)")
            return tick_data
        else:
            print(f"❌ API错误: {result.get('msg', '未知错误')}")
            return None

    except Exception as e:
        print(f"❌ 请求异常: {str(e)}")
        return None

# 获取星展银行实时Tick数据
print("\n🔍 获取星展银行(D05)实时Tick数据:")
get_singapore_tick_data("D05")

    

七、新加坡市场实用指南

1. 股票代码格式

新加坡市场股票代码采用本地代码(如 D05、O39、Z74),iTick API 中通过region=SG参数指定市场。WebSocket 订阅时,格式为代码$SG

2. 主要新加坡股票代码参考

公司名称股票代码所属板块业务简介
DBS Bank (星展银行)D05金融东南亚最大银行
OCBC Bank (华侨银行)O39金融新加坡主要银行
United Overseas Bank (大华银行)U11金融新加坡主要银行
Singapore Telecom (新电信)Z74电信新加坡领先电信运营商
Singtel (新电信)Z74电信同上(注意 Z74 和 Singtel 为同一公司)
Keppel Corp (吉宝企业)BN4综合企业航运、房地产等
Sembcorp Industries (胜科工业)U96能源能源、水务
CapitaLand (凯德集团)C31房地产亚洲领先房地产集团
Yangzijiang Shipbuilding (扬子江船业)BS6航运中国在新加坡上市的造船企业
Wilmar International (丰益国际)F34农业全球领先农产品企业

3. 交易时间

新加坡股市交易时间(新加坡交易所):

  • 开盘时间:当地时间 9:00
  • 收盘时间:当地时间 17:00
  • 午间休市:12:00 - 13:00
  • 对应北京时间:与新加坡时间相同(无时差)

4. 常见问题

Q: 免费套餐有哪些限制? A: 免费套餐包含基础实时行情无限调用、历史日线数据访问和 WebSocket 连接支持,对于个人学习和轻量级开发完全够用。

Q: 如何获取更长时间的历史数据? A: 付费套餐支持超 15 年的完整历史数据,适合专业回测需求。

Q: WebSocket 连接不稳定怎么办? A: 代码中已包含心跳机制(每 30 秒发送 ping),可有效保持连接稳定。如仍遇断开,可实现自动重连逻辑。

Q: 新加坡股票代码是否区分大小写? A: 不区分,但建议使用官方格式(如 D05、O39 等)。

八、总结

通过本文的详细教程,您已经掌握了如何使用 iTick API 接入新加坡股票市场的完整流程:

  1. REST API 实时行情:简单快速地获取新加坡股票的实时报价,适合批量查询和定时任务
  2. REST API 历史数据:获取多周期 K 线数据,支持策略回测和趋势分析
  3. WebSocket 实时推送:低延迟订阅新加坡股票的成交、报价和盘口数据,适用于高频交易和实时监控
  4. Tick 数据获取:逐笔成交数据,满足精细化分析需求

iTick 的新加坡股票数据服务具有以下核心优势:

  • 全面覆盖:支持新加坡交易所全部股票
  • 低延迟:WebSocket 延迟<50ms,满足量化交易需求
  • 开发者友好:Python SDK 完善,对接极简
  • 性价比高:免费版够用,付费版实惠

现在就访问 iTick 官网 注册账号,开始您的新加坡股市数据之旅吧!


延伸阅读: