feat: add output pipeline for seasonal analysis and compound signals

- Charts: anomaly markers (red rings) on months exceeding z-score
threshold; default y-axis labels per indicator
- Maps: new render_hotspot_map() for z-score change visualization
over true-color base layer
- Narrative: generate_compound_signals_text() for cross-indicator
compound signal descriptions
- Report: anomaly column in summary table, compound signals section,
hotspot maps in indicator blocks, confidence breakdown table in
technical annex
- Worker: generates hotspot maps, detects compound signals via
spatial z-score overlap, passes both to PDF report

Generated with [Claude Code](https://claude.ai/code)
via [Happy](https://happy.engineering)

Co-Authored-By: Claude <noreply@anthropic.com>
Co-Authored-By: Happy <yesreply@happy.engineering>

app/outputs/charts.py

@@ -59,6 +59,16 @@ def render_timeseries_chart(
     dates = chart_data.get("dates", [])
     values = chart_data.get("values", [])
 
+    # Default y-axis labels per indicator
+    if not y_label:
+        _default_labels = {
+            "Vegetation (NDVI)": "NDVI (0\u20131)",
+            "Water Bodies": "Water extent (%)",
+            "SAR Backscatter": "VV backscatter (dB)",
+            "Settlement Extent": "Built-up area (%)",
+        }
+        y_label = _default_labels.get(indicator_name, "")
+
     status_color = STATUS_COLORS[status]
     arrow = TREND_ARROWS[trend]
 
@@ -139,6 +149,18 @@ def render_timeseries_chart(
             alpha=0.15, color=status_color,
         )
 
+        # Anomaly markers — red rings on months with |z| > threshold
+        anomaly_flags = chart_data.get("anomaly_flags")
+        if anomaly_flags and len(anomaly_flags) == len(parsed_dates):
+            anomaly_x = [d for d, f in zip(parsed_dates, anomaly_flags) if f]
+            anomaly_y = [v for v, f in zip(values, anomaly_flags) if f]
+            if anomaly_x:
+                ax.scatter(
+                    anomaly_x, anomaly_y,
+                    s=120, facecolors="none", edgecolors="#B83A2A",
+                    linewidths=2, zorder=4, label="Anomaly",
+                )
+
         if has_monthly_baseline or has_summary_baseline:
             ax.legend(fontsize=7, loc="upper left", framealpha=0.8)
 

app/outputs/maps.py

@@ -301,6 +301,67 @@ def render_raster_map(
     plt.close(fig)
 
 
+def render_hotspot_map(
+    *,
+    true_color_path: str | None,
+    zscore_raster: np.ndarray,
+    hotspot_mask: np.ndarray,
+    extent: list[float],
+    aoi: AOI,
+    status: StatusLevel,
+    output_path: str,
+    label: str = "Z-score",
+) -> None:
+    """Render a change hotspot map: significant pixels over true-color base.
+
+    Only pixels where |z-score| > threshold are shown; non-significant
+    pixels are transparent, letting the true-color base show through.
+    """
+    import rasterio
+
+    fig, ax = plt.subplots(figsize=(6, 5), dpi=200, facecolor=SHELL)
+    ax.set_facecolor(SHELL)
+
+    # True-color base layer
+    if true_color_path is not None:
+        with rasterio.open(true_color_path) as src:
+            rgb = src.read([1, 2, 3]).astype(np.float32)
+            tc_extent = [src.bounds.left, src.bounds.right, src.bounds.bottom, src.bounds.top]
+        rgb_max = max(rgb.max(), 1.0)
+        scale = 3000.0 if rgb_max > 255 else 255.0
+        rgb_normalized = np.clip(rgb / scale, 0, 1).transpose(1, 2, 0)
+        ax.imshow(rgb_normalized, extent=tc_extent, aspect="auto", zorder=0)
+
+    # Hotspot overlay — only significant pixels, masked elsewhere
+    masked_z = np.ma.masked_where(~hotspot_mask, zscore_raster)
+    vmax = min(float(np.nanmax(np.abs(zscore_raster))), 5.0)
+    im = ax.imshow(
+        masked_z, extent=extent, cmap="RdBu_r", alpha=0.8,
+        vmin=-vmax, vmax=vmax, aspect="auto", zorder=1,
+    )
+    cbar = fig.colorbar(im, ax=ax, fraction=0.03, pad=0.04, shrink=0.85)
+    cbar.set_label(f"{label} (decline \u2190 \u2192 increase)", fontsize=7, color=INK_MUTED)
+    cbar.ax.tick_params(labelsize=6, colors=INK_MUTED)
+
+    # AOI outline
+    ax.set_xlim(extent[0], extent[1])
+    ax.set_ylim(extent[2], extent[3])
+    color = STATUS_COLORS[status]
+    _draw_aoi_rect(ax, aoi, color)
+
+    ax.tick_params(labelsize=6, colors=INK_MUTED)
+    ax.set_xlabel("Longitude", fontsize=7, color=INK_MUTED)
+    ax.set_ylabel("Latitude", fontsize=7, color=INK_MUTED)
+
+    for spine in ax.spines.values():
+        spine.set_color(INK_MUTED)
+        spine.set_linewidth(0.5)
+
+    plt.tight_layout()
+    fig.savefig(output_path, dpi=200, bbox_inches="tight", facecolor=SHELL)
+    plt.close(fig)
+
+
 def render_overview_map(
     *,
     true_color_path: str,

app/outputs/narrative.py

@@ -82,3 +82,26 @@ def generate_narrative(results: Sequence[IndicatorResult]) -> str:
             break
 
     return " ".join(parts)
+
+
+def generate_compound_signals_text(signals: list) -> str:
+    """Generate text for compound signal section.
+
+    Parameters
+    ----------
+    signals : list of CompoundSignal objects.
+
+    Returns text describing triggered compound signals.
+    """
+    if not signals:
+        return "No compound signals detected across the indicator set."
+
+    triggered = [s for s in signals if s.triggered]
+    if not triggered:
+        return "No compound signals detected across the indicator set."
+
+    parts = []
+    for s in triggered:
+        parts.append(f"**{s.name.replace('_', ' ').title()}** ({s.confidence}): {s.description}")
+
+    return " ".join(parts)

app/outputs/report.py

@@ -163,6 +163,7 @@ def _indicator_block(
     styles: dict,
     map_path: str = "",
     chart_path: str = "",
+    hotspot_path: str = "",
 ) -> list:
     """Build the flowables for a single indicator section."""
     from app.outputs.narrative import get_interpretation
@@ -217,6 +218,15 @@ def _indicator_block(
         elements.append(img)
         elements.append(Spacer(1, 2 * mm))
 
+    # Hotspot change map (if available)
+    hotspot_exists = hotspot_path and os.path.exists(hotspot_path)
+    if hotspot_exists:
+        from reportlab.platypus import Image as RLImage
+        hotspot_img = RLImage(hotspot_path, width=14 * cm, height=5.5 * cm)
+        hotspot_img.hAlign = "CENTER"
+        elements.append(hotspot_img)
+        elements.append(Spacer(1, 2 * mm))
+
     # What the data shows
     elements.append(Paragraph("<b>What the data shows</b>", styles["body_muted"]))
     elements.append(Paragraph(result.summary, styles["body"]))
@@ -264,8 +274,10 @@ def generate_pdf_report(
     output_path: str,
     summary_map_path: str = "",
     indicator_map_paths: dict[str, str] | None = None,
+    indicator_hotspot_paths: dict[str, str] | None = None,
     overview_score: dict | None = None,
     overview_map_path: str = "",
+    compound_signals: list | None = None,
 ) -> None:
     """Generate a styled PDF report and save to ``output_path``.
 
@@ -425,6 +437,7 @@ def generate_pdf_report(
         Paragraph("<b>Status</b>", styles["body"]),
         Paragraph("<b>Trend</b>", styles["body"]),
         Paragraph("<b>Confidence</b>", styles["body"]),
+        Paragraph("<b>Anomalies</b>", styles["body"]),
         Paragraph("<b>Headline</b>", styles["body"]),
     ]
     summary_rows = [summary_header]
@@ -445,18 +458,20 @@ def generate_pdf_report(
             status_cell,
             Paragraph(result.trend.value.capitalize(), styles["body_muted"]),
             Paragraph(result.confidence.value.capitalize(), styles["body_muted"]),
-            Paragraph(result.headline[:80], styles["body_muted"]),
+            Paragraph(f"{result.anomaly_months}/12", styles["body_muted"]),
+            Paragraph(result.headline[:70], styles["body_muted"]),
         ])
 
     ov_col_w = PAGE_W - 2 * MARGIN
     ov_table = Table(
         summary_rows,
         colWidths=[
-            ov_col_w * 0.15,
-            ov_col_w * 0.10,
-            ov_col_w * 0.12,
-            ov_col_w * 0.13,
-            ov_col_w * 0.50,
+            ov_col_w * 0.14,
+            ov_col_w * 0.09,
+            ov_col_w * 0.11,
+            ov_col_w * 0.11,
+            ov_col_w * 0.09,
+            ov_col_w * 0.46,
         ],
     )
     ov_ts = TableStyle([
@@ -475,6 +490,27 @@ def generate_pdf_report(
     ov_table.setStyle(ov_ts)
     story.append(ov_table)
 
+    # Compound signals section
+    if compound_signals:
+        triggered = [s for s in compound_signals if s.triggered]
+        if triggered:
+            story.append(Spacer(1, 4 * mm))
+            story.append(Paragraph("Compound Signals", styles["section_heading"]))
+            story.append(Spacer(1, 2 * mm))
+            for s in triggered:
+                signal_text = (
+                    f"<b>{s.name.replace('_', ' ').title()}</b> "
+                    f"({s.confidence}): {s.description}"
+                )
+                story.append(Paragraph(signal_text, styles["body"]))
+                story.append(Spacer(1, 2 * mm))
+        else:
+            story.append(Spacer(1, 2 * mm))
+            story.append(Paragraph(
+                "No compound signals detected across the indicator set.",
+                styles["body_muted"],
+            ))
+
     story.append(Spacer(1, 6 * mm))
     story.append(HRFlowable(width="100%", thickness=0.5, color=colors.HexColor("#D8D5CF")))
     story.append(Spacer(1, 4 * mm))
@@ -488,6 +524,7 @@ def generate_pdf_report(
     for result in results:
         indicator_label = _indicator_label(result.indicator_id)
         map_path = (indicator_map_paths or {}).get(result.indicator_id, "")
+        hotspot_path = (indicator_hotspot_paths or {}).get(result.indicator_id, "")
 
         # Auto-detect chart path from output directory
         chart_path = os.path.join(output_dir, f"{result.indicator_id}_chart.png")
@@ -495,7 +532,7 @@ def generate_pdf_report(
             chart_path = ""
 
         block = [Paragraph(indicator_label, styles["section_heading"])]
-        block += _indicator_block(result, styles, map_path=map_path, chart_path=chart_path)
+        block += _indicator_block(result, styles, map_path=map_path, chart_path=chart_path, hotspot_path=hotspot_path)
         story.append(KeepTogether(block))
 
     # ================================================================== #
@@ -516,6 +553,47 @@ def generate_pdf_report(
         )
     story.append(Spacer(1, 4 * mm))
 
+    # Confidence breakdown table
+    story.append(Paragraph("Confidence Breakdown", styles["section_heading"]))
+    story.append(Spacer(1, 2 * mm))
+
+    conf_header = [
+        Paragraph("<b>Indicator</b>", styles["body"]),
+        Paragraph("<b>Temporal</b>", styles["body"]),
+        Paragraph("<b>Obs. Density</b>", styles["body"]),
+        Paragraph("<b>Baseline Depth</b>", styles["body"]),
+        Paragraph("<b>Spatial Compl.</b>", styles["body"]),
+        Paragraph("<b>Overall</b>", styles["body"]),
+    ]
+    conf_rows = [conf_header]
+    for result in results:
+        f = result.confidence_factors
+        if f:
+            conf_rows.append([
+                Paragraph(_indicator_label(result.indicator_id), styles["body_muted"]),
+                Paragraph(f"{f.get('temporal', 0):.2f}", styles["body_muted"]),
+                Paragraph(f"{f.get('observation_density', 0):.2f}", styles["body_muted"]),
+                Paragraph(f"{f.get('baseline_depth', 0):.2f}", styles["body_muted"]),
+                Paragraph(f"{f.get('spatial_completeness', 0):.2f}", styles["body_muted"]),
+                Paragraph(result.confidence.value.capitalize(), styles["body_muted"]),
+            ])
+
+    if len(conf_rows) > 1:
+        conf_col_w = PAGE_W - 2 * MARGIN
+        conf_table = Table(
+            conf_rows,
+            colWidths=[conf_col_w * 0.18] + [conf_col_w * 0.164] * 5,
+        )
+        conf_table.setStyle(TableStyle([
+            ("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#E8E6E0")),
+            ("GRID", (0, 0), (-1, -1), 0.3, colors.HexColor("#D8D5CF")),
+            ("TOPPADDING", (0, 0), (-1, -1), 3),
+            ("BOTTOMPADDING", (0, 0), (-1, -1), 3),
+            ("LEFTPADDING", (0, 0), (-1, -1), 4),
+        ]))
+        story.append(conf_table)
+    story.append(Spacer(1, 4 * mm))
+
     # Limitations subsection — deduplicated across all indicators
     all_limitations: list[str] = []
     seen: set[str] = set()

app/worker.py

@@ -228,6 +228,79 @@ async def process_job(job_id: str, db: Database, registry: IndicatorRegistry) ->
             spatial_json_path = os.path.join(results_dir, f"{result.indicator_id}_spatial.json")
             _save_spatial_json(spatial, result.status.value, spatial_json_path)
 
+        # Generate hotspot maps for indicators with z-score data
+        from app.outputs.maps import render_hotspot_map
+        indicator_hotspot_paths = {}
+        for result in job.results:
+            indicator_obj = registry.get(result.indicator_id)
+            zscore_raster = getattr(indicator_obj, '_zscore_raster', None)
+            hotspot_mask = getattr(indicator_obj, '_hotspot_mask', None)
+            true_color_path_ind = getattr(indicator_obj, '_true_color_path', None)
+
+            if zscore_raster is not None and hotspot_mask is not None:
+                hotspot_path = os.path.join(results_dir, f"{result.indicator_id}_hotspot.png")
+
+                raster_path = getattr(indicator_obj, '_indicator_raster_path', None)
+                if raster_path:
+                    import rasterio
+                    with rasterio.open(raster_path) as src:
+                        extent = [src.bounds.left, src.bounds.right, src.bounds.bottom, src.bounds.top]
+                else:
+                    b = job.request.aoi.bbox
+                    extent = [b[0], b[2], b[1], b[3]]
+
+                render_hotspot_map(
+                    true_color_path=true_color_path_ind,
+                    zscore_raster=zscore_raster,
+                    hotspot_mask=hotspot_mask,
+                    extent=extent,
+                    aoi=job.request.aoi,
+                    status=result.status,
+                    output_path=hotspot_path,
+                    label=result.indicator_id.upper(),
+                )
+                indicator_hotspot_paths[result.indicator_id] = hotspot_path
+                output_files.append(hotspot_path)
+
+        # Cross-indicator compound signal detection
+        from app.analysis.compound import detect_compound_signals
+        import numpy as np
+
+        zscore_rasters = {}
+        for result in job.results:
+            indicator_obj = registry.get(result.indicator_id)
+            z = getattr(indicator_obj, '_zscore_raster', None)
+            if z is not None:
+                zscore_rasters[result.indicator_id] = z
+
+        compound_signals = []
+        if len(zscore_rasters) >= 2:
+            shapes = [z.shape for z in zscore_rasters.values()]
+            target_shape = min(shapes, key=lambda s: s[0] * s[1])
+
+            resampled = {}
+            for ind_id, z in zscore_rasters.items():
+                if z.shape != target_shape:
+                    from scipy.ndimage import zoom
+                    factors = (target_shape[0] / z.shape[0], target_shape[1] / z.shape[1])
+                    resampled[ind_id] = zoom(z, factors, order=0)
+                else:
+                    resampled[ind_id] = z
+
+            pixel_area_ha = (job.request.aoi.area_km2 * 100) / (target_shape[0] * target_shape[1])
+
+            compound_signals = detect_compound_signals(
+                zscore_rasters=resampled,
+                pixel_area_ha=pixel_area_ha,
+                threshold=2.0,
+            )
+
+        if compound_signals:
+            signals_path = os.path.join(results_dir, "compound_signals.json")
+            with open(signals_path, "w") as f:
+                json.dump([s.model_dump() for s in compound_signals], f, indent=2)
+            output_files.append(signals_path)
+
         # Build map paths dict for PDF
         indicator_map_paths = {}
         for result in job.results:
@@ -281,8 +354,10 @@ async def process_job(job_id: str, db: Database, registry: IndicatorRegistry) ->
             output_path=report_path,
             summary_map_path=summary_map_path,
             indicator_map_paths=indicator_map_paths,
+            indicator_hotspot_paths=indicator_hotspot_paths,
             overview_score=overview_score,
             overview_map_path=overview_map_path if true_color_path else "",
+            compound_signals=compound_signals,
         )
         output_files.append(report_path)
 

tests/test_narrative.py

@@ -0,0 +1,55 @@
+"""Tests for updated narrative generation."""
+import pytest
+
+
+def test_generate_narrative_includes_zscore_context(mock_indicator_result):
+    """Narrative references z-score context when anomaly data is present."""
+    from app.outputs.narrative import generate_narrative
+
+    results = [
+        mock_indicator_result(
+            indicator_id="ndvi",
+            status="amber",
+            headline="Vegetation decline (z=-1.8)",
+            z_score_current=-1.8,
+            anomaly_months=3,
+        ),
+    ]
+    text = generate_narrative(results)
+    assert "concern" in text.lower() or "monitoring" in text.lower()
+
+
+def test_generate_compound_signals_text():
+    """Compound signal text generated from CompoundSignal objects."""
+    from app.outputs.narrative import generate_compound_signals_text
+    from app.models import CompoundSignal
+
+    signals = [
+        CompoundSignal(
+            name="land_conversion",
+            triggered=True,
+            confidence="strong",
+            description="NDVI decline overlaps with settlement growth (45% overlap, 120 ha).",
+            indicators=["ndvi", "buildup"],
+            overlap_pct=45.0,
+            affected_ha=120.0,
+        ),
+        CompoundSignal(
+            name="flood_event",
+            triggered=False,
+            confidence="weak",
+            description="No flood signal detected.",
+            indicators=["sar", "water"],
+        ),
+    ]
+    text = generate_compound_signals_text(signals)
+    assert "Land Conversion" in text
+    assert "NDVI decline" in text
+
+
+def test_no_compound_signals_text():
+    """When no signals triggered, text says so explicitly."""
+    from app.outputs.narrative import generate_compound_signals_text
+
+    text = generate_compound_signals_text([])
+    assert "no compound" in text.lower()