Automate Price Drop Alerts for Quick Commerce Products

import requests
import json
from datetime import datetime

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://api.quickcommerceapi.com/v1"

# Define products to track with target prices
WATCHLIST = [
    {"query": "Surf Excel 2kg", "platform": "blinkit", "target_price": 280},
    {"query": "Surf Excel 2kg", "platform": "zepto", "target_price": 280},
    {"query": "Tata Tea Gold 1kg", "platform": "blinkit", "target_price": 420},
    {"query": "Aashirvaad Atta 5kg", "platform": "swiggy", "target_price": 280},
    {"query": "Amul Butter 500g", "platform": "bigbasket", "target_price": 260},
]

def search_product(query: str, platform: str) -> dict:
    """Search for a product on a specific platform."""
    resp = requests.get(
        f"{BASE_URL}/search",
        params={"query": query, "platform": platform},
        headers={"X-API-Key": API_KEY},
    )
    resp.raise_for_status()
    return resp.json()

def check_prices():
    """Check all watchlist items and collect alerts."""
    alerts = []
    for item in WATCHLIST:
        data = search_product(item["query"], item["platform"])
        if not data.get("results"):
            continue

        product = data["results"][0]  # Best match
        current_price = product["offer_price"]

        if current_price <= item["target_price"]:
            alerts.append({
                "product": product["name"],
                "platform": item["platform"],
                "current_price": current_price,
                "mrp": product["mrp"],
                "target_price": item["target_price"],
                "discount": product.get("discount", "N/A"),
                "time": datetime.now().isoformat(),
            })
            print(f"ALERT: {product['name']} on {item['platform']} "
                  f"is Rs {current_price} (target: Rs {item['target_price']})")
        else:
            print(f"OK: {product['name']} on {item['platform']} "
                  f"is Rs {current_price} (target: Rs {item['target_price']})")

    return alerts

if __name__ == "__main__":
    alerts = check_prices()
    if alerts:
        print(f"\n{len(alerts)} price alert(s) triggered!")
        # Send notifications (Telegram, email, Slack, etc.)
        send_notifications(alerts)
    else:
        print("\nNo price drops detected. Prices are above target.")

import requests

TELEGRAM_BOT_TOKEN = "YOUR_BOT_TOKEN"
TELEGRAM_CHAT_ID = "YOUR_CHAT_ID"

def send_telegram_alert(alerts: list[dict]):
    """Send price drop alerts to Telegram."""
    for alert in alerts:
        message = (
            f"Price Drop Alert!\n\n"
            f"Product: {alert['product']}\n"
            f"Platform: {alert['platform'].title()}\n"
            f"Current Price: Rs {alert['current_price']}\n"
            f"MRP: Rs {alert['mrp']}\n"
            f"Discount: {alert['discount']}\n"
            f"Your Target: Rs {alert['target_price']}\n"
            f"Savings vs Target: Rs {alert['target_price'] - alert['current_price']}\n\n"
            f"Hurry! Prices can change quickly."
        )

        requests.post(
            f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage",
            json={
                "chat_id": TELEGRAM_CHAT_ID,
                "text": message,
                "parse_mode": "HTML",
            },
        )
        print(f"Telegram alert sent for {alert['product']}")

# Usage:
# alerts = check_prices()  # from the monitoring script
# if alerts:
#     send_telegram_alert(alerts)

import sqlite3
from datetime import datetime

DB_PATH = "price_history.db"

def init_db():
    """Create the price history table."""
    conn = sqlite3.connect(DB_PATH)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS prices (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            product_name TEXT NOT NULL,
            platform TEXT NOT NULL,
            offer_price REAL NOT NULL,
            mrp REAL NOT NULL,
            discount_pct REAL,
            in_stock INTEGER,
            recorded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    conn.commit()
    conn.close()

def record_price(product_name: str, platform: str,
                 offer_price: float, mrp: float, in_stock: bool):
    """Save a price observation."""
    discount_pct = round((1 - offer_price / mrp) * 100, 1) if mrp > 0 else 0
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        """INSERT INTO prices
           (product_name, platform, offer_price, mrp, discount_pct, in_stock)
           VALUES (?, ?, ?, ?, ?, ?)""",
        (product_name, platform, offer_price, mrp, discount_pct, int(in_stock)),
    )
    conn.commit()
    conn.close()

def get_price_trend(product_name: str, platform: str, days: int = 30):
    """Get price history for the last N days."""
    conn = sqlite3.connect(DB_PATH)
    rows = conn.execute(
        """SELECT offer_price, mrp, discount_pct, recorded_at
           FROM prices
           WHERE product_name = ? AND platform = ?
             AND recorded_at >= datetime('now', ?)
           ORDER BY recorded_at DESC""",
        (product_name, platform, f"-{days} days"),
    ).fetchall()
    conn.close()
    return rows

def detect_percentage_drop(product_name: str, platform: str,
                           current_price: float, threshold_pct: float = 10):
    """Alert if current price is X% below the 7-day average."""
    conn = sqlite3.connect(DB_PATH)
    row = conn.execute(
        """SELECT AVG(offer_price) FROM prices
           WHERE product_name = ? AND platform = ?
             AND recorded_at >= datetime('now', '-7 days')""",
        (product_name, platform),
    ).fetchone()
    conn.close()

    if row and row[0]:
        avg_price = row[0]
        drop_pct = ((avg_price - current_price) / avg_price) * 100
        if drop_pct >= threshold_pct:
            return {
                "product": product_name,
                "platform": platform,
                "avg_7d": round(avg_price, 2),
                "current": current_price,
                "drop_pct": round(drop_pct, 1),
            }
    return None