← back to Labj Cre Banks
Add LA Socrata CRE-prospect ingester (Active Businesses + LADBS permits -> realestate.socrata_cre_prospects); first load 15,805 net-new firms
b6b50389d06ee8ecf172847cc1a80953ef41caa4 · 2026-08-10 21:53:35 -0700 · steve
Files touched
A ingest/socrata/.gitignoreA ingest/socrata/README.mdA ingest/socrata/ingest.py
Diff
commit b6b50389d06ee8ecf172847cc1a80953ef41caa4
Author: steve <steve@designerwallcoverings.com>
Date: Mon Aug 10 21:53:35 2026 -0700
Add LA Socrata CRE-prospect ingester (Active Businesses + LADBS permits -> realestate.socrata_cre_prospects); first load 15,805 net-new firms
---
ingest/socrata/.gitignore | 2 +
ingest/socrata/README.md | 30 ++++++
ingest/socrata/ingest.py | 262 ++++++++++++++++++++++++++++++++++++++++++++++
3 files changed, 294 insertions(+)
diff --git a/ingest/socrata/.gitignore b/ingest/socrata/.gitignore
new file mode 100644
index 0000000..e0fef3d
--- /dev/null
+++ b/ingest/socrata/.gitignore
@@ -0,0 +1,2 @@
+socrata_prospects.tsv
+socrata_prospects.jsonl
diff --git a/ingest/socrata/README.md b/ingest/socrata/README.md
new file mode 100644
index 0000000..25282c9
--- /dev/null
+++ b/ingest/socrata/README.md
@@ -0,0 +1,30 @@
+# LA Socrata CRE-prospect ingester
+
+Pulls two FREE, keyless LA City open-data (SODA) datasets into a de-duplicated
+CRE advertiser-prospect table, surfacing only firms NOT already in the curated
+LABJ set (`data/labj_master.json`).
+
+| Source | Dataset | Yields | Role |
+|---|---|---|---|
+| Listing of Active Businesses | `6rrh-rzua` (current) | ~65k CRE-NAICS firms | Broker / Property Manager / Architect-Engineer / Developer |
+| LADBS Building Permits | `d9aa-v8bm` (**historical 2013–2023**) | contractors on Commercial/Apartment permits | Contractor |
+
+Output: Postgres `realestate.socrata_cre_prospects` (upsert on `dedup_key`,
+`first_seen`/`last_seen` tracked) + a `socrata_prospects.jsonl` staging file.
+
+## Run
+```
+python3 ingest.py # 15k businesses + 24mo permits -> PG
+python3 ingest.py --dry-run # no DB write; prints role breakdown
+python3 ingest.py --businesses-only --max-businesses 40000
+python3 ingest.py --include-lessors # also pull the 43k "Lessors of real estate" (property owners)
+```
+`SOCRATA_APP_TOKEN` (optional) avoids throttling on large pulls. Cost: $0 (keyless, local PG).
+
+## Known limitation / next step
+`d9aa-v8bm` is a HISTORICAL snapshot ending 2023-05-19, so the permit window is
+computed relative to the dataset's own max date (not today). For a live permit
+feed, source the current LADBS "Permit Information" dataset and swap `PERMITS_ID`.
+
+To keep this fresh automatically it should run on a schedule (launchd/cron) —
+that's a Steve-gated scheduled-job install, not auto-added here.
diff --git a/ingest/socrata/ingest.py b/ingest/socrata/ingest.py
new file mode 100644
index 0000000..5b8e1b9
--- /dev/null
+++ b/ingest/socrata/ingest.py
@@ -0,0 +1,262 @@
+#!/usr/bin/env python3
+"""
+LA Socrata CRE-prospect ingester.
+
+Pulls two free, keyless LA City open-data (Socrata / SODA) datasets and turns them
+into de-duplicated CRE advertiser-prospect firm records:
+
+ 1. LADBS Building Permits (d9aa-v8bm) — Commercial + Apartment permits, recent,
+ with a named contractor -> role=Contractor (+ applicant person, if owner).
+ 2. Listing of Active Businesses (6rrh-rzua) — NAICS-filtered to CRE-relevant firms
+ (RE brokers, property managers, RE activities, commercial/multifamily builders,
+ architecture/engineering) -> role by NAICS.
+
+Output: a Postgres staging table `socrata_cre_prospects` in the local `realestate`
+DB (same DB as rentv_licensed_targets) via TSV + psql \\copy + upsert, PLUS a JSONL
+staging file. Firms already in the curated LABJ set (labj_master.json) are skipped so
+this surfaces only NET-NEW names.
+
+Zero third-party deps: urllib (fetch) + psql (load). Free ($0) + keyless; set
+SOCRATA_APP_TOKEN to avoid throttling on large pulls.
+
+Usage:
+ python3 ingest.py # bounded default pull, load to PG
+ python3 ingest.py --businesses-only
+ python3 ingest.py --permits-only
+ python3 ingest.py --max-businesses 20000 --permit-months 18 --min-valuation 250000
+ python3 ingest.py --dry-run # fetch + normalize + dedup, no DB write
+"""
+import argparse, json, os, re, subprocess, sys, urllib.parse, urllib.request, datetime
+
+BASE = "https://data.lacity.org/resource"
+PERMITS_ID = "d9aa-v8bm"
+BUSINESS_ID = "6rrh-rzua"
+APP_TOKEN = os.environ.get("SOCRATA_APP_TOKEN", "")
+PGDB = os.environ.get("PGDATABASE", "realestate")
+PGHOST = os.environ.get("PGHOST", "/tmp")
+HERE = os.path.dirname(os.path.abspath(__file__))
+CURATED = os.path.join(HERE, "..", "..", "data", "labj_master.json")
+TABLE = "socrata_cre_prospects"
+
+# NAICS prefix -> role for the Active Businesses pull. "Lessors" (5311) is huge and
+# low-signal per-record, so it's OFF by default (enable with --include-lessors).
+NAICS_ROLE = {
+ "5312": "Broker", # offices of real estate agents & brokers
+ "5313": "Property Manager", # property managers / other RE activities
+ "2362": "Developer/Builder", # nonresidential building construction
+ "23611": "Developer/Builder",# residential (incl. multifamily) construction
+ "5413": "Architect/Engineer",# architectural, engineering & related services
+}
+LESSOR_PREFIX = "5311"
+
+SUF = re.compile(r"\b(inc|llc|llp|lp|corp|corporation|company|co|ltd|the)\b")
+def norm(s):
+ s = SUF.sub(" ", re.sub(r"[^a-z0-9 ]", " ", (s or "").lower()))
+ return re.sub(r"\s+", " ", s).strip()
+
+def clean(v):
+ """TSV-safe single-line string."""
+ if v is None:
+ return ""
+ return re.sub(r"[\t\r\n]+", " ", str(v)).strip()
+
+def soda_get(dataset, params, cap):
+ """Paged SODA fetch (limit/offset). Returns list of dict rows, up to cap."""
+ rows, offset, page = [], 0, 5000
+ while len(rows) < cap:
+ q = dict(params)
+ q["$limit"] = min(page, cap - len(rows))
+ q["$offset"] = offset
+ url = f"{BASE}/{dataset}.json?" + urllib.parse.urlencode(q, safe="()' ,>=<")
+ req = urllib.request.Request(url, headers={"Accept": "application/json",
+ **({"X-App-Token": APP_TOKEN} if APP_TOKEN else {})})
+ with urllib.request.urlopen(req, timeout=60) as r:
+ batch = json.load(r)
+ if not batch:
+ break
+ rows.extend(batch)
+ offset += len(batch)
+ if len(batch) < q["$limit"]:
+ break
+ return rows
+
+def role_for_naics(naics):
+ n = (naics or "").strip()
+ for pref, role in NAICS_ROLE.items():
+ if n.startswith(pref):
+ return role
+ if n.startswith(LESSOR_PREFIX):
+ return "Property Owner (Lessor)"
+ return "Other RE"
+
+def fetch_businesses(cap, include_lessors):
+ prefixes = list(NAICS_ROLE.keys()) + ([LESSOR_PREFIX] if include_lessors else [])
+ where = " OR ".join(f"starts_with(naics,'{p}')" for p in prefixes)
+ where = f"({where}) AND business_name IS NOT NULL"
+ rows = soda_get(BUSINESS_ID, {
+ "$select": "location_account,business_name,street_address,city,zip_code,naics,"
+ "primary_naics_description,council_district,location_start_date",
+ "$where": where, "$order": "location_account",
+ }, cap)
+ out = []
+ for r in rows:
+ out.append({
+ "source": "la_active_businesses", "dataset_id": BUSINESS_ID,
+ "role": role_for_naics(r.get("naics")),
+ "firm_name": clean(r.get("business_name")), "contact_name": "",
+ "address": clean(r.get("street_address")), "city": clean(r.get("city")),
+ "state": "CA", "zip": clean(r.get("zip_code")),
+ "naics": clean(r.get("naics")), "naics_desc": clean(r.get("primary_naics_description")),
+ "permit_type": "", "permit_sub_type": "", "valuation": "",
+ "issue_date": clean(r.get("location_start_date"))[:10],
+ "work_description": "", "council_district": clean(r.get("council_district")),
+ "src_id": clean(r.get("location_account")),
+ })
+ return out
+
+def permits_max_date():
+ """This dataset (d9aa-v8bm) is a historical snapshot; window relative to ITS max
+ issue_date, not today, so the ingester always gets the freshest available permits.
+ (A live LADBS feed should replace this ID when sourced — see README.)"""
+ rows = soda_get(PERMITS_ID, {"$select": "max(issue_date)"}, 1)
+ m = (rows[0].get("max_issue_date") if rows else "") or "2023-05-19T00:00:00"
+ return datetime.date.fromisoformat(m[:10])
+
+def fetch_permits(months, min_valuation, cap):
+ cutoff = permits_max_date() - datetime.timedelta(days=int(months * 30))
+ where = ("(permit_sub_type='Commercial' OR permit_sub_type='Apartment') "
+ "AND contractors_business_name IS NOT NULL "
+ f"AND issue_date > '{cutoff.isoformat()}T00:00:00'")
+ rows = soda_get(PERMITS_ID, {
+ "$select": "pcis_permit,permit_type,permit_sub_type,issue_date,valuation,work_description,"
+ "address_start,street_direction,street_name,street_suffix,zip_code,"
+ "contractors_business_name,contractor_address,contractor_city,contractor_state,"
+ "applicant_first_name,applicant_last_name,applicant_relationship,council_district",
+ "$where": where, "$order": "issue_date DESC",
+ }, cap)
+ out = []
+ for r in rows:
+ try:
+ val = float(re.sub(r"[^0-9.]", "", r.get("valuation") or "") or 0)
+ except ValueError:
+ val = 0
+ if val < min_valuation:
+ continue
+ appl = " ".join(x for x in [r.get("applicant_first_name"), r.get("applicant_last_name")] if x).strip()
+ out.append({
+ "source": "ladbs_permits", "dataset_id": PERMITS_ID, "role": "Contractor",
+ "firm_name": clean(r.get("contractors_business_name")),
+ "contact_name": clean(appl) + (f" ({clean(r.get('applicant_relationship'))})" if r.get("applicant_relationship") else ""),
+ "address": clean(r.get("contractor_address")), "city": clean(r.get("contractor_city")),
+ "state": clean(r.get("contractor_state")) or "CA", "zip": "",
+ "naics": "", "naics_desc": "",
+ "permit_type": clean(r.get("permit_type")), "permit_sub_type": clean(r.get("permit_sub_type")),
+ "valuation": str(int(val)),
+ "issue_date": clean(r.get("issue_date"))[:10],
+ "work_description": clean(r.get("work_description"))[:300],
+ "council_district": clean(r.get("council_district")),
+ "src_id": clean(r.get("pcis_permit")),
+ })
+ return out
+
+COLS = ["dedup_key", "source", "dataset_id", "role", "firm_name", "contact_name",
+ "address", "city", "state", "zip", "naics", "naics_desc", "permit_type",
+ "permit_sub_type", "valuation", "issue_date", "work_description",
+ "council_district", "src_id"]
+
+def load_pg(records, tsv_path):
+ with open(tsv_path, "w") as f:
+ for r in records:
+ f.write("\t".join(clean(r.get(c, "")) for c in COLS) + "\n")
+ ddl = f"""
+CREATE TABLE IF NOT EXISTS {TABLE} (
+ dedup_key text PRIMARY KEY, source text, dataset_id text, role text,
+ firm_name text, contact_name text, address text, city text, state text, zip text,
+ naics text, naics_desc text, permit_type text, permit_sub_type text,
+ valuation text, issue_date text, work_description text, council_district text,
+ src_id text, first_seen timestamptz DEFAULT now(), last_seen timestamptz DEFAULT now()
+);
+CREATE TEMP TABLE _stg (LIKE {TABLE} INCLUDING DEFAULTS);
+ALTER TABLE _stg DROP COLUMN first_seen, DROP COLUMN last_seen;
+\\copy _stg ({','.join(COLS)}) FROM '{tsv_path}' WITH (FORMAT text, DELIMITER E'\\t');
+INSERT INTO {TABLE} ({','.join(COLS)})
+ SELECT {','.join(COLS)} FROM _stg
+ON CONFLICT (dedup_key) DO UPDATE SET last_seen = now(),
+ firm_name = EXCLUDED.firm_name, address = EXCLUDED.address,
+ role = EXCLUDED.role, naics_desc = EXCLUDED.naics_desc;
+SELECT source, role, count(*) FROM {TABLE} GROUP BY 1,2 ORDER BY 1,3 DESC;
+"""
+ p = subprocess.run(["psql", "-h", PGHOST, "-d", PGDB, "-v", "ON_ERROR_STOP=1"],
+ input=ddl, text=True, capture_output=True)
+ if p.returncode != 0:
+ print("PG load failed:\n", p.stderr, file=sys.stderr)
+ sys.exit(1)
+ print(p.stdout)
+
+def main():
+ ap = argparse.ArgumentParser()
+ ap.add_argument("--businesses-only", action="store_true")
+ ap.add_argument("--permits-only", action="store_true")
+ ap.add_argument("--max-businesses", type=int, default=15000)
+ ap.add_argument("--max-permits", type=int, default=20000)
+ ap.add_argument("--permit-months", type=int, default=24)
+ ap.add_argument("--min-valuation", type=int, default=250000)
+ ap.add_argument("--include-lessors", action="store_true")
+ ap.add_argument("--dry-run", action="store_true")
+ a = ap.parse_args()
+
+ recs = []
+ if not a.permits_only:
+ print(f"Fetching Active Businesses (CRE NAICS, cap {a.max_businesses})…", flush=True)
+ recs += fetch_businesses(a.max_businesses, a.include_lessors)
+ if not a.businesses_only:
+ print(f"Fetching LADBS permits (Commercial/Apartment, last {a.permit_months}mo, ≥${a.min_valuation:,})…", flush=True)
+ recs += fetch_permits(a.permit_months, a.min_valuation, a.max_permits)
+ print(f"Raw records: {len(recs)}")
+
+ # dedup vs curated LABJ set + within-run
+ curated = set()
+ try:
+ for r in json.load(open(CURATED))["records"]:
+ curated.add(norm(r["company_name"]))
+ for al in (r.get("aliases") or "").split(";"):
+ if al.strip():
+ curated.add(norm(al))
+ except Exception as e:
+ print("warn: could not read curated set:", e)
+
+ seen, out, skip_cur = set(), [], 0
+ for r in recs:
+ nf = norm(r["firm_name"])
+ if not nf:
+ continue
+ if nf in curated:
+ skip_cur += 1
+ continue
+ key = f"{r['source']}|{nf}|{r['zip'] or r['city']}"
+ if key in seen:
+ continue
+ seen.add(key)
+ r["dedup_key"] = key
+ out.append(r)
+
+ print(f"After dedup: {len(out)} net-new (skipped {skip_cur} already-curated, "
+ f"{len(recs)-len(out)-skip_cur} intra-run dups)")
+
+ jsonl = os.path.join(HERE, "socrata_prospects.jsonl")
+ with open(jsonl, "w") as f:
+ for r in out:
+ f.write(json.dumps(r) + "\n")
+ print(f"Wrote staging JSONL: {jsonl}")
+
+ if a.dry_run:
+ from collections import Counter
+ c = Counter((r["source"], r["role"]) for r in out)
+ for (s, role), n in c.most_common():
+ print(f" {n:>6} {s} · {role}")
+ print("(dry-run — no DB write)")
+ return
+ load_pg(out, os.path.join(HERE, "socrata_prospects.tsv"))
+
+if __name__ == "__main__":
+ main()
← 3fc6c4d Contrarian verify pass on 48 mid-market firms: 1 refuted (Ab
·
back to Labj Cre Banks
·
Socrata ingest: drafted weekly launchd plist (gated -> pendi 50ee64c →