from __future__ import annotations import argparse import math import os import time from typing import Any, cast from dotenv import load_dotenv load_dotenv() import requests from mcp.server.fastmcp import FastMCP from tradingview_screener.column import col from tradingview_screener.query import And, Or, Query from tradingview_screener.screeners import ( bond, cfd, coin, crypto, crypto_dex, forex, futures, options, stocks, ) mcp = FastMCP("tradingview-screener") _session_cookies: dict[str, str] | None = None def _resolve_cookies(sessionid: str | None = None) -> dict[str, str] | None: if sessionid: return {"sessionid": sessionid} if _session_cookies is not None: return _session_cookies env = os.environ.get("TV_SESSION_ID") if env: return {"sessionid": env} return None def _exec_query( q: Query, sessionid: str | None = None, ) -> tuple[int, list[dict[str, Any]]]: cookies = _resolve_cookies(sessionid) total, df = q.get_scanner_data(cookies=cookies) return total, cast("list[dict[str, Any]]", df.to_dict(orient="records")) def _build_condition(f: dict[str, Any]) -> Any: if "field" in f: c = col(f["field"]) op = f["operator"] if op in ("empty", "not_empty"): return c.empty() if op == "empty" else c.not_empty() val = f["value"] match op: case ">" | "greater": return c > val case ">=" | "egreater": return c >= val case "<" | "less": return c < val case "<=" | "eless": return c <= val case "==" | "equal": return c == val case "!=" | "nequal": return c != val case "between": return c.between(val[0], val[1]) case "not_between": return c.not_between(val[0], val[1]) case "isin": return c.isin(val) case "not_in": return c.not_in(val) case "has": return c.has(val) case "has_none_of": return c.has_none_of(val) case "like": return c.like(val) case "not_like": return c.not_like(val) case "crosses": return c.crosses(val) case "crosses_above": return c.crosses_above(val) case "crosses_below": return c.crosses_below(val) case "above_pct": return c.above_pct(val[0], val[1]) case "below_pct": return c.below_pct(val[0], val[1]) case "between_pct": return c.between_pct(val[0], val[1], val[2]) case "in_day_range": return c.in_day_range(val[0], val[1]) case "in_week_range": return c.in_week_range(val[0], val[1]) case "in_month_range": return c.in_month_range(val[0], val[1]) if "operator" in f and "filters" in f: sub = [_build_condition(sf) for sf in f["filters"]] if f["operator"] == "or": return Or(*sub) return And(*sub) msg = f"Invalid filter: {f}" raise ValueError(msg) def _apply_filters(q: Query, filters: list[dict[str, Any]] | None) -> Query: if not filters: return q simple: list[Any] = [] nested: list[Any] = [] for f in filters: if "operator" in f and "filters" in f: nested.append(_build_condition(f)) else: simple.append(_build_condition(f)) if simple: q = q.where(*simple) if nested: default_filter2 = q.query.get("filter2") user_op = nested[0] if len(nested) == 1 else And(*nested) user_filter2 = user_op["operation"] if default_filter2: q.query["filter2"] = { "operator": "and", "operands": [{"operation": user_filter2}, {"operation": default_filter2}], } else: q.query["filter2"] = user_filter2 return q MARKET_FACTORIES: dict[str, Any] = { "stocks": stocks, "crypto": crypto, "crypto_dex": crypto_dex, "coin": coin, "forex": forex, "futures": futures, "bond": bond, "cfd": cfd, "options": options, } STOCK_COUNTRY_MARKETS: list[str] = [ "america", "argentina", "australia", "austria", "bangladesh", "belgium", "brazil", "bulgaria", "canada", "chile", "china", "colombia", "croatia", "cyprus", "czech", "denmark", "egypt", "estonia", "finland", "france", "germany", "ghana", "greece", "hongkong", "hungary", "iceland", "india", "indonesia", "ireland", "israel", "italy", "japan", "jordan", "kazakhstan", "kenya", "kuwait", "latvia", "lithuania", "luxembourg", "malaysia", "malta", "mexico", "morocco", "netherlands", "newzealand", "nigeria", "norway", "oman", "pakistan", "peru", "philippines", "poland", "portugal", "qatar", "romania", "saudiarabia", "serbia", "singapore", "slovakia", "slovenia", "southafrica", "southkorea", "spain", "srilanka", "sweden", "switzerland", "taiwan", "thailand", "tunisia", "turkey", "uganda", "ukraine", "unitedarabemirates", "unitedkingdom", "vietnam", ] FIELD_CATEGORIES: dict[str, list[dict[str, str]]] = { "price": [ {"name": "open", "description": "Open price"}, {"name": "high", "description": "High price"}, {"name": "low", "description": "Low price"}, {"name": "close", "description": "Close price"}, {"name": "volume", "description": "Trading volume"}, {"name": "change", "description": "Percent change"}, {"name": "change_abs", "description": "Absolute change"}, {"name": "premarket_change", "description": "Premarket percent change"}, {"name": "postmarket_change", "description": "Postmarket percent change"}, {"name": "VWAP", "description": "Volume Weighted Average Price"}, {"name": "52 Week High", "description": "52-week high price"}, {"name": "52 Week Low", "description": "52-week low price"}, ], "technical": [ {"name": "RSI", "description": "Relative Strength Index (14)"}, {"name": "RSI.5", "description": "RSI (5)"}, {"name": "Stoch.K", "description": "Stochastic %K"}, {"name": "Stoch.D", "description": "Stochastic %D"}, {"name": "MACD.macd", "description": "MACD line"}, {"name": "MACD.signal", "description": "MACD signal line"}, {"name": "MACD.histogram", "description": "MACD histogram"}, {"name": "BB.upper", "description": "Bollinger Band upper"}, {"name": "BB.middle", "description": "Bollinger Band middle (SMA 20)"}, {"name": "BB.lower", "description": "Bollinger Band lower"}, {"name": "SMA", "description": "Simple Moving Average"}, {"name": "EMA", "description": "Exponential Moving Average"}, {"name": "SMA20", "description": "SMA 20"}, {"name": "SMA50", "description": "SMA 50"}, {"name": "SMA200", "description": "SMA 200"}, {"name": "EMA5", "description": "EMA 5"}, {"name": "EMA20", "description": "EMA 20"}, {"name": "EMA50", "description": "EMA 50"}, {"name": "EMA200", "description": "EMA 200"}, {"name": "ADX", "description": "Average Directional Index (14)"}, {"name": "ATR", "description": "Average True Range"}, {"name": "AO", "description": "Awesome Oscillator"}, {"name": "OBV", "description": "On Balance Volume"}, {"name": "CCI20", "description": "Commodity Channel Index (20)"}, {"name": "ROC", "description": "Rate of Change"}, {"name": "Williams %R", "description": "Williams Percent Range (14)"}, {"name": "relative_volume_10d_calc", "description": "Relative volume vs 10-day average"}, {"name": "volume_ma_10", "description": "Volume moving average (10)"}, {"name": "TechRating_1D", "description": "Technical rating (1 day)"}, ], "fundamental": [ {"name": "market_cap_basic", "description": "Market cap"}, {"name": "price_earnings_ttm", "description": "Price/Earnings (TTM)"}, {"name": "earnings_per_share_diluted_ttm", "description": "EPS diluted (TTM)"}, {"name": "earnings_per_share_diluted_yoy_growth_ttm", "description": "EPS YoY growth (TTM)"}, {"name": "dividends_yield_current", "description": "Dividend yield"}, {"name": "dividends_payout_ratio", "description": "Dividend payout ratio"}, {"name": "price_to_book_fq", "description": "Price/Book"}, {"name": "price_to_sales", "description": "Price/Sales"}, {"name": "price_to_cash_flow", "description": "Price/Cash Flow"}, {"name": "return_on_equity", "description": "Return on Equity"}, {"name": "return_on_assets", "description": "Return on Assets"}, {"name": "operating_margin", "description": "Operating margin"}, {"name": "profit_margin", "description": "Profit margin"}, {"name": "revenue_growth", "description": "Revenue growth"}, {"name": "earnings_growth", "description": "Earnings growth"}, {"name": "debt_to_equity", "description": "Debt/Equity"}, {"name": "current_ratio_fq", "description": "Current ratio"}, {"name": "beta_1_year", "description": "Beta (1 year)"}, {"name": "float", "description": "Shares float"}, {"name": "shares_outstanding", "description": "Shares outstanding"}, {"name": "short_ratio_fq", "description": "Short ratio"}, {"name": "short_float", "description": "Short % of float"}, {"name": "insider_ownership", "description": "Insider ownership"}, {"name": "institutional_ownership", "description": "Institutional ownership"}, {"name": "price_earnings_ttm", "description": "P/E ratio (TTM)"}, {"name": "AnalystRating", "description": "Analyst rating (numeric)"}, ], "general": [ {"name": "name", "description": "Short name / ticker"}, {"name": "description", "description": "Company description"}, {"name": "ticker", "description": "Full ticker (EXCHANGE:SYMBOL)"}, {"name": "exchange", "description": "Exchange name"}, {"name": "market", "description": "Market/country"}, {"name": "sector", "description": "Sector"}, {"name": "industry", "description": "Industry"}, {"name": "country", "description": "Country"}, {"name": "currency", "description": "Quote currency"}, {"name": "type", "description": "Instrument type"}, {"name": "typespecs", "description": "Instrument sub-type"}, {"name": "is_primary", "description": "Primary listing flag"}, ], } def _get_market_factory(market_type: str, country_or_param: str | None = None) -> Query: if market_type == "stocks" or market_type not in MARKET_FACTORIES: fact = stocks return fact(country_or_param or "america") fact = MARKET_FACTORIES[market_type] if market_type == "options": return fact(country_or_param or "CME_MINI:ESM2026") return fact() # --------------------------------------------------------------------------- # Historical Candles & Chart Rendering # --------------------------------------------------------------------------- TV_CHART_HEADERS = { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36" ), "Origin": "https://www.tradingview.com", "Referer": "https://www.tradingview.com/", } def _resolve_symbol(ticker: str, market_type: str) -> str: if ":" in ticker: return ticker if market_type == "crypto": return f"BINANCE:{ticker}" return ticker def _to_yahoo_ticker(ticker: str) -> str: """Convert various ticker formats to Yahoo Finance format.""" if ".JK" in ticker or ".JK" in ticker.upper(): return ticker.replace(".jk", ".JK").replace(".Jk", ".JK") if ticker.startswith("IDX:"): return ticker.replace("IDX:", "") + ".JK" if ":" in ticker: return ticker.split(":", 1)[1] return ticker def _fetch_candles( ticker: str, resolution: str = "D", count: int = 30, ) -> list[dict[str, Any]]: # Try TradingView chart API first sym = _resolve_symbol(ticker, "stocks") to_ts = int(time.time()) resolution_seconds = {"1": 60, "5": 300, "15": 900, "30": 1800, "60": 3600, "240": 14400, "D": 86400, "W": 604800, "M": 2592000} from_ts = to_ts - (count * resolution_seconds.get(resolution, 86400)) tv_urls = [ "https://chart-data.tradingview.com/history", "https://scanner.tradingview.com/history", ] for url in tv_urls: try: params = {"symbol": sym, "resolution": resolution, "from": from_ts, "to": to_ts} resp = requests.get(url, headers=TV_CHART_HEADERS, params=params, timeout=10) data = resp.json() if data.get("s") == "ok" and len(data.get("t", [])) > 0: candles = [] for i in range(len(data["t"])): candles.append({ "time": data["t"][i], "open": data["o"][i], "high": data["h"][i], "low": data["l"][i], "close": data["c"][i], "volume": data["v"][i], }) return candles except requests.RequestException: continue # Fallback: Yahoo Finance API try: yahoo_sym = _to_yahoo_ticker(ticker) interval_map = {"1": "1m", "5": "5m", "15": "15m", "30": "30m", "60": "60m", "240": "4h", "D": "1d", "W": "1wk", "M": "1mo"} interval = interval_map.get(resolution, "1d") range_map = {30: "1mo", 60: "3mo", 120: "6mo", 250: "1y", 500: "2y"} y_range = "1mo" for cnt, r in sorted(range_map.items()): if count <= cnt: y_range = r break else: y_range = "5y" url = f"https://query1.finance.yahoo.com/v8/finance/chart/{yahoo_sym}" params = {"interval": interval, "range": y_range} headers = {"User-Agent": TV_CHART_HEADERS["User-Agent"]} resp = requests.get(url, headers=headers, params=params, timeout=15) data = resp.json() result = data.get("chart", {}).get("result", [None])[0] if not result: result = data.get("chart", {}).get("result", [None])[0] if result: timestamps = result.get("timestamp", []) quotes = result.get("indicators", {}).get("quote", [None])[0] if timestamps and quotes: opens = quotes.get("open", []) highs = quotes.get("high", []) lows = quotes.get("low", []) closes = quotes.get("close", []) volumes = quotes.get("volume", []) candles = [] for i in range(len(timestamps)): o, h, l, c, v = opens[i], highs[i], lows[i], closes[i], volumes[i] if None in (o, h, l, c, v): continue candles.append({ "time": timestamps[i], "open": o, "high": h, "low": l, "close": c, "volume": v, }) return candles[-count:] except requests.RequestException: pass return [] def _detect_candle_patterns(candles: list[dict[str, Any]]) -> list[str]: patterns: list[str] = [] if len(candles) < 2: return patterns c = candles[-1] body = abs(c["close"] - c["open"]) upper = c["high"] - max(c["open"], c["close"]) lower = min(c["open"], c["close"]) - c["low"] total_range = c["high"] - c["low"] # Doji if total_range > 0 and body / total_range < 0.1: patterns.append("DOJI (indecision / potential reversal)") # Hammer / Shooting Star if body > 0 and total_range > 0: upper_pct = upper / total_range * 100 lower_pct = lower / total_range * 100 body_pct = body / total_range * 100 if lower_pct > 60 and upper_pct < 10 and body_pct < 30: if c["close"] < c["open"]: patterns.append("HAMMER (potential bullish reversal at bottom)") else: patterns.append("INVERTED HAMMER (potential bullish reversal at bottom)") if upper_pct > 60 and lower_pct < 10 and body_pct < 30: if c["close"] > c["open"]: patterns.append("SHOOTING STAR (potential bearish reversal at top)") else: patterns.append("HANGING MAN (potential bearish reversal at top)") # Marubozu if upper < total_range * 0.05 and lower < total_range * 0.05 and body > total_range * 0.8: if c["close"] > c["open"]: patterns.append("MARUBOZU (strong bullish — no wicks)") else: patterns.append("MARUBOZU (strong bearish — no wicks)") # Engulfing if len(candles) >= 2: p = candles[-2] p_body = abs(p["close"] - p["open"]) if p_body > 0 and body > 0: if c["close"] > c["open"] and c["open"] < p["close"] and c["close"] > p["open"] and body > p_body * 1.2: patterns.append("BULLISH ENGULFING (strong reversal up)") if c["close"] < c["open"] and c["open"] > p["close"] and c["close"] < p["open"] and body > p_body * 1.2: patterns.append("BEARISH ENGULFING (strong reversal down)") # Three white soldiers / Three black crows if len(candles) >= 3: c1, c2, c3 = candles[-3], candles[-2], candles[-1] if all(x["close"] > x["open"] for x in (c1, c2, c3)): b1, b2, b3 = c1["close"] - c1["open"], c2["close"] - c2["open"], c3["close"] - c3["open"] if b2 > 0 and b3 > 0 and b2 >= b1 * 0.9 and b3 >= b2 * 0.9: if c2["open"] > c1["open"] and c3["open"] > c2["open"]: patterns.append("3 WHITE SOLDIERS (strong sustained bullish momentum)") elif b2 > 0 and b3 > 0 and b2 < b1 * 0.7 and b3 < b2 * 0.7: patterns.append("BULLISH IN CONCLUSION (rising but weakening momentum)") if all(x["close"] < x["open"] for x in (c1, c2, c3)): b1, b2, b3 = c1["open"] - c1["close"], c2["open"] - c2["close"], c3["open"] - c3["close"] if b2 > 0 and b3 > 0 and b2 >= b1 * 0.9 and b3 >= b2 * 0.9: if c2["open"] < c1["open"] and c3["open"] < c2["open"]: patterns.append("3 BLACK CROWS (strong sustained bearish momentum)") # Piercing / Dark Cloud if len(candles) >= 2: p = candles[-2] if c["close"] > c["open"] and p["close"] < p["open"]: mid_point = p["open"] - (p["open"] - p["close"]) / 2 if c["close"] > mid_point and c["open"] < p["close"]: patterns.append("PIERCING PATTERN (bullish reversal)") if c["close"] < c["open"] and p["close"] > p["open"]: mid_point = p["close"] - (p["close"] - p["open"]) / 2 if c["close"] < mid_point and c["open"] > p["close"]: patterns.append("DARK CLOUD COVER (bearish reversal)") # Morning/Evening Star (3-candle) if len(candles) >= 3: c1, c2, c3 = candles[-3], candles[-2], candles[-1] c1_body = abs(c1["close"] - c1["open"]) c2_body = abs(c2["close"] - c2["open"]) c3_body = abs(c3["close"] - c3["open"]) if c1["close"] < c1["open"] and c3["close"] > c3["open"]: if c2_body < c1_body * 0.5 and c2_body < c3_body * 0.5: patterns.append("MORNING STAR (bullish reversal — doji middle candle)") if c1["close"] > c1["open"] and c3["close"] < c3["open"]: if c2_body < c1_body * 0.5 and c2_body < c3_body * 0.5: patterns.append("EVENING STAR (bearish reversal — doji middle candle)") if not patterns: patterns.append("No significant candlestick pattern detected") return patterns @mcp.tool() def get_historical_candles( ticker: str, resolution: str = "D", count: int = 30, market_type: str = "stocks", detect_patterns: bool = True, ) -> str: """Fetch historical OHLCV candlestick data for candle pattern and trend analysis. Parameters: ticker: Full ticker (e.g. "IDX:BBRI", "NASDAQ:AAPL", "IDX:BUMI"). resolution: Candle resolution — 1, 3, 5, 15, 30, 60, 240 (minutes), D (daily, default), W (weekly), M (monthly). count: Number of candles to fetch (max 500, default 30). market_type: Market type (stocks, crypto, forex, etc.). detect_patterns: Whether to detect candlestick patterns (default True). """ count = min(count, 500) candles = _fetch_candles(ticker, resolution, count) if not candles: return f"No historical data returned for {ticker}" # Summary stats first, last = candles[0], candles[-1] total_change = ((last["close"] - first["close"]) / first["close"]) * 100 if first["close"] else 0 highest = max(c["high"] for c in candles) lowest = min(c["low"] for c in candles) avg_vol = sum(c["volume"] for c in candles) / len(candles) lines: list[str] = [] lines.append(f"📊 {ticker} — {count}candle chart ({resolution} resolution)") lines.append(f" Period: {time.strftime('%Y-%m-%d', time.gmtime(first['time']))}" f" → {time.strftime('%Y-%m-%d', time.gmtime(last['time']))}") lines.append(f" Latest: O={last['open']:.2f} H={last['high']:.2f} L={last['low']:.2f} C={last['close']:.2f}") lines.append(f" Change: {total_change:.2f}% | Range: {highest:.2f} - {lowest:.2f} | Avg Vol: {avg_vol:,.0f}") lines.append("") # Current candle analysis c = last body = abs(c["close"] - c["open"]) upper_wick = c["high"] - max(c["open"], c["close"]) lower_wick = min(c["open"], c["close"]) - c["low"] total = c["high"] - c["low"] lines.append("── Latest Candle ──") lines.append(f" Type: {'🟢 BULLISH' if c['close'] >= c['open'] else '🔴 BEARISH'}") lines.append(f" Body: {body:.2f} | Upper Wick: {upper_wick:.2f} | Lower Wick: {lower_wick:.2f}") if total > 0: lines.append(f" Body Ratio: {body/total*100:.0f}% | Wick Ratio: {(upper_wick+lower_wick)/total*100:.0f}%") if detect_patterns: lines.append("") lines.append("── Pattern Detection ──") for p in _detect_candle_patterns(candles): lines.append(f" • {p}") # Simple trend analysis if len(candles) >= 10: lines.append("") lines.append("── Trend (last 10 candles) ──") recent = candles[-10:] up_count = sum(1 for x in recent if x["close"] >= x["open"]) down_count = 10 - up_count net = recent[-1]["close"] - recent[0]["close"] lines.append(f" Bullish candles: {up_count} | Bearish candles: {down_count}") lines.append(f" Net change: {net:+.2f} ({net/recent[0]['close']*100:+.2f}%)") if up_count >= 7: lines.append(" → Strong bullish momentum") elif up_count >= 6: lines.append(" → Moderate bullish bias") elif down_count >= 7: lines.append(" → Strong bearish momentum") elif down_count >= 6: lines.append(" → Moderate bearish bias") else: lines.append(" → Sideways / indecision") return "\n".join(lines) # --------------------------------------------------------------------------- # Tools # --------------------------------------------------------------------------- @mcp.tool() def get_stock_quotes( tickers: list[str], columns: list[str] | None = None, market_type: str = "stocks", market_country: str | None = None, sessionid: str | None = None, ) -> str: """Get quote data for specific ticker symbols across any market. Examples: - get_stock_quotes(tickers=["NASDAQ:NVDA", "NASDAQ:AAPL"]) - get_stock_quotes(tickers=["BINANCE:BTCUSDT"], market_type="crypto") - get_stock_quotes(tickers=["NYSE:SPY"], columns=["name","close","volume","RSI"]) """ q = _get_market_factory(market_type, market_country) if columns: q = q.select(*columns) q = q.set_tickers(*tickers) total, data = _exec_query(q, sessionid) return f"Found {total} result(s)\n{data}" @mcp.tool() def screen_market( columns: list[str] | None = None, filters: list[dict[str, Any]] | None = None, order_by: str | None = None, ascending: bool = False, limit: int = 50, offset: int = 0, market_type: str = "stocks", market_country: str | None = None, sessionid: str | None = None, ) -> str: """General-purpose screener across any market type. Parameters: columns: Field names to return (e.g. ["name","close","RSI","volume"]). filters: Structured filter conditions. Simple condition: {"field": "", "operator": "", "value": } Nested group: {"operator": "and"|"or", "filters": []} Operators: >, >=, <, <=, ==, !=, between, not_between, isin, not_in, has, has_none_of, like, not_like, empty, not_empty, crosses, crosses_above, crosses_below, above_pct, below_pct, between_pct, in_day_range, in_week_range, in_month_range Examples: [{"field": "RSI", "operator": ">", "value": 70}] [{"field": "close", "operator": "between", "value": [100, 200]}] [{"field": "exchange", "operator": "isin", "value": ["NASDAQ","NYSE"]}] [{"field": "MACD.macd", "operator": "crosses_above", "value": "MACD.signal"}] Nested: [{"operator": "or", "filters": [ {"field": "RSI", "operator": ">", "value": 70}, {"field": "RSI", "operator": "<", "value": 30} ]}] order_by: Field to sort by (e.g. "volume", "market_cap_basic"). ascending: Sort ascending (default False = descending). limit: Max rows to return (1-1000, default 50). offset: Row offset for pagination. market_type: Market type: stocks, crypto, forex, futures, bond, cfd, coin, crypto_dex, options. market_country: For stocks, a country name (e.g. "america", "india", "germany"). For options, the underlying symbol (e.g. "CME_MINI:ESM2026"). sessionid: Optional TradingView session ID for real-time data. """ q = _get_market_factory(market_type, market_country) if columns: q = q.select(*columns) q = _apply_filters(q, filters) if order_by: q = q.order_by(order_by, ascending=ascending) q = q.limit(limit).offset(offset) total, data = _exec_query(q, sessionid) return f"Total: {total}\n{data}" @mcp.tool() def find_top_gainers( limit: int = 20, min_price: float | None = None, market_type: str = "stocks", market_country: str | None = None, sessionid: str | None = None, ) -> str: """Find top gainers by percent change in any market.""" q = _get_market_factory(market_type, market_country) q = q.select("name", "close", "change", "change_abs", "volume") filters = [] if min_price is not None: filters.append({"field": "close", "operator": ">", "value": min_price}) q = _apply_filters(q, filters) q = q.order_by("change", ascending=False).limit(limit) total, data = _exec_query(q, sessionid) return f"Total: {total}\n{data}" @mcp.tool() def find_top_losers( limit: int = 20, min_price: float | None = None, market_type: str = "stocks", market_country: str | None = None, sessionid: str | None = None, ) -> str: """Find top losers by percent change in any market.""" q = _get_market_factory(market_type, market_country) q = q.select("name", "close", "change", "change_abs", "volume") filters = [] if min_price is not None: filters.append({"field": "close", "operator": ">", "value": min_price}) q = _apply_filters(q, filters) q = q.order_by("change", ascending=True).limit(limit) total, data = _exec_query(q, sessionid) return f"Total: {total}\n{data}" @mcp.tool() def find_most_active( limit: int = 20, min_price: float | None = None, market_type: str = "stocks", market_country: str | None = None, sessionid: str | None = None, ) -> str: """Find most active instruments by volume.""" q = _get_market_factory(market_type, market_country) q = q.select("name", "close", "change", "volume") filters = [] if min_price is not None: filters.append({"field": "close", "operator": ">", "value": min_price}) q = _apply_filters(q, filters) q = q.order_by("volume", ascending=False).limit(limit) total, data = _exec_query(q, sessionid) return f"Total: {total}\n{data}" @mcp.tool() def technical_scan( filters: list[dict[str, Any]], columns: list[str] | None = None, limit: int = 50, market_type: str = "stocks", market_country: str | None = None, sessionid: str | None = None, ) -> str: """Screen using technical indicator conditions. Example filters: [{"field": "RSI", "operator": "<", "value": 30}, # oversold {"field": "MACD.macd", "operator": "crosses_above", "value": "MACD.signal"}, # bullish MACD {"field": "close", "operator": "above_pct", "value": ["SMA50", 1.02]}] # 2% above SMA50 """ default_cols = ["name", "close", "change", "volume", "RSI", "MACD.macd", "SMA50", "VWAP"] q = _get_market_factory(market_type, market_country) q = q.select(*(columns or default_cols)) q = _apply_filters(q, filters) q = q.limit(limit) total, data = _exec_query(q, sessionid) return f"Total: {total}\n{data}" @mcp.tool() def fundamental_scan( filters: list[dict[str, Any]], columns: list[str] | None = None, limit: int = 50, market_type: str = "stocks", market_country: str | None = None, sessionid: str | None = None, ) -> str: """Screen by fundamental metrics (market cap, P/E, dividend, sector, etc.). Example filters: [{"field": "market_cap_basic", "operator": "between", "value": [1e9, 1e11}], {"field": "price_earnings_ttm", "operator": "<", "value": 20}, {"field": "dividends_yield_current", "operator": ">", "value": 2}, {"field": "sector", "operator": "isin", "value": ["Technology", "Healthcare"]}] """ default_cols = ["name", "close", "market_cap_basic", "price_earnings_ttm", "dividends_yield_current", "sector", "industry"] q = _get_market_factory(market_type, market_country) q = q.select(*(columns or default_cols)) q = _apply_filters(q, filters) q = q.limit(limit) total, data = _exec_query(q, sessionid) return f"Total: {total}\n{data}" @mcp.tool() def list_markets() -> str: """List all available market types and stock-country markets.""" markets = { "asset_types": [ {"id": "stocks", "description": "Stocks (common, preferred, DRs, funds) — use market_country param"}, {"id": "crypto", "description": "Centralised-exchange crypto pairs"}, {"id": "crypto_dex", "description": "Decentralised-exchange crypto pairs (USD)"}, {"id": "coin", "description": "CoinMarketCap crypto coins"}, {"id": "forex", "description": "Forex currency pairs"}, {"id": "futures", "description": "Futures contracts"}, {"id": "bond", "description": "Bonds"}, {"id": "cfd", "description": "Contracts for Difference"}, {"id": "options", "description": "Options (use market_country param for underlying symbol)"}, ], "stock_countries": STOCK_COUNTRY_MARKETS, } return str(markets) @mcp.tool() def list_fields(category: str | None = None) -> str: """List available screener fields, optionally filtered by category. Categories: price, technical, fundamental, general If no category given, returns all categories. """ if category: cat = category.lower() if cat in FIELD_CATEGORIES: return str(FIELD_CATEGORIES[cat]) return f"Category '{category}' not found. Available: price, technical, fundamental, general" return str(FIELD_CATEGORIES) @mcp.tool() def set_session(sessionid: str) -> str: """Store a TradingView session ID for real-time data access. Get your sessionid from: 1. Go to tradingview.com and log in 2. Open DevTools > Application > Cookies > tradingview.com 3. Copy the 'sessionid' cookie value """ global _session_cookies _session_cookies = {"sessionid": sessionid} return "Session ID stored successfully" @mcp.tool() def clear_session() -> str: """Clear stored TradingView session ID.""" global _session_cookies _session_cookies = None return "Session ID cleared" if __name__ == "__main__": parser = argparse.ArgumentParser(description="TradingView Screener MCP Server") parser.add_argument( "--transport", choices=["stdio", "sse", "streamable-http"], default="stdio", help="Transport protocol (default: stdio)", ) parser.add_argument("--host", default=None, help="Host to bind (default: 127.0.0.1)") parser.add_argument("--port", type=int, default=None, help="Port to bind (default: 8000)") args = parser.parse_args() if args.host: mcp.settings.host = args.host if args.host not in ("127.0.0.1", "localhost", "::1"): # FastMCP auto-sets restrictive transport_security for localhost at init time. # Reset it when binding to a non-localhost address so external hosts # (e.g. host.docker.internal) are not rejected with 421. mcp.settings.transport_security = None if args.port: mcp.settings.port = args.port mcp.run(transport=args.transport)