fix: coerce earnings_calendar EPS/revenue to FLOAT64 (#140) (#147)

C00ldudeNoonan · claude · web-flow · commit 22149a0be33e · 2026-07-12T07:42:44.000-04:00
* fix: coerce earnings_calendar EPS/revenue to FLOAT64 (#140) earnings_calendar_job failed on every run with "Value of type STRING cannot be assigned to T.eps_actual, which has type FLOAT64" because Yahoo's feed returns EPS fields as strings/None. Polars inferred a String column, the staging load typed it STRING, and the MERGE into the FLOAT64 target was rejected. Coerce eps_estimated/eps_actual via a _to_float helper (str/float/None -> float|None), and pin all four numeric columns to Float64 on the frame so an all-None batch can't reintroduce the mismatch through a Null/String staging column. Also closes the same latent hazard on the always-None revenue_estimated/revenue_actual columns. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * fix: narrow _to_float param type to satisfy ty typecheck ty rejected float(value) when value was typed `object`. Type the parameter as the values Yahoo actually returns (str | float | int | None), which is convertible to float after the None/empty guard. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * fix: normalize existing earnings_calendar numeric columns before MERGE (#140) Address Codex P1: a target table first created by pre-fix code can still hold eps_*/revenue_* as STRING. The FLOAT64 staging casts would then fail the MERGE assignment into a STRING target for existing deployments. Call bq.normalize_column_types("earnings_calendar", FLOAT64 types) before upsert_data — SAFE_CASTing any drifted target columns (no-op when already FLOAT64 or the table doesn't yet exist), mirroring the treasury-yields path. Hoist the numeric column list/types to module constants and add a test asserting normalize runs before upsert with FLOAT64 types. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * style: apply ruff format to calendars.py Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
diff --git a/macro_agents/src/macro_agents/defs/domains/calendars.py b/macro_agents/src/macro_agents/defs/domains/calendars.py
@@ -60,6 +60,17 @@
 
 WEEKDAY_ABBRS = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]
 
+# earnings_calendar numeric columns typed as FLOAT64 in BigQuery. Used both to
+# pin the staging frame's dtypes and to normalize any drifted target columns
+# (older tables created by pre-fix code can hold these as STRING).
+EARNINGS_NUMERIC_COLUMNS = [
+    "eps_estimated",
+    "eps_actual",
+    "revenue_estimated",
+    "revenue_actual",
+]
+EARNINGS_NUMERIC_COLUMN_TYPES = {col: "FLOAT64" for col in EARNINGS_NUMERIC_COLUMNS}
+
 
 def _nth_weekday_of_month(year: int, month: int, weekday: int, n: int) -> date:
     """Return the date for the nth weekday of a month (weekday: 0=Mon)."""
@@ -337,6 +348,23 @@ def parse_numeric_value(value: str | None) -> float | None:
         return None
 
 
+def _to_float(value: str | float | int | None) -> float | None:
+    """Coerce a source EPS value (str, float, or None) to float, or None.
+
+    Yahoo's earnings feed returns EPS fields as strings, floats, or None
+    depending on the row. The BigQuery ``earnings_calendar`` table types these
+    columns as FLOAT64, so blank/None values must map to NULL and everything
+    else must be a real float before the frame is built (otherwise Polars
+    infers a String column and the MERGE is rejected).
+    """
+    if value is None or value == "":
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
 @dg.asset(
     group_name=CALENDAR_GROUP,
     kinds={"api", "economic_calendar"},
@@ -475,8 +503,8 @@ def earnings_calendar(
                 "company_name": row.get("company_name", ""),
                 "report_date": report_date,
                 "fiscal_date_ending": "",
-                "eps_estimated": row.get("eps_estimated"),
-                "eps_actual": row.get("eps_actual"),
+                "eps_estimated": _to_float(row.get("eps_estimated")),
+                "eps_actual": _to_float(row.get("eps_actual")),
                 "revenue_estimated": None,
                 "revenue_actual": None,
                 "report_time": row.get("report_time", ""),
@@ -489,8 +517,22 @@ def earnings_calendar(
         )
 
     processed_df = pl.DataFrame(processed_records)
+    # Pin numeric columns to Float64 so the staging table always matches the
+    # FLOAT64 raw-table schema — even for a batch where every value is None,
+    # which would otherwise infer a Null/String column and break the MERGE.
+    processed_df = processed_df.with_columns(
+        pl.col(col).cast(pl.Float64, strict=False) for col in EARNINGS_NUMERIC_COLUMNS
+    )
     context.log.info(f"Fetched {len(processed_df)} earnings announcements")
 
+    # Repair existing deployments: a target table first created by pre-fix code
+    # can still hold these columns as STRING. The FLOAT64 staging columns above
+    # would then fail the MERGE assignment into a STRING target, so normalize
+    # the target's numeric columns first (SAFE_CAST; no-op if already FLOAT64
+    # or the table doesn't yet exist).
+    bq.normalize_column_types(
+        "earnings_calendar", EARNINGS_NUMERIC_COLUMN_TYPES, context=context
+    )
     bq.upsert_data("earnings_calendar", processed_df, ["event_id"], context=context)
 
     return dg.MaterializeResult(
diff --git a/macro_agents/tests/test_earnings_calendar_types.py b/macro_agents/tests/test_earnings_calendar_types.py
@@ -0,0 +1,128 @@
+"""Regression tests for earnings_calendar EPS/revenue type coercion (issue #140).
+
+The BigQuery ``earnings_calendar`` table types EPS and revenue columns as
+FLOAT64. Yahoo's source feed returns EPS fields as strings, floats, or None, so
+the values must be coerced to float (and the columns pinned to Float64) before
+the frame is loaded into a staging table and MERGEd — otherwise BigQuery rejects
+the MERGE with ``Value of type STRING cannot be assigned to ... FLOAT64``.
+"""
+
+from unittest.mock import MagicMock
+
+import dagster as dg
+import polars as pl
+
+from macro_agents.defs.domains.calendars import (
+    EARNINGS_NUMERIC_COLUMN_TYPES,
+    _to_float,
+    earnings_calendar,
+)
+
+
+class TestToFloat:
+    def test_string_number_is_parsed(self):
+        assert _to_float("1.23") == 1.23
+
+    def test_float_passthrough(self):
+        assert _to_float(2.5) == 2.5
+
+    def test_integer_is_coerced(self):
+        assert _to_float(3) == 3.0
+
+    def test_none_maps_to_none(self):
+        assert _to_float(None) is None
+
+    def test_empty_string_maps_to_none(self):
+        assert _to_float("") is None
+
+    def test_non_numeric_string_maps_to_none(self):
+        assert _to_float("N/A") is None
+
+
+class TestNumericColumnDtypes:
+    """The staging frame must expose FLOAT64 columns even for all-None batches."""
+
+    NUMERIC_COLUMNS = [
+        "eps_estimated",
+        "eps_actual",
+        "revenue_estimated",
+        "revenue_actual",
+    ]
+
+    def _pin(self, records: list[dict]) -> pl.DataFrame:
+        # Mirror the cast applied in earnings_calendar before upsert_data.
+        df = pl.DataFrame(records)
+        return df.with_columns(
+            pl.col(col).cast(pl.Float64, strict=False) for col in self.NUMERIC_COLUMNS
+        )
+
+    def test_all_none_batch_yields_float64(self):
+        records = [
+            {c: None for c in self.NUMERIC_COLUMNS},
+            {c: None for c in self.NUMERIC_COLUMNS},
+        ]
+        df = self._pin(records)
+        for col in self.NUMERIC_COLUMNS:
+            assert df.schema[col] == pl.Float64
+
+    def test_coerced_values_yield_float64(self):
+        records = [
+            {
+                "eps_estimated": _to_float("1.1"),
+                "eps_actual": _to_float("1.3"),
+                "revenue_estimated": None,
+                "revenue_actual": None,
+            }
+        ]
+        df = self._pin(records)
+        for col in self.NUMERIC_COLUMNS:
+            assert df.schema[col] == pl.Float64
+        assert df["eps_actual"][0] == 1.3
+
+
+class TestTargetNormalization:
+    """The asset must normalize drifted STRING target columns before the MERGE.
+
+    Existing deployments can hold eps_*/revenue_* as STRING (inferred by the
+    old code's first load). The FLOAT64 staging columns would then fail the
+    MERGE into a STRING target, so normalize_column_types must run first.
+    """
+
+    def test_normalize_runs_before_upsert_with_float64_types(self):
+        source_df = pl.DataFrame(
+            [
+                {
+                    "symbol": "AAPL",
+                    "company_name": "Apple",
+                    "report_date": "2026-07-15",
+                    "eps_estimated": "1.10",
+                    "eps_actual": None,
+                    "report_time": "amc",
+                    "timing": "after_market",
+                }
+            ]
+        )
+        yahoo = MagicMock()
+        yahoo.get_earnings_range.return_value = source_df
+
+        bq = MagicMock()
+        call_order: list[str] = []
+        bq.normalize_column_types.side_effect = lambda *a, **k: call_order.append(
+            "normalize"
+        )
+        bq.upsert_data.side_effect = lambda *a, **k: call_order.append("upsert")
+
+        ctx = dg.build_asset_context()
+        earnings_calendar(ctx, yahoo, bq)
+
+        # normalize must run, with FLOAT64 types, before the upsert.
+        assert call_order == ["normalize", "upsert"]
+        norm_args = bq.normalize_column_types.call_args
+        assert norm_args.args[0] == "earnings_calendar"
+        assert norm_args.args[1] == EARNINGS_NUMERIC_COLUMN_TYPES
+        assert set(EARNINGS_NUMERIC_COLUMN_TYPES.values()) == {"FLOAT64"}
+
+        # The frame handed to upsert_data has FLOAT64 numeric columns.
+        upsert_df = bq.upsert_data.call_args.args[1]
+        for col in EARNINGS_NUMERIC_COLUMN_TYPES:
+            assert upsert_df.schema[col] == pl.Float64
