Deploy version 0.1.0

Dockerfile

@@ -0,0 +1,34 @@
+FROM python:3.11-slim
+
+# Install uv
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
+
+WORKDIR /app
+
+# Copy dependency files first (changes rarely)
+COPY pyproject.toml uv.lock ./
+
+# Put venv outside of /app so it won't be affected by volume mounts
+ENV UV_PROJECT_ENVIRONMENT=/opt/venv
+
+# Install dependencies (this will now create venv at /opt/venv)
+RUN uv sync --locked --no-group docs --no-group notebook
+
+# Create a non-root user
+RUN useradd -m appuser
+
+# Copy all source code
+COPY . .
+
+# Set ownership of writable dirs
+RUN mkdir -p /app/memory && chown -R appuser:appuser /app/memory
+
+# Switch to non-root user
+USER appuser
+
+# Set environment variables
+ENV PYTHONUNBUFFERED=1
+
+# Default command - use uv run to execute with the virtual environment
+CMD ["bash", "-c", "set -e && uv run modal deploy -m src.modal_services.entry && uv run main.py"]
+# CMD ["uv", "run", "main.py"]

README.md

@@ -1,11 +1,56 @@
 ---
 title: Snapr
-emoji: 📈
+emoji: 🏷️
 colorFrom: yellow
-colorTo: green
+colorTo: blue
 sdk: docker
 pinned: false
-short_description: 🏷️ Multi-AI agent system trained to snap the best deals
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# 🏷️ Snapr - Multi-AI Agent Snaps Best Deals
+
+- [🚀 Live Demo on Hugging Face](https://huggingface.co/spaces/lisekarimi/snapr)
+- [📚 Full Documentation](https://lisekarimi.github.io/snapr)
+
+📷 **Screenshots**
+
+<a href="docs/styles/assets/ui/full_app.png">
+  <img src="docs/styles/assets/ui/full_app.png" width="400">
+</a>
+
+---
+
+## 📖 Overview 
+**Snapr** is an AI-powered system that finds the best online deals using a pipeline of collaborative agents powered by LLMs, traditional ML models, embeddings, and retrieval-augmented generation (RAG).
+
+### ⚙️ Key Capabilities
+- Central agent orchestrates deal processing with specialized agents
+- Real-time deal fetching and price prediction using LLMs and traditional models
+- Automatic filtering and memory to avoid duplicates
+- Remote execution on Modal with live activity logs
+- DevOps: Docker, CI/CD with GitHub Actions, pre-commit hooks
+- Unit testing, security checks, and code quality enforcement
+- Auto-generated changelogs with LLM and MkDocs-based documentation
+
+👉 For full details, see [Core Features and Development & DevOps.](https://lisekarimi.github.io/snapr).
+
+---
+
+## ⚙️ Setup & Installation 
+
+To set up locally and install the app, see the "Local Deployment" section in the [technical documentation](https://lisekarimi.github.io/snapr/technical/localdev/).
+
+---
+
+## 📚 Docs & Architecture
+
+- Documentation includes:
+
+    - Technical and functional docs
+    - Schematics of technical architecture, functional workflow, and user flow
+
+---
+
+## 🪪 License
+
+MIT

main.py

@@ -0,0 +1,8 @@
+"""Entry point for the application."""
+
+from src.ui.gradio_app import build_ui
+
+demo = build_ui()
+
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0", server_port=7860)

pyproject.toml

@@ -0,0 +1,52 @@
+[project]
+name = "snapr"
+version = "0.1.0"
+description = "Multi-AI agent system trained to snap the best deals"
+readme = "README.md"
+requires-python = ">=3.11, <3.12"
+dependencies = [
+    "beautifulsoup4>=4.13.4",
+    "feedparser>=6.0.11",
+    "gradio==5.29.1",
+    "modal==0.74.20",
+    "openai==1.65.5",
+    "python-dotenv>=1.1.0",
+    "requests>=2.32.3",
+    "rich>=14.0.0",
+]
+
+[dependency-groups]
+docs = [
+    "mkdocs-glightbox>=0.4.0",
+    "mkdocs-macros-plugin>=1.3.7",
+    "mkdocs-material>=9.6.14",
+]
+notebook = [
+    "datasets==2.21.0",
+    "ipykernel>=6.29.5",
+    "ipywidgets>=8.1.7",
+    "matplotlib>=3.10.3",
+    "numpy>=2.2.6",
+    "pandas>=2.2.3",
+    "transformers>=4.52.3",
+]
+
+[tool.pytest.ini_options]
+pythonpath = ["."]
+filterwarnings = [
+    "ignore::DeprecationWarning:websockets.legacy",
+]
+
+[tool.ruff]
+target-version = "py39"
+line-length = 88
+
+[tool.ruff.lint]
+select = ["E", "F", "I"]
+ignore = ["F821"]
+
+[tool.ruff.lint.per-file-ignores]
+"src/agents/*.py" = ["ANN"]
+"tests/*.py" = ["ANN"]
+"notebooks/**/*" = ["ALL"]
+"**/*.ipynb" = ["ALL"]

src/agents/base_agent.py

@@ -0,0 +1,21 @@
+"""Base class for Agents, providing colorful logging with Rich."""
+
+from src.config.logging_queue import log_queue
+from src.utils.logger import console
+
+
+class Agent:
+    """Abstract superclass for Agents, with colorful Rich logging."""
+
+    name: str = ""
+    color: str = "white"
+
+    def log(self, message: str) -> None:
+        """Print log with colored message using Rich."""
+        # Terminal (Rich)
+        console.print(f"[{self.color} on black][{self.name}] {message}[/]")
+
+        # Gradio UI (HTML)
+        log_queue.put(
+            f"<span style='color:{self.color}'>[{self.name}] {message}</span><br>"
+        )

src/agents/deal_scanner_agent.py

@@ -0,0 +1,150 @@
+"""Scan deals and return top 5 via OpenAI.
+
+1. Fetch deals from RSS feeds.
+2. Prompt OpenAI with the deal list.
+3. Return top 5 detailed, clearly priced deals as structured JSON.
+"""
+
+import json
+import os
+from typing import Any, Dict, List, Optional
+
+from src.agents.base_agent import Agent
+from src.config.constants import DEALS_FILE
+from src.deals.raw_deals import ScrapedDeal
+from src.deals.structured_deals import OpportunitiesCollection
+from src.models.frontier_model import OPENAI_MODEL, openai
+
+
+class DealScannerAgent(Agent):
+    """Agent for scanning and filtering deals.
+
+    Processes them via OpenAI.
+    """
+
+    name = "Deal Scanner Agent"
+    color = "green"
+
+    SYSTEM_PROMPT = """
+    You are a deal filtering assistant.
+
+    Your task is to identify the 5 deals with the most detailed product descriptions 
+    and clearly stated prices. Focus only on the product itself — not the deal terms, 
+    discounts, or promotions.
+
+    Only include deals where the price is explicitly mentioned and easy to extract. 
+    Avoid entries with phrases like "$XXX off" or "reduced by $XXX" — those are not 
+    valid prices. Only include deals when you are confident about the actual 
+    product price.
+
+    Respond strictly in JSON with no explanation, using the following format:
+
+    {
+    "deals": [
+        {
+        "product_description": "A clear, 4–5 sentence summary of the product.",
+        "price": 99.99,
+        "url": "..."
+        },
+        ...
+    ]
+    }"""
+
+    def __init__(self, memory_path: str = DEALS_FILE) -> None:
+        """Initialize OpenAI client."""
+        self.openai = openai
+        self.memory_path = memory_path
+        self.log("is ready")
+
+    def _load_memory(self) -> Dict[str, List[Dict[str, Any]]]:
+        """Load memory from file, returning seen URLs and the full memory."""
+        if os.path.exists(self.memory_path):
+            try:
+                with open(self.memory_path, "r") as f:
+                    memory_json = json.load(f)
+                seen_urls = [op["url"] for op in memory_json.get("opportunities", [])]
+                return {
+                    "seen_urls": seen_urls,
+                    "memory": memory_json.get("opportunities", []),
+                }
+            except (json.JSONDecodeError, KeyError, IOError) as e:
+                self.log(f"Error loading memory: {e}. Creating new memory.")
+                return {"seen_urls": [], "memory": []}
+        else:
+            self.log("No memory file found. Assuming first run")
+            return {"seen_urls": [], "memory": []}
+
+    def fetch_deals(self, categories: List[str]) -> List[ScrapedDeal]:
+        """Fetch new RSS deals not present in memory."""
+        self.log("is fetching deals from RSS feed")
+
+        # Load memory to get seen URLs
+        memory_data = self._load_memory()
+        seen_urls = set(memory_data["seen_urls"])
+
+        # Fetch all deals and filter out seen ones
+        try:
+            scraped = ScrapedDeal.fetch(categories)
+            result = [deal for deal in scraped if deal.url not in seen_urls]
+            overlap = [deal for deal in scraped if deal.url in seen_urls]
+            self.log(f"{len(overlap)} deals skipped")
+            self.log(f"{len(result)} new deals fetched")
+            return result
+        except Exception as e:
+            self.log(f"Error fetching deals: {e}")
+            return []
+
+    def make_user_prompt(self, scraped: List[ScrapedDeal]) -> str:
+        """Build the full user prompt for OpenAI."""
+        return (
+            "Select the 5 best deals with the clearest product descriptions "
+            "and exact prices. Here is the list:\n\n"
+            + "\n\n".join(deal.describe() for deal in scraped)
+        )
+
+    def scan(self, categories: List[str]) -> Optional[OpportunitiesCollection]:
+        """Return top 5 new deals."""
+        # Step 1: Fetch new deals not already in memory
+        scraped = self.fetch_deals(categories)
+        if not scraped:
+            self.log("❌ found no new deals to process ")
+            return None
+
+        # Step 2: Construct prompt with all new deals
+        user_prompt = self.make_user_prompt(scraped)
+
+        # Step 3: Call OpenAI - allow RuntimeError to propagate
+        result = self._call_openai(user_prompt)
+
+        # Step 4: Filter out invalid deals
+        filtered_result = self._filter_invalid_deals(result)
+
+        return filtered_result if filtered_result.opportunities else None
+
+    def _call_openai(self, user_prompt: str) -> OpportunitiesCollection:
+        """Call OpenAI API to get the processed deals."""
+        self.log("📞 is calling OpenAI")
+        try:
+            result = self.openai.beta.chat.completions.parse(
+                model=OPENAI_MODEL,
+                messages=[
+                    {"role": "system", "content": self.SYSTEM_PROMPT},
+                    {"role": "user", "content": user_prompt},
+                ],
+                response_format=OpportunitiesCollection,
+            )
+        except Exception as e:
+            self.log(f"[ERROR] OpenAI call failed: {e}")
+            raise RuntimeError(
+                "DealScannerAgent failed to get response from OpenAI."
+            ) from e
+        return result
+
+    def _filter_invalid_deals(
+        self, result: OpportunitiesCollection
+    ) -> OpportunitiesCollection:
+        """Filter out deals with invalid prices."""
+        result = result.choices[0].message.parsed
+        result.opportunities = [op for op in result.opportunities if op.price > 0]
+        self.log(f"✅ received {len(result.opportunities)} valid opportunities ")
+        return result

src/agents/ensemble_price_agent.py

@@ -0,0 +1,57 @@
+"""Agent calling a remote ensemble model on Modal.
+
+Computes final price from multiple predictions.
+"""
+
+import modal
+
+from src.agents.base_agent import Agent
+from src.agents.ft_price_agent import FTPriceAgent
+from src.agents.rag_price_agent import RAGPriceAgent
+from src.agents.xgb_price_agent import XGBoostPriceAgent
+from src.config.constants import CURRENCY
+from src.modal_services.app_config import APP_NAME
+
+
+class EnsemblePriceAgent(Agent):
+    """Agent that aggregates FT, RAG, and XGB predictions.
+
+    Sends them to the remote EnsemblePricer on Modal.
+    """
+
+    name = "EnsemblePrice Agent"
+    color = "magenta"
+
+    def __init__(self) -> None:
+        """Initialize the agent."""
+        self._modal_called = False
+        self.ft_agent = FTPriceAgent()
+        self.rag_agent = RAGPriceAgent()
+        self.xgb_agent = XGBoostPriceAgent()
+        remote_ensemble = modal.Cls.from_name(APP_NAME, "EnsemblePricer")
+        self.ensemble = remote_ensemble()
+        self.log("is ready")
+
+    def price(self, description: str) -> float:
+        """Get individual predictions and pass them to the ensemble model."""
+        ft_pred = self.ft_agent.price(description)
+        rag_pred = self.rag_agent.price(description)
+        xgb_pred = self.xgb_agent.price(description)
+
+        if not self._modal_called:
+            self.log("📡 Connecting to Modal — Loading trained linear model...")
+            self._modal_called = True
+
+        self.log(
+            f"Predictions — FT={CURRENCY}{ft_pred}, "
+            f"RAG={CURRENCY}{rag_pred}, "
+            f"XGB={CURRENCY}{xgb_pred}"
+        )
+
+        try:
+            result = self.ensemble.price.remote(ft_pred, rag_pred, xgb_pred)
+            self.log(f"Final estimate: {CURRENCY}{result:.2f}")
+            return result
+        except Exception as e:
+            self.log(f"[ERROR] Remote EnsemblePricer failed: {e}")
+            raise RuntimeError("EnsemblePriceAgent failed to get final price.") from e

src/agents/ft_price_agent.py

@@ -0,0 +1,38 @@
+"""FTPriceAgent uses a remote fine-tuned LLM on Modal.
+
+Predicts item prices from descriptions.
+"""
+
+import modal
+
+from src.agents.base_agent import Agent
+from src.modal_services.app_config import APP_NAME
+
+
+class FTPriceAgent(Agent):
+    """Agent for running a fine-tuned LLM remotely."""
+
+    name = "FTPrice Agent"
+    color = "red"
+
+    def __init__(self) -> None:
+        """Initialize agent with Modal class instance."""
+        self._modal_called = False
+        ft_pricer = modal.Cls.from_name(APP_NAME, "FTPricer")
+        self.ftpricer = ft_pricer()
+        self.log("is ready")
+
+    def price(self, description: str) -> float:
+        """Remote call to estimate price, with error handling."""
+        if not self._modal_called:
+            self.log("🧠 Calling Modal's fine-tuned LLM...")
+            self._modal_called = True
+        try:
+            result = self.ftpricer.price.remote(
+                description
+            )  # 2nd API call: run price method
+            # self.log(f"predicting ${result:.2f} ✅")
+            return result
+        except Exception as e:
+            self.log(f"[ERROR] Remote pricing failed: {e}")
+            raise RuntimeError("FTPriceAgent failed to get price from Modal.") from e

src/agents/pipeline.py

@@ -0,0 +1,198 @@
+"""Executes planning agent and streams logs/results.
+
+Integrates with the Gradio UI.
+"""
+
+import queue
+import threading
+import time
+import traceback
+from typing import Any, Generator, List, Optional, Tuple, Union
+
+import gradio as gr
+
+from src.agents.planning_agent import PlanningAgent
+from src.config.constants import (
+    DEALS_FILE,
+    IS_DEMO_VERSION,
+    MAX_CATEGORY_SELECTION,
+)
+from src.config.logging_queue import log_queue
+from src.ui.formatting import format_deals_table, html_for
+from src.utils.cleanup import delete_if_old
+from src.utils.state_manager import can_run_app, get_state, update_state
+
+
+def run_pipeline(log_queue: queue.Queue, selected_categories: List[str]) -> None:
+    """Runs the planning agent pipeline and stores accepted deals."""
+    try:
+        delete_if_old(DEALS_FILE)
+        agent = PlanningAgent()
+        results = agent.plan(selected_categories)
+
+        global accepted_deals
+        accepted_deals = [
+            [
+                opp.product_description,
+                f"${opp.price:.2f}",
+                f"${opp.estimate:.2f}",
+                f"${opp.discount:.2f}",
+                opp.url,
+            ]
+            for opp in results
+        ]
+
+    except Exception as e:
+        log_queue.put(
+            f"<span style='color:red'>❌ Error during pipeline execution: "
+            f"{str(e)}</span>"
+        )
+        log_queue.put(f"<pre>{traceback.format_exc()}</pre>")
+
+
+def validate_categories(
+    selected_categories: Union[str, List[str]],
+) -> Tuple[bool, Optional[str]]:
+    """Validates the selected categories."""
+    if isinstance(selected_categories, str):
+        selected_categories = [selected_categories]
+
+    if not selected_categories:
+        return False, "⚠️ Please select at least one category before running."
+
+    if len(selected_categories) > MAX_CATEGORY_SELECTION:
+        return (
+            False,
+            f"⚠️ You can select up to {MAX_CATEGORY_SELECTION} categories only.",
+        )
+
+    return True, None
+
+
+def check_demo_restrictions() -> Tuple[bool, Optional[str], Optional[str]]:
+    """Checks if the app can run under demo restrictions."""
+    can_run, message = can_run_app()
+
+    if not can_run:
+        return False, f"⚠️ {message}", None
+
+    if IS_DEMO_VERSION:
+        # Just update the run count, but use the message from can_run_app
+        update_state({"run_count": get_state()["run_count"] + 1})
+        # Get fresh message after updating the state
+        _, status_msg = can_run_app()
+    else:
+        status_msg = ""
+
+    return True, None, status_msg
+
+
+def initial_ui_update(
+    log_data: List[str], status_msg: str
+) -> Tuple[str, str, Any, str]:
+    """Returns initial UI state for the app."""
+    disable_btn = gr.update(
+        interactive=False, elem_classes=["run-button", "btn-disabled"]
+    )
+    return html_for(log_data), format_deals_table([]), disable_btn, status_msg
+
+
+def run_pipeline_threaded(
+    selected_categories: List[str],
+    log_data: List[str],
+    status_msg: str,
+    enable_btn: Any,  # noqa: ANN401
+) -> Generator[Tuple[str, str, Any, str], None, None]:
+    """Runs pipeline in background thread with log streaming.
+
+    Yields UI updates until completion.
+
+    """
+    thread = threading.Thread(
+        target=run_pipeline, args=(log_queue, selected_categories)
+    )
+    thread.start()
+
+    disable_btn = gr.update(
+        interactive=False, elem_classes=["run-button", "btn-disabled"]
+    )
+
+    while thread.is_alive() or not log_queue.empty():
+        while not log_queue.empty():
+            log_msg = log_queue.get()
+            log_data.append(log_msg)
+            yield (
+                html_for(log_data),
+                format_deals_table(accepted_deals),
+                disable_btn,
+                status_msg,
+            )
+
+        if thread.is_alive():
+            time.sleep(0.2)
+            yield (
+                html_for(log_data),
+                format_deals_table(accepted_deals),
+                disable_btn,
+                status_msg,
+            )
+
+    # Final UI update after thread finishes
+    yield html_for(log_data), format_deals_table(accepted_deals), enable_btn, status_msg
+
+
+def handle_pipeline_error(
+    e: Exception,
+    log_data: List[str],
+    enable_btn: gr.components.Component,
+    status_msg: str,
+) -> Tuple[str, str, gr.components.Component, str]:
+    """Handles exceptions and appends error logs.
+
+    Returns the final UI update tuple.
+    """
+    log_data.append(f"<span style='color:red'>❌ Unexpected error: {str(e)}</span>")
+    log_data.append(f"<pre>{traceback.format_exc()}</pre>")
+    return html_for(log_data), format_deals_table([]), enable_btn, status_msg
+
+
+def run_and_stream_logs(
+    selected_categories: Union[str, List[str]],
+) -> Generator[Tuple[str, str, bool, str], None, None]:
+    """Runs pipeline in a thread, streaming logs and results to the UI.
+
+    Returns HTML logs, deal table, button state, and status message.
+    """
+    global accepted_deals
+    accepted_deals = []
+    log_data = []
+
+    # Step 1: Validate categories
+    is_valid, error_msg = validate_categories(selected_categories)
+    if not is_valid:
+        yield None, None, gr.update(interactive=True), error_msg
+        return
+
+    # Step 2: Check demo restrictions
+    can_run, error_msg, status_msg = check_demo_restrictions()
+    if not can_run:
+        yield (
+            html_for([error_msg]),
+            format_deals_table([]),
+            gr.update(interactive=True),
+            error_msg,
+        )
+        return
+
+    # Step 3: Initial UI update showing we're starting
+    enable_btn = gr.update(interactive=True, elem_classes=["run-button"])
+    yield initial_ui_update(log_data, status_msg)
+
+    try:
+        # Step 4: Run the pipeline in a thread
+        yield from run_pipeline_threaded(
+            selected_categories, log_data, status_msg, enable_btn
+        )
+
+    except Exception as e:
+        yield handle_pipeline_error(e, log_data, enable_btn, status_msg)

src/agents/planning_agent.py

@@ -0,0 +1,116 @@
+"""PlanningAgent coordinates deal scanning and enrichment."""
+
+import json
+from typing import List
+
+from rich import print_json
+
+from src.agents.base_agent import Agent
+from src.agents.deal_scanner_agent import DealScannerAgent
+from src.agents.ensemble_price_agent import EnsemblePriceAgent
+from src.config.constants import CURRENCY, DEAL_THRESHOLD
+from src.deals.structured_deals import OpportunitiesCollection, Opportunity
+from src.utils.logger import console
+from src.utils.memory_utils import save_opportunities_to_memory
+
+
+class PlanningAgent(Agent):
+    """Create instances of the Agents that this planner coordinates across."""
+
+    name = "Planning Agent"
+    color = "cyan"
+
+    def __init__(self) -> None:
+        """Initialize agents."""
+        self.log("🧠 Let’s wake up the agents — time to sniff out some sweet deals!")
+        self.log("is ready")
+        self.scanner = DealScannerAgent()
+        self.ensemble = EnsemblePriceAgent()
+        self.log("🚀 All AI Agents are caffeinated, calibrated, and ready to hustle..")
+
+    def scan_deals(self, categories: List[str]) -> List[Opportunity]:
+        """Scans deals and returns GPT-processed opportunities."""
+        result = self.scanner.scan(categories)
+        if result is None:
+            self.log("❌ No valid deals found.")
+            return []
+        return result.opportunities
+
+    def enrich(self, opportunity: Opportunity) -> Opportunity:
+        """Add estimated market price and discount to an opportunity."""
+        estimate = self.ensemble.price(opportunity.product_description)
+        discount = round(estimate - opportunity.price, 2)
+        opportunity.estimate = estimate
+        opportunity.discount = discount
+        return opportunity
+
+    def _log_result(self, idx: int, opportunity: Opportunity) -> None:
+        """Logs if a deal was accepted or rejected.
+
+        Decision is based on discount vs. threshold.
+
+        """
+        if opportunity.discount >= DEAL_THRESHOLD:
+            self.log(
+                f"✅ Deal #{idx} accepted — discount: "
+                f"{CURRENCY}{opportunity.discount:.2f}"
+            )
+        else:
+            self.log(
+                f"❌ Deal #{idx} rejected — discount below threshold: "
+                f"{CURRENCY}{opportunity.discount:.2f}"
+            )
+
+    def _report_summary(self, enriched: List[Opportunity]) -> None:
+        """Display a summary of accepted opportunities after enrichment."""
+        if not enriched:
+            self.log("❌ No opportunities met the discount threshold.")
+        else:
+            for opp in enriched:
+                console.print(
+                    f"- {opp.product_description}\n"
+                    f"  Price: {CURRENCY}{opp.price:.2f} | "
+                    f"AI Estimate: {CURRENCY}{opp.estimate:.2f} | "
+                    f"Discount: {CURRENCY}{opp.discount:.2f}\n"
+                    f"  URL: {opp.url}\n"
+                )
+
+    def plan(self, categories: List[str]) -> List[Opportunity]:
+        """Full pipeline: scan → enrich → filter → save."""
+        self.log(
+            "************** SCANNING INITIATED — HUNTING JUICY DEALS...**************"
+        )
+
+        deals = self.scan_deals(categories)
+        if not deals:
+            self.log("❌ No deals found from scanner.")
+            return []
+        print_json(
+            data=json.loads(
+                OpportunitiesCollection(opportunities=deals).model_dump_json()
+            )
+        )  # For debugging/inspection
+
+        self.log(
+            "************** SCANNING COMPLETE — STARTING ENRICHMENT **************"
+        )
+
+        enriched = []
+        for idx, deal in enumerate(deals, start=1):
+            opportunity = self.enrich(deal)
+            self._log_result(idx, opportunity)
+            if opportunity.discount >= DEAL_THRESHOLD:
+                enriched.append(opportunity)
+
+        self.log(
+            "************** ENRICHMENT COMPLETE — SAVING OPPORTUNITIES **************"
+        )
+        save_opportunities_to_memory([opp.model_dump() for opp in enriched])
+        self.log(f"💾 {len(enriched)} top deals saved to memory.")
+
+        self._report_summary(enriched)
+        self.log(
+            "************** ✅ MISSION COMPLETE — BEST DEALS LOCKED IN **************"
+        )
+
+        return enriched

src/agents/rag_price_agent.py

@@ -0,0 +1,36 @@
+"""Handles the integration of RAG model with Modal to predict item prices."""
+
+import modal
+
+from src.agents.base_agent import Agent
+from src.modal_services.app_config import APP_NAME
+
+
+class RAGPriceAgent(Agent):
+    """RAGPriceAgent connects to a remote Modal container.
+
+    Uses a RAG pipeline to predict item prices.
+    """
+
+    name = "RAGPrice Agent"
+    color = "blue"
+
+    def __init__(self) -> None:
+        """Initialize the agent."""
+        self._modal_called = False
+        remote_rag_pricer = modal.Cls.from_name(APP_NAME, "RAGPricer")
+        self.rag = remote_rag_pricer()
+        self.log("is ready")
+
+    def price(self, description: str) -> float:
+        """Call the remote RAGPricer to estimate price."""
+        if not self._modal_called:
+            self.log("📡 Connecting to Modal — loading embedding model and ChromaDB...")
+            self._modal_called = True
+        try:
+            result = self.rag.price.remote(description)
+            # self.log(f"predicting ${result:.2f} ✅")
+            return result
+        except Exception as e:
+            self.log(f"[ERROR] Remote RAGPricer failed: {e}")
+            raise RuntimeError("RAGPriceAgent failed to get price from Modal.") from e

src/agents/xgb_price_agent.py

@@ -0,0 +1,38 @@
+"""Handles the integration of XGBoost model with Modal to predict item prices."""
+
+import modal
+
+from src.agents.base_agent import Agent
+from src.modal_services.app_config import APP_NAME
+
+
+class XGBoostPriceAgent(Agent):
+    """XGBoostPriceAgent connects to a remote Modal container.
+
+    Uses E5 and XGBoost to predict prices from descriptions.
+    """
+
+    name = "XGBPrice Agent"
+    color = "yellow"
+
+    def __init__(self) -> None:
+        """Initialize the agent."""
+        self._modal_called = False
+        remote_xgb_pricer = modal.Cls.from_name(APP_NAME, "XGBPricer")
+        self.xgb = remote_xgb_pricer()
+        self.log("is ready")
+
+    def price(self, description: str) -> float:
+        """Call the remote XGBPricer to estimate price."""
+        if not self._modal_called:
+            self.log("📡 Connecting to Modal — loading XGBoost and embedding model...")
+            self._modal_called = True
+        try:
+            result = self.xgb.price.remote(description)
+            # self.log(f"predicting ${result:.2f} ✅")
+            return result
+        except Exception as e:
+            self.log(f"[ERROR] Remote XGBPricer failed: {e}")
+            raise RuntimeError(
+                "XGBoostPriceAgent failed to get price from Modal."
+            ) from e

src/config/constants.py

@@ -0,0 +1,39 @@
+# src/config/constants.py
+"""Constants for configuration across the project."""
+
+from pathlib import Path
+
+import tomllib
+
+# ==================== PROJECT METADATA ====================
+root = Path(__file__).parent.parent.parent
+with open(root / "pyproject.toml", "rb") as f:
+    pyproject = tomllib.load(f)
+
+PROJECT_NAME = pyproject["project"]["name"]
+VERSION = pyproject["project"]["version"]
+
+# ==================== ENVIRONMENT CONFIG ====================
+ENV = "PROD"  # or "PROD"
+IS_PROD = ENV == "PROD"
+IS_DEMO_VERSION = IS_PROD  # True if PROD, False if DEV
+
+# App naming for Modal
+APP_NAME = f"{PROJECT_NAME}-{ENV.lower()}-{VERSION}"
+
+# ==================== APPLICATION LIMITS ====================
+MAX_DEMO_RUNS_PER_DAY = 5
+MAX_LOG_LINES = 50
+MAX_DEALS_PER_FEED = 20
+MAX_CATEGORY_SELECTION = 3
+MEMORY_EXPIRATION_DAYS = 5
+
+# ==================== BUSINESS LOGIC ====================
+CURRENCY = "$"
+DEAL_THRESHOLD = 50
+
+# ==================== PATHS ====================
+BASE_DIR = Path(__file__).resolve().parent.parent.parent
+MEMORY_DIR = BASE_DIR / "memory"
+STATE_FILE = MEMORY_DIR / "demo_state.json"
+DEALS_FILE = MEMORY_DIR / "memory.json"

src/config/feeds.py

@@ -0,0 +1,18 @@
+"""Maps deal categories to RSS feed URLs.
+
+Shared across UI, agents, and scrapers.
+"""
+
+CATEGORY_FEEDS = {
+    "Home & Garden": "https://www.dealnews.com/c196/Home-Garden/?rss=1",
+    "Clothing & Accessories": "https://www.dealnews.com/c202/Clothing-Accessories/?rss=1",
+    "Electronics": "https://www.dealnews.com/c142/Electronics/?rss=1",
+    "Health & Beauty": "https://www.dealnews.com/c756/Health-Beauty/?rss=1",
+    "Computers": "https://www.dealnews.com/c39/Computers/?rss=1",
+    "Sports & Fitness": "https://www.dealnews.com/c211/Sports-Fitness/?rss=1",
+    "Gaming & Toys": "https://www.dealnews.com/c186/Gaming-Toys/?rss=1",
+    "Automotive": "https://www.dealnews.com/c238/Automotive/?rss=1",
+    "Movies, Music & Books": "https://www.dealnews.com/c178/Movies-Music-Books/?rss=1",
+    "Office & School Supplies": "https://www.dealnews.com/c182/Office-School-Supplies/?rss=1",
+    "Special Occasion": "https://www.dealnews.com/c636/Special-Occasion/?rss=1",
+}

src/config/logging_queue.py

@@ -0,0 +1,8 @@
+"""Initializes a thread-safe global log queue.
+
+Used to manage log messages across threads.
+"""
+
+import queue
+
+log_queue = queue.Queue()

src/deals/raw_deals.py

@@ -0,0 +1,146 @@
+"""Handles scraping and preprocessing logic before OpenAI interaction."""
+
+import time
+from typing import Dict, List, Self
+
+import feedparser
+import requests
+from bs4 import BeautifulSoup
+
+from src.config.constants import MAX_DEALS_PER_FEED
+from src.config.feeds import CATEGORY_FEEDS
+from src.utils.logger import console
+
+
+def extract(html_snippet: str) -> str:
+    """Cleans text from messy HTML with fallback handling."""
+    soup = BeautifulSoup(html_snippet, "html.parser")
+    snippet = soup.find("div", class_="snippet summary")
+
+    # Extract inner HTML or fallback to full snippet
+    raw_html = snippet.decode_contents() if snippet else html_snippet
+
+    # Parse again to clean any nested/malformed HTML
+    clean_soup = BeautifulSoup(raw_html, "html.parser")
+    text = clean_soup.get_text(" ", strip=True)
+
+    return text.replace("\n", " ")
+
+
+class ScrapedDeal:
+    """Represents a deal from an RSS feed.
+
+    Flow: fetch() → __init__ → _load_content() → use methods.
+    """
+
+    category: str  # Deal type
+    title: str  # Deal title
+    summary: str  # RSS summary
+    url: str  # Deal link
+    details: str  # Full description
+    features: str  # Feature list
+
+    def __init__(self, entry: Dict[str, str]) -> None:
+        """Initialize deal from RSS entry and fetch content."""
+        # Basic metadata from RSS
+        self.title = entry["title"]
+        self.summary = extract(entry["summary"])
+        self.url = entry["links"][0]["href"]
+
+        # Initialize placeholders
+        self.details = ""
+        self.features = ""
+
+        # Fetch and parse full deal content
+        self._load_content()
+
+    def _load_content(self) -> None:
+        """Fetches and parses deal content; raises on failure to skip."""
+        try:
+            res = requests.get(self.url, timeout=5)
+            res.raise_for_status()
+
+            soup = BeautifulSoup(res.content, "html.parser")
+            content = soup.find("div", class_="content-section")
+
+            if content:
+                text = content.get_text().replace("\nmore", "").replace("\n", " ")
+                if "Features" in text:
+                    self.details, self.features = text.split("Features", 1)
+                else:
+                    self.details = text
+                    self.features = ""
+            else:
+                raise ValueError("No content section found.")
+
+        except Exception as e:
+            raise RuntimeError(f"Failed to load deal content from {self.url}: {e}")
+
+    def __repr__(self) -> str:
+        """Quick string representation of the deal."""
+        return f"<{self.title}>"
+
+    def describe(self) -> str:
+        """Detailed description of the deal."""
+        return (
+            f"Title: {self.title.strip()}\n"
+            f"Details: {self.details.strip()}\n"
+            f"Features: {self.features.strip()}\n"
+            f"URL: {self.url.strip()}"
+        )
+
+    @classmethod
+    def fetch(cls, selected_categories: List[str]) -> List[Self]:
+        """Parses RSS feeds into ScrapedDeal instances.
+
+        Skips failed deals; stops app if all fail.
+        """
+        deals = []
+        feed_urls = [
+            CATEGORY_FEEDS[cat] for cat in selected_categories if cat in CATEGORY_FEEDS
+        ]
+
+        for feed_url in feed_urls:
+            feed = cls._parse_feed(feed_url)
+            if feed is None:
+                continue
+
+            console.print(
+                f"[bold blue]DEBUG[/] {len(feed.entries)} entries found in feed: "
+                f"{feed_url}"
+            )
+
+            for entry in feed.entries[:MAX_DEALS_PER_FEED]:
+                cls._process_deal(entry, deals)
+
+            # Throttle requests to avoid hitting servers too fast
+            time.sleep(0.5)
+
+        if not deals:
+            raise RuntimeError("❌ All deals failed to load. Stopping.")
+
+        return deals
+
+    @staticmethod
+    def _parse_feed(feed_url: str) -> feedparser.FeedParserDict | None:
+        """Helper method to parse the RSS feed and return the feed data."""
+        feed = feedparser.parse(feed_url)
+        if feed.bozo:
+            console.print(
+                f"[bold red]ERROR[/] Failed to parse RSS feed: {feed_url} "
+                f"({feed.bozo_exception})"
+            )
+            return None
+        return feed
+
+    @staticmethod
+    def _process_deal(entry: Dict[str, str], deals: List[Self]) -> None:
+        """Helper method to process each RSS entry and add valid deals."""
+        try:
+            deal = ScrapedDeal(entry)
+            deals.append(deal)
+        except Exception as e:
+            console.print(
+                f"[bold yellow]WARN[/] Skipped deal "
+                f"'{entry.get('title', 'Unknown')}' due to error: {e}"
+            )

src/deals/structured_deals.py

@@ -0,0 +1,24 @@
+"""Defines post-GPT deal classes.
+
+Used for further analysis as structured opportunities.
+"""
+
+from typing import List, Optional
+
+from pydantic import BaseModel
+
+
+class Opportunity(BaseModel):
+    """A single opportunity (final deal) after GPT response."""
+
+    product_description: str  # Full description of the product
+    price: float  # Listed price from the deal feed
+    url: str  # Link to the product
+    estimate: Optional[float] = None  # predicted price
+    discount: Optional[float] = None  # estimate - price
+
+
+class OpportunitiesCollection(BaseModel):
+    """A list of top opportunities selected by GPT."""
+
+    opportunities: List[Opportunity]  # High-quality final deals

src/modal_services/app_config.py

@@ -0,0 +1,50 @@
+"""Defines shared Modal configuration.
+
+Includes constants, image, volume, secrets, and app setup.
+"""
+
+from modal import App, Image, Secret, Volume
+
+from src.config.constants import APP_NAME
+
+CACHE_PATH = "/cache"
+GPU = "T4"
+
+# Modal image, volume, and secrets
+image = (
+    Image.debian_slim()
+    .pip_install(
+        "huggingface",
+        "torch",
+        "transformers",
+        "bitsandbytes",
+        "accelerate",
+        "peft",
+        "sentence-transformers",
+        "xgboost",
+        "joblib",
+        "chromadb",
+        "openai",
+        "numpy",
+        "pandas",
+    )
+    .env({"HF_HUB_CACHE": CACHE_PATH})
+    .add_local_file(local_path="pyproject.toml", remote_path="/root/pyproject.toml")
+)
+
+# Modal setup
+app = App(APP_NAME, image=image)
+
+cache_vol = Volume.from_name("hf-hub-cache", create_if_missing=True)
+secrets = [Secret.from_name("HF_TOKEN"), Secret.from_name("OPENAI_API_KEY")]
+
+# Shared Modal class config for all model agents
+modal_class_kwargs = dict(
+    image=image,
+    secrets=secrets,
+    volumes={CACHE_PATH: cache_vol},  # Mount volume into /cache
+    gpu=GPU,
+    timeout=1800,  # 30-minute max runtime
+    min_containers=0,  # 1 = always-on, uses credits
+    scaledown_window=180,  # Shuts down the container
+)

src/modal_services/e5_model_base.py

@@ -0,0 +1,37 @@
+"""Base class for E5 model handling.
+
+Downloads, caches, and loads the model for reuse.
+"""
+
+import logging
+import os
+
+# Import CACHE_PATH from your config
+from src.modal_services.app_config import CACHE_PATH
+
+# Define the model directory using the imported CACHE_PATH
+E5_MODEL_DIR = f"{CACHE_PATH}/e5_model"
+
+
+class E5ModelBase:
+    """Base class for downloading and loading the E5 model."""
+
+    def setup_e5_model(self) -> None:
+        """Downloads and loads the E5 embedding model."""
+        try:
+            # Lazy imports to avoid issues in Docker
+            from huggingface_hub import snapshot_download
+            from sentence_transformers import SentenceTransformer
+
+            # Cache E5 embedding model into /cache/e5_model
+            os.makedirs(E5_MODEL_DIR, exist_ok=True)
+            if not os.listdir(E5_MODEL_DIR):
+                snapshot_download("intfloat/e5-small-v2", local_dir=E5_MODEL_DIR)
+                logging.info("E5 model downloaded.")
+
+            self.vectorizer = SentenceTransformer(E5_MODEL_DIR, device="cuda")
+            logging.info("E5 model loaded on GPU.")
+
+        except Exception as e:
+            logging.error(f"[E5ModelBase] Failed to setup E5 model: {e}")
+            raise RuntimeError("[E5ModelBase] E5 model setup failed.") from e

src/modal_services/ensemble_pricer.py

@@ -0,0 +1,77 @@
+"""Remote Modal service for price prediction.
+
+Combines outputs from multiple agents using an ensemble model.
+"""
+
+# Standard library imports
+import logging
+
+import modal
+
+# Third-party imports
+# Local imports
+from src.modal_services.app_config import (
+    CACHE_PATH,
+    app,
+    modal_class_kwargs,
+)
+
+# Configure logging after all imports
+logging.basicConfig(level=logging.INFO)
+
+REPO_ID = "lisekarimi/smart-deal-finder-models"
+
+# Local paths inside Modal volume
+ENSEMBLE_MODEL_DIR = f"{CACHE_PATH}/ensemble_model"
+ENSEMBLE_MODEL_FILENAME = "ensemble_model.pkl"
+
+
+@app.cls(**modal_class_kwargs)
+class EnsemblePricer:
+    """Modal class for ensemble price prediction from agent outputs."""
+
+    @modal.enter()
+    def setup(self) -> None:
+        """Loads ensemble model from Hugging Face into Modal cache."""
+        try:
+            # Lazy load hf_hub_download and joblib
+            import joblib
+            from huggingface_hub import hf_hub_download
+
+            logging.info("Downloading Ensemble model...")
+            model_path = hf_hub_download(
+                repo_id=REPO_ID,
+                filename=ENSEMBLE_MODEL_FILENAME,
+                cache_dir=ENSEMBLE_MODEL_DIR,
+            )
+
+            logging.info("Ensemble model downloaded.")
+            self.model = joblib.load(model_path)
+            logging.info("Ensemble model loaded successfully.")
+
+        except Exception as e:
+            logging.error(f"[EnsemblePricer] Failed during setup: {e}")
+            raise RuntimeError("[EnsemblePricer] Setup failed.") from e
+
+    @modal.method()
+    def price(self, ft: float, rag: float, xgb: float) -> float:
+        """Predicts final price using ensemble of 3 models."""
+        try:
+            # Lazy load pandas and numpy for feature creation
+            import numpy as np
+            import pandas as pd
+
+            features = pd.DataFrame(
+                {
+                    "FT_LLaMA": [ft],
+                    "GPT4oMini": [rag],
+                    "XGBoost": [xgb],
+                    "Max": [max(ft, rag, xgb)],
+                    "Mean": [np.mean([ft, rag, xgb])],
+                }
+            )
+            prediction = self.model.predict(features)[0]
+            return round(float(prediction), 2)
+        except Exception as e:
+            logging.error(f"[EnsemblePricer] Prediction failed: {e}")
+            return 0.0

src/modal_services/entry.py

@@ -0,0 +1,24 @@
+"""Entry point for creating Modal classes."""
+
+import os
+
+import modal
+from dotenv import load_dotenv
+
+from src.modal_services.app_config import app
+from src.modal_services.ensemble_pricer import EnsemblePricer
+from src.modal_services.ft_pricer import FTPricer
+from src.modal_services.rag_pricer import RAGPricer
+from src.modal_services.xgb_pricer import XGBPricer
+
+# Load environment variables after imports
+load_dotenv()
+
+MODAL_TOKEN_ID = os.getenv("MODAL_TOKEN_ID")
+MODAL_TOKEN_SECRET = os.getenv("MODAL_TOKEN_SECRET")
+
+if not MODAL_TOKEN_ID or not MODAL_TOKEN_SECRET:
+    raise ValueError("❌ Missing Modal tokens!")
+
+# These imports are required for Modal class registration
+__all__ = ["FTPricer", "XGBPricer", "RAGPricer", "EnsemblePricer", "app", "modal"]

src/modal_services/ft_pricer.py

@@ -0,0 +1,126 @@
+"""Defines Pricer agent using fine-tuned LLaMA on Modal."""
+
+import logging
+import os
+from typing import Any
+
+import modal
+
+from src.modal_services.app_config import CACHE_PATH, app, modal_class_kwargs
+from src.utils.text_utils import extract_tagged_price
+
+logging.basicConfig(level=logging.INFO)
+
+# Model identifiers
+BASE_MODEL = "meta-llama/Meta-Llama-3.1-8B"
+FINETUNED_MODEL = "ed-donner/pricer-2024-09-13_13.04.39"
+REVISION = "e8d637df551603dc86cd7a1598a8f44af4d7ae36"
+
+# Local model paths in volume
+BASE_MODEL_DIR = f"{CACHE_PATH}/llama_base_model"
+FINETUNED_MODEL_DIR = f"{CACHE_PATH}/llama_finetuned_model"
+
+QUESTION = "How much does this cost to the nearest dollar?"
+PREFIX = "Price is $"
+
+
+@app.cls(**modal_class_kwargs)
+class FTPricer:
+    """Remote pricing with LLaMA, PEFT, and 4-bit quantization."""
+
+    @staticmethod
+    def _build_prompt(description: str) -> str:
+        return f"{QUESTION}\n\n{description}\n\n{PREFIX}"
+
+    @staticmethod
+    def _generate_output(
+        model: Any,  # noqa: ANN401
+        inputs: dict,
+        tokenizer: Any,  # noqa: ANN401
+    ) -> str:
+        """Generate output from model."""
+        import torch
+
+        with torch.no_grad():
+            outputs = model.generate(**inputs, max_new_tokens=5, num_return_sequences=1)
+        return tokenizer.decode(outputs[0])
+
+    @staticmethod
+    def _download_models() -> None:
+        from huggingface_hub import snapshot_download
+
+        snapshot_download(BASE_MODEL, local_dir=BASE_MODEL_DIR)
+        snapshot_download(
+            FINETUNED_MODEL, revision=REVISION, local_dir=FINETUNED_MODEL_DIR
+        )
+
+    def _load_tokenizer(self) -> None:
+        from transformers import AutoTokenizer
+
+        self.tokenizer = AutoTokenizer.from_pretrained(BASE_MODEL_DIR)
+        self.tokenizer.pad_token = self.tokenizer.eos_token
+        self.tokenizer.padding_side = "right"
+        logging.info("Tokenizer loaded.")
+
+    def _load_models(self) -> None:
+        import torch
+        from peft import PeftModel
+        from transformers import AutoModelForCausalLM, BitsAndBytesConfig
+
+        base_model = AutoModelForCausalLM.from_pretrained(
+            BASE_MODEL_DIR,
+            quantization_config=BitsAndBytesConfig(
+                load_in_4bit=True,
+                bnb_4bit_use_double_quant=True,
+                bnb_4bit_compute_dtype=torch.bfloat16,
+                bnb_4bit_quant_type="nf4",
+            ),
+            device_map="auto",
+        )
+
+        self.fine_tuned_model = PeftModel.from_pretrained(
+            base_model, FINETUNED_MODEL_DIR, revision=REVISION
+        )
+        self.fine_tuned_model.eval()
+        gen_config = self.fine_tuned_model.generation_config
+        gen_config.pad_token_id = self.tokenizer.pad_token_id
+        gen_config.eos_token_id = self.tokenizer.eos_token_id
+        logging.info("Models loaded.")
+
+    @modal.enter()
+    def setup(self) -> None:
+        """Load base and fine-tuned models with tokenizer and quantization."""
+        try:
+            os.makedirs(CACHE_PATH, exist_ok=True)
+            self._download_models()
+            logging.info("Base and fine-tuned models downloaded.")
+            self._load_tokenizer()
+            self._load_models()
+        except Exception as e:
+            logging.error(f"[FTPricer] Setup failed: {e}")
+            raise RuntimeError("[FTPricer] Model setup failed") from e
+
+    @modal.method()
+    def price(self, description: str) -> float:
+        """Generate a price estimate based on a product description."""
+        from transformers import set_seed
+
+        try:
+            set_seed(42)
+            logging.info("[FTPricer] Generating price...")
+
+            prompt = self._build_prompt(description)
+            inputs = self.tokenizer(prompt, return_tensors="pt", padding=True).to(
+                "cuda"
+            )
+            result = self._generate_output(
+                self.fine_tuned_model, inputs, self.tokenizer
+            )
+            price = extract_tagged_price(result)
+
+            logging.info(f"[FTPricer] Predicted price: {price}")
+            return price
+
+        except Exception as e:
+            logging.error(f"[FTPricer] Prediction failed: {e}")
+            return 0.0

src/modal_services/rag_pricer.py

@@ -0,0 +1,141 @@
+"""Predicts item prices using RAG (Retrieval Augmented Generation).
+
+With ChromaDB, E5 embeddings, and GPT-4o-mini.
+"""
+
+# Standard library imports
+import logging
+import os
+import zipfile
+
+import modal
+
+# Third-party imports
+import numpy as np
+import requests
+
+# Local imports
+from src.modal_services.app_config import CACHE_PATH, app, modal_class_kwargs
+from src.modal_services.e5_model_base import E5ModelBase
+from src.models.frontier_model import OPENAI_MODEL
+from src.utils.text_utils import extract_price
+
+# Configure logging after all imports
+logging.basicConfig(level=logging.INFO)
+
+# Paths
+E5_MODEL_DIR = f"{CACHE_PATH}/e5_model"
+CHROMA_DIR = f"{CACHE_PATH}/chroma"
+CHROMA_ZIP_URL = "https://aiprojects-lise-karimi.s3.eu-west-3.amazonaws.com/smart-deal-finder/chroma.zip"
+COLLECTION_NAME = "price_items"
+
+
+@app.cls(**modal_class_kwargs)
+class RAGPricer(E5ModelBase):
+    """Remote class for pricing products using RAG pipeline."""
+
+    @modal.enter()
+    def setup(self) -> None:
+        """Load E5 embedding model, ChromaDB and OpenAI client."""
+        try:
+            # Lazy load the required modules
+            import chromadb
+
+            # Setup E5 model using the base class method
+            self.setup_e5_model()
+
+            # ChromaDB setup remains the same
+            if not os.path.exists(CHROMA_DIR):
+                os.makedirs(CHROMA_DIR, exist_ok=True)
+                r = requests.get(CHROMA_ZIP_URL)
+                with open("/tmp/chroma.zip", "wb") as f:
+                    f.write(r.content)
+                with zipfile.ZipFile("/tmp/chroma.zip", "r") as zip_ref:
+                    zip_ref.extractall(CHROMA_DIR)
+            logging.info("ChromaDB ready.")
+
+            self.chroma_client = chromadb.PersistentClient(path=CHROMA_DIR)
+            self.collection = self.chroma_client.get_collection(name=COLLECTION_NAME)
+            logging.info("ChromaDB client ready.")
+
+        except Exception as e:
+            logging.error(f"[RAGPricer] Failed during setup: {e}")
+            raise RuntimeError("[RAGPricer] Setup failed.") from e
+
+    def _get_embedding(self, item: str) -> np.ndarray:
+        """Encodes the item description into embeddings using the E5 model."""
+        return self.vectorizer.encode(["passage: " + item], normalize_embeddings=True)
+
+    def _find_similar_items(self, item: str) -> tuple[list[str], list[float]]:
+        """Finds similar items from ChromaDB based on embeddings."""
+        query_emb = self._get_embedding(item).astype(float).tolist()
+        results = self.collection.query(query_embeddings=query_emb, n_results=5)
+        documents = results["documents"][0][:]
+        prices = [m["price"] for m in results["metadatas"][0][:]]
+
+        # Log similar items and their prices
+        for doc, price in zip(documents, prices):
+            logging.info(f"[RAGPricer] Similar item: '{doc}' | Price: ${price:.2f}")
+
+        return documents, prices
+
+    def _format_context(self, similars: list[str], prices: list[float]) -> str:
+        """Formats the context for the RAG pipeline."""
+        message = "To provide some context, here are some other items "
+        message += "that might be similar to the item you need to estimate.\n\n"
+
+        for similar, price in zip(similars, prices):
+            message += (
+                f"Potentially related product:\n{similar}\nPrice is ${price:.2f}\n\n"
+            )
+
+        return message
+
+    def _build_messages(
+        self, item: dict, similars: list[str], prices: list[float]
+    ) -> list[dict[str, str]]:
+        """Builds messages for the GPT-4o-mini model to predict the price."""
+        system_message = (
+            "You are a pricing expert. "
+            "Given a product description and a few similar products with their prices, "
+            "you must estimate the most likely price for the given product. "
+            "Always respond ONLY with a number, no words or explanation."
+        )
+        context = self._format_context(similars, prices)
+        user_prompt = (
+            "Estimate the price for the following product:\n\n"
+            + item["description"]
+            + "\n\n"
+            + context
+        )
+
+        return [
+            {"role": "system", "content": system_message},
+            {"role": "user", "content": user_prompt},
+            {"role": "assistant", "content": "Price is $"},
+        ]
+
+    @modal.method()
+    def price(self, description: str) -> float:
+        """Predicts price from description using RAG and Frontier."""
+        try:
+            logging.info("[RAGPricer] Searching similar items...")
+            documents, prices = self._find_similar_items(description)
+            messages = self._build_messages(
+                {"description": description}, documents, prices
+            )
+
+            # Lazy import OpenAI API
+            import openai
+
+            response = openai.chat.completions.create(
+                model=OPENAI_MODEL, messages=messages, seed=42, max_tokens=5
+            )
+            reply = response.choices[0].message.content
+            price = extract_price(reply)
+
+            logging.info(f"[RAGPricer] Predicted price: {price}")
+            return price
+        except Exception as e:
+            logging.error(f"[RAGPricer] Failed to predict price: {e}")
+            return 0.0

src/modal_services/xgb_pricer.py

@@ -0,0 +1,61 @@
+"""Predicts prices using XGBoost and E5 embeddings."""
+
+import logging
+
+import modal
+
+from src.modal_services.app_config import (
+    CACHE_PATH,
+    app,
+    modal_class_kwargs,
+)
+from src.modal_services.e5_model_base import E5ModelBase
+
+REPO_ID = "lisekarimi/smart-deal-finder-models"
+
+# Local paths inside Modal volume
+E5_MODEL_DIR = f"{CACHE_PATH}/e5_model"
+XGB_MODEL_DIR = f"{CACHE_PATH}/xgb_model"
+XGB_MODEL_FILENAME = "xgboost_model.pkl"
+
+
+@app.cls(**modal_class_kwargs)
+class XGBPricer(E5ModelBase):
+    """Remote pricing via E5 and XGBoost."""
+
+    @modal.enter()
+    def setup(self) -> None:
+        """Loads E5 and XGBoost into Modal cache."""
+        try:
+            # Setup E5 model using the base class method
+            self.setup_e5_model()
+
+            # Lazy load XGBoost model and download it inside the setup method
+            import joblib
+            from huggingface_hub import hf_hub_download
+
+            logging.info("Downloading XGBoost model...")
+            model_path = hf_hub_download(
+                repo_id=REPO_ID, filename=XGB_MODEL_FILENAME, cache_dir=XGB_MODEL_DIR
+            )
+
+            logging.info("XGBoost model downloaded.")
+            self.model = joblib.load(model_path)
+            logging.info("XGBoost model loaded.")
+
+        except Exception as e:
+            logging.error(f"[XGBPricer] Failed during setup: {e}")
+            raise RuntimeError("[XGBPricer] Setup failed.") from e
+
+    @modal.method()
+    def price(self, description: str) -> float:
+        """Predict price from product description using E5 + XGBoost."""
+        try:
+            logging.info("[XGBPricer] Encoding description...")
+            vector = self.vectorizer.encode(["passage: " + description])
+            pred = self.model.predict(vector)[0]
+            logging.info(f"[XGBPricer] Predicted price: {pred}")
+            return round(float(max(0, pred)), 2)
+        except Exception as e:
+            logging.error(f"[XGBPricer] Failed to predict price: {e}")
+            return 0.0

src/models/frontier_model.py

@@ -0,0 +1,16 @@
+"""Initializes access to frontier AI models using environment variables."""
+
+import os
+
+from dotenv import load_dotenv
+from openai import OpenAI
+
+load_dotenv(override=True)
+
+OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
+
+if not OPENAI_API_KEY:
+    raise ValueError("❌ OpenAI API Key is missing!")
+
+openai = OpenAI(api_key=OPENAI_API_KEY)
+OPENAI_MODEL = "gpt-4o-mini"

src/ui/assets/styles.css

@@ -0,0 +1,242 @@
+/* ==== Global Reset & Layout ==== */
+html, body, #app, body > div, .gradio-container {
+    background-color: #1f2937 !important;
+    margin: 0;
+    padding: 0;
+    font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
+    display: flex;
+    justify-content: center;
+}
+
+/* ==== Main Container ==== */
+#app-container {
+    background-color: #0e2d53 !important;
+    margin: 0 auto;
+    padding: 20px;
+    border-radius: 50px;
+    box-shadow: 0 4px 30px 5px rgba(0, 0, 0, 0.7);
+    border: 1px solid #1f2f46;
+    max-width: 1000px;
+    width: 100%;
+    color: white;
+    margin-bottom: 10px !important;
+}
+
+#app-container > *:last-child {
+    margin-bottom: 0 !important;
+}
+
+#app-container h4,
+#app-container p,
+#app-container ol,
+#app-container li,
+#app-container strong {
+    font-size: 16px;
+    line-height: 1.6;
+    color: white !important; 
+}
+
+.gradio-container {
+    margin-bottom: 0 !important;
+    padding-bottom: 0 !important;
+}
+
+.version-banner {
+    margin-bottom: 0 !important;
+    padding-bottom: 0 !important;
+    margin-top: 10px;
+}
+
+/* ==== Titles ==== */
+#app-title {
+    font-size: 40px;
+    font-weight: 800;
+    text-align: center;
+    margin-bottom: 6px;
+    background: linear-gradient(135deg, #FFA500, #FF4500);
+    -webkit-background-clip: text;
+    background-clip: text;
+    color: transparent;
+}
+
+#app-title::before {
+    filter: brightness(0.95);
+}
+
+#app-subtitle {
+    font-size: 24px;
+    font-weight: 600;
+    text-align: center;
+    margin-top: 0;
+    background: linear-gradient(135deg, #FFA500, #FF4500);
+    -webkit-background-clip: text;
+    background-clip: text;
+    color: transparent;
+}
+
+/* ==== Intro Text ==== */
+#intro-text {
+    font-size: 16px;
+    color: white !important; 
+    margin-top: 20px;
+    line-height: 1.6;
+}
+
+h4 ~ p {
+    padding-left: 20px;
+}
+ 
+.custom-links a {
+    color: #ffa500;
+    text-decoration: none;
+}
+
+.custom-links a:hover {
+    text-decoration: underline;
+}
+
+/* ==== Category Selector ==== */
+#category-selector span {
+    font-size: 16px;
+    color: #ffa500;
+    display: flex;
+    align-items: center;
+    gap: 6px;
+  }
+
+#category-selector {
+    background-color: #111827;
+    border-radius: 8px;
+    padding: 10px;
+  }
+#category-selector .wrap {
+    background-color: #111827;
+    border: 1px solid #26313f;
+}
+
+.token {
+    background-color: #3f3f46 !important;
+}
+
+.wrap .options li.item {
+    background-color: #1f2937;
+    color: white;
+    padding: 8px;
+}
+
+.wrap .options li.item:hover {
+    background-color: #4b5563;
+}
+
+/* ==== Buttons ==== */
+#run-btn {
+    background: linear-gradient(135deg, #FFA500, #FF4500);
+    color: #0a0f1a !important;
+    font-weight: bold;
+    border: none;
+    padding: 10px 20px;
+    border-radius: 8px;
+    cursor: pointer;
+    transition: filter 0.3s ease;
+}
+
+#run-btn:hover {
+    filter: brightness(1.1);
+}
+
+.btn-disabled {
+    opacity: 0.5;
+    pointer-events: none;
+    filter: grayscale(1);
+}
+
+/* ==== Status Message / Demo Notice ==== */
+.html-container .prose {
+    font-size: 16px !important;
+    color: #e2e8f0;
+    font-weight: 500;
+}
+
+/* ==== Logs ==== */
+#logs-label {
+    font-weight: bold;
+    color: #ffa500;
+    font-size: 16px;
+}
+
+#scrollContent {
+    height: 400px;
+    overflow-y: auto;
+    display: flex;
+    flex-direction: column-reverse;
+    border: 1px solid #444;
+    background-color: #111827;
+    font-family: monospace;
+    font-size: 16px;
+    color: #fefefe;
+    padding: 10px;
+    scroll-behavior: smooth;
+}
+
+.log-entry {
+    line-height: 1.5;       
+    margin-bottom: 2px;  
+    white-space: pre-wrap;
+}
+
+
+/* ==== Table ==== */
+#deals-label {
+    font-weight: bold;
+    color: #ffa500;
+    margin-bottom: 6px;
+    font-size: 16px;
+}
+
+#deal-table {
+    overflow-x: auto;
+}
+
+#deal-table table {
+    width: 100% !important;
+    box-sizing: border-box;
+    border-collapse: collapse;
+    table-layout: fixed;
+    min-width: 700px;
+}
+
+#deal-table table th,
+#deal-table table td {
+    padding: 10px;
+    text-align: left;
+    overflow-wrap: break-word;
+    word-wrap: break-word;
+    word-break: break-word;
+}
+
+#deal-table table th {
+    background-color: #1f2937 !important;
+    color: #f4a261 !important;
+    font-weight: bold;
+    border: 1px solid #334155;
+}
+
+#deal-table table td {
+    background-color: #1f2937 !important;
+    color: white !important;
+    border: 1px solid #334155;
+}
+
+#deal-table table th:first-child,
+#deal-table table td:first-child {
+    width: 40% !important;
+}
+
+/* ==== Version ==== */
+.version-banner {
+    text-align: center;
+    font-size: 0.9em;
+}
+
+
+

src/ui/formatting.py

@@ -0,0 +1,66 @@
+"""Formatting utilities for the Gradio UI."""
+
+from typing import List
+
+from src.config.constants import MAX_LOG_LINES
+
+
+def html_for(log_data: list[str]) -> str:
+    """Generate HTML for displaying the log lines in a scrollable container."""
+    logs = reversed(log_data[-MAX_LOG_LINES:])
+    output = "".join(f"<div class='log-entry'>{line}</div>" for line in logs)
+    return (
+        "<div id='logs-label'>📜 Live Agent Logs</div>"
+        "<div id='scrollContent'>" + output + "</div>"
+    )
+
+
+def format_deals_table(deals: List[List[str]]) -> str:
+    """Formats accepted deals as an HTML table with styled links."""
+    html = """
+    <div id="deal-table">
+        <div id="deals-label">🛍️ Best Deals Found</div>
+        <table>
+            <thead>
+                <tr>
+                    <th>Description</th>
+                    <th>Price</th>
+                    <th>AI Estimate</th>
+                    <th>Discount</th>
+                    <th>URL</th>
+                </tr>
+            </thead>
+            <tbody>
+    """
+    for desc, price, estimate, discount, url in deals:
+        html += f"""
+                <tr>
+                    <td>{desc}</td>
+                    <td>{price}</td>
+                    <td>{estimate}</td>
+                    <td>{discount}</td>
+                    <td><a href="{url}" target="_blank">Link</a></td>
+                </tr>
+        """
+    html += """
+            </tbody>
+        </table>
+    </div>
+    """
+    return html
+
+
+def get_server_timezone() -> str:
+    """Get the server's timezone information."""
+    import datetime
+
+    current_time = datetime.datetime.now()
+    try:
+        timezone_name = current_time.astimezone().tzinfo.tzname(current_time)
+    except AttributeError:
+        timezone_name = "Unknown"
+
+    return (
+        f"Server Time: {current_time.strftime('%Y-%m-%d %H:%M:%S')} "
+        f"(Timezone: {timezone_name})"
+    )

src/ui/gradio_app.py

@@ -0,0 +1,170 @@
+"""Launches a Gradio app that runs an AI planning agent to find smart online deals.
+
+Uses Python's built-in logging system for consistent logs.
+Includes demo mode restrictions: MAX_DEMO_RUNS_PER_DAY runs per day.
+"""
+
+import os
+from typing import Tuple
+
+import gradio as gr
+
+from src.agents.pipeline import run_and_stream_logs
+from src.config.constants import (
+    IS_DEMO_VERSION,
+    MAX_CATEGORY_SELECTION,
+    MAX_DEMO_RUNS_PER_DAY,
+    PROJECT_NAME,
+    VERSION,
+)
+from src.config.feeds import CATEGORY_FEEDS
+from src.ui.formatting import format_deals_table
+from src.utils.logger import console
+from src.utils.state_manager import can_run_app
+
+PROJECT_NAME_CAP = PROJECT_NAME.capitalize()
+REPO_URL = f"https://github.com/lisekarimi/{PROJECT_NAME}"
+DOC_URL = f"https://lisekarimi.github.io/{PROJECT_NAME}"
+
+
+def build_ui() -> gr.Blocks:
+    """Constructs and returns the Gradio UI interface, with error handling."""
+    try:
+        with open(
+            os.path.join(os.path.dirname(__file__), "assets", "styles.css"), "r"
+        ) as f:
+            css = f.read()
+    except Exception as e:
+        css = ""
+        console.print(f"[bold yellow]⚠️ Failed to load CSS:[/] {e}")
+
+    # Set initial UI state here
+    disable_btn = False
+    initial_status = "Loading demo status..."
+
+    # Building the UI
+    try:
+        with gr.Blocks(css=css, title=f"🏷️{PROJECT_NAME_CAP}") as ui:
+            with gr.Column(elem_id="app-container"):
+                # 🔝 Top content
+                gr.Markdown(f"<h1 id='app-title'>🏷️ {PROJECT_NAME_CAP} </h1>")
+                gr.Markdown(
+                    "<h2 id='app-subtitle'>Autonomous AI Agents Snapping the Best "
+                    "Deals Online</h2>"
+                )
+                gr.HTML(
+                    f"""
+                    <div id="intro-text">
+                        <p>🏷️ <strong>Snapr</strong>, 
+                        <strong>an Agentic AI System</strong>, 
+                        discovers the best online deals for you—smart, 
+                        simple, and automatic. 
+                        Let <strong>AI</strong> do the work, so you can save money 
+                        without any effort.</p>
+
+
+                        <h4>🤖 How It Works:</h4>
+                            <p>1️⃣ Choose up to {MAX_CATEGORY_SELECTION} 
+                            categories to search.</p>
+                            <p>2️⃣ Click "Find Smart Deals" — AI scans, estimates prices, 
+                            and filters top discounts.</p>
+                            <p>3️⃣ See the best deals in a table with prices, discounts, 
+                            and direct links.</p>
+                    </div>
+                    """
+                )
+                if IS_DEMO_VERSION:
+                    gr.Markdown(
+                        f"""
+                        <p>⚠️ This is a demo version — limited to 
+                        {MAX_DEMO_RUNS_PER_DAY} global runs per day for all users.</p>
+                        """
+                    )
+
+                gr.Markdown(
+                    f"""
+                    <p class="custom-links">📦 Want more?
+                    You can <a href="{REPO_URL}" 
+                    target="_blank">run it locally</a> 
+                    with full <a href="{DOC_URL}/technical/localdev/" 
+                    target="_blank">instructions</a> and
+                    <a href="{DOC_URL}" 
+                    target="_blank">documentation</a>.</p>
+                    """
+                )
+
+                with gr.Column(elem_id="left-col"):
+                    category_selector = gr.Dropdown(
+                        choices=list(CATEGORY_FEEDS.keys()),
+                        value="Electronics",
+                        multiselect=True,
+                        label=(
+                            f"🧭 Select up to {MAX_CATEGORY_SELECTION} Deal Categories"
+                        ),
+                        elem_id="category-selector",
+                        elem_classes="custom-dropdown",
+                    )
+                    run_btn = gr.Button(
+                        "🔍 Find Smart Deals",
+                        elem_id="run-btn",
+                        variant="primary",
+                        elem_classes="run-button",
+                        interactive=not disable_btn,
+                    )
+
+                status_msg = gr.HTML(value=initial_status, elem_id="status-message")
+
+                # Logs + deals
+                logs_output = gr.HTML(
+                    value="""
+                    <div id="logs-label">📜 Live Agent Logs</div>
+                        <div id="scrollContent">
+                            🕵️‍♀️ Click "🔍 Find Smart Deals" to wake the agents and 
+                            stream logs here!
+                        </div> 
+                    """
+                )
+
+                deals_output = gr.HTML(
+                    value=format_deals_table([]), elem_id="deal-table"
+                )
+
+                # Connect button
+                run_btn.click(
+                    fn=run_and_stream_logs,
+                    inputs=[category_selector],
+                    outputs=[logs_output, deals_output, run_btn, status_msg],
+                )
+
+                # Status update on load
+                @ui.load(outputs=[status_msg, run_btn])
+                def update_status_on_load() -> Tuple[str, gr.update]:
+                    """Sets demo status and button state on UI load."""
+                    can_run, status = can_run_app()
+                    btn_state = gr.update(
+                        interactive=can_run,
+                        elem_classes=(
+                            ["run-button", "btn-disabled"]
+                            if not can_run
+                            else ["run-button"]
+                        ),
+                    )
+                    return status, btn_state
+
+            # Bottom: version info
+            gr.Markdown(
+                f"""
+                <p class="version-banner">
+                    🔖 <strong>
+                    <a href="{DOC_URL}/changelog" 
+                    target="_blank">Version {VERSION}</a>
+                    </strong>
+                </p>
+                """
+            )
+
+        return ui
+
+    except Exception as e:
+        console.print(f"[bold red]❌ Failed to build UI:[/] {e}")
+        raise

src/utils/cleanup.py

@@ -0,0 +1,14 @@
+"""Deletes a file if it's older than a specified number of days."""
+
+import os
+import time
+
+from src.config.constants import MEMORY_EXPIRATION_DAYS
+
+
+def delete_if_old(path: str, max_age_days: int = MEMORY_EXPIRATION_DAYS) -> None:
+    """Deletes file if older than max_age_days."""
+    if os.path.exists(path):
+        age = time.time() - os.path.getmtime(path)
+        if age > max_age_days * 86400:
+            os.remove(path)

src/utils/file_io.py

@@ -0,0 +1,48 @@
+"""Utility functions for safely handling JSON file operations."""
+
+import json
+import os
+from typing import Any, Dict
+
+from src.utils.logger import console
+
+
+def write_json(path: str, data: Dict[str, Any]) -> None:
+    """Writes JSON to file, ensuring parent folder exists..
+
+    Args:
+        path (str): Full file path to write to.
+        data (Dict[str, Any]): The data to write as JSON.
+    """
+    try:
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, "w") as f:
+            json.dump(data, f, indent=2)
+    except Exception as e:
+        console.print(f"Error writing to {path}: {e}", style="red")
+
+
+def load_json(path: str) -> Dict[str, Any]:
+    """Safely loads and returns JSON data from the given file path.
+
+    Args:
+        path (str): Full file path to read from.
+
+    Returns:
+        Dict[str, Any]: Parsed JSON content, or an empty dict if loading fails.
+    """
+    try:
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+
+        if not os.path.exists(path):
+            console.print(
+                f"[yellow]File not found at {path}. Assuming first run.[/yellow]"
+            )
+            return {}
+
+        with open(path, "r") as f:
+            return json.load(f)
+
+    except Exception as e:
+        console.print(f"[red]Error reading from {path}: {e}[/red]")
+        return {}

src/utils/logger.py

@@ -0,0 +1,5 @@
+"""Provides a shared Rich console instance for styled logging."""
+
+from rich.console import Console
+
+console = Console()

src/utils/memory_utils.py

@@ -0,0 +1,26 @@
+"""Utilities for handling deal memory and persistence."""
+
+from datetime import datetime, timezone
+from typing import Dict, List
+
+from src.config.constants import DEALS_FILE
+from src.utils.file_io import load_json, write_json
+
+
+def save_opportunities_to_memory(
+    new_opportunities: List[Dict], memory_path: str = DEALS_FILE
+) -> None:
+    """Updates opportunities with deduplication and saves to memory."""
+    existing_data = load_json(memory_path) or {"opportunities": []}
+
+    all_opportunities = existing_data.get("opportunities", []) + new_opportunities
+    unique_by_url = {op["url"]: op for op in all_opportunities}
+    final_list = list(unique_by_url.values())
+
+    write_json(
+        memory_path,
+        {
+            "opportunities": final_list,
+            "last_updated": datetime.now(timezone.utc).isoformat(),
+        },
+    )

src/utils/state_manager.py

@@ -0,0 +1,81 @@
+"""Handles persistent state management and access control for the application."""
+
+import datetime
+import os
+import threading
+from typing import Any, Dict, Tuple
+
+from src.config.constants import IS_DEMO_VERSION, MAX_DEMO_RUNS_PER_DAY, STATE_FILE
+from src.ui.formatting import get_server_timezone
+from src.utils.file_io import load_json, write_json
+from src.utils.logger import console
+
+state_lock = threading.Lock()
+
+
+def get_default_state() -> Dict[str, Any]:
+    """Returns the default initial state dictionary."""
+    return {"date": datetime.datetime.now().strftime("%Y-%m-%d"), "run_count": 0}
+
+
+def get_state() -> Dict[str, Any]:
+    """Get the current state from the JSON file, initializing if needed."""
+    if not os.path.exists(STATE_FILE):
+        default_state = get_default_state()
+        write_json(STATE_FILE, default_state)
+        return default_state
+
+    try:
+        state = load_json(STATE_FILE)
+
+        # Reset counter if it's a new day
+        current_date = datetime.datetime.now().strftime("%Y-%m-%d")
+        if state["date"] != current_date:
+            state = get_default_state()
+            write_json(STATE_FILE, state)
+
+        return state
+
+    except Exception as e:
+        console.print(f"Error reading state file: {e}", style="red")
+        return get_default_state()
+
+
+def update_state(state_updates: Dict[str, Any]) -> Dict[str, Any]:
+    """Update the state file with new values."""
+    current_state = get_state()
+    current_state.update(state_updates)
+    write_json(STATE_FILE, current_state)
+    return current_state
+
+
+def can_run_app() -> Tuple[bool, str]:
+    """Check if the app can be run based on demo restrictions."""
+    # No restrictions if not running in demo mode
+    if not IS_DEMO_VERSION:
+        return True, ""
+
+    # Get server timezone for both cases
+    server_time = get_server_timezone()
+    server_time = f"🌐 {server_time}"
+
+    # Load current run state
+    state = get_state()
+    runs_left = MAX_DEMO_RUNS_PER_DAY - state["run_count"]
+
+    # Block if daily limit reached
+    if state["run_count"] >= MAX_DEMO_RUNS_PER_DAY:
+        return (
+            False,
+            f"⛔ Daily limit reached ({MAX_DEMO_RUNS_PER_DAY} runs per day "
+            f"in demo mode). Please try again tomorrow!"
+            f"<br>{server_time}",
+        )
+
+    # Otherwise, allow and return runs remaining
+    return (
+        True,
+        f"🕒 Demo mode: {runs_left} run"
+        f"{'s' if runs_left != 1 else ''} left today."
+        f"<br>{server_time}",
+    )

src/utils/text_utils.py

@@ -0,0 +1,29 @@
+"""Utility functions for text processing."""
+
+import re
+
+
+def extract_tagged_price(output: str) -> float:
+    """Extracts a float price from a string based on 'Price is $' keyword."""
+    try:
+        contents = output.split("Price is $")[1].replace(",", "")
+        match = re.search(r"[-+]?\d*\.\d+|\d+", contents)
+        return float(match.group()) if match else 0.0
+    except Exception:
+        return 0.0
+
+
+def extract_price(output: str) -> float:
+    """Extracts a float price from a string.
+
+    If no number is found, returns 0.0.
+    """
+    try:
+        cleaned_text = output.replace("$", "").replace(",", "")
+        match = re.search(r"[-+]?\d*\.\d+|\d+", cleaned_text)
+        if match:
+            return round(float(match.group()), 2)
+        return 0.0
+    except Exception:
+        # Optionally log the exception or handle differently
+        return 0.0