# PythonLand 投資×Python シリーズ 第10回 # 「配当利回りスクリーナーを作る（pandas × CSV）」 # https://pythonland.tech/dividend-yield-screener.html # # 記事2〜8章のコードをつないだ完成版です（2026-09-04 時点）。 # 架空銘柄のCSVを読み、配当利回り・PER・PBR を計算して、 # 指定した条件に合う行を機械的に抽出します。 # # ⚠️ これは投資判断をするプログラムではありません。条件に合う行を抜き出すだけの道具で、 # 抽出された銘柄が良い投資対象であることを意味しません。 # ⚠️ しきい値（--min-yield / --max-per / --max-pbr）に既定値は用意していません。 # 既定値を置くとその数字が推奨に見えるためです。どの値で絞るかは利用者ご自身の判断です。 # ⚠️ 通信について: 既定の動作（CSVの読み込み・絞り込み・書き出し）は完全にローカルで、 # 外部との通信はありません。--fetch を付けたときだけ yfinance 経由で # Yahoo! Finance と通信します。送るのは銘柄コードだけで、CSVも条件も送信しません。 # ⚠️ --init-sample で書き出すサンプルCSVの銘柄コード・社名・業種・株価・配当額・ # 1株利益・1株純資産はすべて架空です。実在の企業とは無関係で、推奨銘柄でも # 運用実績でもありません。 # ⚠️ 出力CSVは「自分がどの条件でどの銘柄を見ているか」の記録です。共有フォルダ・ # クラウド同期・Webサーバーの公開ディレクトリには置かないでください。 # ライセンス: MIT（https://pythonland.tech/ のコードは MIT ライセンスで公開しています） """架空銘柄のCSVを配当利回り・PER・PBR で絞り込む CLI スクリーナー。 使い方: python dividend_screener.py --init-sample # 架空サンプルCSVを書き出す python dividend_screener.py --min-yield 3.5 # 1条件で絞る python dividend_screener.py --min-yield 3.5 --max-pbr 1.0 # 条件を重ねる python dividend_screener.py --min-yield 3.5 --out result.csv # 結果をCSVに書き出す python dividend_screener.py --fetch 7203.T # ここだけ外部と通信する しきい値は必須です（3つのうち少なくとも1つ）。数値はいずれも動作確認用の例で、 推奨値ではありません。 """ import argparse import sys from pathlib import Path import pandas as pd BASE_DIR = Path(__file__).resolve().parent DEFAULT_CSV = BASE_DIR / "sample_stocks.csv" NUMERIC_COLS = ("price", "dividend", "dividend_prev", "eps", "bps") # 架空のサンプルデータ（25行）。実在の企業・銘柄コードとは一切関係がありません。 # 欠損・赤字・債務超過・減配・記念配当を意図的に混ぜてあります。 SAMPLE_CSV = """code,name,sector,price,dividend,dividend_prev,eps,bps,dividend_note SMPL-01,アルファ精機,セクターA,1200,36,34,95,880,普通 SMPL-02,ベータ電装,セクターA,2450,49,45,180,1500,普通 SMPL-03,ガンマ商事,セクターB,780,31.2,30,52,610,普通 SMPL-04,デルタ製薬,セクターC,3100,31,31,140,1900,普通 SMPL-05,イプシロン物流,セクターB,640,22.4,21,48,520,普通 SMPL-06,ゼータ食品,セクターD,1580,47.4,45,98,1210,普通 SMPL-07,イータ工業,セクターA,920,18.4,18,61,700,普通 SMPL-08,シータ建材,セクターE,2100,84,80,155,1650,普通 SMPL-09,イオタ化学,セクターC,1450,0,0,88,1100,無配 SMPL-10,カッパ精密,セクターE,660,0,0,25,540,無配 SMPL-11,ラムダ運輸,セクターB,305,0,0,12,260,無配 SMPL-12,ミュー電機,セクターA,1870,,,120,1400, SMPL-13,ニュー繊維,セクターD,512,,12,33,480, SMPL-14,クサイ金属,セクターE,430,8.6,12,-22,380,普通 SMPL-15,オミクロン印刷,セクターD,268,0,0,-15,240,無配 SMPL-16,パイ機械,セクターA,1120,22.4,30,-40,900,普通 SMPL-17,ロー通信,セクターC,2050,41,40,,1600,普通 SMPL-18,シグマ銀行,セクターC,480,19.2,18,55,620,普通 SMPL-19,タウ鉄鋼,セクターE,1150,46,44,130,1500,普通 SMPL-20,ウプシロン海運,セクターB,890,35.6,32,145,1250,普通 SMPL-21,ファイ興業,セクターE,210,0,6,-30,-55,無配 SMPL-22,カイ不動産,セクターC,620,31,44,40,700,普通 SMPL-23,プサイ電設,セクターA,102,3.57,5.2,9,105,普通 SMPL-24,オメガ商会,セクターD,1500,75,30,90,1150,記念配当を含む SMPL-25,サンプル建設,セクターE,840,42,21,60,700,特別配当を含む """ def write_sample(path: Path, force: bool) -> None: """架空のサンプルCSVを書き出す。既存ファイルは --force なしでは上書きしない。""" if path.exists() and not force: raise SystemExit( f"[中止] {path} はすでにあります。上書きするなら --force を付けてください。" ) path.write_text(SAMPLE_CSV, encoding="utf-8", newline="\n") print(f"架空のサンプルCSVを書き出しました: {path}（25行・実在の銘柄ではありません）") def load_stocks(path: Path) -> pd.DataFrame: """CSVを読む。codeは必ず文字列、指標のもとになる列は数値に強制する。""" df = pd.read_csv(path, dtype={"code": str, "name": str, "sector": str, "dividend_note": str}) for col in NUMERIC_COLS: # 「-」「—」などが混ざると列がobjectになるので、数値以外はNaNに落とす df[col] = pd.to_numeric(df[col], errors="coerce") return df def add_metrics(df: pd.DataFrame) -> pd.DataFrame: """配当利回り・PER・PBR を計算する。算出できない行はNaNにする。""" df = df.copy() price = df["price"].where(df["price"] > 0) df["yield_pct"] = df["dividend"] / price * 100 df["per"] = price / df["eps"].where(df["eps"] > 0) # 赤字（eps<=0）は算出不能 df["pbr"] = price / df["bps"].where(df["bps"] > 0) # 債務超過（bps<=0）は算出不能 return df def sen(series: pd.Series) -> pd.Series: """円を銭（1/100円）の整数に直す。比較を整数で行い、割り算の誤差を持ち込まない。""" return series.mul(100).round().astype("Float64").astype("Int64") def cond_min_yield(df: pd.DataFrame, value: float): """配当利回り >= value(%)。除算をせず、両辺に掛けて整数で比べる。""" bp = round(value * 100) # 3.5% -> 350（ベーシスポイント） usable = df["dividend"].notna() & (df["price"] > 0) ok = (sen(df["dividend"]) * 10000 >= bp * sen(df["price"])).fillna(False) return ok.astype(bool) & usable, usable def cond_max_per(df: pd.DataFrame, value: float): """PER <= value(倍)。1株利益が0以下の行は「算出不能」として判定から外す。""" hundredths = round(value * 100) # 15.0倍 -> 1500 usable = (df["eps"] > 0) & (df["price"] > 0) ok = (sen(df["price"]) * 100 <= hundredths * sen(df["eps"])).fillna(False) return ok.astype(bool) & usable, usable def cond_max_pbr(df: pd.DataFrame, value: float): """PBR <= value(倍)。1株純資産が0以下の行は「算出不能」として判定から外す。""" hundredths = round(value * 100) usable = (df["bps"] > 0) & (df["price"] > 0) ok = (sen(df["price"]) * 100 <= hundredths * sen(df["bps"])).fillna(False) return ok.astype(bool) & usable, usable def build_conditions(df: pd.DataFrame, args) -> dict: """指定されたしきい値だけを条件に組み立てる。指定が無い条件は評価しない。""" conds = {} if args.min_yield is not None: conds[f"配当利回り >= {args.min_yield}%"] = cond_min_yield(df, args.min_yield) if args.max_per is not None: conds[f"PER <= {args.max_per}倍"] = cond_max_per(df, args.max_per) if args.max_pbr is not None: conds[f"PBR <= {args.max_pbr}倍"] = cond_max_pbr(df, args.max_pbr) return conds def screen(df: pd.DataFrame, conds: dict): """通過・条件落ち・判定不能の3つに分ける。この3つの合計が必ず全件になる。""" ok_all = pd.Series(True, index=df.index) usable_all = pd.Series(True, index=df.index) for ok, usable in conds.values(): ok_all &= ok usable_all &= usable passed = usable_all & ok_all dropped = usable_all & ~ok_all unknown = ~usable_all return passed, dropped, unknown def report(df: pd.DataFrame, conds: dict, passed, dropped, unknown, sort_by: str, source: Path) -> pd.DataFrame: """適用した条件・件数の内訳・通過した行を標準出力に出す。""" print(f"入力: {source}（{len(df)}件）") print("適用した条件（いずれも動作確認用の値で、推奨値ではありません）:") for label in conds: print(f" - {label}") print("-" * 64) print(f" 条件を満たした : {int(passed.sum())}件") print(f" 条件を満たさなかった: {int(dropped.sum())}件") print(f" 判定できなかった : {int(unknown.sum())}件（データ欠損・算出不能）") print(f" 合計 : {len(df)}件") print("-" * 64) for label, (ok, usable) in conds.items(): print(f" [{label}] 判定できた {int(usable.sum())}件 / うち通過 {int((ok & usable).sum())}件") result = df.loc[passed].sort_values(sort_by, ascending=False, na_position="last") view = result[["code", "name", "sector", "price", "yield_pct", "per", "pbr", "dividend", "dividend_prev", "dividend_note"]] print("\n■ 条件を満たした行（並び順は指標の大小であって、投資対象としての順位ではありません）") print(view.to_string(index=False, float_format=lambda v: f"{v:,.2f}")) cut = int((result["dividend"] < result["dividend_prev"]).sum()) onetime = int(result["dividend_note"].isin(["記念配当を含む", "特別配当を含む"]).sum()) print(f"\n参考（判定には使っていません）: 上の{len(result)}件のうち、" f"前期より配当額が少ない行が{cut}件、一過性の配当を含むと記載された行が{onetime}件あります。") missing = {col: int(df[col].isna().sum()) for col in NUMERIC_COLS} print("欠損の内訳: " + " / ".join(f"{k}={v}" for k, v in missing.items())) return view def save_csv(view: pd.DataFrame, out: Path, conds: dict, source: Path, total: int) -> None: """条件と件数を先頭にコメント行として刷ってからCSVを書く。""" with out.open("w", encoding="utf-8-sig", newline="") as f: f.write(f"# source,{source.name}\n") f.write(f"# total_rows,{total}\n") f.write(f"# matched_rows,{len(view)}\n") for i, label in enumerate(conds, 1): f.write(f"# condition_{i},{label}\n") f.write("# note,条件に合う行を機械的に抽出しただけのファイルです。投資判断ではありません。\n") view.to_csv(f, index=False, float_format="{:.2f}") print(f"\n書き出しました: {out}（読み戻すときは pd.read_csv(path, comment=\"#\")）") def fetch_one(ticker: str) -> None: """指定した1銘柄の生データを取る。ここだけ Yahoo! Finance と通信する。""" import yfinance as yf print(f"[通信します] {ticker} の情報を Yahoo! Finance から取得します（送るのは銘柄コードだけです）") info = yf.Ticker(ticker).info keys = ["currentPrice", "previousClose", "dividendYield", "dividendRate", "trailingPE", "forwardPE", "priceToBook", "bookValue", "currency", "financialCurrency"] for key in keys: print(f" {key:<18}: {info.get(key, '(キーなし)')}") rate, price = info.get("dividendRate"), info.get("currentPrice") if rate is not None and price: print(f" dividendRate / currentPrice * 100 = {rate / price * 100:.4f}") print(" ↑ この自己計算と dividendYield を見比べて、単位（3.2 なのか 0.032 なのか）を確かめてください。") else: print(" dividendRate か currentPrice が無いため、利回りの自己計算はできません。") print(" 取得した値は取得時点のものです。数値の良し悪しは本ツールでは判断しません。") def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( description="架空銘柄のCSVを配当利回り・PER・PBRで絞り込む（投資助言ではありません）") p.add_argument("--csv", type=Path, default=DEFAULT_CSV, help="入力CSV（既定: sample_stocks.csv）") p.add_argument("--init-sample", action="store_true", help="架空のサンプルCSVを書き出して終了する") p.add_argument("--force", action="store_true", help="--init-sample で既存ファイルを上書きする") # しきい値に default は置かない（既定値を置くと推奨値に見えるため） p.add_argument("--min-yield", type=float, help="配当利回りの下限（%%）") p.add_argument("--max-per", type=float, help="PERの上限（倍）") p.add_argument("--max-pbr", type=float, help="PBRの上限（倍）") p.add_argument("--sort", default="yield_pct", choices=["yield_pct", "per", "pbr", "price"], help="並べ替えに使う列（既定: yield_pct）") p.add_argument("--out", type=Path, help="結果を書き出すCSVのパス") p.add_argument("--fetch", metavar="TICKER", help="1銘柄の生データを取得する（外部と通信する）") return p def main(argv=None) -> int: parser = build_parser() args = parser.parse_args(argv) if args.init_sample: write_sample(args.csv, args.force) return 0 if args.fetch: fetch_one(args.fetch) return 0 if args.min_yield is None and args.max_per is None and args.max_pbr is None: parser.error("しきい値を少なくとも1つ指定してください（--min-yield / --max-per / --max-pbr）。" "既定値は用意していません。") if not args.csv.exists(): raise SystemExit(f"[中止] {args.csv} がありません。--init-sample で作れます。") df = add_metrics(load_stocks(args.csv)) conds = build_conditions(df, args) passed, dropped, unknown = screen(df, conds) view = report(df, conds, passed, dropped, unknown, args.sort, args.csv) if args.out: save_csv(view, args.out, conds, args.csv, len(df)) return 0 if __name__ == "__main__": sys.exit(main())