分享两个监控资产价格，当达成条件时发出预警。代码展示的是基础应用场景，可以根据自己实际需要调整判断条件，也可以接入交易接口，实现预警后的自动交易。
比特币价格预警
通过HTTP接口查询比特币的实时价格，如果价格达到预设值，比如7万3美金，则发邮件提醒。
import requests
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart# 设置AllTick API的URL和请求参数
API_URL = "https://api.alltick.com/v1/marketdata/btcusd"
API_KEY = "your_api_key_here" # 替换为您的API密钥# API KEY可以在alltick.co注册获取# 邮件通知设置
SMTP_SERVER = "smtp.your-email-provider.com" # 替换为SMTP服务器地址
SMTP_PORT = 587 # 通常587或465
EMAIL_ADDRESS = "your_email@example.com" # 发件人邮箱
EMAIL_PASSWORD = "your_password_here" # 发件人邮箱密码
TO_EMAIL = "recipient@example.com" # 收件人邮箱# 获取比特币最新报价
def get_btc_price():headers = {"Authorization": f"Bearer {API_KEY}","Content-Type": "application/json"}response = requests.get(API_URL, headers=headers)if response.status_code == 200:data = response.json()return float(data["price"]) else:print("Failed to retrieve data:", response.status_code, response.text)return None# 发送邮件通知
def send_email_notification(price):subject = "比特币价格提醒"body = f"比特币当前价格为 ${price}，已超过73000美元！"message = MIMEMultipart()message["From"] = EMAIL_ADDRESSmessage["To"] = TO_EMAILmessage["Subject"] = subjectmessage.attach(MIMEText(body, "plain"))try:with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:server.starttls() # 启用TLSserver.login(EMAIL_ADDRESS, EMAIL_PASSWORD)server.sendmail(EMAIL_ADDRESS, TO_EMAIL, message.as_string())print("邮件已发送！")except Exception as e:print("邮件发送失败:", e)# 主函数，检查价格并发送通知
def main():price = get_btc_price()if price is not None:print(f"当前比特币价格: ${price}")if price > 73000:send_email_notification(price)if __name__ == "__main__":main()
通过Websocket订阅BTC的实时价格，当出现金叉，并且交易量激增的情况下发出预警。
import json
import websocket
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from collections import deque# WebSocket和邮件参数
API_URL = 'wss://quote.tradeswitcher.com/quote-b-ws-api?token=your_token_here' # 替换为AllTick的实际WebSocket URL和Token
SMTP_SERVER = "smtp.your-email-provider.com"
SMTP_PORT = 587
EMAIL_ADDRESS = "your_email@example.com"
EMAIL_PASSWORD = "your_password_here"
TO_EMAIL = "recipient@example.com"# AllTick接口注册：alltick.co# 初始化均线和交易量窗口
short_window = deque(maxlen=50) # 50个数据点窗口
long_window = deque(maxlen=200) # 200个数据点窗口
prev_short_avg = 0
prev_long_avg = 0# 邮件通知功能
def send_email_notification(price, volume):subject = "比特币价格金叉提醒"body = f"比特币当前价格为 ${price}，且交易量显著增加！"message = MIMEMultipart()message["From"] = EMAIL_ADDRESSmessage["To"] = TO_EMAILmessage["Subject"] = subjectmessage.attach(MIMEText(body, "plain"))try:with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:server.starttls()server.login(EMAIL_ADDRESS, EMAIL_PASSWORD)server.sendmail(EMAIL_ADDRESS, TO_EMAIL, message.as_string())print("邮件已发送！")except Exception as e:print("邮件发送失败:", e)# WebSocket连接类
class Feed:def __init__(self):self.url = API_URLself.ws = Nonedef on_open(self, ws):print("WebSocket连接已打开！")sub_param = {"cmd_id": 22002,"seq_id": 123,"data": {"symbol_list": [{"code": "BTCUSD", "depth_level": 5}]}}ws.send(json.dumps(sub_param))print("已订阅BTCUSD行情数据！")def on_message(self, ws, message):global prev_short_avg, prev_long_avgdata = json.loads(message)# 提取价格和交易量price = float(data["price"])volume = float(data["volume"])# 更新均线窗口short_window.append(price)long_window.append(price)# 计算短期和长期均线if len(short_window) == short_window.maxlen and len(long_window) == long_window.maxlen:short_avg = sum(short_window) / len(short_window)long_avg = sum(long_window) / len(long_window)# 检查金叉条件：短期均线从下向上突破长期均线if prev_short_avg < prev_long_avg and short_avg > long_avg:# 检查交易量激增（假设过去平均交易量增幅超过50%为激增）avg_volume = sum(long_window) / len(long_window)if volume > avg_volume * 1.5:send_email_notification(price, volume)prev_short_avg = short_avgprev_long_avg = long_avgdef on_error(self, ws, error):print("WebSocket出错:", error)def on_close(self, ws, close_status_code, close_msg):print("WebSocket连接关闭！")def start(self):self.ws = websocket.WebSocketApp(self.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()# 启动WebSocket客户端
if __name__ == "__main__":feed = Feed()feed.start()
