#!/usr/bin/env python3 """Recent funding announcements -> a research spreadsheet (CSV), with investor organisations where stated. Runs the Apify Actor nexgensignal/funding-round-intelligence-api (SEC Form D filings + TechCrunch funding news, plus PR Newswire funding releases with named investor organisations), waits for it, and upserts the records into a CSV keyed on record_id, so the same financing never appears twice in the file. A state file remembers a started run until its results are saved, so an interrupted save is finished from that run instead of paying for a new one. Billing note: the CSV is de-duplicated, the charges are not. A financing delivered again by a later run (for example when weekly windows overlap) is charged again even though its row is only updated. Needs Python 3.8+ and nothing else. Set your own token first: export APIFY_TOKEN=your_apify_token (Windows: set APIFY_TOKEN=...) python funding_to_research_csv.py Options: --days 7 --max 25 --min-amount 1000000 --query "robotics" --out funding_research.csv --cap 2.50 Recovery: --resume-run RUN_ID, --start-new (see the tutorial). """ import argparse, csv, datetime as dt, http.client, json, os, sys, time, urllib.error, urllib.parse, urllib.request ACTOR = "nexgensignal~funding-round-intelligence-api" KEY = "record_id" COLUMNS = ["record_id", "date", "company_name", "source", "event_type", "raise_status", "financing_type", "amount", "currency", "amount_type", "amount_display", "round_stage", "lead_investors", "investor_organizations", "investor_info_status", "industry", "location", "issuer", "title", "url", "other_reports", "cik", "first_seen", "last_seen"] LOG_COLUMNS = ["run_at", "run_id", "outcome", "records_received", "new_rows", "updated_rows", "charged_records", "status_message"] TECH_HEALTH = ["Other Technology", "Computers", "Biotechnology", "Pharmaceuticals", "Other Health Care"] # --- apify run core (shared by the three example scripts; edit _runcore.py and run build_scripts.py) --- import threading _STATE_LOCK = threading.Lock() API = os.environ.get("APIFY_API_BASE", "https://api.apify.com/v2") CONSOLE_RUNS = "https://console.apify.com/actors/runs" class StartUnknown(Exception): """The request that starts a paid run was sent, but no answer came back.""" def api(method, path, token, body=None, query=None): """Reads (GET) are retried on network errors and busy/5xx answers. A run start (POST) is sent exactly once: the Apify API has no idempotency key for starting runs, so repeating it could start a second paid run.""" url = f"{API}{path}" + ("?" + urllib.parse.urlencode(query) if query else "") data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(url, data=data, method=method, headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json"}) tries = 5 if method == "GET" else 1 for attempt in range(tries): try: with urllib.request.urlopen(req, timeout=120) as r: return json.loads(r.read() or b"null") except urllib.error.HTTPError as e: if e.code == 404 and "/records/" in path: return None # no such record (for example a run that failed on bad input) if method == "GET" and e.code in (429, 500, 502, 503, 504) and attempt < tries - 1: time.sleep(2 ** attempt); continue if method != "GET" and e.code >= 500: raise StartUnknown(f"HTTP {e.code}") # the server may have created the run before failing raise SystemExit(f"Apify API refused {method} {path}: HTTP {e.code} {e.read()[:300]!r}") except (urllib.error.URLError, OSError, http.client.HTTPException) as e: if method == "GET" and attempt < tries - 1: time.sleep(2 ** attempt); continue if method != "GET": raise StartUnknown(f"{type(e).__name__}: {e}") raise def _unknown_message(job, at, extra=""): return (f"START OUTCOME UNKNOWN ({job}): a request to start a paid run was sent at {at} but no answer came " f"back, so it is not known whether Apify started that run. Nothing was started again. {extra}" f"Next: run this script again (it looks for that run and reuses it if found), or check {CONSOLE_RUNS} " f"and rerun with --resume-run RUN_ID to collect that run's results without paying again, or with " f"--start-new if you are sure no run was started.") def find_runs_since(token, actor, since_iso, run_input): """Runs of this Actor (yours) started since the uncertain request, whose saved input equals ours.""" since = dt.datetime.fromisoformat(since_iso) - dt.timedelta(seconds=120) # allow for clock differences runs = (api("GET", f"/acts/{actor}/runs", token, query={"desc": "true", "limit": 25}) or {}).get("data", {}) found = [] for r in runs.get("items", []): started = dt.datetime.fromisoformat(r["startedAt"].replace("Z", "+00:00")) if started < since: continue saved = api("GET", f"/key-value-stores/{r['defaultKeyValueStoreId']}/records/INPUT", token) if saved == run_input: found.append(r["id"]) return found def start_or_resume(token, actor, run_input, cap_usd, job_state, save, resume_id=None, start_new=False, job="run"): """Return the finished run for this job. Reuses a run already started (saved run ID, --resume-run, or a run found after an uncertain start) instead of paying for a new one. job_state is saved to disk by save().""" now = dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds") run = None if resume_id: run = api("GET", f"/actor-runs/{resume_id}", token)["data"] if run["actId"] != api("GET", f"/acts/{actor}", token)["data"]["id"]: raise SystemExit(f"Run {resume_id} belongs to a different Actor; nothing was done.") elif job_state.get("pending_run"): run = api("GET", f"/actor-runs/{job_state['pending_run']}", token)["data"] print(f"Reusing run {run['id']} started earlier (its results were not saved yet); no new run started.") elif job_state.get("start_attempt") and not start_new: att = job_state["start_attempt"] found = find_runs_since(token, actor, att["at"], att["input"]) if len(found) == 1: run = api("GET", f"/actor-runs/{found[0]}", token)["data"] print(f"Found run {run['id']} started by the earlier uncertain request; reusing it, no new run started.") else: extra = (f"{len(found)} matching runs were found: {', '.join(found)}. " if found else "No matching run was found yet. ") raise SystemExit(_unknown_message(job, att["at"], extra)) if run is None: job_state.pop("start_attempt", None) job_state["start_attempt"] = {"at": now, "input": run_input} save() try: run = api("POST", f"/acts/{actor}/runs", token, run_input, {"waitForFinish": 60, "maxTotalChargeUsd": cap_usd})["data"] except StartUnknown as e: raise SystemExit(_unknown_message(job, now, f"({e}) ")) job_state.pop("start_attempt", None) job_state["pending_run"] = run["id"] # kept until this run's results are saved save() while run["status"] in ("READY", "RUNNING", "TIMING-OUT", "ABORTING"): run = api("GET", f"/actor-runs/{run['id']}", token, query={"waitForFinish": 60})["data"] return run def run_results(token, run): items = api("GET", f"/datasets/{run['defaultDatasetId']}/items", token, query={"clean": "true", "format": "json"}) receipt = api("GET", f"/key-value-stores/{run['defaultKeyValueStoreId']}/records/RUN_RECEIPT", token) or {} return items or [], receipt def load_state(path): return json.load(open(path)) if os.path.exists(path) else {} def save_state(path, state): with _STATE_LOCK: # two runs (UK numbers + names) may save at the same moment tmp = path + ".tmp" with open(tmp, "w") as f: json.dump(state, f, indent=1) os.replace(tmp, path) def outcome(run, n_items): """complete | no_match | partial | source_unavailable | failed — from the run status and its status message.""" msg = run.get("statusMessage") or "" if run["status"] != "SUCCEEDED": return "failed" if msg.startswith("SOURCE UNAVAILABLE"): return "source_unavailable" if (msg.startswith(("PARTIAL", "CAPPED", "INCOMPLETE SEARCH")) or "PARTIAL SEARCH" in msg or msg.startswith("The sources that answered")): return "partial" return "no_match" if n_items == 0 else "complete" def read_csv(path): if not os.path.exists(path): return [] with open(path, newline="", encoding="utf-8-sig") as f: return list(csv.DictReader(f)) def write_csv(path, rows, columns): tmp = path + ".tmp" with open(tmp, "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=columns, extrasaction="ignore") w.writeheader() w.writerows(rows) os.replace(tmp, path) # --- end of apify run core --- def names(v, with_role=False): out = [] for x in v or []: if isinstance(x, dict): n = x.get("name") or "" out.append(f"{n} ({x['role']})" if with_role and x.get("role") else n) else: out.append(str(x)) return "; ".join(o for o in out if o) def to_row(it): refs = [r for r in (it.get("source_references") or []) if r.get("record_id") != it.get("record_id")] row = {"record_id": it.get("record_id"), "date": it.get("date"), "company_name": it.get("company_name"), "source": it.get("source"), "event_type": it.get("event_type"), "raise_status": it.get("raise_status"), "financing_type": it.get("financing_type"), "amount": it.get("amount"), "currency": it.get("currency"), "amount_type": it.get("amount_type"), "amount_display": it.get("amount_display"), "round_stage": it.get("round_stage"), "lead_investors": names(it.get("lead_investors")), "investor_organizations": names(it.get("investor_organizations"), with_role=True), "investor_info_status": it.get("investor_info_status"), "industry": it.get("industry"), "location": it.get("location"), "issuer": it.get("issuer"), "title": it.get("title"), "url": it.get("url"), "other_reports": " | ".join(f"{r.get('source')}: {r.get('url')}" for r in refs), "cik": it.get("cik")} return {k: ("" if v is None else v) for k, v in row.items()} def main(): p = argparse.ArgumentParser(description=__doc__.split("\n")[0]) p.add_argument("--days", type=int, default=7, help="how many days back to read") p.add_argument("--max", type=int, default=25, help="most records per run (each is charged)") p.add_argument("--min-amount", type=int, default=0, help="only events with a known USD amount at or above this") p.add_argument("--query", default=None, help="optional company or keyword filter") p.add_argument("--all-industries", action="store_true", help="do not limit SEC filings to tech and health") p.add_argument("--no-prnewswire", action="store_true", help="skip PR Newswire (no investor organisations)") p.add_argument("--cap", type=float, default=2.50, help="maximum charge for one run, USD (includes the start fee)") p.add_argument("--out", default="funding_research.csv") p.add_argument("--resume-run", default=None, help="collect this existing run's results instead of starting one") p.add_argument("--start-new", action="store_true", help="after an uncertain start you have checked: start anew") a = p.parse_args() token = os.environ.get("APIFY_TOKEN") if not token: sys.exit("Set APIFY_TOKEN to your own Apify API token first.") state_path = a.out + ".state.json" state = load_state(state_path) job = state.setdefault("job", {}) save = lambda: save_state(state_path, state) run_input = {"mode": "all_sources", "includePrNewswire": not a.no_prnewswire, "daysBack": a.days, "maxResults": a.max} if not a.all_industries: run_input["industryGroups"] = TECH_HEALTH # applies to SEC filings only; news is any sector if a.min_amount: run_input["minAmount"] = a.min_amount if a.query: run_input["searchQuery"] = a.query run = start_or_resume(token, ACTOR, run_input, a.cap, job, save, a.resume_run, a.start_new) items, receipt = run_results(token, run) result = outcome(run, len(items)) now = dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds") rows = read_csv(a.out) by_key = {r[KEY]: r for r in rows} new = updated = 0 for it in items: row = to_row(it) k = row.get(KEY) if not k: continue if k in by_key: by_key[k].update(row); by_key[k]["last_seen"] = now; updated += 1 else: row["first_seen"] = row["last_seen"] = now by_key[k] = row; rows.append(row); new += 1 if items: write_csv(a.out, rows, COLUMNS) # if this fails, the run ID stays saved: rerun to finish log = read_csv(a.out + ".runlog.csv") log.append({"run_at": now, "run_id": run["id"], "outcome": result, "records_received": len(items), "new_rows": new, "updated_rows": updated, "charged_records": receipt.get("delivered"), "status_message": run.get("statusMessage")}) write_csv(a.out + ".runlog.csv", log, LOG_COLUMNS) job.pop("pending_run", None) # results saved: forget the run save() print(f"{result}: {len(items)} received, {new} new, {updated} already in {a.out}. {run.get('statusMessage')}") if updated: print(f"Note: {updated} financing(s) were already in the file from an earlier run. Their rows were updated, " f"not duplicated, but a newly started run charges for them again.") if result in ("failed", "source_unavailable"): sys.exit(2) if __name__ == "__main__": main()