Providers: IMF multi-country accumulation, CoinGecko honesty,

ExchangeRate keyed endpoint, retrieval-layer concurrency

- IMF SDMX multi-country batches accumulate first-success-per-country
  (a 3-country comparison returned after country #1, silently dropping
  the rest); CSV series group by the DSD structure's real dimension ids
  so attribute columns (STATUS) can't fragment a series; DataMapper
  source links use the code's actual dataset anchor, not @weo.
- CoinGecko: the free-tier 365-day history clamp is disclosed via a
  metadata note on both paths, and frequency is derived from the ACTUAL
  point spacing (8-90 day hourly windows were labeled "daily").
- ExchangeRate: the keyed v6 endpoint returns conversion_rates, not
  rates — setting an API key would have hard-broken every spot query.
- Indicator retrieval: FTS query builder strips ALL non-word characters
  (an enumerated escape list let '"2025,"*' through — an FTS5 syntax
  error that killed the whole lexical arm for any punctuated query, seen
  in prod); sqlite connections are per-thread (the shared connection
  corrupted in-flight cursors under concurrent searches, ~40/day
  InterfaceError in prod, silently degrading retrieval).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: hanlulong

backend/providers/coingecko.py

@@ -47,6 +47,34 @@ class CoinGeckoProvider(BaseProvider):
     def provider_name(self) -> str:
         return "CoinGecko"
 
+    @staticmethod
+    def _frequency_from_point_spacing(raw_points: Any, requested_days: float) -> str:
+        """Label frequency from the ACTUAL spacing of the returned points.
+
+        CoinGecko's market_chart auto-granularity (intraday / hourly / daily)
+        depends on the window in ways that have shifted over time; deriving
+        the label from the served data can never mislabel (a request-days
+        ladder called 8-90 day hourly windows "daily").
+        """
+        timestamps = [pt[0] for pt in (raw_points or []) if isinstance(pt, (list, tuple)) and pt]
+        if len(timestamps) >= 3:
+            deltas = sorted(
+                t2 - t1 for t1, t2 in zip(timestamps, timestamps[1:]) if t2 > t1
+            )
+            if deltas:
+                median_ms = deltas[len(deltas) // 2]
+                if median_ms <= 30 * 60 * 1000:
+                    return "5-minute"
+                if median_ms <= 6 * 60 * 60 * 1000:
+                    return "hourly"
+                return "daily"
+        # Too few points to measure — fall back to the request-window ladder.
+        if requested_days <= 1:
+            return "5-minute"
+        if requested_days <= 90:
+            return "hourly"
+        return "daily"
+
     def __init__(self, api_key: Optional[str] = None, timeout: float = 30.0):
         """
         Initialize CoinGecko provider.
@@ -308,9 +336,11 @@ async def get_historical_data(
 
         # CoinGecko free/demo tier has a 365-day limit for historical data
         # Pro tier doesn't have this limit
+        was_clamped_to_free_tier = False
         if not self.is_pro and days > 365:
             logger.warning(f"⚠️ CoinGecko: Free tier limited to 365 days, requested {days} days. Capping at 365.")
             days = 365
+            was_clamped_to_free_tier = True
         params = {
             "vs_currency": vs_currency,
             "days": str(days),
@@ -349,13 +379,11 @@ async def get_historical_data(
 
         logger.info(f"📊 Created {len(data_points)} data points for {metric}")
 
-        # Determine frequency
-        if days == 1:
-            frequency = "5-minute"
-        elif days <= 7:
-            frequency = "hourly"
-        else:
-            frequency = "daily"
+        # Determine frequency from the ACTUAL point spacing the API returned
+        # (its auto-granularity: intraday for 1 day, hourly for 2-90 days,
+        # daily beyond). The old request-days ladder mislabeled 8-90 day
+        # windows as "daily" while the points were hourly.
+        frequency = self._frequency_from_point_spacing(data.get(metric_key, []), days)
 
         # Determine indicator name and unit based on metric
         metric_display_mapping = {
@@ -374,6 +402,14 @@ async def get_historical_data(
             lastUpdated=datetime.now().isoformat(),
             seriesId=coin_id,
             apiUrl=api_url,
+            notes=(
+                [
+                    "CoinGecko's free tier limits history to the last 365 days; "
+                    "the requested range was truncated accordingly."
+                ]
+                if was_clamped_to_free_tier
+                else None
+            ),
         )
 
         return [NormalizedData(metadata=metadata, data=data_points)]
@@ -419,9 +455,11 @@ async def get_historical_data_range(
         max_days_ago = 365 * 24 * 60 * 60  # 365 days in seconds
         min_allowed_timestamp = now_timestamp - max_days_ago
 
+        was_clamped_to_free_tier = False
         if not self.is_pro and from_timestamp < min_allowed_timestamp:
             logger.warning(f"⚠️ CoinGecko: Free tier limited to 365 days. Adjusting from_date.")
             from_timestamp = min_allowed_timestamp
+            was_clamped_to_free_tier = True
             logger.info(f"   - Adjusted from_timestamp: {datetime.fromtimestamp(from_timestamp, tz=timezone.utc).isoformat()}")
 
         params = {
@@ -456,14 +494,9 @@ async def get_historical_data_range(
             except (ValueError, TypeError):
                 continue
 
-        # Determine frequency based on date range
+        # Frequency from actual point spacing (see _frequency_from_point_spacing)
         days_diff = (to_timestamp - from_timestamp) / 86400
-        if days_diff <= 1:
-            frequency = "5-minute"
-        elif days_diff <= 7:
-            frequency = "hourly"
-        else:
-            frequency = "daily"
+        frequency = self._frequency_from_point_spacing(data.get(metric_key, []), days_diff)
 
         # Determine indicator name and unit based on metric
         metric_display_mapping = {
@@ -482,6 +515,14 @@ async def get_historical_data_range(
             lastUpdated=datetime.now().isoformat(),
             seriesId=coin_id,
             apiUrl=api_url,
+            notes=(
+                [
+                    "CoinGecko's free tier limits history to the last 365 days; "
+                    "the requested range was truncated accordingly."
+                ]
+                if was_clamped_to_free_tier
+                else None
+            ),
         )
 
         return [NormalizedData(metadata=metadata, data=data_points)]

backend/providers/exchangerate.py

@@ -133,7 +133,7 @@ async def fetch_exchange_rate(
             response = await self._get_with_retry(client, full_url, timeout=15.0)
             logger.info(f"📊 Response status: {response.status_code}")
             data = response.json()
-            logger.info(f"✅ Response received. Result: {data.get('result')}, Rates count: {len(data.get('rates', {}))}")
+            logger.info(f"✅ Response received. Result: {data.get('result')}, Rates count: {len(data.get('rates') or data.get('conversion_rates') or {})}")
 
             if data.get("result") != "success":
                 error_msg = data.get('error-type', 'Unknown error')
@@ -156,7 +156,10 @@ async def fetch_exchange_rate(
             logger.error(f"Unexpected error fetching exchange rates: {str(e)}")
             raise DataNotAvailableError(f"Failed to fetch exchange rates: {str(e)}")
 
-        rates = data.get("rates", {})
+        # Keyless open endpoint returns "rates"; the keyed v6 API returns
+        # "conversion_rates" for the same call — read whichever is present so
+        # setting an API key doesn't hard-break every spot-rate query.
+        rates = data.get("rates") or data.get("conversion_rates") or {}
         time_last_update = data.get("time_last_update_utc", "")
 
         logger.info(f"📈 Processing {len(rates)} exchange rates")

backend/providers/imf.py

@@ -707,13 +707,32 @@ async def fetch_batch_indicator(
         # an EMPTY unit. Use the catalog unit when present; fall back to the
         # heuristic only when it is missing. Graceful on any lookup failure.
         catalog_unit = ""
+        # DataMapper website anchor for the verification sourceUrl. The catalog
+        # records the DataMapper dataset a code belongs to (WEO, FM, GDD,
+        # AFRREO, ...) in the raw payload and in the `category` column; the
+        # legacy hardcoded "@WEO" mislabeled every non-WEO code. Prefer the
+        # payload's own dataset id, then the catalog category. If neither is
+        # derivable, the URL below links the generic indicator page instead of
+        # asserting a wrong dataset.
+        dataset_anchor = ""
         try:
             from ..services.indicator_database import get_indicator_lookup
             _imf_meta = get_indicator_lookup().get("IMF", indicator_code)
             if _imf_meta:
                 catalog_unit = str(_imf_meta.get("unit") or "").strip()
+                raw_meta = _imf_meta.get("raw_metadata")
+                if raw_meta:
+                    try:
+                        dataset_anchor = str((json.loads(raw_meta) or {}).get("dataset") or "").strip()
+                    except (ValueError, TypeError):
+                        dataset_anchor = ""
+                if not dataset_anchor:
+                    category = str(_imf_meta.get("category") or "").strip()
+                    if category and category.upper() != "INDICATOR":
+                        dataset_anchor = category
         except Exception:
             catalog_unit = ""
+            dataset_anchor = ""
 
         # Process each requested country
         results = []
@@ -771,9 +790,16 @@ async def fetch_batch_indicator(
             # see the FRED provider note on the removed value-sniffing
             # percent normalizer (it corrupted legitimately small series).
 
-            # Human-readable URL for data verification on IMF DataMapper website
-            # Format: https://www.imf.org/external/datamapper/{INDICATOR_CODE}@WEO/{COUNTRY}
-            source_url = f"https://www.imf.org/external/datamapper/{indicator_code}@WEO/{country_code}"
+            # Human-readable URL for data verification on IMF DataMapper website.
+            # Format with a known dataset:
+            #   https://www.imf.org/external/datamapper/{CODE}@{DATASET}/{COUNTRY}
+            # When the dataset is not derivable from the catalog, link the
+            # dataset-agnostic indicator page (IMF's routing picks the dataset)
+            # rather than asserting a wrong "@WEO" anchor.
+            if dataset_anchor:
+                source_url = f"https://www.imf.org/external/datamapper/{indicator_code}@{dataset_anchor}/{country_code}"
+            else:
+                source_url = f"https://www.imf.org/external/datamapper/{indicator_code}"
 
             # Build country-specific API URL for reproducibility
             # Format: https://www.imf.org/external/datamapper/api/v1/{INDICATOR_CODE}/{COUNTRY}
@@ -1556,34 +1582,66 @@ def _parse_sdmx_structure_specific_xml(
     def _parse_sdmx_csv(
         self,
         response_text: str,
+        dimension_ids: Optional[List[str]] = None,
     ) -> List[tuple[Dict[str, str], List[Dict[str, str]]]]:
-        """Parse IMF SDMX CSV into the same series/observation shape as XML."""
+        """Parse IMF SDMX CSV into the same series/observation shape as XML.
+
+        Series identity is defined by the DSD's structural dimensions only.
+        When the caller supplies the dataflow's ``dimension_ids`` (from
+        ``_get_imf_dataflow_structure``), group strictly by those columns and
+        treat every other column as a per-observation attribute. Without
+        structure metadata, fall back to a conservative exclusion list of the
+        observation/attribute columns the live IMF.STA CSV is known to emit.
+        """
         rows = [
             {str(key): str(value) for key, value in row.items() if key is not None and value is not None}
             for row in csv.DictReader(io.StringIO(str(response_text or "")))
         ]
         grouped: Dict[tuple[tuple[str, str], ...], List[Dict[str, str]]] = {}
         series_dimensions_by_key: Dict[tuple[tuple[str, str], ...], Dict[str, str]] = {}
-        observation_columns = {
+
+        allowed_dimensions: Optional[set[str]] = None
+        if dimension_ids:
+            allowed_dimensions = {
+                str(dim).strip().upper() for dim in dimension_ids if str(dim or "").strip()
+            }
+
+        # Observation values and per-observation attributes are never
+        # series-defining. STATUS/SCALE/DECIMALS_DISPLAYED are the real column
+        # names emitted by the live IMF.STA SDMX 2.1 CSV; the SDMX-standard
+        # OBS_STATUS/UNIT_MULT/DECIMALS names are kept for other flows/versions.
+        attribute_columns = {
             "TIME_PERIOD",
             "OBS_VALUE",
             "OBS_STATUS",
             "OBS_CONF",
             "UNIT_MULT",
             "DECIMALS",
+            "STATUS",
+            "SCALE",
+            "DECIMALS_DISPLAYED",
         }
 
         for row in rows:
             if not row.get("TIME_PERIOD") or "OBS_VALUE" not in row:
                 continue
-            dimensions = {
-                key: value
-                for key, value in row.items()
-                if key not in observation_columns
-                and not key.startswith("OBS_")
-                and key
-                and value != ""
-            }
+            if allowed_dimensions is not None:
+                dimensions = {
+                    key: value
+                    for key, value in row.items()
+                    if key
+                    and str(key).strip().upper() in allowed_dimensions
+                    and value != ""
+                }
+            else:
+                dimensions = {
+                    key: value
+                    for key, value in row.items()
+                    if key
+                    and str(key).strip().upper() not in attribute_columns
+                    and not str(key).startswith("OBS_")
+                    and value != ""
+                }
             key = tuple(sorted(dimensions.items()))
             series_dimensions_by_key.setdefault(key, dimensions)
             grouped.setdefault(key, []).append(row)
@@ -1609,9 +1667,24 @@ async def _fetch_sdmx_exact_indicator_family(
         last_error: Optional[Exception] = None
         indicator_name = indicator_label or indicator_code
 
+        # Accumulate the first successful candidate per requested country so a
+        # multi-country comparison returns every country that resolves, not just
+        # the first one. Candidates are country-major (see
+        # _build_sdmx_series_candidates), so the previous early ``return`` on the
+        # first success silently dropped every country after the first. This
+        # mirrors the DataMapper JSON path, which iterates all requested
+        # countries before returning. Partial coverage is intentional; the
+        # pipeline's coverage-warning machinery discloses the missing countries.
+        results_by_country: Dict[str, List[NormalizedData]] = {}
+
         for candidate in candidates:
             flow = candidate["flow"]
             key = candidate["key"]
+            candidate_country = candidate.get("country") or key.split(".", 1)[0]
+            # A country already satisfied by an earlier candidate (for CPI, a
+            # lower frequency) does not need its remaining candidates attempted.
+            if candidate_country in results_by_country:
+                continue
             url = self._sdmx_data_url(
                 flow=flow,
                 key=key,
@@ -1640,14 +1713,26 @@ async def _fetch_sdmx_exact_indicator_family(
                 response_text = getattr(response, "text", "")
                 content_type = str(getattr(response, "headers", {}).get("content-type", "")).lower()
                 if "csv" in content_type or str(response_text).lstrip().startswith("DATAFLOW,"):
-                    series_payloads = self._parse_sdmx_csv(response_text)
+                    # Series identity must be defined by the DSD's structural
+                    # dimensions only. The live IMF.STA CSV also emits per-obs
+                    # attribute columns (STATUS/SCALE/DECIMALS_DISPLAYED) that,
+                    # if treated as dimensions, fragment a series whose STATUS
+                    # varies mid-history. Pass the real dimension ids so the CSV
+                    # parser groups strictly by them (cached structure lookup).
+                    structure = await self._get_imf_dataflow_structure(flow)
+                    dimension_ids = (
+                        structure.get("dimension_ids")
+                        if isinstance(structure, dict)
+                        else None
+                    )
+                    series_payloads = self._parse_sdmx_csv(response_text, dimension_ids)
                 else:
                     series_payloads = self._parse_sdmx_structure_specific_xml(response_text)
             except Exception as exc:
                 last_error = exc
                 continue
 
-            results: List[NormalizedData] = []
+            candidate_results: List[NormalizedData] = []
             for series_attrs, observations in series_payloads:
                 data_points = [
                     {
@@ -1695,10 +1780,18 @@ async def _fetch_sdmx_exact_indicator_family(
                     startDate=data_points[0]["date"],
                     endDate=data_points[-1]["date"],
                 )
-                results.append(NormalizedData(metadata=metadata, data=data_points))
+                candidate_results.append(NormalizedData(metadata=metadata, data=data_points))
+
+            if candidate_results:
+                results_by_country[candidate_country] = candidate_results
 
-            if results:
-                return results
+        all_results = [
+            series
+            for country_results in results_by_country.values()
+            for series in country_results
+        ]
+        if all_results:
+            return all_results
 
         diagnostic = f" Attempted SDMX keys: {', '.join(attempted)}." if attempted else ""
         error_suffix = f" Last error: {last_error}." if last_error else ""