Python 脚本监控亿万富翁更改股票投资组合

Source: Dev.to

每个季度，管理超过 $100 M 的对冲基金和机构投资者必须通过 13F 报告 向 SEC 披露其股票持仓。这是公开数据——它揭示了沃伦·巴菲特、雷·达里奥、迈克尔·伯里以及成千上万其他大玩家的买卖情况。

在本教程中，我们将编写一个 Python 脚本来监控这些报告并检测投资组合的变化——新持仓、退出以及显著的增减。

13F 备案的工作原理

SEC 要求机构投资经理每季度提交 Form 13F。每份备案列出基金持有的所有股票头寸，包括股份数量和市值。通过比较连续的备案，你可以准确看到哪些发生了变化。

问题在哪？SEC 的 EDGAR 系统众所周知难以使用——格式不一致、XML 解析、速率限制以及 CIK 编号查询。我们将通过使用一个为我们处理这些的 API 来跳过所有这些麻烦。

Setup

我们将使用 RapidAPI 上的 SEC EDGAR Financial Data API，它提供对 10,000 多家公司 13F 持仓的干净、结构化访问。

1. 订阅该 API 以获取您的密钥，然后安装所需的包：

pip install requests

第一步：查找投资者的 CIK 编号

每个 SEC 备案人都有一个 CIK（Central Index Key，中心索引键）。我们来搜索一个：

import requests

RAPIDAPI_KEY = "YOUR_RAPIDAPI_KEY"
BASE_URL = "https://sec-edgar-financial-data-api.p.rapidapi.com"
HEADERS = {
    "x-rapidapi-host": "sec-edgar-financial-data-api.p.rapidapi.com",
    "x-rapidapi-key": RAPIDAPI_KEY,
}

def search_company(query):
    resp = requests.get(
        f"{BASE_URL}/companies/search",
        params={"query": query},
        headers=HEADERS,
    )
    resp.raise_for_status()
    return resp.json()

results = search_company("Berkshire Hathaway")
for company in results.get("companies", [])[:5]:
    print(f"{company['name']} — CIK: {company['cik']}")

这将为您提供任何备案人的 CIK。Berkshire Hathaway 的 CIK 为 0001067983。

第2步：获取 13F 持仓

现在获取实际的投资组合持仓：

def get_holdings(cik):
    resp = requests.get(
        f"{BASE_URL}/holdings/13f/{cik}",
        headers=HEADERS,
    )
    resp.raise_for_status()
    return resp.json()

holdings = get_holdings("0001067983")  # Berkshire Hathaway

# Show top 10 positions by value
positions = holdings.get("holdings", [])
positions_sorted = sorted(positions, key=lambda x: x.get("value", 0), reverse=True)

print(f"Total positions: {len(positions)}")
print("\nTop 10 Holdings:")
for pos in positions_sorted[:10]:
    name = pos.get("name", "Unknown")
    value = pos.get("value", 0)
    shares = pos.get("shares", 0)
    print(f"  {name}: ${value:,.0f} ({shares:,.0f} shares)")

第三步：检测投资组合变动

真实价值来自于比较两个季度。下面是一个函数，用于比较两组持仓并突出显示发生了哪些变化：

def detect_changes(current_holdings, previous_holdings):
    current = {h["cusip"]: h for h in current_holdings}
    previous = {h["cusip"]: h for h in previous_holdings}

    changes = {
        "new_positions": [],
        "exited_positions": [],
        "increased": [],
        "decreased": [],
    }

    # New positions (in current but not previous)
    for cusip in current:
        if cusip not in previous:
            changes["new_positions"].append(current[cusip])

    # Exited positions (in previous but not current)
    for cusip in previous:
        if cusip not in current:
            changes["exited_positions"].append(previous[cusip])

    # Changed positions
    for cusip in current:
        if cusip in previous:
            curr_shares = current[cusip].get("shares", 0)
            prev_shares = previous[cusip].get("shares", 0)
            if prev_shares == 0:
                continue
            pct_change = ((curr_shares - prev_shares) / prev_shares) * 100

            if pct_change > 5:  # increased by more than 5%
                changes["increased"].append({
                    **current[cusip],
                    "prev_shares": prev_shares,
                    "pct_change": pct_change,
                })
            elif pct_change < -5:  # decreased by more than 5%
                changes["decreased"].append({
                    **current[cusip],
                    "prev_shares": prev_shares,
                    "pct_change": pct_change,
                })

    return changes

# Example usage (assuming `current` and `previous` are already fetched):
# changes = detect_changes(current["holdings"], previous["holdings"])
# print("New positions:", len(changes["new_positions"]))
# print("Exited positions:", len(changes["exited_positions"]))
# print("Increased positions:", len(changes["increased"]))
# print("Decreased positions:", len(changes["decreased"]))

第五步：监控多个基金

跟踪一份投资者观察列表，并一次性检查所有基金：

watchlist = {
    "Berkshire Hathaway": "0001067983",
    "Bridgewater Associates": "0001350694",
    "Citadel Advisors": "0001423053",
    "Renaissance Technologies": "0001037389",
}

def monitor_watchlist(watchlist):
    for name, cik in watchlist.items():
        print(f"\nFetching data for {name}...")
        try:
            data = get_holdings(cik)
            holdings = data.get("holdings", [])
            total_value = sum(h.get("value", 0) for h in holdings)
            print(f"  Positions: {len(holdings)}")
            print(f"  Total reported value: ${total_value:,.0f}")
        except Exception as e:
            print(f"  Error: {e}")

monitor_watchlist(watchlist)

Ideas to Extend This

• Scheduled monitoring — 将其作为 cron 任务运行，并在出现新备案时发送电子邮件或 Slack 警报
• Consensus picks — 找出多个顶级基金同时买入的股票
• Historical tracking — 将季度快照存储在数据库中，并随时间绘制仓位规模图表
• Sector analysis — 按行业对持仓进行分类，以观察大资金是否在向科技、能源、医疗等行业轮动

重要说明

13F 报告在每个季度结束后约 45 天才公布。这不是实时数据——它是一个季度快照。但它仍然极具价值，可帮助了解全球最聪明的资金在其投资组合中做了什么。

总结

使用 SEC EDGAR Financial Data API 和一点 Python，您可以以编程方式跟踪任何提交 13F 报告的机构投资者的投资组合变化。再也不需要手动浏览 SEC 网站或解析原始 XML。

在 RapidAPI 订阅 并开始跟踪您关心的基金。

您首先想跟踪哪个投资者？ 在评论中告诉我。