Spaces:
Sleeping
Sleeping
File size: 5,604 Bytes
0310410 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 | """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}"
|