diff --git a/alphaforge/pit/__init__.py b/alphaforge/pit/__init__.py index 4e0fdec..6774a44 100644 --- a/alphaforge/pit/__init__.py +++ b/alphaforge/pit/__init__.py @@ -1,14 +1,20 @@ from .accessor import PITAccessor, ensure_pit_table +from .adapters import PITAdapter +from .catalog import SeriesCatalog from .contract import PIT_CONTRACT_VERSION, PITContractVersion, get_pit_contract_version from .exceptions import ( + PITAdapterError, PITCausalityError, PITContractError, PITEngineError, PITError, PITExperimentalFeatureError, + PITNotSupportedError, PITUnsupportedOperationError, PITValidationError, PITValidationWarning, + SourceFetchError, + VintageNotFoundError, ) from .gdp import ( GDPC1_QOQ_SAAR_RF_REL_GRAPH_ID, @@ -35,6 +41,13 @@ coerce_pit_tape_spec, coerce_snapshot_series_spec, ) +from .observation import ( + PITMode, + PITObservation, + SeriesMetadata, + create_pit_dataframe, + create_wide_view, +) from .pipelines import ( PITPipelineResult, PITPipelineSpec, @@ -48,6 +61,18 @@ RealtimeResolver, VintageResolver, ) +from .target import ( + TargetPolicy, + get_quarterly_release_observation_stream, + list_quarterly_target_releases_asof, + list_quarterly_target_releases_asof_multi, + quarter_end_date, + quarter_obs_date, + quarter_start_date, + resolve_quarterly_final_target, + resolve_target_from_releases, + resolve_target_obs_date_anchor, +) from .tasks import ( build_snapshot_tape, first_vintage_snapshot, @@ -79,6 +104,17 @@ "PITCausalityError", "PITEngineError", "PITValidationWarning", + "PITAdapterError", + "PITNotSupportedError", + "VintageNotFoundError", + "SourceFetchError", + "PITMode", + "PITObservation", + "SeriesMetadata", + "create_pit_dataframe", + "create_wide_view", + "PITAdapter", + "SeriesCatalog", "PITTransformSpec", "PITTransformResult", "PITPipelineStep", @@ -134,4 +170,14 @@ "RealtimeResolver", "LatestResolver", "FrozenResolver", + "TargetPolicy", + "quarter_end_date", + "quarter_start_date", + "quarter_obs_date", + "resolve_target_obs_date_anchor", + "list_quarterly_target_releases_asof", + "list_quarterly_target_releases_asof_multi", + "resolve_target_from_releases", + "resolve_quarterly_final_target", + "get_quarterly_release_observation_stream", ] diff --git a/alphaforge/pit/adapters/__init__.py b/alphaforge/pit/adapters/__init__.py new file mode 100644 index 0000000..1efa52f --- /dev/null +++ b/alphaforge/pit/adapters/__init__.py @@ -0,0 +1,5 @@ +"""PIT data source adapters.""" + +from alphaforge.pit.adapters.base import PITAdapter + +__all__ = ["PITAdapter"] diff --git a/alphaforge/pit/adapters/alphaforge_adapter.py b/alphaforge/pit/adapters/alphaforge_adapter.py new file mode 100644 index 0000000..95b5f82 --- /dev/null +++ b/alphaforge/pit/adapters/alphaforge_adapter.py @@ -0,0 +1,272 @@ +"""Adapter for fetching data from AlphaForge PIT storage.""" + +from __future__ import annotations + +from collections.abc import Iterable +from datetime import date, datetime +from typing import Any + +import pandas as pd + +from alphaforge.data.context import DataContext +from alphaforge.data.query import Query +from alphaforge.pit.adapters.alphaforge_layer import AlphaForgePITLayer +from alphaforge.pit.adapters.base import PITAdapter +from alphaforge.pit.observation import PITObservation, SeriesMetadata +from alphaforge.time.ref_period import RefFreq, RefPeriod + + +def _normalize_obs_date_key(value: object) -> date: + """Normalize AlphaForge obs_date timestamps to canonical date keys. + + AlphaForge may return obs_date timestamps with non-midnight UTC times. + We round by adding 12 hours before taking the date to recover period-end keys. + """ + if isinstance(value, date) and not isinstance(value, datetime): + return value + + ts = pd.Timestamp(value) + if pd.isna(ts): + raise ValueError("obs_date is missing or NaT") + if ts.tz is None: + ts = ts.tz_localize("UTC") + else: + ts = ts.tz_convert("UTC") + if ts != ts.normalize(): + ts = ts + pd.Timedelta(hours=12) + return ts.normalize().date() + + +class AlphaForgePITAdapter(PITAdapter): + """Point-in-time data adapter for AlphaForge.""" + + def __init__(self, ctx: DataContext) -> None: + self._ctx = ctx + self._layer = AlphaForgePITLayer(ctx) + + @property + def name(self) -> str: + return "alphaforge" + + def supports_pit(self, series_id: str) -> bool: + return True + + def list_vintages(self, query_series_key: str) -> list[date]: + conn = self._ctx.pit.conn + rows = conn.execute( + "SELECT DISTINCT asof_utc FROM pit_observations WHERE series_key = ?", + [query_series_key], + ).fetchall() + if not rows: + return [] + return sorted( + {pd.Timestamp(row[0], tz="UTC").date() for row in rows if row[0] is not None} + ) + + def list_pit_observations_asof( + self, + *, + series_key: str, + obs_date: date, + asof_date: date, + ) -> pd.DataFrame: + return self._layer.list_pit_observations_asof( + series_key=series_key, + obs_date=obs_date, + asof_date=asof_date, + ) + + def list_pit_observations_asof_multi(self, requests: pd.DataFrame) -> pd.DataFrame: + return self._layer.list_pit_observations_asof_multi(requests) + + def fetch_asof( + self, + series_id: str, + asof_date: date, + start: date | None = None, + end: date | None = None, + *, + metadata: SeriesMetadata | None = None, + ingest_from_ctx_source: bool = True, + **kwargs: Any, + ) -> list[PITObservation]: + source_series_id = metadata.source_series_id if metadata else series_id + query_series_key = metadata.series_key if metadata else series_id + + asof_ts = pd.Timestamp(asof_date, tz="UTC") + start_ts = pd.Timestamp(start, tz="UTC") if start else None + end_ts = pd.Timestamp(end, tz="UTC") if end else None + + if ingest_from_ctx_source and "fred" in self._ctx.sources: + query = Query( + table="fred_series", + columns=["value"], + entities=[source_series_id], + start=start_ts, + end=end_ts, + asof=asof_ts, + ) + panel = self._ctx.fetch_panel("fred", query) + panel_df = panel.df.reset_index() + required = {"entity_id", "ts_utc", "asof_utc", "value"} + missing = required - set(panel_df.columns) + if missing: + raise ValueError( + f"Unexpected alphaforge panel schema; missing {sorted(missing)}" + ) + + if metadata is None: + series_key_values = panel_df["entity_id"] + else: + series_key_values = [query_series_key] * len(panel_df) + pit_df = pd.DataFrame( + { + "series_key": series_key_values, + "obs_date": pd.to_datetime(panel_df["ts_utc"], utc=True).dt.floor("D"), + "asof_utc": pd.to_datetime(panel_df["asof_utc"], utc=True), + "value": panel_df["value"], + "source": pd.NA, + "revision_id": pd.NA, + "meta_json": pd.NA, + "release_time_utc": pd.NaT, + } + ) + self._ctx.pit.upsert_pit_observations(pit_df) + + snap = self._layer.snapshot(query_series_key, asof=asof_ts, start=start_ts, end=end_ts) + + observations = [] + series_key = metadata.series_key if metadata else series_id + source_series_id = metadata.source_series_id if metadata else series_id + frequency = metadata.frequency if metadata else "" + source = metadata.source if metadata else "alphaforge" + for obs_date_val, value in snap.items(): + obs = PITObservation( + series_key=series_key, + source=source, + source_series_id=source_series_id, + asof_date=asof_date, + vintage_date=asof_date, + obs_date=_normalize_obs_date_key(obs_date_val), + value=float(value), + frequency=frequency, + ) + observations.append(obs) + return observations + + def fetch_asof_many( + self, + *, + series_keys: Iterable[str], + asof_date: date, + start: date | None = None, + end: date | None = None, + metadata_by_key: dict[str, SeriesMetadata] | None = None, + ingest_from_ctx_source: bool = True, + ) -> dict[str, list[PITObservation]]: + """Fetch multiple series snapshots at once using a single PIT query.""" + keys = [str(key) for key in series_keys] + if not keys: + return {} + + if ingest_from_ctx_source: + return { + key: self.fetch_asof( + key, + asof_date, + start=start, + end=end, + metadata=(metadata_by_key or {}).get(key), + ingest_from_ctx_source=ingest_from_ctx_source, + ) + for key in keys + } + + asof_ts = pd.Timestamp(asof_date, tz="UTC") + start_ts = pd.Timestamp(start, tz="UTC") if start else None + end_ts = pd.Timestamp(end, tz="UTC") if end else None + batch = self._layer.snapshot_multi(keys, asof=asof_ts, start=start_ts, end=end_ts) + grouped: dict[str, list[PITObservation]] = {key: [] for key in keys} + + if batch.empty: + return grouped + + metadata_map = metadata_by_key or {} + for row in batch.itertuples(index=False): + query_series_key = str(row.series_key) + value = row.value + obs_date_val = _normalize_obs_date_key(row.obs_date) + if pd.isna(value): + continue + + meta = metadata_map.get(query_series_key) + source_series_id = meta.source_series_id if meta else query_series_key + frequency = meta.frequency if meta else "" + source = meta.source if meta else "alphaforge" + obs = PITObservation( + series_key=query_series_key, + source=source, + source_series_id=source_series_id, + asof_date=asof_date, + vintage_date=asof_date, + obs_date=obs_date_val, + value=float(value), + frequency=frequency, + ) + grouped.setdefault(query_series_key, []).append(obs) + + return grouped + + def fetch_asof_ref( + self, + series_id: str, + asof_date: date, + start_ref: str | RefPeriod | None = None, + end_ref: str | RefPeriod | None = None, + *, + freq: RefFreq | None = None, + metadata: SeriesMetadata | None = None, + ) -> list[PITObservation]: + query_series_key = metadata.series_key if metadata else series_id + asof_ts = pd.Timestamp(asof_date, tz="UTC") + snap = self._layer.snapshot_ref( + query_series_key, asof=asof_ts, start_ref=start_ref, end_ref=end_ref, freq=freq + ) + observations = [] + series_key = metadata.series_key if metadata else series_id + source_series_id = metadata.source_series_id if metadata else series_id + source = metadata.source if metadata else "alphaforge" + frequency = metadata.frequency if metadata else (freq.value if freq else "") + for obs_date_val, value in snap.items(): + obs = PITObservation( + series_key=series_key, + source=source, + source_series_id=source_series_id, + asof_date=asof_date, + vintage_date=asof_date, + obs_date=_normalize_obs_date_key(obs_date_val), + value=float(value), + frequency=frequency, + ) + observations.append(obs) + return observations + + def fetch_revisions_ref( + self, + series_id: str, + ref: str | RefPeriod, + start_asof: date | None = None, + end_asof: date | None = None, + *, + freq: RefFreq | None = None, + metadata: SeriesMetadata | None = None, + ) -> pd.Series: + start_ts = pd.Timestamp(start_asof, tz="UTC") if start_asof else None + end_ts = pd.Timestamp(end_asof, tz="UTC") if end_asof else None + query_series_key = metadata.series_key if metadata else series_id + series = self._layer.revisions_ref( + query_series_key, ref, start_asof=start_ts, end_asof=end_ts, freq=freq + ) + if metadata is not None: + series = series.rename(metadata.series_key) + return series diff --git a/alphaforge/pit/adapters/alphaforge_layer.py b/alphaforge/pit/adapters/alphaforge_layer.py new file mode 100644 index 0000000..c144a85 --- /dev/null +++ b/alphaforge/pit/adapters/alphaforge_layer.py @@ -0,0 +1,246 @@ +"""Thin adapter wrapper around AlphaForge PIT APIs.""" + +from __future__ import annotations + +from collections.abc import Iterable +from datetime import date +from typing import Any + +import pandas as pd + +from alphaforge.data.context import DataContext +from alphaforge.pit.transforms import PITTransformResult, PITTransformSpec +from alphaforge.pit.utils.timestamps import coerce_utc_timestamp, normalize_utc_day +from alphaforge.time.ref_period import RefFreq, RefPeriod + + +class AlphaForgePITLayer: + """Wrapper for AlphaForge PIT accessors.""" + + def __init__(self, ctx: DataContext) -> None: + if ctx.pit is None: + raise ValueError("PIT requires DuckDBParquetStore-backed DataContext") + self._ctx = ctx + + def snapshot( + self, + series_key: str, + asof: pd.Timestamp, + start: pd.Timestamp | None = None, + end: pd.Timestamp | None = None, + ) -> pd.Series: + return self._ctx.pit.get_snapshot(series_key, asof=asof, start=start, end=end) + + def snapshot_multi( + self, + series_keys: Iterable[str], + *, + asof: pd.Timestamp, + start: pd.Timestamp | None = None, + end: pd.Timestamp | None = None, + ) -> pd.DataFrame: + if self._ctx.pit is None: + raise ValueError("PIT store is not available; cannot query snapshots.") + return self._ctx.pit.get_snapshot_multi( + list(series_keys), + asof=asof, + start=start, + end=end, + ) + + def snapshot_ref( + self, + series_key: str, + asof: pd.Timestamp, + start_ref: str | RefPeriod | None = None, + end_ref: str | RefPeriod | None = None, + *, + freq: RefFreq | None = None, + ) -> pd.Series: + if isinstance(start_ref, str): + start_ref = RefPeriod.parse(start_ref) + if isinstance(end_ref, str): + end_ref = RefPeriod.parse(end_ref) + return self._ctx.pit.get_snapshot_ref( + series_key, + asof=asof, + start_ref=start_ref, + end_ref=end_ref, + freq=freq, + ) + + def revisions( + self, + series_key: str, + obs_date: pd.Timestamp, + start_asof: pd.Timestamp | None = None, + end_asof: pd.Timestamp | None = None, + ) -> pd.Series: + return self._ctx.pit.get_revision_timeline( + series_key, obs_date=obs_date, start_asof=start_asof, end_asof=end_asof + ) + + def revision_path( + self, + series_key: str, + obs_date: pd.Timestamp, + start_asof: pd.Timestamp | None = None, + end_asof: pd.Timestamp | None = None, + ) -> pd.DataFrame: + return self._ctx.pit.get_revision_path( + series_key, obs_date=obs_date, start_asof=start_asof, end_asof=end_asof + ) + + def revision_path_multi(self, requests: pd.DataFrame) -> pd.DataFrame: + return self._ctx.pit.get_revision_path_multi(requests) + + def revisions_ref( + self, + series_key: str, + ref: str | RefPeriod, + start_asof: pd.Timestamp | None = None, + end_asof: pd.Timestamp | None = None, + *, + freq: RefFreq | None = None, + ) -> pd.Series: + if isinstance(ref, str): + ref = RefPeriod.parse(ref) + return self._ctx.pit.get_revision_timeline_ref( + series_key, ref=ref, start_asof=start_asof, end_asof=end_asof, freq=freq + ) + + def upsert(self, df: pd.DataFrame) -> None: + self._ctx.pit.upsert_pit_observations(df) + + def apply_transform( + self, + spec: PITTransformSpec | dict[str, Any], + *, + overwrite: bool = False, + persist: bool = True, + allow_experimental: bool = False, + on_engine_mismatch: str = "error", + ) -> PITTransformResult: + return self._ctx.pit.apply_transform( + spec, + overwrite=overwrite, + persist=persist, + allow_experimental=allow_experimental, + on_engine_mismatch=on_engine_mismatch, + ) + + def explain_transform( + self, + spec: PITTransformSpec | dict[str, Any], + *, + allow_experimental: bool = False, + on_engine_mismatch: str = "error", + ) -> dict[str, Any]: + return self._ctx.pit.explain_transform( + spec, + allow_experimental=allow_experimental, + on_engine_mismatch=on_engine_mismatch, + ) + + def list_pit_observations_asof( + self, + *, + series_key: str, + obs_date: date, + asof_date: date, + ) -> pd.DataFrame: + if self._ctx.pit is None: + raise ValueError("PIT store is not available; cannot list PIT observations.") + conn = self._ctx.pit.conn + obs_day = normalize_utc_day(obs_date) + asof_cutoff = coerce_utc_timestamp(asof_date).normalize() + pd.Timedelta(days=1) + df = conn.execute( + """ + SELECT series_key, obs_date, asof_utc, value + FROM pit_observations + WHERE series_key = ? + AND DATE(obs_date) = ? + AND asof_utc < ? + ORDER BY asof_utc ASC + """, + [series_key, obs_day, asof_cutoff], + ).fetchdf() + if df.empty: + return pd.DataFrame( + { + "series_key": pd.Series(dtype="object"), + "obs_date": pd.Series(dtype="datetime64[ns, UTC]"), + "asof_utc": pd.Series(dtype="datetime64[ns, UTC]"), + "value": pd.Series(dtype="float64"), + } + ) + + df["obs_date"] = pd.to_datetime(df["obs_date"], utc=True).dt.floor("D") + df["asof_utc"] = pd.to_datetime(df["asof_utc"], utc=True) + return df + + def list_pit_observations_asof_multi(self, requests: pd.DataFrame) -> pd.DataFrame: + """Batch list PIT observations for multiple (series_key, obs_date, asof_date) requests.""" + required = {"request_id", "series_key", "obs_date", "asof_date"} + missing = required - set(requests.columns) + if missing: + raise ValueError(f"Missing required request columns: {sorted(missing)}") + + if self._ctx.pit is None: + raise ValueError("PIT store is not available; cannot list PIT observations.") + + _empty = pd.DataFrame( + { + "request_id": pd.Series(dtype="object"), + "series_key": pd.Series(dtype="object"), + "obs_date": pd.Series(dtype="datetime64[ns, UTC]"), + "asof_utc": pd.Series(dtype="datetime64[ns, UTC]"), + "value": pd.Series(dtype="float64"), + } + ) + + if requests.empty: + return _empty + + req = requests.loc[:, ["request_id", "series_key", "obs_date", "asof_date"]].copy() + req["obs_day"] = pd.to_datetime(req["obs_date"], utc=True, errors="coerce").dt.date + req["asof_cutoff"] = ( + pd.to_datetime(req["asof_date"], utc=True, errors="coerce").dt.normalize() + + pd.Timedelta(days=1) + ).dt.tz_localize(None) + req = req.dropna(subset=["obs_day", "asof_cutoff", "series_key", "request_id"]) + if req.empty: + return _empty + + conn = self._ctx.pit.conn + conn.register("pit_asof_requests", req) + try: + df = conn.execute( + """ + SELECT + r.request_id, + p.series_key, + p.obs_date, + p.asof_utc, + p.value + FROM pit_asof_requests r + LEFT JOIN pit_observations p + ON p.series_key = r.series_key + AND DATE(p.obs_date) = r.obs_day + AND p.asof_utc < r.asof_cutoff + ORDER BY r.request_id, p.asof_utc + """ + ).fetchdf() + finally: + conn.unregister("pit_asof_requests") + + if df.empty: + return _empty + + df = df[df["series_key"].notna()].copy() + if df.empty: + return _empty + df["obs_date"] = pd.to_datetime(df["obs_date"], utc=True).dt.floor("D") + df["asof_utc"] = pd.to_datetime(df["asof_utc"], utc=True) + df["value"] = pd.to_numeric(df["value"], errors="coerce") + return df.reset_index(drop=True) diff --git a/alphaforge/pit/adapters/base.py b/alphaforge/pit/adapters/base.py new file mode 100644 index 0000000..9c6a06e --- /dev/null +++ b/alphaforge/pit/adapters/base.py @@ -0,0 +1,142 @@ +"""Base adapter interface for PIT data sources.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from datetime import date +from typing import TYPE_CHECKING, Any + +import pandas as pd + +from alphaforge.pit.observation import PITObservation, SeriesMetadata + +if TYPE_CHECKING: + from alphaforge.time.ref_period import RefFreq, RefPeriod + + +class PITAdapter(ABC): + """Base class for point-in-time data adapters. + + Concrete subclasses fetch vintage data from external sources (FRED/ALFRED, + ECB, BoE, etc.) and return :class:`PITObservation` records. + """ + + @property + @abstractmethod + def name(self) -> str: + """Adapter name/identifier.""" + + @abstractmethod + def supports_pit(self, series_id: str) -> bool: + """Check if a series supports point-in-time retrieval. + + Args: + series_id: Source-specific series identifier. + + Returns: + True if PIT is supported, False otherwise. + """ + + @abstractmethod + def list_vintages(self, series_id: str) -> list[date]: + """List available vintage dates for a series. + + Args: + series_id: Source-specific series identifier. + + Returns: + Sorted list of vintage dates. + + Raises: + PITNotSupportedError: If series doesn't support vintages. + SourceFetchError: If fetching fails. + """ + + @abstractmethod + def fetch_asof( + self, + series_id: str, + asof_date: date, + start: date | None = None, + end: date | None = None, + *, + metadata: SeriesMetadata | None = None, + **kwargs: Any, + ) -> list[PITObservation]: + """Fetch observations as they were known on *asof_date*. + + Args: + series_id: Source-specific series identifier. + asof_date: Point-in-time evaluation date. + start: Optional start date for observation period. + end: Optional end date for observation period. + metadata: Optional series metadata. + + Returns: + List of PIT observations. + + Raises: + PITNotSupportedError: If series doesn't support PIT. + VintageNotFoundError: If no vintage available at asof_date. + SourceFetchError: If fetching fails. + """ + + def fetch_asof_ref( + self, + series_id: str, + asof_date: date, + start_ref: str | RefPeriod | None = None, + end_ref: str | RefPeriod | None = None, + *, + freq: RefFreq | None = None, + metadata: SeriesMetadata | None = None, + ) -> list[PITObservation]: + """Optional ref-period snapshot query.""" + raise NotImplementedError("Ref-period snapshot queries not supported.") + + def fetch_revisions_ref( + self, + series_id: str, + ref: str | RefPeriod, + start_asof: date | None = None, + end_asof: date | None = None, + *, + freq: RefFreq | None = None, + metadata: SeriesMetadata | None = None, + ) -> pd.Series: + """Optional ref-period revision timeline query.""" + raise NotImplementedError("Ref-period revision queries not supported.") + + def fetch_vintage( + self, + series_id: str, + vintage_date: date, + start: date | None = None, + end: date | None = None, + ) -> list[PITObservation]: + """Fetch a specific vintage of observations. + + Default implementation delegates to :meth:`fetch_asof`. + """ + return self.fetch_asof(series_id, vintage_date, start, end) + + def list_pit_observations_asof( + self, + *, + series_key: str, + obs_date: date, + asof_date: date, + ) -> pd.DataFrame: + """List all PIT observations for a series/obs_date up to an as-of date. + + Returns: + DataFrame with columns: series_key, obs_date, asof_utc, value. + """ + raise NotImplementedError("PIT observation listing not supported.") + + def list_pit_observations_asof_multi(self, requests: pd.DataFrame) -> pd.DataFrame: + """Optional batched PIT observation listing. + + Expected request columns: request_id, series_key, obs_date, asof_date. + """ + raise NotImplementedError("Batched PIT observation listing not supported.") diff --git a/alphaforge/pit/adapters/boe.py b/alphaforge/pit/adapters/boe.py new file mode 100644 index 0000000..0beb02d --- /dev/null +++ b/alphaforge/pit/adapters/boe.py @@ -0,0 +1,39 @@ +"""Bank of England Real-Time Database adapter for UK data.""" + +from __future__ import annotations + +from datetime import date +from typing import Any + +from alphaforge.pit.adapters.base import PITAdapter +from alphaforge.pit.exceptions import PITNotSupportedError +from alphaforge.pit.observation import PITObservation, SeriesMetadata + + +class BOERTDBAdapter(PITAdapter): + """Adapter for Bank of England Real-Time Database (UK GDP). + + Stub implementation — to be completed with actual BoE RTDB integration. + """ + + @property + def name(self) -> str: + return "BOE_RTDB" + + def supports_pit(self, series_id: str) -> bool: + return False + + def list_vintages(self, series_id: str) -> list[date]: + raise PITNotSupportedError(series_id, "BoE RTDB adapter not yet implemented") + + def fetch_asof( + self, + series_id: str, + asof_date: date, + start: date | None = None, + end: date | None = None, + *, + metadata: SeriesMetadata | None = None, + **kwargs: Any, + ) -> list[PITObservation]: + raise PITNotSupportedError(series_id, "BoE RTDB adapter not yet implemented") diff --git a/alphaforge/pit/adapters/ecb.py b/alphaforge/pit/adapters/ecb.py new file mode 100644 index 0000000..603b131 --- /dev/null +++ b/alphaforge/pit/adapters/ecb.py @@ -0,0 +1,39 @@ +"""ECB Real-Time Database adapter for Euro Area data.""" + +from __future__ import annotations + +from datetime import date +from typing import Any + +from alphaforge.pit.adapters.base import PITAdapter +from alphaforge.pit.exceptions import PITNotSupportedError +from alphaforge.pit.observation import PITObservation, SeriesMetadata + + +class ECBRTDBAdapter(PITAdapter): + """Adapter for ECB Real-Time Database (Euro Area). + + Stub implementation — to be completed with actual ECB RTDB integration. + """ + + @property + def name(self) -> str: + return "ECB_RTDB" + + def supports_pit(self, series_id: str) -> bool: + return False + + def list_vintages(self, series_id: str) -> list[date]: + raise PITNotSupportedError(series_id, "ECB RTDB adapter not yet implemented") + + def fetch_asof( + self, + series_id: str, + asof_date: date, + start: date | None = None, + end: date | None = None, + *, + metadata: SeriesMetadata | None = None, + **kwargs: Any, + ) -> list[PITObservation]: + raise PITNotSupportedError(series_id, "ECB RTDB adapter not yet implemented") diff --git a/alphaforge/pit/adapters/fred.py b/alphaforge/pit/adapters/fred.py new file mode 100644 index 0000000..57350e8 --- /dev/null +++ b/alphaforge/pit/adapters/fred.py @@ -0,0 +1,167 @@ +"""FRED/ALFRED adapter for US macroeconomic data. + +Requires ``requests`` (install via ``pip install alphaforge[fred]``). +""" + +from __future__ import annotations + +import os +import time +from datetime import date, datetime +from typing import Any + +import requests + +from alphaforge.pit.adapters.base import PITAdapter +from alphaforge.pit.exceptions import PITNotSupportedError, SourceFetchError, VintageNotFoundError +from alphaforge.pit.observation import PITObservation, SeriesMetadata +from alphaforge.pit.vintage import select_vintage_for_asof + + +class FREDALFREDAdapter(PITAdapter): + """Adapter for FRED/ALFRED (Federal Reserve Economic Data). + + Uses the FRED API to retrieve point-in-time data using realtime_start/realtime_end. + """ + + BASE_URL = "https://api.stlouisfed.org/fred" + + def __init__(self, api_key: str | None = None) -> None: + """Initialize FRED adapter. + + Args: + api_key: FRED API key. If None, reads from FRED_API_KEY env var. + """ + self._api_key = api_key or os.getenv("FRED_API_KEY") + if not self._api_key: + raise ValueError( + "FRED API key required. Set FRED_API_KEY environment variable " + "or pass api_key parameter." + ) + self._session = requests.Session() + self._vintage_cache: dict[str, tuple[list[date], float]] = {} + self._cache_ttl = 86400 # 24 hours + + @property + def name(self) -> str: + return "FRED_ALFRED" + + def supports_pit(self, series_id: str) -> bool: + try: + vintages = self.list_vintages(series_id) + return len(vintages) > 0 + except Exception: + return False + + def list_vintages(self, series_id: str) -> list[date]: + cache_key = series_id + if cache_key in self._vintage_cache: + cached_data, cached_time = self._vintage_cache[cache_key] + if time.time() - cached_time < self._cache_ttl: + return cached_data + + url = f"{self.BASE_URL}/series/vintagedates" + params = { + "series_id": series_id, + "api_key": self._api_key, + "file_type": "json", + } + + try: + response = self._request_with_retry(url, params) + data = response.json() + vintage_dates_str = data.get("vintage_dates", []) + if not vintage_dates_str: + vintages: list[date] = [] + else: + vintages = [ + datetime.strptime(v, "%Y-%m-%d").date() for v in vintage_dates_str + ] + self._vintage_cache[cache_key] = (vintages, time.time()) + return vintages + except Exception as e: + raise SourceFetchError("FRED_ALFRED", original_error=e) from e + + def fetch_asof( + self, + series_id: str, + asof_date: date, + start: date | None = None, + end: date | None = None, + *, + metadata: SeriesMetadata | None = None, + **kwargs: Any, + ) -> list[PITObservation]: + vintages = self.list_vintages(series_id) + if not vintages: + raise PITNotSupportedError(series_id, "No vintages available (non-revised series)") + + selected_vintage = select_vintage_for_asof(vintages, asof_date) + if selected_vintage is None: + raise VintageNotFoundError(series_id, asof_date) + + url = f"{self.BASE_URL}/series/observations" + params: dict[str, str] = { + "series_id": series_id, + "api_key": self._api_key, + "file_type": "json", + "realtime_start": asof_date.strftime("%Y-%m-%d"), + "realtime_end": asof_date.strftime("%Y-%m-%d"), + } + if start: + params["observation_start"] = start.strftime("%Y-%m-%d") + if end: + params["observation_end"] = end.strftime("%Y-%m-%d") + + try: + response = self._request_with_retry(url, params) + data = response.json() + + observations: list[PITObservation] = [] + for obs_dict in data.get("observations", []): + value_str = obs_dict["value"] + if value_str == ".": + continue + try: + value = float(value_str) + except (ValueError, TypeError): + continue + + obs = PITObservation( + series_key=series_id, + source="FRED_ALFRED", + source_series_id=series_id, + asof_date=asof_date, + vintage_date=asof_date, + obs_date=datetime.strptime(obs_dict["date"], "%Y-%m-%d").date(), + value=value, + value_raw=value_str, + frequency=self._infer_frequency(obs_dict["date"]), + realtime_start=asof_date, + realtime_end=asof_date, + ingested_at=datetime.utcnow(), + provenance={"realtime_start": obs_dict.get("realtime_start")}, + ) + observations.append(obs) + return observations + except SourceFetchError: + raise + except Exception as e: + raise SourceFetchError("FRED_ALFRED", original_error=e) from e + + def _request_with_retry( + self, url: str, params: dict[str, str], max_retries: int = 3 + ) -> requests.Response: + for attempt in range(max_retries): + try: + response = self._session.get(url, params=params, timeout=30) + response.raise_for_status() + return response + except requests.exceptions.RequestException: + if attempt == max_retries - 1: + raise + time.sleep(2**attempt) + raise SourceFetchError("FRED_ALFRED", "Max retries exceeded") + + def _infer_frequency(self, date_str: str) -> str: + return "M" # Default to monthly diff --git a/alphaforge/pit/adapters/statcan.py b/alphaforge/pit/adapters/statcan.py new file mode 100644 index 0000000..adef1a2 --- /dev/null +++ b/alphaforge/pit/adapters/statcan.py @@ -0,0 +1,39 @@ +"""Statistics Canada Real-Time Database adapter for Canadian data.""" + +from __future__ import annotations + +from datetime import date +from typing import Any + +from alphaforge.pit.adapters.base import PITAdapter +from alphaforge.pit.exceptions import PITNotSupportedError +from alphaforge.pit.observation import PITObservation, SeriesMetadata + + +class StatCanRealTimeAdapter(PITAdapter): + """Adapter for Statistics Canada Real-Time Tables. + + Stub implementation — to be completed with actual StatCan integration. + """ + + @property + def name(self) -> str: + return "STATCAN_REALTIME" + + def supports_pit(self, series_id: str) -> bool: + return False + + def list_vintages(self, series_id: str) -> list[date]: + raise PITNotSupportedError(series_id, "StatCan adapter not yet implemented") + + def fetch_asof( + self, + series_id: str, + asof_date: date, + start: date | None = None, + end: date | None = None, + *, + metadata: SeriesMetadata | None = None, + **kwargs: Any, + ) -> list[PITObservation]: + raise PITNotSupportedError(series_id, "StatCan adapter not yet implemented") diff --git a/alphaforge/pit/adapters/swiss.py b/alphaforge/pit/adapters/swiss.py new file mode 100644 index 0000000..2b8ec24 --- /dev/null +++ b/alphaforge/pit/adapters/swiss.py @@ -0,0 +1,39 @@ +"""Switzerland macro data adapter (SECO/SNB).""" + +from __future__ import annotations + +from datetime import date +from typing import Any + +from alphaforge.pit.adapters.base import PITAdapter +from alphaforge.pit.exceptions import PITNotSupportedError +from alphaforge.pit.observation import PITObservation, SeriesMetadata + + +class SwissAdapter(PITAdapter): + """Adapter for Swiss macro data (SECO/SNB). + + Stub implementation — to be completed with actual Swiss data source integration. + """ + + @property + def name(self) -> str: + return "SWISS_SECO" + + def supports_pit(self, series_id: str) -> bool: + return False + + def list_vintages(self, series_id: str) -> list[date]: + raise PITNotSupportedError(series_id, "Swiss adapter not yet implemented") + + def fetch_asof( + self, + series_id: str, + asof_date: date, + start: date | None = None, + end: date | None = None, + *, + metadata: SeriesMetadata | None = None, + **kwargs: Any, + ) -> list[PITObservation]: + raise PITNotSupportedError(series_id, "Swiss adapter not yet implemented") diff --git a/alphaforge/pit/catalog.py b/alphaforge/pit/catalog.py new file mode 100644 index 0000000..f6ef5e9 --- /dev/null +++ b/alphaforge/pit/catalog.py @@ -0,0 +1,105 @@ +"""Series catalog configuration management. + +Parses YAML-based series metadata catalogs that describe macro series, +their PIT modes, and release rules. + +Requires ``pyyaml`` (install via ``pip install alphaforge[catalog]``). +""" + +from __future__ import annotations + +from pathlib import Path + +import yaml + +from alphaforge.pit.observation import SeriesMetadata +from alphaforge.pit.release_rules import ReleaseRule + + +class SeriesCatalog: + """Manages series metadata from configuration.""" + + def __init__(self, config_path: Path | None = None) -> None: + """Initialize catalog from configuration file. + + Args: + config_path: Path to YAML configuration file. If None, starts empty. + """ + self._metadata: dict[str, SeriesMetadata] = {} + if config_path: + self.load(config_path) + + def load(self, config_path: Path) -> None: + """Load series metadata from YAML configuration.""" + with open(config_path) as f: + config = yaml.safe_load(f) + + if not config: + return + + for series_key, series_config in config.items(): + release_rule_raw = series_config.get("release_rule") + release_rule = ( + ReleaseRule.from_dict(release_rule_raw) + if release_rule_raw is not None + else None + ) + + metadata = SeriesMetadata( + series_key=series_key, + country=series_config.get("country", ""), + source=series_config.get("source", ""), + source_series_id=series_config.get("source_series_id", ""), + frequency=series_config.get("frequency", ""), + pit_mode=series_config.get("pit_mode", "NO_PIT"), + seasonal_adjustment=series_config.get("seasonal_adjustment"), + units=series_config.get("units"), + description=series_config.get("description"), + transforms=series_config.get("transforms"), + adapter=series_config.get("adapter"), + obs_date_anchor=series_config.get("obs_date_anchor"), + publication_lag_months=series_config.get("publication_lag_months"), + release_rule=release_rule, + blocks=series_config.get("blocks", {}), + ) + self._metadata[series_key] = metadata + + def get(self, series_key: str) -> SeriesMetadata | None: + """Get metadata for a series.""" + return self._metadata.get(series_key) + + def get_all(self) -> dict[str, SeriesMetadata]: + """Get all series metadata.""" + return self._metadata.copy() + + def add(self, metadata: SeriesMetadata) -> None: + """Add or update series metadata.""" + self._metadata[metadata.series_key] = metadata + + def list_series( + self, country: str | None = None, source: str | None = None + ) -> list[str]: + """List series keys, optionally filtered. + + Args: + country: Filter by country code. + source: Filter by source name. + + Returns: + Sorted list of series keys. + """ + series = [] + for key, meta in self._metadata.items(): + if country and meta.country != country: + continue + if source and meta.source != source: + continue + series.append(key) + return sorted(series) + + def supports_pit(self, series_key: str) -> bool: + """Check if a series supports PIT retrieval.""" + meta = self.get(series_key) + if not meta: + return False + return meta.pit_mode != "NO_PIT" diff --git a/alphaforge/pit/exceptions.py b/alphaforge/pit/exceptions.py index 2b8ee20..ea9403e 100644 --- a/alphaforge/pit/exceptions.py +++ b/alphaforge/pit/exceptions.py @@ -31,3 +31,54 @@ class PITCausalityError(PITValidationError): class PITEngineError(PITError): """Raised when requested transform engine cannot be satisfied.""" + + +# --------------------------------------------------------------------------- +# Adapter-level exceptions +# --------------------------------------------------------------------------- + + +class PITAdapterError(PITError): + """Base exception for PIT adapter errors.""" + + +class PITNotSupportedError(PITAdapterError): + """Raised when a series does not support point-in-time retrieval.""" + + def __init__(self, series_key: str, message: str | None = None): + self.series_key = series_key + if message is None: + message = f"Series '{series_key}' does not support point-in-time retrieval" + super().__init__(message) + + +class VintageNotFoundError(PITAdapterError): + """Raised when no vintage is available for the requested asof date.""" + + def __init__(self, series_key: str, asof_date: object, message: str | None = None): + self.series_key = series_key + self.asof_date = asof_date + if message is None: + message = ( + f"No vintage available for series '{series_key}' " + f"at or before asof_date={asof_date}" + ) + super().__init__(message) + + +class SourceFetchError(PITAdapterError): + """Raised when data fetching from a source fails.""" + + def __init__( + self, + source: str, + message: str | None = None, + original_error: Exception | None = None, + ): + self.source = source + self.original_error = original_error + if message is None: + message = f"Failed to fetch data from source '{source}'" + if original_error: + message += f": {original_error!s}" + super().__init__(message) diff --git a/alphaforge/pit/observation.py b/alphaforge/pit/observation.py new file mode 100644 index 0000000..f6199ad --- /dev/null +++ b/alphaforge/pit/observation.py @@ -0,0 +1,149 @@ +"""Data models for PIT adapter observations and series metadata. + +These types define the adapter-level contract for point-in-time data: + +- :class:`PITObservation` — a single vintage observation from a data source. +- :class:`SeriesMetadata` — metadata describing a macro series and its PIT mode. +- :data:`PITMode` — enumeration of PIT access modes. +- :func:`create_pit_dataframe` — convert observations to a canonical DataFrame. +- :func:`create_wide_view` — pivot a single-asof DataFrame to wide format. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import date, datetime +from typing import TYPE_CHECKING, Any, Literal + +import pandas as pd + +if TYPE_CHECKING: + from alphaforge.pit.release_rules import ReleaseRule + +PITMode = Literal["ALFRED_REALTIME", "DISCRETE_VINTAGES_SNAP", "NO_PIT"] + + +@dataclass +class SeriesMetadata: + """Metadata for a macro series.""" + + series_key: str + country: str + source: str + source_series_id: str + frequency: str + pit_mode: PITMode + seasonal_adjustment: str | None = None + units: str | None = None + description: str | None = None + transforms: list | None = None + adapter: str | None = None + obs_date_anchor: Literal["start", "end"] | None = None + publication_lag_months: int | None = None + release_rule: ReleaseRule | None = None + blocks: dict[str, int] = field(default_factory=dict) + + +@dataclass +class PITObservation: + """A single point-in-time observation.""" + + series_key: str + source: str + source_series_id: str + asof_date: date + vintage_date: date + obs_date: date + value: float + frequency: str + value_raw: str | None = None + units: str | None = None + seasonal_adjustment: str | None = None + realtime_start: date | None = None + realtime_end: date | None = None + ingested_at: datetime | None = None + provenance: dict[str, Any] | None = None + + def to_dict(self) -> dict[str, Any]: + """Convert to dictionary for DataFrame construction.""" + return { + "series_key": self.series_key, + "source": self.source, + "source_series_id": self.source_series_id, + "asof_date": self.asof_date, + "vintage_date": self.vintage_date, + "obs_date": self.obs_date, + "value": self.value, + "value_raw": self.value_raw, + "frequency": self.frequency, + "units": self.units, + "seasonal_adjustment": self.seasonal_adjustment, + "realtime_start": self.realtime_start, + "realtime_end": self.realtime_end, + "ingested_at": self.ingested_at, + "provenance": self.provenance, + } + + +_PIT_DF_COLUMNS = [ + "series_key", + "source", + "source_series_id", + "asof_date", + "vintage_date", + "obs_date", + "value", + "value_raw", + "frequency", + "units", + "seasonal_adjustment", + "realtime_start", + "realtime_end", + "ingested_at", + "provenance", +] + + +def create_pit_dataframe(observations: list[PITObservation]) -> pd.DataFrame: + """Create a canonical PIT DataFrame from observations. + + Schema columns: series_key, source, source_series_id, asof_date, + vintage_date, obs_date, value, value_raw, frequency, units, + seasonal_adjustment, realtime_start, realtime_end, ingested_at, provenance. + """ + if not observations: + return pd.DataFrame(columns=_PIT_DF_COLUMNS) + + data = [obs.to_dict() for obs in observations] + df = pd.DataFrame(data) + + date_cols = ["asof_date", "vintage_date", "obs_date", "realtime_start", "realtime_end"] + for col in date_cols: + if col in df.columns: + df[col] = pd.to_datetime(df[col]) + + if "ingested_at" in df.columns: + df["ingested_at"] = pd.to_datetime(df["ingested_at"]) + + return df + + +def create_wide_view(df: pd.DataFrame) -> pd.DataFrame: + """Create a wide view for modeling from PIT DataFrame. + + Args: + df: PIT DataFrame with single asof_date. + + Returns: + Wide DataFrame with index=obs_date, columns=series_key, values=value. + """ + if df.empty: + return pd.DataFrame() + + asof_dates = df["asof_date"].unique() + if len(asof_dates) > 1: + raise ValueError( + f"Wide view requires single asof_date, got {len(asof_dates)} different dates" + ) + + return df.pivot(index="obs_date", columns="series_key", values="value") diff --git a/alphaforge/pit/resolvers.py b/alphaforge/pit/resolvers.py index 034c027..70ad335 100644 --- a/alphaforge/pit/resolvers.py +++ b/alphaforge/pit/resolvers.py @@ -152,5 +152,9 @@ def resolve( vintages = self._revision_map.get((series_key, obs_date)) if vintages is None: return requested_asof # no revision history → fall through - idx = min(self._n, len(vintages)) - 1 # 0-indexed - return vintages[idx] + # Filter to vintages on or before requested_asof to prevent lookahead. + valid = [v for v in vintages if v <= requested_asof] + if not valid: + return requested_asof + idx = min(self._n, len(valid)) - 1 # 0-indexed + return valid[idx] diff --git a/alphaforge/pit/target.py b/alphaforge/pit/target.py new file mode 100644 index 0000000..307a505 --- /dev/null +++ b/alphaforge/pit/target.py @@ -0,0 +1,416 @@ +"""Target policy — release-selection logic for PIT series. + +Provides :class:`TargetPolicy` and helpers for resolving target values +from point-in-time release streams. The functions are generic and work +with any :class:`~alphaforge.pit.adapters.base.PITAdapter` implementation. +""" + +from __future__ import annotations + +import re +from collections.abc import Iterable +from dataclasses import dataclass +from datetime import date +from typing import TYPE_CHECKING, Literal + +import pandas as pd + +from alphaforge.pit.adapters.base import PITAdapter + +if TYPE_CHECKING: + from alphaforge.pit.catalog import SeriesCatalog + + +# --------------------------------------------------------------------------- +# Dataclass +# --------------------------------------------------------------------------- + + +@dataclass +class TargetPolicy: + """Configuration for selecting a target value from available releases. + + Attributes: + mode: Selection strategy. + + * ``"latest_available"`` — pick the most recent non-null release. + * ``"nth_release"`` — pick exactly the *n*-th non-null release. + * ``"next_release"`` — signal the rank of the next expected release. + + nth: Required when *mode* is ``"nth_release"``. Must be >= 1. + max_release_rank: Optional cap on the number of releases considered. + """ + + mode: Literal["latest_available", "nth_release", "next_release"] + nth: int | None = None + max_release_rank: int | None = None + + +# --------------------------------------------------------------------------- +# Quarter calendar helpers +# --------------------------------------------------------------------------- + + +def _quarter_end_from_string(ref_quarter: str) -> date: + """Parse ``YYYYQn`` and return the calendar quarter-end date.""" + match = re.match(r"^(\d{4})[Qq](\d+)$", str(ref_quarter).strip()) + if not match: + raise ValueError(f"Expected quarterly reference in format YYYYQn, got {ref_quarter!r}") + year = int(match.group(1)) + quarter = int(match.group(2)) + if quarter not in {1, 2, 3, 4}: + raise ValueError(f"Invalid quarter: {quarter}. Must be 1, 2, 3, or 4") + month = quarter * 3 + # Calendar quarter-end day + if month in {3, 12}: + day = 31 + elif month in {6, 9}: + day = 30 + else: # pragma: no cover — unreachable for valid quarters + raise ValueError(f"Invalid quarter month: {month}") + return date(year, month, day) + + +def quarter_end_date(ref_quarter: str | pd.Period) -> date: + """Return the quarter-end date for a reference quarter. + + Args: + ref_quarter: Quarterly reference in ``YYYYQn`` format or a pandas Period + (e.g., ``"2025Q1"``, ``pd.Period("2025Q1", freq="Q")``). + + Returns: + The calendar quarter-end date. + + Raises: + ValueError: If the reference quarter does not match ``YYYYQn``. + """ + return _quarter_end_from_string(str(ref_quarter)) + + +def quarter_start_date(ref_quarter: str | pd.Period) -> date: + """Return the quarter-start date for a reference quarter.""" + quarter_end = quarter_end_date(ref_quarter) + if quarter_end.month == 3: + return date(quarter_end.year, 1, 1) + if quarter_end.month == 6: + return date(quarter_end.year, 4, 1) + if quarter_end.month == 9: + return date(quarter_end.year, 7, 1) + return date(quarter_end.year, 10, 1) + + +def quarter_obs_date( + ref_quarter: str | pd.Period, obs_date_anchor: Literal["end", "start"] = "end" +) -> date: + """Return the quarter observation date under the configured anchor policy.""" + if obs_date_anchor == "end": + return quarter_end_date(ref_quarter) + if obs_date_anchor == "start": + return quarter_start_date(ref_quarter) + raise ValueError(f"obs_date_anchor must be 'end' or 'start', got {obs_date_anchor}") + + +# --------------------------------------------------------------------------- +# Observation anchor resolution +# --------------------------------------------------------------------------- + + +def resolve_target_obs_date_anchor( + target_obs_date_anchor: Literal["auto", "end", "start"], + *, + target_series_key: str, + catalog: SeriesCatalog | None, +) -> Literal["end", "start"]: + """Resolve the effective target observation anchor. + + Resolution policy: + 1. Explicit ``"start"`` or ``"end"`` is honored. + 2. ``"auto"`` uses catalog metadata ``obs_date_anchor`` for the target series. + 3. If metadata is missing/unknown, fallback is ``"end"``. + """ + if target_obs_date_anchor in {"start", "end"}: + return target_obs_date_anchor # type: ignore[return-value] + if target_obs_date_anchor != "auto": + raise ValueError( + f"target_obs_date_anchor must be one of 'auto', 'start', 'end', got " + f"{target_obs_date_anchor!r}" + ) + if catalog is not None: + meta = catalog.get(target_series_key) + if meta is not None and meta.obs_date_anchor in {"start", "end"}: + return meta.obs_date_anchor # type: ignore[return-value] + return "end" + + +# --------------------------------------------------------------------------- +# Release listing +# --------------------------------------------------------------------------- + + +def list_quarterly_target_releases_asof( + adapter: PITAdapter, + *, + series_key: str, + ref_quarter: str | pd.Period, + asof_date: date, + obs_date_anchor: Literal["end", "start"] = "end", +) -> pd.DataFrame: + """List available quarterly releases up to an as-of date. + + Args: + adapter: PIT adapter that supports ``list_pit_observations_asof``. + series_key: PIT series key for the target series. + ref_quarter: Quarterly reference in ``YYYYQn`` format or a pandas Period. + asof_date: Vintage cut-off date (treated as end-of-day UTC). + obs_date_anchor: Observation date anchor policy. + + Returns: + DataFrame sorted by ``asof_utc`` (ascending) with columns: + ``obs_date`` (quarter start/end per anchor), ``asof_utc`` (release timestamp), + and ``value``. + """ + obs_date = quarter_obs_date(ref_quarter, obs_date_anchor=obs_date_anchor) + releases = adapter.list_pit_observations_asof( + series_key=series_key, + obs_date=obs_date, + asof_date=asof_date, + ) + releases = releases.loc[:, ["obs_date", "asof_utc", "value"]].copy() + return releases.sort_values("asof_utc", kind="mergesort").reset_index(drop=True) + + +def list_quarterly_target_releases_asof_multi( + adapter: PITAdapter, + *, + series_key: str, + ref_quarters: Iterable[str | pd.Period], + asof_date: date, + obs_date_anchor: Literal["end", "start"] = "end", +) -> dict[str, pd.DataFrame]: + """Batch release lookup for multiple quarterly references at the same as-of date.""" + refs = [str(ref) for ref in ref_quarters] + if not refs: + return {} + + if hasattr(adapter, "list_pit_observations_asof_multi"): + requests = pd.DataFrame( + { + "request_id": refs, + "series_key": [series_key] * len(refs), + "obs_date": [ + quarter_obs_date(ref, obs_date_anchor=obs_date_anchor) for ref in refs + ], + "asof_date": [asof_date] * len(refs), + } + ) + try: + releases = adapter.list_pit_observations_asof_multi(requests) # type: ignore[attr-defined] + except NotImplementedError: + releases = pd.DataFrame() + out: dict[str, pd.DataFrame] = {} + if not releases.empty and "request_id" in releases.columns: + grouped = releases.groupby("request_id", sort=False) + for request_id, frame in grouped: + out[str(request_id)] = ( + frame.loc[:, ["obs_date", "asof_utc", "value"]] + .sort_values("asof_utc", kind="mergesort") + .reset_index(drop=True) + ) + empty = pd.DataFrame( + { + "obs_date": pd.Series(dtype="datetime64[ns, UTC]"), + "asof_utc": pd.Series(dtype="datetime64[ns, UTC]"), + "value": pd.Series(dtype="float64"), + } + ) + for ref in refs: + out.setdefault(ref, empty.copy()) + return out + + return { + ref: list_quarterly_target_releases_asof( + adapter, + series_key=series_key, + ref_quarter=ref, + asof_date=asof_date, + obs_date_anchor=obs_date_anchor, + ) + for ref in refs + } + + +# --------------------------------------------------------------------------- +# Policy resolution +# --------------------------------------------------------------------------- + + +def resolve_target_from_releases( + releases: pd.DataFrame, + policy: TargetPolicy, +) -> tuple[float | None, dict]: + """Resolve a target value from available releases. + + Args: + releases: DataFrame containing ``obs_date``, ``asof_utc``, and ``value`` columns. + policy: Target policy configuration. + + Returns: + Tuple of (value, metadata). ``value`` is ``None`` if the policy cannot select + a release. Metadata includes: + ``policy_mode``, ``selected_release_rank``, ``selected_release_asof_utc``, + ``target_release_rank``, ``available_release_ranks``, ``n_releases_available``, + and ``obs_date``. ``selected_*`` fields only describe an available release, + while ``target_release_rank`` is used to signal a future (next) release. + + Raises: + ValueError: If the policy configuration is invalid. + """ + releases_sorted = releases.sort_values("asof_utc", kind="mergesort").reset_index(drop=True) + raw_k = len(releases_sorted) + obs_date = releases_sorted["obs_date"].iloc[0] if raw_k > 0 else None + + nonnull_releases_all = releases_sorted.loc[releases_sorted["value"].notna()].copy() + nonnull_releases_all["__row_rank"] = nonnull_releases_all.index + 1 + + if policy.max_release_rank is not None: + if policy.max_release_rank < 1: + raise ValueError("max_release_rank must be >= 1") + nonnull_releases = nonnull_releases_all.head(policy.max_release_rank).reset_index(drop=True) + else: + nonnull_releases = nonnull_releases_all.reset_index(drop=True) + + k = len(nonnull_releases) + available_release_ranks = list(range(1, k + 1)) + nonnull_release_row_ranks = nonnull_releases["__row_rank"].astype(int).tolist() + n_nonnull = k + + value: float | None = None + selected_rank: int | None = None + selected_row_rank: int | None = None + selected_asof_utc = None + target_release_rank: int | None = None + + if policy.mode == "latest_available": + if n_nonnull > 0: + selected_rank = n_nonnull + row = nonnull_releases.iloc[selected_rank - 1] + value = float(row["value"]) + selected_asof_utc = row["asof_utc"] + selected_row_rank = int(row["__row_rank"]) + elif raw_k > 0: + # Preserve historical metadata semantics: latest_available reports rank=1 when + # release rows exist but all values are null. + selected_rank = 1 + elif policy.mode == "nth_release": + if policy.nth is None: + raise ValueError("TargetPolicy.nth must be set for nth_release mode") + if policy.nth < 1: + raise ValueError("TargetPolicy.nth must be >= 1") + selected_rank = policy.nth + if policy.nth <= n_nonnull: + row = nonnull_releases.iloc[selected_rank - 1] + value = float(row["value"]) + selected_asof_utc = row["asof_utc"] + selected_row_rank = int(row["__row_rank"]) + elif policy.mode == "next_release": + target_release_rank = n_nonnull + 1 + if policy.max_release_rank is not None: + target_release_rank = min(target_release_rank, policy.max_release_rank) + else: + raise ValueError(f"Unknown target policy mode: {policy.mode}") + + meta = { + "policy_mode": policy.mode, + "selected_release_rank": selected_rank, + "selected_release_asof_utc": selected_asof_utc if value is not None else None, + "selected_release_row_rank": selected_row_rank if value is not None else None, + "target_release_rank": target_release_rank, + "available_release_ranks": available_release_ranks, + "available_nonnull_release_row_ranks": nonnull_release_row_ranks, + "n_releases_available": k, + "n_nonnull_releases_available": n_nonnull, + "n_raw_releases_available": raw_k, + "obs_date": obs_date, + } + return value, meta + + +# --------------------------------------------------------------------------- +# High-level resolution +# --------------------------------------------------------------------------- + + +def resolve_quarterly_final_target( + adapter: PITAdapter, + *, + series_key: str, + ref_quarter: str | pd.Period, + evaluation_asof_date: date, + policy: TargetPolicy = TargetPolicy(mode="nth_release", nth=3, max_release_rank=3), + obs_date_anchor: Literal["end", "start"] = "end", +) -> tuple[float | None, dict]: + """Resolve a final target value for a quarterly reference. + + Args: + adapter: PIT adapter that supports ``list_pit_observations_asof``. + series_key: PIT series key for the target series. + ref_quarter: Quarterly reference in ``YYYYQn`` format or a pandas Period. + evaluation_asof_date: As-of date used to resolve the final target proxy. + policy: Target selection policy (defaults to nth_release=3 capped to 3 releases). + obs_date_anchor: Observation date anchor policy. + + Returns: + Tuple of (value, metadata) from ``resolve_target_from_releases``. + """ + releases = list_quarterly_target_releases_asof( + adapter, + series_key=series_key, + ref_quarter=ref_quarter, + asof_date=evaluation_asof_date, + obs_date_anchor=obs_date_anchor, + ) + return resolve_target_from_releases(releases, policy) + + +def get_quarterly_release_observation_stream( + adapter: PITAdapter, + *, + series_key: str, + ref_quarter: str | pd.Period, + asof_date: date, + max_release_rank: int | None = None, + obs_date_anchor: Literal["end", "start"] = "end", +) -> pd.DataFrame: + """Return a time-ordered observation stream of quarterly releases.""" + releases = list_quarterly_target_releases_asof( + adapter, + series_key=series_key, + ref_quarter=ref_quarter, + asof_date=asof_date, + obs_date_anchor=obs_date_anchor, + ) + if max_release_rank is not None: + if max_release_rank < 1: + raise ValueError("max_release_rank must be >= 1") + releases = releases.head(max_release_rank).reset_index(drop=True) + + if releases.empty: + return pd.DataFrame( + { + "series_key": pd.Series(dtype="object"), + "ref_quarter": pd.Series(dtype="object"), + "obs_date": pd.Series(dtype="datetime64[ns, UTC]"), + "release_rank": pd.Series(dtype="int64"), + "asof_utc": pd.Series(dtype="datetime64[ns, UTC]"), + "value": pd.Series(dtype="float64"), + } + ) + + release_rank = pd.Series(range(1, len(releases) + 1), name="release_rank") + stream = releases.copy() + stream.insert(0, "series_key", series_key) + stream.insert(1, "ref_quarter", str(ref_quarter)) + stream.insert(3, "release_rank", release_rank) + return stream.loc[ + :, ["series_key", "ref_quarter", "obs_date", "release_rank", "asof_utc", "value"] + ] diff --git a/alphaforge/pit/utils/__init__.py b/alphaforge/pit/utils/__init__.py new file mode 100644 index 0000000..ea78e39 --- /dev/null +++ b/alphaforge/pit/utils/__init__.py @@ -0,0 +1,17 @@ +"""PIT utility helpers.""" + +from alphaforge.pit.utils.timestamps import ( + coerce_utc_timestamp, + normalize_utc_day, + to_month_end, + to_quarter_end_ts, + to_utc_naive, +) + +__all__ = [ + "coerce_utc_timestamp", + "normalize_utc_day", + "to_utc_naive", + "to_month_end", + "to_quarter_end_ts", +] diff --git a/alphaforge/pit/utils/timestamps.py b/alphaforge/pit/utils/timestamps.py new file mode 100644 index 0000000..1358e35 --- /dev/null +++ b/alphaforge/pit/utils/timestamps.py @@ -0,0 +1,75 @@ +"""Consolidated timestamp and date conversion utilities. + +This module provides canonical implementations of UTC timestamp coercion +and date normalization for PIT data operations. +""" + +from __future__ import annotations + +from datetime import date + +import pandas as pd + + +def coerce_utc_timestamp(value: date | pd.Timestamp | str) -> pd.Timestamp: + """Coerce a date-like value to a UTC-aware pandas Timestamp. + + Accepts date, pd.Timestamp (tz-naive or tz-aware), or parseable string. + """ + ts = pd.Timestamp(value) + if ts.tzinfo is None: + ts = ts.tz_localize("UTC") + else: + ts = ts.tz_convert("UTC") + return ts + + +def normalize_utc_day(value: date | pd.Timestamp) -> date: + """Normalize a date-like value to a UTC midnight date object.""" + return coerce_utc_timestamp(value).normalize().date() + + +def to_utc_naive( + values: pd.Series | pd.DatetimeIndex | object, +) -> pd.Series | pd.DatetimeIndex | pd.Timestamp: + """Convert datetime-like values to UTC-naive timestamps. + + Handles pd.Series, pd.DatetimeIndex, and scalar values. + """ + if isinstance(values, pd.Series) and pd.api.types.is_datetime64_any_dtype(values): + parsed = values + elif isinstance(values, pd.DatetimeIndex): + parsed = values + else: + parsed = pd.to_datetime(values, utc=True, errors="coerce") + if isinstance(values, pd.Series): + if parsed.dt.tz is None: + parsed = parsed.dt.tz_localize("UTC") + else: + parsed = parsed.dt.tz_convert("UTC") + return parsed.dt.tz_localize(None) + if isinstance(parsed, pd.DatetimeIndex): + if parsed.tz is None: + parsed = parsed.tz_localize("UTC") + else: + parsed = parsed.tz_convert("UTC") + return parsed.tz_localize(None) + if pd.isna(parsed): + return parsed + return parsed.tz_convert("UTC").tz_localize(None) + + +def to_month_end(ts: pd.Timestamp) -> pd.Timestamp: + """Convert a timestamp to the UTC-aware end of its month.""" + ts = coerce_utc_timestamp(ts) + naive = ts.tz_localize(None) + month_end = naive.to_period("M").end_time.normalize() + return month_end.tz_localize("UTC") + + +def to_quarter_end_ts(ts: pd.Timestamp) -> pd.Timestamp: + """Convert a timestamp to the UTC-aware end of its quarter.""" + ts = coerce_utc_timestamp(ts) + naive = ts.tz_localize(None) + quarter_end = naive.to_period("Q").end_time.normalize() + return quarter_end.tz_localize("UTC") diff --git a/docs/api/pit-target.md b/docs/api/pit-target.md new file mode 100644 index 0000000..3bdc6fa --- /dev/null +++ b/docs/api/pit-target.md @@ -0,0 +1,3 @@ +# PIT Target Policy + +::: alphaforge.pit.target diff --git a/mkdocs.yml b/mkdocs.yml index 1d79985..f976ea7 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -66,6 +66,7 @@ nav: - PIT Validation: api/pit-validation.md - PIT Guards: api/pit-guards.md - PIT Tasks: api/pit-tasks.md + - PIT Target Policy: api/pit-target.md - PIT Data Source: api/pit-source.md - Stores: api/stores.md - Public Web: api/public-web.md diff --git a/tests/test_pit_target.py b/tests/test_pit_target.py new file mode 100644 index 0000000..b930c5c --- /dev/null +++ b/tests/test_pit_target.py @@ -0,0 +1,377 @@ +"""Tests for alphaforge.pit.target — TargetPolicy and release resolution.""" + +from __future__ import annotations + +from datetime import date +from unittest.mock import MagicMock + +import pandas as pd +import pytest + +from alphaforge.pit.target import ( + TargetPolicy, + get_quarterly_release_observation_stream, + list_quarterly_target_releases_asof, + list_quarterly_target_releases_asof_multi, + quarter_end_date, + quarter_obs_date, + quarter_start_date, + resolve_quarterly_final_target, + resolve_target_from_releases, + resolve_target_obs_date_anchor, +) + +# --------------------------------------------------------------------------- +# TargetPolicy dataclass +# --------------------------------------------------------------------------- + + +class TestTargetPolicy: + def test_latest_available(self): + p = TargetPolicy(mode="latest_available") + assert p.mode == "latest_available" + assert p.nth is None + assert p.max_release_rank is None + + def test_nth_release(self): + p = TargetPolicy(mode="nth_release", nth=3, max_release_rank=3) + assert p.nth == 3 + assert p.max_release_rank == 3 + + def test_next_release(self): + p = TargetPolicy(mode="next_release") + assert p.mode == "next_release" + + +# --------------------------------------------------------------------------- +# Quarter calendar helpers +# --------------------------------------------------------------------------- + + +class TestQuarterCalendar: + @pytest.mark.parametrize( + ("ref", "expected"), + [ + ("2025Q1", date(2025, 3, 31)), + ("2025Q2", date(2025, 6, 30)), + ("2025Q3", date(2025, 9, 30)), + ("2025Q4", date(2025, 12, 31)), + ("2024q2", date(2024, 6, 30)), + ], + ) + def test_quarter_end_date(self, ref, expected): + assert quarter_end_date(ref) == expected + + def test_quarter_end_date_pd_period(self): + period = pd.Period("2025Q1", freq="Q") + assert quarter_end_date(period) == date(2025, 3, 31) + + def test_quarter_end_date_invalid(self): + with pytest.raises(ValueError, match="Expected quarterly"): + quarter_end_date("2025-01") + + @pytest.mark.parametrize( + ("ref", "expected"), + [ + ("2025Q1", date(2025, 1, 1)), + ("2025Q2", date(2025, 4, 1)), + ("2025Q3", date(2025, 7, 1)), + ("2025Q4", date(2025, 10, 1)), + ], + ) + def test_quarter_start_date(self, ref, expected): + assert quarter_start_date(ref) == expected + + def test_quarter_obs_date_end(self): + assert quarter_obs_date("2025Q1", obs_date_anchor="end") == date(2025, 3, 31) + + def test_quarter_obs_date_start(self): + assert quarter_obs_date("2025Q1", obs_date_anchor="start") == date(2025, 1, 1) + + def test_quarter_obs_date_invalid_anchor(self): + with pytest.raises(ValueError, match="obs_date_anchor"): + quarter_obs_date("2025Q1", obs_date_anchor="middle") # type: ignore[arg-type] + + +# --------------------------------------------------------------------------- +# resolve_target_obs_date_anchor +# --------------------------------------------------------------------------- + + +class TestResolveTargetObsDateAnchor: + def test_explicit_end(self): + assert resolve_target_obs_date_anchor("end", target_series_key="X", catalog=None) == "end" + + def test_explicit_start(self): + assert ( + resolve_target_obs_date_anchor("start", target_series_key="X", catalog=None) == "start" + ) + + def test_auto_fallback_no_catalog(self): + assert resolve_target_obs_date_anchor("auto", target_series_key="X", catalog=None) == "end" + + def test_auto_from_catalog(self): + catalog = MagicMock() + meta = MagicMock() + meta.obs_date_anchor = "start" + catalog.get.return_value = meta + result = resolve_target_obs_date_anchor("auto", target_series_key="GDP", catalog=catalog) + assert result == "start" + + def test_auto_catalog_missing_key(self): + catalog = MagicMock() + catalog.get.return_value = None + result = resolve_target_obs_date_anchor("auto", target_series_key="GDP", catalog=catalog) + assert result == "end" + + def test_invalid_anchor(self): + with pytest.raises(ValueError, match="target_obs_date_anchor"): + resolve_target_obs_date_anchor("foo", target_series_key="X", catalog=None) # type: ignore[arg-type] + + +# --------------------------------------------------------------------------- +# resolve_target_from_releases +# --------------------------------------------------------------------------- + + +def _make_releases(values, asof_dates=None): + """Helper to build a releases DataFrame.""" + n = len(values) + if asof_dates is None: + base = pd.Timestamp("2024-04-15", tz="UTC") + asof_dates = [base + pd.Timedelta(days=30 * i) for i in range(n)] + return pd.DataFrame( + { + "obs_date": [pd.Timestamp("2024-03-31", tz="UTC")] * n, + "asof_utc": asof_dates, + "value": values, + } + ) + + +class TestResolveTargetFromReleases: + def test_latest_available_single(self): + releases = _make_releases([1.5]) + val, meta = resolve_target_from_releases(releases, TargetPolicy(mode="latest_available")) + assert val == 1.5 + assert meta["selected_release_rank"] == 1 + assert meta["n_releases_available"] == 1 + + def test_latest_available_multiple(self): + releases = _make_releases([1.5, 2.0, 2.5]) + val, meta = resolve_target_from_releases(releases, TargetPolicy(mode="latest_available")) + assert val == 2.5 + assert meta["selected_release_rank"] == 3 + + def test_latest_available_all_null(self): + releases = _make_releases([None, None]) + val, meta = resolve_target_from_releases(releases, TargetPolicy(mode="latest_available")) + assert val is None + assert meta["selected_release_rank"] == 1 + + def test_nth_release_hit(self): + releases = _make_releases([1.5, 2.0, 2.5]) + val, meta = resolve_target_from_releases(releases, TargetPolicy(mode="nth_release", nth=2)) + assert val == 2.0 + assert meta["selected_release_rank"] == 2 + + def test_nth_release_miss(self): + releases = _make_releases([1.5]) + val, meta = resolve_target_from_releases(releases, TargetPolicy(mode="nth_release", nth=3)) + assert val is None + + def test_nth_release_no_nth(self): + releases = _make_releases([1.5]) + with pytest.raises(ValueError, match="nth must be set"): + resolve_target_from_releases(releases, TargetPolicy(mode="nth_release")) + + def test_nth_release_invalid_nth(self): + releases = _make_releases([1.5]) + with pytest.raises(ValueError, match="nth must be >= 1"): + resolve_target_from_releases(releases, TargetPolicy(mode="nth_release", nth=0)) + + def test_next_release(self): + releases = _make_releases([1.5, 2.0]) + val, meta = resolve_target_from_releases(releases, TargetPolicy(mode="next_release")) + assert val is None + assert meta["target_release_rank"] == 3 + + def test_next_release_with_cap(self): + releases = _make_releases([1.5, 2.0]) + val, meta = resolve_target_from_releases( + releases, TargetPolicy(mode="next_release", max_release_rank=2) + ) + assert meta["target_release_rank"] == 2 + + def test_max_release_rank(self): + releases = _make_releases([1.5, 2.0, 2.5, 3.0]) + val, meta = resolve_target_from_releases( + releases, TargetPolicy(mode="latest_available", max_release_rank=2) + ) + assert val == 2.0 + assert meta["n_releases_available"] == 2 + + def test_max_release_rank_invalid(self): + releases = _make_releases([1.5]) + with pytest.raises(ValueError, match="max_release_rank must be >= 1"): + resolve_target_from_releases( + releases, TargetPolicy(mode="latest_available", max_release_rank=0) + ) + + def test_empty_releases(self): + releases = pd.DataFrame( + { + "obs_date": pd.Series(dtype="datetime64[ns, UTC]"), + "asof_utc": pd.Series(dtype="datetime64[ns, UTC]"), + "value": pd.Series(dtype="float64"), + } + ) + val, meta = resolve_target_from_releases(releases, TargetPolicy(mode="latest_available")) + assert val is None + assert meta["n_raw_releases_available"] == 0 + + def test_unknown_mode(self): + releases = _make_releases([1.5]) + with pytest.raises(ValueError, match="Unknown target policy mode"): + resolve_target_from_releases(releases, TargetPolicy(mode="bad")) # type: ignore[arg-type] + + def test_nan_skipped(self): + releases = _make_releases([float("nan"), 2.0, float("nan")]) + val, meta = resolve_target_from_releases(releases, TargetPolicy(mode="latest_available")) + assert val == 2.0 + assert meta["n_nonnull_releases_available"] == 1 + + +# --------------------------------------------------------------------------- +# Adapter-dependent functions (mock PITAdapter) +# --------------------------------------------------------------------------- + + +def _mock_adapter(releases_df): + adapter = MagicMock(spec=["list_pit_observations_asof"]) + adapter.list_pit_observations_asof.return_value = releases_df + return adapter + + +class TestListQuarterlyTargetReleasesAsof: + def test_basic(self): + releases = _make_releases([1.5, 2.0]) + adapter = _mock_adapter(releases) + result = list_quarterly_target_releases_asof( + adapter, series_key="GDP", ref_quarter="2024Q1", asof_date=date(2024, 12, 31) + ) + assert len(result) == 2 + assert list(result.columns) == ["obs_date", "asof_utc", "value"] + adapter.list_pit_observations_asof.assert_called_once() + + def test_sorted_by_asof(self): + releases = pd.DataFrame( + { + "obs_date": [pd.Timestamp("2024-03-31", tz="UTC")] * 2, + "asof_utc": [ + pd.Timestamp("2024-06-01", tz="UTC"), + pd.Timestamp("2024-05-01", tz="UTC"), + ], + "value": [2.0, 1.5], + } + ) + adapter = _mock_adapter(releases) + result = list_quarterly_target_releases_asof( + adapter, series_key="GDP", ref_quarter="2024Q1", asof_date=date(2024, 12, 31) + ) + assert result.iloc[0]["value"] == 1.5 + assert result.iloc[1]["value"] == 2.0 + + +class TestListQuarterlyMulti: + def test_fallback_to_single(self): + releases = _make_releases([1.5]) + adapter = MagicMock(spec=["list_pit_observations_asof"]) + adapter.list_pit_observations_asof.return_value = releases + result = list_quarterly_target_releases_asof_multi( + adapter, + series_key="GDP", + ref_quarters=["2024Q1", "2024Q2"], + asof_date=date(2024, 12, 31), + ) + assert "2024Q1" in result + assert "2024Q2" in result + assert adapter.list_pit_observations_asof.call_count == 2 + + def test_empty_refs(self): + adapter = MagicMock() + result = list_quarterly_target_releases_asof_multi( + adapter, series_key="GDP", ref_quarters=[], asof_date=date(2024, 12, 31) + ) + assert result == {} + + +class TestResolveQuarterlyFinalTarget: + def test_basic(self): + releases = _make_releases([1.5, 2.0, 2.5]) + adapter = _mock_adapter(releases) + val, meta = resolve_quarterly_final_target( + adapter, series_key="GDP", ref_quarter="2024Q1", evaluation_asof_date=date(2024, 12, 31) + ) + assert val == 2.5 + assert meta["selected_release_rank"] == 3 + + +class TestGetQuarterlyReleaseObservationStream: + def test_basic(self): + releases = _make_releases([1.5, 2.0]) + adapter = _mock_adapter(releases) + stream = get_quarterly_release_observation_stream( + adapter, series_key="GDP", ref_quarter="2024Q1", asof_date=date(2024, 12, 31) + ) + assert list(stream.columns) == [ + "series_key", + "ref_quarter", + "obs_date", + "release_rank", + "asof_utc", + "value", + ] + assert len(stream) == 2 + assert list(stream["release_rank"]) == [1, 2] + assert stream.iloc[0]["series_key"] == "GDP" + assert stream.iloc[0]["ref_quarter"] == "2024Q1" + + def test_empty(self): + empty_df = pd.DataFrame( + { + "obs_date": pd.Series(dtype="datetime64[ns, UTC]"), + "asof_utc": pd.Series(dtype="datetime64[ns, UTC]"), + "value": pd.Series(dtype="float64"), + } + ) + adapter = _mock_adapter(empty_df) + stream = get_quarterly_release_observation_stream( + adapter, series_key="GDP", ref_quarter="2024Q1", asof_date=date(2024, 12, 31) + ) + assert stream.empty + assert "release_rank" in stream.columns + + def test_max_release_rank(self): + releases = _make_releases([1.5, 2.0, 2.5]) + adapter = _mock_adapter(releases) + stream = get_quarterly_release_observation_stream( + adapter, + series_key="GDP", + ref_quarter="2024Q1", + asof_date=date(2024, 12, 31), + max_release_rank=2, + ) + assert len(stream) == 2 + + def test_max_release_rank_invalid(self): + releases = _make_releases([1.5]) + adapter = _mock_adapter(releases) + with pytest.raises(ValueError, match="max_release_rank must be >= 1"): + get_quarterly_release_observation_stream( + adapter, + series_key="GDP", + ref_quarter="2024Q1", + asof_date=date(2024, 12, 31), + max_release_rank=0, + ) diff --git a/tests/test_vintage_resolvers.py b/tests/test_vintage_resolvers.py index 1282eb2..f2c7751 100644 --- a/tests/test_vintage_resolvers.py +++ b/tests/test_vintage_resolvers.py @@ -241,6 +241,31 @@ def test_different_series_different_vintages(self, revision_map: dict) -> None: assert payems_result == date(2024, 6, 7) assert gdp_result != payems_result + def test_lookahead_guard_filters_future_vintages(self, revision_map: dict) -> None: + """FrozenResolver must not return vintages after requested_asof.""" + resolver = FrozenResolver(revision_map=revision_map, n_releases=3) + # requested_asof is before the 3rd release (2024-06-27), + # so only 2 releases are valid → returns 2nd release + result = resolver.resolve( + series_key="GDP", + obs_date=date(2024, 3, 31), + requested_asof=date(2024, 6, 1), + has_pit=True, + ) + assert result == date(2024, 5, 30) # 2nd release (latest valid) + + def test_lookahead_guard_no_valid_vintages(self, revision_map: dict) -> None: + """If requested_asof is before all vintages, fall back to requested_asof.""" + resolver = FrozenResolver(revision_map=revision_map, n_releases=3) + asof = date(2024, 1, 1) + result = resolver.resolve( + series_key="GDP", + obs_date=date(2024, 3, 31), + requested_asof=asof, + has_pit=True, + ) + assert result == asof # no valid vintages → fall through + def test_empty_revision_map(self) -> None: resolver = FrozenResolver(revision_map={}, n_releases=3) asof = date(2024, 6, 15)