track_changes.py

"""
Example: Track Portfolio Changes Across Quarters

Compares an institution's current 13F holdings against a snapshot you
provide (or a previous API call) to highlight new positions, closed
positions, and significant size changes.

This example uses Citadel (CIK 1423053) but works with any CIK.

Get your free API key:
https://rapidapi.com/dapdev-dapdev-default/api/sec-edgar-financial-data-api
"""

import os
from edgar_client import EdgarAPI, Portfolio

API_KEY = os.environ.get("RAPIDAPI_KEY", "YOUR_API_KEY_HERE")

# Threshold: flag a position if its value changed by more than this fraction
CHANGE_THRESHOLD = 0.20


def holdings_by_cusip(portfolio: Portfolio) -> dict[str, int]:
    """Return a dict mapping CUSIP -> value_usd for quick lookup."""
    return {h.cusip: h.value_usd for h in portfolio.holdings}


def holdings_name(portfolio: Portfolio) -> dict[str, str]:
    """Return a dict mapping CUSIP -> issuer name."""
    return {h.cusip: h.name_of_issuer for h in portfolio.holdings}


def compare(old: Portfolio, new: Portfolio) -> None:
    """Print a diff between two Portfolio snapshots."""
    old_map = holdings_by_cusip(old)
    new_map = holdings_by_cusip(new)
    names = {**holdings_name(old), **holdings_name(new)}

    all_cusips = set(old_map) | set(new_map)
    new_positions = []
    closed_positions = []
    increased = []
    decreased = []

    for cusip in all_cusips:
        old_val = old_map.get(cusip, 0)
        new_val = new_map.get(cusip, 0)
        name = names.get(cusip, cusip)

        if old_val == 0 and new_val > 0:
            new_positions.append((name, new_val))
        elif old_val > 0 and new_val == 0:
            closed_positions.append((name, old_val))
        elif old_val > 0 and new_val > 0:
            change = (new_val - old_val) / old_val
            if change >= CHANGE_THRESHOLD:
                increased.append((name, old_val, new_val, change))
            elif change <= -CHANGE_THRESHOLD:
                decreased.append((name, old_val, new_val, change))

    print(f"\nComparing {old.company_name}")
    print(f"  {old.report_date}  ->  {new.report_date}\n")

    if new_positions:
        print(f"NEW POSITIONS ({len(new_positions)})")
        print("-" * 50)
        for name, val in sorted(new_positions, key=lambda x: x[1], reverse=True):
            print(f"  + {name[:40]:<40}  ${val/1e6:,.1f}M")

    if closed_positions:
        print(f"\nCLOSED POSITIONS ({len(closed_positions)})")
        print("-" * 50)
        for name, val in sorted(closed_positions, key=lambda x: x[1], reverse=True):
            print(f"  - {name[:40]:<40}  was ${val/1e6:,.1f}M")

    if increased:
        print(f"\nSIGNIFICANTLY INCREASED (>={CHANGE_THRESHOLD*100:.0f}%)")
        print("-" * 50)
        for name, old_v, new_v, chg in sorted(increased, key=lambda x: x[3], reverse=True):
            print(f"  ^ {name[:35]:<35}  ${old_v/1e6:,.1f}M -> ${new_v/1e6:,.1f}M  ({chg*100:+.1f}%)")

    if decreased:
        print(f"\nSIGNIFICANTLY DECREASED (>={CHANGE_THRESHOLD*100:.0f}%)")
        print("-" * 50)
        for name, old_v, new_v, chg in sorted(decreased, key=lambda x: x[3]):
            print(f"  v {name[:35]:<35}  ${old_v/1e6:,.1f}M -> ${new_v/1e6:,.1f}M  ({chg*100:+.1f}%)")

    total_old = old.total_value_usd / 1e9
    total_new = new.total_value_usd / 1e9
    total_chg = (new.total_value_usd - old.total_value_usd) / old.total_value_usd * 100
    print(f"\nPortfolio total: ${total_old:.2f}B -> ${total_new:.2f}B  ({total_chg:+.1f}%)")


def main():
    with EdgarAPI(api_key=API_KEY) as api:
        print("Fetching Citadel portfolio (current)...")
        current = api.holdings(EdgarAPI.CITADEL, limit=200)

        # To compare across real quarters you would store the previous
        # snapshot and reload it here. For demonstration we simulate a
        # "previous" snapshot by applying a small artificial delta.
        print("Building simulated previous snapshot for demonstration...\n")

        import copy
        import random

        random.seed(42)
        previous_holdings = copy.deepcopy(current.holdings)

        # Simulate: remove two positions, inflate three, shrink two
        dropped = previous_holdings[:2]
        for h in previous_holdings[2:5]:
            h.value_usd = int(h.value_usd * 0.60)
            h.shares = int(h.shares * 0.60)
        for h in previous_holdings[5:8]:
            h.value_usd = int(h.value_usd * 1.50)
            h.shares = int(h.shares * 1.50)
        previous_holdings = previous_holdings[2:]  # drop first two (now "new")

        from edgar_client import Portfolio, Holding
        previous = Portfolio(
            cik=current.cik,
            company_name=current.company_name,
            report_date="2024-09-30",  # simulated prior quarter
            total_value_usd=sum(h.value_usd for h in previous_holdings),
            total_holdings=len(previous_holdings),
            holdings=previous_holdings,
        )

        compare(previous, current)


if __name__ == "__main__":
    main()