使用API获取新加坡股票数据的完整指南
原创
使用API获取新加坡股票数据的完整指南
原创
用户11961746
发布于 2026-03-13 10:32:44
发布于 2026-03-13 10:32:44
使用API获取新加坡股票数据的完整指南
在金融科技开发和量化交易领域，获取准确、实时的股票数据是构建分析系统和交易策略的基础。新加坡作为亚洲重要的金融中心，其股票市场数据对于开发者和投资者具有重要价值。本文将详细介绍如何通过API接口获取新加坡股票数据，并提供完整的实现示例。
一、数据源选择与准备工作
在开始之前，我们需要选择一个可靠的数据源（仅供参考，不构成任何投资建议）。
1.1 获取访问凭证
要使用该数据服务，首先需要获取API密钥。
1.2 安装必要依赖
对于Python开发者，建议安装以下库：
pip install requests pandas matplotlib二、核心接口详解
2.1 获取新加坡股票列表
要获取新加坡交易所（SGX）的股票列表，可以使用以下接口：
import requests
def get\_singapore\_stocks(api\_key, page\_size=20, page=1):
"""获取新加坡股票列表"""
url = "https://api.stocktv.top/stock/stocks"
params = {
"key": api\_key,
"countryId": 43, # 新加坡国家ID
"pageSize": page\_size,
"page": page
}
response = requests.get(url, params=params)
if response.status\_code == 200:
return response.json()
else:
print(f"请求失败: {response.status\_code}")
return None
# 使用示例
api\_key = "您的API密钥"
stocks\_data = get\_singapore\_stocks(api\_key)
if stocks\_data and stocks\_data.get("code") == 200:
for stock in stocks\_data.get("data", {}).get("records", []):
print(f"代码: {stock['symbol']}, 名称: {stock['name']}, 最新价: {stock['last']}")该接口返回的数据包含股票代码、名称、最新价格、涨跌幅、成交量等关键信息。
2.2 获取历史K线数据
对于技术分析和策略回测，历史K线数据至关重要：
def get\_historical\_kline(api\_key, pid, interval="P1D", limit=100):
"""获取股票历史K线数据
参数:
- pid: 股票产品ID
- interval: 时间间隔
PT5M: 5分钟
PT15M: 15分钟
PT1H: 1小时
P1D: 日线
P1W: 周线
P1M: 月线
- limit: 数据条数
"""
url = "https://api.stocktv.top/stock/kline"
params = {
"pid": pid,
"interval": interval,
"limit": limit,
"key": api\_key
}
response = requests.get(url, params=params)
if response.status\_code == 200:
return response.json()
else:
print(f"请求失败: {response.status\_code}")
return None
# 使用示例
kline\_data = get\_historical\_kline(api\_key, pid=60231, interval="P1D", limit=50)
if kline\_data and kline\_data.get("code") == 200:
for candle in kline\_data.get("data", []):
print(f"时间: {candle['time']}, 开盘: {candle['open']}, 最高: {candle['high']}, 最低: {candle['low']}, 收盘: {candle['close']}")2.3 获取新加坡海峡时报指数（STI）
海峡时报指数是衡量新加坡市场表现的核心指标：
def get\_singapore\_indices(api\_key):
"""获取新加坡指数数据"""
url = "https://api.stocktv.top/stock/indices"
params = {
"countryId": 15, # 新加坡指数标识
"key": api\_key
}
response = requests.get(url, params=params)
if response.status\_code == 200:
return response.json()
else:
print(f"请求失败: {response.status\_code}")
return None三、实战应用场景
3.1 构建个股监控系统
通过简单的API调用，可以实时监控特定股票的价位变动和成交量异常：
import time
from datetime import datetime
class StockMonitor:
def \_\_init\_\_(self, api\_key, stock\_pid, alert\_threshold=0.05):
self.api\_key = api\_key
self.stock\_pid = stock\_pid
self.alert\_threshold = alert\_threshold
self.last\_price = None
def monitor\_price(self):
"""监控股票价格变动"""
while True:
current\_data = self.get\_current\_price()
if current\_data:
current\_price = current\_data['last']
if self.last\_price is not None:
price\_change = (current\_price - self.last\_price) / self.last\_price
if abs(price\_change) > self.alert\_threshold:
self.send\_alert(f"价格异常波动: {price\_change\*100:.2f}%")
self.last\_price = current\_price
print(f"{datetime.now()}: 当前价格: {current\_price}")
time.sleep(60) # 每分钟检查一次
def get\_current\_price(self):
"""获取当前价格"""
url = "https://api.stocktv.top/stock/queryStocks"
params = {
"id": self.stock\_pid,
"key": self.api\_key
}
response = requests.get(url, params=params)
if response.status\_code == 200:
data = response.json()
if data.get("code") == 200:
return data.get("data", [{}])[0]
return None
def send\_alert(self, message):
"""发送警报"""
print(f"警报: {message}")
# 这里可以集成邮件、短信或推送通知3.2 开发量化交易策略
获取历史数据后，可以进行策略回测分析：
import pandas as pd
import numpy as np
class StrategyBacktester:
def \_\_init\_\_(self, api\_key):
self.api\_key = api\_key
def backtest\_moving\_average(self, pid, short\_window=10, long\_window=30):
"""移动平均线策略回测"""
# 获取历史数据
kline\_data = get\_historical\_kline(self.api\_key, pid, interval="P1D", limit=200)
if not kline\_data or kline\_data.get("code") != 200:
return None
# 转换为DataFrame
df = pd.DataFrame(kline\_data.get("data", []))
df['time'] = pd.to\_datetime(df['time'], unit='ms')
df.set\_index('time', inplace=True)
# 计算移动平均线
df['short\_ma'] = df['close'].rolling(window=short\_window).mean()
df['long\_ma'] = df['close'].rolling(window=long\_window).mean()
# 生成交易信号
df['signal'] = 0
df['signal'][short\_window:] = np.where(
df['short\_ma'][short\_window:] > df['long\_ma'][short\_window:], 1, 0
)
df['positions'] = df['signal'].diff()
# 计算收益率
df['returns'] = df['close'].pct\_change()
df['strategy\_returns'] = df['returns'] \* df['signal'].shift(1)
return df四、开发注意事项
4.1 交易时间处理
新加坡股市交易时间通常为北京时间09:00-17:00（含午休），在非交易时段，价格数据将保持为收盘价。开发时需要考虑这一特性。
4.2 错误处理与频率限制
建议在代码中增加完善的错误处理机制：
def safe\_api\_call(func, max\_retries=3):
"""安全的API调用装饰器"""
def wrapper(\*args, \*\*kwargs):
for attempt in range(max\_retries):
try:
result = func(\*args, \*\*kwargs)
if result and result.get("code") == 200:
return result
elif result and result.get("code") != 200:
print(f"API返回错误: {result.get('message')}")
time.sleep(2 \*\* attempt) # 指数退避
except Exception as e:
print(f"第{attempt+1}次尝试失败: {e}")
time.sleep(2 \*\* attempt)
return None
return wrapper4.3 数据单位与货币
SGX股票通常以新加坡元（SGD）计价，处理跨境投资应用时需要注意汇率转换。
五、性能优化建议
5.1 使用WebSocket实时数据
对于需要实时数据的应用场景，建议使用WebSocket协议：
import websocket
import json
class RealTimeDataClient:
def \_\_init\_\_(self, api\_key):
self.api\_key = api\_key
self.ws = None
def connect(self):
"""连接WebSocket服务器"""
ws\_url = f"wss://api.stocktv.top/ws?key={self.api\_key}"
self.ws = websocket.WebSocketApp(
ws\_url,
on\_message=self.on\_message,
on\_error=self.on\_error,
on\_close=self.on\_close
)
self.ws.on\_open = self.on\_open
self.ws.run\_forever()
def subscribe\_stock(self, symbol):
"""订阅股票实时数据"""
if self.ws:
subscribe\_msg = {
"action": "subscribe",
"symbol": symbol,
"channel": "stock"
}
self.ws.send(json.dumps(subscribe\_msg))
def on\_message(self, ws, message):
"""处理接收到的消息"""
data = json.loads(message)
print(f"实时数据: {data}")
def on\_error(self, ws, error):
print(f"WebSocket错误: {error}")
def on\_close(self, ws, close\_status\_code, close\_msg):
print("WebSocket连接关闭")
def on\_open(self, ws):
print("WebSocket连接已建立")5.2 数据缓存策略
对于频繁访问的数据，建议实现缓存机制以减少API调用：
from functools import lru\_cache
import time
class CachedDataFetcher:
def \_\_init\_\_(self, api\_key, cache\_ttl=300):
self.api\_key = api\_key
self.cache\_ttl = cache\_ttl
self.cache = {}
@lru\_cache(maxsize=128)
def get\_cached\_data(self, endpoint, params):
"""带缓存的数据获取"""
cache\_key = f"{endpoint}\_{hash(frozenset(params.items()))}"
if cache\_key in self.cache:
cached\_data, timestamp = self.cache[cache\_key]
if time.time() - timestamp < self.cache\_ttl:
return cached\_data
# 调用API获取新数据
url = f"https://api.stocktv.top/{endpoint}"
params['key'] = self.api\_key
response = requests.get(url, params=params)
if response.status\_code == 200:
data = response.json()
self.cache[cache\_key] = (data, time.time())
return data
return None六、总结
通过本文介绍的API接口，开发者可以轻松获取新加坡股票市场的实时行情、历史数据和指数信息。这些数据接口设计合理，支持RESTful API和WebSocket两种协议，能够满足不同场景下的数据需求。
无论是构建个股监控系统、开发量化交易策略，还是进行市场分析研究，这些接口都提供了可靠的数据支持。在实际开发过程中，建议关注错误处理、频率限制和数据缓存等关键点，以确保应用的稳定性和性能。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
评论
登录后参与评论
推荐阅读
目录
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号