"""Detect new funding, hiring, and news signals for a Dealroom watchlist. Install: pip install authlib requests python-dotenv Run: Create a .env file with DEALROOM_CLIENT_ID and DEALROOM_CLIENT_SECRET. python startup_watchlist.py The first run saves a baseline and emits no alerts. Later runs compare the live API response with that local state file and print only changes. """ import argparse import json import os import time from datetime import datetime, timezone from pathlib import Path from authlib.integrations.requests_client import OAuth2Session from dotenv import load_dotenv load_dotenv() API_BASE = os.environ.get("DEALROOM_API_BASE", "https://api.beta.dealroom.app") AUTH_URL = os.environ.get( "DEALROOM_AUTH_URL", "https://accounts.beta.dealroom.co/oauth/token" ) AUDIENCE = os.environ.get( "DEALROOM_AUDIENCE", "https://api-next.beta.dealroom.co" ) USER_AGENT = os.environ.get( "DEALROOM_USER_AGENT", "dealroom-cookbook-startup-watchlist/1.0" ) DEFAULT_WATCHLIST = [ "a3741a91-cbe8-4a50-9dba-69b4ee612973", # Cerrion "3e03e957-749f-4d40-99e9-67ccab228390", # Aris Machina "97214a00-4ed4-4e62-9ebb-fc78bb57a5ca", # sensmore "51dd7312-3b04-4a7a-9489-625ab82f08ff", # Oversonic "f15afc9c-2261-488a-bf85-548422ae0446", # Juna.Ai "022f868e-7d95-4078-aec3-73d1ae28e19b", # Manex AI "07515066-ae36-4e7c-8fc8-ec9ab463f7e5", # Xelerit "22de85c0-0d32-423b-932f-ea2eb5e84d1f", # Raeon ] class DealroomClient: """OAuth2 client with one automatic token refresh after a 401 response.""" def __init__(self): client_id = os.environ["DEALROOM_CLIENT_ID"] client_secret = os.environ["DEALROOM_CLIENT_SECRET"] self.session = OAuth2Session( client_id=client_id, client_secret=client_secret, token_endpoint=AUTH_URL, ) self.session.headers.update( {"X-Client-Id": client_id, "User-Agent": USER_AGENT} ) self._fetch_token() def _fetch_token(self): self.session.fetch_token( url=AUTH_URL, grant_type="client_credentials", audience=AUDIENCE, ) def get(self, path, **kwargs): for attempt in range(5): response = self.session.get(f"{API_BASE}{path}", **kwargs) if response.status_code == 401 and attempt == 0: self._fetch_token() continue if response.status_code == 429 or response.status_code >= 500: if attempt == 4: response.raise_for_status() retry_after = response.headers.get("Retry-After") try: delay = float(retry_after) if retry_after else min(2**attempt, 8) except ValueError: delay = min(2**attempt, 8) time.sleep(delay) continue response.raise_for_status() return response.json() raise RuntimeError(f"Dealroom request failed after retries: {path}") def list_all(self, path, params=None, page_size=100, max_pages=20): query = dict(params or {}) query["limit"] = page_size records = [] for _ in range(max_pages): payload = self.get(path, params=query) page_records = payload.get("data", []) records.extend(page_records) page = payload.get("page", {}) next_cursor = page.get("next_cursor") total = page.get("total") if next_cursor: query["cursor"] = next_cursor elif total is not None and len(records) < total and page_records: query["offset"] = len(records) else: break return records def watchlist_ids(): configured = os.environ.get("DEALROOM_WATCHLIST_IDS", "") if configured.strip(): return [value.strip() for value in configured.split(",") if value.strip()] return DEFAULT_WATCHLIST def round_date(round_record): year = round_record.get("year") if not year: return None month = round_record.get("month") or 1 return f"{year}-{month:02d}-01" def find_news_company(article, watched_ids): entities = article.get("entities", []) for entity in entities: if entity.get("uuid") in watched_ids and entity.get("mention_role") == "subject": return entity return next( (entity for entity in entities if entity.get("uuid") in watched_ids), None ) def fetch_snapshot(client, ids): id_filter = "|".join(ids) company_rows = client.list_all( "/data/companies", {"filter": f"id[in_any]:{id_filter}"} ) companies = { row["uuid"]: { "uuid": row["uuid"], "name": row["name"], "employee_count": row.get("employee_count"), "open_jobs_count": row.get("company", {}).get("open_jobs_count", 0), "is_hiring": row.get("company", {}).get("is_hiring", False), } for row in company_rows } funding = {} for company_id in ids: company = companies.get(company_id, {"name": company_id}) rounds = client.list_all( f"/data/companies/{company_id}/funding-rounds", {"sort": "-date"} ) for item in rounds: funding[str(item["id"])] = { "id": str(item["id"]), "company_id": company_id, "company_name": company["name"], "date": round_date(item), "amount_usd": item.get("amount"), "round_type": item.get("standardized_round") or item.get("round_type") or "Funding", "source_url": item.get("source_url"), } time.sleep(0.25) news = {} news_rows = client.list_all( "/data/news", {"filter": f"entity_id[in_any]:{id_filter}", "sort": "-publish_date"}, ) watched_ids = set(ids) for item in news_rows: entity = find_news_company(item, watched_ids) if not entity: continue news[str(item["id"])] = { "id": str(item["id"]), "company_id": entity["uuid"], "company_name": entity["name"], "title": item.get("title"), "publish_date": item.get("publish_date"), "article_type": item.get("article_type"), "url": item.get("url"), } jobs = {} job_rows = client.list_all( "/data/jobs", {"filter": f"entity_id[in_any]:{id_filter}", "sort": "-date_posted"}, ) for item in job_rows: entity = item.get("entity", {}) jobs[str(item["id"])] = { "id": str(item["id"]), "company_id": entity.get("uuid"), "company_name": entity.get("name"), "title": item.get("title"), "location": item.get("formatted_location"), "date_posted": item.get("date_posted"), "url": item.get("url"), } return { "version": 1, "saved_at": datetime.now(timezone.utc).isoformat(), "companies": companies, "funding": funding, "news": news, "jobs": jobs, } def money(value): if value is None or value == 0: return "an undisclosed amount" if value >= 1_000_000: return f"${value / 1_000_000:g}M" return f"${value / 1_000:g}K" def detect_changes(previous, current): alerts = [] for record_id in current["funding"].keys() - previous["funding"].keys(): item = current["funding"][record_id] alerts.append( { "kind": "new_funding", "company": item["company_name"], "message": f'{item["company_name"]} recorded a {item["round_type"]} round of {money(item["amount_usd"])}.', "record": item, } ) for record_id in current["news"].keys() - previous["news"].keys(): item = current["news"][record_id] alerts.append( { "kind": "new_news", "company": item["company_name"], "message": f'{item["company_name"]}: {item["title"]}', "record": item, } ) for record_id in current["jobs"].keys() - previous["jobs"].keys(): item = current["jobs"][record_id] alerts.append( { "kind": "job_opened", "company": item["company_name"], "message": f'{item["company_name"]} opened a {item["title"]} role.', "record": item, } ) for record_id in previous["jobs"].keys() - current["jobs"].keys(): item = previous["jobs"][record_id] alerts.append( { "kind": "job_removed", "company": item["company_name"], "message": f'{item["company_name"]} no longer lists its {item["title"]} role.', "record": item, } ) for company_id, company in current["companies"].items(): old = previous["companies"].get(company_id) if not old: continue if company["open_jobs_count"] != old.get("open_jobs_count"): alerts.append( { "kind": "hiring_count_changed", "company": company["name"], "message": f'{company["name"]} now lists {company["open_jobs_count"]} open roles, previously {old.get("open_jobs_count", 0)}.', "record": { "company_id": company_id, "before": old.get("open_jobs_count", 0), "after": company["open_jobs_count"], }, } ) return sorted( alerts, key=lambda alert: ( alert["kind"], alert.get("company") or "", str(alert.get("record", {}).get("id") or ""), ), ) def read_state(path): if not path.exists(): return None return json.loads(path.read_text(encoding="utf-8")) def write_state(path, state): path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(f"{path.suffix}.tmp") temporary.write_text(f"{json.dumps(state, indent=2)}\n", encoding="utf-8") temporary.replace(path) def result_payload(current, previous, alerts, state_path): return { "mode": "changes_detected" if previous else "baseline_created", "checked_at": current["saved_at"], "state_file": str(state_path), "summary": { "companies_tracked": len(current["companies"]), "funding_rounds_seen": len(current["funding"]), "news_items_seen": len(current["news"]), "open_jobs_seen": len(current["jobs"]), "changes": len(alerts), }, "alerts": alerts, } def print_human(result): summary = result["summary"] if result["mode"] == "baseline_created": print(f'Baseline created for {summary["companies_tracked"]} companies.') print("No alerts were emitted on the first run.") elif not result["alerts"]: print(f'Checked {summary["companies_tracked"]} companies. No changes found.') else: print(f'Found {summary["changes"]} changes:') for alert in result["alerts"]: print(f' [{alert["kind"]}] {alert["message"]}') print(f'State saved to {result["state_file"]}') def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--state", type=Path, default=Path(".dealroom-watchlist-state.json"), help="Path to the persisted baseline JSON file.", ) parser.add_argument( "--json", action="store_true", help="Print machine-readable JSON output." ) args = parser.parse_args() previous = read_state(args.state) current = fetch_snapshot(DealroomClient(), watchlist_ids()) alerts = detect_changes(previous, current) if previous else [] write_state(args.state, current) result = result_payload(current, previous, alerts, args.state) if args.json: print(json.dumps(result, indent=2)) else: print_human(result) if __name__ == "__main__": main()