feat(fred): get_observations, aligned and downsampled

plugins/fred/.claude-plugin/plugin.json

@@ -1,6 +1,6 @@
 {
   "name": "fred",
-  "version": "0.2.0",
+  "version": "0.3.0",
   "description": "MCP server for the FRED API: economic time series from the St. Louis Fed, with search, aligned multi-series observations, revision history, and the release calendar.",
   "author": {
     "name": "Walker Hughes"

plugins/fred/src/dates.py

@@ -0,0 +1,87 @@
+"""Date parsing for arguments FRED will only accept as YYYY-MM-DD.
+
+A model asked for "the last five years" writes ``start="5y"`` or ``start="2020"`` or
+``start="last 5 years"``. FRED answers all three with HTTP 400 and a message about a
+variable name. Since none of these are ambiguous, they are translated rather than
+rejected.
+"""
+
+import re
+from datetime import date
+
+# "5y", "5 years", "last 5 years", "past 18 months", "10yr".
+_RELATIVE = re.compile(
+    r"^(?:last|past)?\s*(\d+)\s*(d|w|m|q|y|yr|day|days|week|weeks|month|months|quarter|quarters|year|years)$"
+)
+
+_UNIT_MONTHS = {"m": 1, "mo": 1, "month": 1, "months": 1, "q": 3, "quarter": 3, "quarters": 3}
+_UNIT_YEARS = {"y": 1, "yr": 1, "year": 1, "years": 1}
+_UNIT_DAYS = {"d": 1, "day": 1, "days": 1, "w": 7, "week": 7, "weeks": 7}
+
+
+def today() -> date:
+    """Indirection so tests can pin the clock without patching the stdlib."""
+    return date.today()
+
+
+def parse(value: str, *, field: str) -> str:
+    """Return a YYYY-MM-DD string, or "" for an unset value.
+
+    Accepted, beyond a full date: a bare year ("2020"), a year-month ("2020-01"),
+    "today"/"now", "ytd", and a relative span ("5y", "18 months", "last 10 years")
+    measured back from today.
+    """
+    text = " ".join(str(value).strip().lower().split())
+    if not text:
+        return ""
+
+    if text in ("today", "now"):
+        return today().isoformat()
+    if text in ("ytd", "year to date", "this year"):
+        return today().replace(month=1, day=1).isoformat()
+
+    if re.fullmatch(r"\d{4}", text):
+        return f"{text}-01-01"
+    if re.fullmatch(r"\d{4}-\d{2}", text):
+        return f"{text}-01"
+    if re.fullmatch(r"\d{4}-\d{2}-\d{2}", text):
+        # Round-tripped through date() so 2020-13-01 is caught here rather than by FRED.
+        try:
+            return date.fromisoformat(text).isoformat()
+        except ValueError as exc:
+            raise ValueError(f"{field}={value!r} is not a real date: {exc}") from exc
+
+    match = _RELATIVE.match(text)
+    if match:
+        amount, unit = int(match.group(1)), match.group(2)
+        return _ago(amount, unit).isoformat()
+
+    raise ValueError(
+        f"{field}={value!r} could not be read as a date. Use YYYY-MM-DD, a year "
+        '("2020"), a year-month ("2020-01"), a span back from today ("5y", '
+        '"18 months"), "ytd", or "today".'
+    )
+
+
+def _ago(amount: int, unit: str) -> date:
+    """Today, minus ``amount`` of ``unit``. Calendar arithmetic, no dependency."""
+    now = today()
+    if unit in _UNIT_DAYS:
+        return date.fromordinal(max(1, now.toordinal() - amount * _UNIT_DAYS[unit]))
+    if unit in _UNIT_YEARS:
+        months = amount * 12
+    else:
+        months = amount * _UNIT_MONTHS[unit]
+
+    total = now.year * 12 + (now.month - 1) - months
+    year, month = divmod(total, 12)
+    month += 1
+    # Clamp the day: three months before the 31st is not the 31st of a 30-day month.
+    day = min(now.day, _days_in_month(year, month))
+    return date(year, month, day)
+
+
+def _days_in_month(year: int, month: int) -> int:
+    if month == 12:
+        return 31
+    return (date(year + (month == 12), month % 12 + 1, 1).toordinal()) - date(year, month, 1).toordinal()

plugins/fred/src/schemas.py

@@ -14,6 +14,8 @@
 
 from pydantic import BaseModel, Field, field_validator, model_validator
 
+from . import dates
+
 MAX_SERIES_PER_CALL = 20
 
 # FRED's frequency codes, and the words a model writes instead.
@@ -75,6 +77,45 @@
 
 SERIES_INCLUDES = ("metadata", "notes", "release", "categories", "tags")
 
+# What each FRED units code actually does, echoed in the response so the numbers are
+# not left to be guessed at.
+UNITS_MEANING: dict[str, str] = {
+    "lin": "levels, as published",
+    "chg": "change from the previous period",
+    "ch1": "change from a year ago",
+    "pch": "percent change from the previous period",
+    "pc1": "percent change from a year ago",
+    "pca": "compounded annual rate of change",
+    "cch": "continuously compounded rate of change",
+    "cca": "continuously compounded annual rate of change",
+    "log": "natural log",
+}
+
+# The single highest-value table in this file. "Year over year percent change" is the
+# most-asked transformation in all of economics and FRED spells it "pc1".
+UNITS_ALIASES: dict[str, str] = {
+    "": "lin", "lin": "lin", "level": "lin", "levels": "lin", "none": "lin",
+    "raw": "lin", "as published": "lin", "nominal": "lin",
+    "pc1": "pc1", "yoy": "pc1", "y/y": "pc1", "yoy%": "pc1", "year over year": "pc1",
+    "year-over-year": "pc1", "annual percent change": "pc1", "percent change from year ago": "pc1",
+    "percent change from a year ago": "pc1", "inflation": "pc1",
+    "pch": "pch", "mom": "pch", "m/m": "pch", "month over month": "pch",
+    "percent change": "pch", "pct change": "pch", "pct_change": "pch",
+    "percent change from previous period": "pch",
+    "chg": "chg", "change": "chg", "diff": "chg", "difference": "chg",
+    "change from previous period": "chg",
+    "ch1": "ch1", "change from year ago": "ch1", "change from a year ago": "ch1",
+    "pca": "pca", "annualized": "pca", "saar": "pca", "annual rate": "pca",
+    "compounded annual rate of change": "pca",
+    "cch": "cch", "cca": "cca", "log": "log", "natural log": "log", "ln": "log",
+}  # fmt: skip
+
+AGGREGATION_ALIASES: dict[str, str] = {
+    "avg": "avg", "average": "avg", "mean": "avg",
+    "sum": "sum", "total": "sum",
+    "eop": "eop", "end": "eop", "end of period": "eop", "last": "eop", "close": "eop",
+}  # fmt: skip
+
 
 def normalize_series_ids(value: Any) -> Any:
     """Accept the several ways a model writes a list of series IDs.
@@ -231,3 +272,72 @@ def _known_includes(cls, value: list[str]) -> list[str]:
 
     def wants(self, part: str) -> bool:
         return part in self.include
+
+
+class ObservationArgs(BaseModel):
+    """Arguments for get_observations."""
+
+    series_ids: list[str] = Field(min_length=1, max_length=MAX_SERIES_PER_CALL)
+    start: str = ""
+    end: str = ""
+    units: str = "lin"
+    frequency: str = ""
+    aggregation_method: str = "avg"
+    # Floor of 2 because downsampling always keeps the first and last point; a cap of
+    # one cannot express a series.
+    max_points: int = Field(default=120, ge=2, le=2000)
+
+    @model_validator(mode="before")
+    @classmethod
+    def _correct(cls, data: Any) -> Any:
+        if not isinstance(data, dict):
+            return data
+        data = dict(data)
+        if "series_ids" in data:
+            data["series_ids"] = normalize_series_ids(data["series_ids"])
+        for field in ("start", "end"):
+            if data.get(field) is not None:
+                data[field] = dates.parse(data[field], field=field)
+        if data.get("units") is not None:
+            raw = str(data["units"]).strip().lower()
+            data["units"] = UNITS_ALIASES.get(raw, raw)
+        if "frequency" in data:
+            data["frequency"] = normalize_frequency(data["frequency"])
+        if data.get("aggregation_method") is not None:
+            raw = str(data["aggregation_method"]).strip().lower()
+            data["aggregation_method"] = AGGREGATION_ALIASES.get(raw, raw)
+        return data
+
+    @field_validator("units")
+    @classmethod
+    def _known_units(cls, value: str) -> str:
+        if value not in UNITS_MEANING:
+            raise ValueError(
+                f"unknown units {value!r}; use a FRED code ({', '.join(UNITS_MEANING)}) "
+                "or a phrase such as 'yoy', 'percent change', 'level'"
+            )
+        return value
+
+    @field_validator("frequency")
+    @classmethod
+    def _known_frequency(cls, value: str) -> str:
+        if value and value not in FREQUENCY_TAGS:
+            raise ValueError(f"unknown frequency {value!r}; use one of {', '.join(FREQUENCY_TAGS)}")
+        return value
+
+    @field_validator("aggregation_method")
+    @classmethod
+    def _known_aggregation(cls, value: str) -> str:
+        if value not in ("avg", "sum", "eop"):
+            raise ValueError(f"unknown aggregation_method {value!r}; use avg, sum, or eop")
+        return value
+
+    @model_validator(mode="after")
+    def _range_is_the_right_way_round(self) -> "ObservationArgs":
+        if self.start and self.end and self.start > self.end:
+            raise ValueError(f"start ({self.start}) is after end ({self.end}); swap them")
+        return self
+
+    @property
+    def units_meaning(self) -> str:
+        return UNITS_MEANING[self.units]

plugins/fred/src/shaping.py

@@ -66,3 +66,106 @@ def series_list(payload: dict, *, notes: bool = False) -> list[dict]:
 def first_or_empty(payload: dict, key: str) -> dict[str, Any]:
     items = payload.get(key) or []
     return items[0] if items else {}
+
+
+# --- observations ------------------------------------------------------------------
+#
+# FRED returns one object per observation:
+#
+#     {"realtime_start": "2026-08-05", "realtime_end": "2026-08-05",
+#      "date": "2025-01-01", "value": "2.99098"}
+#
+# On a normal (non-vintage) request the two realtime fields hold the same value on
+# every single row, and the key names repeat once per observation. Twenty years of a
+# daily series is around 5,000 of those. Columnar output drops all of it.
+
+Number = float | None
+
+
+def parse_value(raw: object) -> Number:
+    """FRED writes values as strings and missing ones as ".", not null."""
+    if raw is None:
+        return None
+    text = str(raw).strip()
+    if text in ("", "."):
+        return None
+    try:
+        return float(text)
+    except ValueError:
+        return None
+
+
+def observation_pairs(payload: dict) -> list[tuple[str, Number]]:
+    """(date, value) for each observation, values parsed."""
+    return [(obs["date"], parse_value(obs.get("value"))) for obs in payload.get("observations", []) if obs.get("date")]
+
+
+def align(per_series: dict[str, list[tuple[str, Number]]]) -> tuple[list[str], dict[str, list[Number]]]:
+    """Put several series on one shared, sorted date index.
+
+    This is what makes comparison a single tool call. A monthly and a quarterly series
+    will not line up, so the index is the union of their dates and the gaps are null
+    rather than silently dropped or forward-filled, which would invent data.
+    """
+    dates = sorted({d for pairs in per_series.values() for d, _ in pairs})
+    index = {date: position for position, date in enumerate(dates)}
+
+    columns: dict[str, list[Number]] = {}
+    for series_id, pairs in per_series.items():
+        column: list[Number] = [None] * len(dates)
+        for date, value in pairs:
+            column[index[date]] = value
+        columns[series_id] = column
+    return dates, columns
+
+
+def summarize(pairs: list[tuple[str, Number]]) -> dict:
+    """The figures worth having, computed over every observation.
+
+    Deliberately computed *before* downsampling. A 5,000-point daily series thinned to
+    120 still reports its true min, max, and latest value; a summary computed after
+    would quietly report the extremes of the sample instead of the series.
+    """
+    observed = [(date, value) for date, value in pairs if value is not None]
+    if not observed:
+        return {"observations": len(pairs), "count": 0}
+
+    values = [value for _, value in observed]
+    latest_date, latest = observed[-1]
+    summary: dict[str, Any] = {
+        "latest": latest,
+        "latest_date": latest_date,
+        "min": min(values),
+        "max": max(values),
+        "mean": round(sum(values) / len(values), 6),
+        "start_date": observed[0][0],
+        "end_date": latest_date,
+        "count": len(observed),
+        "observations": len(pairs),
+    }
+    if len(observed) > 1:
+        prior = observed[-2][1]
+        summary["prior"] = prior
+        summary["change"] = round(latest - prior, 6)
+        if prior:
+            summary["pct_change"] = round((latest - prior) / abs(prior) * 100, 4)
+    return summary
+
+
+def downsample(dates: list[str], columns: dict[str, list[Number]], max_points: int) -> tuple[list, dict, int]:
+    """Thin to evenly spaced points, always keeping the first and the last.
+
+    Returns the kept dates, the kept columns, and how many points were dropped, so a
+    truncated series is never mistaken for a complete one.
+    """
+    total = len(dates)
+    if total <= max_points:
+        return dates, columns, 0
+
+    step = (total - 1) / (max_points - 1)
+    keep = sorted({round(i * step) for i in range(max_points)} | {0, total - 1})
+    return (
+        [dates[i] for i in keep],
+        {series_id: [column[i] for i in keep] for series_id, column in columns.items()},
+        total - len(keep),
+    )