Python量化交易:如何接入金融行情数据?
【摘要】 本文为大家介绍如何使用Python调用已经封装好的高频数据API。这里以Infoway API的tick数据接口作为演示。
本文为大家介绍如何使用Python调用已经封装好的高频数据API。这里以Infoway API的实时行情接口作为演示。下面是代码示例。
请求K线数据
import requests api_url = 'https://data.infoway.io/stock/v2/batch_kline/1/10/002594.SZ%2C00285.HK%2CTSLA.US' # 设置请求头 headers = { 'User-Agent': 'Mozilla/5.0', 'Accept': 'application/json', 'apiKey': 'yourApikey' } # 发送GET请求 response = requests.get(api_url, headers=headers) # 输出结果 print(f"HTTP code: {response.status_code}") print(f"message: {response.text}")
上面代码同时查询三个不同市场的股票(002594.SZ, 00285.HK, TSLA.US),批量K线的请求地址统一为:
https://data.infoway.io/stock/v2/batch_kline
入参示例:
{ "klineType":1, //kline类型 "klineNum":10, //k线数量 "codes":"TSLA.US,AAPL.US", //查询产品,逗号分隔 "timestamp":1758553860 //最近截止时间戳 }
请求最新报价成交数据
import requests url = "https://data.infoway.io/stock/batch_trade/002594.SZ%2C00285.HK%2CTSLA.US%2CAUDCAD%2CHK50%2CXNIUSD" headers = {"accept": "application/json"} response = requests.get(url, headers=headers) print(response.text)
这个接口支持批量查询,只需传入多个股票代码即可。
获取最新盘口报价数据
import requests url = "https://data.infoway.io/stock/batch_depth/002594.SZ%2C00285.HK%2CTSLA.US%2CAUDCAD%2CHK50%2CXNIUSD" headers = {"accept": "application/json"} response = requests.get(url, headers=headers) print(response.text)
同上,symbol_list支持传入多个值。
通过Websocket订阅实时行情数据
import json import time import schedule import threading import websocket from loguru import logger class WebsocketExample: def __init__(self): self.session = None self.ws_url = "wss://data.infoway.io/ws?business=crypto&apikey=yourApikey" self.reconnecting = False def connect_all(self): """建立WebSocket连接并启动自动重连机制""" try: self.connect(self.ws_url) self.start_reconnection(self.ws_url) except Exception as e: logger.error(f"Failed to connect to {self.ws_url}: {str(e)}") def start_reconnection(self, url): """启动定时重连检查""" def check_connection(): if not self.is_connected(): logger.debug("Reconnection attempt...") self.connect(url) # 使用线程定期检查连接状态 threading.Thread(target=lambda: schedule.every(10).seconds.do(check_connection), daemon=True).start() def is_connected(self): """检查WebSocket连接状态""" return self.session and self.session.connected def connect(self, url): """建立WebSocket连接""" try: if self.is_connected(): self.session.close() self.session = websocket.WebSocketApp( url, on_open=self.on_open, on_message=self.on_message, on_error=self.on_error, on_close=self.on_close ) # 启动WebSocket连接(非阻塞模式) threading.Thread(target=self.session.run_forever, daemon=True).start() except Exception as e: logger.error(f"Failed to connect to the server: {str(e)}") def on_open(self, ws): """WebSocket连接建立成功后的回调""" logger.info(f"Connection opened") try: # 发送实时成交明细订阅请求 trade_send_obj = { "code": 10000, "trace": "01213e9d-90a0-426e-a380-ebed633cba7a", "data": {"codes": "BTCUSDT"} } self.send_message(trade_send_obj) # 不同请求之间间隔一段时间 time.sleep(5) # 发送实时盘口数据订阅请求 depth_send_obj = { "code": 10003, "trace": "01213e9d-90a0-426e-a380-ebed633cba7a", "data": {"codes": "BTCUSDT"} } self.send_message(depth_send_obj) # 不同请求之间间隔一段时间 time.sleep(5) # 发送实时K线数据订阅请求 kline_data = { "arr": [ { "type": 1, "codes": "BTCUSDT" } ] } kline_send_obj = { "code": 10006, "trace": "01213e9d-90a0-426e-a380-ebed633cba7a", "data": kline_data } self.send_message(kline_send_obj) # 启动定时心跳任务 threading.Thread(target=lambda: schedule.every(30).seconds.do(self.ping), daemon=True).start() except Exception as e: logger.error(f"Error sending initial messages: {str(e)}") def on_message(self, ws, message): """接收消息的回调""" try: logger.info(f"Message received: {message}") except Exception as e: logger.error(f"Error processing message: {str(e)}") def on_close(self, ws, close_status_code, close_msg): """连接关闭的回调""" logger.info(f"Connection closed: {close_status_code} - {close_msg}") def on_error(self, ws, error): """错误处理的回调""" logger.error(f"WebSocket error: {str(error)}") def send_message(self, message_obj): """发送消息到WebSocket服务器""" if self.is_connected(): try: self.session.send(json.dumps(message_obj)) except Exception as e: logger.error(f"Error sending message: {str(e)}") else: logger.warning("Cannot send message: Not connected") def ping(self): """发送心跳包""" ping_obj = { "code": 10010, "trace": "01213e9d-90a0-426e-a380-ebed633cba7a" } self.send_message(ping_obj) # 使用示例 if __name__ == "__main__": ws_client = WebsocketExample() ws_client.connect_all() # 保持主线程运行 try: while True: schedule.run_pending() time.sleep(1) except KeyboardInterrupt: logger.info("Exiting...") if ws_client.is_connected(): ws_client.session.close()
Infoway API实时行情数据接口
Infoway API实时行情接口涵盖各类资产,港股、美股以及加密货币的数据都有:
市场 | API |
港股 | 实时行情数据 |
实时十档盘口数据 | |
实时K线数据 | |
美股 | 实时行情数据 |
实时一档盘口数 | |
实时K线数据 | |
A股 | 沪深A股实时行情数据 |
沪深实时五档盘口数据 | |
沪深实时K线数据 | |
加密货币 | 实时行情数据 |
实时多档盘口数据 | |
实时K线数据 | |
外汇 | 实时行情数据 |
实时五档盘口数据 | |
实时K线数据 | |
贵金属 | 实时行情数据 |
实时五档盘口数据 | |
实时K线数据 |
【Github】 https://github.com/infoway-api/realtime-market-data-api
【声明】本内容来自华为云开发者社区博主,不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息,否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱:
cloudbbs@huaweicloud.com
- 点赞
- 收藏
- 关注作者
评论(0)