Infoway 财经新闻 API 接口正式上线

id / ctime(发布时间) / title / source(来源) / picUrl(封面图) / url(网页链接) / description(描述)

id / ctime(发布时间) / title / source(来源) / picUrl(封面图) / url(网页链接) / description(描述)

dk(去重键) / title / content(正文) / sd(摘要) / published(时间戳)
/ urgency(紧急度) / provider(来源) / symbols(关联标的) / country / lang / link

dk(去重键) / title / content(正文) / sd(摘要) / published(时间戳)
/ urgency(紧急度) / provider(来源) / symbols(关联标的) / country / lang / link

import time

def measure_latency(news_item: dict) -> float:
    """端到端时延 = 本地接收时刻 - 新闻发布时间戳（均为 UTC 秒）"""
    return time.time() - news_item["published"]

# 对轮询接口：记录你"发起请求"到"看到这条新闻"的时间差，
#   并叠加你的轮询间隔，才是真实的发现延迟。
# 对 WebSocket 推送：直接用上式，收到即测。

import time

def measure_latency(news_item: dict) -> float:
    """端到端时延 = 本地接收时刻 - 新闻发布时间戳（均为 UTC 秒）"""
    return time.time() - news_item["published"]

# 对轮询接口：记录你"发起请求"到"看到这条新闻"的时间差，
#   并叠加你的轮询间隔，才是真实的发现延迟。
# 对 WebSocket 推送：直接用上式，收到即测。

"symbols": ["000660.KS", "005930.KS", "NVDA.US"]

"symbols": ["000660.KS", "005930.KS", "NVDA.US"]

def route_news_to_holdings(news: dict, my_holdings: set[str]) -> bool:
    """新闻关联标的与持仓有交集 → 触发提醒/风控"""
    hit = set(news.get("symbols") or []) & my_holdings
    if hit:
        print(f"⚠️ 持仓相关新闻 [{news['urgency']}] {news['title']}  命中: {hit}")
        return True
    return False

route_news_to_holdings(news, {"005930.KS", "AAPL.US"})

def route_news_to_holdings(news: dict, my_holdings: set[str]) -> bool:
    """新闻关联标的与持仓有交集 → 触发提醒/风控"""
    hit = set(news.get("symbols") or []) & my_holdings
    if hit:
        print(f"⚠️ 持仓相关新闻 [{news['urgency']}] {news['title']}  命中: {hit}")
        return True
    return False

route_news_to_holdings(news, {"005930.KS", "AAPL.US"})

import heapq

# 用 urgency 做小顶堆，突发消息优先出队
queue = []
heapq.heappush(queue, (news["urgency"], news["published"], news))

import heapq

# 用 urgency 做小顶堆，突发消息优先出队
queue = []
heapq.heappush(queue, (news["urgency"], news["published"], news))

seen = set()

def dedup(news: dict) -> bool:
    dk = news.get("dk")
    if not dk or dk in seen:
        return False          # 重复，丢弃
    seen.add(dk)
    return True               # 新新闻，处理

seen = set()

def dedup(news: dict) -> bool:
    dk = news.get("dk")
    if not dk or dk in seen:
        return False          # 重复，丢弃
    seen.add(dk)
    return True               # 新新闻，处理

{
    "code": 10020,
    "trace": "5dd7e89cdcc247d78e817b67269167e0",
    "data": { "lang": "zh-Hans" }
}

{
    "code": 10020,
    "trace": "5dd7e89cdcc247d78e817b67269167e0",
    "data": { "lang": "zh-Hans" }
}

{ "code": 200, "msg": "ws connect success" }

{ "code": 200, "msg": "ws connect success" }

{
    "code": 10021,
    "trace": "5dd7e89cdcc247d78e817b67269167e0",
    "msg": "ok",
    "data": { "lang": "zh-hans" }
}

{
    "code": 10021,
    "trace": "5dd7e89cdcc247d78e817b67269167e0",
    "msg": "ok",
    "data": { "lang": "zh-hans" }
}

{
    "code": 10022,
    "data": {
        "dk": "e9ecbfbc6799c39d5ec09c8c771ac0fe",
        "country": "CN",
        "lang": "zh-Hans",
        "route": "lang",
        "title": "中国债市动态：资金转松支撑中短券走强，科技股大涨压制长债",
        "published": 1785815048,
        "urgency": 2,
        "provider": "reuters",
        "symbols": ["399006.SZ", "000688.SH"],
        "link": "",
        "content": "中国债市周二走势分化，中短券在资金面转松带动下表现较好，收益率小幅下行；长端和超长端则延续弱势，30年期国债收益率上行……（正文略）",
        "sd": "中国债市周二走势分化，中短券在资金面转松带动下表现较好，收益率小幅下行；长端和超长端则延续弱势……"
    }
}

{
    "code": 10022,
    "data": {
        "dk": "e9ecbfbc6799c39d5ec09c8c771ac0fe",
        "country": "CN",
        "lang": "zh-Hans",
        "route": "lang",
        "title": "中国债市动态：资金转松支撑中短券走强，科技股大涨压制长债",
        "published": 1785815048,
        "urgency": 2,
        "provider": "reuters",
        "symbols": ["399006.SZ", "000688.SH"],
        "link": "",
        "content": "中国债市周二走势分化，中短券在资金面转松带动下表现较好，收益率小幅下行；长端和超长端则延续弱势，30年期国债收益率上行……（正文略）",
        "sd": "中国债市周二走势分化，中短券在资金面转松带动下表现较好，收益率小幅下行；长端和超长端则延续弱势……"
    }
}

import os
import asyncio
import json
import uuid
import logging
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("infoway-news")

REQ_SUBSCRIBE = 10020
REQ_HEARTBEAT = 10010
ACK_SUBSCRIBE = 10021
PUSH_NEWS = 10022
HANDSHAKE_OK = 200

SUBSCRIBE_LANG = "zh-Hans"   # 可改为 en / ja / ko / zh-Hant 等


class NewsWSClient:
    """Infoway 财经新闻 WebSocket 客户端（心跳 + 指数退避重连 + 去重）"""

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

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

    async def _subscribe(self) -> None:
        await self._send({"code": REQ_SUBSCRIBE, "trace": uuid.uuid4().hex,
                          "data": {"lang": self.lang}})
        logger.info("已发送订阅请求，语言=%s", self.lang)

    def _start_heartbeat(self) -> None:
        self._cancel_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": uuid.uuid4().hex})
            except (ConnectionClosed, asyncio.CancelledError):
                pass

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

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

    def _on_news(self, data: dict) -> None:
        dk = data.get("dk")
        if dk and dk in self.seen_dk:
            return                       # 去重
        if dk:
            self.seen_dk.add(dk)
        symbols = ", ".join(data.get("symbols") or []) or "-"
        logger.info("[新闻] 紧急度=%s 来源=%s 标的=%s\n  标题: %s",
                    data.get("urgency"), data.get("provider"), symbols, data.get("title"))
        # TODO: 写入数据库 / 推送消息队列 / 触发策略信号

    def _on_message(self, raw: str) -> None:
        try:
            msg = json.loads(raw)
        except json.JSONDecodeError:
            return
        code = msg.get("code")
        if code == PUSH_NEWS:
            self._on_news(msg.get("data", {}))
        elif code == ACK_SUBSCRIBE:
            logger.info("订阅确认成功，当前语言=%s", msg.get("data", {}).get("lang"))
        elif code == HANDSHAKE_OK:
            logger.info("握手成功：%s", msg.get("msg"))

    async def _connect_once(self) -> None:
        async with websockets.connect(self.ws_url) as ws:
            self.ws = ws
            logger.info("WebSocket 连接成功")
            await self._subscribe()          # 重连后必须重新订阅
            self._start_heartbeat()
            try:
                async for message in ws:
                    self._on_message(message)
            finally:
                self._cancel_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
            except ConnectionClosed as e:
                logger.warning("连接关闭: %s", e)
            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)


async def main():
    api_key = os.environ.get("INFOWAY_API_KEY", "YOUR_API_KEY")
    await NewsWSClient(api_key, lang="zh-Hans").start()


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

import os
import asyncio
import json
import uuid
import logging
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("infoway-news")

REQ_SUBSCRIBE = 10020
REQ_HEARTBEAT = 10010
ACK_SUBSCRIBE = 10021
PUSH_NEWS = 10022
HANDSHAKE_OK = 200

SUBSCRIBE_LANG = "zh-Hans"   # 可改为 en / ja / ko / zh-Hant 等


class NewsWSClient:
    """Infoway 财经新闻 WebSocket 客户端（心跳 + 指数退避重连 + 去重）"""

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

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

    async def _subscribe(self) -> None:
        await self._send({"code": REQ_SUBSCRIBE, "trace": uuid.uuid4().hex,
                          "data": {"lang": self.lang}})
        logger.info("已发送订阅请求，语言=%s", self.lang)

    def _start_heartbeat(self) -> None:
        self._cancel_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": uuid.uuid4().hex})
            except (ConnectionClosed, asyncio.CancelledError):
                pass

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

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

    def _on_news(self, data: dict) -> None:
        dk = data.get("dk")
        if dk and dk in self.seen_dk:
            return                       # 去重
        if dk:
            self.seen_dk.add(dk)
        symbols = ", ".join(data.get("symbols") or []) or "-"
        logger.info("[新闻] 紧急度=%s 来源=%s 标的=%s\n  标题: %s",
                    data.get("urgency"), data.get("provider"), symbols, data.get("title"))
        # TODO: 写入数据库 / 推送消息队列 / 触发策略信号

    def _on_message(self, raw: str) -> None:
        try:
            msg = json.loads(raw)
        except json.JSONDecodeError:
            return
        code = msg.get("code")
        if code == PUSH_NEWS:
            self._on_news(msg.get("data", {}))
        elif code == ACK_SUBSCRIBE:
            logger.info("订阅确认成功，当前语言=%s", msg.get("data", {}).get("lang"))
        elif code == HANDSHAKE_OK:
            logger.info("握手成功：%s", msg.get("msg"))

    async def _connect_once(self) -> None:
        async with websockets.connect(self.ws_url) as ws:
            self.ws = ws
            logger.info("WebSocket 连接成功")
            await self._subscribe()          # 重连后必须重新订阅
            self._start_heartbeat()
            try:
                async for message in ws:
                    self._on_message(message)
            finally:
                self._cancel_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
            except ConnectionClosed as e:
                logger.warning("连接关闭: %s", e)
            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)


async def main():
    api_key = os.environ.get("INFOWAY_API_KEY", "YOUR_API_KEY")
    await NewsWSClient(api_key, lang="zh-Hans").start()


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

from collections import defaultdict, deque

# symbol -> 最近 N 条新闻（供各品种页面直接读取）
news_by_symbol: dict[str, deque] = defaultdict(lambda: deque(maxlen=50))

def index_news(news: dict) -> None:
    """每条推送按关联标的分发到对应品种的新闻流"""
    item = {
        "title": news["title"],
        "sd": news.get("sd"),
        "published": news["published"],
        "urgency": news["urgency"],      # 越小越紧急，前端可据此置顶飘红
        "provider": news["provider"],
        "link": news.get("link"),
    }
    for sym in news.get("symbols") or []:
        news_by_symbol[sym].appendleft(item)

def instrument_news(symbol: str) -> list:
    """交易终端渲染某品种页面时，直接取它的相关新闻"""
    return list(news_by_symbol.get(symbol, []))

# 例：用户打开 DELL.US 页面
for n in instrument_news("DELL.US"):
    print(f"[{n['urgency']}] {n['title']} — {n['provider']}")

from collections import defaultdict, deque

# symbol -> 最近 N 条新闻（供各品种页面直接读取）
news_by_symbol: dict[str, deque] = defaultdict(lambda: deque(maxlen=50))

def index_news(news: dict) -> None:
    """每条推送按关联标的分发到对应品种的新闻流"""
    item = {
        "title": news["title"],
        "sd": news.get("sd"),
        "published": news["published"],
        "urgency": news["urgency"],      # 越小越紧急，前端可据此置顶飘红
        "provider": news["provider"],
        "link": news.get("link"),
    }
    for sym in news.get("symbols") or []:
        news_by_symbol[sym].appendleft(item)

def instrument_news(symbol: str) -> list:
    """交易终端渲染某品种页面时，直接取它的相关新闻"""
    return list(news_by_symbol.get(symbol, []))

# 例：用户打开 DELL.US 页面
for n in instrument_news("DELL.US"):
    print(f"[{n['urgency']}] {n['title']} — {n['provider']}")