Add SQLite-cached check_holidays MCP tool with pre-analysis checklist
- Add check_holidays MCP tool in server.py with SQLite caching (cache-first, API-fallback) - Add Section 6 Pre-Analysis Checklist (market status, holidays, news, analysis type) to SKILL.md - Add holidays-data named volume to docker-compose.yml for DB persistence - Add API_CO_ID_KEY and HOLIDAYS_DB_PATH env vars to Dockerfile and .env.example - Remove scripts/ directory (replaced by MCP tool), slim down Dockerfile
This commit is contained in:
Achmad
@@ -817,6 +817,249 @@ def clear_session() -> str:
|
||||
return "Session ID cleared"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Holiday Checker (api.co.id) — SQLite backed
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
import sqlite3
|
||||
from datetime import date as dt_date, datetime
|
||||
from pathlib import Path
|
||||
|
||||
_HOLIDAYS_API_KEY: str | None = None
|
||||
_HOLIDAYS_DB: str | None = None
|
||||
|
||||
|
||||
def _get_holidays_api_key() -> str | None:
|
||||
global _HOLIDAYS_API_KEY
|
||||
if _HOLIDAYS_API_KEY is None:
|
||||
_HOLIDAYS_API_KEY = os.environ.get("API_CO_ID_KEY") or ""
|
||||
return _HOLIDAYS_API_KEY or None
|
||||
|
||||
|
||||
def _get_holidays_db_path() -> str:
|
||||
global _HOLIDAYS_DB
|
||||
if _HOLIDAYS_DB is None:
|
||||
_HOLIDAYS_DB = os.environ.get("HOLIDAYS_DB_PATH") or str(Path(__file__).parent / "holidays.db")
|
||||
return _HOLIDAYS_DB
|
||||
|
||||
|
||||
def _init_holidays_db() -> sqlite3.Connection:
|
||||
db_path = _get_holidays_db_path()
|
||||
conn = sqlite3.connect(db_path)
|
||||
conn.execute("""
|
||||
CREATE TABLE IF NOT EXISTS holidays (
|
||||
date TEXT PRIMARY KEY,
|
||||
name TEXT NOT NULL,
|
||||
type TEXT NOT NULL,
|
||||
is_national INTEGER NOT NULL DEFAULT 1
|
||||
)
|
||||
""")
|
||||
conn.execute("CREATE INDEX IF NOT EXISTS idx_holidays_date ON holidays(date)")
|
||||
conn.commit()
|
||||
return conn
|
||||
|
||||
|
||||
def _cache_holidays(items: list[dict]) -> None:
|
||||
conn = _init_holidays_db()
|
||||
for h in items:
|
||||
conn.execute(
|
||||
"INSERT OR REPLACE INTO holidays (date, name, type, is_national) VALUES (?, ?, ?, ?)",
|
||||
(h["date"], h["name"], h.get("type", "Public Holiday"), 1 if h.get("is_national_holiday", True) else 0),
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
|
||||
def _query_cached_holidays(
|
||||
start_date: str = "",
|
||||
end_date: str = "",
|
||||
limit: int = 0,
|
||||
) -> list[dict]:
|
||||
conn = _init_holidays_db()
|
||||
today = datetime.now().strftime("%Y-%m-%d")
|
||||
if start_date and end_date:
|
||||
rows = conn.execute(
|
||||
"SELECT date, name, type, is_national FROM holidays WHERE date >= ? AND date <= ? ORDER BY date",
|
||||
(start_date, end_date),
|
||||
).fetchall()
|
||||
elif start_date:
|
||||
rows = conn.execute(
|
||||
"SELECT date, name, type, is_national FROM holidays WHERE date >= ? ORDER BY date",
|
||||
(start_date,),
|
||||
).fetchall()
|
||||
elif limit:
|
||||
rows = conn.execute(
|
||||
"SELECT date, name, type, is_national FROM holidays WHERE date >= ? ORDER BY date LIMIT ?",
|
||||
(today, limit),
|
||||
).fetchall()
|
||||
else:
|
||||
rows = conn.execute(
|
||||
"SELECT date, name, type, is_national FROM holidays WHERE date >= ? ORDER BY date",
|
||||
(today,),
|
||||
).fetchall()
|
||||
conn.close()
|
||||
return [
|
||||
{"date": r[0], "name": r[1], "type": r[2], "is_national_holiday": bool(r[3])}
|
||||
for r in rows
|
||||
]
|
||||
|
||||
|
||||
def _fetch_and_cache_holidays(year: int) -> list[dict]:
|
||||
"""Fetch holidays for a year from API and cache them."""
|
||||
api_key = _get_holidays_api_key()
|
||||
if not api_key:
|
||||
return []
|
||||
|
||||
resp = requests.get(
|
||||
"https://use.api.co.id/holidays/indonesia/",
|
||||
headers={"x-api-co-id": api_key},
|
||||
params={"year": year},
|
||||
timeout=10,
|
||||
)
|
||||
if resp.status_code != 200:
|
||||
return []
|
||||
|
||||
data = resp.json()
|
||||
if not data.get("is_success"):
|
||||
return []
|
||||
|
||||
items = data.get("data", [])
|
||||
if not isinstance(items, list):
|
||||
items = [items]
|
||||
|
||||
if items:
|
||||
_cache_holidays(items)
|
||||
return items
|
||||
|
||||
|
||||
def _format_holidays(items: list[dict]) -> str:
|
||||
if not items:
|
||||
return "No holidays found."
|
||||
|
||||
today = datetime.now().date()
|
||||
lines: list[str] = []
|
||||
lines.append(f"{'Date':<15} {'Day':<10} {'Holiday':<45} {'Type':<20}")
|
||||
lines.append("-" * 90)
|
||||
for h in items:
|
||||
d_str = h.get("date", "?")
|
||||
name = h.get("name", "?")
|
||||
typ = h.get("type", "?")
|
||||
try:
|
||||
dt = datetime.strptime(d_str, "%Y-%m-%d").date()
|
||||
day_name = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"][dt.weekday()]
|
||||
delta = (dt - today).days
|
||||
if delta == 0:
|
||||
suffix = " ⬥ TODAY"
|
||||
elif delta == 1:
|
||||
suffix = " ⬥ Tomorrow"
|
||||
elif delta > 0:
|
||||
suffix = f" ({delta}d away)"
|
||||
elif delta < 0:
|
||||
suffix = f" ({abs(delta)}d ago)"
|
||||
else:
|
||||
suffix = ""
|
||||
except ValueError:
|
||||
day_name = "?"
|
||||
suffix = ""
|
||||
lines.append(f"{d_str:<15} {day_name:<10} {name:<45} {typ:<20}{suffix}")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def check_holidays(
|
||||
mode: str = "upcoming",
|
||||
year: int | None = None,
|
||||
date: str | None = None,
|
||||
start_date: str | None = None,
|
||||
end_date: str | None = None,
|
||||
limit: int = 10,
|
||||
refresh: bool = False,
|
||||
) -> str:
|
||||
"""Check Indonesian public holidays (cached in SQLite).
|
||||
|
||||
Modes:
|
||||
upcoming (default): next N holidays
|
||||
year: all holidays for a given year (pass `year`)
|
||||
check: check if a specific date is a holiday (pass `date`, format YYYY-MM-DD)
|
||||
range: holidays in a date range (pass `start_date` & `end_date`)
|
||||
|
||||
Pass refresh=True to re-fetch from API instead of using cache.
|
||||
Requires API_CO_ID_KEY env var to be set for API calls.
|
||||
"""
|
||||
match mode:
|
||||
case "check":
|
||||
if not date:
|
||||
return "Error: `date` param required (YYYY-MM-DD)"
|
||||
# Try cache first
|
||||
cached = _query_cached_holidays(start_date=date, end_date=date)
|
||||
if cached and not refresh:
|
||||
return _format_holidays(cached)
|
||||
# Fall back to API
|
||||
api_key = _get_holidays_api_key()
|
||||
if not api_key:
|
||||
return "⚠ No cached data and API_CO_ID_KEY not set."
|
||||
resp = requests.get(
|
||||
"https://use.api.co.id/holidays/indonesia/check/date",
|
||||
headers={"x-api-co-id": api_key},
|
||||
params={"date": date},
|
||||
timeout=10,
|
||||
)
|
||||
if resp.status_code != 200:
|
||||
return f"API error: HTTP {resp.status_code}"
|
||||
data = resp.json()
|
||||
if not data.get("is_success"):
|
||||
return f"API error: {data.get('message', 'unknown')}"
|
||||
d = data.get("data", {})
|
||||
if not d or not d.get("is_holiday"):
|
||||
return f"{date} is NOT a holiday."
|
||||
items = [d.get("holiday", {})]
|
||||
return _format_holidays(items)
|
||||
|
||||
case "range":
|
||||
if not start_date or not end_date:
|
||||
return "Error: `start_date` and `end_date` required"
|
||||
cached = _query_cached_holidays(start_date=start_date, end_date=end_date)
|
||||
if cached and not refresh:
|
||||
return _format_holidays(cached)
|
||||
api_key = _get_holidays_api_key()
|
||||
if not api_key:
|
||||
return "⚠ No cached data and API_CO_ID_KEY not set."
|
||||
resp = requests.get(
|
||||
"https://use.api.co.id/holidays/indonesia/check/range",
|
||||
headers={"x-api-co-id": api_key},
|
||||
params={"start_date": start_date, "end_date": end_date},
|
||||
timeout=10,
|
||||
)
|
||||
if resp.status_code != 200:
|
||||
return f"API error: HTTP {resp.status_code}"
|
||||
data = resp.json()
|
||||
if not data.get("is_success"):
|
||||
return f"API error: {data.get('message', 'unknown')}"
|
||||
items = data.get("data", [])
|
||||
if not isinstance(items, list):
|
||||
items = [items]
|
||||
return _format_holidays(items)
|
||||
|
||||
case "year":
|
||||
y = year or datetime.now().year
|
||||
cached = _query_cached_holidays(start_date=f"{y}-01-01", end_date=f"{y}-12-31")
|
||||
if cached and not refresh:
|
||||
return _format_holidays(cached)
|
||||
items = _fetch_and_cache_holidays(y)
|
||||
return _format_holidays(items)
|
||||
|
||||
case _: # upcoming
|
||||
cached = _query_cached_holidays(limit=limit)
|
||||
if cached and not refresh:
|
||||
return _format_holidays(cached)
|
||||
# Fetch current & next year to build cache, then return from cache
|
||||
current_year = datetime.now().year
|
||||
_fetch_and_cache_holidays(current_year)
|
||||
_fetch_and_cache_holidays(current_year + 1)
|
||||
cached = _query_cached_holidays(limit=limit)
|
||||
return _format_holidays(cached) if cached else "No upcoming holidays found."
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser(description="TradingView Screener MCP Server")
|
||||
parser.add_argument(
|
||||
|
||||
Reference in New Issue
Block a user