admin/tradingview-screener-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add date/time range support to get_historical_candles

Browse Source 
- _fetch_candles now accepts explicit from_ts/to_ts timestamps; falls
  back to count-based window when omitted
- get_historical_candles accepts from_date/to_date in YYYY-MM-DD,
  YYYY-MM-DD HH:MM, or YYYY-MM-DD HH:MM:SS formats (all UTC)
- Yahoo Finance fallback switched from range= to period1/period2 so
  it respects the exact window in both modes
- Added 11 tests covering date range, datetime, end-of-day bump,
  invalid format, count cap, and pattern detection toggle
- gitignore uv.lock (Dockerfile uses pip, not uv sync)
This commit is contained in:
Achmad
2026-05-17 08:10:03 +00:00
parent b24a424c8f
commit 79d1e0e538
3 changed files with 137 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server.py
+44 -19
View File
@@ -311,13 +311,17 @@ def _fetch_candles(
ticker: str,
resolution: str = "D",
count: int = 30,
from_ts: int | None = None,
to_ts: int | None = None,
) -> 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))
if to_ts is None:
to_ts = int(time.time())
if from_ts is None:
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",
@@ -349,24 +353,14 @@ def _fetch_candles(
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}
params: dict[str, Any] = {"interval": interval, "period1": from_ts, "period2": to_ts}
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]
@@ -385,7 +379,7 @@ def _fetch_candles(
"time": timestamps[i],
"open": o, "high": h, "low": l, "close": c, "volume": v,
})
return candles[-count:]
return candles
except requests.RequestException:
pass
@@ -494,6 +488,8 @@ def get_historical_candles(
count: int = 30,
market_type: str = "stocks",
detect_patterns: bool = True,
from_date: str | None = None,
to_date: str | None = None,
) -> str:
"""Fetch historical OHLCV candlestick data for candle pattern and trend analysis.
@@ -501,12 +497,40 @@ def get_historical_candles(
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).
count: Number of candles to fetch when no date range given (max 500, default 30).
market_type: Market type (stocks, crypto, forex, etc.).
detect_patterns: Whether to detect candlestick patterns (default True).
from_date: Start date/datetime (e.g. "2026-05-08" or "2026-05-17 14:00"). When
provided, fetches candles from this point instead of counting back from now.
to_date: End date/datetime (e.g. "2026-05-13" or "2026-05-17 14:05"). Defaults to
now when from_date is given but to_date is omitted.
"""
import calendar
def _parse_dt(s: str, end_of_period: bool = False) -> int:
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"):
try:
t = time.strptime(s, fmt)
ts = int(calendar.timegm(t))
# For date-only strings, end_of_period bumps to 23:59:59
if end_of_period and fmt == "%Y-%m-%d":
ts += 86399
return ts
except ValueError:
continue
raise ValueError(f"Unrecognised date/datetime format: {s!r}. Use YYYY-MM-DD or YYYY-MM-DD HH:MM")
from_ts: int | None = None
to_ts: int | None = None
if from_date:
from_ts = _parse_dt(from_date)
if to_date:
to_ts = _parse_dt(to_date, end_of_period=True)
elif from_date:
to_ts = int(time.time())
count = min(count, 500)
candles = _fetch_candles(ticker, resolution, count)
candles = _fetch_candles(ticker, resolution, count, from_ts=from_ts, to_ts=to_ts)
if not candles:
return f"No historical data returned for {ticker}"
@@ -518,7 +542,8 @@ def get_historical_candles(
avg_vol = sum(c["volume"] for c in candles) / len(candles)
lines: list[str] = []
lines.append(f"📊 {ticker} — {count}candle chart ({resolution} resolution)")
candle_label = f"{from_date} → {to_date or 'now'}" if from_date else f"{len(candles)} candles"
lines.append(f"📊 {ticker} — {candle_label} ({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}")