尼日利亚NSENG股票API对接实战:实时行情与历史K线

用户头像Fxdund
2026-02-09 发布

作为一名常年对接海外金融数据的开发者,最近接手了尼日利亚股市(NSENG,尼日利亚证券交易所)的数据对接需求,踩了不少坑也积累了些实用经验。不同于A股、美股成熟的API生态,尼日利亚股市的中文资料几乎为零,甚至英文文档也寥寥无几。今天就以个人实战经验,分享如何用 iTick API 对接 NSENG 股票的实时行情和历史 K 线,希望能帮到有同样需求的朋友。
36adbb86851a46108ee5e30bbe98936c.jpg

一、前期准备: API 基础配置

1.1 注册与密钥获取

itick 是尼日利亚本地比较靠谱的金融数据服务商,专门提供 NSENG 股票的实时和历史数据。首先得去它的 iTick 注册开发者账号,完成邮箱验证后,就能拿到 API_KEY —— 这个是后续接口调用的核心凭证,一定要保存好,别泄露。

1.2 环境准备

我用的是 Python 3.8+ 版本,核心依赖就两个:requests(发HTTP请求)和 pandas(处理数据),如果要可视化还需要 mplfinance。安装命令很简单:

pip install requests pandas mplfinance

二、实战1:获取 NSENG 股票实时行情

itick 的实时行情接口是 RESTful 风格,支持按股票代码查询单只或多只股票的实时价格、成交量、涨跌幅等核心数据。

1.2 环境准备

pip install requests pandas mplfinance

二、实战1:获取 NSENG 股票实时行情

新接口:GET /stock/quote?region=NG&code=xxxx

注意:尼日利亚市场代码是 NG(不是以前的 NSENG)。

核心代码示例(已更新)

import requests
import time
from typing import Dict, Optional
import pandas as pd


class ITickNSENGAPI:
    """
    封装 itick.org API 对接尼日利亚(NG)股票数据的类(2026最新版)
    """
    def __init__(self, token: str):
        self.base_url = "//api.itick.org"
        self.token = token
        self.headers = {
            "accept": "application/json",
            "token": self.token
        }

    def get_realtime_quote(self, stock_code: str) -> Optional[Dict]:
        """
        获取单只股票实时行情
        :param stock_code: 股票代码(如 DANGCEM)
        :return: 实时行情字典,失败返回 None
        """
        for retry in range(3):
            try:
                url = f"{self.base_url}/stock/quote"
                params = {
                    "region": "NG",           # 尼日利亚市场代码
                    "code": stock_code.upper()  # 必须大写
                }
                response = requests.get(
                    url,
                    headers=self.headers,
                    params=params,
                    timeout=10
                )

                if response.status_code == 200:
                    result = response.json()
                    if result.get("code") == 0 and "data" in result:
                        quote = result["data"]
                        return {
                            "股票代码": quote["s"],
                            "最新价": quote["ld"],
                            "开盘价": quote["o"],
                            "最高价": quote["h"],
                            "最低价": quote["l"],
                            "成交量": quote["v"],
                            "涨跌额": quote["ch"],
                            "涨跌幅(%)": quote["chp"],
                            "更新时间戳": quote["t"],
                            "交易状态": quote["ts"]  # 0正常 1停牌 2退市 3熔断
                        }
                    else:
                        print(f"未找到 {stock_code} 的实时数据,或接口返回异常")
                        return None
                else:
                    print(f"请求失败,状态码:{response.status_code},重试第{retry+1}次")
                    time.sleep(1.5)

            except requests.exceptions.Timeout:
                print(f"请求超时,重试第{retry+1}次")
                time.sleep(1.5)
            except Exception as e:
                print(f"接口异常:{str(e)}")
                return None

        return None


# ---------------------- 实战调用 ----------------------
if __name__ == "__main__":
    # 替换成你自己的 token
    MY_TOKEN = "your_itick_token_here"

    api = ITickNSENGAPI(MY_TOKEN)

    realtime_data = api.get_realtime_quote("DANGCEM")
    if realtime_data:
        print("=== 尼日利亚 NG 市场实时行情 ===")
        for key, value in realtime_data.items():
            print(f"{key}: {value}")

预期输出示例

=== 尼日利亚 NG 市场实时行情 ===
股票代码: DANGCEM
最新价: 312.5
开盘价: 310.0
最高价: 315.0
最低价: 308.0
成交量: 156800
涨跌额: 2.5
涨跌幅(%): 0.81
更新时间戳: 1765526889000
交易状态: 0

三、实战2:获取历史 K 线数据

新接口:GET /stock/kline?region=NG&code=xxxx&kType=8&limit=100

kType 对应周期

  • 8 = 日K
  • 9 = 周K
  • 10 = 月K
  • 1~7 = 分钟级别(1分钟、5分钟、15分钟等)

核心代码示例(日K)

# 继续使用上面的类,新增方法
def get_historical_klines(
    self,
    stock_code: str,
    k_type: int = 8,           # 8=日K
    limit: int = 100,
    end_timestamp: int = None  # 可选,截止时间戳
) -> Optional[pd.DataFrame]:
    """
    获取历史 K 线数据(日K、周K、月K等)
    """
    try:
        url = f"{self.base_url}/stock/kline"
        params = {
            "region": "NG",
            "code": stock_code.upper(),
            "kType": k_type,
            "limit": limit
        }
        if end_timestamp:
            params["et"] = end_timestamp

        response = requests.get(
            url,
            headers=self.headers,
            params=params,
            timeout=12
        )

        if response.status_code == 200:
            result = response.json()
            if result.get("code") == 0 and isinstance(result.get("data"), list):
                data_list = result["data"]
                if not data_list:
                    print(f"{stock_code} 无历史数据")
                    return None

                df = pd.DataFrame(data_list)
                # 字段映射(符合国内习惯)
                df = df[["t", "o", "h", "l", "c", "v"]]
                df.rename(
                    columns={
                        "t": "时间戳",
                        "o": "开盘价",
                        "h": "最高价",
                        "l": "最低价",
                        "c": "收盘价",
                        "v": "成交量"
                    },
                    inplace=True
                )

                # 时间戳转可读日期(UTC → 北京时间)
                df["日期"] = pd.to_datetime(df["时间戳"], unit="ms", utc=True)
                df["日期"] = df["日期"].dt.tz_convert("Asia/Shanghai")
                df["日期"] = df["日期"].dt.strftime("%Y-%m-%d %H:%M:%S")

                # 调整列顺序
                df = df[["日期", "开盘价", "最高价", "最低价", "收盘价", "成交量", "时间戳"]]

                return df
            else:
                print("接口返回格式异常")
                return None
        else:
            print(f"K线请求失败,状态码:{response.status_code}")
            return None

    except Exception as e:
        print(f"获取K线出错:{str(e)}")
        return None


# ---------------------- 调用示例 ----------------------
if __name__ == "__main__":
    # (前面的初始化代码不变)
    kline_df = api.get_historical_klines(
        stock_code="DANGCEM",
        k_type=8,          # 日K
        limit=60           # 最近60天
    )

    if kline_df is not None:
        print("\n=== 历史日K线数据(前5条) ===")
        print(kline_df.head())

        # 可视化(可选)
        import mplfinance as mpf
        plot_df = kline_df.copy()
        plot_df["日期"] = pd.to_datetime(plot_df["日期"])
        plot_df.set_index("日期", inplace=True)
        plot_df.rename(
            columns={
                "开盘价": "Open",
                "最高价": "High",
                "最低价": "Low",
                "收盘价": "Close",
                "成交量": "Volume"
            },
            inplace=True
        )
        mpf.plot(
            plot_df,
            type='candle',
            volume=True,
            title='DANGCEM (NG) 日K线',
            ylabel='价格 (NGN)',
            figratio=(16,9)
        )

四、重要注意事项(2026最新版)

  1. 市场代码:尼日利亚统一用 region=NG(不再是 NSENG)
  2. 认证方式:现在只用 token 请求头,不再是 Bearer api_key:secret
  3. 时间戳:返回的是毫秒级 Unix 时间戳(UTC),需自行转换为北京时间
  4. 频率限制:免费账号有调用限制,建议加上合理的 sleep 或使用付费套餐
  5. 价格单位:仍为尼日利亚奈拉(NGN)

五、总结

对接尼日利亚 NSENG 股票 API 虽然不如对接A股/美股顺手,但只要找对工具、踩平时区和网络的坑,其实难度不大。总结几个核心点:

关键点回顾

  1. 实时行情和历史 K 线的核心接口都是 GET 请求,重点做好超时重试时区转换
  2. 数据解析时优先提取核心字段,并用 pandas 格式化,方便后续分析和可视化。

温馨提示:本文仅供代码参考,不构成任何投资建议。市场有风险,投资需谨慎

参考文档:https://docs.itick.org/rest-api/stocks/stock-kline
GitHub:https://github.com/itick-org/

评论