手把手教你用Python监控原油价格,自动推送企业微信

pip install oilpriceapi python-dotenv requests

OILPRICE_API_KEY=你的API密钥WECHAT_WEBHOOK_URL=https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=你的机器人密钥

import osimport timeimport requestsfrom datetime import datetime, time as dt_timefrom dotenv import load_dotenvfrom oilpriceapi import OilPriceAPI# 加载 .env 中的配置load_dotenv()API_KEY = os.getenv("OILPRICE_API_KEY")WECHAT_WEBHOOK_URL = os.getenv("WECHAT_WEBHOOK_URL")if not API_KEY or not WECHAT_WEBHOOK_URL:    raise RuntimeError("请在 .env 文件中配置 OILPRICE_API_KEY 和 WECHAT_WEBHOOK_URL")# 初始化油价 API 客户端client = OilPriceAPI(api_key=API_KEY)# 要监控的原油品种commodities = [    ("BRENT_CRUDE_USD", "布伦特原油"),    ("WTI_USD", "WTI原油"),    ("NATURAL_GAS_USD", "天然气"),]def send_to_wechat(message):    """发送消息到企业微信群"""    data = {"msgtype": "text", "text": {"content": message}}    try:        resp = requests.post(WECHAT_WEBHOOK_URL, json=data, timeout=10)        result = resp.json()        if result.get('errcode') == 0:            print(f"[{datetime.now()}] 消息发送成功")        else:            print(f"发送失败: {result.get('errmsg')}")    except Exception as e:        print(f"发送异常: {e}")def fetch_and_send():    """获取油价并发送到企业微信"""    lines = []    lines.append(f"原油价格行情 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")    lines.append("=" * 30)    for code, name in commodities:        try:            price_data = client.prices.get(code)            price = price_data.value            # API 返回的时间是 UTC，这里直接格式化            time_str = price_data.timestamp.strftime("%Y-%m-%d %H:%M:%S")            line = f"{name}: ${price:.2f} / 桶 (更新时间: {time_str})"            print(line)            lines.append(line)        except Exception as e:            error_line = f"{name}: 获取失败 - {e}"            print(error_line)            lines.append(error_line)    send_to_wechat("\n".join(lines))def is_trading_time(now: datetime) -> bool:    """判断当前时间是否在交易时间段内"""    current = now.time()    morning = (dt_time(8, 0) <= current <= dt_time(11, 30))    afternoon = (dt_time(14, 0) <= current <= dt_time(18, 0))    return morning or afternoondef main_loop():    print("油价监控已启动")    print("时间段: 8:00~11:30, 14:00~18:00 | 间隔: 5分钟")    print("按 Ctrl+C 停止\n")    while True:        now = datetime.now()        if is_trading_time(now):            fetch_and_send()            time.sleep(300)   # 5分钟        else:            time.sleep(60)    # 非交易时段，每分钟检查一次if __name__ == "__main__":    try:        main_loop()    except KeyboardInterrupt:        print("\n程序已停止")

python oilprice_monitor.py

nohup python oilprice_monitor.py > oil.log 2>&1 &

morning = (dt_time(0, 0) <= current <= dt_time(23, 59))  # 全天运行

from datetime import timezonebeijing_time = price_data.timestamp.astimezone(timezone(timedelta(hours=8)))

oilprice_monitor/├── .env                 # 存放 API Key 和 Webhook（不提交 Git）├── oilprice_monitor.py  # 主脚本├── .gitignore           # 忽略 .env 和 __pycache__└── README.md            # 项目说明