"""Track institutional and insider ownership across a watchlist. https://vizualstocks.com/api/tutorials/historical-ownership-python Python 3.10+. Install requests; set VIZUALSTOCKS_API_KEY in your environment. Read-only requests. Produces two CSVs, metadata and a self-contained HTML report. """ import argparse import csv from datetime import date, datetime, timedelta, timezone from email.utils import parsedate_to_datetime from html import escape import json import math import os from pathlib import Path import re import time import requests BASE_URL = "https://api.vizualstocks.com/v1" PERCENT_FIELDS = ("institutionsPercent", "insidersPercent", "undisclosedPercent") HISTORY_FIELDS = ["ticker", "observedDate", *PERCENT_FIELDS, "undisclosedPercentDerived", "interpretationStatus", "interpretationFlags", "institutionsChangePp", "insidersChangePp"] SUMMARY_FIELDS = ["ticker", "asOf", "status", "observationCount", "firstObservedDate", "latestObservedDate", "latestAgeDays", *PERCENT_FIELDS, "interpretationStatus", "interpretationFlags"] + [ f"{field}{days}d" for days in (30, 90) for field in ( "targetDate", "baselineDate", "baselineAgeDays", "comparisonStatus", "institutionsChangePp", "insidersChangePp", "baselineFlags")] class ApiError(ValueError): def __init__(self, status, message): self.status = status super().__init__(message) def iso_date(value): if not isinstance(value, str) or not re.fullmatch(r"\d{4}-\d{2}-\d{2}", value): raise ValueError("Use a date in YYYY-MM-DD format.") return date.fromisoformat(value) def normalize_tickers(tickers): result = [] for value in tickers: ticker = value.strip().upper() if not re.fullmatch(r"[A-Z0-9][A-Z0-9.-]{0,19}", ticker): raise ValueError("Use stock tickers such as PLTR, NVDA or BRK.B.") if ticker not in result: result.append(ticker) if not result: raise ValueError("Choose at least one ticker.") return result def make_session(): key = os.environ.get("VIZUALSTOCKS_API_KEY", "").strip() if not key: raise ValueError("Set VIZUALSTOCKS_API_KEY in this terminal first.") session = requests.Session() session.headers.update({"X-API-Key": key, "Accept": "application/json"}) return session def retry_delay(value, attempt): try: delay = float(value) except (ValueError, TypeError): try: delay = (parsedate_to_datetime(value) - datetime.now(timezone.utc)).total_seconds() except (ValueError, TypeError, OverflowError): delay = 2 ** attempt return max(0, min(30, delay)) if math.isfinite(delay) else 2 ** attempt def fetch_history(session, ticker, as_of=None): ticker = normalize_tickers([ticker])[0] params = {"order": "asc"} if as_of is not None: params["to"] = as_of.isoformat() for attempt in range(4): try: response = session.get(f"{BASE_URL}/companies/{ticker}/ownership-history", params=params, timeout=(10, 60), allow_redirects=False) except requests.RequestException: if attempt == 3: raise ApiError(None, "Connection failed after four attempts.") from None time.sleep(2 ** attempt) continue if response.status_code in (429, 500, 502, 503, 504) and attempt < 3: time.sleep(retry_delay(response.headers.get("Retry-After"), attempt)) continue if response.status_code != 200: hints = {401: "Check your API key.", 403: "Ticker unavailable on this API plan.", 404: "No ownership history found for this ticker.", 429: "Rate or quota limit reached; try again later."} raise ApiError(response.status_code, f"HTTP {response.status_code}: " + hints.get(response.status_code, "Request could not be completed.")) try: payload = response.json() except ValueError: raise ApiError(200, "The API returned invalid JSON.") from None if not isinstance(payload, dict) or not isinstance(payload.get("ticker"), str) or payload["ticker"].upper() != ticker: raise ApiError(200, "The response did not match the requested ticker.") return payload def numeric(value): return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value) def normalize_history(payload): """Keep missing values and signed remainders; never fill gaps or clamp to 100.""" if not isinstance(payload, dict) or not isinstance(payload.get("ticker"), str): raise ValueError("Expected a ticker in the ownership response.") ticker = normalize_tickers([payload["ticker"]])[0] if not isinstance(payload.get("data"), list): raise ValueError("Expected a data array in the ownership response.") rows, dates = [], set() for point in payload["data"]: if not isinstance(point, dict): raise ValueError("Expected an object for each ownership observation.") observed = iso_date(point.get("observedDate")).isoformat() if observed in dates: raise ValueError("Duplicate observation date; inspect the response before comparing.") dates.add(observed) row = {"ticker": ticker, "observedDate": observed} for field in PERCENT_FIELDS: value = point.get(field) if value is not None and not numeric(value): raise ValueError(f"Expected a finite number or null for {field}.") row[field] = value flags = point.get("interpretationFlags", []) if not isinstance(flags, list) or not all(isinstance(flag, str) for flag in flags): raise ValueError("Expected an array of interpretation flags.") row.update(undisclosedPercentDerived=point.get("undisclosedPercentDerived"), interpretationStatus=point.get("interpretationStatus"), interpretationFlags=list(flags)) rows.append(row) rows.sort(key=lambda row: row["observedDate"]) previous = None for row in rows: for field, output in (("institutionsPercent", "institutionsChangePp"), ("insidersPercent", "insidersChangePp")): row[output] = difference(row[field], previous[field]) if previous else None previous = row return rows def difference(current, baseline): return round(current - baseline, 6) if numeric(current) and numeric(baseline) else None def on_or_before(rows, cutoff): return max((row for row in rows if row["observedDate"] <= cutoff.isoformat()), key=lambda row: row["observedDate"], default=None) def summarize(ticker, rows, as_of, max_age_days=14): """Use one cutoff across tickers; never substitute an observation after a boundary.""" if max_age_days < 0: raise ValueError("max_age_days cannot be negative.") rows = [row for row in rows if row["observedDate"] <= as_of.isoformat()] latest = on_or_before(rows, as_of) result = dict.fromkeys(SUMMARY_FIELDS) result.update(ticker=ticker, asOf=as_of.isoformat(), status="no_history", observationCount=len(rows), interpretationFlags=[]) if rows: result["firstObservedDate"] = min(row["observedDate"] for row in rows) if latest: age = (as_of - iso_date(latest["observedDate"])).days result.update({field: latest[field] for field in PERCENT_FIELDS}) result.update(latestObservedDate=latest["observedDate"], latestAgeDays=age, status="ok" if age <= max_age_days else "old_latest_observation", interpretationStatus=latest["interpretationStatus"], interpretationFlags=latest["interpretationFlags"]) for days in (30, 90): target = as_of - timedelta(days=days) baseline = on_or_before(rows, target) suffix = f"{days}d" result["targetDate" + suffix] = target.isoformat() result["baselineFlags" + suffix] = baseline["interpretationFlags"] if baseline else [] if baseline: age = (target - iso_date(baseline["observedDate"])).days result["baselineDate" + suffix] = baseline["observedDate"] result["baselineAgeDays" + suffix] = age if latest is None: status = "no_history" elif result["status"] != "ok": status = "old_latest_observation" elif baseline is None: status = "insufficient_history" elif result["baselineAgeDays" + suffix] > max_age_days: status = "old_baseline_observation" else: status = "ok" for field, output in (("institutionsPercent", "institutionsChangePp"), ("insidersPercent", "insidersChangePp")): result[output + suffix] = difference(latest[field], baseline[field]) if any(result[field + suffix] is None for field in ("institutionsChangePp", "insidersChangePp")): status = "missing_values" result["comparisonStatus" + suffix] = status return result def rank_watchlist(rows): return sorted(rows, key=lambda row: ( row.get("institutionsChangePp90d") is None, -(row.get("institutionsChangePp90d") or 0), row["ticker"])) def monthly_observations(rows): """Last actual observation in each month, not an average or invented month-end row.""" months = {} for row in sorted(rows, key=lambda item: item["observedDate"]): months[row["observedDate"][:7]] = row return list(months.values()) def display(value, suffix="", signed=False): return (f"{value:+.2f}" if signed else f"{value:.2f}") + suffix if numeric(value) else "—" def cell(value): if isinstance(value, list): value = "; ".join(value) # Text originating outside the script must not become an Excel formula. if isinstance(value, str) and value.lstrip().startswith(("=", "+", "-", "@")): return "'" + value return value def write_csv(path, fields, rows): with path.open("x", encoding="utf-8-sig", newline="") as file: writer = csv.DictWriter(file, fieldnames=fields, extrasaction="ignore") writer.writeheader() writer.writerows({key: cell(value) for key, value in row.items()} for row in rows) def history_chart(rows, field, title, color): points = monthly_observations(rows)[-18:] valid = [row[field] for row in points if numeric(row[field])] if not valid: return f"

{escape(title)}

No numeric observations for this chart.

" high, low = max(1, max(valid)), min(0, min(valid)) scale = 135 / (high - low) zero = 160 + low * scale # Calendar spacing: an absent month remains a gap, not a compressed time step. month_index = lambda row: iso_date(row["observedDate"]).year * 12 + iso_date(row["observedDate"]).month first, last = month_index(points[0]), month_index(points[-1]) step = 282 / (last - first + 1) bars = [] for row in points: value = row[field] if not numeric(value): continue x, y = 42 + (month_index(row) - first) * step, zero - value * scale tip = escape(f"{row['observedDate']}: {value:.2f}%") bars.append(f'{tip}') return f'''

{escape(title)}

{high:.1f}%{''.join(bars)} {low:.0f}%{points[0]['observedDate']}{points[-1]['observedDate']}''' REPORT_CSS = """ :root{color-scheme:dark}*{box-sizing:border-box}body{margin:0;background:#0b1017;color:#e8edf6;font:15px/1.7 system-ui,sans-serif}main{max-width:1160px;margin:auto;padding:48px 24px}h1{font-size:clamp(27px,4vw,40px);line-height:1.2;letter-spacing:-.035em}h2{font-size:24px;margin:44px 0 18px}h3{margin:0 0 12px}h4{margin:12px 0 0}p,small,dt{color:#a5b4ca}a{color:#7cbbef}a:hover{color:#66dabb}.eyebrow{font-size:11px;color:#66dabb;letter-spacing:.13em}.downloads{display:flex;flex-wrap:wrap;gap:12px 24px;margin:24px 0}.cards{display:grid;grid-template-columns:repeat(auto-fit,minmax(min(100%,290px),1fr));gap:16px}.card{min-width:0;padding:22px;background:#111925;border:1px solid #2b374a;border-radius:12px}.metrics{display:grid;grid-template-columns:1fr 1fr;gap:12px}.metrics div{min-width:0}.metrics dt{font-size:12px}.metrics dd{margin:2px 0 0;font-size:20px;font-weight:600}.rise{color:#78bdb1}.fall{color:#b1a3cf}.change-row{display:grid;grid-template-columns:55px 1fr 86px;gap:12px;align-items:center;margin:16px 0;font-size:13px}.change-track{height:22px;position:relative;background:linear-gradient(90deg,transparent 49.7%,#65758a 49.7%,#65758a 50.3%,transparent 50.3%)}.change-track i{position:absolute;height:100%;border-radius:3px}.change-row strong{text-align:right}.axis{display:flex;justify-content:space-between;font-size:11px;color:#a5b4ca;margin-left:67px;margin-right:98px}svg{width:100%;display:block}svg text{fill:#a5b4ca;font:11px system-ui}details{margin-top:14px}summary{cursor:pointer;color:#7cbbef}details p{font-size:12px;overflow-wrap:anywhere}.note{font-size:12px}.flag{color:#d5bb88}.status{font-size:12px;overflow-wrap:anywhere}footer{border-top:1px solid #2b374a;margin-top:36px;padding-top:20px}@media(max-width:450px){main{padding:28px 16px}.card{padding:16px}.change-row{grid-template-columns:44px 1fr 76px;gap:8px}.axis{margin-left:52px;margin-right:84px}.metrics dd{font-size:18px}}@media(prefers-color-scheme:light){:root{color-scheme:light}body{background:#f5f7fb;color:#18253b}.card{background:white;border-color:#d4dfeb}p,small,dt,.axis{color:#50637c}a,summary{color:#25649a}.eyebrow,.rise{color:#287767}.fall{color:#75618e}.flag{color:#895c1e}svg text{fill:#50637c}} """ def render_report(summary, histories, metadata): bars, cards, charts = [], [], [] ranked = rank_watchlist(summary) maximum = max([abs(row["institutionsChangePp90d"]) for row in ranked if numeric(row.get("institutionsChangePp90d"))] or [1]) or 1 for row in ranked: ticker, value = escape(row["ticker"]), row.get("institutionsChangePp90d") if numeric(value): width = abs(value) / maximum * 50 left = 50 if value >= 0 else 50 - width color = "#78bdb1" if value >= 0 else "#b1a3cf" bars.append(f'
{ticker}
{display(value," pp",True)}
') metrics = [("Institutions", display(row.get("institutionsPercent"), "%")), ("Insiders", display(row.get("insidersPercent"), "%")), ("30-day change", display(row.get("institutionsChangePp30d"), " pp", True)), ("90-day change", display(value, " pp", True))] details = [] for days in (30, 90): suffix = f"{days}d" flags = "; ".join(row.get("baselineFlags" + suffix) or []) details.append(escape(f"{days}d target {row.get('targetDate'+suffix) or '—'}; baseline {row.get('baselineDate'+suffix) or '—'}; baseline gap {row.get('baselineAgeDays'+suffix) if row.get('baselineAgeDays'+suffix) is not None else '—'} days; {row.get('comparisonStatus'+suffix) or row['status']}." + (" Flags: " + flags if flags else ""))) flags = "; ".join(row.get("interpretationFlags") or []) flag_html = f'

{escape(flags)}

' if flags else "" cards.append(f'''

{ticker}

Observed {escape(row.get('latestObservedDate') or '—')} · {escape(row['status'].replace('_',' '))}

{''.join(f'
{label}
{value}
' for label,value in metrics)}
{flag_html}
Comparison dates

{'

'.join(details)}

{row.get('observationCount',0)} observations. First: {escape(row.get('firstObservedDate') or '—')}. Latest snapshot gap: {row.get('latestAgeDays') if row.get('latestAgeDays') is not None else '—'} days.

''') history = histories.get(row["ticker"], []) if history: charts.append(f'

{ticker}

' + history_chart(history,"institutionsPercent","Institutional ownership", "#78bdb1") + history_chart(history,"insidersPercent","Insider ownership", "#7c9ece") + '
') errors = ''.join(f'
  • {escape(item["ticker"])}: {escape(item["message"])}
    • ' for item in metadata["errors"]) return f'''Ownership watchlist · {metadata['asOf']}

    VIZUALSTOCKS · OWNERSHIP RESEARCH

    Is institutional ownership rising in your stocks?

    Watchlist cutoff: {metadata['asOf']} · Retrieved {metadata['retrievedAtUtc']}. Changes are percentage points, not trade counts.

    90-day ownership changes

    Decreasing0Increasing
    {''.join(bars) or '

    No comparable institutional readings for this window.

    '}

    Only numeric comparisons are charted. All requested tickers appear below, including missing history and access errors.

    Your watchlist

    {''.join(cards)}
    {'

    Request notes

    ' if errors else ''}

    Ownership over time

    Separate percentage scales, both starting at zero for nonnegative data. Last actual observation in each month; up to 18 observed months per chart. Hover a bar for its exact date and value. The CSV retains all returned observations.

    {''.join(charts) or '

    No history available.

    '}
    ''' def build_report(tickers, as_of, max_age_days, output_dir): tickers = normalize_tickers(tickers) if max_age_days < 0: raise ValueError("--max-age-days must be zero or greater.") if as_of > datetime.now(timezone.utc).date(): raise ValueError("Choose an --as-of date no later than today (UTC).") output = Path(output_dir) if output.exists(): raise ValueError("Output folder already exists. Choose a new --output-dir.") histories, summaries, methods, errors = {}, [], {}, [] retrieved = datetime.now(timezone.utc).isoformat() with make_session() as session: for ticker in tickers: try: payload = fetch_history(session, ticker, as_of) rows = [row for row in normalize_history(payload) if row["observedDate"] <= as_of.isoformat()] histories[ticker] = rows methods[ticker] = payload.get("methodology", {}) summaries.append(summarize(ticker, rows, as_of, max_age_days)) except (ApiError, ValueError) as error: if isinstance(error, ApiError) and error.status == 401: raise errors.append({"ticker": ticker, "message": str(error)}) row = summarize(ticker, [], as_of, max_age_days) row["status"] = "request_error" summaries.append(row) if not histories: raise ValueError("No ticker requests succeeded. " + " ".join(f"{e['ticker']}: {e['message']}" for e in errors)) summaries = rank_watchlist(summaries) metadata = {"retrievedAtUtc": retrieved, "asOf": as_of.isoformat(), "tickers": tickers, "maxAgeDays": max_age_days, "lookbackDays": [30, 90], "errors": errors, "endpoint": BASE_URL + "/companies/{ticker}/ownership-history", "methodologyByTicker": methods, "comparisonMethod": "Last observation on or before each boundary; no interpolation; changes in percentage points.", "observationDateMeaning": "Stored snapshot date, not a trade date or a source freshness guarantee."} output.mkdir(parents=True, exist_ok=False) write_csv(output / "watchlist.csv", SUMMARY_FIELDS, summaries) write_csv(output / "ownership-history.csv", HISTORY_FIELDS, [row for ticker in tickers for row in histories.get(ticker, [])]) with (output / "metadata.json").open("x", encoding="utf-8") as file: json.dump(metadata, file, indent=2, allow_nan=False) with (output / "report.html").open("x", encoding="utf-8") as file: file.write(render_report(summaries, histories, metadata)) return summaries, metadata def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--tickers", nargs="+", default=["PLTR", "NVDA", "TSLA"]) parser.add_argument("--as-of", type=iso_date, default=datetime.now(timezone.utc).date(), help="Common cutoff, YYYY-MM-DD; defaults to today in UTC.") parser.add_argument("--max-age-days", type=int, default=14, help="Largest allowed gap before each comparison boundary (default: 14 days).") parser.add_argument("--output-dir", default="ownership-research") args = parser.parse_args() try: summary, metadata = build_report(args.tickers, args.as_of, args.max_age_days, args.output_dir) except (ValueError, OSError) as error: parser.exit(1, f"Error: {error}\n") print(f"Saved {len(summary)} watchlist rows and full histories to {args.output_dir}/report.html") if metadata["errors"]: print(f"{len(metadata['errors'])} ticker request(s) unavailable; see Request notes and metadata.json.") if __name__ == "__main__": main()