trading-agent/tools/web.py

"""Our own web tools — no search API key.

web_search hits DuckDuckGo's keyless HTML endpoint and parses results ourselves (lxml,
which ships with trafilatura). Falls back to the `ddgs` library if the HTML layout shifts
or we get rate-limited. web_fetch pulls a page and extracts readable text with trafilatura.
"""

from __future__ import annotations

from urllib.parse import parse_qs, unquote, urlparse

import httpx
import trafilatura
from langchain_core.tools import tool
from lxml import html as lxml_html
from tenacity import retry, stop_after_attempt, wait_exponential

_UA = (
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/125.0 Safari/537.36"
)
_MAX_RESULTS = 6
_FETCH_CHAR_LIMIT = 12000  # ~3k tokens


def _clean_url(href: str) -> str:
    """DDG HTML wraps results in a /l/?uddg=<encoded> redirect — unwrap to the real URL."""
    if href.startswith("//"):
        href = "https:" + href
    if "duckduckgo.com/l/" in href or href.startswith("/l/"):
        qs = parse_qs(urlparse(href).query)
        if "uddg" in qs:
            return unquote(qs["uddg"][0])
    return href


@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, max=8), reraise=True)
def _ddg_html_search(query: str, max_results: int = _MAX_RESULTS) -> list[dict]:
    resp = httpx.post(
        "https://html.duckduckgo.com/html/",
        data={"q": query},
        headers={"User-Agent": _UA},
        timeout=15,
        follow_redirects=True,
    )
    resp.raise_for_status()
    doc = lxml_html.fromstring(resp.text)
    anchors = doc.xpath('//a[contains(@class, "result__a")]')
    snippets = doc.xpath('//a[contains(@class, "result__snippet")]')
    results = []
    for i, a in enumerate(anchors[:max_results]):
        title = a.text_content().strip()
        url = _clean_url(a.get("href", ""))
        snippet = snippets[i].text_content().strip() if i < len(snippets) else ""
        if title and url:
            results.append({"title": title, "url": url, "snippet": snippet})
    return results


def _ddgs_search(query: str, max_results: int = _MAX_RESULTS) -> list[dict]:
    from ddgs import DDGS

    with DDGS() as d:
        return [
            {"title": r.get("title", ""), "url": r.get("href", ""), "snippet": r.get("body", "")}
            for r in d.text(query, max_results=max_results)
        ]


@tool
def web_search(query: str) -> str:
    """Search the web; returns the top results as title + URL + snippet.

    Use for news, analyst opinions, macro context, catalysts — anything the structured
    market-data tools don't cover. Follow up with web_fetch to read a promising result.
    """
    results: list[dict] = []
    try:
        results = _ddg_html_search(query)
    except Exception:
        pass
    if not results:
        try:
            results = _ddgs_search(query)
        except Exception as e:
            return f"ERROR: web_search failed (both backends): {e}"
    if not results:
        return f"No results for {query!r}."
    return "\n".join(
        f"{i}. {r['title']}\n   {r['url']}\n   {r['snippet']}" for i, r in enumerate(results, 1)
    )


@retry(stop=stop_after_attempt(2), wait=wait_exponential(multiplier=1, max=6), reraise=True)
def _http_get(url: str) -> str:
    resp = httpx.get(url, headers={"User-Agent": _UA}, timeout=20, follow_redirects=True)
    resp.raise_for_status()
    return resp.text


@tool
def web_fetch(url: str) -> str:
    """Fetch a URL and return its main readable text (truncated).

    Use after web_search to read an article or page in depth. Strips nav/ads/boilerplate.
    """
    try:
        html_text = _http_get(url)
    except Exception as e:
        return f"ERROR: fetch failed: {e}"
    text = trafilatura.extract(html_text, include_comments=False, include_tables=False)
    if not text:
        return "ERROR: no readable content extracted from that URL."
    return text[:_FETCH_CHAR_LIMIT]