如何用Python获取Dukascopy JS Widget实时股票行情数据?
如何用Python接入Dukascopy JS Widget获取实时股票数据
问题描述
我正在开展一个数据分析项目,需要获取股票价格的实时数据feed。我发现了一款可嵌入网站的免费Dukascopy JS Widget,嵌入代码如下。请问能否通过Python接入该工具实时读取数据?我可以采用每秒发送请求的无限循环方式获取数据,但这种方法效率极低,尤其是追踪多支交易标的时。
服务提供的代码片段:
<script type="text/javascript">DukascopyApplet = {"type":"chart","params":{"showUI":true,"showTabs":true,"showParameterToolbar":true,"showOfferSide":true,"allowInstrumentChange":true,"allowPeriodChange":true,"allowOfferSideChange":true,"showAdditionalToolbar":true,"showDetachButton":true,"presentationType":"candle","axisX":true,"axisY":true,"legend":true,"timeline":true,"showDateSeparators":true,"showZoom":true,"showScrollButtons":true,"showAutoShiftButton":true,"crosshair":true,"borders":false,"theme":"Pastelle","uiColor":"#000","availableInstruments":"l:","instrument":"EUR/USD","period":"7","offerSide":"BID","timezone":0,"live":true,"panLock":false,"width":"100%","height":"600","adv":"popup"}};</script><script type="text/javascript" src="https://freeserv-static.dukascopy.com/2.0/core.js"></script>
可行解决方案
1. 直接调用Dukascopy官方API(推荐)
这款JS Widget本质是通过调用Dukascopy的后台接口获取数据,你可以绕过Widget直接对接官方提供的实时数据服务,这是效率最高的方式。
- 找到API端点:用浏览器开发者工具的「Network」标签,观察Widget加载时的请求,能定位到实时数据的WebSocket或REST接口(Dukascopy主要用WebSocket推送实时行情)。
- Python实现示例:用
websocket-client库连接WebSocket并订阅标的:
import websocket import json def on_message(ws, message): # 解析实时推送的数据 data = json.loads(message) print(f"实时行情: {data}") def on_error(ws, error): print(f"错误: {error}") def on_close(ws): print("连接已关闭") def on_open(ws): # 发送订阅请求,以EUR/USD为例 subscribe_payload = json.dumps({ "action": "subscribe", "instrument": "EUR/USD", "offerSide": "BID" }) ws.send(subscribe_payload) if __name__ == "__main__": websocket.enableTrace(True) # 替换为浏览器中捕获到的实际WebSocket端点 ws = websocket.WebSocketApp( "wss://实际Dukascopy WS端点地址", on_open=on_open, on_message=on_message, on_error=on_error, on_close=on_close ) ws.run_forever()
2. 用无头浏览器解析Widget(备选)
如果暂时找不到直接的API,可以用无头浏览器加载嵌入Widget的页面,通过执行JavaScript提取实时更新的数据,这种方式比轮询请求更高效,因为浏览器会自动接收Widget的实时推送。
- 示例(Playwright实现):
from playwright.sync_api import sync_playwright import time def get_realtime_price(page): # 根据Widget的DOM结构或内部变量提取数据,需自行调整选择器 return page.evaluate("""() => { const priceElement = document.querySelector('.dkc-price-display'); return priceElement ? priceElement.textContent : '暂无数据'; }""") with sync_playwright() as p: browser = p.chromium.launch(headless=True) page = browser.new_page() # 加载本地创建的嵌入Widget的HTML页面 page.goto("file:///本地路径/your-widget-page.html") # 持续监听数据更新 while True: current_price = get_realtime_price(page) print(f"当前价格: {current_price}") time.sleep(1) browser.close()
3. 避免低效轮询
你提到的每秒循环发送HTTP请求是最不推荐的方式:不仅占用大量资源,还容易触发服务端限流。优先选择WebSocket对接API的方案,其次考虑无头浏览器方案。
内容的提问来源于stack exchange,提问作者Tuppitappi
相关产品推荐
相关产品推荐

