#!/usr/bin/env python3 """Map an LP to recorded managers and those managers to fund vehicles.""" 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 from urllib.parse import urlparse 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 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]: params = {key: str(value).lower() if isinstance(value, bool) else value for key, value in params.items()} 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": "dealroom-lp-manager-cookbook/1.0", "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 domain(value: str | None) -> str: raw = (value or "").strip() if not raw: return "" parsed = urlparse(raw if "://" in raw else f"https://{raw}") return (parsed.hostname or "").lower().removeprefix("www.") def all_managers(client: DealroomClient, lp_id: str) -> list[dict[str, Any]]: result = {} offset = 0 while True: payload = client.get(f"/data/investors/{lp_id}/lp-funds", { "limit": 100, "offset": offset, "include_total": True, }) batch = rows(payload) for item in batch: result[item["uuid"]] = item offset += len(batch) total = (payload.get("page") or {}).get("total") if not batch or (total is not None and offset >= total): break return sorted(result.values(), key=lambda item: (item.get("name") or "", item["uuid"])) def build_network(client: DealroomClient, name: str, website: str, manager_limit: int, since: int, currency: str) -> dict[str, Any]: hits = rows(client.get("/data/search", {"q": name, "types": "investor", "limit": 10})) matches = [hit for hit in hits if hit.get("name") == name and domain(hit.get("website_domain")) == domain(website)] if len(matches) != 1: raise ValueError("Expected one LP matching both name and domain; refine the input.") lp = matches[0] managers = all_managers(client, lp["uuid"]) selected = managers[:manager_limit] output = [] for manager in selected: funds = {} offset = 0 while True: payload = client.get(f"/data/investors/{manager['uuid']}/funds", { "filter": f"year[gte]:{since}", "sort": "-fund_date", "limit": 100, "offset": offset, "include_total": True, "currency": currency, }) if payload.get("currency") != currency: raise ValueError("Unexpected response currency.") batch = rows(payload) for fund in batch: funds[fund["id"]] = fund offset += len(batch) total = (payload.get("page") or {}).get("total") if not batch or (total is not None and offset >= total): break output.append({ "manager": manager, "relationship": "LP recorded against manager", "funds": list(funds.values()), "lp_commitment_to_these_funds": "not established by these endpoints", }) return { "generated_at": datetime.now(timezone.utc).isoformat(), "currency": currency, "lp": {"uuid": lp["uuid"], "name": lp["name"], "website_domain": lp["website_domain"]}, "scope": {"since": since, "managers_available": len(managers), "managers_shown": len(selected), "selection": "manager name ascending", "funds_shown": sum(len(item["funds"]) for item in output)}, "caveat": "LP relationships are manager-level. Fund sizes are not LP commitments. " "No fund-level LP allocation is inferred. Missing records are not proof of no relationship.", "managers": output, } def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--lp", default="European Investment Fund") parser.add_argument("--domain", default="eif.org") parser.add_argument("--manager-limit", type=int, default=12) parser.add_argument("--since", type=int, default=2020) parser.add_argument("--currency", choices=["EUR", "USD", "GBP"], default="EUR") parser.add_argument("--output", default="network.json") args = parser.parse_args() if not 1 <= args.manager_limit <= 100 or not 1900 <= args.since <= datetime.now().year: parser.error("manager-limit must be 1–100 and since must be a year from 1900 through today.") load_dotenv() client_id = os.getenv("DEALROOM_CLIENT_ID") client_secret = os.getenv("DEALROOM_CLIENT_SECRET") if not client_id or not client_secret: parser.error("Set DEALROOM_CLIENT_ID and DEALROOM_CLIENT_SECRET in .env.") result = build_network(DealroomClient(client_id, client_secret), args.lp, args.domain, args.manager_limit, args.since, args.currency) with open(args.output, "w", encoding="utf-8") as handle: json.dump(result, handle, ensure_ascii=False, indent=2) handle.write("\n") print(json.dumps(result["scope"])) if __name__ == "__main__": main()