fix: data integrity — built-up persistence, status gating, SAR drift detection, direction-aware narrative

Second-pass fixes based on the Nyala South (Sudan) test report showing
AMBER flags driven by statistical artifacts rather than real signals.

1. Built-up areas: persistence-based classification
- Replace monthly NDBI>0 classification with a persistence mask:
pixel is built-up only if NDBI>0 AND NDVI<0.15 in ≥60% of months
across the analysis period. Removes the seasonal cycling artifact
where bare soil got classified as built-up in dry seasons only.
- Tighten NDVI threshold from 0.2 → 0.15 (sparse-veg regions).
- Replace monthly z-score with direct percentage change between
current and baseline persistent masks. Change thresholds:
|change|≥15% → RED, ≥5% → AMBER, else GREEN.
- Chart now shows baseline vs current as two bars instead of an
oscillating monthly time series that misrepresents reality.
- Shared _analyze_persistence helper used by both batch and
non-batch harvest paths (eliminates duplication).
- Smoke-tested: synthetic seasonal pixel correctly rejected,
persistent pixel correctly kept.

2. Status gating (_classify_zscore)
- Aggregate z-score alone is no longer sufficient to trigger RED.
Moderate z (>1) requires supporting evidence: multiple anomaly
months OR hotspot ≥ 5%. Strong z (>2) without any supporting
evidence caps at AMBER, not RED.
- New min_coverage_pct gate: when water coverage < 0.5% of AOI,
the indicator cannot leave GREEN (prevents noise-driven alerts
on near-dry landscapes).
- Water AMBER on 0.1% coverage now correctly stays GREEN.
- NDVI z=+1.9 with 2/24 anomaly months stays AMBER; same z with
0 anomaly months and no hotspots drops to GREEN.

3. SAR baseline drift detection
- When >40% of months are flagged anomalous, the finding is more
likely a Sentinel-1 IPF version change, orbit geometry shift, or
regime shift than a per-month anomaly pattern. Flag as AMBER with
"baseline may be unreliable" headline and add a limitation noting
the drift check.
- Applied to both batch and non-batch harvest paths.

4. Direction-aware compound signals and narrative
- Cross-pattern matcher now checks z-score direction (up/down/stable),
not just status level. Fixes the bug where a greening + contraction
result produced "vegetation loss coincides with settlement expansion"
in the situation narrative.
- New rule for conflict-context greening + built-up contraction:
"Vegetation recovery coincides with apparent built-up contraction —
in conflict or displacement contexts this can reflect abandoned
land returning to vegetation".
- Direction threshold is product-aware: buildup uses 5% change (its
AMBER cutoff); other indicators use z=1.0.

5. Formatting fixes
- Narrative no longer produces ".." — strip trailing period from
each sentence before re-adding exactly one.
- Overview map date range moved from below the axes (where it
collided with x-axis ticks) into the title block as a second line.

Smoke-tested: all gating cases, direction matching, persistence mask,
and a Nyala-like synthetic scenario confirm no regressions and the
specific bugs in the review report are fixed.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

app/eo_products/base.py

@@ -95,14 +95,66 @@ class BaseProduct(abc.ABC):
         return float(valid / total) if total > 0 else 0.0
 
     @staticmethod
-    def _classify_zscore(z_score: float, hotspot_pct: float) -> "StatusLevel":
-        """Classify status using z-score and hotspot percentage."""
+    def _classify_zscore(
+        z_score: float,
+        hotspot_pct: float,
+        *,
+        anomaly_months: int = 0,
+        total_months: int = 0,
+        min_coverage_pct: float | None = None,
+    ) -> "StatusLevel":
+        """Classify status using z-score, hotspot %, and evidence gates.
+
+        Evidence rules (all apply):
+
+        1. **Minimum coverage gate.** If the caller passes ``min_coverage_pct``
+           and it is below that threshold, the indicator cannot go above GREEN.
+           Used by water: when AOI water fraction is <0.5%, a large z-score
+           is dominated by noise from a handful of pixels.
+
+        2. **Aggregate-z-score alone is not enough.** A z-score between 1.0
+           and 2.0 only raises status to AMBER if supported by *at least one*
+           of: (a) two or more monthly anomalies, or (b) hotspot coverage ≥ 5%.
+           This prevents single-number z-scores from driving false alarms on
+           otherwise stable time series.
+
+        3. **RED requires either strong aggregate evidence or widespread
+           hotspots.** |z| > 2 AND ≥1 supporting month OR hotspot ≥ 25%.
+        """
         from app.models import StatusLevel
         from app.config import ZSCORE_THRESHOLD
-        if abs(z_score) > ZSCORE_THRESHOLD or hotspot_pct > 25:
+
+        z = safe_float(z_score)
+        hot = safe_float(hotspot_pct)
+
+        # Gate 1: minimum coverage
+        if min_coverage_pct is not None and min_coverage_pct < 0.5:
+            return StatusLevel.GREEN
+
+        strong_z = abs(z) > ZSCORE_THRESHOLD
+        moderate_z = abs(z) > 1.0
+        any_monthly_anomaly = anomaly_months >= 1
+        multiple_monthly_anomalies = anomaly_months >= 2
+        strong_hot = hot > 25
+        moderate_hot = hot > 10
+        minor_hot = hot > 5
+
+        # RED: widespread hotspots alone, OR strong z backed by any evidence
+        if strong_hot:
+            return StatusLevel.RED
+        if strong_z and (any_monthly_anomaly or moderate_hot):
             return StatusLevel.RED
-        if abs(z_score) > 1.0 or hotspot_pct > 10:
+
+        # AMBER: moderate z backed by supporting evidence, or moderate hotspots
+        if moderate_hot:
+            return StatusLevel.AMBER
+        if moderate_z and (multiple_monthly_anomalies or minor_hot):
             return StatusLevel.AMBER
+
+        # Strong z without any supporting evidence → only AMBER, not RED
+        if strong_z:
+            return StatusLevel.AMBER
+
         return StatusLevel.GREEN
 
     @staticmethod

app/eo_products/buildup.py

@@ -46,7 +46,12 @@ logger = logging.getLogger(__name__)
 
 BASELINE_YEARS = 5
 NDBI_THRESHOLD = 0.0  # NDBI > 0 = potential built-up
-NDVI_BUILDUP_MAX = 0.2  # NDVI < 0.2 required to exclude vegetation (combined with NDBI threshold)
+NDVI_BUILDUP_MAX = 0.15  # NDVI < 0.15 to exclude even sparse vegetation (tighter than generic 0.2)
+PERSISTENCE_MIN_FRAC = 0.6  # Pixel must meet built-up condition in ≥60% of months to be "persistent" (excludes dry-season-only false positives)
+
+# Change thresholds for status classification (persistent fraction, not z-score)
+CHANGE_AMBER_PCT = 5.0   # |change| ≥ 5% → AMBER
+CHANGE_RED_PCT = 15.0    # |change| ≥ 15% → RED
 
 
 class BuiltupProduct(BaseProduct):
@@ -152,143 +157,13 @@ class BuiltupProduct(BaseProduct):
         spatial_completeness = self._compute_spatial_completeness(current_path)
 
         if baseline_path:
-            seasonal_stats = compute_seasonal_stats_aoi(baseline_path, n_years=BASELINE_YEARS)
-            baseline_stats = self._compute_stats(baseline_path)
-            baseline_frac = baseline_stats["overall_buildup_fraction"]
-            baseline_ha = baseline_frac * aoi_ha
-
-            start_month = time_range.start.month
-            most_recent_month = ((start_month + n_current_bands - 2) % 12) + 1
-
-            # Z-score for overall current mean NDBI vs seasonal baseline
-            if most_recent_month in seasonal_stats and seasonal_stats[most_recent_month]["n_years"] > 0:
-                s = seasonal_stats[most_recent_month]
-                z_current = safe_float(compute_zscore(current_mean, s["mean"], s["std"], MIN_STD_BUILDUP))
-            else:
-                z_current = 0.0
-
-            # Per-month z-scores and anomaly count
-            anomaly_months = 0
-            monthly_zscores = []
-            for i, val in enumerate(current_stats["monthly_means"]):
-                cal_month = ((start_month + i - 1) % 12) + 1
-                if cal_month in seasonal_stats and seasonal_stats[cal_month]["n_years"] > 0:
-                    z = safe_float(compute_zscore(val, seasonal_stats[cal_month]["mean"],
-                                       seasonal_stats[cal_month]["std"], MIN_STD_BUILDUP))
-                    monthly_zscores.append(z)
-                    if abs(z) > ZSCORE_THRESHOLD:
-                        anomaly_months += 1
-                else:
-                    monthly_zscores.append(0.0)
-
-            # Pixel-level hotspot detection
-            month_map = group_bands_by_calendar_month(baseline_stats["valid_months_total"], BASELINE_YEARS)
-            hotspot_pct = 0.0
-            self._zscore_raster = None
-            self._hotspot_mask = None
-            if most_recent_month in month_map and len(month_map[most_recent_month]) > 0:
-                pixel_stats = compute_seasonal_stats_pixel(baseline_path, month_map[most_recent_month])
-                with rasterio.open(current_path) as src:
-                    current_band_idx = min(n_current_bands, src.count)
-                    current_data = src.read(current_band_idx).astype(np.float32)
-                    if src.nodata is not None:
-                        current_data[current_data == src.nodata] = np.nan
-
-                z_raster = compute_zscore_raster(current_data, pixel_stats["mean"],
-                                                  pixel_stats["std"], MIN_STD_BUILDUP)
-                hotspot_mask, hotspot_pct = detect_hotspots(z_raster, ZSCORE_THRESHOLD)
-                self._zscore_raster = z_raster
-                self._hotspot_mask = hotspot_mask
-
-            # Four-factor confidence scoring
-            baseline_depth = sum(1 for m in range(1, 13)
-                                 if m in seasonal_stats and seasonal_stats[m]["n_years"] > 0)
-            mean_baseline_years = (sum(seasonal_stats[m]["n_years"] for m in range(1, 13)
-                                       if m in seasonal_stats) / max(baseline_depth, 1))
-            conf = compute_confidence(
-                valid_months=n_current_bands,
-
-                baseline_years_with_data=int(mean_baseline_years),
+            return self._analyze_persistence(
+                current_path=current_path,
+                baseline_path=baseline_path,
+                aoi=aoi,
+                results_dir=results_dir,
                 spatial_completeness=spatial_completeness,
             )
-            confidence = conf["level"]
-            confidence_factors = conf["factors"]
-
-            status = self._classify_zscore(z_current, hotspot_pct)
-            trend = self._compute_trend_zscore(monthly_zscores)
-
-            baseline_buildup_fractions = self._build_seasonal_buildup_fractions(
-                baseline_stats["monthly_buildup_fractions"], BASELINE_YEARS,
-            )
-            chart_data = self._build_seasonal_chart_data(
-                current_stats["monthly_buildup_fractions"], baseline_buildup_fractions,
-                time_range, monthly_zscores, aoi_ha,
-            )
-
-            headline = self._generate_headline(
-                status=status,
-                z_current=z_current,
-                hotspot_pct=hotspot_pct,
-                anomaly_months=anomaly_months,
-                total_months=n_current_bands,
-                value_phrase=f"{current_ha:.0f} ha built-up",
-                indicator_label="Built-up areas",
-                direction_up="expansion",
-                direction_down="contraction",
-            )
-
-            # Write change raster for map rendering
-            change_map_path = os.path.join(results_dir, "buildup_change.tif")
-            self._write_change_raster(current_path, baseline_path, change_map_path)
-
-            self._spatial_data = SpatialData(
-                map_type="raster",
-                label="Built-up Change",
-                colormap="PiYG",
-                vmin=-1,
-                vmax=1,
-            )
-            self._product_raster_path = change_map_path
-            self._render_band = 1
-
-            return ProductResult(
-                product_id=self.id,
-                headline=headline,
-                status=status,
-                trend=trend,
-                confidence=confidence,
-                map_layer_path=change_map_path,
-                chart_data=chart_data,
-                data_source="satellite",
-                anomaly_months=anomaly_months,
-                z_score_current=round(z_current, 2),
-                hotspot_pct=round(hotspot_pct, 1),
-                confidence_factors=confidence_factors,
-                summary=(
-                    f"Built-up area covers {current_frac*100:.1f}% of the AOI "
-                    f"({current_ha:.0f} ha), mean NDBI {current_mean:.3f} "
-                    f"(z-score {z_current:+.1f} vs seasonal baseline). "
-                    f"{anomaly_months} of {n_current_bands} months show significant anomalies. "
-                    f"{hotspot_pct:.0f}% of AOI has statistically significant change. "
-                    f"Pixel-level NDBI analysis at {BUILDUP_RESOLUTION_M}m resolution."
-                ),
-                methodology=(
-                    f"Sentinel-2 L2A pixel-level NDBI = (B11 \u2212 B08) / (B11 + B08). "
-                    f"Built-up classified as NDBI > {NDBI_THRESHOLD}. "
-                    f"Cloud-masked using SCL band. "
-                    f"Monthly median composites at {BUILDUP_RESOLUTION_M}m native resolution. "
-                    f"Baseline: {BASELINE_YEARS}-year seasonal baselines (per calendar month). "
-                    f"Anomaly detection via z-scores (threshold: \u00b1{ZSCORE_THRESHOLD}). "
-                    f"Processed via CDSE openEO batch jobs."
-                ),
-                limitations=[
-                    f"Resampled to {BUILDUP_RESOLUTION_M}m \u2014 detects settlement extent, not individual buildings.",
-                    "NDBI may confuse bare rock/sand with built-up in arid landscapes.",
-                    "Seasonal vegetation cycles can cause false positives at settlement fringes.",
-                    "For building-level analysis, the SR4S pipeline (GPU-dependent) would be needed.",
-                    "Z-score anomalies assume baseline is representative of normal conditions.",
-                ],
-            )
         else:
             # Degraded mode — no baseline
             z_current = 0.0
@@ -415,105 +290,108 @@ class BuiltupProduct(BaseProduct):
         )
 
         self._true_color_path = true_color_path
+        spatial_completeness = self._compute_spatial_completeness(current_path)
 
-        # --- Seasonal baseline analysis ---
-        current_stats = self._compute_stats(current_path)
-        baseline_stats = self._compute_stats(baseline_path)
-        current_mean = current_stats["overall_mean"]
-        current_frac = current_stats["overall_buildup_fraction"]
-        n_current_bands = current_stats["valid_months"]
+        return self._analyze_persistence(
+            current_path=current_path,
+            baseline_path=baseline_path,
+            aoi=aoi,
+            results_dir=results_dir,
+            spatial_completeness=spatial_completeness,
+        )
+
+    def _analyze_persistence(
+        self,
+        current_path: str,
+        baseline_path: str,
+        aoi: AOI,
+        results_dir: str,
+        spatial_completeness: float,
+    ) -> ProductResult:
+        """Shared persistence-based analysis for built-up change.
+
+        Used by both the batch harvest path and the non-batch process path.
+        Removes the monthly z-score approach (which tracks seasonal
+        vegetation cycles) and replaces it with a persistent-mask
+        comparison between the current and baseline periods.
+        """
         aoi_ha = aoi.area_km2 * 100  # km² → hectares
-        current_ha = current_frac * aoi_ha
-        baseline_frac = baseline_stats["overall_buildup_fraction"]
-        baseline_ha = baseline_frac * aoi_ha
 
-        spatial_completeness = self._compute_spatial_completeness(current_path)
+        current_mask, current_persist_frac, n_current_months, _ = (
+            self._compute_persistent_buildup_mask(current_path)
+        )
+        baseline_mask, baseline_persist_frac, n_baseline_months, _ = (
+            self._compute_persistent_buildup_mask(baseline_path)
+        )
 
-        seasonal_stats = compute_seasonal_stats_aoi(baseline_path, n_years=BASELINE_YEARS)
-        start_month = time_range.start.month
-        most_recent_month = ((start_month + n_current_bands - 2) % 12) + 1
+        current_ha = current_persist_frac * aoi_ha
+        baseline_ha = baseline_persist_frac * aoi_ha
 
-        # Z-score for overall current mean NDBI vs seasonal baseline
-        if most_recent_month in seasonal_stats and seasonal_stats[most_recent_month]["n_years"] > 0:
-            s = seasonal_stats[most_recent_month]
-            z_current = compute_zscore(current_mean, s["mean"], s["std"], MIN_STD_BUILDUP)
+        if baseline_persist_frac > 1e-6:
+            change_pct = safe_float(
+                (current_persist_frac - baseline_persist_frac)
+                / baseline_persist_frac * 100.0
+            )
         else:
-            z_current = 0.0
+            change_pct = 0.0
 
-        # Per-month z-scores and anomaly count
-        anomaly_months = 0
-        monthly_zscores = []
-        for i, val in enumerate(current_stats["monthly_means"]):
-            cal_month = ((start_month + i - 1) % 12) + 1
-            if cal_month in seasonal_stats and seasonal_stats[cal_month]["n_years"] > 0:
-                z = compute_zscore(val, seasonal_stats[cal_month]["mean"],
-                                   seasonal_stats[cal_month]["std"], MIN_STD_BUILDUP)
-                monthly_zscores.append(z)
-                if abs(z) > ZSCORE_THRESHOLD:
-                    anomaly_months += 1
-            else:
-                monthly_zscores.append(0.0)
-
-        # Pixel-level hotspot detection
-        month_map = group_bands_by_calendar_month(baseline_stats["valid_months_total"], BASELINE_YEARS)
-        hotspot_pct = 0.0
-        self._zscore_raster = None
-        self._hotspot_mask = None
-        if most_recent_month in month_map and len(month_map[most_recent_month]) > 0:
-            pixel_stats = compute_seasonal_stats_pixel(baseline_path, month_map[most_recent_month])
-            with rasterio.open(current_path) as src:
-                current_band_idx = min(n_current_bands, src.count)
-                current_data = src.read(current_band_idx).astype(np.float32)
-                if src.nodata is not None:
-                    current_data[current_data == src.nodata] = np.nan
-            z_raster = compute_zscore_raster(current_data, pixel_stats["mean"],
-                                              pixel_stats["std"], MIN_STD_BUILDUP)
-            hotspot_mask, hotspot_pct = detect_hotspots(z_raster, ZSCORE_THRESHOLD)
-            self._zscore_raster = z_raster
-            self._hotspot_mask = hotspot_mask
-
-        # Four-factor confidence scoring
-        baseline_depth = sum(1 for m in range(1, 13)
-                             if m in seasonal_stats and seasonal_stats[m]["n_years"] > 0)
-        mean_baseline_years = (sum(seasonal_stats[m]["n_years"] for m in range(1, 13)
-                                   if m in seasonal_stats) / max(baseline_depth, 1))
-        conf = compute_confidence(
-            valid_months=n_current_bands,
-            baseline_years_with_data=int(mean_baseline_years),
-            spatial_completeness=spatial_completeness,
+        status = self._classify_buildup_change(change_pct)
+        trend = (
+            TrendDirection.DETERIORATING if change_pct > CHANGE_AMBER_PCT
+            else (TrendDirection.IMPROVING if change_pct < -CHANGE_AMBER_PCT
+                  else TrendDirection.STABLE)
         )
-        confidence = conf["level"]
-        confidence_factors = conf["factors"]
 
-        status = self._classify_zscore(z_current, hotspot_pct)
-        trend = self._compute_trend_zscore(monthly_zscores)
-        baseline_buildup_fractions = self._build_seasonal_buildup_fractions(
-            baseline_stats["monthly_buildup_fractions"], BASELINE_YEARS,
+        change_raster = current_mask - baseline_mask
+        newly_built_pct = safe_float(
+            np.sum(change_raster > 0) / change_raster.size * 100.0
         )
-        chart_data = self._build_seasonal_chart_data(
-            current_stats["monthly_buildup_fractions"], baseline_buildup_fractions,
-            time_range, monthly_zscores, aoi_ha,
+        lost_built_pct = safe_float(
+            np.sum(change_raster < 0) / change_raster.size * 100.0
         )
+        hotspot_pct = newly_built_pct + lost_built_pct
+        self._zscore_raster = change_raster.astype(np.float32) * 3.0
+        self._hotspot_mask = np.abs(change_raster) > 0.5
 
-        headline = self._generate_headline(
-            status=status,
-            z_current=z_current,
-            hotspot_pct=hotspot_pct,
-            anomaly_months=anomaly_months,
-            total_months=n_current_bands,
-            value_phrase=f"{current_ha:.0f} ha built-up",
-            indicator_label="Built-up areas",
-            direction_up="expansion",
-            direction_down="contraction",
+        conf = compute_confidence(
+            valid_months=n_current_months,
+            baseline_years_with_data=max(1, n_baseline_months // 12),
+            spatial_completeness=spatial_completeness,
         )
 
-        # Write change raster for map rendering
+        chart_data = {
+            "dates": ["Baseline period", "Current period"],
+            "values": [round(baseline_ha, 1), round(current_ha, 1)],
+            "baseline_mean": [round(baseline_ha, 1), round(baseline_ha, 1)],
+            "baseline_min": [round(baseline_ha, 1), round(baseline_ha, 1)],
+            "baseline_max": [round(baseline_ha, 1), round(baseline_ha, 1)],
+            "anomaly_flags": [False, abs(change_pct) >= CHANGE_AMBER_PCT],
+            "label": "Persistent built-up area (hectares)",
+        }
+
+        if status == StatusLevel.GREEN:
+            headline = (
+                f"Built-up areas stable ({current_ha:.0f} ha, "
+                f"{change_pct:+.1f}% vs baseline)."
+            )
+        else:
+            severity = "Major" if status == StatusLevel.RED else "Moderate"
+            direction = "expansion" if change_pct > 0 else "contraction"
+            headline = (
+                f"Built-up areas: {severity.lower()} {direction} "
+                f"({change_pct:+.1f}%, now {current_ha:.0f} ha)."
+            )
+
         change_map_path = os.path.join(results_dir, "buildup_change.tif")
-        self._write_change_raster(current_path, baseline_path, change_map_path)
+        with rasterio.open(current_path) as src:
+            profile = src.profile.copy()
+        profile.update(count=1, dtype="float32")
+        with rasterio.open(change_map_path, "w", **profile) as dst:
+            dst.write(change_raster.astype(np.float32), 1)
 
         self._spatial_data = SpatialData(
             map_type="raster",
-            label="Built-up Change",
+            label="Built-up change (new / lost)",
             colormap="PiYG",
             vmin=-1,
             vmax=1,
@@ -526,40 +404,150 @@ class BuiltupProduct(BaseProduct):
             headline=headline,
             status=status,
             trend=trend,
-            confidence=confidence,
+            confidence=conf["level"],
             map_layer_path=change_map_path,
             chart_data=chart_data,
             data_source="satellite",
-            anomaly_months=anomaly_months,
-            z_score_current=round(z_current, 2),
+            anomaly_months=0,
+            z_score_current=round(change_pct, 2),
             hotspot_pct=round(hotspot_pct, 1),
-            confidence_factors=confidence_factors,
+            confidence_factors=conf["factors"],
             summary=(
-                f"Built-up area covers {current_frac*100:.1f}% of the AOI "
-                f"({current_ha:.0f} ha), mean NDBI {current_mean:.3f} "
-                f"(z-score {z_current:+.1f} vs seasonal baseline). "
-                f"{anomaly_months} of {n_current_bands} months show significant anomalies. "
-                f"{hotspot_pct:.0f}% of AOI has statistically significant change. "
-                f"Pixel-level NDBI analysis at {BUILDUP_RESOLUTION_M}m resolution."
+                f"Persistent built-up area: {current_ha:.0f} ha "
+                f"({current_persist_frac*100:.1f}% of AOI). "
+                f"Baseline ({BASELINE_YEARS}-year average): {baseline_ha:.0f} ha "
+                f"({baseline_persist_frac*100:.1f}%). "
+                f"Change: {change_pct:+.1f}% "
+                f"({newly_built_pct:.1f}% newly built-up, {lost_built_pct:.1f}% lost). "
+                f"Classification uses persistence filter: pixel must be "
+                f"built-up in ≥{int(PERSISTENCE_MIN_FRAC*100)}% of months."
             ),
             methodology=(
-                f"Sentinel-2 L2A pixel-level NDBI = (B11 \u2212 B08) / (B11 + B08). "
-                f"Built-up classified as NDBI > {NDBI_THRESHOLD}. "
+                f"Sentinel-2 L2A: NDBI = (B11 − B08) / (B11 + B08); "
+                f"NDVI = (B08 − B04) / (B08 + B04). "
+                f"A pixel is classified as persistently built-up if "
+                f"NDBI > {NDBI_THRESHOLD} AND NDVI < {NDVI_BUILDUP_MAX} "
+                f"in at least {int(PERSISTENCE_MIN_FRAC*100)}% of valid months "
+                f"across the analysis period. This removes seasonal "
+                f"false-positives from bare soil in dry seasons. "
+                f"Baseline: {BASELINE_YEARS} years preceding the current period. "
+                f"Status thresholds: |change|≥{CHANGE_RED_PCT}% → RED, "
+                f"≥{CHANGE_AMBER_PCT}% → AMBER. "
                 f"Cloud-masked using SCL band. "
-                f"Monthly median composites at {BUILDUP_RESOLUTION_M}m native resolution. "
-                f"Baseline: {BASELINE_YEARS}-year seasonal baselines (per calendar month). "
-                f"Anomaly detection via z-scores (threshold: \u00b1{ZSCORE_THRESHOLD}). "
-                f"Processed server-side via CDSE openEO."
+                f"{BUILDUP_RESOLUTION_M}m native resolution. "
+                f"Processed via CDSE openEO batch jobs."
             ),
             limitations=[
-                f"Resampled to {BUILDUP_RESOLUTION_M}m \u2014 detects settlement extent, not individual buildings.",
-                "NDBI may confuse bare rock/sand with built-up in arid landscapes.",
-                "Seasonal vegetation cycles can cause false positives at settlement fringes.",
-                "For building-level analysis, the SR4S pipeline (GPU-dependent) would be needed.",
-                "Z-score anomalies assume baseline is representative of normal conditions.",
+                f"Resampled to {BUILDUP_RESOLUTION_M}m — detects settlement extent, not individual buildings.",
+                "NDBI-based classification can confuse bare rock/sand with built-up; the NDVI mask and persistence filter reduce but do not eliminate this.",
+                "For sparse, low-density or newly-built structures, detection is unreliable below a few hundred m².",
+                "Persistent-mask approach cannot capture sub-annual growth; use annual windows for tighter monitoring.",
+                "For building-level analysis, a dedicated footprint extraction pipeline (e.g., GHSL, WSF, SR4S) would be needed.",
             ],
         )
 
+    @staticmethod
+    def _classify_buildup_change(change_pct: float) -> StatusLevel:
+        """Classify built-up change using absolute % change, not z-score.
+
+        Z-scores are unstable for built-up because monthly NDBI
+        classification cycles with vegetation. A direct percentage
+        comparison against the baseline persistence mask is more honest.
+        """
+        abs_change = abs(safe_float(change_pct))
+        if abs_change >= CHANGE_RED_PCT:
+            return StatusLevel.RED
+        if abs_change >= CHANGE_AMBER_PCT:
+            return StatusLevel.AMBER
+        return StatusLevel.GREEN
+
+    @staticmethod
+    def _compute_persistent_buildup_mask(tif_path: str) -> tuple[np.ndarray, float, int, dict]:
+        """Compute a persistence-based built-up mask from paired NDBI/NDVI bands.
+
+        A pixel is classified as *persistently* built-up only if it satisfies
+        the combined condition (NDBI > NDBI_THRESHOLD AND NDVI < NDVI_BUILDUP_MAX)
+        in at least PERSISTENCE_MIN_FRAC of the valid months in the stack.
+
+        This removes the seasonal cycling artifact where bare soil in the
+        dry season gets classified as built-up in winter but not summer.
+
+        Returns
+        -------
+        mask : np.ndarray
+            Binary 2D mask (1 = persistently built-up, 0 = not).
+        fraction : float
+            Fraction of non-nodata pixels classified as persistent built-up.
+        valid_months : int
+            Number of valid monthly observations found in the stack.
+        profile : dict
+            Rasterio profile of the source TIF (for downstream writes).
+        """
+        with rasterio.open(tif_path) as src:
+            count = src.count
+            profile = src.profile.copy()
+            paired = count >= 2 and count % 2 == 0
+            n_months = count // 2 if paired else count
+
+            # Accumulate: number of months each pixel satisfies built-up condition
+            # and number of months the pixel had any valid observation at all.
+            built_counts: np.ndarray | None = None
+            obs_counts: np.ndarray | None = None
+            nodata = src.nodata
+
+            for m in range(n_months):
+                if paired:
+                    ndbi = src.read(m * 2 + 1).astype(np.float32)
+                    ndvi = src.read(m * 2 + 2).astype(np.float32)
+                else:
+                    ndbi = src.read(m + 1).astype(np.float32)
+                    ndvi = None
+
+                if nodata is not None:
+                    valid = ndbi != nodata
+                else:
+                    valid = ~np.isnan(ndbi)
+
+                if paired and ndvi is not None:
+                    cond = (ndbi > NDBI_THRESHOLD) & (ndvi < NDVI_BUILDUP_MAX) & valid
+                else:
+                    # Legacy single-band fallback: NDBI only
+                    cond = (ndbi > NDBI_THRESHOLD) & valid
+
+                if built_counts is None:
+                    built_counts = cond.astype(np.uint16)
+                    obs_counts = valid.astype(np.uint16)
+                else:
+                    built_counts += cond.astype(np.uint16)
+                    obs_counts += valid.astype(np.uint16)
+
+        if built_counts is None or obs_counts is None:
+            return np.zeros((1, 1), dtype=np.float32), 0.0, 0, profile
+
+        # A pixel counts as persistently built-up if:
+        #   - it has ≥1 valid observation, AND
+        #   - built_count / obs_count ≥ PERSISTENCE_MIN_FRAC
+        with np.errstate(divide="ignore", invalid="ignore"):
+            persistent_frac_per_pixel = np.where(
+                obs_counts > 0,
+                built_counts.astype(np.float32) / obs_counts.astype(np.float32),
+                0.0,
+            )
+        mask = (
+            (obs_counts >= 1)
+            & (persistent_frac_per_pixel >= PERSISTENCE_MIN_FRAC)
+        ).astype(np.float32)
+
+        # Area fraction: persistent built-up pixels / all pixels with any valid data
+        any_valid = obs_counts >= 1
+        total_valid = int(np.sum(any_valid))
+        if total_valid > 0:
+            fraction = safe_float(np.sum(mask) / total_valid)
+        else:
+            fraction = 0.0
+
+        return mask, fraction, n_months, profile
+
     @staticmethod
     def _compute_stats(tif_path: str) -> dict[str, Any]:
         """Extract monthly built-up fraction and raw NDBI stats from GeoTIFF.

app/eo_products/ndvi.py

@@ -200,7 +200,11 @@ class NdviProduct(BaseProduct):
             confidence = conf["level"]
             confidence_factors = conf["factors"]
 
-            status = self._classify_zscore(z_current, hotspot_pct)
+            status = self._classify_zscore(
+            z_current, hotspot_pct,
+            anomaly_months=anomaly_months,
+            total_months=n_current_bands,
+        )
             trend = self._compute_trend_zscore(monthly_zscores)
 
             chart_data = self._build_seasonal_chart_data(
@@ -397,7 +401,11 @@ class NdviProduct(BaseProduct):
         confidence = conf["level"]
         confidence_factors = conf["factors"]
 
-        status = self._classify_zscore(z_current, hotspot_pct)
+        status = self._classify_zscore(
+            z_current, hotspot_pct,
+            anomaly_months=anomaly_months,
+            total_months=n_current_bands,
+        )
         trend = self._compute_trend_zscore(monthly_zscores)
         chart_data = self._build_seasonal_chart_data(
             current_stats["monthly_means"], seasonal_stats, time_range, monthly_zscores,

app/eo_products/sar.py

@@ -43,6 +43,7 @@ logger = logging.getLogger(__name__)
 BASELINE_YEARS = 5
 CHANGE_THRESHOLD_DB = 3.0  # dB change considered significant
 FLOOD_SIGMA = 2.0  # Standard deviations below baseline mean
+BASELINE_DRIFT_THRESHOLD = 0.4  # If >40% of months anomalous, treat as baseline drift not finding
 
 
 class SarProduct(BaseProduct):
@@ -226,25 +227,45 @@ class SarProduct(BaseProduct):
             ))
             hotspot_pct = safe_float(hotspot_pct)
 
-            status = self._classify_zscore(z_current, hotspot_pct)
-            trend = self._compute_trend_zscore(monthly_zscores)
+            # Baseline drift detection: if a large majority of months are
+            # flagged as anomalous, this is almost certainly a baseline
+            # calibration shift (S1 IPF version change, orbit geometry) or
+            # regime shift, not a real per-month anomaly pattern.
+            drift_frac = (anomaly_months / n_current_bands) if n_current_bands > 0 else 0.0
+            baseline_drift_detected = drift_frac >= BASELINE_DRIFT_THRESHOLD
+
+            if baseline_drift_detected:
+                status = StatusLevel.AMBER
+                trend = TrendDirection.STABLE
+                headline = (
+                    f"Ground surface: baseline may be unreliable — "
+                    f"{anomaly_months} of {n_current_bands} months diverge from the "
+                    f"5-year baseline (possible sensor calibration or regime shift, "
+                    f"not a per-month anomaly pattern)."
+                )
+            else:
+                status = self._classify_zscore(
+                    z_current, hotspot_pct,
+                    anomaly_months=anomaly_months,
+                    total_months=n_current_bands,
+                )
+                trend = self._compute_trend_zscore(monthly_zscores)
+                headline = self._generate_headline(
+                    status=status,
+                    z_current=z_current,
+                    hotspot_pct=hotspot_pct,
+                    anomaly_months=anomaly_months,
+                    total_months=n_current_bands,
+                    value_phrase=f"backscatter {current_mean:.1f} dB",
+                    indicator_label="Ground surface",
+                    direction_up="brightening (drying or new structures)",
+                    direction_down="darkening (possible flooding or moisture)",
+                )
 
             chart_data = self._build_seasonal_chart_data(
                 current_stats["monthly_vv_means"], seasonal_stats, time_range, monthly_zscores,
             )
 
-            headline = self._generate_headline(
-                status=status,
-                z_current=z_current,
-                hotspot_pct=hotspot_pct,
-                anomaly_months=anomaly_months,
-                total_months=n_current_bands,
-                value_phrase=f"backscatter {current_mean:.1f} dB",
-                indicator_label="Ground surface",
-                direction_up="brightening (drying or new structures)",
-                direction_down="darkening (possible flooding or moisture)",
-            )
-
             change_map_path = os.path.join(results_dir, "sar_change.tif")
             self._write_change_raster(current_path, baseline_path, change_map_path)
 
@@ -259,16 +280,30 @@ class SarProduct(BaseProduct):
             self._render_band = 1
             map_layer_path = change_map_path
 
-            summary = (
-                f"Mean VV backscatter: {current_mean:.1f} dB (z-score {z_current:+.1f} vs seasonal baseline). "
-                f"{anomaly_months} of {n_current_bands} months show significant anomalies. "
-                f"{hotspot_pct:.0f}% of AOI has statistically significant change. "
-                f"Mean VV change: {change_db:+.1f} dB. "
-                f"{change_pct:.1f}% area with >{CHANGE_THRESHOLD_DB} dB change. "
-                f"{flood_months} month(s) with potential flood signals. "
-                f"Pixel-level analysis at {SAR_RESOLUTION_M}m resolution."
-            )
-            extra_limitations: list[str] = []
+            if baseline_drift_detected:
+                summary = (
+                    f"Baseline instability detected: {anomaly_months} of "
+                    f"{n_current_bands} months diverge from the {BASELINE_YEARS}-year "
+                    f"baseline. This pattern is more consistent with Sentinel-1 "
+                    f"processor/calibration change or regime shift than a "
+                    f"per-month anomaly signal. Per-month z-scores not reported as "
+                    f"reliable indicators. Mean VV: {current_mean:.1f} dB "
+                    f"(change from baseline: {change_db:+.1f} dB)."
+                )
+                extra_limitations: list[str] = [
+                    "Baseline instability flagged — the 5-year SAR baseline may include calibration changes or regime shifts. Re-check with a shorter baseline window or consistent relative-orbit filtering before interpreting.",
+                ]
+            else:
+                summary = (
+                    f"Mean VV backscatter: {current_mean:.1f} dB (z-score {z_current:+.1f} vs seasonal baseline). "
+                    f"{anomaly_months} of {n_current_bands} months show significant anomalies. "
+                    f"{hotspot_pct:.0f}% of AOI has statistically significant change. "
+                    f"Mean VV change: {change_db:+.1f} dB. "
+                    f"{change_pct:.1f}% area with >{CHANGE_THRESHOLD_DB} dB change. "
+                    f"{flood_months} month(s) with potential flood signals. "
+                    f"Pixel-level analysis at {SAR_RESOLUTION_M}m resolution."
+                )
+                extra_limitations: list[str] = []
         else:
             # Degraded mode — no baseline
             z_current = 0.0
@@ -332,6 +367,8 @@ class SarProduct(BaseProduct):
                 f"Anomaly detection via z-scores (threshold: ±{ZSCORE_THRESHOLD}). "
                 f"Change detection: >{CHANGE_THRESHOLD_DB} dB difference vs baseline. "
                 f"Flood mapping: VV < baseline_mean − {FLOOD_SIGMA}σ. "
+                f"Baseline drift check: if >{int(BASELINE_DRIFT_THRESHOLD*100)}% of months are anomalous, "
+                f"the finding is flagged as likely baseline instability rather than a real signal. "
                 f"Processed via CDSE openEO batch jobs."
             ),
             limitations=[
@@ -494,25 +531,41 @@ class SarProduct(BaseProduct):
         ))
         hotspot_pct = safe_float(hotspot_pct)
 
-        status = self._classify_zscore(z_current, hotspot_pct)
-        trend = self._compute_trend_zscore(monthly_zscores)
+        drift_frac = (anomaly_months / n_current_bands) if n_current_bands > 0 else 0.0
+        baseline_drift_detected = drift_frac >= BASELINE_DRIFT_THRESHOLD
+
+        if baseline_drift_detected:
+            status = StatusLevel.AMBER
+            trend = TrendDirection.STABLE
+            headline = (
+                f"Ground surface: baseline may be unreliable — "
+                f"{anomaly_months} of {n_current_bands} months diverge from the "
+                f"{BASELINE_YEARS}-year baseline (possible sensor calibration "
+                f"or regime shift, not a per-month anomaly pattern)."
+            )
+        else:
+            status = self._classify_zscore(
+                z_current, hotspot_pct,
+                anomaly_months=anomaly_months,
+                total_months=n_current_bands,
+            )
+            trend = self._compute_trend_zscore(monthly_zscores)
+            headline = self._generate_headline(
+                status=status,
+                z_current=z_current,
+                hotspot_pct=hotspot_pct,
+                anomaly_months=anomaly_months,
+                total_months=n_current_bands,
+                value_phrase=f"backscatter {current_mean:.1f} dB",
+                indicator_label="Ground surface",
+                direction_up="brightening (drying or new structures)",
+                direction_down="darkening (possible flooding or moisture)",
+            )
 
         chart_data = self._build_seasonal_chart_data(
             current_stats["monthly_vv_means"], seasonal_stats, time_range, monthly_zscores,
         )
 
-        headline = self._generate_headline(
-            status=status,
-            z_current=z_current,
-            hotspot_pct=hotspot_pct,
-            anomaly_months=anomaly_months,
-            total_months=n_current_bands,
-            value_phrase=f"backscatter {current_mean:.1f} dB",
-            indicator_label="Ground surface",
-            direction_up="brightening (drying or new structures)",
-            direction_down="darkening (possible flooding or moisture)",
-        )
-
         # Store raster path for map rendering — write a change map
         change_map_path = os.path.join(results_dir, "sar_change.tif")
         self._write_change_raster(current_path, baseline_path, change_map_path)
@@ -541,13 +594,24 @@ class SarProduct(BaseProduct):
             hotspot_pct=round(hotspot_pct, 1),
             confidence_factors=confidence_factors,
             summary=(
-                f"Mean VV backscatter: {current_mean:.1f} dB (z-score {z_current:+.1f} vs seasonal baseline). "
-                f"{anomaly_months} of {n_current_bands} months show significant anomalies. "
-                f"{hotspot_pct:.0f}% of AOI has statistically significant change. "
-                f"Mean VV change: {change_db:+.1f} dB. "
-                f"{change_pct:.1f}% area with >{CHANGE_THRESHOLD_DB} dB change. "
-                f"{flood_months} month(s) with potential flood signals. "
-                f"Pixel-level analysis at {SAR_RESOLUTION_M}m resolution."
+                (
+                    f"Baseline instability detected: {anomaly_months} of "
+                    f"{n_current_bands} months diverge from the {BASELINE_YEARS}-year "
+                    f"baseline. This pattern is more consistent with Sentinel-1 "
+                    f"processor/calibration change or regime shift than per-month "
+                    f"anomalies. Mean VV: {current_mean:.1f} dB "
+                    f"(change from baseline: {change_db:+.1f} dB)."
+                )
+                if baseline_drift_detected
+                else (
+                    f"Mean VV backscatter: {current_mean:.1f} dB (z-score {z_current:+.1f} vs seasonal baseline). "
+                    f"{anomaly_months} of {n_current_bands} months show significant anomalies. "
+                    f"{hotspot_pct:.0f}% of AOI has statistically significant change. "
+                    f"Mean VV change: {change_db:+.1f} dB. "
+                    f"{change_pct:.1f}% area with >{CHANGE_THRESHOLD_DB} dB change. "
+                    f"{flood_months} month(s) with potential flood signals. "
+                    f"Pixel-level analysis at {SAR_RESOLUTION_M}m resolution."
+                )
             ),
             methodology=(
                 f"Sentinel-1 GRD IW VV/VH polarizations, ascending orbit. "
@@ -557,6 +621,8 @@ class SarProduct(BaseProduct):
                 f"Anomaly detection via z-scores (threshold: ±{ZSCORE_THRESHOLD}). "
                 f"Change detection: >{CHANGE_THRESHOLD_DB} dB difference vs baseline. "
                 f"Flood mapping: VV < baseline_mean − {FLOOD_SIGMA}σ. "
+                f"Baseline drift check: if >{int(BASELINE_DRIFT_THRESHOLD*100)}% of months are anomalous, "
+                f"the finding is flagged as likely baseline instability rather than a real signal. "
                 f"Processed via CDSE openEO."
             ),
             limitations=[
@@ -565,7 +631,10 @@ class SarProduct(BaseProduct):
                 "Sentinel-1 coverage over East Africa can be inconsistent.",
                 "VV decrease may indicate flooding, moisture, or vegetation change — not uniquely flood.",
                 "Z-score anomalies assume baseline is representative of normal conditions.",
-            ],
+            ] + (
+                ["Baseline instability flagged — re-check with a shorter baseline window or consistent relative-orbit filtering before drawing conclusions."]
+                if baseline_drift_detected else []
+            ),
         )
 
     # ------------------------------------------------------------------

app/eo_products/water.py

@@ -205,7 +205,12 @@ class WaterProduct(BaseProduct):
             confidence = conf["level"]
             confidence_factors = conf["factors"]
 
-            status = self._classify_zscore(z_current, hotspot_pct)
+            status = self._classify_zscore(
+            z_current, hotspot_pct,
+            anomaly_months=anomaly_months,
+            total_months=n_current_bands,
+            min_coverage_pct=current_frac * 100.0,
+        )
             trend = self._compute_trend_zscore(monthly_zscores)
 
             baseline_seasonal_fractions = self._build_seasonal_water_fractions(
@@ -410,7 +415,12 @@ class WaterProduct(BaseProduct):
         confidence = conf["level"]
         confidence_factors = conf["factors"]
 
-        status = self._classify_zscore(z_current, hotspot_pct)
+        status = self._classify_zscore(
+            z_current, hotspot_pct,
+            anomaly_months=anomaly_months,
+            total_months=n_current_bands,
+            min_coverage_pct=current_frac * 100.0,
+        )
         trend = self._compute_trend_zscore(monthly_zscores)
         baseline_seasonal_fractions = self._build_seasonal_water_fractions(
             baseline_stats["monthly_water_fractions"], BASELINE_YEARS,

app/outputs/maps.py

@@ -403,14 +403,15 @@ def render_overview_map(
     # AOI outline
     _draw_aoi_rect(ax, aoi, INK)
 
-    # Title and date range
-    if title:
+    # Title (with date range on a second line if provided) — placed above the
+    # axes so the date no longer collides with the x-axis tick labels.
+    if title and date_range:
+        full_title = f"{title}\n{date_range}"
+        ax.set_title(full_title, fontsize=9, color=INK, fontweight="bold", pad=8)
+    elif title:
         ax.set_title(title, fontsize=10, color=INK, fontweight="bold", pad=8)
-    if date_range:
-        ax.text(
-            0.5, -0.05, date_range,
-            transform=ax.transAxes, ha="center", fontsize=7, color=INK_MUTED,
-        )
+    elif date_range:
+        ax.set_title(date_range, fontsize=8, color=INK_MUTED, pad=6)
 
     ax.set_xlim(extent[0], extent[1])
     ax.set_ylim(extent[2], extent[3])

app/outputs/narrative.py

@@ -42,22 +42,66 @@ def get_verify_suggestion(product_id: str, status: StatusLevel) -> str:
         return ""
     return _VERIFY_SUGGESTIONS.get((product_id, status), "")
 
-# --- Cross-indicator pattern rules ---
-_CROSS_PATTERNS: list[tuple[dict[str, set[StatusLevel]], str]] = [
+# --- Direction-aware cross-indicator pattern rules ---
+#
+# Each rule describes a pattern of (indicator_id, required_direction) pairs.
+# Direction is checked against `z_score_current` sign: "down" means z<-1,
+# "up" means z>+1, "any" means any non-GREEN status regardless of direction.
+#
+# Only triggered when ALL constraints are met. This replaces the previous
+# status-only matcher which produced sentences like "vegetation loss
+# coincides with settlement expansion" even when both were actually going
+# the opposite direction.
+_CROSS_PATTERNS: list[tuple[list[tuple[str, str]], str]] = [
     (
-        {"ndvi": {StatusLevel.RED, StatusLevel.AMBER}, "buildup": {StatusLevel.RED, StatusLevel.AMBER}},
-        "Vegetation loss coincides with settlement expansion, indicating possible land-use conversion.",
+        [("ndvi", "down"), ("buildup", "up")],
+        "Vegetation loss coincides with built-up expansion, indicating possible land-use conversion.",
     ),
     (
-        {"ndvi": {StatusLevel.RED, StatusLevel.AMBER}, "sar": {StatusLevel.RED, StatusLevel.AMBER}},
-        "Vegetation decline paired with SAR backscatter anomalies may indicate flood damage or soil saturation.",
+        [("ndvi", "up"), ("buildup", "down")],
+        "Vegetation recovery coincides with apparent built-up contraction — "
+        "in conflict or displacement contexts this can reflect abandoned land "
+        "returning to vegetation.",
     ),
     (
-        {"water": {StatusLevel.RED, StatusLevel.AMBER}, "sar": {StatusLevel.RED, StatusLevel.AMBER}},
-        "Increased water extent and SAR signal changes suggest flooding or waterlogging.",
+        [("ndvi", "down"), ("sar", "any")],
+        "Vegetation decline paired with ground-surface anomalies may indicate "
+        "damage, soil saturation, or disturbance.",
+    ),
+    (
+        [("water", "up"), ("sar", "down")],
+        "Expanded water extent combined with radar darkening suggests "
+        "flooding or waterlogging.",
+    ),
+    (
+        [("water", "down"), ("ndvi", "down")],
+        "Drying water and declining vegetation together indicate drought stress.",
     ),
 ]
 
+
+def _z_direction(r: "ProductResult") -> str:
+    """Return 'up', 'down', or 'stable'.
+
+    Uses z_score_current (which for most indicators is a z-score with
+    threshold 1.0, but for the buildup indicator is a change percentage
+    with threshold 5.0% — matching its AMBER cutoff).
+    """
+    val = float(r.z_score_current or 0.0)
+    threshold = 5.0 if r.product_id == "buildup" else 1.0
+    if val > threshold:
+        return "up"
+    if val < -threshold:
+        return "down"
+    return "stable"
+
+
+def _direction_matches(actual: str, required: str) -> bool:
+    """Match a direction constraint.  'any' matches any non-stable direction."""
+    if required == "any":
+        return actual in ("up", "down")
+    return actual == required
+
 _LEAD_TEMPLATES = {
     StatusLevel.RED: "One or more indicators show major changes that warrant action and ground verification.",
     StatusLevel.AMBER: "One or more indicators show elevated change that should be monitored.",
@@ -73,10 +117,21 @@ def get_interpretation(product_id: str, status: StatusLevel) -> str:
     )
 
 
+def _clean_sentence(text: str) -> str:
+    """Strip trailing period(s) and whitespace so we can re-add exactly one."""
+    return text.rstrip().rstrip(".").rstrip()
+
+
 def generate_narrative(results: Sequence[ProductResult]) -> str:
-    """Generate a cross-indicator narrative paragraph from indicator results."""
+    """Generate a cross-indicator narrative paragraph from indicator results.
+
+    The narrative is *direction-aware* — it checks z-score signs, not just
+    status levels, so it no longer produces sentences like "vegetation loss
+    coincides with settlement expansion" when aggregates actually show
+    greening and contraction.
+    """
     if not results:
-        return "No EO product data available for narrative generation."
+        return "No indicator data available for narrative generation."
 
     parts: list[str] = []
 
@@ -85,20 +140,21 @@ def generate_narrative(results: Sequence[ProductResult]) -> str:
         (r.status for r in results),
         key=lambda s: [StatusLevel.GREEN, StatusLevel.AMBER, StatusLevel.RED].index(s),
     )
-    parts.append(_LEAD_TEMPLATES[worst])
+    parts.append(_clean_sentence(_LEAD_TEMPLATES[worst]) + ".")
 
-    # 2. Per-indicator sentences
+    # 2. Per-indicator sentences — strip any pre-existing trailing period so we
+    #    don't end up with ".." when concatenating.
     for r in results:
-        parts.append(f"{r.headline}.")
+        parts.append(_clean_sentence(r.headline) + ".")
 
-    # 3. Cross-indicator connection
-    result_map = {r.product_id: r.status for r in results}
+    # 3. Direction-aware cross-indicator connection
+    directions = {r.product_id: _z_direction(r) for r in results}
     for required, sentence in _CROSS_PATTERNS:
         if all(
-            ind_id in result_map and result_map[ind_id] in allowed_statuses
-            for ind_id, allowed_statuses in required.items()
+            ind_id in directions and _direction_matches(directions[ind_id], req_dir)
+            for ind_id, req_dir in required
         ):
-            parts.append(sentence)
+            parts.append(_clean_sentence(sentence) + ".")
             break
 
     return " ".join(parts)