Trong việc quản lý đầu tư, việc theo dõi biến động giá tài sản theo thời gian thực là vô cùng quan trọng. Bài viết này hướng dẫn xây dựng một công cụ tự động hóa bằng Python để theo dõi giá vàng "Easy Store Gold" từ ngân hàng CCB. Công cụ sử dụng Selenium để quét dữ liệu (web scraping) và tích hợp Webhook để gửi thông báo tức thời qua Feishu (Lark Suite).

Các tính năng chính

• Truy xuất dữ liệu thực tế: Tự động lấy giá vàng trực tiếp từ bảng tỷ giá của ngân hàng.
• Thông báo qua Webhook: Gửi thông báo chi tiết đến ứng dụng tin nhắn khi có sự thay đổi về giá.
• Giám sát liên tục: Chế độ chạy ngầm và kiểm tra giá theo chu kỳ (mặc định 5 phút).
• Quản lý lịch sử: Lưu trữ dữ liệu biến động vào file JSON để phân tích sau này.
• Xử lý lỗi thông minh: Tự động thử lại khi mất kết nối hoặc trang web tải chậm.

Yêu cầu hệ thống

Để vận hành công cụ, bạn cần cài đặt các thư viện sau:

pip install selenium requests webdriver-manager

Triển khai mã nguồn

Dưới đây là cấu trúc mã nguồn được tối ưu hóa, chia thành các module chức năng để dễ dàng bảo trì.

import os
import time
import json
import logging
import signal
import requests
from datetime import datetime
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager

# Thiết lập hệ thống Log
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[logging.FileHandler('gold_tracker.log'), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

class LarkAlert:
    """Xử lý gửi thông báo đến Feishu/Lark"""
    def __init__(self, endpoint: str):
        self.endpoint = endpoint

    def notify(self, title: str, message: str, stats: dict = None):
        if not self.endpoint:
            return
        
        payload = {
            "msg_type": "interactive",
            "card": {
                "header": {"title": {"tag": "plain_text", "content": title}, "template": "orange"},
                "elements": [
                    {"tag": "div", "text": {"tag": "lark_md", "content": message}}
                ]
            }
        }
        
        if stats:
            fields = [{"is_short": True, "text": {"tag": "lark_md", "content": f"**{k}:** {v}"}} for k, v in stats.items()]
            payload["card"]["elements"].append({"tag": "div", "fields": fields})

        try:
            res = requests.post(self.endpoint, json=payload, timeout=15)
            logger.info(f"Trạng thái gửi thông báo: {res.status_code}")
        except Exception as err:
            logger.error(f"Lỗi gửi thông báo: {err}")

class GoldPriceTracker:
    def __init__(self, webhook_url=None):
        self.url = 'http://www2.ccb.com/chn/home/gjs_sp/scxq/index.shtml'
        self.history_file = "price_records.json"
        self.is_active = True
        self.notifier = LarkAlert(webhook_url)
        self.last_known_price = 0.0
        
        signal.signal(signal.SIGINT, self._stop_engine)
        self._load_local_data()

    def _stop_engine(self, signum, frame):
        logger.info("Đang dừng dịch vụ giám sát...")
        self.is_active = False

    def _load_local_data(self):
        if os.path.exists(self.history_file):
            with open(self.history_file, 'r', encoding='utf-8') as f:
                data = json.load(f)
                self.last_known_price = data.get("last_price", 0.0)

    def _init_browser(self):
        chrome_options = Options()
        chrome_options.add_argument("--headless")
        chrome_options.add_argument("--no-sandbox")
        chrome_options.add_argument("--disable-gpu")
        service = Service(ChromeDriverManager().install())
        return webdriver.Chrome(service=service, options=chrome_options)

    def fetch_price(self):
        browser = self._init_browser()
        try:
            browser.get(self.url)
            wait = WebDriverWait(browser, 20)
            # Chờ bảng giá xuất hiện
            table = wait.until(EC.presence_of_element_located((By.ID, 'gold_easyto_store')))
            
            # Trích xuất thông tin từ dòng dữ liệu thứ 2
            product_name = table.find_element(By.XPATH, './/tr[2]/td[2]').text.strip()
            price_val = table.find_element(By.XPATH, './/tr[2]/td[3]/span').text.strip()
            
            return {"name": product_name, "price": float(price_val), "time": datetime.now().strftime('%H:%M:%S')}
        except Exception as e:
            logger.error(f"Lỗi khi quét dữ liệu: {e}")
            return None
        finally:
            browser.quit()

    def run_monitoring(self, interval=300):
        logger.info(f"Bắt đầu giám sát chu kỳ {interval} giây.")
        while self.is_active:
            result = self.fetch_price()
            if result:
                current_p = result['price']
                diff = current_p - self.last_known_price
                
                if abs(diff) >= 0.05: # Chỉ thông báo nếu giá biến động đáng kể
                    trend = "📈 Tăng" if diff > 0 else "📉 Giảm"
                    msg = f"Cập nhật giá mới cho **{result['name']}**\nGiá hiện tại: **{current_p}** VNĐ/đơn vị\nBiến động: {diff:+.2f}"
                    self.notifier.notify("Cảnh Báo Giá Vàng", msg, {"Xu hướng": trend, "Thời điểm": result['time']})
                    
                    self.last_known_price = current_p
                    self._save_state(current_p)
                
                logger.info(f"Giá hiện tại: {current_p} (Biến động: {diff:+.2f})")
            
            for _ in range(interval):
                if not self.is_active: break
                time.sleep(1)

    def _save_state(self, price):
        with open(self.history_file, 'w') as f:
            json.dump({"last_price": price, "updated_at": str(datetime.now())}, f)

if __name__ == "__main__":
    WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/YOUR_ID_HERE"
    tracker = GoldPriceTracker(WEBHOOK)
    tracker.run_monitoring(interval=300)

Giải thích logic hoạt động

Mã nguồn trên được thiết kế dựa trên nguyên lý hướng đối tượng:

• Khởi tạo trình duyệt ngầm: Sử dụng --headless để trình duyệt không hiển thị giao diện, giúp tiết kiệm tài nguyên hệ thống khi chạy trên máy chủ (VPS).
• Xử lý tín hiệu (Signals): Sử dụng thư viện signal để bắt lệnh Ctrl+C, giúp chương trình đóng các tiến trình trình duyệt đang chạy dở trước khi thoát hoàn toàn.
• Ngưỡng thông báo (Threshold): Chương trình chỉ gửi tin nhắn nếu giá trị thay đổi vượt mức 0.05 đơn vị. Điều này tránh việc gửi quá nhiều tin nhắn rác khi thị trường đi ngang.
• Quản lý lịch sử: Trạng thái giá gần nhất được lưu vào price_records.json, đảm bảo khi khởi động lại, chương trình vẫn nhớ được giá của lần quét cuối cùng.

Lưu ý khi sử dụng

Khi triển khai công cụ này, bạn cần lưu ý một số vấn đề kỹ thuật:

• Tần suất truy cập: Không nên đặt thời gian kiểm tra quá ngắn (dưới 60 giây) để tránh việc địa chỉ IP bị hệ thống máy chủ ngân hàng chặn do nghi ngờ tấn công DOS.
• Cập nhật XPath: Cấu trúc HTML của trang web ngân hàng có thể thay đổi. Nếu chương trình không tìm thấy phần tử, hãy kiểm tra lại ID gold_easyto_store trong mã nguồn trang web.
• Môi trường máy chủ: Nếu chạy trên Linux (Ubuntu/CentOS), hãy đảm bảo đã cài đặt Google Chrome và các thư viện hỗ trợ render giao diện cần thiết.