新加坡股票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 依赖库
本教程需要安装requests和websocket-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 接入新加坡股票市场的完整流程:
- REST API 实时行情:简单快速地获取新加坡股票的实时报价,适合批量查询和定时任务
- REST API 历史数据:获取多周期 K 线数据,支持策略回测和趋势分析
- WebSocket 实时推送:低延迟订阅新加坡股票的成交、报价和盘口数据,适用于高频交易和实时监控
- Tick 数据获取:逐笔成交数据,满足精细化分析需求
iTick 的新加坡股票数据服务具有以下核心优势:
- 全面覆盖:支持新加坡交易所全部股票
- 低延迟:WebSocket 延迟<50ms,满足量化交易需求
- 开发者友好:Python SDK 完善,对接极简
- 性价比高:免费版够用,付费版实惠
现在就访问 iTick 官网 注册账号,开始您的新加坡股市数据之旅吧!
延伸阅读: