要獲取實(shí)時(shí)加密貨幣更新并將它們發(fā)送到Telegram，我們需要將金融建模準(zhǔn)備（FMP）API與Python腳本集成。為了清晰起見(jiàn)，我們將這個(gè)過(guò)程分為幾個(gè)小標(biāo)題。
pip install websocket-client requestsclass FMPWebSocketClient:
def __init__(self, api_key, symbol, telegram_token, chat_id):
self.api_key = api_key
self.symbol = symbol
self.telegram_token = telegram_token
self.chat_id = chat_id
self.ws_url = "wss://crypto.financialmodelingprep.com"
self.data = []api_key：您的FMP API密鑰。
symbol：您想要跟蹤的加密貨幣符號(hào)（例如，SOLUSD）。
telegram_token：您的Telegram機(jī)器人API令牌。
chat_id：更新將被發(fā)送到的聊天ID。
def on_message(self, ws, message):
data = json.loads(message)
if 'price' in data and 'volume' in data:
price = float(data['price'])
volume = float(data['volume'])
timestamp = datetime.utcnow()
data_point = {"price": price, "volume": volume, "timestamp": timestamp}
self.data.append(data_point)
# Keep only the data for the last 5 minutes
five_minutes_ago = datetime.utcnow() - timedelta(minutes=5)
self.data = [x for x in self.data if x['timestamp'] >= five_minutes_ago]
# Simplified condition to send a message for every new data point
message = f"Symbol: {self.symbol}\nPrice: {price}\nVolume: {volume}\nTime: {timestamp}"
self.send_telegram_message(message)這個(gè)方法處理傳入的WebSocket消息，并將數(shù)據(jù)點(diǎn)添加到列表中。
def send_telegram_message(self, message):
url = f"https://api.telegram.org/bot{self.telegram_token}/sendMessage"
payload = {
"chat_id": self.chat_id,
"text": message
}
response = requests.post(url, data=payload)
print("Response from Telegram API:", response.json())這個(gè)方法構(gòu)建負(fù)載并使用機(jī)器人API發(fā)送消息到Telegram。 它打印出來(lái)自Telegram API的響應(yīng)以供驗(yàn)證。
5. 啟動(dòng)WebSocket客戶端
最后，我們將創(chuàng)建一個(gè)方法來(lái)啟動(dòng)WebSocket客戶端并持續(xù)獲取數(shù)據(jù)。
def on_error(self, ws, error):
print("WebSocket Error:", error)
def on_close(self, ws, close_status_code, close_msg):
print("WebSocket Closed:", close_msg)
def on_open(self, ws):
login_message = json.dumps({
"event": "login",
"data": {"apiKey": self.api_key}
})
ws.send(login_message)
subscribe_message = json.dumps({
"event": "subscribe",
"data": {"symbol": self.symbol}
})
ws.send(subscribe_message)
print("Subscribed to:", self.symbol)on_open：向WebSocket服務(wù)器發(fā)送登錄和訂閱消息。
啟動(dòng)客戶端：
def start(self):
self.ws = websocket.WebSocketApp(self.ws_url,
on_open=self.on_open,
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close)
self.ws.run_forever()初始化并啟動(dòng)WebSocket客戶端以連接到FMP WebSocket端點(diǎn)。
import requests
import time
from datetime import datetime, timedelta
class FMPWebSocketClient:
def __init__(self, api_key, symbol, telegram_token, chat_id):
self.api_key = api_key
self.symbol = symbol
self.telegram_token = telegram_token
self.chat_id = chat_id
self.ws_url = "wss://crypto.financialmodelingprep.com"
self.data = []
def on_message(self, ws, message):
data = json.loads(message)
if 'price' in data and 'volume' in data:
price = float(data['price'])
volume = float(data['volume'])
timestamp = datetime.utcnow()
data_point = {"price": price, "volume": volume, "timestamp": timestamp}
self.data.append(data_point)
# Keep only the data for the last 5 minutes
five_minutes_ago = datetime.utcnow() - timedelta(minutes=5)
self.data = [x for x in self.data if x['timestamp'] >= five_minutes_ago]
# Simplified condition to send a message for every new data point
message = f"Symbol: {self.symbol}\nPrice: {price}\nVolume: {volume}\nTime: {timestamp}"
self.send_telegram_message(message)
def send_telegram_message(self, message):
url = f"https://api.telegram.org/bot{self.telegram_token}/sendMessage"
payload = {"chat_id": self.chat_id, "text": message}
response = requests.post(url, data=payload)
print("Response from Telegram API:", response.json())
def on_error(self, ws, error):
print("WebSocket Error:", error)
def on_close(self, ws, close_status_code, close_msg):
print("WebSocket Closed:", close_msg)
def on_open(self, ws):
login_message = json.dumps({
"event": "login",
"data": {"apiKey": self.api_key}
})
ws.send(login_message)
subscribe_message = json.dumps({
"event": "subscribe",
"data": {"symbol": self.symbol}
})
ws.send(subscribe_message)
print("Subscribed to:", self.symbol)
def start(self):
self.ws = websocket.WebSocketApp(self.ws_url,
on_open=self.on_open,
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close)
self.ws.run_forever()
# Replace with your actual API key, token, and chat ID
api_key = "your_fmp_api_key"
symbol = "SOLUSD" # Use "ADAUSD" for Cardano or any other supported cryptocurrency symbol
telegram_token = "your_telegram_bot_token"
chat_id = "your_chat_id"
client = FMPWebSocketClient(api_key, symbol, telegram_token, chat_id)
client.start()通過(guò)將?Financial Modeling Prep （FMP） API?與 Telegram 機(jī)器人集成，您可以毫不費(fèi)力地隨時(shí)了解實(shí)時(shí)加密貨幣數(shù)據(jù)。本指南向您展示了如何設(shè)置 Telegram 機(jī)器人、從?FMP?獲取實(shí)時(shí)數(shù)據(jù)以及向您的聊天發(fā)送更新。
我們首先設(shè)置 Telegram 機(jī)器人，確保它已準(zhǔn)備好接收消息。接下來(lái)，我們?cè)?Python 中創(chuàng)建了一個(gè) WebSocket 客戶端，用于從?FMP?獲取實(shí)時(shí)數(shù)據(jù)?？蛻舳颂幚磉@些數(shù)據(jù)并根據(jù)預(yù)定義的條件向 Telegram 發(fā)送通知。
該系統(tǒng)允許您接收有關(guān)重大市場(chǎng)變化的即時(shí)更新，幫助您做出明智的交易決策。無(wú)論您是跟蹤 Solana、Cardano 還是任何其他加密貨幣，此設(shè)置都能確保您不會(huì)錯(cuò)過(guò)重要的市場(chǎng)走勢(shì)。
隨意自定義條件并擴(kuò)展系統(tǒng)以監(jiān)控多種加密貨幣。這種實(shí)時(shí)數(shù)據(jù)與即時(shí)消息的集成可以顯著增強(qiáng)您的交易策略和市場(chǎng)意識(shí)。
冪簡(jiǎn)集成是國(guó)內(nèi)領(lǐng)先的API集成管理平臺(tái)，專(zhuān)注于為開(kāi)發(fā)者提供全面、高效、易用的API集成解決方案。冪簡(jiǎn)API平臺(tái)可以通過(guò)以下兩種方式找到所需API：通過(guò)關(guān)鍵詞搜索API、或者從API Hub分類(lèi)頁(yè)進(jìn)入尋找。