"""Download reported short-interest history to CSV and a two-panel PNG chart. Tutorial: https://vizualstocks.com/api/tutorials/historical-short-interest-python Python 3.10+. Install requests, pandas and matplotlib. Set VIZUALSTOCKS_API_KEY. """ import argparse from datetime import date, datetime, timezone import json import math import os from pathlib import Path import re import pandas as pd import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry BASE_URL = "https://api.vizualstocks.com/v1" FIELDS = ["settlementDate", "sharesShorted", "sharesFloat", "shortInterestPercentage"] 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"}) retries = Retry(total=3, backoff_factor=1, status_forcelist=[429, 502, 503, 504], allowed_methods=["GET"], respect_retry_after_header=True) session.mount("https://", HTTPAdapter(max_retries=retries)) return session def api_get(session, path, **params): response = session.get(BASE_URL + path, params=params, timeout=(10, 60), allow_redirects=False) if 300 <= response.status_code < 400: raise ValueError("Unexpected API redirect; check the documented API URL.") response.raise_for_status() return response.json() def iso_date(value): if not isinstance(value, str) or not re.fullmatch(r"\d{4}-\d{2}-\d{2}", value): raise ValueError("Dates must use YYYY-MM-DD.") return date.fromisoformat(value).isoformat() def short_interest_frame(payload, ticker): if not isinstance(payload, list) or not payload: raise ValueError("No history returned. Check ticker, dates and access; this does not mean zero short interest.") rows = [] for item in payload: if not isinstance(item, dict) or not set(FIELDS).issubset(item): raise ValueError("Unexpected short-interest response fields.") row = {"settlementDate": iso_date(item["settlementDate"])} for field in FIELDS[1:]: value = item[field] if value is None: row[field] = None continue if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value < 0: raise ValueError(f"Invalid non-negative numeric value for {field}.") if field != "shortInterestPercentage" and value != int(value): raise ValueError(f"Expected whole shares for {field}.") row[field] = value rows.append(row) frame = pd.DataFrame(rows).sort_values("settlementDate").reset_index(drop=True) if frame["settlementDate"].duplicated().any(): raise ValueError("Duplicate settlement dates returned; inspect before comparing observations.") for field in ["sharesShorted", "sharesFloat"]: frame[field] = frame[field].astype("Int64") frame["shortInterestPercentage"] = frame["shortInterestPercentage"].astype("Float64") if frame["sharesShorted"].isna().all(): raise ValueError("No reported shares-shorted values returned.") frame.insert(0, "ticker", ticker) # Compare adjacent returned observations, never invented daily values. frame["sharesShortedChange"] = frame["sharesShorted"].diff() frame["shortInterestChangePp"] = frame["shortInterestPercentage"].diff().round(8) return frame def save_chart(frame, destination): import matplotlib matplotlib.use("Agg") # Works in terminals, notebooks and headless environments. import matplotlib.pyplot as plt import matplotlib.dates as mdates dates = pd.to_datetime(frame["settlementDate"]) fig, axes = plt.subplots(2, 1, figsize=(10, 6), sharex=True, layout="constrained") for axis, column, divisor, label, color in [ (axes[0], "sharesShorted", 1_000_000, "Shares shorted (millions)", "#3576b9"), (axes[1], "shortInterestPercentage", 1, "Short interest (% of float)", "#17846d"), ]: values = frame[column].to_numpy(dtype=float, na_value=float("nan")) / divisor axis.plot(dates, values, marker="o", markersize=3, linewidth=1.4, color=color) axis.set_ylabel(label) axis.grid(axis="y", alpha=0.2) axis.spines[["top", "right"]].set_visible(False) if frame[column].isna().all(): axis.text(0.5, 0.5, "Not returned for these observations", transform=axis.transAxes, ha="center") locator = mdates.AutoDateLocator(minticks=3, maxticks=8) axes[1].xaxis.set_major_locator(locator) axes[1].xaxis.set_major_formatter(mdates.ConciseDateFormatter(locator)) axes[1].set_xlabel("Settlement date ยท lines connect reported observations, not daily positions") fig.suptitle(f"{frame['ticker'].iloc[0]} reported short-interest history") try: with Path(destination).open("xb") as output: fig.savefig(output, format="png", dpi=160) finally: plt.close(fig) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--ticker", default="PLTR") parser.add_argument("--from", dest="from_date", help="Inclusive settlement date, YYYY-MM-DD") parser.add_argument("--to", dest="to_date", help="Inclusive settlement date, YYYY-MM-DD") parser.add_argument("--list-companies", action="store_true") parser.add_argument("--output-dir", default="pltr-short-interest", help="New directory; never overwrites existing work") args = parser.parse_args() try: ticker = args.ticker.strip().upper() if not re.fullmatch(r"[A-Z0-9][A-Z0-9.-]{0,19}", ticker): raise ValueError("Use a stock ticker containing letters, digits, periods or hyphens.") params = {"order": "asc"} for name, value in [("from", args.from_date), ("to", args.to_date)]: if value: params[name] = iso_date(value) if args.from_date and args.to_date and args.from_date > args.to_date: raise ValueError("--from must be on or before --to.") destination = Path(args.output_dir) if not args.list_companies and destination.exists(): raise ValueError("Output directory already exists. Choose a new --output-dir.") with make_session() as session: if args.list_companies: companies = api_get(session, "/companies") if not isinstance(companies, list): raise ValueError("Unexpected company-list response.") for company in companies: if company.get("hasShortInterest"): print(f"{company['ticker']}: {company['name']}") return payload = api_get(session, f"/companies/{ticker}/short-interest", **params) frame = short_interest_frame(payload, ticker) if (args.from_date and frame["settlementDate"].min() < args.from_date) or (args.to_date and frame["settlementDate"].max() > args.to_date): raise ValueError("API returned dates outside the requested range.") destination.mkdir(parents=True, exist_ok=False) stem = ticker.lower() frame.to_csv(destination / f"{stem}-short-interest.csv", index=False, na_rep="", mode="x") save_chart(frame, destination / f"{stem}-short-interest.png") with (destination / "readme.json").open("x", encoding="utf-8") as output: json.dump({"ticker": ticker, "endpoint": f"/v1/companies/{ticker}/short-interest", "parameters": params, "retrievedAtUtc": datetime.now(timezone.utc).isoformat(), "observations": len(frame), "firstSettlementDate": frame["settlementDate"].min(), "lastSettlementDate": frame["settlementDate"].max(), "notes": "Retrieval time is not publication time. Missing values remain blank. Changes compare adjacent returned records; pp means percentage points."}, output, indent=2) print(f"Saved {len(frame)} observations, chart and retrieval notes in {destination}.") except requests.HTTPError as error: parser.exit(1, f"API HTTP {error.response.status_code}. Check key, access and quota in the API dashboard.\n") except requests.RequestException: parser.exit(1, "API connection failed after retries. Check your connection and try again.\n") except (ValueError, OSError, OverflowError) as error: parser.exit(1, f"{error}\n") if __name__ == "__main__": main()