apps/copilot/tools.py

"""Standalone financial data-fetching functions (framework-agnostic).

Used by the Claude Agent SDK MCP tools and standalone context enrichment.
No dependency on LangChain or any agent framework.
"""

import requests as http_requests

_TV_NEWS_URL = "https://news-mediator.tradingview.com/public/news-flow/v2/news"
_TV_HEADERS = {
    "accept": "*/*",
    "origin": "https://in.tradingview.com",
    "referer": "https://in.tradingview.com/",
    "user-agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/145.0.0.0 Safari/537.36"
    ),
}
_TV_SCREENER_URL = (
    "https://screener-facade.tradingview.com/screener-facade/api/v1/screener-table/scan"
)
_TV_SCREENER_HEADERS = {
    "accept": "application/json",
    "content-type": "text/plain;charset=UTF-8",
    "origin": "https://in.tradingview.com",
    "referer": "https://in.tradingview.com/",
    "user-agent": _TV_HEADERS["user-agent"],
}
_COMMON_EXCHANGES = ["NYSE", "NASDAQ", "AMEX"]


def fetch_stock_news_data(symbol: str) -> str:
    """Fetch latest news for a stock ticker from TradingView."""
    all_items: list[dict] = []
    seen: set[str] = set()
    for exchange in _COMMON_EXCHANGES:
        params = {
            "filter": ["lang:en", f"symbol:{exchange}:{symbol.upper()}"],
            "client": "landing",
            "streaming": "false",
            "user_prostatus": "non_pro",
        }
        try:
            resp = http_requests.get(
                _TV_NEWS_URL, params=params, headers=_TV_HEADERS, timeout=10,
            )
            resp.raise_for_status()
            for item in resp.json().get("items", []):
                iid = str(item.get("id", ""))
                if iid and iid not in seen:
                    seen.add(iid)
                    all_items.append(item)
        except Exception:
            continue

    all_items.sort(key=lambda x: x.get("published", 0), reverse=True)
    top = all_items[:8]
    if not top:
        return f"No recent news found for {symbol}."

    lines = [f"Latest news for {symbol} ({len(top)} items):"]
    for idx, item in enumerate(top, 1):
        provider = ""
        prov = item.get("provider")
        if isinstance(prov, dict):
            provider = prov.get("name", "")
        title = item.get("title", "Untitled")
        lines.append(f"{idx}. [{provider}] {title}")
    return "\n".join(lines)


def fetch_market_benchmarks_data() -> str:
    """Fetch US market index benchmarks with technical signals and RSI."""
    params = {
        "table_id": "indices_quotes.us",
        "version": "52",
        "columnset_id": "technicals",
    }
    body = '{"lang":"en","range":[0,5],"scanner_product_label":"markets-screener"}'
    try:
        resp = http_requests.post(
            _TV_SCREENER_URL,
            params=params,
            headers=_TV_SCREENER_HEADERS,
            data=body,
            timeout=10,
        )
        resp.raise_for_status()
        raw = resp.json()
    except Exception as exc:
        return f"Failed to fetch benchmarks: {exc}"

    symbols = raw.get("symbols", [])
    columns = raw.get("data", [])
    count = len(symbols)
    col_map: dict[str, dict] = {}
    for col in columns:
        col_map.setdefault(col["id"], col)

    lines = ["US Market Benchmarks:"]
    for i in range(count):
        ticker_col = col_map.get("TickerUniversal", {})
        raw_vals = ticker_col.get("rawValues", [])
        info = raw_vals[i] if i < len(raw_vals) else {}
        tech = col_map.get("TechnicalRating", {}).get("rawValues", [])
        rsi = col_map.get("RelativeStrengthIndex", {}).get("rawValues", [])
        name = info.get("description", symbols[i] if i < len(symbols) else "")
        tech_val = tech[i] if i < len(tech) else "N/A"
        rsi_val = (
            round(rsi[i], 2)
            if i < len(rsi) and isinstance(rsi[i], (int, float))
            else "N/A"
        )
        lines.append(f"- {name}: Signal={tech_val}, RSI={rsi_val}")
    return "\n".join(lines)


def fetch_breaking_news_data() -> str:
    """Fetch breaking financial and market news from TradingView."""
    params = {
        "filter": ["lang:en_IN", "priority:important"],
        "client": "screener",
        "streaming": "true",
        "user_prostatus": "non_pro",
    }
    try:
        resp = http_requests.get(
            _TV_NEWS_URL, params=params, headers=_TV_HEADERS, timeout=10,
        )
        resp.raise_for_status()
        items = resp.json().get("items", [])
    except Exception as exc:
        return f"Failed to fetch breaking news: {exc}"

    top = items[:8]
    if not top:
        return "No breaking news available right now."

    lines = [f"Breaking market news ({len(top)} headlines):"]
    for idx, item in enumerate(top, 1):
        title = item.get("title", "Untitled")
        provider = ""
        prov = item.get("provider")
        if isinstance(prov, dict):
            provider = prov.get("name", "")
        lines.append(f"{idx}. [{provider}] {title}")
    return "\n".join(lines)


def calculate_expression(expression: str) -> str:
    """Evaluate a mathematical expression safely."""
    allowed = {"abs": abs, "round": round, "min": min, "max": max, "sum": sum}
    try:
        result = eval(expression, {"__builtins__": {}}, allowed)  # noqa: S307
        return f"Result: {result}"
    except Exception as exc:
        return f"Calculation error: {exc}"