배당수익률 배당성장률 높은 미국주식 찾기! S&P 500개 종목 3분컷 자동 수집

!pip install yfinance openpyxl requests lxml beautifulsoup4 -q

import pandas as pd
import yfinance as yf
import requests
from io import StringIO
from pathlib import Path

# =========================================================
# [입력란] 여기만 바꾸면 됨
# =========================================================
AS_OF_DATE = "2025-12-31"      # 조회 기준일
LOOKBACK_YEARS = 5             # 배당성장률 비교 연수
USE_ADJ_CLOSE = False          # False=Close, True=Adj Close
SORT_BY = "combined"           # "combined", "yield", "growth", None
SAVE_EXCEL = True
VERBOSE = True

# 코랩 저장 위치
OUTPUT_DIR = "/content"

# 파일명 직접 지정하고 싶으면 입력, 아니면 None
OUTPUT_FILE = None

# 테스트용: 예) 30개만 돌려보기. 전체면 None
LIMIT = None
# LIMIT = 30


def get_sp500_symbols():
    url = "https://en.wikipedia.org/wiki/List_of_S%26P_500_companies"
    headers = {"User-Agent": "Mozilla/5.0"}

    response = requests.get(url, headers=headers, timeout=20)
    response.raise_for_status()

    df = pd.read_html(StringIO(response.text))[0]
    symbols = df["Symbol"].tolist()
    symbols = [symbol.replace(".", "-") for symbol in symbols]
    return symbols


def get_last_available_price(series: pd.Series, target_date: str):
    s = series.loc[:target_date].dropna()
    if s.empty:
        return None, None
    return s.index[-1], float(s.iloc[-1])


def get_ttm_dividend(div_series: pd.Series, as_of_date: str):
    end = pd.Timestamp(as_of_date)
    start = end - pd.Timedelta(days=365)
    s = div_series[(div_series.index > start) & (div_series.index <= end)].dropna()
    return float(s.sum()) if not s.empty else 0.0


def calc_cagr(begin_value: float, end_value: float, years: float):
    if begin_value is None or end_value is None:
        return None
    if begin_value <= 0 or end_value <= 0:
        return None
    return (end_value / begin_value) ** (1 / years) - 1


def add_ranking_columns(df: pd.DataFrame, lookback_years: int):
    growth_col = f"Dividend_Growth_{lookback_years}Y_CAGR_%"
    yield_col = "Current_Dividend_Yield_%"

    result = df.copy()

    result["Dividend_Yield_Rank"] = result[yield_col].rank(
        ascending=False, method="min", na_option="bottom"
    )
    result["Dividend_Growth_Rank"] = result[growth_col].rank(
        ascending=False, method="min", na_option="bottom"
    )
    result["Combined_Score"] = result[["Dividend_Yield_Rank", "Dividend_Growth_Rank"]].sum(axis=1)
    result["Combined_Rank"] = result["Combined_Score"].rank(
        ascending=True, method="min", na_option="bottom"
    )

    for col in ["Dividend_Yield_Rank", "Dividend_Growth_Rank", "Combined_Rank"]:
        result[col] = result[col].astype("Int64")

    return result


def run_sp500_dividend_screen(
    as_of_date: str,
    lookback_years: int = 5,
    use_adj_close: bool = False,
    save_excel: bool = True,
    output_dir: str = "/content",
    output_file: str = None,
    sort_by: str = "combined",
    verbose: bool = True,
    limit: int = None
):
    price_col = "Adj Close" if use_adj_close else "Close"

    as_of_ts = pd.Timestamp(as_of_date)
    compare_ts = as_of_ts - pd.DateOffset(years=lookback_years)

    compare_date = compare_ts.strftime("%Y-%m-%d")
    start_lookback = (compare_ts - pd.Timedelta(days=370)).strftime("%Y-%m-%d")
    history_end = (as_of_ts + pd.Timedelta(days=2)).strftime("%Y-%m-%d")

    if verbose:
        print("=" * 60)
        print(f"조회 기준일     : {as_of_date}")
        print(f"비교 기준일     : {compare_date}")
        print(f"배당성장률 기간 : {lookback_years}년")
        print(f"주가 기준 컬럼  : {price_col}")
        print(f"조회 시작일     : {start_lookback}")
        print("=" * 60)

    tickers = get_sp500_symbols()
    if limit is not None:
        tickers = tickers[:limit]

    if verbose:
        print(f"S&P500 대상 티커 수: {len(tickers)}")

    results = []

    for i, ticker in enumerate(tickers, start=1):
        try:
            tk = yf.Ticker(ticker)

            hist = tk.history(
                start=start_lookback,
                end=history_end,
                auto_adjust=False
            )

            dividends = tk.dividends.copy()
            if not dividends.empty:
                dividends.index = pd.to_datetime(dividends.index).tz_localize(None)

            note_list = []

            if hist.empty or price_col not in hist.columns:
                asof_px_date, asof_px = None, None
                note_list.append("주가 데이터 없음")
            else:
                price_series = hist[price_col].copy()
                price_series.index = pd.to_datetime(price_series.index).tz_localize(None)
                asof_px_date, asof_px = get_last_available_price(price_series, as_of_date)

            if dividends.empty:
                ttm_div_current = 0.0
                ttm_div_past = 0.0
                note_list.append("배당 데이터 없음")
            else:
                ttm_div_current = get_ttm_dividend(dividends, as_of_date)
                ttm_div_past = get_ttm_dividend(dividends, compare_date)

                if ttm_div_current == 0:
                    note_list.append(f"{as_of_date} 기준 최근 1년 배당금 없음")
                if ttm_div_past == 0:
                    note_list.append(f"{compare_date} 기준 최근 1년 배당금 없음")

            if asof_px is None or asof_px == 0:
                dividend_yield = None
                if "주가 데이터 없음" not in note_list:
                    note_list.append("기준일 주가 없음")
            else:
                dividend_yield = ttm_div_current / asof_px

            div_growth_cagr = calc_cagr(
                begin_value=ttm_div_past,
                end_value=ttm_div_current,
                years=lookback_years
            )

            results.append({
                "Ticker": ticker,
                "AsOfDate": as_of_date,
                "CompareDate": compare_date,
                "LookbackYears": lookback_years,
                "PriceDateUsed": asof_px_date.strftime("%Y-%m-%d") if asof_px_date is not None else None,
                "PriceUsed": round(asof_px, 4) if asof_px is not None else None,
                f"TTM_Dividend_{compare_date}": round(ttm_div_past, 6),
                f"TTM_Dividend_{as_of_date}": round(ttm_div_current, 6),
                f"Dividend_Growth_{lookback_years}Y_CAGR_%": round(div_growth_cagr * 100, 2) if div_growth_cagr is not None else None,
                "Current_Dividend_Yield_%": round(dividend_yield * 100, 2) if dividend_yield is not None else None,
                "Note": "; ".join(note_list)
            })

            if verbose and (i % 25 == 0 or i == len(tickers)):
                print(f"[진행] {i}/{len(tickers)} 완료")

        except Exception as e:
            results.append({
                "Ticker": ticker,
                "AsOfDate": as_of_date,
                "CompareDate": compare_date,
                "LookbackYears": lookback_years,
                "PriceDateUsed": None,
                "PriceUsed": None,
                f"TTM_Dividend_{compare_date}": None,
                f"TTM_Dividend_{as_of_date}": None,
                f"Dividend_Growth_{lookback_years}Y_CAGR_%": None,
                "Current_Dividend_Yield_%": None,
                "Note": f"오류: {str(e)}"
            })

            if verbose:
                print(f"[오류] {ticker} - {e}")

    df = pd.DataFrame(results)
    df = add_ranking_columns(df, lookback_years=lookback_years)

    growth_col = f"Dividend_Growth_{lookback_years}Y_CAGR_%"

    if sort_by == "combined":
        df = df.sort_values(
            by=["Combined_Rank", "Dividend_Growth_Rank", "Dividend_Yield_Rank"],
            ascending=[True, True, True],
            na_position="last"
        ).reset_index(drop=True)

    elif sort_by == "yield":
        df = df.sort_values(
            by=["Dividend_Yield_Rank", growth_col],
            ascending=[True, False],
            na_position="last"
        ).reset_index(drop=True)

    elif sort_by == "growth":
        df = df.sort_values(
            by=["Dividend_Growth_Rank", "Current_Dividend_Yield_%"],
            ascending=[True, False],
            na_position="last"
        ).reset_index(drop=True)

    saved_path = None

    if save_excel:
        output_dir_path = Path(output_dir)
        output_dir_path.mkdir(parents=True, exist_ok=True)

        if output_file is None:
            safe_asof = as_of_date.replace("-", "")
            output_file = f"sp500_dividend_growth_yield_rank_asof_{safe_asof}_{lookback_years}y.xlsx"

        saved_path = output_dir_path / output_file
        df.to_excel(saved_path, index=False, engine="openpyxl")

        print("\n엑셀 저장 완료")
        print(f"파일명   : {saved_path.name}")
        print(f"저장경로 : {saved_path}")

    return df, saved_path


# =========================================================
# 실행
# =========================================================
df, saved_path = run_sp500_dividend_screen(
    as_of_date=AS_OF_DATE,
    lookback_years=LOOKBACK_YEARS,
    use_adj_close=USE_ADJ_CLOSE,
    save_excel=SAVE_EXCEL,
    output_dir=OUTPUT_DIR,
    output_file=OUTPUT_FILE,
    sort_by=SORT_BY,
    verbose=VERBOSE,
    limit=LIMIT
)

print("\n상위 10개 미리보기")
print(df.head(10))

# =========================================================
# 코랩 자동 다운로드
# =========================================================
if saved_path is not None:
    from google.colab import files
    files.download(str(saved_path))