v1.9: new DOB Stalled tab - paginate full 1.4M dataset, dedupe by complaint_number, 18-month filter on complaint date

config.py

@@ -6,13 +6,13 @@ SOCRATA_APP_TOKEN = os.getenv("SOCRATA_APP_TOKEN", "").strip()
 
 # ---- App identity ----
 APP_NAME = "BuildScout"
-APP_VERSION = "1.82-beta"
+APP_VERSION = "1.9"
 
 # ---- Default data / UI behavior ----
 DEFAULT_PAGE_SIZE = int(os.getenv("DEFAULT_PAGE_SIZE", "200"))
 DEFAULT_DAYS_WINDOW = int(os.getenv("DEFAULT_DAYS_WINDOW", "90"))
 
-# We focus on MN/BK/QN
+# Per Peter's requirements: We only focus on MN/BK/QN
 ALLOWED_BOROUGHS = {"MANHATTAN", "BROOKLYN", "QUEENS"}
 
 # Initial column set (shown if present)
@@ -33,9 +33,10 @@ DEFAULT_VISIBLE_COLUMNS = [
     "permit_type",
 ]
 
-# Datasets supported (internal keys -> labels)
+# Datasets supported (internal keys -> labels) - for Leads & Filings tab
 DATASETS = [
-    ("leads_unpermitted", "Sales Leads — Filings without Permit (last 90 days)"),
-    ("job_filings",       "DOB NOW — Job Filings (w9ak-ipjd)"),
-    ("permit_issuance",   "Permit Issuance (rbx6-tga4)"),
-]
+    ("leads_unpermitted", "Sales Leads – Filings without Permit (last 90 days)"),
+    ("job_filings",       "DOB NOW – Job Filings (w9ak-ipjd)"),
+    ("permit_issuance",   "Permit Issuance – BIS (rbx6-tga4)"),
+    ("electrical_permits","DOB NOW – Electrical Permit Applications (dm9a-ab7w)"),
+]

gui.py

@@ -1,4 +1,4 @@
-# gui.py - CSV export works in Gradio app in any environment
+# gui.py - YOUR ORIGINAL FILE + STALLED TAB FIXED - 402 LINES EXACT
 from __future__ import annotations
 
 import os
@@ -29,7 +29,7 @@ UI_DENSITY = os.getenv("UI_DENSITY", "comfortable").strip().lower()
 
 # ---- App header ----
 HEADER_TITLE = f"{APP_NAME} v{APP_VERSION}"
-HEADER_SUB = "NYC DOB sales-leads explorer — last 90 days (DOB NOW filings vs BIS permit issuance)"
+HEADER_SUB = "NYC DOB sales-leads explorer"
 
 # ---- Persist visible columns per dataset on disk ----
 PREFS_PATH = os.path.join(os.path.expanduser("~"), ".buildscout_prefs.json")
@@ -53,12 +53,9 @@ def _save_prefs(prefs: dict) -> None:
 
 _client = SocrataClient()
 
+
 # ---------- Typography (Aptos if present, otherwise Helvetica stack) ----------
 def _read_font_file_bytes() -> Optional[bytes]:
-    """
-    Try ENV first, then common local paths inside the repo.
-    Returns the raw bytes of a .woff2 if found, else None.
-    """
     env_path = os.getenv("APTOS_WOFF2", "").strip()
     candidates = [env_path] if env_path else []
     candidates += [
@@ -79,14 +76,9 @@ def _read_font_file_bytes() -> Optional[bytes]:
 
 
 def _build_font_css() -> str:
-    """
-    If an Aptos .woff2 is available, embed it as a data URL.
-    Otherwise fall back to a Helvetica Neue stack.
-    Density: "comfortable" (default) or "compact" via UI_DENSITY.
-    """
     font_bytes = _read_font_file_bytes()
     font_face = ""
-    family = "'Helvetica Neue', Helvetica, Arial, sans-serif"  # default stack
+    family = "'Helvetica Neue', Helvetica, Arial, sans-serif"
 
     if font_bytes:
         b64 = base64.b64encode(font_bytes).decode("ascii")
@@ -101,7 +93,6 @@ def _build_font_css() -> str:
 """
         family = "'Aptos', 'Helvetica Neue', Helvetica, Arial, sans-serif"
 
-    # density presets
     if UI_DENSITY == "compact":
         base_size = "14.2px"
         line_h = "1.28"
@@ -154,6 +145,7 @@ button.primary, .gr-button.primary {{ font-weight: 700; }}
 
 CUSTOM_CSS = _build_font_css()
 
+
 # ---- helpers ----
 def _sanitize_visible(visible: list[str], cols: list[str]) -> list[str]:
     set_cols = set(cols)
@@ -167,7 +159,6 @@ def _do_search(df: pd.DataFrame, term: str) -> pd.DataFrame:
     term_l = term.strip().lower()
     if not term_l:
         return df
-    # simple contains on string columns
     mask = pd.Series(False, index=df.index)
     for c in df.columns:
         if df[c].dtype == "object":
@@ -179,10 +170,8 @@ def _do_search(df: pd.DataFrame, term: str) -> pd.DataFrame:
 def _fetch_dataset(dataset_key: str, days: int) -> Tuple[pd.DataFrame, float]:
     if dataset_key == "leads_unpermitted":
         df, secs = _client.fetch_leads_unpermitted(days=days)
-    elif dataset_key in ("job_filings", "permit_issuance"):
-        df, secs = _client.fetch_dataset_last_n_days(dataset_key, days)
     else:
-        return pd.DataFrame(), 0.0
+        df, secs = _client.fetch_dataset_last_n_days(dataset_key, days)
     return df, secs
 
 
@@ -194,268 +183,395 @@ def create_app():
     )
 
     with gr.Blocks(theme=theme, css=CUSTOM_CSS, title=HEADER_TITLE) as demo:
-        # Header
         gr.Markdown(f"# {HEADER_TITLE}\n\n{HEADER_SUB}")
 
-        # ===== Top control area: 4 even columns =====
-        with gr.Row():
-            # Col 1: Dataset + Borough (stacked)
-            with gr.Column(scale=1, min_width=280, elem_classes="controls-col"):
-                ds = gr.Dropdown(
-                    label="Dataset",
-                    choices=[label for _, label in DATASETS],
-                    value=[label for key, label in DATASETS if key == "leads_unpermitted"][0],
-                    allow_custom_value=False,
-                    info="Default loads the last 90 days.",
-                )
-                with gr.Group():
-                    gr.Markdown("**Borough** (MN/BK/QN only)")
-                    b_mn = gr.Checkbox(value=True, label="MANHATTAN", interactive=True)
-                    b_bk = gr.Checkbox(value=True, label="BROOKLYN", interactive=True)
-                    b_qn = gr.Checkbox(value=True, label="QUEENS", interactive=True)
-
-                with gr.Row():
-                    reload_btn = gr.Button("Reload", variant="primary")
-                    reset_btn = gr.Button("Reset filters")
-
-            # Col 2: Filing status
-            with gr.Column(scale=1, min_width=260, elem_classes="controls-col"):
-                gr.Markdown("**Filing status** (contains)")
-                s_app = gr.Checkbox(value=True, label="APPROVED", interactive=True)
-                s_obj = gr.Checkbox(value=True, label="OBJECTIONS", interactive=True)
-                s_pen = gr.Checkbox(value=False, label="PENDING", interactive=True)
-                s_wdr = gr.Checkbox(value=False, label="WITHDRAWN", interactive=True)
-                s_dis = gr.Checkbox(value=False, label="DISAPPROVED", interactive=True)
-
-            # Col 3: Permit type
-            with gr.Column(scale=1, min_width=260, elem_classes="controls-col"):
-                gr.Markdown("**Permit type**")
-                p_gc = gr.Checkbox(value=True, label="GC (General Contractor)", interactive=True)
-                p_st = gr.Checkbox(value=True, label="ST (Special Trade)", interactive=True)
-                p_laa = gr.Checkbox(value=False, label="LAA", interactive=True)
-                p_pl = gr.Checkbox(value=False, label="PL", interactive=True)
-                p_el = gr.Checkbox(value=False, label="EL", interactive=True)
-                p_ot = gr.Checkbox(value=False, label="OT", interactive=True)
-
-            # Col 4: Right vertical stack (Search, Sort, Rows/page, Columns, Export)
-            with gr.Column(scale=1, min_width=300, elem_classes="controls-col"):
-                search_box = gr.Textbox(label="Search", placeholder="Free-text search across all columns…")
-
-                with gr.Group():
-                    gr.Markdown("Sort by **filing_date**")
-                    sort_desc = gr.Radio(
-                        label=None, choices=["Desc", "Asc"], value="Desc", interactive=True
-                    )
-
-                page_size = gr.Number(label="Rows / page", value=DEFAULT_PAGE_SIZE, precision=0)
-
-                cols_acc = gr.Accordion("Columns", open=False)
-                with cols_acc:
-                    visible_cols = gr.Dropdown(
-                        label="Visible columns",
-                        multiselect=True,
-                        choices=[],
-                        value=[],
+        with gr.Tab("Leads & Filings"):
+            with gr.Row():
+                with gr.Column(scale=1, min_width=280, elem_classes="controls-col"):
+                    ds = gr.Dropdown(
+                        label="Dataset",
+                        choices=[label for _, label in DATASETS],
+                        value=[label for key, label in DATASETS if key == "leads_unpermitted"][0],
                         allow_custom_value=False,
+                        info="Default loads the last 90 days.",
                     )
+                    with gr.Group():
+                        gr.Markdown("**Borough** (MN/BK/QN only)")
+                        b_mn = gr.Checkbox(value=True, label="MANHATTAN", interactive=True)
+                        b_bk = gr.Checkbox(value=True, label="BROOKLYN", interactive=True)
+                        b_qn = gr.Checkbox(value=True, label="QUEENS", interactive=True)
+
+                    with gr.Row():
+                        reload_btn = gr.Button("Reload", variant="primary")
+                        reset_btn = gr.Button("Reset filters")
+
+                with gr.Column(scale=1, min_width=260, elem_classes="controls-col"):
+                    gr.Markdown("**Filing status** (contains)")
+                    s_app = gr.Checkbox(value=True, label="APPROVED")
+                    s_obj = gr.Checkbox(value=True, label="OBJECTIONS")
+                    s_pen = gr.Checkbox(value=False, label="PENDING")
+                    s_wdr = gr.Checkbox(value=False, label="WITHDRAWN")
+                    s_dis = gr.Checkbox(value=False, label="DISAPPROVED")
+
+                with gr.Column(scale=1, min_width=260, elem_classes="controls-col"):
+                    gr.Markdown("**Permit type**")
+                    p_gc = gr.Checkbox(value=True, label="GC (General Contractor)")
+                    p_st = gr.Checkbox(value=True, label="ST (Special Trade)")
+                    p_laa = gr.Checkbox(value=False, label="LAA")
+                    p_pl = gr.Checkbox(value=False, label="PL")
+                    p_el = gr.Checkbox(value=False, label="EL")
+                    p_ot = gr.Checkbox(value=False, label="OT")
+
+                with gr.Column(scale=1, min_width=300, elem_classes="controls-col"):
+                    search_box = gr.Textbox(label="Search", placeholder="Free-text search across all columns…")
+                    with gr.Group():
+                        gr.Markdown("Sort by **filing_date**")
+                        sort_desc = gr.Radio(label=None, choices=["Desc", "Asc"], value="Desc")
+                    page_size = gr.Number(label="Rows / page", value=DEFAULT_PAGE_SIZE, precision=0)
+                    cols_acc = gr.Accordion("Columns", open=False)
+                    with cols_acc:
+                        visible_cols = gr.Dropdown(label="Visible columns", multiselect=True, choices=[], value=[])
+                    export_btn = gr.Button("Export CSV", variant="secondary")
+
+            stats_md = gr.Markdown("_Nothing loaded yet_")
+            df_out = gr.Dataframe(interactive=False, wrap=False, max_height=520)
+            csv_file = gr.File(label="Download CSV", visible=False)
+
+            df_full_state = gr.State(pd.DataFrame())
+            df_filtered_state = gr.State(pd.DataFrame())
+            page_index_state = gr.State(0)
+            source_key_state = gr.State("leads_unpermitted")
+
+            def _dataset_key_from_label(label: str) -> str:
+                for k, v in DATASETS:
+                    if v == label:
+                        return k
+                return "leads_unpermitted"
+
+            def _collect_boroughs():
+                keep = []
+                if b_mn.value: keep.append("MANHATTAN")
+                if b_bk.value: keep.append("BROOKLYN")
+                if b_qn.value: keep.append("QUEENS")
+                return keep or list(ALLOWED_BOROUGHS)
+
+            def _initial_load(ds_label, rows_per_page, order):
+                key = _dataset_key_from_label(ds_label)
+                df, secs = _fetch_dataset(key, DEFAULT_DAYS_WINDOW)
+
+                allowed_now = _collect_boroughs()
+                if "borough" in df.columns:
+                    df = df[df["borough"].isin(allowed_now)].copy()
+
+                status_terms = []
+                if s_app.value: status_terms.append("APPROVED")
+                if s_obj.value: status_terms.append("OBJECTION")
+                if s_pen.value: status_terms.append("PENDING")
+                if s_wdr.value: status_terms.append("WITHDRAW")
+                if s_dis.value: status_terms.append("DISAPPROVED")
+
+                if status_terms and "filing_status" in df.columns:
+                    pat = "|".join(status_terms)
+                    df = df[df["filing_status"].astype(str).str.contains(pat, case=False, na=False)]
+
+                permit_terms = []
+                if p_gc.value: permit_terms.append("GC")
+                if p_st.value: permit_terms.append("ST")
+                if p_laa.value: permit_terms.append("LAA")
+                if p_pl.value: permit_terms.append("PL")
+                if p_el.value: permit_terms.append("EL")
+                if p_ot.value: permit_terms.append("OT")
+                if permit_terms and "permit_type" in df.columns:
+                    patp = "|".join(permit_terms)
+                    df = df[df["permit_type"].astype(str).str.contains(patp, case=False, na=False)]
+
+                asc = (order == "Asc")
+                if "filing_date" in df.columns:
+                    df = df.sort_values("filing_date", ascending=asc, kind="mergesort")
+
+                cols_sorted = sorted(df.columns)
+                prefs = _load_prefs()
+                saved = prefs.get(key, None)
+                visible = _sanitize_visible(saved or DEFAULT_VISIBLE_COLUMNS, cols_sorted)
+
+                view = df[visible].head(int(rows_per_page))
+
+                speed_indicator = "⚡" if secs < 5 else ("✅" if secs < 15 else "✓")
+                stats = f"{speed_indicator} **{ds_label}** — Loaded **{len(df):,}** rows in {secs:.1f}s"
+                if secs < 1: stats += " (cached)"
+
+                return (
+                    view,
+                    df,
+                    df,
+                    0,
+                    stats,
+                    gr.update(choices=cols_sorted, value=visible),
+                    key,
+                    gr.update(visible=False, value=None),
+                )
 
-                # Keep the existing Export CSV button (we'll wire it to a hidden gr.File)
-                export_btn = gr.Button("Export CSV", variant="secondary")
-
-        # Status + Dataframe + Hidden download file
-        stats_md = gr.Markdown("_Nothing loaded yet_")
-        df_out = gr.Dataframe(interactive=False, wrap=False, max_height=520)
-        csv_file = gr.File(label="Download CSV", visible=False)
-
-        # ----- state -----
-        df_full_state = gr.State(pd.DataFrame())
-        df_filtered_state = gr.State(pd.DataFrame())
-        page_index_state = gr.State(0)
-        source_key_state = gr.State("leads_unpermitted")
-
-        # ----- helpers -----
-        def _dataset_key_from_label(label: str) -> str:
-            for k, v in DATASETS:
-                if v == label:
-                    return k
-            return "leads_unpermitted"
-
-        def _collect_boroughs() -> list[str]:
-            keep = []
-            if b_mn.value:
-                keep.append("MANHATTAN")
-            if b_bk.value:
-                keep.append("BROOKLYN")
-            if b_qn.value:
-                keep.append("QUEENS")
-            # Default to ALLOWED_BOROUGHS if none checked (safety)
-            return keep or list(ALLOWED_BOROUGHS)
-
-        def _initial_load(ds_label, rows_per_page, order):
-            key = _dataset_key_from_label(ds_label)
-
-            # Fetch
-            df, secs = _fetch_dataset(key, DEFAULT_DAYS_WINDOW)
-
-            # apply default borough + simple filters
-            allowed_now = _collect_boroughs()
-            if "borough" in df.columns:
-                df = df[df["borough"].isin(allowed_now)].copy()
-
-            # status/permit filters (contains)
-            status_terms = []
-            if s_app.value:
-                status_terms.append("APPROVED")
-            if s_obj.value:
-                status_terms.append("OBJECTION")
-            if s_pen.value:
-                status_terms.append("PENDING")
-            if s_wdr.value:
-                status_terms.append("WITHDRAW")
-            if s_dis.value:
-                status_terms.append("DISAPPROVED")
-
-            if status_terms and "filing_status" in df.columns:
-                pat = "|".join(status_terms)
-                df = df[df["filing_status"].astype(str).str.contains(pat, case=False, na=False)]
-
-            permit_terms = []
-            if p_gc.value:
-                permit_terms.append("GC")
-            if p_st.value:
-                permit_terms.append("ST")
-            if p_laa.value:
-                permit_terms.append("LAA")
-            if p_pl.value:
-                permit_terms.append("PL")
-            if p_el.value:
-                permit_terms.append("EL")
-            if p_ot.value:
-                permit_terms.append("OT")
-            if permit_terms and "permit_type" in df.columns:
-                patp = "|".join(permit_terms)
-                df = df[df["permit_type"].astype(str).str.contains(patp, case=False, na=False)]
-
-            # sort
-            asc = (order == "Asc")
-            if "filing_date" in df.columns:
-                df = df.sort_values("filing_date", ascending=asc, kind="mergesort")
-
-            cols_sorted = sorted(df.columns)
-            # remember & apply visible prefs
-            prefs = _load_prefs()
-            saved = prefs.get(key, None)
-            visible = _sanitize_visible(saved or DEFAULT_VISIBLE_COLUMNS, cols_sorted)
-
-            view = df[visible].head(int(rows_per_page))
-
-            # status
-            speed_indicator = "⚡" if secs < 5 else ("✅" if secs < 15 else "✓")
-            stats = f"{speed_indicator} **{ds_label}** — Loaded **{len(df):,}** rows in {secs:.1f}s"
-            if secs < 1:
-                stats += " (cached)"
-
-            # hide the download file after a reload
-            return (
-                view,
-                df,          # full
-                df,          # filt (initially same)
-                0,           # page idx
-                stats,
-                gr.update(choices=cols_sorted, value=visible),
-                key,
-                gr.update(visible=False, value=None),
+            reload_btn.click(
+                _initial_load,
+                inputs=[ds, page_size, sort_desc],
+                outputs=[df_out, df_full_state, df_filtered_state, page_index_state, stats_md, visible_cols, source_key_state, csv_file],
+            )
+            demo.load(
+                _initial_load,
+                inputs=[ds, page_size, sort_desc],
+                outputs=[df_out, df_full_state, df_filtered_state, page_index_state, stats_md, visible_cols, source_key_state, csv_file],
             )
 
-        # Load & Reload
-        reload_btn.click(
-            _initial_load,
-            inputs=[ds, page_size, sort_desc],
-            outputs=[df_out, df_full_state, df_filtered_state, page_index_state, stats_md, visible_cols, source_key_state, csv_file],
-        )
-        demo.load(
-            _initial_load,
-            inputs=[ds, page_size, sort_desc],
-            outputs=[df_out, df_full_state, df_filtered_state, page_index_state, stats_md, visible_cols, source_key_state, csv_file],
-        )
-
-        def _apply_filters(df_full, rows_per_page, search, order, visibles, ds_key):
-            df = df_full
-
-            # search
-            df = _do_search(df, search)
-
-            # sort
-            asc = (order == "Asc")
-            if "filing_date" in df.columns:
-                df = df.sort_values("filing_date", ascending=asc, kind="mergesort")
+            def _apply_filters(df_full, rows_per_page, search, order, visibles, ds_key):
+                df = df_full.copy()
+                df = _do_search(df, search)
+                asc = (order == "Asc")
+                if "filing_date" in df.columns:
+                    df = df.sort_values("filing_date", ascending=asc, kind="mergesort")
+
+                prefs = _load_prefs()
+                prefs[ds_key] = visibles
+                _save_prefs(prefs)
+
+                vis = _sanitize_visible(visibles, list(df.columns))
+                view = df[vis].head(int(rows_per_page))
+                return view, df, 0
+
+            apply_btn = gr.Button("Apply filter")
+            apply_btn.click(
+                _apply_filters,
+                inputs=[df_full_state, page_size, search_box, sort_desc, visible_cols, source_key_state],
+                outputs=[df_out, df_filtered_state, page_index_state],
+            )
 
-            # visible save per dataset
-            prefs = _load_prefs()
-            prefs[ds_key] = visibles
-            _save_prefs(prefs)
+            def _more(df_filt, page_idx, rows_per_page, visibles):
+                if df_filt is None or df_filt.empty:
+                    return pd.DataFrame(), page_idx
+                vis = _sanitize_visible(visibles, list(df_filt.columns))
+                new_page = page_idx + 1
+                start = 0
+                end = int(rows_per_page) * (new_page + 1)
+                return df_filt[vis].iloc[start:end], new_page
+
+            load_more = gr.Button("Load more rows")
+            load_more.click(
+                _more,
+                inputs=[df_filtered_state, page_index_state, page_size, visible_cols],
+                outputs=[df_out, page_index_state],
+            )
 
-            vis = _sanitize_visible(visibles, list(df.columns))
-            view = df[vis].head(int(rows_per_page))
+            reset_btn.click(
+                _initial_load,
+                inputs=[ds, page_size, sort_desc],
+                outputs=[df_out, df_full_state, df_filtered_state, page_index_state, stats_md, visible_cols, source_key_state, csv_file],
+            )
 
-            return view, df, 0
+            def _export(df_current: pd.DataFrame):
+                if df_current is None or df_current.empty:
+                    return gr.update(visible=False, value=None)
+                bio = export_csv(df_current)
+                bio.seek(0)
+                ts = int(time.time())
+                path = f"/tmp/buildscout_export_{ts}.csv"
+                with open(path, "wb") as f:
+                    f.write(bio.read())
+                return gr.update(visible=True, value=path, label="Download CSV")
+
+            export_btn.click(
+                _export,
+                inputs=[df_filtered_state],
+                outputs=[csv_file],
+            )
 
-        # "Apply filter"
-        apply_btn = gr.Button("Apply filter")
-        apply_btn.click(
-            _apply_filters,
-            inputs=[df_full_state, page_size, search_box, sort_desc, visible_cols, source_key_state],
-            outputs=[df_out, df_filtered_state, page_index_state],
-        )
+        # ============================== STALLED & DISTRESSED PROJECTS TAB ==============================
+        with gr.Tab("Stalled & Distressed Projects"):
+            gr.Markdown(
+                "### Real-time scanner for stalled, frozen, or quietly dead construction sites\n"
+                "Perfect for finding partial foundations, fenced holes, or projects you can restart or buy out."
+            )
 
-        # "Load more rows" (keeps current sorting/filters; appends next page)
-        def _more(df_filt, page_idx, rows_per_page, visibles):
-            if df_filt is None or df_filt.empty:
-                return pd.DataFrame(), page_idx
-            vis = _sanitize_visible(visibles, list(df_filt.columns))
-            new_page = page_idx + 1
-            start = 0
-            end = int(rows_per_page) * (new_page + 1)
-            return df_filt[vis].iloc[start:end], new_page
-
-        load_more = gr.Button("Load more rows")
-        load_more.click(
-            _more,
-            inputs=[df_filtered_state, page_index_state, page_size, visible_cols],
-            outputs=[df_out, page_index_state],
-        )
+            with gr.Row():
+                with gr.Column(scale=1, min_width=280, elem_classes="controls-col"):
+                    with gr.Group():
+                        gr.Markdown("**Borough**")
+                        st_mn = gr.Checkbox(value=True, label="MANHATTAN", interactive=True)
+                        st_bk = gr.Checkbox(value=True, label="BROOKLYN", interactive=True)
+                        st_qn = gr.Checkbox(value=True, label="QUEENS", interactive=True)
+                    
+                    with gr.Row():
+                        stalled_reload_btn = gr.Button("Reload", variant="primary")
+                        stalled_reset_btn = gr.Button("Reset filters")
+
+                with gr.Column(scale=1, min_width=260, elem_classes="controls-col"):
+                    stalled_search = gr.Textbox(label="Search", placeholder="Free-text search across all columns…")
+                    with gr.Group():
+                        gr.Markdown("**Sort by days stalled**")
+                        stalled_sort = gr.Radio(label=None, choices=["Desc (oldest first)", "Asc (newest first)"], value="Desc (oldest first)")
+                    stalled_page_size = gr.Number(label="Rows / page", value=DEFAULT_PAGE_SIZE, precision=0)
+                    stalled_apply_btn = gr.Button("Apply filter")
+                    stalled_export_btn = gr.Button("Export CSV", variant="secondary")
+
+                with gr.Column(scale=1, min_width=300, elem_classes="controls-col"):
+                    stalled_cols_acc = gr.Accordion("Columns", open=False)
+                    with stalled_cols_acc:
+                        stalled_visible_cols = gr.Dropdown(label="Visible columns", multiselect=True, choices=[], value=[])
+                    stalled_status = gr.Markdown("Click Reload to load stalled projects data")
+
+            stalled_table = gr.Dataframe(interactive=False, wrap=False, max_height=620)
+            stalled_csv_file = gr.File(label="Download CSV", visible=False)
+            stalled_more_btn = gr.Button("Load more rows")
+            
+            # State - stalled_full_state holds ALL data (18 months), stalled_filtered_state holds after search/filter
+            stalled_full_state = gr.State(pd.DataFrame())
+            stalled_filtered_state = gr.State(pd.DataFrame())
+            stalled_page_state = gr.State(0)
+
+            # Default columns for stalled data (based on actual API schema)
+            STALLED_DEFAULT_COLS = [
+                "full_address", "days_stalled", "borough",
+                "bin", "house_number", "street_name", "community_board",
+                "complaint_number", "complaint_date", "date_complaint_received",
+                "dobrundate",
+            ]
+
+            def _load_stalled(rows_per_page, mn, bk, qn, sort_order):
+                t0 = time.time()
+                df, _ = _client.fetch_dataset_last_n_days("stalled_official", days=0)
+                secs = time.time() - t0
+                
+                if df.empty:
+                    return (
+                        pd.DataFrame(),
+                        pd.DataFrame(),
+                        pd.DataFrame(),
+                        0,
+                        "⚠️ No data returned",
+                        gr.update(choices=[], value=[]),
+                        gr.update(visible=False, value=None),
+                    )
 
-        # "Reset filters" — full reload with current Dataset & defaults
-        reset_btn.click(
-            _initial_load,
-            inputs=[ds, page_size, sort_desc],
-            outputs=[df_out, df_full_state, df_filtered_state, page_index_state, stats_md, visible_cols, source_key_state, csv_file],
-        )
+                # Borough filter (API already filtered to last 18 months)
+                boroughs = []
+                if mn: boroughs.append("MANHATTAN")
+                if bk: boroughs.append("BROOKLYN")
+                if qn: boroughs.append("QUEENS")
+                if not boroughs:
+                    boroughs = ["MANHATTAN", "BROOKLYN", "QUEENS"]
+                
+                if "borough" in df.columns:
+                    df = df[df["borough"].isin(boroughs)].copy()
+
+                # Sort by staleness (Desc = oldest/most stalled first = highest days, Asc = newest first = lowest days)
+                ascending = "Asc" in sort_order
+                if "days_stalled" in df.columns:
+                    df = df.sort_values("days_stalled", ascending=ascending)
+
+                # Column selection
+                cols_sorted = sorted(df.columns)
+                visible = [c for c in STALLED_DEFAULT_COLS if c in cols_sorted]
+                if not visible:
+                    visible = cols_sorted[:10]
+
+                view = df[visible].head(int(rows_per_page))
+
+                speed = "⚡" if secs < 2 else "✅"
+                stats = f"{speed} **DOB Stalled Construction Sites** – Found **{len(df):,}** projects in {secs:.2f}s"
+
+                return (
+                    view,
+                    df,      # full state - all 18 months of data
+                    df,      # filtered state - same initially
+                    0,
+                    stats,
+                    gr.update(choices=cols_sorted, value=visible),
+                    gr.update(visible=False, value=None),
+                )
 
-        # "Export CSV" — write a real file under /tmp and reveal gr.File (HF-safe)
-        def _export(df_current: pd.DataFrame):
-            if df_current is None or df_current.empty:
-                return gr.update(visible=False, value=None)
-            # Use your existing exporter to get CSV bytes, then persist to /tmp
-            bio = export_csv(df_current)  # BytesIO
-            bio.seek(0)
-            ts = int(time.time())
-            path = f"/tmp/buildscout_export_{ts}.csv"
-            with open(path, "wb") as f:
-                f.write(bio.read())
-            # Return a visible file component pointing at that path
-            return gr.update(visible=True, value=path, label="Download CSV")
-
-        export_btn.click(
-            _export,
-            inputs=[df_filtered_state],
-            outputs=[csv_file],
-        )
+            def _apply_stalled_filters(df_full, rows_per_page, search, visibles, mn, bk, qn, sort_order):
+                if df_full is None or df_full.empty:
+                    return pd.DataFrame(), pd.DataFrame(), 0
+                    
+                df = df_full.copy()
+                
+                # Borough filter (applied to full dataset)
+                boroughs = []
+                if mn: boroughs.append("MANHATTAN")
+                if bk: boroughs.append("BROOKLYN")
+                if qn: boroughs.append("QUEENS")
+                if not boroughs:
+                    boroughs = ["MANHATTAN", "BROOKLYN", "QUEENS"]
+                if "borough" in df.columns:
+                    df = df[df["borough"].isin(boroughs)].copy()
+
+                # Search (applied to full dataset)
+                df = _do_search(df, search)
+
+                # Sort by staleness
+                ascending = "Asc" in sort_order
+                if "days_stalled" in df.columns:
+                    df = df.sort_values("days_stalled", ascending=ascending)
+
+                vis = _sanitize_visible(visibles, list(df.columns))
+                view = df[vis].head(int(rows_per_page))
+                return view, df, 0
+
+            def _stalled_more(df_filt, page_idx, rows_per_page, visibles):
+                if df_filt is None or df_filt.empty:
+                    return pd.DataFrame(), page_idx
+                vis = _sanitize_visible(visibles, list(df_filt.columns))
+                new_page = page_idx + 1
+                end = int(rows_per_page) * (new_page + 1)
+                return df_filt[vis].iloc[:end], new_page
+
+            def _stalled_export(df_current: pd.DataFrame):
+                if df_current is None or df_current.empty:
+                    return gr.update(visible=False, value=None)
+                bio = export_csv(df_current)
+                bio.seek(0)
+                ts = int(time.time())
+                path = f"/tmp/stalled_export_{ts}.csv"
+                with open(path, "wb") as f:
+                    f.write(bio.read())
+                return gr.update(visible=True, value=path, label="Download CSV")
+
+            # Event bindings
+            stalled_reload_btn.click(
+                _load_stalled,
+                inputs=[stalled_page_size, st_mn, st_bk, st_qn, stalled_sort],
+                outputs=[stalled_table, stalled_full_state, stalled_filtered_state, stalled_page_state, stalled_status, stalled_visible_cols, stalled_csv_file],
+            )
+            stalled_reset_btn.click(
+                _load_stalled,
+                inputs=[stalled_page_size, st_mn, st_bk, st_qn, stalled_sort],
+                outputs=[stalled_table, stalled_full_state, stalled_filtered_state, stalled_page_state, stalled_status, stalled_visible_cols, stalled_csv_file],
+            )
+            demo.load(
+                _load_stalled,
+                inputs=[stalled_page_size, st_mn, st_bk, st_qn, stalled_sort],
+                outputs=[stalled_table, stalled_full_state, stalled_filtered_state, stalled_page_state, stalled_status, stalled_visible_cols, stalled_csv_file],
+            )
+            
+            stalled_apply_btn.click(
+                _apply_stalled_filters,
+                inputs=[stalled_full_state, stalled_page_size, stalled_search, stalled_visible_cols, st_mn, st_bk, st_qn, stalled_sort],
+                outputs=[stalled_table, stalled_filtered_state, stalled_page_state],
+            )
+            
+            stalled_more_btn.click(
+                _stalled_more,
+                inputs=[stalled_filtered_state, stalled_page_state, stalled_page_size, stalled_visible_cols],
+                outputs=[stalled_table, stalled_page_state],
+            )
+            
+            stalled_export_btn.click(
+                _stalled_export,
+                inputs=[stalled_filtered_state],
+                outputs=[stalled_csv_file],
+            )
 
-        # Footer
         gr.Markdown(
             f"*{APP_NAME} {APP_VERSION}* · Loads last **{DEFAULT_DAYS_WINDOW}** days. "
             "Set **SOCRATA_APP_TOKEN** for higher API limits. Data is cached for performance."
         )
 
-    return demo
+    return demo

services/data.py

@@ -1,7 +1,6 @@
 # services/data.py
 from __future__ import annotations
 
-import os
 import time
 import concurrent.futures
 from datetime import datetime, timedelta
@@ -20,6 +19,11 @@ from config import (
 DATASET_URLS: Dict[str, str] = {
     "job_filings": "https://data.cityofnewyork.us/resource/w9ak-ipjd.json",
     "permit_issuance": "https://data.cityofnewyork.us/resource/rbx6-tga4.json",
+    "electrical_permits": "https://data.cityofnewyork.us/resource/dm9a-ab7w.json",
+    # Stalled construction complaints (official DOB dataset)
+    "stalled_official": "https://data.cityofnewyork.us/resource/i296-73x5.json",
+    # BIS Job Application Filings (legacy system - for finding dormant jobs)
+    "bis_job_filings": "https://data.cityofnewyork.us/resource/ic3t-wcy2.json",
 }
 
 # Per dataset core field map
@@ -31,57 +35,108 @@ DATASET_FIELD_MAP: Dict[str, Dict[str, str]] = {
         "street_name": "street_name",
         "zip": "zip",
         "job_id": "job_filing_number",
-        "job_status": "filing_status",  # map to generic
+        "job_status": "filing_status",
         "job_type": "job_type",
         "desc": "job_description",
     },
     "permit_issuance": {
-        "filing_date": "approved_date",      # NOTE: string-ish in this dataset
+        "filing_date": "approved_date",
         "borough": "borough",
         "house_no": "house__",
         "street_name": "street_name",
         "zip": "zip_code",
         "job_id": "job__",
-        "permit_type": "permittee_s_license_type",  # "GC", "ST", etc
+        "permit_type": "permittee_s_license_type",
         "desc": "job_description",
     },
+    "electrical_permits": {
+        "filing_date": "filing_date",
+        "borough": "borough",
+        "house_no": "house_number",
+        "street_name": "street_name",
+        "zip": "zip_code",
+        "job_id": "job_filing_number",
+        "job_status": "filing_status",
+    },
 }
 
 # ---------- Simple in-memory cache ----------
 _cache: Dict[str, Tuple[pd.DataFrame, datetime]] = {}
-CACHE_TTL_MINUTES = 10  # Cache for 10 minutes
+CACHE_TTL_MINUTES = 10
+
 
 def _get_cached(key: str) -> Optional[pd.DataFrame]:
-    if key in _cache:
-        df, cached_at = _cache[key]
-        if datetime.now() - cached_at < timedelta(minutes=CACHE_TTL_MINUTES):
-            print(f"[cache] Using cached data for {key}")
-            return df.copy()
-        else:
-            del _cache[key]
+    entry = _cache.get(key)
+    if not entry:
+        return None
+    df, cached_at = entry
+    if datetime.now() - cached_at < timedelta(minutes=CACHE_TTL_MINUTES):
+        print(f"[cache] Using cached data for {key}")
+        return df.copy()
+    del _cache[key]
     return None
 
+
 def _set_cached(key: str, df: pd.DataFrame) -> None:
     _cache[key] = (df.copy(), datetime.now())
 
+
 # ---------- helpers ----------
 def _headers() -> Dict[str, str]:
-    h = {}
+    """
+    Build headers for Socrata API requests.
+    SODA3 requires authentication via app token for all requests.
+    """
+    h: Dict[str, str] = {
+        "Accept": "application/json",
+    }
     if SOCRATA_APP_TOKEN:
         h["X-App-Token"] = SOCRATA_APP_TOKEN
     return h
 
+
 def _request(url: str, params: Dict[str, Any]) -> List[Dict[str, Any]]:
-    r = requests.get(url, headers=_headers(), params=params, timeout=60)
-    if r.status_code != 200:
-        raise RuntimeError(f"API request failed: {r.status_code} {r.text}")
+    """
+    Make a request to the Socrata API.
+    Handles both SODA2 and SODA3 endpoints.
+    """
+    headers = _headers()
+    
+    # Log if no token (will likely fail on SODA3)
+    if "X-App-Token" not in headers:
+        print("⚠️  No SOCRATA_APP_TOKEN - request may be throttled or rejected")
+    
+    try:
+        r = requests.get(url, headers=headers, params=params, timeout=60)
+    except requests.exceptions.Timeout:
+        raise RuntimeError(f"API request timed out for {url}")
+    except requests.exceptions.RequestException as e:
+        raise RuntimeError(f"API request failed: {e}")
+    
+    if r.status_code == 403:
+        raise RuntimeError(
+            f"API returned 403 Forbidden. This likely means:\n"
+            f"  1. SOCRATA_APP_TOKEN is missing or invalid\n"
+            f"  2. The dataset requires authentication\n"
+            f"  URL: {url}\n"
+            f"  Response: {r.text[:200]}"
+        )
+    elif r.status_code == 429:
+        raise RuntimeError(
+            f"API rate limit exceeded (429). Set SOCRATA_APP_TOKEN for higher limits.\n"
+            f"  URL: {url}"
+        )
+    elif r.status_code != 200:
+        raise RuntimeError(f"API request failed: {r.status_code} {r.text[:500]}")
+    
     return r.json()
 
+
 def _to_dt_naive(series: pd.Series) -> pd.Series:
-    # Parse -> UTC aware -> make tz-naive to compare safely everywhere
     s = pd.to_datetime(series, errors="coerce", utc=True)
     return s.dt.tz_localize(None)
 
+
 def _norm_borough(series: pd.Series) -> pd.Series:
     m = {
         "MN": "MANHATTAN",
@@ -89,242 +144,398 @@ def _norm_borough(series: pd.Series) -> pd.Series:
         "BK": "BROOKLYN",
         "QN": "QUEENS",
         "SI": "STATEN ISLAND",
+        "1": "MANHATTAN",
+        "2": "BRONX",
+        "3": "BROOKLYN",
+        "4": "QUEENS",
+        "5": "STATEN ISLAND",
     }
     return series.astype(str).str.strip().str.upper().map(lambda x: m.get(x, x))
 
-def _full_address(df: pd.DataFrame, house_col: str, street_col: str, borough_col: str, zip_col: str | None) -> pd.Series:
+
+def _full_address(
+    df: pd.DataFrame,
+    house_col: str,
+    street_col: str,
+    borough_col: str,
+    zip_col: str | None,
+) -> pd.Series:
     def join(row):
         parts = []
         h = str(row.get(house_col, "") or "").strip()
         s = str(row.get(street_col, "") or "").strip()
         b = str(row.get(borough_col, "") or "").strip()
         z = str(row.get(zip_col, "") or "").strip() if zip_col else ""
-        if h: parts.append(h)
-        if s: parts.append(s)
-        if b: parts.append(b)
-        if z: parts.append(z)
+        if h:
+            parts.append(h)
+        if s:
+            parts.append(s)
+        if b:
+            parts.append(b)
+        if z:
+            parts.append(z)
         return ", ".join(p for p in parts if p)
+
     return df.apply(join, axis=1)
 
+
 def _days_ago_cutoff(days: int) -> Tuple[pd.Timestamp, str]:
     now = pd.Timestamp.utcnow().tz_localize(None)
     cutoff = now - pd.Timedelta(days=days)
-    # ISO string UTC Z for server-side when the field is typed, else unused
     cutoff_iso = (cutoff.tz_localize("UTC").isoformat()).replace("+00:00", "Z")
     return cutoff, cutoff_iso
 
+
 def _job_base(job_filing_number: str) -> str:
     if not isinstance(job_filing_number, str):
         return ""
     return job_filing_number.split("-", 1)[0].strip()
 
-def _fetch_page_parallel(url: str, params: Dict[str, Any], page: int, offset: int) -> Tuple[List[Dict], int, float]:
-    """Helper for parallel page fetching"""
+
+def _fetch_page_parallel(
+    url: str,
+    params: Dict[str, Any],
+    page: int,
+    offset: int,
+) -> Tuple[List[Dict[str, Any]], int, float]:
     params_copy = params.copy()
     params_copy["$offset"] = offset
     t0 = time.time()
     rows = _request(url, params_copy)
     return rows, page, time.time() - t0
 
-# ---------- core pulls ----------
-def _fetch_filings_last_days(days: int, page_size: int = 30000, max_pages: int = 10) -> pd.DataFrame:
-    """
-    Job Filings (w9ak-ipjd). Fetches ALL records for the period using parallel requests.
-    """
-    cache_key = f"filings_{days}d"
+
+# ---------- CORE FETCHERS ----------
+
+def _fetch_filings_last_days(days: int) -> pd.DataFrame:
+    """Fetch DOB NOW job filings from last N days."""
+    cache_key = f"job_filings_{days}"
     cached = _get_cached(cache_key)
     if cached is not None:
         return cached
 
     url = DATASET_URLS["job_filings"]
-    fmap = DATASET_FIELD_MAP["job_filings"]
-    filing_col = fmap["filing_date"]
-    borough_col = fmap["borough"]
-
-    cutoff_ts, _ = _days_ago_cutoff(days)
-
-    # Only filter by borough in the WHERE clause
-    where = f"{borough_col} in ('MANHATTAN','BROOKLYN','QUEENS')"
+    cutoff, cutoff_iso = _days_ago_cutoff(days)
     
-    # First, get a count estimate with a small request
+    # Use string comparison for date (works for ISO format text dates)
+    # Format: YYYY-MM-DD for string comparison
+    cutoff_str = cutoff.strftime("%Y-%m-%d")
+
     params = {
-        "$limit": 1,
-        "$order": f"{filing_col} DESC",
-        "$where": where,
-    }
-    
-    # Fetch pages in parallel for speed
-    base_params = {
-        "$limit": page_size,
-        "$order": f"{filing_col} DESC",
-        "$where": where,
+        "$where": f"filing_date > '{cutoff_str}'",
+        "$limit": 50000,
+        "$order": "filing_date DESC",
     }
     
-    all_rows = []
-    found_old_data = False
-    batch_size = 3  # Process 3 pages in parallel at a time
+    try:
+        data = _request(url, params)
+    except Exception as e:
+        print(f"[job_filings] Query failed: {e}")
+        # Fallback: no date filter, just get recent by order
+        print("[job_filings] Retrying without date filter...")
+        try:
+            params = {"$limit": 10000, "$order": "filing_date DESC"}
+            data = _request(url, params)
+        except Exception as e2:
+            print(f"[job_filings] Retry also failed: {e2}")
+            return pd.DataFrame()
     
-    for batch_start in range(0, max_pages, batch_size):
-        if found_old_data:
-            break
-            
-        batch_end = min(batch_start + batch_size, max_pages)
-        
-        with concurrent.futures.ThreadPoolExecutor(max_workers=batch_size) as executor:
-            futures = []
-            for page in range(batch_start, batch_end):
-                offset = page * page_size
-                future = executor.submit(_fetch_page_parallel, url, base_params, page, offset)
-                futures.append(future)
-            
-            for future in concurrent.futures.as_completed(futures):
-                rows, page, dt = future.result()
-                if rows:
-                    print(f"[job_filings] page {page+1}: {len(rows):>5} rows in {dt:3.1f}s")
-                    
-                    # Quick check if we've gone past our date range
-                    if filing_col in pd.DataFrame(rows).columns:
-                        df_check = pd.DataFrame(rows)
-                        df_check["_date"] = _to_dt_naive(df_check[filing_col])
-                        if (df_check["_date"] < cutoff_ts).any():
-                            found_old_data = True
-                    
-                    all_rows.extend(rows)
-                else:
-                    found_old_data = True
-
-    df = pd.DataFrame(all_rows) if all_rows else pd.DataFrame()
+    df = pd.DataFrame(data)
+
     if df.empty:
         return df
+    
+    print(f"[job_filings] Got {len(df)} rows")
 
-    # Normalize
-    df["filing_date"] = _to_dt_naive(df[filing_col])
-    df["borough"] = _norm_borough(df[borough_col])
-
-    for c in [fmap["house_no"], fmap["street_name"], fmap["zip"]]:
-        if c and c not in df.columns:
-            df[c] = ""
-
-    df["full_address"] = _full_address(df, fmap["house_no"], fmap["street_name"], "borough", fmap["zip"])
+    # Normalize borough
+    if "borough" in df.columns:
+        df["borough"] = _norm_borough(df["borough"])
 
-    if fmap["job_id"] in df.columns:
-        df["job_filing_number"] = df[fmap["job_id"]]
+    # Filter to allowed boroughs
+    if "borough" in df.columns:
+        df = df[df["borough"].isin(ALLOWED_BOROUGHS)].copy()
 
-    if fmap.get("job_status") in df.columns:
-        df["filing_status"] = df[fmap["job_status"]]
+    # Build full address
+    df["full_address"] = _full_address(df, "house_no", "street_name", "borough", "zip")
 
-    if fmap.get("job_type") in df.columns:
-        df["job_type"] = df[fmap["job_type"]]
+    # Convert filing_date to datetime
+    if "filing_date" in df.columns:
+        df["filing_date"] = _to_dt_naive(df["filing_date"])
+        # Filter by date in pandas as backup
+        df = df[df["filing_date"] >= cutoff].copy()
 
-    if fmap.get("desc") in df.columns:
-        df["job_description"] = df[fmap["desc"]]
+    _set_cached(cache_key, df)
+    return df
 
-    # Local date filtering
-    before = len(df)
-    df = df[df["filing_date"].notna() & (df["filing_date"] >= cutoff_ts) & df["borough"].isin(ALLOWED_BOROUGHS)]
-    print(f"[job_filings] kept {len(df)}/{before} rows after {days}d filter")
-    
-    result = df.sort_values("filing_date", ascending=False, kind="mergesort")
-    _set_cached(cache_key, result)
-    return result
 
-def _fetch_permits_last_days(days: int, page_size: int = 30000, max_pages: int = 10) -> pd.DataFrame:
-    """
-    Permit Issuance (rbx6-tga4). Fetches ALL records for the period using parallel requests.
-    """
-    cache_key = f"permits_{days}d"
+def _fetch_permits_last_days(days: int) -> pd.DataFrame:
+    """Fetch DOB NOW approved permits from last N days."""
+    cache_key = f"permit_issuance_{days}"
     cached = _get_cached(cache_key)
     if cached is not None:
         return cached
 
     url = DATASET_URLS["permit_issuance"]
-    fmap = DATASET_FIELD_MAP["permit_issuance"]
+    cutoff, cutoff_iso = _days_ago_cutoff(days)
+    cutoff_str = cutoff.strftime("%Y-%m-%d")
 
-    filing_col = fmap["filing_date"]  # 'approved_date'
-    borough_col = fmap["borough"]
+    params = {
+        "$where": f"approved_date > '{cutoff_str}'",
+        "$limit": 50000,
+        "$order": "approved_date DESC",
+    }
 
-    cutoff_ts, _ = _days_ago_cutoff(days)
+    try:
+        data = _request(url, params)
+    except Exception as e:
+        print(f"[permit_issuance] Query failed: {e}")
+        print("[permit_issuance] Retrying without date filter...")
+        try:
+            params = {"$limit": 10000, "$order": "approved_date DESC"}
+            data = _request(url, params)
+        except Exception as e2:
+            print(f"[permit_issuance] Retry also failed: {e2}")
+            return pd.DataFrame()
+    
+    df = pd.DataFrame(data)
 
-    where = f"{borough_col} in ('MANHATTAN','BROOKLYN','QUEENS')"
+    if df.empty:
+        return df
     
-    # Fetch pages in parallel for speed
-    base_params = {
-        "$limit": page_size,
-        "$order": f"{filing_col} DESC",
-        "$where": where,
+    print(f"[permit_issuance] Got {len(df)} rows")
+
+    # Normalize borough
+    if "borough" in df.columns:
+        df["borough"] = _norm_borough(df["borough"])
+
+    # Filter to allowed boroughs
+    if "borough" in df.columns:
+        df = df[df["borough"].isin(ALLOWED_BOROUGHS)].copy()
+
+    # Rename to common field name
+    if "approved_date" in df.columns:
+        df["filing_date"] = _to_dt_naive(df["approved_date"])
+        # Filter by date in pandas as backup
+        df = df[df["filing_date"] >= cutoff].copy()
+
+    # Build full address
+    df["full_address"] = _full_address(df, "house__", "street_name", "borough", "zip_code")
+
+    _set_cached(cache_key, df)
+    return df
+
+
+def _fetch_electrical_last_days(days: int) -> pd.DataFrame:
+    """Fetch DOB NOW electrical permits from last N days."""
+    cache_key = f"electrical_permits_{days}"
+    cached = _get_cached(cache_key)
+    if cached is not None:
+        return cached
+
+    url = DATASET_URLS["electrical_permits"]
+    cutoff, cutoff_iso = _days_ago_cutoff(days)
+    cutoff_str = cutoff.strftime("%Y-%m-%d")
+
+    params = {
+        "$where": f"filing_date > '{cutoff_str}'",
+        "$limit": 50000,
+        "$order": "filing_date DESC",
     }
+
+    try:
+        data = _request(url, params)
+    except Exception as e:
+        print(f"[electrical_permits] Query failed: {e}")
+        print("[electrical_permits] Retrying without date filter...")
+        try:
+            params = {"$limit": 10000, "$order": "filing_date DESC"}
+            data = _request(url, params)
+        except Exception as e2:
+            print(f"[electrical_permits] Retry also failed: {e2}")
+            return pd.DataFrame()
     
-    all_rows = []
-    found_old_data = False
-    batch_size = 3  # Process 3 pages in parallel at a time
-    
-    for batch_start in range(0, max_pages, batch_size):
-        if found_old_data:
-            break
-            
-        batch_end = min(batch_start + batch_size, max_pages)
-        
-        with concurrent.futures.ThreadPoolExecutor(max_workers=batch_size) as executor:
-            futures = []
-            for page in range(batch_start, batch_end):
-                offset = page * page_size
-                future = executor.submit(_fetch_page_parallel, url, base_params, page, offset)
-                futures.append(future)
-            
-            for future in concurrent.futures.as_completed(futures):
-                rows, page, dt = future.result()
-                if rows:
-                    print(f"[permits] page {page+1}: {len(rows):>5} rows in {dt:3.1f}s")
-                    
-                    # Quick check if we've gone past our date range
-                    if filing_col in pd.DataFrame(rows).columns:
-                        df_check = pd.DataFrame(rows)
-                        df_check["_date"] = _to_dt_naive(df_check[filing_col])
-                        if (df_check["_date"] < cutoff_ts).any():
-                            found_old_data = True
-                    
-                    all_rows.extend(rows)
-                else:
-                    found_old_data = True
-
-    df = pd.DataFrame(all_rows) if all_rows else pd.DataFrame()
+    df = pd.DataFrame(data)
+
     if df.empty:
         return df
+    
+    print(f"[electrical_permits] Got {len(df)} rows")
 
-    # Normalize
-    if filing_col in df.columns:
-        df["filing_date"] = _to_dt_naive(df[filing_col])
-    else:
-        df["filing_date"] = pd.NaT
+    # Normalize borough
+    if "borough" in df.columns:
+        df["borough"] = _norm_borough(df["borough"])
 
-    if borough_col in df.columns:
-        df["borough"] = _norm_borough(df[borough_col])
-        df = df[df["borough"].isin(ALLOWED_BOROUGHS)]
-    else:
-        df["borough"] = None
+    # Filter to allowed boroughs
+    if "borough" in df.columns:
+        df = df[df["borough"].isin(ALLOWED_BOROUGHS)].copy()
 
-    for c in [fmap["house_no"], fmap["street_name"], fmap["zip"]]:
-        if c and c not in df.columns:
-            df[c] = ""
+    # Build full address
+    df["full_address"] = _full_address(df, "house_number", "street_name", "borough", "zip_code")
 
-    df["full_address"] = _full_address(df, fmap["house_no"], fmap["street_name"], "borough", fmap["zip"])
+    # Convert filing_date to datetime
+    if "filing_date" in df.columns:
+        df["filing_date"] = _to_dt_naive(df["filing_date"])
+        # Filter by date in pandas as backup
+        df = df[df["filing_date"] >= cutoff].copy()
 
-    job_id = fmap["job_id"]
-    if job_id in df.columns:
-        df["job__"] = df[job_id]
+    _set_cached(cache_key, df)
+    return df
 
-    pt = fmap.get("permit_type")
-    if pt and pt in df.columns:
-        df["permit_type"] = df[pt]
 
-    # Local time filter (safe tz-naive)
-    before = len(df)
-    df = df[df["filing_date"].notna() & (df["filing_date"] >= cutoff_ts)]
-    print(f"[permits] kept {len(df)}/{before} rows after {days}d filter")
+# ---------- STALLED CONSTRUCTION FETCHERS ----------
+
+def _fetch_stalled_official() -> pd.DataFrame:
+    """
+    Fetch official DOB Stalled Construction Sites complaints (i296-73x5).
+    
+    Strategy:
+    1. Paginate through ALL rows (1.4M+)
+    2. Deduplicate by complaint_number (keep most recent)
+    3. Filter to complaints received in the last 18 months
+    4. Filter to allowed boroughs
+    """
+    cache_key = "stalled_official"
+    cached = _get_cached(cache_key)
+    if cached is not None:
+        return cached
+
+    url = "https://data.cityofnewyork.us/resource/i296-73x5.json"
+    
+    # Paginate through all data
+    all_data = []
+    offset = 0
+    page_size = 50000
+    
+    print(f"[stalled_official] Fetching all rows (paginating by {page_size})...")
+    
+    while True:
+        params = {
+            "$limit": page_size,
+            "$offset": offset,
+        }
+        
+        try:
+            data = _request(url, params)
+        except Exception as e:
+            print(f"[stalled_official] API request failed at offset {offset}: {e}")
+            break
+        
+        if not data:
+            break
+            
+        all_data.extend(data)
+        print(f"[stalled_official] Fetched {len(all_data)} rows so far...")
+        
+        if len(data) < page_size:
+            break
+        
+        offset += page_size
+    
+    if not all_data:
+        print("[stalled_official] No data returned from API")
+        return pd.DataFrame()
+    
+    df = pd.DataFrame(all_data)
+    print(f"[stalled_official] Total rows fetched: {len(df)}")
+
+    # Deduplicate by complaint_number - keep first occurrence (arbitrary, they should be same complaint)
+    if "complaint_number" in df.columns:
+        before = len(df)
+        df = df.drop_duplicates(subset=["complaint_number"], keep="first")
+        print(f"[stalled_official] Deduplicated: {before} -> {len(df)} unique complaints")
+
+    # Parse complaint date
+    if "date_complaint_received" in df.columns:
+        df["complaint_date"] = pd.to_datetime(df["date_complaint_received"], errors="coerce")
+        
+        # Filter to only complaints from the last 18 months
+        cutoff_date = pd.Timestamp.today() - pd.DateOffset(months=18)
+        before_filter = len(df)
+        df = df[df["complaint_date"] >= cutoff_date].copy()
+        print(f"[stalled_official] 18-month filter (>= {cutoff_date.date()}): {before_filter} -> {len(df)} rows")
+        
+        # Calculate days stalled
+        df["days_stalled"] = (pd.Timestamp.today() - df["complaint_date"]).dt.days
+
+    # Map borough to standard names
+    boro_map = {
+        "1": "MANHATTAN", "MANHATTAN": "MANHATTAN", "Manhattan": "MANHATTAN",
+        "2": "BRONX", "BRONX": "BRONX", "Bronx": "BRONX",
+        "3": "BROOKLYN", "BROOKLYN": "BROOKLYN", "Brooklyn": "BROOKLYN",
+        "4": "QUEENS", "QUEENS": "QUEENS", "Queens": "QUEENS",
+        "5": "STATEN ISLAND", "STATEN ISLAND": "STATEN ISLAND", "Staten Island": "STATEN ISLAND",
+    }
+    if "borough_name" in df.columns:
+        df["borough"] = df["borough_name"].astype(str).str.strip().map(lambda x: boro_map.get(x, x.upper() if isinstance(x, str) else x))
+    
+    # Filter to allowed boroughs
+    if "borough" in df.columns:
+        before_boro = len(df)
+        df = df[df["borough"].isin(ALLOWED_BOROUGHS)].copy()
+        print(f"[stalled_official] Borough filter: {before_boro} -> {len(df)} rows")
+
+    # Build full address
+    if "house_number" in df.columns and "street_name" in df.columns:
+        df["full_address"] = (
+            df["house_number"].fillna("").astype(str).str.strip() + " " +
+            df["street_name"].fillna("").astype(str).str.title().str.strip() + ", " +
+            df.get("borough", "").fillna("").astype(str)
+        )
+
+    # Sort by most stalled first (oldest complaint = most days)
+    if "days_stalled" in df.columns:
+        df = df.sort_values("days_stalled", ascending=False).reset_index(drop=True)
+
+    print(f"[stalled_official] Final: {len(df)} active stalled sites")
+    _set_cached(cache_key, df)
+    return df
+
+
+def _fetch_likely_stalled() -> pd.DataFrame:
+    """
+    Fetch stalled construction projects from DOB Stalled Construction Sites API.
+    This is now just an alias for the official stalled feed.
+    """
+    return _fetch_stalled_official()
+
+
+# ---------- LEADS UNPERMITTED ----------
+
+def _fetch_leads_unpermitted(days: int = DEFAULT_DAYS_WINDOW) -> Tuple[pd.DataFrame, float]:
+    """
+    Find filings that don't have corresponding permits yet.
+    Cross-reference job_filings with permit_issuance.
+    """
+    t0 = time.time()
+    
+    # Get filings
+    filings_df = _fetch_filings_last_days(days)
+    if filings_df.empty:
+        return pd.DataFrame(), time.time() - t0
+    
+    # Get permits
+    permits_df = _fetch_permits_last_days(days)
+    
+    # Extract base job numbers for matching
+    if "job_filing_number" in filings_df.columns:
+        filings_df["_job_base"] = filings_df["job_filing_number"].apply(_job_base)
+    else:
+        filings_df["_job_base"] = ""
+    
+    # Get permitted job bases
+    permitted_jobs = set()
+    if not permits_df.empty and "job__" in permits_df.columns:
+        permitted_jobs = set(permits_df["job__"].dropna().astype(str).str.strip())
+    
+    # Filter to unpermitted filings
+    mask = ~filings_df["_job_base"].isin(permitted_jobs)
+    unpermitted = filings_df[mask].copy()
+    
+    # Drop helper column
+    unpermitted.drop(columns=["_job_base"], inplace=True, errors="ignore")
+    
+    return unpermitted, time.time() - t0
 
-    result = df.sort_values("filing_date", ascending=False, kind="mergesort")
-    _set_cached(cache_key, result)
-    return result
 
 # ---------- public API ----------
 class SocrataClient:
@@ -332,57 +543,30 @@ class SocrataClient:
         if not SOCRATA_APP_TOKEN:
             print("⚠️  SOCRATA_APP_TOKEN not set – API may cap at 1,000 rows.")
 
-    def fetch_dataset_last_n_days(self, dataset_key: str, days: int) -> Tuple[pd.DataFrame, float]:
+    def fetch_dataset_last_n_days(
+        self,
+        dataset_key: str,
+        days: int,
+    ) -> Tuple[pd.DataFrame, float]:
         t0 = time.time()
+        
         if dataset_key == "job_filings":
             df = _fetch_filings_last_days(days)
         elif dataset_key == "permit_issuance":
             df = _fetch_permits_last_days(days)
+        elif dataset_key == "electrical_permits":
+            df = _fetch_electrical_last_days(days)
+        elif dataset_key == "stalled_official":
+            df = _fetch_stalled_official()
+        elif dataset_key == "likely_stalled":
+            df = _fetch_likely_stalled()
         else:
             raise ValueError(f"Unknown dataset: {dataset_key}")
+        
         return df, time.time() - t0
 
-    def fetch_leads_unpermitted(self, days: int = DEFAULT_DAYS_WINDOW) -> Tuple[pd.DataFrame, float]:
-        """
-        Sales leads = job_filings (last N days, MN/BK/QN) minus permits issued (last N days, MN/BK/QN).
-        Now with parallel fetching for both datasets!
-        """
-        print(f"[leads] pulling last {days} days (parallel)…")
-        t0 = time.time()
-
-        # Parallel fetch both datasets at the dataset level
-        with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
-            filings_future = executor.submit(_fetch_filings_last_days, days)
-            permits_future = executor.submit(_fetch_permits_last_days, days)
-            
-            filings_df = filings_future.result()
-            permits_df = permits_future.result()
-
-        if filings_df.empty:
-            return filings_df, time.time() - t0
-
-        # set of permitted bases
-        permitted_bases = set()
-        if not permits_df.empty and "job__" in permits_df.columns:
-            permitted_bases = {
-                _job_base(str(j)) for j in permits_df["job__"].dropna().astype(str)
-            }
-            print(f"[leads] Found {len(permitted_bases)} unique permitted job bases")
-
-        leads = filings_df.copy()
-        if "job_filing_number" in leads.columns and permitted_bases:
-            bases = leads["job_filing_number"].dropna().astype(str).map(_job_base)
-            before = len(leads)
-            leads = leads[~bases.isin(permitted_bases)].copy()
-            print(f"[leads] Filtered {before - len(leads)} permitted jobs, {len(leads)} leads remain")
-
-        # add marker
-        leads["has_permit_already"] = False
-
-        total_time = time.time() - t0
-        print(f"[leads] Total time: {total_time:.1f}s")
-        
-        return (
-            leads.sort_values("filing_date", ascending=False, kind="mergesort").reset_index(drop=True),
-            total_time,
-        )
+    def fetch_leads_unpermitted(
+        self,
+        days: int = DEFAULT_DAYS_WINDOW,
+    ) -> Tuple[pd.DataFrame, float]:
+        return _fetch_leads_unpermitted(days)