edgartools/investor_holdings.py

#!/usr/bin/env python3

import os
import json
import hashlib

# Check if virtual environment is activated
if not os.environ.get('VIRTUAL_ENV'):
    print("Virtual environment is not activated.")
    print("To activate: . venv/bin/activate")
    print("Then run: python investor_holdings.py")
    exit(1)

import pandas as pd
from edgar import Company, set_identity, set_local_storage_path, use_local_storage

# Step-by-step implementation of local storage from https://www.edgartools.io/speedup-jobs-with-local-storage/
# Step-by-step implementation of local storage from https://www.edgartools.io/speedup-jobs-with-local-storage/
# Enable local storage for caching filings
LOCAL_STORAGE_PATH = os.path.abspath("./edgar_cache")
os.makedirs(LOCAL_STORAGE_PATH, exist_ok=True)
use_local_storage(LOCAL_STORAGE_PATH)

# Set your identity (required by SEC) - after local storage
set_identity("your.email@example.com")

# Cache directory
CACHE_DIR = 'cache'
os.makedirs(CACHE_DIR, exist_ok=True)

def get_cache_key(name, ticker, inv_type, num_filings):
    key = f"{name}_{ticker}_{inv_type}_{num_filings}"
    return hashlib.md5(key.encode()).hexdigest() + '.json'

def load_cache(cache_file):
    if os.path.exists(cache_file):
        with open(cache_file, 'r') as f:
            return json.load(f)
    return None

def save_cache(cache_file, data):
    with open(cache_file, 'w') as f:
        json.dump(data, f, indent=2)

def get_holdings(ticker_or_cik, name, num_filings=1):
    cache_file = os.path.join(CACHE_DIR, get_cache_key(name, ticker_or_cik, '13f', num_filings))
    cached_data = load_cache(cache_file)
    if cached_data:
        print(f"\n{name} - 13F Holdings (from cache):")
        for entry in cached_data:
            print(f"\n{name} ({entry['company_name']}) - 13F Holdings as of {entry['report_period']}:")
            print("=" * 60)
            for holding in entry['top_holdings']:
                print(holding)
        return cached_data

    try:
        company = Company(ticker_or_cik)
        if company.not_found:
            print(f"{name}: Company not found")
            return None

        # Get last num_filings 13F filings
        filings_13f = company.get_filings(form="13F-HR").head(num_filings)
        if not filings_13f:
            print(f"{name}: No 13F filings found")
            return None

        data_to_cache = []
        for i, filing_13f in enumerate(filings_13f):
            # Get holdings data
            holdings_obj = filing_13f.obj()
            if not holdings_obj.has_infotable():
                print(f"{name} - Filing {i+1}: No holdings data")
                continue
            holdings_df = holdings_obj.infotable

            report_period = holdings_obj.report_period
            print(f"\n{name} ({company.name}) - 13F Holdings as of {report_period}:")
            print("=" * 60)

            holdings_list = []

            # Ensure SharesPrnAmount is numeric
            holdings_df['SharesPrnAmount'] = pd.to_numeric(holdings_df['SharesPrnAmount'], errors='coerce')

            # Separate long and short positions
            # Puts are bearish (short exposure), Calls are bullish (long exposure)
            put_call_col = 'PutCall' if 'PutCall' in holdings_df.columns else None
            shares_type_col = 'SharesPrnType' if 'SharesPrnType' in holdings_df.columns else None

            if put_call_col:
                long_positions = holdings_df[(holdings_df[put_call_col] != 'Put') & (holdings_df.get(shares_type_col, '') != 'SH')]
                short_positions = holdings_df[(holdings_df[put_call_col] == 'Put') | (holdings_df.get(shares_type_col, '') == 'SH')]
            else:
                long_positions = holdings_df[holdings_df.get(shares_type_col, '') != 'SH']
                short_positions = holdings_df[holdings_df.get(shares_type_col, '') == 'SH']

            # Display top 10 long holdings by value
            if len(long_positions) > 0:
                top_long = long_positions.nlargest(10, 'Value')
                print("\nTop Long Positions:")
                for idx, row in top_long.iterrows():
                    shares = int(row['SharesPrnAmount']) if pd.notna(row['SharesPrnAmount']) else 'N/A'
                    put_call = row.get('PutCall', '')
                    title = row.get('TitleOfClass', '')
                    position_type = put_call if put_call else (title if title else 'Shares')
                    line = f"{row['Issuer']}: ${row['Value']/1e6:.1f}M ({shares} shares, {position_type})"
                    print(line)
                    holdings_list.append(f"Long: {line}")

            # Display top 10 short positions by absolute value
            if len(short_positions) > 0:
                short_positions = short_positions.copy()
                short_positions['AbsValue'] = short_positions['Value'].abs()
                top_short = short_positions.nlargest(10, 'AbsValue')
                print("\nTop Short Positions:")
                for idx, row in top_short.iterrows():
                    shares = int(row['SharesPrnAmount']) if pd.notna(row['SharesPrnAmount']) else 'N/A'
                    line = f"{row['Issuer']}: ${row['Value']/1e6:.1f}M ({shares} shares)"
                    print(line)
                    holdings_list.append(f"Short: {line}")
            else:
                # If no 'SH' type, check for negative shares as shorts
                potential_shorts = holdings_df[holdings_df['SharesPrnAmount'] < 0]
                if len(potential_shorts) > 0:
                    potential_shorts = potential_shorts.copy()
                    potential_shorts['AbsValue'] = potential_shorts['Value'].abs()
                    top_short = potential_shorts.nlargest(10, 'AbsValue')
                    print("\nTop Short Positions (inferred from negative shares):")
                    for idx, row in top_short.iterrows():
                        shares = int(row['SharesPrnAmount']) if pd.notna(row['SharesPrnAmount']) else 'N/A'
                        line = f"{row['Issuer']}: ${row['Value']/1e6:.1f}M ({shares} shares)"
                        print(line)
                        holdings_list.append(f"Short: {line}")
                else:
                    print("\nNo short positions reported.")

            data_to_cache.append({
                'report_period': str(report_period),
                'company_name': company.name,
                'top_holdings': holdings_list
            })

        save_cache(cache_file, data_to_cache)
        return data_to_cache

    except Exception as e:
        print(f"Error getting holdings for {name}: {e}")
        return None

def get_insider_transactions(ticker, insider_name=None, num_filings=5):
    cache_file = os.path.join(CACHE_DIR, get_cache_key(insider_name or 'all', ticker, 'insider', num_filings))
    cached_data = load_cache(cache_file)
    if cached_data:
        print(f"\nAll insiders found in recent Form 4 filings for {cached_data['company_name']} ({ticker}) (from cache):")
        for insider in cached_data['all_insiders']:
            print(f"  - {insider}")

        if insider_name:
            if cached_data['found_transactions']:
                print(f"\nInsider Transactions for {insider_name} at {cached_data['company_name']} ({ticker}) (from cache):")
                print("=" * 80)
                for tx in cached_data['found_transactions'][:10]:
                    print(f"Date: {tx['date']}, Code: {tx['code']}, Shares: {tx['shares']}, Price: ${tx['price']}, Amount: ${tx['amount']}")
            else:
                print(f"No transactions found for {insider_name} at {ticker}")
        return cached_data['found_transactions']

    try:
        company = Company(ticker)
        if company.not_found:
            print(f"Company {ticker}: Not found")
            return None

        # Get recent Form 4 filings
        filings_4 = company.get_filings(form="4").head(num_filings)
        if not filings_4:
            print(f"No Form 4 filings found for {ticker}")
            return None

        all_insiders = set()
        found_transactions = []
        for filing in filings_4:
            form4 = filing.obj()
            if hasattr(form4, 'transactions') and form4.transactions is not None:
                if hasattr(form4, 'reporting_owner') and form4.reporting_owner:
                    owner_name = form4.reporting_owner.name
                    all_insiders.add(owner_name)
                    if insider_name and insider_name.lower() in owner_name.lower():
                        for transaction in form4.transactions:
                            found_transactions.append({
                                'date': str(filing.filing_date),
                                'owner': owner_name,
                                'security': transaction.security_title,
                                'code': transaction.transaction_code,
                                'shares': str(transaction.transaction_shares),
                                'price': str(transaction.transaction_price),
                                'amount': str(transaction.transaction_amount)
                            })

        print(f"\nAll insiders found in recent Form 4 filings for {company.name} ({ticker}):")
        for insider in sorted(all_insiders):
            print(f"  - {insider}")

        if insider_name:
            if found_transactions:
                print(f"\nInsider Transactions for {insider_name} at {company.name} ({ticker}):")
                print("=" * 80)
                for tx in found_transactions[:10]:  # Limit to 10 most recent
                    print(f"Date: {tx['date']}, Code: {tx['code']}, Shares: {tx['shares']}, Price: ${tx['price']}, Amount: ${tx['amount']}")
            else:
                print(f"No transactions found for {insider_name} at {ticker}")

        data_to_cache = {
            'company_name': company.name,
            'all_insiders': sorted(list(all_insiders)),
            'found_transactions': found_transactions
        }
        save_cache(cache_file, data_to_cache)
        return found_transactions

    except Exception as e:
        print(f"Error getting insider transactions: {e}")
        return None

# Load investors from external file
with open('investors.json', 'r') as f:
    config = json.load(f)

investors = config['investors']

def select_investors():
    print("\nAvailable Investors:")
    for i, inv in enumerate(investors, 1):
        print(f"{i}. {inv['name']} ({inv['type'].upper()})")
    print("0. All investors")
    print("q. Quit")

    while True:
        choice = input("\nSelect investor(s) (comma-separated numbers, 'all', or 'q' to quit): ").strip().lower()
        if choice == 'q':
            return []
        elif choice == 'all' or choice == '0':
            return investors
        else:
            try:
                indices = [int(x.strip()) - 1 for x in choice.split(',')]
                selected = [investors[i] for i in indices if 0 <= i < len(investors)]
                if selected:
                    return selected
                else:
                    print("Invalid selection. Try again.")
            except ValueError:
                print("Invalid input. Enter numbers separated by commas, 'all', or 'q'.")

if __name__ == "__main__":
    import sys
    if len(sys.argv) > 1:
        try:
            idx = int(sys.argv[1]) - 1
            if 0 <= idx < len(investors):
                selected_investors = [investors[idx]]
            else:
                print("Invalid index")
                selected_investors = []
        except ValueError:
            print("Invalid argument")
            selected_investors = []
    else:
        selected_investors = select_investors()
        if not selected_investors:
            print("Exiting.")
            exit(0)

    for investor in selected_investors:
        name = investor['name']
        ticker = investor['ticker']
        inv_type = investor['type']
        num_filings = investor.get('filings', 1)

        if inv_type == '13f':
            get_holdings(ticker, name, num_filings)
        elif inv_type == 'insider':
            get_insider_transactions(ticker, name, num_filings)

    print("\n" + "="*80)
    print("Processing complete.")