Python 스크립트로 억만장자들의 주식 포트폴리오 변동을 모니터링

Source: Dev.to

매 분기마다 1억 달러 이상을 관리하는 헤지펀드와 기관 투자자는 13F 보고서를 통해 SEC에 주식 보유 현황을 공개해야 합니다. 이는 공개 데이터이며, 워런 버핏, 레이 달리오, 마이클 버리 및 수천 명의 다른 대형 투자자들이 무엇을 사고 팔고 있는지 정확히 보여줍니다.

이 튜토리얼에서는 이러한 보고서를 모니터링하고 포트폴리오 변화를 감지하는 파이썬 스크립트를 만들 것입니다 — 새로운 포지션, 청산, 그리고 큰 폭의 증가 또는 감소를 탐지합니다.

13F 제출이 작동하는 방식

SEC는 기관 투자 매니저가 분기마다 Form 13F를 제출하도록 요구합니다. 각 제출서에는 펀드가 보유하고 있는 모든 주식 포지션이 나열되며, 주식 수와 시장 가치가 포함됩니다. 연속적인 제출서를 비교하면 정확히 무엇이 변했는지 확인할 수 있습니다.

문제는? SEC의 EDGAR 시스템은 형식이 일관되지 않고, XML 파싱, 속도 제한, CIK 번호 조회 등으로 작업하기가 매우 어렵습니다. 우리는 이를 대신 처리해 주는 API를 사용함으로써 그 모든 번거로움을 건너뛰겠습니다.

설정

우리는 RapidAPI에서 제공하는 SEC EDGAR Financial Data API 를 사용합니다. 이 API는 10,000개 이상의 기업에 대한 13F 보유 정보를 깔끔하고 구조화된 형태로 제공합니다.

1. API에 구독하여 키를 발급받은 후, 필요한 패키지를 설치합니다:

pip install requests

Step 1: 투자자의 CIK 번호 찾기

모든 SEC 제출자는 CIK(중앙 인덱스 키)를 가지고 있습니다. 하나를 검색해 보겠습니다:

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)")

단계 3: 포트폴리오 변동 감지

두 분기를 비교할 때 실제 가치가 드러납니다. 다음은 두 보유 목록을 비교하여 변동 사항을 강조하는 함수입니다:

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"]))

단계 5: 여러 펀드 모니터링

투자자 감시 목록을 추적하고 한 번에 모두 확인합니다:

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)

이 내용을 확장하는 아이디어

• Scheduled monitoring — 새로운 파일이 나타날 때마다 크론 작업으로 실행하고 이메일 또는 Slack 알림을 보냅니다
• Consensus picks — 여러 상위 펀드가 동시에 매수하고 있는 주식을 찾습니다
• Historical tracking — 데이터베이스에 분기별 스냅샷을 저장하고 시간에 따라 포지션 규모를 차트로 표시합니다
• Sector analysis — 보유 종목을 섹터별로 분류하여 대규모 자금이 기술, 에너지, 헬스케어 등으로 회전하고 있는지 확인합니다

중요 사항

13F 보고서는 각 분기가 끝난 후 약 45일 정도 지연되어 보고됩니다. 이는 실시간 데이터가 아니라 분기별 스냅샷입니다. 하지만 전 세계에서 가장 현명한 자금이 포트폴리오를 어떻게 운용하고 있는지 이해하는 데 여전히 매우 가치가 있습니다.

마무리

With the SEC EDGAR Financial Data API and a bit of Python, you can programmatically track portfolio changes for any institutional investor filing 13F reports. No more manually navigating the SEC website or parsing raw XML.

Subscribe on RapidAPI and start tracking the funds you care about.

Which investor would you track first? Let me know in the comments.