Show ACS margins of error in the charts and tables

backend/.sqlfluff

@@ -38,3 +38,4 @@ rule_string = CAST("Residential_Min_Lot" AS VARCHAR) AS residential_min_lot
 indicators = 'ASTHMA', 'DIABETES'
 path = dummy.csv
 min_acres = 10
+margins = True

backend/api/routes/post_routes/post_acs5_db.py

@@ -448,12 +448,14 @@ def dp_combined_tree():
 # (see query/dp_variable_names.py).
 DP_IDENTITY_COLUMNS = ("variable_code", "source_label")
 
+# Each value's 90% margin of error and reliability label, under the names the
+# frontend reads beside Value. A warehouse built before margins were collected
+# has neither, and responses then leave them out.
+DP_MARGIN_COLUMNS = {"moe": "Value_MOE", "reliability": "Value_Reliability"}
 
-def _dp_identity_columns() -> list[str]:
-    present = {
-        row[0] for row in DB.execute("DESCRIBE acs5_dp_combined_tidy").fetchall()
-    }
-    return [c for c in DP_IDENTITY_COLUMNS if c in present]
+
+def _dp_columns() -> set[str]:
+    return {row[0] for row in DB.execute("DESCRIBE acs5_dp_combined_tidy").fetchall()}
 
 
 @router.post("/load/acs5-db/dp-combined/series")
@@ -465,8 +467,12 @@ def dp_combined_series(request: DPSeriesRequest):
     in a stable order so each keeps its place across years: by variable code
     when the warehouse has it, otherwise by load order.
     """
-    identity = _dp_identity_columns()
+    present = _dp_columns()
+    identity = [c for c in DP_IDENTITY_COLUMNS if c in present]
     extra = "".join(f", {c}" for c in identity)
+    extra += "".join(
+        f", {c} AS {name}" for c, name in DP_MARGIN_COLUMNS.items() if c in present
+    )
     tiebreak = "variable_code" if "variable_code" in identity else "rowid"
     rows = DB.execute(
         f"""--sql
@@ -512,11 +518,12 @@ def county_comparison_indicators():
 @router.post("/load/acs5-db/county-comparison")
 def county_comparison(request: CountyComparisonRequest):
     """Return {place, geo_type, year, value} for each county, Vermont and the
-    United States, every year the indicator has data."""
+    United States, every year the indicator has data, with value_MOE and
+    value_Reliability where the warehouse has margins."""
     indicator = INDICATORS_BY_KEY.get(request.indicator)
     if indicator is None or not indicator.available:
         raise HTTPException(404, f"Unknown indicator: {request.indicator}")
-    sql, params = comparison_query(indicator)
+    sql, params = comparison_query(indicator, margins="moe" in _dp_columns())
     rows = DB.execute(sql, params).df()
     return make_response(
         data=rows,

backend/query/acs5.py

@@ -126,6 +126,8 @@
 
 # The cleaned timeseries tables store their measure columns in snake_case, but
 # the frontend charts read these Title_Case field names, so responses keep them.
+# Where a series has a 90% margin of error, it follows as `<field>_MOE`, with
+# its reliability label (high, medium or low) as `<field>_Reliability`.
 TIMESERIES_FIELD_NAMES = {
     "median_age": "Median_Age",
     "age_dependency_ratio": "Age_Dependency_Ratio",
@@ -135,6 +137,26 @@
     "median_home_value": "Median_Home_Value",
     "percent": "Percent",
 }
+TIMESERIES_FIELD_NAMES |= {
+    f"{column}_{suffix.lower()}": f"{field}_{suffix}"
+    for column, field in TIMESERIES_FIELD_NAMES.items()
+    for suffix in ("MOE", "Reliability")
+}
+
+# The tidy tables' columns the tidy endpoints return, under the Title Case
+# names the frontend reads. A warehouse built before margins were collected
+# has no margin columns; its responses just leave them out.
+TIDY_FIELD_NAMES = {
+    "year": "year",
+    "section": "Section",
+    "variable": "Variable",
+    "value": "Value",
+    "value_moe": "Value_MOE",
+    "value_reliability": "Value_Reliability",
+    "percent": "Percent",
+    "percent_moe": "Percent_MOE",
+    "percent_reliability": "Percent_Reliability",
+}
 
 
 def _acs5_source(
@@ -172,6 +194,9 @@ def get_acs5_tidy(dataset: str, filters: dict | None = None) -> pd.DataFrame:
     sql, params = sql_filter_block(sql_path / "acs5_tidy.sql", [source])
 
     result = DB.execute(sql, params).df()
+    result = result[[c for c in TIDY_FIELD_NAMES if c in result.columns]].rename(
+        columns=TIDY_FIELD_NAMES
+    )
 
     if result.empty:
         logger.error(

backend/query/comparison.py

@@ -30,6 +30,8 @@
 logger = logging.getLogger(__name__)
 sql_dir = Path(__file__).resolve().parent / "sql"
 
+# A value's 90% margin of error and reliability label, where the source has them
+MARGIN_COLUMNS = ("moe", "reliability")
 MIN_COMPLETE_ROWS = 3  # fewest geographies a composite index can be fit from
 # Census ACS sentinel for a suppressed/not-applicable estimate (e.g. median age
 # in a town too small to estimate). Never a real value, so drop it up front.
@@ -223,14 +225,24 @@ def dataset_for_table(table: str) -> str:
         raise ValueError(f"unregistered comparison table: {table}") from e
 
 
+def _has_margins(table: str) -> bool:
+    """Whether `table` carries ACS margins of error. A warehouse built before
+    margins were collected has none, and the ACS templates then leave their
+    `moe`/`reliability` columns out (CDC PLACES never has them)."""
+    columns = {row[0] for row in DB.execute(f"DESCRIBE {table}").fetchall()}
+    return "percent_moe" in columns
+
+
 def _fetch(cfg: dict, extra_filters: dict | None = None) -> pd.DataFrame:
     filters = dict(extra_filters or {})
     if "geo_type" in cfg:
         filters["geo_type"] = [cfg["geo_type"]]
     filters["year"] = [_latest_year(cfg["table"])]
 
     source = FilterSource(filter_table=cfg["table"], filters=filters)
-    sql, params = sql_filter_block(cfg["sql"], [source])
+    sql, params = sql_filter_block(
+        cfg["sql"], [source], margins=_has_margins(cfg["table"])
+    )
     df = DB.execute(sql, params).df()
 
     df = df.rename(
@@ -241,6 +253,9 @@ def _fetch(cfg: dict, extra_filters: dict | None = None) -> pd.DataFrame:
             cfg["name_col"]: "name",
         }
     )
+    for column in MARGIN_COLUMNS:
+        if column not in df.columns:
+            df[column] = None
     return df[df["data_value"] != ACS_SENTINEL]
 
 
@@ -261,7 +276,8 @@ def _single_variable(
     filters = dict(extra_filters or {})
     filters[cfg["var_col"]] = [var]
     df = _fetch(cfg, filters)
-    sub = df[df["measure"] == var][["geoid", "geometry", "data_value", "name"]].copy()
+    columns = ["geoid", "geometry", "data_value", "name", *MARGIN_COLUMNS]
+    sub = df[df["measure"] == var][columns].copy()
     # Rows whose Jurisdiction/geo-id failed to match a geometry (a handful of
     # unparsed or non-geographic entries in the source tidy table) carry no
     # geoid at all. Left in, two such rows from either side of a comparison
@@ -299,7 +315,8 @@ def compare_variables(
     its Crude and Age-adjusted rows, doubling every geography on the map).
     """
     d1 = _single_variable(dataset1, level, var1, filters1)
-    d2 = _single_variable(dataset2, level, var2, filters2)[["geoid", "data_value"]]
+    d2 = _single_variable(dataset2, level, var2, filters2)
+    d2 = d2[["geoid", "data_value", *MARGIN_COLUMNS]]
 
     d1["bin"], edges_x = _bin3(d1["data_value"])
     d2["bin"], edges_y = _bin3(d2["data_value"])
@@ -321,6 +338,20 @@ def compare_variables(
             var1: round(r.data_value_1, 2),
             var2: round(r.data_value_2, 2),
         }
+        margins = {
+            var: {
+                "moe": round(moe, 2),
+                "reliability": reliability if pd.notna(reliability) else None,
+            }
+            for var, moe, reliability in [
+                (var1, r.moe_1, r.reliability_1),
+                (var2, r.moe_2, r.reliability_2),
+            ]
+            if pd.notna(moe)
+        }
+        if margins:
+            # Read by the map tooltip, which shows each value with its margin.
+            tooltip["__margins__"] = margins
         features.append(
             {
                 "type": "Feature",

backend/query/county_comparison.py

@@ -19,11 +19,12 @@
 GEO_TYPES = ("county", "state", "national")
 
 
-def comparison_query(indicator: Indicator) -> tuple[str, list]:
+def comparison_query(indicator: Indicator, margins: bool = False) -> tuple[str, list]:
     """Render the comparison query for one indicator; returns (sql, params).
 
     Rows are {place, geo_type, year, value}, where `place` is the short display
-    name ("Essex", "Vermont", "United States"). Values travel as $N
+    name ("Essex", "Vermont", "United States"), plus value_MOE and
+    value_Reliability when `margins` (the table has them). Values travel as $N
     placeholders: execute with `DB.execute(sql, params)`.
     """
     params: list = []
@@ -46,4 +47,7 @@ def comparison_query(indicator: Indicator) -> tuple[str, list]:
     # with nulls, as query/acs5.py does.
     clauses.append("value >= 0")
     where_string = "WHERE " + "\n    AND ".join(clauses)
-    return render_sql(SQL_PATH, table=SOURCE_TABLE, where_string=where_string), params
+    sql = render_sql(
+        SQL_PATH, table=SOURCE_TABLE, where_string=where_string, margins=margins
+    )
+    return sql, params

backend/query/sql/acs5/acs5_county_compare.sql

@@ -7,6 +7,17 @@ SELECT
     -- present; count-only variables (Population, Median Age, ...) have no
     -- Percent and fall back to Value unchanged.
     COALESCE(a.percent, a.value) AS value,
+    {% if margins %}
+        -- The margin and reliability label of whichever figure `value` is. A
+        -- warehouse built before margins were collected has neither column.
+        CASE
+            WHEN a.percent IS NOT NULL THEN a.percent_moe ELSE a.value_moe
+        END AS moe,
+        CASE
+            WHEN a.percent IS NOT NULL THEN a.percent_reliability
+            ELSE a.value_reliability
+        END AS reliability,
+    {% endif %}
     c.geoid,
     c.county,
     ST_ASGEOJSON(ST_GeomFromWKB(c.geometry)) AS geometry

backend/query/sql/acs5/acs5_dp_county_comparison.sql

@@ -9,7 +9,11 @@ SELECT
     REGEXP_REPLACE(name, ' County, Vermont$', '') AS place,
     geo_type,
     CAST(year AS INTEGER) AS year,
-    CAST(value AS DOUBLE) AS value
+    CAST(value AS DOUBLE) AS value{% if margins %},
+    -- Named for the frontend, which reads a field's margin as <field>_MOE.
+    moe AS value_MOE,
+    reliability AS value_Reliability
+{% endif %}
 FROM {{ table }}
 {{ where_string }}
 ORDER BY year, place

backend/query/sql/acs5/acs5_tidy.sql

@@ -1,11 +1,6 @@
--- The response keeps the Title Case field names the frontend reads; the tidy
--- tables store them lowercase, and DuckDB would otherwise echo that casing.
-SELECT
-    year,
-    section AS Section,
-    variable AS Variable,
-    value AS Value,
-    percent AS Percent
+-- Every column, so the margin columns come along where the warehouse has
+-- them; get_acs5_tidy keeps and renames the fields the frontend reads.
+SELECT *
 FROM {{ table }}
 {{ where_string }}
-ORDER BY year, Section, Variable
+ORDER BY year, section, variable

backend/query/sql/acs5/acs5_town_compare.sql

@@ -3,6 +3,17 @@ SELECT
     -- see acs5_county_compare.sql: prefer the share-of-total (Percent) over
     -- the raw count (Value) when the variable has one.
     COALESCE(a.percent, a.value) AS value,
+    {% if margins %}
+        -- see acs5_county_compare.sql: the margin of whichever figure
+        -- `value` is.
+        CASE
+            WHEN a.percent IS NOT NULL THEN a.percent_moe ELSE a.value_moe
+        END AS moe,
+        CASE
+            WHEN a.percent IS NOT NULL THEN a.percent_reliability
+            ELSE a.value_reliability
+        END AS reliability,
+    {% endif %}
     t.FIPS_ID,
     t.TOWN_NAME,
     ST_ASGEOJSON(ST_GeomFromWKB(t.geometry)) AS geometry

backend/query/sql/acs5/acs5_tract_compare.sql

@@ -3,6 +3,12 @@
 SELECT
     a.variable,
     COALESCE(a.percent, a.value) AS value,
+    {% if margins %}
+        -- The share's margin and reliability label, where the warehouse has
+        -- them.
+        a.percent_moe AS moe,
+        a.percent_reliability AS reliability,
+    {% endif %}
     a.geoid,
     a.name,
     ST_ASGEOJSON(ST_GEOMFROMWKB(t.geometry)) AS geometry

backend/query/sql_render.py

@@ -129,18 +129,21 @@ def compile_filters(sources: list[FilterSource], params: list) -> tuple[str, str
     return cte_block, join_block
 
 
-def sql_filter_block(sql_path: Path, sources: list[FilterSource]) -> tuple[str, list]:
+def sql_filter_block(
+    sql_path: Path, sources: list[FilterSource], **extra
+) -> tuple[str, list]:
     """Render a query template with compiled filters; returns (sql, params).
 
     Only the fragments the template actually references are compiled, so `params`
     contains exactly the values behind the $N placeholders present in the SQL.
+    `extra` is passed to the template as is (e.g. a flag for optional columns).
     Execute with `DB.execute(sql, params)`.
     """
     source_text = sql_path.read_text()
     used = meta.find_undeclared_variables(JINJA_ENV.parse(source_text))
 
     params: list = []
-    context: dict = {}
+    context: dict = dict(extra)
     if used & {"cte_filter_block", "join_filter_block"}:
         cte_block, join_block = compile_filters(sources, params)
         context["cte_filter_block"] = cte_block

backend/tests/test_acs5_tidy_query.py

@@ -0,0 +1,69 @@
+"""
+The tidy ACS endpoints' query, run against a small in-memory tidy table
+instead of the warehouse: Title Case fields, with each estimate's margin of
+error where the warehouse has one.
+"""
+
+import importlib
+
+import duckdb
+import pytest
+
+from query import production_db
+
+ESTIMATES = (
+    "year BIGINT, geoid VARCHAR, name VARCHAR, section VARCHAR, "
+    "variable VARCHAR, value DOUBLE, percent DOUBLE"
+)
+MARGINS = (
+    "value_moe DOUBLE, value_reliability VARCHAR, "
+    "percent_moe DOUBLE, percent_reliability VARCHAR"
+)
+ROW = (2023, "5000144350", "Middlebury town", "Race", "Asian", 150.0, 1.8)
+
+
+@pytest.fixture
+def warehouse(monkeypatch):
+    conn = duckdb.connect()
+    monkeypatch.setattr(production_db, "get_db", lambda: conn)
+    module = importlib.import_module("query.acs5")
+    monkeypatch.setattr(module, "DB", conn)
+    return module, conn
+
+
+def test_margins_come_back_with_their_estimates(warehouse):
+    acs5, conn = warehouse
+    conn.execute(f"CREATE TABLE acs5_demographics_tidy ({ESTIMATES}, {MARGINS})")
+    conn.execute(
+        "INSERT INTO acs5_demographics_tidy VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+        [*ROW, 80.0, "low", 1.0, "low"],
+    )
+    df = acs5.get_acs5_tidy("demographics", {"Location": "Middlebury town"})
+
+    assert df.to_dict("records") == [
+        {
+            "year": 2023,
+            "Section": "Race",
+            "Variable": "Asian",
+            "Value": 150.0,
+            "Value_MOE": 80.0,
+            "Value_Reliability": "low",
+            "Percent": 1.8,
+            "Percent_MOE": 1.0,
+            "Percent_Reliability": "low",
+        }
+    ]
+
+
+def test_warehouse_without_margins_still_serves_its_estimates(warehouse):
+    # Built before margins were collected: the endpoints must not fail on the
+    # missing columns while the ETL catches up.
+    acs5, conn = warehouse
+    conn.execute(f"CREATE TABLE acs5_demographics_tidy ({ESTIMATES})")
+    conn.execute(
+        "INSERT INTO acs5_demographics_tidy VALUES (?, ?, ?, ?, ?, ?, ?)", list(ROW)
+    )
+    df = acs5.get_acs5_tidy("demographics", {"Location": "Middlebury town"})
+
+    assert list(df.columns) == ["year", "Section", "Variable", "Value", "Percent"]
+    assert df.iloc[0].Value == 150.0