#!/usr/bin/env python3 """Build a review queue of startups that pass Dealroom next-round signals.""" from __future__ import annotations import argparse import json import os import random import time from dataclasses import dataclass from datetime import datetime, timezone from email.utils import parsedate_to_datetime from typing import Any import requests from dotenv import load_dotenv API_BASE = "https://api.beta.dealroom.app" TOKEN_URL = "https://accounts.beta.dealroom.co/oauth/token" AUDIENCE = "https://api-next.beta.dealroom.co" RETRYABLE_STATUS = {429, 500, 502, 503, 504} def retry_delay(value: str | None, fallback: float) -> float: if not value: return fallback try: seconds = float(value) except ValueError: try: retry_at = parsedate_to_datetime(value) if retry_at.tzinfo is None: retry_at = retry_at.replace(tzinfo=timezone.utc) seconds = (retry_at - datetime.now(timezone.utc)).total_seconds() except (TypeError, ValueError, OverflowError): return fallback return min(max(seconds, 0.0), 30.0) @dataclass class DealroomClient: client_id: str client_secret: str user_agent: str token: str | None = None def authenticate(self) -> None: response = requests.post( TOKEN_URL, json={ "grant_type": "client_credentials", "client_id": self.client_id, "client_secret": self.client_secret, "audience": AUDIENCE, }, headers={"Accept": "application/json"}, timeout=30, ) response.raise_for_status() self.token = response.json()["access_token"] def get(self, path: str, params: dict[str, Any]) -> dict[str, Any]: if not self.token: self.authenticate() last_error: requests.HTTPError | None = None for attempt in range(4): response = requests.get( f"{API_BASE}{path}", params=params, headers={ "Authorization": f"Bearer {self.token}", "X-Client-Id": self.client_id, "User-Agent": self.user_agent, "Accept": "application/json", }, timeout=30, ) if response.ok: payload = response.json() if (payload.get("page") or {}).get("tier") or payload.get("locked"): raise RuntimeError("The API treated this as a capped non-M2M request.") return payload if response.status_code == 401 and attempt == 0: self.authenticate() continue try: response.raise_for_status() except requests.HTTPError as error: last_error = error if response.status_code not in RETRYABLE_STATUS or attempt == 3: raise last_error or RuntimeError(response.text) delay = retry_delay(response.headers.get("Retry-After"), 0.3 * (2**attempt)) time.sleep(delay + random.uniform(0, 0.2)) raise last_error or RuntimeError("Dealroom request failed") def rows(payload: dict[str, Any]) -> list[dict[str, Any]]: value = payload.get("data") return value if isinstance(value, list) else [] def exact_match(items: list[dict[str, Any]], label: str, source_type: str) -> dict[str, Any]: wanted = label.strip().lower() for item in items: name = str(item.get("label") or item.get("name") or "").strip().lower() item_type = item.get("source_type") or item.get("type") if name == wanted and item_type == source_type: return item raise ValueError(f"Could not resolve taxonomy value: {label}") def compact_money(value: Any) -> str: if value is None: return "undisclosed" number = float(value) if number >= 1_000_000_000: return f"${number / 1_000_000_000:.1f}B" if number >= 1_000_000: return f"${number / 1_000_000:.1f}M" return f"${number:,.0f}" def build_candidates(client: DealroomClient, args: argparse.Namespace) -> dict[str, Any]: locations = client.get( "/reference/filters/location/values", {"q": args.geography, "type": "continent", "limit": 20}, ) tags = client.get( "/reference/filters/search", {"q": args.market, "scope": "companies", "limit": 20}, ) location = exact_match(rows(locations), args.geography, "continent") tag = exact_match(rows(tags), args.market, "sector") location_id = location.get("value") or location.get("id") tag_id = tag.get("value") or tag.get("id") filters = [ f"tag_id[eq]:{tag_id}", f"hq_location[eq]:{location_id}", "is_startup[eq]:true", "company_status[eq]:operational", f"launch_date[gte]:{args.founded_since}", f"total_funding[gte]:{args.funding_min}", f"total_funding[lte]:{args.funding_max}", f"signal_timing[gte]:{args.timing_min}", f"signal_team[gte]:{args.team_min}", ] query: dict[str, Any] = { "filter": f"and({','.join(filters)})", "sort": "-signal_rating", "limit": args.pool, "include_total": "true", "currency": "USD", } payload = client.get("/data/companies", query) companies = [] for rank, entity in enumerate(rows(payload)[: args.limit], start=1): company = entity.get("company") or {} companies.append( { "rank": rank, "uuid": entity.get("uuid"), "name": entity.get("name"), "tagline": entity.get("tagline"), "dealroom_url": entity.get("dealroom_url"), "website": entity.get("website"), "hq_city": entity.get("hq_city"), "hq_country": entity.get("hq_country"), "launch_year": entity.get("launch_year"), "employee_count": entity.get("employee_count"), "employee_count_1y_growth_percent": entity.get("employee_count_1y_growth"), "total_funding_usd": company.get("total_funding"), "signal_rating": company.get("signal_rating"), "is_hiring": bool(company.get("is_hiring")), "open_jobs_count": company.get("open_jobs_count") or 0, } ) if not companies: raise RuntimeError("The candidate query returned no matches.") return { "generated_at": datetime.now(timezone.utc).isoformat(), "methodology": ( "Screen by Dealroom Timing and Team signals, then rank the returned " "company records by overall Signal score for human review." ), "query": {**query, "include_total": True}, "summary": { "available_matches": (payload.get("page") or {}).get("total"), "candidates_shown": len(companies), }, "companies": companies, "limitations": [ "Signal prioritizes research; it does not predict a financing event.", "Funding totals do not reveal runway or capital need.", "Hiring and headcount growth are context, not proof of fundraising activity.", ], } def markdown(snapshot: dict[str, Any]) -> str: lines = ["# Next-round research queue", "", snapshot["methodology"], ""] for company in snapshot["companies"]: location = ", ".join( value for value in [company["hq_city"], company["hq_country"]] if value ) or "Location unavailable" lines.extend( [ f"## {company['rank']}. {company['name']}", company.get("tagline") or "No tagline available.", f"- Headquarters: {location}", f"- Overall Signal: {company['signal_rating']}", f"- Total funding: {compact_money(company['total_funding_usd'])}", f"- Employees: {company['employee_count'] or 'Unavailable'}", f"- Open roles: {company['open_jobs_count']}", f"- Dealroom: {company.get('dealroom_url') or 'Not available'}", "", ] ) return "\n".join(lines) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Find startups approaching a next raise.") parser.add_argument("--market", default="Climate Tech") parser.add_argument("--geography", default="Europe") parser.add_argument("--founded-since", type=int, default=2019) parser.add_argument("--funding-min", type=int, default=1_000_000) parser.add_argument("--funding-max", type=int, default=50_000_000) parser.add_argument("--timing-min", type=int, default=70) parser.add_argument("--team-min", type=int, default=50) parser.add_argument("--pool", type=int, default=50) parser.add_argument("--limit", type=int, default=12) parser.add_argument("--json", action="store_true") return parser.parse_args() def main() -> None: load_dotenv() args = parse_args() if not 1 <= args.limit <= args.pool <= 500: raise SystemExit("Require 1 <= --limit <= --pool <= 500") if not 0 <= args.timing_min <= 100 or not 0 <= args.team_min <= 100: raise SystemExit("Signal thresholds must be between 0 and 100") if args.funding_min < 0 or args.funding_max <= args.funding_min: raise SystemExit("Funding max must exceed funding min") client_id = os.environ.get("DEALROOM_CLIENT_ID") client_secret = os.environ.get("DEALROOM_CLIENT_SECRET") if not client_id or not client_secret: raise SystemExit("Set DEALROOM_CLIENT_ID and DEALROOM_CLIENT_SECRET in .env") client = DealroomClient( client_id, client_secret, os.environ.get("DEALROOM_USER_AGENT", "your-next-round-research/1.0"), ) snapshot = build_candidates(client, args) print(json.dumps(snapshot, indent=2) if args.json else markdown(snapshot)) if __name__ == "__main__": main()