Infoway 韩国股票实时行情 API 正式上线

import requests

# 同时查询三星电子和 SK 海力士的最新成交
codes = "005930.KS,000660.KS"
url = f"https://data.infoway.io/korea/batch_trade/{codes}"

headers = {
    "apiKey": "YOUR_API_KEY",
    "Accept": "application/json"
}

resp = requests.get(url, headers=headers)
data = resp.json()

for item in data["data"]:
    direction = {0: "未知", 1: "买入", 2: "卖出"}.get(item["td"], "未知")
    print(f"{item['s']}: 最新价={item['p']}, 成交量={item['v']}, 方向={direction}")

import requests

# 同时查询三星电子和 SK 海力士的最新成交
codes = "005930.KS,000660.KS"
url = f"https://data.infoway.io/korea/batch_trade/{codes}"

headers = {
    "apiKey": "YOUR_API_KEY",
    "Accept": "application/json"
}

resp = requests.get(url, headers=headers)
data = resp.json()

for item in data["data"]:
    direction = {0: "未知", 1: "买入", 2: "卖出"}.get(item["td"], "未知")
    print(f"{item['s']}: 最新价={item['p']}, 成交量={item['v']}, 方向={direction}")

import requests

codes = "005930.KS"
url = f"https://data.infoway.io/korea/batch_depth/{codes}"

headers = {"apiKey": "YOUR_API_KEY"}

resp = requests.get(url, headers=headers)
result = resp.json()["data"][0]

ask_price = result["a"][0][0]   # 卖一价
ask_vol   = result["a"][1][0]   # 卖一量
bid_price = result["b"][0][0]   # 买一价
bid_vol   = result["b"][1][0]   # 买一量

print(f"标的：{result['s']}")
print(f"卖一：₩{float(ask_price):,.0f} × {float(ask_vol):.0f} 股")
print(f"买一：₩{float(bid_price):,.0f} × {float(bid_vol):.0f} 股")
print(f"价差：₩{float(ask_price) - float(bid_price):,.0f}")

import requests

codes = "005930.KS"
url = f"https://data.infoway.io/korea/batch_depth/{codes}"

headers = {"apiKey": "YOUR_API_KEY"}

resp = requests.get(url, headers=headers)
result = resp.json()["data"][0]

ask_price = result["a"][0][0]   # 卖一价
ask_vol   = result["a"][1][0]   # 卖一量
bid_price = result["b"][0][0]   # 买一价
bid_vol   = result["b"][1][0]   # 买一量

print(f"标的：{result['s']}")
print(f"卖一：₩{float(ask_price):,.0f} × {float(ask_vol):.0f} 股")
print(f"买一：₩{float(bid_price):,.0f} × {float(bid_vol):.0f} 股")
print(f"价差：₩{float(ask_price) - float(bid_price):,.0f}")

import requests

url = "https://data.infoway.io/korea/v2/batch_kline"

headers = {
    "apiKey": "YOUR_API_KEY",
    "Content-Type": "application/json"
}

# 查询 NAVER 最近 20 根 5 分钟 K 线
payload = {
    "klineType": 2,        # 5分钟K
    "klineNum": 20,
    "codes": "035420.KS"
    # "timestamp": 1234567890  # 可选，传秒时间戳向前翻页查历史
}

resp = requests.post(url, json=payload, headers=headers)
result = resp.json()

for item in result["data"]:
    print(f"\n{item['s']} K线列表（最新在前）：")
    for bar in item["respList"]:
        print(f"  时间={bar['t']}, 开={bar['o']}, 高={bar['h']}, 低={bar['l']}, 收={bar['c']}, 量={bar['v']}, 涨跌幅={bar['pc']}")

import requests

url = "https://data.infoway.io/korea/v2/batch_kline"

headers = {
    "apiKey": "YOUR_API_KEY",
    "Content-Type": "application/json"
}

# 查询 NAVER 最近 20 根 5 分钟 K 线
payload = {
    "klineType": 2,        # 5分钟K
    "klineNum": 20,
    "codes": "035420.KS"
    # "timestamp": 1234567890  # 可选，传秒时间戳向前翻页查历史
}

resp = requests.post(url, json=payload, headers=headers)
result = resp.json()

for item in result["data"]:
    print(f"\n{item['s']} K线列表（最新在前）：")
    for bar in item["respList"]:
        print(f"  时间={bar['t']}, 开={bar['o']}, 高={bar['h']}, 低={bar['l']}, 收={bar['c']}, 量={bar['v']}, 涨跌幅={bar['pc']}")

import requests

url = "https://data.infoway.io/common/basic/financial/income_statement"

headers = {"apiKey": "YOUR_API_KEY"}

params = {
    "symbol": "005930.KS",
    "type": "STOCK_KR",
    "period_type": "fq"    # fq=季度, fy=年度, fh=半年
}

resp = requests.get(url, params=params, headers=headers)
items = resp.json()["data"]

# 找出净利润条目
for item in items:
    if item["itemId"] == "net_income":
        print(f"报告期: {item['periodDate']}, 净利润: {item['itemValue']:,.0f} 韩元, TTM: {item['ttm']:,.0f} 韩元")
        break

import requests

url = "https://data.infoway.io/common/basic/financial/income_statement"

headers = {"apiKey": "YOUR_API_KEY"}

params = {
    "symbol": "005930.KS",
    "type": "STOCK_KR",
    "period_type": "fq"    # fq=季度, fy=年度, fh=半年
}

resp = requests.get(url, params=params, headers=headers)
items = resp.json()["data"]

# 找出净利润条目
for item in items:
    if item["itemId"] == "net_income":
        print(f"报告期: {item['periodDate']}, 净利润: {item['itemValue']:,.0f} 韩元, TTM: {item['ttm']:,.0f} 韩元")
        break

/common/basic/financial/income_statement    # 利润表
/common/basic/financial/cash_flow          # 现金流量表
/common/basic/financial/balance_sheet      # 资产负债表
/common/basic/financial/statistics         # 估值与财务比率（含实时PE/PB）
/common/basic/financial/dividend           # 股息指标
/common/basic/financial/earnings           # EPS 与营收预期对比（Beat/Miss）

/common/basic/financial/income_statement    # 利润表
/common/basic/financial/cash_flow          # 现金流量表
/common/basic/financial/balance_sheet      # 资产负债表
/common/basic/financial/statistics         # 估值与财务比率（含实时PE/PB）
/common/basic/financial/dividend           # 股息指标
/common/basic/financial/earnings           # EPS 与营收预期对比（Beat/Miss）

wss://data.infoway.io/ws?business=korea&apikey=YOUR_API_KEY

wss://data.infoway.io/ws?business=korea&apikey=YOUR_API_KEY

import asyncio
import json
import logging
import os
import uuid
from typing import Optional

import websockets
from websockets.asyncio.client import ClientConnection
from websockets.exceptions import ConnectionClosed

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s"
)
logger = logging.getLogger("korea-ws")

# --- 协议号 ------------------------------------------------------------------
REQ_TRADE     = 10000   # 订阅实时成交明细
REQ_DEPTH     = 10003   # 订阅盘口
REQ_KLINE     = 10006   # 订阅K线
REQ_HEARTBEAT = 10010   # 心跳

PUSH_TRADE = 10002
PUSH_DEPTH = 10005
PUSH_KLINE = 10008
ACK_CODES  = {10001, 10004, 10007}

# 要订阅的韩国股票，多个用逗号分隔
SUBSCRIBE_CODES = "005930.KS,000660.KS,035420.KS"


class KoreaMarketWsClient:
    """韩国行情 WebSocket 客户端，带自动重连和心跳保活。"""

    def __init__(self, api_key: str):
        self.ws_url = f"wss://data.infoway.io/ws?business=korea&apikey={api_key}"
        self.ws: Optional[ClientConnection] = None
        self.running = True
        self.reconnect_base = 5
        self.reconnect_max  = 60
        self.heartbeat_interval = 30
        self.heartbeat_task: Optional[asyncio.Task] = None

    @staticmethod
    def _trace() -> str:
        return str(uuid.uuid4())

    async def _send(self, msg: dict) -> None:
        assert self.ws is not None
        await self.ws.send(json.dumps(msg))

    async def _subscribe_all(self) -> None:
        # 1. 实时成交明细
        await self._send({"code": REQ_TRADE, "trace": self._trace(),
                          "data": {"codes": SUBSCRIBE_CODES}})
        logger.info("已订阅成交明细: %s", SUBSCRIBE_CODES)

        # 2. 一档盘口
        await self._send({"code": REQ_DEPTH, "trace": self._trace(),
                          "data": {"codes": SUBSCRIBE_CODES}})
        logger.info("已订阅盘口: %s", SUBSCRIBE_CODES)

        # 3. 1 分钟 K 线
        await self._send({"code": REQ_KLINE, "trace": self._trace(),
                          "data": {"arr": [{"type": 1, "codes": SUBSCRIBE_CODES}]}})
        logger.info("已订阅1分钟K线: %s", SUBSCRIBE_CODES)

    def _start_heartbeat(self) -> None:
        self._stop_heartbeat()

        async def _loop():
            try:
                while True:
                    await asyncio.sleep(self.heartbeat_interval)
                    if self.ws is None or self.ws.close_code is not None:
                        break
                    await self._send({"code": REQ_HEARTBEAT, "trace": self._trace()})
                    logger.debug("心跳已发送")
            except (ConnectionClosed, asyncio.CancelledError):
                pass

        self.heartbeat_task = asyncio.create_task(_loop())

    def _stop_heartbeat(self) -> None:
        if self.heartbeat_task and not self.heartbeat_task.done():
            self.heartbeat_task.cancel()
        self.heartbeat_task = None

    def _on_message(self, raw: str) -> None:
        try:
            msg = json.loads(raw)
        except json.JSONDecodeError:
            logger.error("消息解析失败: %s", raw[:200])
            return

        code = msg.get("code")
        data = msg.get("data", {})

        if code == PUSH_TRADE:
            logger.info("[成交] %s: 价格=%s 量=%s 方向=%s",
                        data.get("s"), data.get("p"), data.get("v"),
                        {0: "未知", 1: "买入", 2: "卖出"}.get(data.get("td", 0)))
        elif code == PUSH_DEPTH:
            asks = list(zip(data.get("a", [[]])[0], data.get("a", [[], []])[1]))
            bids = list(zip(data.get("b", [[]])[0], data.get("b", [[], []])[1]))
            logger.info("[盘口] %s: 卖一=%s 买一=%s",
                        data.get("s"),
                        asks[0] if asks else "N/A",
                        bids[0] if bids else "N/A")
        elif code == PUSH_KLINE:
            logger.info("[K线] %s: 收=%s 涨幅=%s",
                        data.get("s"), data.get("c"), data.get("pfr"))
        elif code in ACK_CODES:
            logger.info("[订阅确认] code=%s msg=%s", code, msg.get("msg"))
        elif code == 200:
            logger.info("[连接成功] %s", msg.get("msg"))
        elif code == REQ_HEARTBEAT:
            logger.debug("[心跳响应] trace=%s", msg.get("trace"))
        else:
            logger.warning("[未知消息] code=%s", code)

    async def _connect_once(self) -> None:
        async with websockets.connect(self.ws_url) as ws:
            self.ws = ws
            logger.info("WebSocket 连接成功")
            await self._subscribe_all()
            self._start_heartbeat()
            try:
                async for message in ws:
                    self._on_message(message)
            finally:
                self._stop_heartbeat()
                self.ws = None

    async def start(self) -> None:
        backoff = self.reconnect_base
        while self.running:
            try:
                await self._connect_once()
                backoff = self.reconnect_base
                logger.warning("连接被服务端关闭")
            except ConnectionClosed as e:
                logger.warning("连接关闭: %s", e)
                backoff = self.reconnect_base
            except Exception as e:
                logger.error("连接异常: %s", e)

            if not self.running:
                break
            logger.info("%.0f 秒后重连...", backoff)
            await asyncio.sleep(backoff)
            backoff = min(backoff * 2, self.reconnect_max)

    def stop(self) -> None:
        self.running = False


async def main():
    api_key = os.environ.get("INFOWAY_API_KEY", "YOUR_API_KEY")
    client = KoreaMarketWsClient(api_key=api_key)
    await client.start()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info("已退出")

import asyncio
import json
import logging
import os
import uuid
from typing import Optional

import websockets
from websockets.asyncio.client import ClientConnection
from websockets.exceptions import ConnectionClosed

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s"
)
logger = logging.getLogger("korea-ws")

# --- 协议号 ------------------------------------------------------------------
REQ_TRADE     = 10000   # 订阅实时成交明细
REQ_DEPTH     = 10003   # 订阅盘口
REQ_KLINE     = 10006   # 订阅K线
REQ_HEARTBEAT = 10010   # 心跳

PUSH_TRADE = 10002
PUSH_DEPTH = 10005
PUSH_KLINE = 10008
ACK_CODES  = {10001, 10004, 10007}

# 要订阅的韩国股票，多个用逗号分隔
SUBSCRIBE_CODES = "005930.KS,000660.KS,035420.KS"


class KoreaMarketWsClient:
    """韩国行情 WebSocket 客户端，带自动重连和心跳保活。"""

    def __init__(self, api_key: str):
        self.ws_url = f"wss://data.infoway.io/ws?business=korea&apikey={api_key}"
        self.ws: Optional[ClientConnection] = None
        self.running = True
        self.reconnect_base = 5
        self.reconnect_max  = 60
        self.heartbeat_interval = 30
        self.heartbeat_task: Optional[asyncio.Task] = None

    @staticmethod
    def _trace() -> str:
        return str(uuid.uuid4())

    async def _send(self, msg: dict) -> None:
        assert self.ws is not None
        await self.ws.send(json.dumps(msg))

    async def _subscribe_all(self) -> None:
        # 1. 实时成交明细
        await self._send({"code": REQ_TRADE, "trace": self._trace(),
                          "data": {"codes": SUBSCRIBE_CODES}})
        logger.info("已订阅成交明细: %s", SUBSCRIBE_CODES)

        # 2. 一档盘口
        await self._send({"code": REQ_DEPTH, "trace": self._trace(),
                          "data": {"codes": SUBSCRIBE_CODES}})
        logger.info("已订阅盘口: %s", SUBSCRIBE_CODES)

        # 3. 1 分钟 K 线
        await self._send({"code": REQ_KLINE, "trace": self._trace(),
                          "data": {"arr": [{"type": 1, "codes": SUBSCRIBE_CODES}]}})
        logger.info("已订阅1分钟K线: %s", SUBSCRIBE_CODES)

    def _start_heartbeat(self) -> None:
        self._stop_heartbeat()

        async def _loop():
            try:
                while True:
                    await asyncio.sleep(self.heartbeat_interval)
                    if self.ws is None or self.ws.close_code is not None:
                        break
                    await self._send({"code": REQ_HEARTBEAT, "trace": self._trace()})
                    logger.debug("心跳已发送")
            except (ConnectionClosed, asyncio.CancelledError):
                pass

        self.heartbeat_task = asyncio.create_task(_loop())

    def _stop_heartbeat(self) -> None:
        if self.heartbeat_task and not self.heartbeat_task.done():
            self.heartbeat_task.cancel()
        self.heartbeat_task = None

    def _on_message(self, raw: str) -> None:
        try:
            msg = json.loads(raw)
        except json.JSONDecodeError:
            logger.error("消息解析失败: %s", raw[:200])
            return

        code = msg.get("code")
        data = msg.get("data", {})

        if code == PUSH_TRADE:
            logger.info("[成交] %s: 价格=%s 量=%s 方向=%s",
                        data.get("s"), data.get("p"), data.get("v"),
                        {0: "未知", 1: "买入", 2: "卖出"}.get(data.get("td", 0)))
        elif code == PUSH_DEPTH:
            asks = list(zip(data.get("a", [[]])[0], data.get("a", [[], []])[1]))
            bids = list(zip(data.get("b", [[]])[0], data.get("b", [[], []])[1]))
            logger.info("[盘口] %s: 卖一=%s 买一=%s",
                        data.get("s"),
                        asks[0] if asks else "N/A",
                        bids[0] if bids else "N/A")
        elif code == PUSH_KLINE:
            logger.info("[K线] %s: 收=%s 涨幅=%s",
                        data.get("s"), data.get("c"), data.get("pfr"))
        elif code in ACK_CODES:
            logger.info("[订阅确认] code=%s msg=%s", code, msg.get("msg"))
        elif code == 200:
            logger.info("[连接成功] %s", msg.get("msg"))
        elif code == REQ_HEARTBEAT:
            logger.debug("[心跳响应] trace=%s", msg.get("trace"))
        else:
            logger.warning("[未知消息] code=%s", code)

    async def _connect_once(self) -> None:
        async with websockets.connect(self.ws_url) as ws:
            self.ws = ws
            logger.info("WebSocket 连接成功")
            await self._subscribe_all()
            self._start_heartbeat()
            try:
                async for message in ws:
                    self._on_message(message)
            finally:
                self._stop_heartbeat()
                self.ws = None

    async def start(self) -> None:
        backoff = self.reconnect_base
        while self.running:
            try:
                await self._connect_once()
                backoff = self.reconnect_base
                logger.warning("连接被服务端关闭")
            except ConnectionClosed as e:
                logger.warning("连接关闭: %s", e)
                backoff = self.reconnect_base
            except Exception as e:
                logger.error("连接异常: %s", e)

            if not self.running:
                break
            logger.info("%.0f 秒后重连...", backoff)
            await asyncio.sleep(backoff)
            backoff = min(backoff * 2, self.reconnect_max)

    def stop(self) -> None:
        self.running = False


async def main():
    api_key = os.environ.get("INFOWAY_API_KEY", "YOUR_API_KEY")
    client = KoreaMarketWsClient(api_key=api_key)
    await client.start()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info("已退出")

pip install websockets

pip install websockets

export INFOWAY_API_KEY="你的API Key"
python korea_ws.py

export INFOWAY_API_KEY="你的API Key"
python korea_ws.py

import time
import requests

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://data.infoway.io"
HEADERS  = {"apiKey": API_KEY}

CODES = "005930.KS,000660.KS"   # 三星电子, SK 海力士

def get_trade():
    url = f"{BASE_URL}/korea/batch_trade/{CODES}"
    return {item["s"]: item for item in requests.get(url, headers=HEADERS).json()["data"]}

def get_kline_daily(code: str):
    """获取最近 2 根日K，用于计算昨收价。"""
    resp = requests.post(
        f"{BASE_URL}/korea/v2/batch_kline",
        json={"klineType": 8, "klineNum": 2, "codes": code},
        headers={**HEADERS, "Content-Type": "application/json"}
    )
    bars = resp.json()["data"][0]["respList"]
    return bars   # bars[0] 是当日（未收盘），bars[1] 是昨日收盘

def calc_change(current_price: str, prev_close: str) -> float:
    return (float(current_price) - float(prev_close)) / float(prev_close) * 100

if __name__ == "__main__":
    # 拉昨收
    samsung_bars  = get_kline_daily("005930.KS")
    skhynix_bars  = get_kline_daily("000660.KS")
    samsung_prev  = samsung_bars[1]["c"]
    skhynix_prev  = skhynix_bars[1]["c"]

    print(f"三星昨收: ₩{float(samsung_prev):,.0f}")
    print(f"SK海力士昨收: ₩{float(skhynix_prev):,.0f}")
    print("-" * 40)

    while True:
        trades = get_trade()

        samsung_pct  = calc_change(trades["005930.KS"]["p"], samsung_prev)
        skhynix_pct  = calc_change(trades["000660.KS"]["p"], skhynix_prev)
        spread = samsung_pct - skhynix_pct

        print(f"三星: {samsung_pct:+.2f}%  SK海力士: {skhynix_pct:+.2f}%  价差: {spread:+.2f}%", end="")

        if abs(spread) > 2.0:
            print(f"  ⚠ 两者涨幅差超过 2%，关注联动异常！", end="")
        print()

        time.sleep(5)

import time
import requests

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://data.infoway.io"
HEADERS  = {"apiKey": API_KEY}

CODES = "005930.KS,000660.KS"   # 三星电子, SK 海力士

def get_trade():
    url = f"{BASE_URL}/korea/batch_trade/{CODES}"
    return {item["s"]: item for item in requests.get(url, headers=HEADERS).json()["data"]}

def get_kline_daily(code: str):
    """获取最近 2 根日K，用于计算昨收价。"""
    resp = requests.post(
        f"{BASE_URL}/korea/v2/batch_kline",
        json={"klineType": 8, "klineNum": 2, "codes": code},
        headers={**HEADERS, "Content-Type": "application/json"}
    )
    bars = resp.json()["data"][0]["respList"]
    return bars   # bars[0] 是当日（未收盘），bars[1] 是昨日收盘

def calc_change(current_price: str, prev_close: str) -> float:
    return (float(current_price) - float(prev_close)) / float(prev_close) * 100

if __name__ == "__main__":
    # 拉昨收
    samsung_bars  = get_kline_daily("005930.KS")
    skhynix_bars  = get_kline_daily("000660.KS")
    samsung_prev  = samsung_bars[1]["c"]
    skhynix_prev  = skhynix_bars[1]["c"]

    print(f"三星昨收: ₩{float(samsung_prev):,.0f}")
    print(f"SK海力士昨收: ₩{float(skhynix_prev):,.0f}")
    print("-" * 40)

    while True:
        trades = get_trade()

        samsung_pct  = calc_change(trades["005930.KS"]["p"], samsung_prev)
        skhynix_pct  = calc_change(trades["000660.KS"]["p"], skhynix_prev)
        spread = samsung_pct - skhynix_pct

        print(f"三星: {samsung_pct:+.2f}%  SK海力士: {skhynix_pct:+.2f}%  价差: {spread:+.2f}%", end="")

        if abs(spread) > 2.0:
            print(f"  ⚠ 两者涨幅差超过 2%，关注联动异常！", end="")
        print()

        time.sleep(5)