admin/tradingview-screener-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
38f9b980eceb8adf1591d5ecd1730297f5552c16
tradingview-screener-mcp/server.py
T
achmad 38f9b980ec Initial commit: MCP server wrapping tradingview-screener as MCP tools 
- 11 MCP tools: get_stock_quotes, screen_market, find_top_gainers,
  find_top_losers, find_most_active, technical_scan, fundamental_scan,
  list_markets, list_fields, set_session, clear_session
- Structured filter conditions with nested And/Or groups
- Session cookie resolution: per-call > in-memory > env var
- auto-load .env via python-dotenv
- 69 unit tests covering helpers, conditions, filters, and all tools
2026-05-17 11:09:26 +07:00

523 lines
20 KiB
Python
Raw Blame History

from __future__ import annotations
import os
from typing import Any, cast
from dotenv import load_dotenv
load_dotenv()
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()
# ---------------------------------------------------------------------------
# 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": "<name>", "operator": "<op>", "value": <value>}
Nested group: {"operator": "and"|"or", "filters": [<conditions>]}
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__":
mcp.run(transport="stdio")
Reference in New Issue View Git Blame Copy Permalink