获取新加坡股票历史数据分析：使用接口文档的完整指南

在金融数据分析和量化交易领域，获取准确、实时的股票历史数据是进行策略回测和市场分析的基础。新加坡作为亚洲重要的金融中心之一，其股票市场数据对于投资者和开发者具有重要价值。本文将详细介绍如何通过API获取新加坡股票的历史数据，并提供完整的Python实现示例。

二、准备工作

1. 获取API密钥

准备密钥

2. 安装必要依赖

pip install requests pandas matplotlib

三、新加坡市场数据获取

1. 确定新加坡市场标识

根据StockTV API文档，不同市场通过countryId参数进行区分。虽然公开文档中没有明确列出新加坡的countryId，但根据API的全球覆盖特性，新加坡市场是支持的。在实际使用中，您可以通过以下方式获取正确的标识符：

查阅官方API文档
联系技术支持获取准确参数

2. 核心接口说明

获取股票列表接口

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": "SG",  # 新加坡市场标识（示例）
        "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

获取历史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 = {
   
        "key": api_key,
        "pid": pid,
        "interval": interval,
        "limit": limit
    }

    response = requests.get(url, params=params)
    if response.status_code == 200:
        return response.json()
    else:
        print(f"请求失败: {response.status_code}")
        return None

四、完整示例：新加坡股票历史数据分析

下面是一个完整的Python示例，展示如何获取新加坡某只股票的历史数据并进行基本分析：

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

class SingaporeStockAnalyzer:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.stocktv.top"

    def get_stock_info(self, symbol):
        """查询特定股票信息"""
        url = f"{self.base_url}/stock/queryStocks"
        params = {
   
            "key": self.api_key,
            "symbol": symbol
        }

        response = requests.get(url, params=params)
        if response.status_code == 200:
            data = response.json()
            if data.get("code") == 200 and data.get("data"):
                return data["data"][0]
        return None

    def get_historical_data(self, pid, interval="P1D", days=30):
        """获取历史数据并转换为DataFrame"""
        url = f"{self.base_url}/stock/kline"
        params = {
   
            "key": self.api_key,
            "pid": pid,
            "interval": interval
        }

        response = requests.get(url, params=params)
        if response.status_code == 200:
            data = response.json()
            if data.get("code") == 200 and data.get("data"):
                # 转换为DataFrame
                df = pd.DataFrame(data["data"])

                # 转换时间戳
                df['date'] = pd.to_datetime(df['time'], unit='ms')
                df.set_index('date', inplace=True)

                # 保留核心字段
                df = df[['open', 'high', 'low', 'close', 'volume']]

                return df
        return None

    def calculate_technical_indicators(self, df):
        """计算技术指标"""
        # 移动平均线
        df['MA5'] = df['close'].rolling(window=5).mean()
        df['MA20'] = df['close'].rolling(window=20).mean()

        # 相对强弱指数（RSI）
        delta = df['close'].diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
        rs = gain / loss
        df['RSI'] = 100 - (100 / (1 + rs))

        # 布林带
        df['Middle Band'] = df['close'].rolling(window=20).mean()
        df['Upper Band'] = df['Middle Band'] + 2 * df['close'].rolling(window=20).std()
        df['Lower Band'] = df['Middle Band'] - 2 * df['close'].rolling(window=20).std()

        return df

    def plot_stock_chart(self, df, title="新加坡股票价格走势"):
        """绘制股票图表"""
        fig, axes = plt.subplots(3, 1, figsize=(14, 10))

        # 价格走势图
        axes[0].plot(df.index, df['close'], label='收盘价', color='blue', linewidth=2)
        axes[0].plot(df.index, df['MA5'], label='5日均线', color='orange', linestyle='--')
        axes[0].plot(df.index, df['MA20'], label='20日均线', color='green', linestyle='--')
        axes[0].fill_between(df.index, df['Upper Band'], df['Lower Band'], alpha=0.2, color='gray')
        axes[0].set_title(f'{title} - 价格走势')
        axes[0].set_ylabel('价格')
        axes[0].legend()
        axes[0].grid(True, alpha=0.3)

        # 成交量图
        axes[1].bar(df.index, df['volume'], color='blue', alpha=0.6)
        axes[1].set_title('成交量')
        axes[1].set_ylabel('成交量')
        axes[1].grid(True, alpha=0.3)

        # RSI图
        axes[2].plot(df.index, df['RSI'], label='RSI', color='purple', linewidth=2)
        axes[2].axhline(y=70, color='red', linestyle='--', alpha=0.5, label='超买线')
        axes[2].axhline(y=30, color='green', linestyle='--', alpha=0.5, label='超卖线')
        axes[2].set_title('相对强弱指数(RSI)')
        axes[2].set_ylabel('RSI')
        axes[2].set_xlabel('日期')
        axes[2].legend()
        axes[2].grid(True, alpha=0.3)

        plt.tight_layout()
        plt.show()

    def generate_analysis_report(self, df):
        """生成分析报告"""
        latest_price = df['close'].iloc[-1]
        price_change = ((latest_price - df['close'].iloc[0]) / df['close'].iloc[0]) * 100

        avg_volume = df['volume'].mean()
        volatility = df['close'].pct_change().std() * 100

        report = {
   
            "分析期间": f"{df.index[0].date()} 至 {df.index[-1].date()}",
            "数据点数": len(df),
            "最新收盘价": f"{latest_price:.2f}",
            "期间涨跌幅": f"{price_change:.2f}%",
            "平均日成交量": f"{avg_volume:,.0f}",
            "价格波动率": f"{volatility:.2f}%",
            "最高价": f"{df['high'].max():.2f}",
            "最低价": f"{df['low'].min():.2f}",
            "当前RSI": f"{df['RSI'].iloc[-1]:.2f}" if 'RSI' in df.columns else "N/A"
        }

        return report

# 使用示例
def main():
    # 替换为您的API密钥
    API_KEY = "YOUR_API_KEY_HERE"

    # 初始化分析器
    analyzer = SingaporeStockAnalyzer(API_KEY)

    # 示例：分析新加坡银行股（假设股票代码为DBS）
    stock_info = analyzer.get_stock_info("DBS")
    if stock_info:
        print(f"股票信息: {stock_info['name']} ({stock_info['symbol']})")
        print(f"最新价格: {stock_info['last']}")
        print(f"涨跌幅: {stock_info['chgPct']}%")

        # 获取历史数据
        historical_data = analyzer.get_historical_data(
            pid=stock_info['id'],
            interval="P1D",
            days=90
        )

        if historical_data is not None:
            # 计算技术指标
            analyzed_data = analyzer.calculate_technical_indicators(historical_data)

            # 生成分析报告
            report = analyzer.generate_analysis_report(analyzed_data)
            print("\n=== 分析报告 ===")
            for key, value in report.items():
                print(f"{key}: {value}")

            # 绘制图表
            analyzer.plot_stock_chart(analyzed_data, title=f"{stock_info['name']} 技术分析")
        else:
            print("无法获取历史数据")
    else:
        print("无法获取股票信息")

if __name__ == "__main__":
    main()

五、数据处理与存储建议

1. 数据缓存策略

由于股票数据更新频繁但某些数据变化不大，建议实现缓存机制：

import redis
import json
from datetime import timedelta

class StockDataCache:
    def __init__(self, redis_host='localhost', redis_port=6379):
        self.redis_client = redis.Redis(host=redis_host, port=redis_port, decode_responses=True)

    def get_cached_data(self, cache_key):
        """获取缓存数据"""
        cached = self.redis_client.get(cache_key)
        if cached:
            return json.loads(cached)
        return None

    def set_cache_data(self, cache_key, data, expiry_hours=1):
        """设置缓存数据"""
        self.redis_client.setex(
            cache_key,
            timedelta(hours=expiry_hours),
            json.dumps(data)
        )

2. 数据持久化

import sqlite3
import pandas as pd

class StockDataStorage:
    def __init__(self, db_path='stock_data.db'):
        self.conn = sqlite3.connect(db_path)

    def save_historical_data(self, symbol, df):
        """保存历史数据到数据库"""
        df['symbol'] = symbol
        df['created_at'] = pd.Timestamp.now()

        df.to_sql('historical_prices', self.conn, if_exists='append', index=False)

    def get_historical_data(self, symbol, start_date, end_date):
        """从数据库获取历史数据"""
        query = """
        SELECT * FROM historical_prices 
        WHERE symbol = ? AND date BETWEEN ? AND ?
        ORDER BY date
        """
        return pd.read_sql_query(query, self.conn, params=[symbol, start_date, end_date])

六、注意事项

API限制：注意查看API的请求频率限制，避免过度请求导致被封禁
错误处理：妥善处理网络错误和API返回的错误信息
数据准确性：重要决策前应验证数据的准确性和时效性
密钥安全：妥善保管API密钥，不要在前端代码中暴露
时区处理：新加坡市场时间与北京时间相同，无需时区转换

七、总结

本文不构成任何投资建议。本文提供的代码示例和技术方案，可以帮助您快速上手新加坡股票历史数据的获取与分析。

获取新加坡股票历史数据分析：使用接口文档的完整指南

获取新加坡股票历史数据分析：使用接口文档的完整指南

二、准备工作

1. 获取API密钥

2. 安装必要依赖

三、新加坡市场数据获取

1. 确定新加坡市场标识

2. 核心接口说明

获取股票列表接口

获取历史K线数据接口

四、完整示例：新加坡股票历史数据分析

五、数据处理与存储建议

1. 数据缓存策略

2. 数据持久化

六、注意事项

七、总结

云原生

热门文章

最新文章

相关电子书