Phase 4: Data Validation & Type Safety

- CSV validation in currency.py: _validate_currency_csv() (8-column check),
  _parse_currency_dates() (format validation), _parse_currency_types() (type
  conversion with error handling); _get_symbol() now uses all three validators
- SGSCode refactored to @DataClass(frozen=True) with from_code() and
  from_named() classmethods; added _validate_sgs_code() to reject non-positive
  codes; updated _codes() generator to use classmethods + validation
- Added CurrencyTextResult type alias and improved get() overloads to
  distinguish single-symbol (str) vs multi-symbol (Dict[str, str]) text returns
- Added from __future__ import annotations to all 10 bcb/ modules
- Converted all relative imports to absolute imports across the package
- Updated tests to use new SGSCode classmethod constructors

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>

Author: wilsonfreitas

bcb/__init__.py

@@ -1,11 +1,13 @@
-from .exceptions import (
+from __future__ import annotations
+
+from bcb.exceptions import (
     BCBError,
     BCBAPIError,
     CurrencyNotFoundError,
     SGSError,
     ODataError,
 )
-from .odata.api import (
+from bcb.odata.api import (
     ODataAPI,
     Expectativas,
     PTAX,

bcb/currency.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import re
 import threading
 from datetime import date, timedelta
@@ -359,6 +361,93 @@ def _fetch_symbol_response(
     return res
 
 
+def _validate_currency_csv(csv_text: str) -> pd.DataFrame:
+    """Parse and validate currency CSV format.
+
+    Parameters
+    ----------
+    csv_text : str
+        CSV content from BCB API
+
+    Returns
+    -------
+    pd.DataFrame
+        Parsed DataFrame with all columns
+
+    Raises
+    ------
+    BCBAPIError
+        If CSV format is invalid (wrong column count)
+    """
+    df = pd.read_csv(StringIO(csv_text), delimiter=";", header=None, dtype=str)
+
+    # Validate column count
+    if len(df.columns) != 8:
+        raise BCBAPIError(
+            f"Invalid CSV format: expected 8 columns, got {len(df.columns)}",
+            status_code=400,
+        )
+
+    # Assign meaningful names
+    df.columns = ["Date", "_col1", "_col2", "_col3", "bid", "ask", "_col6", "_col7"]
+    return df
+
+
+def _parse_currency_dates(df: pd.DataFrame) -> pd.DataFrame:
+    """Parse and validate date column in currency CSV.
+
+    Parameters
+    ----------
+    df : pd.DataFrame
+        DataFrame with Date column as strings
+
+    Returns
+    -------
+    pd.DataFrame
+        DataFrame with parsed DatetimeIndex
+
+    Raises
+    ------
+    BCBAPIError
+        If date parsing fails
+    """
+    try:
+        df["Date"] = pd.to_datetime(df["Date"], format="%d%m%Y")
+    except ValueError as e:
+        raise BCBAPIError(
+            f"Failed to parse currency date column: {str(e)}", status_code=400
+        )
+    return df
+
+
+def _parse_currency_types(df: pd.DataFrame) -> pd.DataFrame:
+    """Parse and validate data types in currency DataFrame.
+
+    Parameters
+    ----------
+    df : pd.DataFrame
+        DataFrame with mixed types
+
+    Returns
+    -------
+    pd.DataFrame
+        DataFrame with validated types
+
+    Raises
+    ------
+    BCBAPIError
+        If type conversion fails
+    """
+    try:
+        df["bid"] = df["bid"].str.replace(",", ".").astype(np.float64)
+        df["ask"] = df["ask"].str.replace(",", ".").astype(np.float64)
+    except (ValueError, TypeError) as e:
+        raise BCBAPIError(
+            f"Failed to parse currency numeric columns: {str(e)}", status_code=400
+        )
+    return df
+
+
 def _get_symbol(
     symbol: str, start_date: DateInput, end_date: DateInput
 ) -> pd.DataFrame:
@@ -383,18 +472,12 @@ def _get_symbol(
     CurrencyNotFoundError
         If currency not found
     BCBAPIError
-        If API returns error
+        If API returns error or data format is invalid
     """
     res = _fetch_symbol_response(symbol, start_date, end_date)
-    columns = ["Date", "aa", "bb", "cc", "bid", "ask", "dd", "ee"]
-    df = pd.read_csv(
-        StringIO(res.text), delimiter=";", header=None, names=columns, dtype=str
-    )
-    df = df.assign(
-        Date=lambda x: pd.to_datetime(x["Date"], format="%d%m%Y"),
-        bid=lambda x: x["bid"].str.replace(",", ".").astype(np.float64),
-        ask=lambda x: x["ask"].str.replace(",", ".").astype(np.float64),
-    )
+    df = _validate_currency_csv(res.text)
+    df = _parse_currency_dates(df)
+    df = _parse_currency_types(df)
     df1 = df.set_index("Date")
     n = ["bid", "ask"]
     df1 = df1[n]
@@ -431,9 +514,13 @@ def _get_symbol_text(symbol: str, start_date: DateInput, end_date: DateInput) ->
     return res.text
 
 
+# Type alias for text output with multiple symbols
+CurrencyTextResult = Dict[str, str]  # Maps symbol → CSV text
+
+
 @overload
 def get(
-    symbols: Union[str, List[str]],
+    symbols: str,
     start: DateInput,
     end: DateInput,
     side: str = ...,
@@ -444,13 +531,35 @@ def get(
 
 @overload
 def get(
-    symbols: Union[str, List[str]],
+    symbols: List[str],
+    start: DateInput,
+    end: DateInput,
+    side: str = ...,
+    groupby: str = ...,
+    output: Literal["dataframe"] = ...,
+) -> pd.DataFrame: ...
+
+
+@overload
+def get(
+    symbols: str,
+    start: DateInput,
+    end: DateInput,
+    side: str = ...,
+    groupby: str = ...,
+    output: Literal["text"] = ...,
+) -> str: ...
+
+
+@overload
+def get(
+    symbols: List[str],
     start: DateInput,
     end: DateInput,
     side: str = ...,
     groupby: str = ...,
     output: Literal["text"] = ...,
-) -> Union[str, Dict[str, str]]: ...
+) -> CurrencyTextResult: ...
 
 
 def get(

bcb/exceptions.py

@@ -1,3 +1,6 @@
+from __future__ import annotations
+
+
 class BCBError(Exception):
     """Base exception for all python-bcb errors."""
 

bcb/http.py

@@ -1,5 +1,7 @@
 """Shared HTTP client and utilities for python-bcb."""
 
+from __future__ import annotations
+
 from typing import Callable, TypeVar
 
 import httpx

bcb/odata/__init__.py

@@ -1,4 +1,6 @@
-from .api import (
+from __future__ import annotations
+
+from bcb.odata.api import (
     ODataAPI,
     Expectativas,
     PTAX,

bcb/odata/api.py

@@ -1,5 +1,8 @@
+from __future__ import annotations
+
 from typing import Any, Literal, Optional, Union, overload
-from .framework import (
+
+from bcb.odata.framework import (
     ODataEntitySet,
     ODataFunctionImport,
     ODataQuery,

bcb/odata/framework.py

@@ -1,6 +1,9 @@
+from __future__ import annotations
+
 import threading
 from io import BytesIO
 from typing import Any, Optional, Union
+
 from lxml import etree
 import json
 from urllib.parse import quote

bcb/sgs/__init__.py

@@ -1,4 +1,7 @@
+from __future__ import annotations
+
 import json
+from dataclasses import dataclass
 from io import StringIO
 from typing import (
     Dict,
@@ -29,18 +32,58 @@
 """
 
 
+@dataclass(frozen=True)
 class SGSCode:
-    def __init__(self, code: Union[str, int], name: Optional[str] = None) -> None:
-        if name is None:
-            if isinstance(code, int) or isinstance(code, str):
-                self.name = str(code)
-                self.value = int(code)
-        else:
-            self.name = str(name)
-            self.value = int(code)
+    """SGS time series code with optional human-readable name.
+
+    Attributes
+    ----------
+    value : int
+        Numeric SGS code
+    name : str
+        Human-readable name or string representation of code
+    """
+
+    value: int
+    name: str
+
+    @classmethod
+    def from_code(cls, code: int | str) -> "SGSCode":
+        """Create SGSCode from numeric or string code.
+
+        Parameters
+        ----------
+        code : int | str
+            SGS code
+
+        Returns
+        -------
+        SGSCode
+            New instance with name = str(code)
+        """
+        code_int = int(code)
+        return cls(value=code_int, name=str(code_int))
+
+    @classmethod
+    def from_named(cls, code: int | str, name: str) -> "SGSCode":
+        """Create SGSCode with explicit name.
+
+        Parameters
+        ----------
+        code : int | str
+            SGS code
+        name : str
+            Human-readable name
+
+        Returns
+        -------
+        SGSCode
+            New instance with value and name
+        """
+        return cls(value=int(code), name=name)
 
     def __repr__(self) -> str:
-        return f"{self.value} - {self.name}" if self.name else f"{self.value}"
+        return f"{self.value} - {self.name}"
 
 
 SGSCodeInput: TypeAlias = Union[
@@ -52,20 +95,62 @@ def __repr__(self) -> str:
 ]
 
 
+def _validate_sgs_code(code: SGSCode) -> None:
+    """Validate SGSCode value.
+
+    Parameters
+    ----------
+    code : SGSCode
+        Code to validate
+
+    Raises
+    ------
+    ValueError
+        If code value is not positive integer
+    """
+    if code.value <= 0:
+        raise ValueError(f"SGS code must be positive integer, got {code.value}")
+
+
 def _codes(codes: SGSCodeInput) -> Generator[SGSCode, None, None]:
+    """Normalize various SGSCodeInput formats to SGSCode instances.
+
+    Parameters
+    ----------
+    codes : SGSCodeInput
+        Input in various formats: int, str, tuple, list, or mapping
+
+    Yields
+    ------
+    SGSCode
+        Validated SGSCode instances
+
+    Raises
+    ------
+    ValueError
+        If any code is a non-positive integer
+    """
     if isinstance(codes, int) or isinstance(codes, str):
-        yield SGSCode(codes)
+        code_obj = SGSCode.from_code(codes)
+        _validate_sgs_code(code_obj)
+        yield code_obj
     elif isinstance(codes, tuple):
-        yield SGSCode(codes[1], codes[0])
+        code_obj = SGSCode.from_named(codes[1], codes[0])
+        _validate_sgs_code(code_obj)
+        yield code_obj
     elif isinstance(codes, list):
         for cd in codes:
             if isinstance(cd, tuple):
-                yield SGSCode(cd[1], cd[0])
+                code_obj = SGSCode.from_named(cd[1], cd[0])
             else:
-                yield SGSCode(cd)
+                code_obj = SGSCode.from_code(cd)
+            _validate_sgs_code(code_obj)
+            yield code_obj
     elif isinstance(codes, Mapping):
         for name, code in codes.items():
-            yield SGSCode(code, name)
+            code_obj = SGSCode.from_named(code, name)
+            _validate_sgs_code(code_obj)
+            yield code_obj
 
 
 def _get_url_and_payload(

bcb/sgs/regional_economy.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from typing import Dict, List, Optional, Union
 
 import pandas as pd

bcb/utils.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from datetime import date, datetime
 from typing import TypeAlias, Union